fix(shutdown): Update known lifecycle flow to warn not error (#25458)

* Update known lifecycle flow to warn not error

* Adding changelog

* Refactor the stat so its done on every shutdown regardless of cause

* Adding tests

* fix issue found by codex

* update comment on new behavior

* More doc updates

* fmt update

* Rename and move logic to call site

* Move function to shared helper file

* fmt pass

Author: taylorchandleryoung

changelog.d/tcp_ack_graceful_tls_shutdown.fix.md

@@ -0,0 +1,5 @@
+TCP-based sources that emit acknowledgements (`fluent`, `logstash`) no longer log a spurious `Error writing acknowledgement, dropping connection.` at ERROR level when the ack write fails because the peer cleanly closed its TLS session (for example, during a rolling pod restart). These graceful shutdowns now log at WARN and no longer increment `component_errors_total{error_code="ack_failed", ...}`, preventing operator dashboards/alerts from firing on routine peer disconnects. Genuine ack write failures are still logged at ERROR and continue to increment `component_errors_total`.
+
+The `connection_shutdown_total{mode="tcp"}` counter is now incremented exactly once per accepted source connection when its per-connection task exits, pairing with `ConnectionOpen` — regardless of cause (TLS handshake failure, shutdown signal during handshake, graceful peer EOF, decoder failure, downstream closed, ack write failure, tripwire, max connection duration). Previously it was not emitted by TCP sources at all.
+
+authors: taylorchandleryoung

src/internal_events/tcp.rs

@@ -46,6 +46,22 @@ impl InternalEvent for TcpSocketConnectionShutdown {
     }
 }
 
+/// Emitted once per accepted TCP source connection, after the per-connection
+/// task exits — regardless of cause. This includes pre-loop exits (TLS
+/// handshake failure, shutdown signal arriving during handshake) as well as
+/// every read/ack loop exit (graceful peer EOF, decoder failure, downstream
+/// closed, ack write failure, shutdown signal, tripwire, max connection
+/// duration). Pairs exactly with `ConnectionOpen`.
+#[derive(Debug, NamedInternalEvent)]
+pub struct TcpSourceConnectionClosed;
+
+impl InternalEvent for TcpSourceConnectionClosed {
+    fn emit(self) {
+        debug!(message = "Connection closed.");
+        counter!(CounterName::ConnectionShutdownTotal, "mode" => "tcp").increment(1);
+    }
+}
+
 #[cfg(all(unix, feature = "sources-dnstap"))]
 #[derive(Debug, NamedInternalEvent)]
 pub struct TcpSocketError<'a, E> {
@@ -158,3 +174,94 @@ impl InternalEvent for TcpBytesReceived {
         .increment(self.byte_size as u64);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::io;
+
+    use serial_test::serial;
+    use vector_lib::event::MetricValue;
+    use vector_lib::internal_event::InternalEvent;
+    use vector_lib::metrics::Controller;
+
+    use super::{TcpSendAckError, TcpSourceConnectionClosed};
+
+    /// Returns the current value of a counter matching `name` and all `tags`.
+    /// Counters that have not yet been touched aren't in the snapshot and
+    /// register as 0.0 here.
+    fn counter_value(name: &str, tags: &[(&str, &str)]) -> f64 {
+        Controller::get()
+            .expect("metrics controller initialized")
+            .capture_metrics()
+            .into_iter()
+            .find(|m| {
+                m.name() == name
+                    && tags
+                        .iter()
+                        .all(|(k, v)| m.tags().is_some_and(|t| t.get(k) == Some(*v)))
+            })
+            .map(|m| match m.value() {
+                MetricValue::Counter { value } => *value,
+                other => panic!("expected counter for {name}, got {other:?}"),
+            })
+            .unwrap_or(0.0)
+    }
+
+    /// `TcpSourceConnectionClosed` MUST bump `connection_shutdown_total{mode="tcp"}`
+    /// once per emit. Pins the contract that this event is the sole owner of the
+    /// connection-close counter on the source side.
+    #[test]
+    #[serial]
+    fn tcp_source_connection_closed_increments_shutdown_total() {
+        crate::test_util::trace_init();
+        let before = counter_value("connection_shutdown_total", &[("mode", "tcp")]);
+
+        TcpSourceConnectionClosed.emit();
+
+        let after = counter_value("connection_shutdown_total", &[("mode", "tcp")]);
+        assert_eq!(after - before, 1.0);
+    }
+
+    /// `TcpSendAckError` is an `*Error` event and per the instrumentation spec MUST
+    /// only emit on real errors — bumping `component_errors_total` with the
+    /// `ack_failed` error_code.
+    #[test]
+    #[serial]
+    fn tcp_send_ack_error_emit_always_increments_component_errors_total() {
+        crate::test_util::trace_init();
+        let errors_before = counter_value(
+            "component_errors_total",
+            &[
+                ("error_code", "ack_failed"),
+                ("error_type", "writer_failed"),
+                ("stage", "sending"),
+                ("mode", "tcp"),
+            ],
+        );
+        let shutdown_before = counter_value("connection_shutdown_total", &[("mode", "tcp")]);
+
+        TcpSendAckError {
+            error: io::Error::from(io::ErrorKind::ConnectionReset),
+        }
+        .emit();
+
+        assert_eq!(
+            counter_value(
+                "component_errors_total",
+                &[
+                    ("error_code", "ack_failed"),
+                    ("error_type", "writer_failed"),
+                    ("stage", "sending"),
+                    ("mode", "tcp"),
+                ],
+            ) - errors_before,
+            1.0,
+        );
+        assert_eq!(
+            counter_value("connection_shutdown_total", &[("mode", "tcp")]),
+            shutdown_before,
+            "TcpSendAckError must not bump the connection-close counter — \
+             that is TcpSourceConnectionClosed's responsibility.",
+        );
+    }
+}

src/net.rs

@@ -50,3 +50,129 @@ where
 pub fn set_keepalive(socket: &TcpStream, ttl: Duration) -> io::Result<()> {
     SockRef::from(socket).set_tcp_keepalive(&TcpKeepalive::new().with_time(ttl))
 }
+
+// SSL_R_PROTOCOL_IS_SHUTDOWN from openssl/include/openssl/sslerr.h. Stable across
+// OpenSSL 1.1.1 and 3.x. Not re-exported by the `openssl-sys` crate so we pin it here.
+const SSL_R_PROTOCOL_IS_SHUTDOWN: std::ffi::c_int = 207;
+
+/// Returns true when an `io::Error` represents a peer-initiated, graceful TLS
+/// shutdown (close_notify), rather than a real I/O failure.
+///
+/// Two cases are recognized:
+/// - `SSL_ERROR_ZERO_RETURN`: the peer sent `close_notify` and we observed it
+///   during this I/O call.
+/// - `SSL_R_PROTOCOL_IS_SHUTDOWN`: a subsequent write after the session was
+///   already shut down ("ssl session has been shut down").
+pub fn is_graceful_tls_shutdown(err: &io::Error) -> bool {
+    let Some(ssl) = err
+        .get_ref()
+        .and_then(|inner| inner.downcast_ref::<openssl::ssl::Error>())
+    else {
+        return false;
+    };
+    if ssl.code() == openssl::ssl::ErrorCode::ZERO_RETURN {
+        return true;
+    }
+    ssl.ssl_error().is_some_and(|stack| {
+        stack
+            .errors()
+            .iter()
+            .any(|e| e.reason_code() == SSL_R_PROTOCOL_IS_SHUTDOWN)
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use std::pin::Pin;
+
+    use openssl::ssl::{SslAcceptor, SslConnector, SslFiletype, SslMethod, SslVerifyMode};
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::TcpListener;
+    use tokio_openssl::SslStream;
+
+    use crate::tls::{TEST_PEM_CA_PATH, TEST_PEM_CRT_PATH, TEST_PEM_KEY_PATH};
+
+    use super::{TcpStream, io, is_graceful_tls_shutdown};
+
+    #[test]
+    fn plain_io_errors_are_not_graceful() {
+        for err in [
+            io::Error::from(io::ErrorKind::BrokenPipe),
+            io::Error::from(io::ErrorKind::ConnectionReset),
+            io::Error::from(io::ErrorKind::UnexpectedEof),
+            io::Error::other("not an ssl error"),
+        ] {
+            assert!(
+                !is_graceful_tls_shutdown(&err),
+                "expected non-graceful, got graceful for {err:?}",
+            );
+        }
+    }
+
+    // Drives a real TLS handshake between two local sockets and completes a
+    // bidirectional SSL shutdown. A subsequent write surfaces a `std::io::Error`
+    // wrapping an `openssl::ssl::Error` from the same code path production hits,
+    // validating that the helper correctly identifies it as a graceful shutdown
+    // — without having to synthesize an `openssl::ssl::Error` (whose fields are
+    // crate-private). Bidirectional shutdown is what reliably elicits
+    // SSL_R_PROTOCOL_IS_SHUTDOWN; a half-closed session would still permit
+    // writes per RFC 5246.
+    #[tokio::test]
+    async fn detects_graceful_shutdown_from_real_ssl_stream() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+
+        let server = tokio::spawn(async move {
+            let (stream, _) = listener.accept().await.unwrap();
+            let mut acceptor = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap();
+            acceptor
+                .set_private_key_file(TEST_PEM_KEY_PATH, SslFiletype::PEM)
+                .unwrap();
+            acceptor
+                .set_certificate_chain_file(TEST_PEM_CRT_PATH)
+                .unwrap();
+            let acceptor = acceptor.build();
+            let ssl = openssl::ssl::Ssl::new(acceptor.context()).unwrap();
+            let mut tls = SslStream::new(ssl, stream).unwrap();
+            Pin::new(&mut tls).accept().await.unwrap();
+            // Cleanly close the SSL session — sends close_notify and waits for the peer's.
+            Pin::new(&mut tls).shutdown().await.unwrap();
+        });
+
+        let mut connector = SslConnector::builder(SslMethod::tls()).unwrap();
+        connector.set_ca_file(TEST_PEM_CA_PATH).unwrap();
+        connector.set_verify(SslVerifyMode::NONE);
+        let ssl = connector
+            .build()
+            .configure()
+            .unwrap()
+            .into_ssl("localhost")
+            .unwrap();
+        let stream = TcpStream::connect(addr).await.unwrap();
+        let mut tls = SslStream::new(ssl, stream).unwrap();
+        Pin::new(&mut tls).connect().await.unwrap();
+
+        // Drain the server's close_notify so our SSL state observes the peer shutdown.
+        let mut buf = [0u8; 1];
+        let n = tls.read(&mut buf).await.unwrap();
+        assert_eq!(n, 0, "expected EOF from peer's close_notify");
+
+        // Complete the bidirectional SSL shutdown locally. Once both sides are
+        // shut down, OpenSSL marks the session as SHUTDOWN and any further write
+        // returns SSL_R_PROTOCOL_IS_SHUTDOWN ("ssl session has been shut down").
+        Pin::new(&mut tls).shutdown().await.unwrap();
+
+        let err = tls
+            .write_all(b"too late")
+            .await
+            .expect_err("write after bidirectional shutdown should fail");
+
+        assert!(
+            is_graceful_tls_shutdown(&err),
+            "expected graceful shutdown detection, got: {err:?} (inner: {:?})",
+            err.get_ref(),
+        );
+
+        server.await.unwrap();
+    }
+}

src/sources/util/net/tcp/mod.rs

@@ -37,8 +37,9 @@ use crate::{
     internal_events::{
         ConnectionOpen, OpenGauge, SocketBindError, SocketEventsReceived, SocketMode,
         SocketReceiveError, StreamClosedError, TcpBytesReceived, TcpSendAckError,
-        TcpSocketTlsConnectionError,
+        TcpSocketTlsConnectionError, TcpSourceConnectionClosed,
     },
+    net::is_graceful_tls_shutdown,
     sources::util::{AfterReadExt, LenientFramedRead},
 };
 
@@ -220,6 +221,12 @@ where
                             tokio::spawn(
                                 fut.map(move |()| {
                                     drop(open_token);
+                                    // Paired with the ConnectionOpen emit above:
+                                    // fires exactly once per accepted connection,
+                                    // including paths that return early from
+                                    // handle_stream (TLS handshake failure,
+                                    // shutdown during handshake).
+                                    emit!(TcpSourceConnectionClosed);
                                     drop(tcp_connection_permit);
                                 })
                                 .instrument(span.or_current()),
@@ -401,7 +408,19 @@ async fn handle_stream<T>(
                                 if let Some(ack_bytes) = acker.build_ack(ack){
                                     let stream = reader.get_mut().get_mut();
                                     if let Err(error) = stream.write_all(&ack_bytes).await {
-                                        emit!(TcpSendAckError{ error });
+                                        // Per spec, `*Error` events MUST only be
+                                        // emitted on real errors. A peer-initiated
+                                        // graceful TLS shutdown during the ack
+                                        // write is a lifecycle event, not an error
+                                        // — log at warn and skip the emit.
+                                        if is_graceful_tls_shutdown(&error) {
+                                            warn!(
+                                                message = "Connection closed by peer before acknowledgement could be sent.",
+                                                error = %error,
+                                            );
+                                        } else {
+                                            emit!(TcpSendAckError { error });
+                                        }
                                         break;
                                     }
                                 }