Renew gRPC and REST connections after max_connection_age (#6547)

Author: guilload

config/quickwit.yaml

@@ -49,6 +49,12 @@ version: 0.8
 #     # How often cert_path/key_path are polled for changes and hot-reloaded; an
 #     # immediate reload can also be triggered with SIGHUP. Defaults to 5m.
 #     cert_poll_interval: 5m
+#     # Maximum lifetime of a connection before the server sends an HTTP/2 GOAWAY and the
+#     # client reconnects. Disabled when unset.
+#     max_connection_age: 30m
+#     # Grace period after the GOAWAY before a still-draining connection is forcefully
+#     # closed. Requires `max_connection_age` to be set.
+#     max_connection_age_grace: 30s
 #
 # Optional plaintext health-check server. Disabled unless `listen_port` is set (or the
 # `QW_HEALTH_LISTEN_PORT` environment variable). It serves only `/health/livez` and
@@ -75,6 +81,12 @@ version: 0.8
 #     # How often cert_path/key_path are polled for changes and hot-reloaded; an
 #     # immediate reload can also be triggered with SIGHUP. Defaults to 5m.
 #     cert_poll_interval: 5m
+#     # Maximum lifetime of an inbound connection before the server sends an HTTP/2 GOAWAY
+#     # and the peer reconnects. Disabled when unset.
+#     max_connection_age: 30m
+#     # Grace period after the GOAWAY before a still-draining connection is forcefully
+#     # closed. Requires `max_connection_age` to be set.
+#     max_connection_age_grace: 30s
 #
 # IP address advertised by the node, i.e. the IP address that peer nodes should use to connect to the node for RPCs.
 # The environment variable `QW_ADVERTISE_ADDRESS` can also be used to override this value.

quickwit/Cargo.lock

@@ -24,13 +24,30 @@
         "extra_headers": {
             "x-header-1": "header-value-1",
             "x-header-2": "header-value-2"
-        }
+        },
+        "tls": {
+            "cert_path": "/path/to/rest.crt",
+            "key_path": "/path/to/rest.key",
+            "ca_path": "/path/to/ca.crt",
+            "verify_client_cert": true
+        },
+        "max_connection_age": "30m",
+        "max_connection_age_grace": "30s"
     },
     "health": {
         "listen_port": 4444
     },
     "grpc": {
-        "max_message_size": "10 MB"
+        "max_message_size": "10 MB",
+        "tls": {
+            "cert_path": "/path/to/grpc.crt",
+            "key_path": "/path/to/grpc.key",
+            "ca_path": "/path/to/ca.crt",
+            "verify_client_cert": true,
+            "expected_name": "quickwit.local"
+        },
+        "max_connection_age": "1h",
+        "max_connection_age_grace": "10s"
     },
     "storage": {
         "azure": {

quickwit/quickwit-config/resources/tests/node_config/quickwit.json

@@ -15,16 +15,33 @@ default_index_root_uri = "s3://quickwit-indexes"
 
 [rest]
 listen_port = 1111
+max_connection_age = "30m"
+max_connection_age_grace = "30s"
 
 [rest.extra_headers]
 x-header-1 = "header-value-1"
 x-header-2 = "header-value-2"
 
+[rest.tls]
+cert_path = "/path/to/rest.crt"
+key_path = "/path/to/rest.key"
+ca_path = "/path/to/ca.crt"
+verify_client_cert = true
+
 [health]
 listen_port = 4444
 
 [grpc]
 max_message_size = "10 MB"
+max_connection_age = "1h"
+max_connection_age_grace = "10s"
+
+[grpc.tls]
+cert_path = "/path/to/grpc.crt"
+key_path = "/path/to/grpc.key"
+ca_path = "/path/to/ca.crt"
+verify_client_cert = true
+expected_name = "quickwit.local"
 
 [storage.azure]
 account = "quickwit-dev"

quickwit/quickwit-config/resources/tests/node_config/quickwit.toml

@@ -22,12 +22,27 @@ rest:
   extra_headers:
     x-header-1: header-value-1
     x-header-2: header-value-2
+  tls:
+    cert_path: /path/to/rest.crt
+    key_path: /path/to/rest.key
+    ca_path: /path/to/ca.crt
+    verify_client_cert: true
+  max_connection_age: 30m
+  max_connection_age_grace: 30s
 
 health:
   listen_port: 4444
 
 grpc:
   max_message_size: 10 MB
+  tls:
+    cert_path: /path/to/grpc.crt
+    key_path: /path/to/grpc.key
+    ca_path: /path/to/ca.crt
+    verify_client_cert: true
+    expected_name: quickwit.local
+  max_connection_age: 1h
+  max_connection_age_grace: 10s
 
 storage:
   azure:

quickwit/quickwit-config/resources/tests/node_config/quickwit.yaml

@@ -52,6 +52,13 @@ pub struct RestConfig {
     pub extra_headers: HeaderMap,
     #[serde(default, rename = "tls")]
     pub tls_config: Option<TlsConfig>,
+    // See `GrpcConfig::max_connection_age`. Closes long-lived keep-alive connections so an updated
+    // TLS certificate is eventually presented.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub max_connection_age: Option<HumanDuration>,
+    // See `GrpcConfig::max_connection_age_grace`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub max_connection_age_grace: Option<HumanDuration>,
 }
 
 /// Configuration for the optional plaintext health-check HTTP server.
@@ -77,6 +84,14 @@ pub struct GrpcConfig {
     // keep alive ping request.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub keep_alive: Option<KeepAliveConfig>,
+    // Maximum lifetime of an inbound connection before the server sends an HTTP/2 GOAWAY and the
+    // peer reconnects. Disabled when unset.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub max_connection_age: Option<HumanDuration>,
+    // Grace period after the GOAWAY before a still-draining connection is forcefully closed.
+    // Requires `max_connection_age` to be set. Waits indefinitely when unset.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub max_connection_age_grace: Option<HumanDuration>,
 }
 
 fn default_http2_keep_alive_interval() -> HumanDuration {
@@ -116,6 +131,10 @@ impl GrpcConfig {
         if let Some(tls_config) = &self.tls_config {
             tls_config.validate()?;
         }
+        ensure!(
+            !(self.max_connection_age_grace.is_some() && self.max_connection_age.is_none()),
+            "`grpc.max_connection_age_grace` requires `grpc.max_connection_age` to be set"
+        );
         Ok(())
     }
 }
@@ -126,6 +145,8 @@ impl Default for GrpcConfig {
             max_message_size: Self::default_max_message_size(),
             tls_config: None,
             keep_alive: None,
+            max_connection_age: None,
+            max_connection_age_grace: None,
         }
     }
 }
@@ -1065,19 +1086,37 @@ mod tests {
     fn test_grpc_config_validate() {
         let grpc_config = GrpcConfig {
             max_message_size: ByteSize::mb(1),
-            tls_config: None,
-            keep_alive: None,
+            ..Default::default()
         };
         assert!(grpc_config.validate().is_ok());
 
         let grpc_config = GrpcConfig {
             max_message_size: ByteSize::kb(1),
-            tls_config: None,
-            keep_alive: None,
+            ..Default::default()
         };
         assert!(grpc_config.validate().is_err());
     }
 
+    #[test]
+    fn test_grpc_config_validate_rejects_connection_age_grace_without_age() {
+        let grpc_config = GrpcConfig {
+            max_connection_age_grace: Some(HumanDuration::try_from("10s".to_string()).unwrap()),
+            ..Default::default()
+        };
+        let error = grpc_config.validate().unwrap_err().to_string();
+        assert!(
+            error.contains("requires `grpc.max_connection_age`"),
+            "unexpected error: {error}"
+        );
+
+        let grpc_config = GrpcConfig {
+            max_connection_age: Some(HumanDuration::try_from("1h".to_string()).unwrap()),
+            max_connection_age_grace: Some(HumanDuration::try_from("10s".to_string()).unwrap()),
+            ..Default::default()
+        };
+        assert!(grpc_config.validate().is_ok());
+    }
+
     fn tls_config(poll_interval: &str) -> TlsConfig {
         TlsConfig {
             cert_path: "/path/to/server.crt".to_string(),
@@ -1105,7 +1144,7 @@ mod tests {
         let grpc_config = GrpcConfig {
             max_message_size: ByteSize::mib(20),
             tls_config: Some(tls_config("0s")),
-            keep_alive: None,
+            ..Default::default()
         };
         assert!(grpc_config.validate().is_err());
     }

quickwit/quickwit-config/src/node_config/mod.rs

@@ -17,7 +17,7 @@ use std::net::{IpAddr, SocketAddr};
 use std::str::FromStr;
 use std::time::Duration;
 
-use anyhow::{Context, bail};
+use anyhow::{Context, bail, ensure};
 use bytesize::ByteSize;
 use http::HeaderMap;
 use quickwit_common::fs::get_disk_size;
@@ -31,6 +31,7 @@ use tracing::{info, warn};
 use super::{GrpcConfig, HealthConfig, RestConfig};
 use crate::config_value::ConfigValue;
 use crate::qw_env_vars::*;
+use crate::serde_utils::HumanDuration;
 use crate::service::QuickwitService;
 use crate::storage_config::StorageConfigs;
 use crate::templating::render_config;
@@ -457,6 +458,10 @@ struct RestConfigBuilder {
     pub extra_headers: HeaderMap,
     #[serde(default, rename = "tls")]
     pub tls_config: Option<TlsConfig>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub max_connection_age: Option<HumanDuration>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub max_connection_age_grace: Option<HumanDuration>,
 }
 
 impl RestConfigBuilder {
@@ -475,11 +480,17 @@ impl RestConfigBuilder {
         if let Some(tls_config) = &self.tls_config {
             tls_config.validate()?;
         }
+        ensure!(
+            !(self.max_connection_age_grace.is_some() && self.max_connection_age.is_none()),
+            "`rest.max_connection_age_grace` requires `rest.max_connection_age` to be set"
+        );
         let rest_config = RestConfig {
             listen_addr: SocketAddr::new(listen_ip, listen_port),
             cors_allow_origins: self.cors_allow_origins,
             extra_headers: self.extra_headers,
             tls_config: self.tls_config,
+            max_connection_age: self.max_connection_age,
+            max_connection_age_grace: self.max_connection_age_grace,
         };
         Ok(rest_config)
     }
@@ -550,6 +561,8 @@ pub fn node_config_for_tests_from_ports(
         cors_allow_origins: Vec::new(),
         extra_headers: HeaderMap::new(),
         tls_config: None,
+        max_connection_age: None,
+        max_connection_age_grace: None,
     };
     NodeConfig {
         cluster_id: default_cluster_id().unwrap(),
@@ -624,17 +637,54 @@ mod tests {
             config.rest_config.extra_headers.get("x-header-2").unwrap(),
             "header-value-2"
         );
+        let rest_tls_config = config.rest_config.tls_config.unwrap();
+        assert_eq!(
+            rest_tls_config,
+            TlsConfig {
+                cert_path: "/path/to/rest.crt".to_string(),
+                key_path: "/path/to/rest.key".to_string(),
+                ca_path: "/path/to/ca.crt".to_string(),
+                expected_name: None,
+                verify_client_cert: true,
+                cert_poll_interval: HumanDuration::try_from("5m".to_string()).unwrap(),
+            }
+        );
+        assert_eq!(
+            config.rest_config.max_connection_age,
+            Some(HumanDuration::try_from("30m".to_string()).unwrap())
+        );
+        assert_eq!(
+            config.rest_config.max_connection_age_grace,
+            Some(HumanDuration::try_from("30s".to_string()).unwrap())
+        );
         assert_eq!(config.grpc_config.max_message_size, ByteSize::mb(10));
 
+        let grpc_tls_config = config.grpc_config.tls_config.unwrap();
         assert_eq!(
-            config
-                .health_config
-                .as_ref()
-                .expect("health config should be set")
-                .listen_addr,
-            SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 4444)
+            grpc_tls_config,
+            TlsConfig {
+                cert_path: "/path/to/grpc.crt".to_string(),
+                key_path: "/path/to/grpc.key".to_string(),
+                ca_path: "/path/to/ca.crt".to_string(),
+                expected_name: Some("quickwit.local".to_string()),
+                verify_client_cert: true,
+                cert_poll_interval: HumanDuration::try_from("5m".to_string()).unwrap(),
+            }
+        );
+        assert_eq!(
+            config.grpc_config.max_connection_age,
+            Some(HumanDuration::try_from("1h".to_string()).unwrap())
+        );
+        assert_eq!(
+            config.grpc_config.max_connection_age_grace,
+            Some(HumanDuration::try_from("10s".to_string()).unwrap())
         );
 
+        let health_config = config.health_config.unwrap();
+        assert_eq!(
+            health_config.listen_addr,
+            SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 4444)
+        );
         assert_eq!(
             config.gossip_listen_addr,
             SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 2222)

quickwit/quickwit-config/src/node_config/serialize.rs

@@ -50,6 +50,7 @@ quickwit-parquet-engine = { workspace = true, optional = true }
 anyhow = { workspace = true }
 aws-sdk-sqs = { workspace = true }
 futures-util = { workspace = true }
+http-body-util = { workspace = true }
 hyper = { workspace = true }
 hyper-util = { workspace = true }
 itertools = { workspace = true }