enhancement(sources): add a runtime-swappable TLS acceptor to TCP sources

lib/vector-core/Cargo.toml

@@ -9,6 +9,7 @@ publish = false
 workspace = true
 
 [dependencies]
+arc-swap = { workspace = true }
 async-trait.workspace = true
 bitmask-enum = { version = "2.2.5", default-features = false }
 bytes = { workspace = true, features = ["serde"] }
@@ -34,7 +35,7 @@ metrics-util.workspace = true
 mlua = { workspace = true, optional = true }
 no-proxy = { version = "0.3.6", default-features = false, features = ["serialize"] }
 ordered-float.workspace = true
-openssl = { version = "0.10.80", default-features = false, features = ["vendored"] }
+openssl = { version = "0.10.81", default-features = false, features = ["vendored"] }
 parking_lot = { version = "0.12.5", default-features = false }
 pin-project.workspace = true
 proptest = { workspace = true, optional = true }

lib/vector-core/src/tls/incoming.rs

@@ -7,6 +7,7 @@ use std::{
     task::{Context, Poll},
 };
 
+use arc_swap::ArcSwap;
 use futures::{FutureExt, Stream, future::BoxFuture, stream};
 use ipnet::IpNet;
 use openssl::{
@@ -24,7 +25,7 @@ use tonic::transport::{Certificate, server::Connected};
 
 use super::{
     CreateAcceptorSnafu, HandshakeSnafu, IncomingListenerSnafu, MaybeTlsSettings, MaybeTlsStream,
-    SslBuildSnafu, TcpBindSnafu, TlsError, TlsSettings,
+    SslBuildSnafu, TcpBindSnafu, TlsAcceptorReloader, TlsError, TlsSettings,
 };
 use crate::tcp::{self, TcpKeepaliveConfig};
 
@@ -43,43 +44,48 @@ impl TlsSettings {
 
 impl MaybeTlsSettings {
     pub async fn bind(&self, addr: &SocketAddr) -> crate::tls::Result<MaybeTlsListener> {
-        let listener = TcpListener::bind(addr).await.context(TcpBindSnafu)?;
-
-        let acceptor = match self {
-            Self::Tls(tls) => Some(tls.acceptor()?),
-            Self::Raw(()) => None,
-        };
-
-        Ok(MaybeTlsListener {
-            listener,
-            acceptor,
-            origin_filter: None,
-        })
+        self.bind_reloadable(addr, None).await
     }
 
     pub async fn bind_with_allowlist(
         &self,
         addr: &SocketAddr,
         allow_origin: Vec<IpNet>,
+    ) -> crate::tls::Result<MaybeTlsListener> {
+        Ok(self
+            .bind_reloadable(addr, None)
+            .await?
+            .with_allowlist(Some(allow_origin)))
+    }
+
+    /// Bind a listener, optionally sharing a [`TlsAcceptorReloader`] so its acceptor can be swapped
+    /// at runtime. When `reloader` is `Some` and TLS is enabled, connections accepted by the
+    /// returned listener use the acceptor the reloader currently holds; otherwise the acceptor is
+    /// fixed for the lifetime of the listener.
+    pub async fn bind_reloadable(
+        &self,
+        addr: &SocketAddr,
+        reloader: Option<TlsAcceptorReloader>,
     ) -> crate::tls::Result<MaybeTlsListener> {
         let listener = TcpListener::bind(addr).await.context(TcpBindSnafu)?;
 
-        let acceptor = match self {
-            Self::Tls(tls) => Some(tls.acceptor()?),
-            Self::Raw(()) => None,
+        let acceptor = match (self, reloader) {
+            (Self::Raw(()), _) => None,
+            (Self::Tls(_), Some(reloader)) => Some(reloader.shared()),
+            (Self::Tls(tls), None) => Some(Arc::new(ArcSwap::from_pointee(tls.acceptor()?))),
         };
 
         Ok(MaybeTlsListener {
             listener,
             acceptor,
-            origin_filter: Some(allow_origin),
+            origin_filter: None,
         })
     }
 }
 
 pub struct MaybeTlsListener {
     listener: TcpListener,
-    acceptor: Option<SslAcceptor>,
+    acceptor: Option<Arc<ArcSwap<SslAcceptor>>>,
     origin_filter: Option<Vec<IpNet>>,
 }
 
@@ -90,7 +96,11 @@ impl MaybeTlsListener {
             .accept()
             .await
             .map(|(stream, peer_addr)| {
-                MaybeTlsIncomingStream::new(stream, peer_addr, self.acceptor.clone())
+                let acceptor = self
+                    .acceptor
+                    .as_ref()
+                    .map(|accptr| accptr.load().as_ref().clone());
+                MaybeTlsIncomingStream::new(stream, peer_addr, acceptor)
             })
             .context(IncomingListenerSnafu)?;
 
@@ -421,10 +431,7 @@ impl From<X509> for CertificateMetadata {
     fn from(cert: X509) -> Self {
         let mut subject_metadata: HashMap<String, String> = HashMap::new();
         for entry in cert.subject_name().entries() {
-            let data_string = match entry.data().as_utf8() {
-                Ok(data) => data.to_string(),
-                Err(_) => String::new(),
-            };
+            let data_string = entry.data().to_string().unwrap_or_default();
             subject_metadata.insert(entry.object().to_string(), data_string);
         }
         Self {

lib/vector-core/src/tls/mod.rs

@@ -15,10 +15,12 @@ use crate::tcp::{self, TcpKeepaliveConfig};
 mod incoming;
 mod maybe_tls;
 mod outgoing;
+mod reload;
 mod settings;
 
 pub use incoming::{CertificateMetadata, MaybeTlsIncomingStream, MaybeTlsListener};
 pub use maybe_tls::MaybeTls;
+pub use reload::{TlsAcceptorReloader, WeakTlsAcceptorReloader};
 pub use settings::{
     MaybeTlsSettings, PEM_START_MARKER, TEST_PEM_CA_PATH, TEST_PEM_CLIENT_CRT_PATH,
     TEST_PEM_CLIENT_KEY_PATH, TEST_PEM_CRT_PATH, TEST_PEM_INTERMEDIATE_CA_PATH, TEST_PEM_KEY_PATH,