vector/internal_events/
websocket.rs

1#![allow(dead_code)] // TODO requires optional feature compilation
2
3use std::{
4    error::Error,
5    fmt::{Debug, Display, Formatter, Result},
6};
7
8use tokio_tungstenite::tungstenite::error::Error as TungsteniteError;
9use vector_common::{
10    internal_event::{error_stage, error_type},
11    json_size::JsonSize,
12};
13use vector_lib::{
14    NamedInternalEvent, counter, histogram,
15    internal_event::{CounterName, HistogramName, InternalEvent},
16};
17
18pub const PROTOCOL: &str = "websocket";
19
20#[derive(Debug, NamedInternalEvent)]
21pub struct WebSocketConnectionEstablished;
22
23impl InternalEvent for WebSocketConnectionEstablished {
24    fn emit(self) {
25        debug!(message = "Connected.");
26        counter!(CounterName::ConnectionEstablishedTotal).increment(1);
27    }
28}
29
30#[derive(Debug, NamedInternalEvent)]
31pub struct WebSocketConnectionFailedError {
32    pub error: Box<dyn Error>,
33}
34
35impl InternalEvent for WebSocketConnectionFailedError {
36    fn emit(self) {
37        error!(
38            message = "WebSocket connection failed.",
39            error = %self.error,
40            error_code = "websocket_connection_error",
41            error_type = error_type::CONNECTION_FAILED,
42            stage = error_stage::SENDING,
43        );
44        counter!(
45            CounterName::ComponentErrorsTotal,
46            "protocol" => PROTOCOL,
47            "error_code" => "websocket_connection_failed",
48            "error_type" => error_type::CONNECTION_FAILED,
49            "stage" => error_stage::SENDING,
50        )
51        .increment(1);
52    }
53}
54
55#[derive(Debug, NamedInternalEvent)]
56pub struct WebSocketConnectionShutdown;
57
58impl InternalEvent for WebSocketConnectionShutdown {
59    fn emit(self) {
60        warn!(message = "Closed by the server.");
61        counter!(CounterName::ConnectionShutdownTotal).increment(1);
62    }
63}
64
65#[derive(Debug, NamedInternalEvent)]
66pub struct WebSocketConnectionError {
67    pub error: tokio_tungstenite::tungstenite::Error,
68}
69
70impl InternalEvent for WebSocketConnectionError {
71    fn emit(self) {
72        error!(
73            message = "WebSocket connection error.",
74            error = %self.error,
75            error_code = "websocket_connection_error",
76            error_type = error_type::WRITER_FAILED,
77            stage = error_stage::SENDING,
78        );
79        counter!(
80            CounterName::ComponentErrorsTotal,
81            "protocol" => PROTOCOL,
82            "error_code" => "websocket_connection_error",
83            "error_type" => error_type::WRITER_FAILED,
84            "stage" => error_stage::SENDING,
85        )
86        .increment(1);
87    }
88}
89
90#[allow(dead_code)]
91#[derive(Debug, Copy, Clone)]
92pub enum WebSocketKind {
93    Ping,
94    Pong,
95    Text,
96    Binary,
97    Close,
98    Frame,
99}
100
101impl Display for WebSocketKind {
102    fn fmt(&self, f: &mut Formatter<'_>) -> Result {
103        write!(f, "{self:?}")
104    }
105}
106
107#[derive(Debug, NamedInternalEvent)]
108pub struct WebSocketBytesReceived<'a> {
109    pub byte_size: usize,
110    pub url: &'a str,
111    pub protocol: &'static str,
112    pub kind: WebSocketKind,
113}
114
115impl InternalEvent for WebSocketBytesReceived<'_> {
116    fn emit(self) {
117        trace!(
118            message = "Bytes received.",
119            byte_size = %self.byte_size,
120            url = %self.url,
121            protocol = %self.protocol,
122            kind = %self.kind
123        );
124        let counter = counter!(
125            CounterName::ComponentReceivedBytesTotal,
126            "url" => self.url.to_string(),
127            "protocol" => self.protocol,
128            "kind" => self.kind.to_string()
129        );
130        counter.increment(self.byte_size as u64);
131    }
132}
133
134#[derive(Debug, NamedInternalEvent)]
135pub struct WebSocketMessageReceived<'a> {
136    pub count: usize,
137    pub byte_size: JsonSize,
138    pub url: &'a str,
139    pub protocol: &'static str,
140    pub kind: WebSocketKind,
141}
142
143impl InternalEvent for WebSocketMessageReceived<'_> {
144    fn emit(self) {
145        trace!(
146            message = "Events received.",
147            count = %self.count,
148            byte_size = %self.byte_size,
149            url =  %self.url,
150            protocol = %self.protocol,
151            kind = %self.kind
152        );
153
154        let histogram = histogram!(HistogramName::ComponentReceivedEventsCount);
155        histogram.record(self.count as f64);
156        let counter = counter!(
157            CounterName::ComponentReceivedEventsTotal,
158            "uri" => self.url.to_string(),
159            "protocol" => PROTOCOL,
160            "kind" => self.kind.to_string()
161        );
162        counter.increment(self.count as u64);
163        let counter = counter!(
164            CounterName::ComponentReceivedEventBytesTotal,
165            "url" => self.url.to_string(),
166            "protocol" => PROTOCOL,
167            "kind" => self.kind.to_string()
168        );
169        counter.increment(self.byte_size.get() as u64);
170    }
171}
172
173#[derive(Debug, NamedInternalEvent)]
174pub struct WebSocketReceiveError<'a> {
175    pub error: &'a TungsteniteError,
176}
177
178impl InternalEvent for WebSocketReceiveError<'_> {
179    fn emit(self) {
180        error!(
181            message = "Error receiving message from websocket.",
182            error = %self.error,
183            error_code = "websocket_receive_error",
184            error_type = error_type::CONNECTION_FAILED,
185            stage = error_stage::PROCESSING,
186        );
187        counter!(
188            CounterName::ComponentErrorsTotal,
189            "protocol" => PROTOCOL,
190            "error_code" => "websocket_receive_error",
191            "error_type" => error_type::CONNECTION_FAILED,
192            "stage" => error_stage::PROCESSING,
193        )
194        .increment(1);
195    }
196}
197
198#[derive(Debug, NamedInternalEvent)]
199pub struct WebSocketSendError<'a> {
200    pub error: &'a TungsteniteError,
201}
202
203impl InternalEvent for WebSocketSendError<'_> {
204    fn emit(self) {
205        error!(
206            message = "Error sending message to websocket.",
207            error = %self.error,
208            error_code = "websocket_send_error",
209            error_type = error_type::CONNECTION_FAILED,
210            stage = error_stage::PROCESSING,
211        );
212        counter!(
213            CounterName::ComponentErrorsTotal,
214            "protocol" => PROTOCOL,
215            "error_code" => "websocket_send_error",
216            "error_type" => error_type::CONNECTION_FAILED,
217            "stage" => error_stage::PROCESSING,
218        )
219        .increment(1);
220    }
221}
vector/internal_events/websocket.rs

vector/internal_events/
websocket.rs
