feat(sources): add gRPC max connection age (#25660)

Co-authored-by: Pavlos Rontidis <pavlos.rontidis@gmail.com>

Author: fpytloun

changelog.d/25660_grpc_source_max_connection_age.feature.md

@@ -0,0 +1,5 @@
+Added support for configuring gRPC source maximum connection age, allowing
+long-lived client connections to be gracefully recycled for better load balancer
+distribution.
+
+authors: fpytloun

src/components/validation/runner/io.rs

@@ -22,7 +22,7 @@ use crate::{
         Client as VectorClient, HealthCheckRequest, HealthCheckResponse, PushEventsRequest,
         PushEventsResponse, Server as VectorServer, Service as VectorService, ServingStatus,
     },
-    sources::util::grpc::run_grpc_server,
+    sources::util::grpc::{GrpcKeepaliveConfig, run_grpc_server},
 };
 
 #[derive(Clone)]
@@ -166,6 +166,7 @@ pub fn spawn_grpc_server<S>(
             listen_addr.as_socket_addr(),
             tls_settings,
             service,
+            GrpcKeepaliveConfig::default(),
             shutdown_signal,
         );
         pin!(server);

src/sources/opentelemetry/config.rs

@@ -14,7 +14,7 @@ use crate::{
             grpc::Service,
             http::{build_warp_filter, run_http_server},
         },
-        util::grpc::run_grpc_server_with_routes,
+        util::grpc::{GrpcKeepaliveConfig, run_grpc_server_with_routes},
     },
 };
 use futures::FutureExt;
@@ -173,12 +173,17 @@ pub struct GrpcConfig {
     #[configurable(derived)]
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub tls: Option<TlsEnableableConfig>,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub keepalive: GrpcKeepaliveConfig,
 }
 
 fn example_grpc_config() -> GrpcConfig {
     GrpcConfig {
         address: "0.0.0.0:4317".parse().unwrap(),
         tls: None,
+        keepalive: GrpcKeepaliveConfig::default(),
     }
 }
 
@@ -325,6 +330,7 @@ impl SourceConfig for OpentelemetryConfig {
             self.grpc.address,
             grpc_tls_settings,
             builder.routes(),
+            self.grpc.keepalive.clone(),
             cx.shutdown.clone(),
         )
         .map_err(|error| {

src/sources/opentelemetry/integration_tests.rs

@@ -53,6 +53,7 @@ async fn receive_logs_legacy_namespace() {
             grpc: GrpcConfig {
                 address: source_grpc_address().parse().unwrap(),
                 tls: Default::default(),
+                keepalive: Default::default(),
             },
             http: HttpConfig {
                 address: source_http_address().parse().unwrap(),
@@ -152,6 +153,7 @@ async fn receive_trace() {
             grpc: GrpcConfig {
                 address: source_grpc_address().parse().unwrap(),
                 tls: Default::default(),
+                keepalive: Default::default(),
             },
             http: HttpConfig {
                 address: source_http_address().parse().unwrap(),
@@ -257,6 +259,7 @@ async fn receive_metric() {
             grpc: GrpcConfig {
                 address: source_grpc_address().parse().unwrap(),
                 tls: Default::default(),
+                keepalive: Default::default(),
             },
             http: HttpConfig {
                 address: source_http_address().parse().unwrap(),

src/sources/opentelemetry/tests.rs

@@ -206,6 +206,30 @@ fn generate_config() {
     test_util::test_generate_config::<OpentelemetryConfig>();
 }
 
+#[test]
+fn config_grpc_keepalive() {
+    let config: OpentelemetryConfig = toml::from_str(
+        r#"
+            [grpc]
+            address = "0.0.0.0:4317"
+
+            [grpc.keepalive]
+            max_connection_age_secs = 300
+            max_connection_age_grace_secs = 30
+
+            [http]
+            address = "0.0.0.0:4318"
+        "#,
+    )
+    .unwrap();
+
+    assert_eq!(config.grpc.keepalive.max_connection_age_secs, Some(300));
+    assert_eq!(
+        config.grpc.keepalive.max_connection_age_grace_secs,
+        Some(30)
+    );
+}
+
 #[tokio::test]
 async fn receive_grpc_logs_vector_namespace() {
     assert_source_compliance(&SOURCE_TAGS, async {
@@ -1175,6 +1199,7 @@ fn get_source_config_with_headers(
         grpc: GrpcConfig {
             address: grpc_addr,
             tls: Default::default(),
+            keepalive: Default::default(),
         },
         http: HttpConfig {
             address: http_addr,
@@ -1510,6 +1535,7 @@ pub async fn build_otlp_test_env(
         grpc: GrpcConfig {
             address: grpc_addr,
             tls: Default::default(),
+            keepalive: Default::default(),
         },
         http: HttpConfig {
             address: http_addr,
@@ -1589,6 +1615,7 @@ async fn http_logs_use_otlp_decoding_emits_metric() {
         grpc: GrpcConfig {
             address: grpc_addr,
             tls: Default::default(),
+            keepalive: Default::default(),
         },
         http: HttpConfig {
             address: http_addr,
@@ -1816,6 +1843,7 @@ mod otlp_decoding_config_tests {
             grpc: GrpcConfig {
                 address: "0.0.0.0:4317".parse().unwrap(),
                 tls: None,
+                keepalive: Default::default(),
             },
             http: HttpConfig {
                 address: "0.0.0.0:4318".parse().unwrap(),
@@ -1856,6 +1884,7 @@ mod otlp_decoding_config_tests {
             grpc: GrpcConfig {
                 address: "0.0.0.0:4317".parse().unwrap(),
                 tls: None,
+                keepalive: Default::default(),
             },
             http: HttpConfig {
                 address: "0.0.0.0:4318".parse().unwrap(),
@@ -1899,6 +1928,7 @@ mod otlp_decoding_config_tests {
             grpc: GrpcConfig {
                 address: "0.0.0.0:4317".parse().unwrap(),
                 tls: None,
+                keepalive: Default::default(),
             },
             http: HttpConfig {
                 address: "0.0.0.0:4318".parse().unwrap(),