Adding keep alive setting. (#5769)

By default, the behavior is unchanged.

Co-authored-by: fulmicoton <[email protected]>

Author: fulmicoton

quickwit/quickwit-cli/src/tool.rs

@@ -26,7 +26,7 @@ use colored::{ColoredString, Colorize};
 use humantime::format_duration;
 use quickwit_actors::{ActorExitStatus, ActorHandle, Mailbox, Universe};
 use quickwit_cluster::{
-    ChannelTransport, Cluster, ClusterMember, FailureDetectorConfig, make_client_tls_config,
+    ChannelTransport, Cluster, ClusterMember, FailureDetectorConfig, make_client_grpc_config,
 };
 use quickwit_common::pubsub::EventBroker;
 use quickwit_common::runtimes::RuntimesConfig;
@@ -939,6 +939,7 @@ async fn create_empty_cluster(config: &NodeConfig) -> anyhow::Result<Cluster> {
         indexing_cpu_capacity: CpuCapacity::zero(),
         indexing_tasks: Vec::new(),
     };
+    let client_grpc_config = make_client_grpc_config(&config.grpc_config)?;
     let cluster = Cluster::join(
         config.cluster_id.clone(),
         self_node,
@@ -947,12 +948,7 @@ async fn create_empty_cluster(config: &NodeConfig) -> anyhow::Result<Cluster> {
         config.gossip_interval,
         FailureDetectorConfig::default(),
         &ChannelTransport::default(),
-        config
-            .grpc_config
-            .tls
-            .as_ref()
-            .map(make_client_tls_config)
-            .transpose()?,
+        client_grpc_config,
     )
     .await?;
 

quickwit/quickwit-cluster/src/change.rs

@@ -21,11 +21,11 @@ use chitchat::{ChitchatId, NodeState};
 use futures::Stream;
 use pin_project::pin_project;
 use quickwit_common::sorted_iter::{KeyDiff, SortedByKeyIterator};
-use quickwit_common::tower::{make_channel, warmup_channel};
+use quickwit_common::tower::{ClientGrpcConfig, make_channel, warmup_channel};
 use quickwit_proto::types::NodeId;
 use tokio::sync::mpsc;
 use tokio_stream::wrappers::UnboundedReceiverStream;
-use tonic::transport::{Channel, ClientTlsConfig};
+use tonic::transport::Channel;
 use tracing::{info, warn};
 
 use crate::ClusterNode;
@@ -74,7 +74,7 @@ pub(crate) async fn compute_cluster_change_events(
     previous_nodes: &mut BTreeMap<NodeId, ClusterNode>,
     previous_node_states: &BTreeMap<ChitchatId, NodeState>,
     new_node_states: &BTreeMap<ChitchatId, NodeState>,
-    tls_config: Option<&ClientTlsConfig>,
+    client_grpc_config: &ClientGrpcConfig,
 ) -> Vec<ClusterChange> {
     let mut cluster_events = Vec::new();
 
@@ -91,7 +91,7 @@ pub(crate) async fn compute_cluster_change_events(
                     chitchat_id,
                     node_state,
                     previous_nodes,
-                    tls_config,
+                    client_grpc_config.clone(),
                 )
                 .await;
 
@@ -136,7 +136,7 @@ async fn compute_cluster_change_events_on_added(
     new_chitchat_id: &ChitchatId,
     new_node_state: &NodeState,
     previous_nodes: &mut BTreeMap<NodeId, ClusterNode>,
-    tls_config: Option<&ClientTlsConfig>,
+    client_grpc_config: ClientGrpcConfig,
 ) -> Vec<ClusterChange> {
     let is_self_node = self_chitchat_id == new_chitchat_id;
     let new_node_id: NodeId = new_chitchat_id.node_id.clone().into();
@@ -169,7 +169,7 @@ async fn compute_cluster_change_events_on_added(
         new_chitchat_id,
         new_node_state,
         is_self_node,
-        tls_config,
+        &client_grpc_config,
     )
     .await
     else {
@@ -304,11 +304,11 @@ async fn try_new_node(
     chitchat_id: &ChitchatId,
     node_state: &NodeState,
     is_self_node: bool,
-    tls_config: Option<&ClientTlsConfig>,
+    grpc_config: &ClientGrpcConfig,
 ) -> Option<ClusterNode> {
     match node_state.grpc_advertise_addr() {
         Ok(socket_addr) => {
-            let channel = make_channel(socket_addr, tls_config.cloned()).await;
+            let channel = make_channel(socket_addr, grpc_config.clone()).await;
             try_new_node_with_channel(cluster_id, chitchat_id, node_state, channel, is_self_node)
         }
         Err(error) => {
@@ -448,7 +448,7 @@ pub(crate) mod tests {
                 &new_chitchat_id,
                 &new_node_state,
                 &mut previous_nodes,
-                None,
+                Default::default(),
             )
             .await;
             assert!(events.is_empty());
@@ -471,7 +471,7 @@ pub(crate) mod tests {
                 &new_chitchat_id,
                 &new_node_state,
                 &mut previous_nodes,
-                None,
+                Default::default(),
             )
             .await;
             assert!(events.is_empty());
@@ -500,7 +500,7 @@ pub(crate) mod tests {
                 &new_chitchat_id,
                 &new_node_state,
                 &mut previous_nodes,
-                None,
+                Default::default(),
             )
             .await;
 
@@ -523,7 +523,7 @@ pub(crate) mod tests {
                 &rejoined_chitchat_id,
                 &new_node_state,
                 &mut previous_nodes,
-                None,
+                Default::default(),
             )
             .await;
             assert_eq!(events.len(), 2);
@@ -552,7 +552,7 @@ pub(crate) mod tests {
                 &new_chitchat_id,
                 &new_node_state,
                 &mut previous_nodes,
-                None,
+                Default::default(),
             )
             .await;
             assert!(events.is_empty());
@@ -577,7 +577,7 @@ pub(crate) mod tests {
                 &new_chitchat_id,
                 &new_node_state,
                 &mut previous_nodes,
-                None,
+                Default::default(),
             )
             .await;
             assert_eq!(events.len(), 1);
@@ -908,7 +908,7 @@ pub(crate) mod tests {
                 &mut previous_nodes,
                 &previous_node_states,
                 &new_node_states,
-                None,
+                &Default::default(),
             )
             .await;
             assert!(events.is_empty());
@@ -938,7 +938,7 @@ pub(crate) mod tests {
                 &mut previous_nodes,
                 &previous_node_states,
                 &new_node_states,
-                None,
+                &Default::default(),
             )
             .await;
             assert!(events.is_empty());
@@ -956,7 +956,7 @@ pub(crate) mod tests {
                 &mut previous_nodes,
                 &previous_node_states,
                 &new_node_states,
-                None,
+                &Default::default(),
             )
             .await;
             assert_eq!(events.len(), 1);
@@ -971,7 +971,7 @@ pub(crate) mod tests {
                 &mut previous_nodes,
                 &new_node_states,
                 &new_node_states,
-                None,
+                &Default::default(),
             )
             .await;
             assert_eq!(events.len(), 0);
@@ -1004,7 +1004,7 @@ pub(crate) mod tests {
                 &mut previous_nodes,
                 &previous_node_states,
                 &new_node_states,
-                None,
+                &Default::default(),
             )
             .await;
             assert_eq!(events.len(), 1);
@@ -1024,7 +1024,7 @@ pub(crate) mod tests {
                 &mut previous_nodes,
                 &previous_node_states,
                 &new_node_states,
-                None,
+                &Default::default(),
             )
             .await;
             assert_eq!(events.len(), 1);

quickwit/quickwit-cluster/src/cluster.rs

@@ -26,14 +26,14 @@ use chitchat::{
     FailureDetectorConfig, KeyChangeEvent, ListenerHandle, NodeState, spawn_chitchat,
 };
 use itertools::Itertools;
+use quickwit_common::tower::ClientGrpcConfig;
 use quickwit_proto::indexing::{IndexingPipelineId, IndexingTask, PipelineMetrics};
 use quickwit_proto::types::{NodeId, NodeIdRef, PipelineUid, ShardId};
 use serde::{Deserialize, Serialize};
 use tokio::sync::{Mutex, RwLock, mpsc, watch};
 use tokio::time::timeout;
 use tokio_stream::StreamExt;
 use tokio_stream::wrappers::WatchStream;
-use tonic::transport::ClientTlsConfig;
 use tracing::{info, warn};
 
 use crate::change::{ClusterChange, ClusterChangeStreamFactory, compute_cluster_change_events};
@@ -62,9 +62,10 @@ pub struct Cluster {
     self_chitchat_id: ChitchatId,
     /// Socket address (UDP) the node listens on for receiving gossip messages.
     pub gossip_listen_addr: SocketAddr,
-    // TODO this should become an ArcSwap<ClienTlsConfig> or something so that
-    // some task can watch for new certificates and update this (hot reloading)
-    tls_config: Option<ClientTlsConfig>,
+    // TODO this object contains a tls config. We might want to change it to a
+    // ArcSwap<ClientGrpcConfig> or something so that some task can watch for new certificates
+    // and update this (hot reloading)
+    client_grpc_config: ClientGrpcConfig,
     gossip_interval: Duration,
     inner: Arc<RwLock<InnerCluster>>,
 }
@@ -115,7 +116,7 @@ impl Cluster {
         gossip_interval: Duration,
         failure_detector_config: FailureDetectorConfig,
         transport: &dyn Transport,
-        tls_config: Option<ClientTlsConfig>,
+        client_grpc_config: ClientGrpcConfig,
     ) -> anyhow::Result<Self> {
         info!(
             cluster_id=%cluster_id,
@@ -186,7 +187,7 @@ impl Cluster {
             weak_chitchat,
             live_nodes_rx,
             catchup_callback_rx.clone(),
-            tls_config.clone(),
+            client_grpc_config.clone(),
         )
         .await;
 
@@ -204,7 +205,7 @@ impl Cluster {
             gossip_listen_addr,
             gossip_interval,
             inner: Arc::new(RwLock::new(inner)),
-            tls_config,
+            client_grpc_config,
         };
         spawn_change_stream_task(cluster.clone()).await;
         Ok(cluster)
@@ -560,7 +561,7 @@ fn chitchat_kv_to_indexing_task(key: &str, value: &str) -> Option<IndexingTask>
 async fn spawn_change_stream_task(cluster: Cluster) {
     let cluster_guard = cluster.inner.read().await;
     let cluster_id = cluster_guard.cluster_id.clone();
-    let tls_config = cluster.tls_config.clone();
+    let client_grpc_config = cluster.client_grpc_config.clone();
     let self_chitchat_id = cluster_guard.self_chitchat_id.clone();
     let chitchat = cluster_guard.chitchat_handle.chitchat();
     let weak_cluster = Arc::downgrade(&cluster.inner);
@@ -584,7 +585,7 @@ async fn spawn_change_stream_task(cluster: Cluster) {
                 previous_live_nodes,
                 &previous_live_node_states,
                 &new_live_node_states,
-                tls_config.as_ref(),
+                &client_grpc_config,
             )
             .await;
             if !events.is_empty() {
@@ -701,7 +702,7 @@ pub async fn create_cluster_for_test_with_id(
         Duration::from_millis(25),
         failure_detector_config,
         transport,
-        None,
+        Default::default(),
     )
     .await?;
     cluster.set_self_node_readiness(self_node_readiness).await;

quickwit/quickwit-cluster/src/grpc_gossip.rs

@@ -21,12 +21,12 @@ use std::time::{Duration, Instant};
 use chitchat::{Chitchat, ChitchatId, NodeState, VersionedValue};
 use futures::Future;
 use quickwit_common::pretty::PrettyDisplay;
+use quickwit_common::tower::ClientGrpcConfig;
 use quickwit_proto::cluster::{ClusterService, ClusterServiceClient, FetchClusterStateRequest};
 use rand::seq::IteratorRandom;
 use tokio::sync::{Mutex, watch};
 use tokio_stream::StreamExt;
 use tokio_stream::wrappers::WatchStream;
-use tonic::transport::ClientTlsConfig;
 use tracing::{info, warn};
 
 use crate::grpc_service::cluster_grpc_client;
@@ -42,7 +42,7 @@ pub(crate) async fn spawn_catchup_callback_task(
     weak_chitchat: Weak<Mutex<Chitchat>>,
     live_nodes_rx: watch::Receiver<BTreeMap<ChitchatId, NodeState>>,
     mut catchup_callback_rx: watch::Receiver<()>,
-    tls_config: Option<ClientTlsConfig>,
+    client_grpc_config: ClientGrpcConfig,
 ) {
     let catchup_callback_future = async move {
         let mut interval = tokio::time::interval(Duration::from_secs(60));
@@ -57,8 +57,7 @@ pub(crate) async fn spawn_catchup_callback_task(
                 &self_chitchat_id,
                 chitchat,
                 live_nodes_rx.clone(),
-                cluster_grpc_client,
-                tls_config.clone(),
+                |socket_addr| cluster_grpc_client(socket_addr, client_grpc_config.clone()),
             )
             .await;
 
@@ -72,15 +71,14 @@ pub(crate) async fn spawn_catchup_callback_task(
     tokio::spawn(catchup_callback_future);
 }
 
-async fn perform_grpc_gossip_rounds<Factory, Fut>(
+async fn perform_grpc_gossip_rounds<ClusterServiceClientFactory, Fut>(
     cluster_id: String,
     self_chitchat_id: &ChitchatId,
     chitchat: Arc<Mutex<Chitchat>>,
     live_nodes_rx: watch::Receiver<BTreeMap<ChitchatId, NodeState>>,
-    grpc_client_factory: Factory,
-    tls_client_config: Option<ClientTlsConfig>,
+    grpc_client_factory: ClusterServiceClientFactory,
 ) where
-    Factory: Fn(SocketAddr, Option<ClientTlsConfig>) -> Fut,
+    ClusterServiceClientFactory: Fn(SocketAddr) -> Fut,
     Fut: Future<Output = ClusterServiceClient>,
 {
     wait_for_gossip_candidates(
@@ -101,8 +99,7 @@ async fn perform_grpc_gossip_rounds<Factory, Fut>(
     info!("pulling cluster state from node(s): {node_ids:?}");
 
     for (node_id, grpc_advertise_addr) in zip(node_ids, grpc_advertise_addrs) {
-        let cluster_client =
-            grpc_client_factory(grpc_advertise_addr, tls_client_config.clone()).await;
+        let cluster_client = grpc_client_factory(grpc_advertise_addr).await;
 
         let request = FetchClusterStateRequest {
             cluster_id: cluster_id.clone(),
@@ -272,7 +269,7 @@ mod tests {
         let self_chitchat_id = cluster.self_chitchat_id();
         let chitchat = cluster.chitchat().await;
 
-        let grpc_client_factory = |_: SocketAddr, _: Option<_>| {
+        let grpc_client_factory = |_: SocketAddr| {
             Box::pin(async {
                 let mut mock_cluster_service = MockClusterService::new();
                 mock_cluster_service
@@ -336,7 +333,6 @@ mod tests {
             chitchat.clone(),
             live_nodes_rx,
             grpc_client_factory,
-            None,
         )
         .await;
 

quickwit/quickwit-cluster/src/grpc_service.rs

@@ -17,15 +17,14 @@ use std::net::SocketAddr;
 use bytesize::ByteSize;
 use itertools::Itertools;
 use once_cell::sync::Lazy;
-use quickwit_common::tower::{GrpcMetricsLayer, make_channel};
+use quickwit_common::tower::{ClientGrpcConfig, GrpcMetricsLayer, make_channel};
 use quickwit_proto::cluster::cluster_service_grpc_server::ClusterServiceGrpcServer;
 use quickwit_proto::cluster::{
     ChitchatId as ProtoChitchatId, ClusterError, ClusterResult, ClusterService,
     ClusterServiceClient, ClusterServiceGrpcServerAdapter, FetchClusterStateRequest,
     FetchClusterStateResponse, NodeState as ProtoNodeState, VersionedKeyValue,
 };
 use tonic::async_trait;
-use tonic::transport::ClientTlsConfig;
 
 use crate::Cluster;
 
@@ -38,9 +37,9 @@ static CLUSTER_GRPC_SERVER_METRICS_LAYER: Lazy<GrpcMetricsLayer> =
 
 pub(crate) async fn cluster_grpc_client(
     socket_addr: SocketAddr,
-    tls_config: Option<ClientTlsConfig>,
+    client_grpc_config: ClientGrpcConfig,
 ) -> ClusterServiceClient {
-    let channel = make_channel(socket_addr, tls_config).await;
+    let channel = make_channel(socket_addr, client_grpc_config).await;
 
     ClusterServiceClient::tower()
         .stack_layer(CLUSTER_GRPC_CLIENT_METRICS_LAYER.clone())