vector/internal_events/
tcp.rs

1use std::net::SocketAddr;
2
3use vector_lib::{
4    NamedInternalEvent, counter,
5    internal_event::{CounterName, InternalEvent, error_stage, error_type},
6};
7
8use crate::{internal_events::SocketOutgoingConnectionError, tls::TlsError};
9
10#[derive(Debug, NamedInternalEvent)]
11pub struct TcpSocketConnectionEstablished {
12    pub peer_addr: Option<SocketAddr>,
13}
14
15impl InternalEvent for TcpSocketConnectionEstablished {
16    fn emit(self) {
17        if let Some(peer_addr) = self.peer_addr {
18            debug!(message = "Connected.", %peer_addr);
19        } else {
20            debug!(message = "Connected.", peer_addr = "unknown");
21        }
22        counter!(CounterName::ConnectionEstablishedTotal, "mode" => "tcp").increment(1);
23    }
24}
25
26#[derive(Debug, NamedInternalEvent)]
27pub struct TcpSocketOutgoingConnectionError<E> {
28    pub error: E,
29}
30
31impl<E: std::error::Error> InternalEvent for TcpSocketOutgoingConnectionError<E> {
32    fn emit(self) {
33        // ## skip check-duplicate-events ##
34        // ## skip check-validity-events ##
35        emit!(SocketOutgoingConnectionError { error: self.error });
36    }
37}
38
39#[derive(Debug, NamedInternalEvent)]
40pub struct TcpSocketConnectionShutdown;
41
42impl InternalEvent for TcpSocketConnectionShutdown {
43    fn emit(self) {
44        warn!(message = "Received EOF from the server, shutdown.");
45        counter!(CounterName::ConnectionShutdownTotal, "mode" => "tcp").increment(1);
46    }
47}
48
49/// Emitted once per accepted TCP source connection, after the per-connection
50/// task exits — regardless of cause. This includes pre-loop exits (TLS
51/// handshake failure, shutdown signal arriving during handshake) as well as
52/// every read/ack loop exit (graceful peer EOF, decoder failure, downstream
53/// closed, ack write failure, shutdown signal, tripwire, max connection
54/// duration). Pairs exactly with `ConnectionOpen`.
55#[derive(Debug, NamedInternalEvent)]
56pub struct TcpSourceConnectionClosed;
57
58impl InternalEvent for TcpSourceConnectionClosed {
59    fn emit(self) {
60        debug!(message = "Connection closed.");
61        counter!(CounterName::ConnectionShutdownTotal, "mode" => "tcp").increment(1);
62    }
63}
64
65#[cfg(all(unix, feature = "sources-dnstap"))]
66#[derive(Debug, NamedInternalEvent)]
67pub struct TcpSocketError<'a, E> {
68    pub(crate) error: &'a E,
69    pub peer_addr: SocketAddr,
70}
71
72#[cfg(all(unix, feature = "sources-dnstap"))]
73impl<E: std::fmt::Display> InternalEvent for TcpSocketError<'_, E> {
74    fn emit(self) {
75        error!(
76            message = "TCP socket error.",
77            error = %self.error,
78            peer_addr = ?self.peer_addr,
79            error_type = error_type::CONNECTION_FAILED,
80            stage = error_stage::PROCESSING,
81        );
82        counter!(
83            CounterName::ComponentErrorsTotal,
84            "error_type" => error_type::CONNECTION_FAILED,
85            "stage" => error_stage::PROCESSING,
86        )
87        .increment(1);
88    }
89}
90
91#[derive(Debug, NamedInternalEvent)]
92pub struct TcpSocketTlsConnectionError {
93    pub error: TlsError,
94}
95
96impl InternalEvent for TcpSocketTlsConnectionError {
97    fn emit(self) {
98        match self.error {
99            // Specific error that occurs when the other side is only
100            // doing SYN/SYN-ACK connections for healthcheck.
101            // https://github.com/vectordotdev/vector/issues/7318
102            TlsError::Handshake { ref source }
103                if source.code() == openssl::ssl::ErrorCode::SYSCALL
104                    && source.io_error().is_none() =>
105            {
106                debug!(
107                    message = "Connection error, probably a healthcheck.",
108                    error = %self.error,
109                );
110            }
111            _ => {
112                error!(
113                    message = "Connection error.",
114                    error = %self.error,
115                    error_code = "connection_failed",
116                    error_type = error_type::WRITER_FAILED,
117                    stage = error_stage::SENDING,
118                );
119                counter!(
120                    CounterName::ComponentErrorsTotal,
121                    "error_code" => "connection_failed",
122                    "error_type" => error_type::WRITER_FAILED,
123                    "stage" => error_stage::SENDING,
124                    "mode" => "tcp",
125                )
126                .increment(1);
127            }
128        }
129    }
130}
131
132/// Emitted when a TLS handshake does not complete within the configured
133/// handshake timeout. Without this, a peer that opens a TCP connection but
134/// never completes (or never starts) the TLS handshake — a bare TCP
135/// healthcheck, a stalled client, or deliberate slow-handshake abuse — would
136/// hold its connection-limit slot indefinitely, since neither TCP keepalive
137/// nor `max_connection_duration_secs` are wired up until after the handshake
138/// succeeds.
139#[derive(Debug, NamedInternalEvent)]
140pub struct TcpSocketTlsHandshakeTimeout {
141    pub peer_addr: SocketAddr,
142    pub timeout: std::time::Duration,
143}
144
145impl InternalEvent for TcpSocketTlsHandshakeTimeout {
146    fn emit(self) {
147        warn!(
148            message = "TLS handshake timed out.",
149            peer_addr = %self.peer_addr,
150            timeout_secs = self.timeout.as_secs(),
151            error_code = "tls_handshake_timeout",
152            error_type = error_type::TIMED_OUT,
153            stage = error_stage::RECEIVING,
154        );
155        counter!(
156            CounterName::ComponentErrorsTotal,
157            "error_code" => "tls_handshake_timeout",
158            "error_type" => error_type::TIMED_OUT,
159            "stage" => error_stage::RECEIVING,
160            "mode" => "tcp",
161        )
162        .increment(1);
163    }
164}
165
166#[derive(Debug, NamedInternalEvent)]
167pub struct TcpSendAckError {
168    pub error: std::io::Error,
169}
170
171impl InternalEvent for TcpSendAckError {
172    fn emit(self) {
173        error!(
174            message = "Error writing acknowledgement, dropping connection.",
175            error = %self.error,
176            error_code = "ack_failed",
177            error_type = error_type::WRITER_FAILED,
178            stage = error_stage::SENDING,
179        );
180        counter!(
181            CounterName::ComponentErrorsTotal,
182            "error_code" => "ack_failed",
183            "error_type" => error_type::WRITER_FAILED,
184            "stage" => error_stage::SENDING,
185            "mode" => "tcp",
186        )
187        .increment(1);
188    }
189}
190
191#[derive(Debug, NamedInternalEvent)]
192pub struct TcpBytesReceived {
193    pub byte_size: usize,
194    pub peer_addr: SocketAddr,
195}
196
197impl InternalEvent for TcpBytesReceived {
198    fn emit(self) {
199        trace!(
200            message = "Bytes received.",
201            protocol = "tcp",
202            byte_size = %self.byte_size,
203            peer_addr = %self.peer_addr,
204        );
205        counter!(
206            CounterName::ComponentReceivedBytesTotal, "protocol" => "tcp"
207        )
208        .increment(self.byte_size as u64);
209    }
210}
211
212#[cfg(test)]
213mod tests {
214    use std::io;
215
216    use serial_test::serial;
217    use vector_lib::event::MetricValue;
218    use vector_lib::internal_event::InternalEvent;
219    use vector_lib::metrics::Controller;
220
221    use super::{TcpSendAckError, TcpSourceConnectionClosed};
222
223    /// Returns the current value of a counter matching `name` and all `tags`.
224    /// Counters that have not yet been touched aren't in the snapshot and
225    /// register as 0.0 here.
226    fn counter_value(name: &str, tags: &[(&str, &str)]) -> f64 {
227        Controller::get()
228            .expect("metrics controller initialized")
229            .capture_metrics()
230            .into_iter()
231            .find(|m| {
232                m.name() == name
233                    && tags
234                        .iter()
235                        .all(|(k, v)| m.tags().is_some_and(|t| t.get(k) == Some(*v)))
236            })
237            .map(|m| match m.value() {
238                MetricValue::Counter { value } => *value,
239                other => panic!("expected counter for {name}, got {other:?}"),
240            })
241            .unwrap_or(0.0)
242    }
243
244    /// `TcpSourceConnectionClosed` MUST bump `connection_shutdown_total{mode="tcp"}`
245    /// once per emit. Pins the contract that this event is the sole owner of the
246    /// connection-close counter on the source side.
247    #[test]
248    #[serial]
249    fn tcp_source_connection_closed_increments_shutdown_total() {
250        crate::test_util::trace_init();
251        let before = counter_value("connection_shutdown_total", &[("mode", "tcp")]);
252
253        TcpSourceConnectionClosed.emit();
254
255        let after = counter_value("connection_shutdown_total", &[("mode", "tcp")]);
256        assert_eq!(after - before, 1.0);
257    }
258
259    /// `TcpSendAckError` is an `*Error` event and per the instrumentation spec MUST
260    /// only emit on real errors — bumping `component_errors_total` with the
261    /// `ack_failed` error_code.
262    #[test]
263    #[serial]
264    fn tcp_send_ack_error_emit_always_increments_component_errors_total() {
265        crate::test_util::trace_init();
266        let errors_before = counter_value(
267            "component_errors_total",
268            &[
269                ("error_code", "ack_failed"),
270                ("error_type", "writer_failed"),
271                ("stage", "sending"),
272                ("mode", "tcp"),
273            ],
274        );
275        let shutdown_before = counter_value("connection_shutdown_total", &[("mode", "tcp")]);
276
277        TcpSendAckError {
278            error: io::Error::from(io::ErrorKind::ConnectionReset),
279        }
280        .emit();
281
282        assert_eq!(
283            counter_value(
284                "component_errors_total",
285                &[
286                    ("error_code", "ack_failed"),
287                    ("error_type", "writer_failed"),
288                    ("stage", "sending"),
289                    ("mode", "tcp"),
290                ],
291            ) - errors_before,
292            1.0,
293        );
294        assert_eq!(
295            counter_value("connection_shutdown_total", &[("mode", "tcp")]),
296            shutdown_before,
297            "TcpSendAckError must not bump the connection-close counter — \
298             that is TcpSourceConnectionClosed's responsibility.",
299        );
300    }
301}
vector/internal_events/tcp.rs

vector/internal_events/
tcp.rs
