fix(meta): add default etcd client options with keep-alive settings (#7363)

src/common/meta/src/distributed_time_constants.rs

@@ -14,6 +14,8 @@
 
 use std::time::Duration;
 
+use etcd_client::ConnectOptions;
+
 /// Heartbeat interval time (is the basic unit of various time).
 pub const HEARTBEAT_INTERVAL_MILLIS: u64 = 3000;
 
@@ -52,5 +54,16 @@ pub const HEARTBEAT_CHANNEL_KEEP_ALIVE_INTERVAL_SECS: Duration =
 pub const HEARTBEAT_CHANNEL_KEEP_ALIVE_TIMEOUT_SECS: Duration =
     Duration::from_secs(META_KEEP_ALIVE_INTERVAL_SECS + 1);
 
+/// The default options for the etcd client.
+pub fn default_etcd_client_options() -> ConnectOptions {
+    ConnectOptions::new()
+        .with_keep_alive_while_idle(true)
+        .with_keep_alive(
+            Duration::from_secs(META_KEEP_ALIVE_INTERVAL_SECS + 1),
+            Duration::from_secs(10),
+        )
+        .with_connect_timeout(Duration::from_secs(10))
+}
+
 /// The default mailbox round-trip timeout.
 pub const MAILBOX_RTT_SECS: u64 = 1;

src/meta-srv/src/bootstrap.rs

@@ -22,6 +22,7 @@ use api::v1::meta::store_server::StoreServer;
 use common_base::Plugins;
 use common_config::Configurable;
 use common_error::ext::BoxedError;
+use common_meta::distributed_time_constants::default_etcd_client_options;
 #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
 use common_meta::distributed_time_constants::META_LEASE_SECS;
 use common_meta::kv_backend::chroot::ChrootKvBackend;
@@ -441,7 +442,8 @@ pub async fn create_etcd_client(store_addrs: &[String]) -> Result<Client> {
         .map(|x| x.trim())
         .filter(|x| !x.is_empty())
         .collect::<Vec<_>>();
-    Client::connect(&etcd_endpoints, None)
+    let options = default_etcd_client_options();
+    Client::connect(&etcd_endpoints, Some(options))
         .await
         .context(error::ConnectEtcdSnafu)
 }

src/meta-srv/src/election/etcd.rs

@@ -63,22 +63,6 @@ pub struct EtcdElection {
 }
 
 impl EtcdElection {
-    pub async fn with_endpoints<E, S>(
-        leader_value: E,
-        endpoints: S,
-        store_key_prefix: String,
-    ) -> Result<ElectionRef>
-    where
-        E: AsRef<str>,
-        S: AsRef<[E]>,
-    {
-        let client = Client::connect(endpoints, None)
-            .await
-            .context(error::ConnectEtcdSnafu)?;
-
-        Self::with_etcd_client(leader_value, client, store_key_prefix).await
-    }
-
     pub async fn with_etcd_client<E>(
         leader_value: E,
         client: Client,