feat(grpc connection management): add jitter controls to grpc max age

changelog.d/grpc_connection_age_jitter.enhancement.md

@@ -0,0 +1,4 @@
+Added the `max_connection_age_jitter_factor` gRPC keepalive option to vary the maximum
+connection age for each connection and avoid synchronized reconnects.
+
+authors: hiporox

src/sources/util/grpc/mod.rs

@@ -14,6 +14,7 @@ use futures::{FutureExt, StreamExt, future::BoxFuture};
 use http::{HeaderMap, Request, Response};
 use hyper::{Body, body::HttpBody};
 use pin_project::pin_project;
+use rand::Rng;
 use tokio::{
     io::{AsyncRead, AsyncWrite, ReadBuf},
     net::TcpStream,
@@ -68,7 +69,7 @@ pub(crate) mod test_support {
 
 /// Configuration of gRPC server keepalive parameters.
 #[configurable_component]
-#[derive(Clone, Debug, Default, PartialEq, Eq)]
+#[derive(Clone, Debug, Default, PartialEq)]
 #[serde(deny_unknown_fields)]
 pub struct GrpcKeepaliveConfig {
     /// The maximum amount of time a connection may exist before the server closes it.
@@ -80,6 +81,14 @@ pub struct GrpcKeepaliveConfig {
     #[configurable(metadata(docs::human_name = "Maximum Connection Age"))]
     pub max_connection_age_secs: Option<u64>,
 
+    /// The factor by which to jitter `max_connection_age_secs` for each connection.
+    ///
+    /// A value of 0.1 gives each connection an age between 90% and 110% of the configured
+    /// maximum. The default of zero preserves the configured age exactly.
+    #[serde(default)]
+    #[configurable(validation(range(min = 0.0, max = 1.0)))]
+    pub max_connection_age_jitter_factor: f64,
+
     /// The grace period added to `max_connection_age_secs` before the server closes the connection.
     ///
     /// This setting only applies when `max_connection_age_secs` is set.
@@ -93,7 +102,11 @@ pub struct GrpcKeepaliveConfig {
 impl GrpcKeepaliveConfig {
     fn max_connection_lifetime(&self) -> Option<Duration> {
         self.max_connection_age_secs.map(|max_connection_age_secs| {
-            let age = Duration::from_secs(max_connection_age_secs);
+            let age = jittered_duration(
+                Duration::from_secs(max_connection_age_secs),
+                self.max_connection_age_jitter_factor,
+                rand::rng().random_range(-1.0..=1.0),
+            );
             let grace = self
                 .max_connection_age_grace_secs
                 .map(Duration::from_secs)
@@ -104,6 +117,14 @@ impl GrpcKeepaliveConfig {
     }
 }
 
+fn jittered_duration(duration: Duration, jitter_factor: f64, jitter: f64) -> Duration {
+    let multiplier = 1.0 + jitter_factor.clamp(0.0, 1.0) * jitter.clamp(-1.0, 1.0);
+
+    Duration::try_from_secs_f64(duration.as_secs_f64() * multiplier)
+        .unwrap_or(Duration::MAX)
+        .max(Duration::from_secs(1))
+}
+
 struct MaxConnectionAgeIo {
     inner: MaybeTlsIncomingStream<TcpStream>,
     state: MaxConnectionAgeState,
@@ -372,10 +393,9 @@ where
     let span = Span::current();
     let (tx, rx) = tokio::sync::oneshot::channel::<ShutdownSignalToken>();
     let listener = tls_settings.bind_reloadable(&address, tls_reloader).await?;
-    let max_connection_lifetime = keepalive.max_connection_lifetime();
-    let stream = listener
-        .accept_stream()
-        .map(move |stream| stream.map(|io| MaxConnectionAgeIo::new(io, max_connection_lifetime)));
+    let stream = listener.accept_stream().map(move |stream| {
+        stream.map(|io| MaxConnectionAgeIo::new(io, keepalive.max_connection_lifetime()))
+    });
 
     info!(%address, "Building gRPC server.");
 
@@ -414,10 +434,9 @@ pub async fn run_grpc_server_with_routes(
     let span = Span::current();
     let (tx, rx) = tokio::sync::oneshot::channel::<ShutdownSignalToken>();
     let listener = tls_settings.bind_reloadable(&address, tls_reloader).await?;
-    let max_connection_lifetime = keepalive.max_connection_lifetime();
-    let stream = listener
-        .accept_stream()
-        .map(move |stream| stream.map(|io| MaxConnectionAgeIo::new(io, max_connection_lifetime)));
+    let stream = listener.accept_stream().map(move |stream| {
+        stream.map(|io| MaxConnectionAgeIo::new(io, keepalive.max_connection_lifetime()))
+    });
 
     info!(%address, "Building gRPC server.");
 
@@ -499,6 +518,41 @@ mod tests {
         }
     }
 
+    #[test]
+    fn connection_age_jitter_is_bounded() {
+        let duration = Duration::from_secs(100);
+
+        assert_eq!(jittered_duration(duration, 0.0, 1.0), duration);
+        assert_eq!(
+            jittered_duration(duration, 0.2, -1.0),
+            Duration::from_secs(80)
+        );
+        assert_eq!(
+            jittered_duration(duration, 0.2, 1.0),
+            Duration::from_secs(120)
+        );
+    }
+
+    #[test]
+    fn connection_age_jitter_saturates_on_overflow() {
+        assert_eq!(
+            jittered_duration(Duration::from_secs(u64::MAX), 1.0, 1.0),
+            Duration::MAX
+        );
+    }
+
+    #[test]
+    fn connection_age_jitter_has_one_second_minimum() {
+        assert_eq!(
+            jittered_duration(Duration::from_secs(1), 1.0, -1.0),
+            Duration::from_secs(1)
+        );
+        assert_eq!(
+            jittered_duration(Duration::ZERO, 0.0, 0.0),
+            Duration::from_secs(1)
+        );
+    }
+
     #[tokio::test]
     async fn max_connection_age_service_tracks_response_body_until_drop() {
         let active_requests = Arc::new(AtomicUsize::new(0));

src/sources/vector/mod.rs

@@ -279,12 +279,14 @@ mod test {
 
                 [keepalive]
                 max_connection_age_secs = 300
+                max_connection_age_jitter_factor = 0.2
                 max_connection_age_grace_secs = 30
             "#,
         )
         .unwrap();
 
         assert_eq!(config.keepalive.max_connection_age_secs, Some(300));
+        assert_eq!(config.keepalive.max_connection_age_jitter_factor, 0.2);
         assert_eq!(config.keepalive.max_connection_age_grace_secs, Some(30));
     }
 

website/cue/reference/components/sources/generated/opentelemetry.cue

@@ -29,8 +29,9 @@ generated: components: sources: opentelemetry: configuration: {
 			examples: [{
 				address: "0.0.0.0:4317"
 				keepalive: {
-					max_connection_age_grace_secs: null
-					max_connection_age_secs:       null
+					max_connection_age_grace_secs:    null
+					max_connection_age_jitter_factor: 0.0
+					max_connection_age_secs:          null
 				}
 			}]
 			options: {
@@ -59,6 +60,16 @@ generated: components: sources: opentelemetry: configuration: {
 								unit: "seconds"
 							}
 						}
+						max_connection_age_jitter_factor: {
+							description: """
+																The factor by which to jitter `max_connection_age_secs` for each connection.
+
+																A value of 0.1 gives each connection an age between 90% and 110% of the configured
+																maximum. The default of zero preserves the configured age exactly.
+																"""
+							required: false
+							type: float: default: 0.0
+						}
 						max_connection_age_secs: {
 							description: """
 																The maximum amount of time a connection may exist before the server closes it.

website/cue/reference/components/sources/generated/vector.cue

@@ -47,6 +47,16 @@ generated: components: sources: vector: configuration: {
 					unit: "seconds"
 				}
 			}
+			max_connection_age_jitter_factor: {
+				description: """
+					The factor by which to jitter `max_connection_age_secs` for each connection.
+
+					A value of 0.1 gives each connection an age between 90% and 110% of the configured
+					maximum. The default of zero preserves the configured age exactly.
+					"""
+				required: false
+				type: float: default: 0.0
+			}
 			max_connection_age_secs: {
 				description: """
 					The maximum amount of time a connection may exist before the server closes it.