feat: add election logic for PgElection (#5249)

* feat: init PgElection

* fix: release advisory lock

* fix: handle duplicate keys

* chore: update comments

* fix: unlock if acquired the lock

* chore: add TODO and avoid unwrap

* refactor: check both lock and expire time, add more comments

* test: add unit test for pg election

* chore: fmt

* chore: typo

* fix: add feature gate

* chore: visibility

* chore: follow review comments

src/meta-srv/src/bootstrap.rs

@@ -48,6 +48,10 @@ use tonic::transport::server::{Router, TcpIncoming};
 
 use crate::election::etcd::EtcdElection;
 #[cfg(feature = "pg_kvbackend")]
+use crate::election::postgres::PgElection;
+#[cfg(feature = "pg_kvbackend")]
+use crate::election::CANDIDATE_LEASE_SECS;
+#[cfg(feature = "pg_kvbackend")]
 use crate::error::InvalidArgumentsSnafu;
 use crate::error::{InitExportMetricsTaskSnafu, TomlFormatSnafu};
 use crate::metasrv::builder::MetasrvBuilder;
@@ -229,7 +233,15 @@ pub async fn metasrv_builder(
             let kv_backend = PgStore::with_pg_client(pg_client)
                 .await
                 .context(error::KvBackendSnafu)?;
-            (kv_backend, None)
+            let election_client = create_postgres_client(opts).await?;
+            let election = PgElection::with_pg_client(
+                opts.server_addr.clone(),
+                election_client,
+                opts.store_key_prefix.clone(),
+                CANDIDATE_LEASE_SECS,
+            )
+            .await?;
+            (kv_backend, Some(election))
         }
     };
 

src/meta-srv/src/election.rs

@@ -19,7 +19,9 @@ pub mod postgres;
 use std::fmt::{self, Debug};
 use std::sync::Arc;
 
-use tokio::sync::broadcast::Receiver;
+use common_telemetry::{info, warn};
+use tokio::sync::broadcast::error::RecvError;
+use tokio::sync::broadcast::{self, Receiver, Sender};
 
 use crate::error::Result;
 use crate::metasrv::MetasrvNodeInfo;
@@ -75,6 +77,37 @@ impl fmt::Display for LeaderChangeMessage {
     }
 }
 
+fn listen_leader_change(leader_value: String) -> Sender<LeaderChangeMessage> {
+    let (tx, mut rx) = broadcast::channel(100);
+    let _handle = common_runtime::spawn_global(async move {
+        loop {
+            match rx.recv().await {
+                Ok(msg) => match msg {
+                    LeaderChangeMessage::Elected(key) => {
+                        info!(
+                            "[{leader_value}] is elected as leader: {:?}, lease: {}",
+                            String::from_utf8_lossy(key.name()),
+                            key.lease_id()
+                        );
+                    }
+                    LeaderChangeMessage::StepDown(key) => {
+                        warn!(
+                            "[{leader_value}] is stepping down: {:?}, lease: {}",
+                            String::from_utf8_lossy(key.name()),
+                            key.lease_id()
+                        );
+                    }
+                },
+                Err(RecvError::Lagged(_)) => {
+                    warn!("Log printing is too slow or leader changed too fast!");
+                }
+                Err(RecvError::Closed) => break,
+            }
+        }
+    });
+    tx
+}
+
 #[async_trait::async_trait]
 pub trait Election: Send + Sync {
     type Leader;

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

@@ -23,13 +23,12 @@ use etcd_client::{
 };
 use snafu::{ensure, OptionExt, ResultExt};
 use tokio::sync::broadcast;
-use tokio::sync::broadcast::error::RecvError;
 use tokio::sync::broadcast::Receiver;
 use tokio::time::{timeout, MissedTickBehavior};
 
 use crate::election::{
-    Election, LeaderChangeMessage, LeaderKey, CANDIDATES_ROOT, CANDIDATE_LEASE_SECS, ELECTION_KEY,
-    KEEP_ALIVE_INTERVAL_SECS,
+    listen_leader_change, Election, LeaderChangeMessage, LeaderKey, CANDIDATES_ROOT,
+    CANDIDATE_LEASE_SECS, ELECTION_KEY, KEEP_ALIVE_INTERVAL_SECS,
 };
 use crate::error;
 use crate::error::Result;
@@ -88,36 +87,7 @@ impl EtcdElection {
         E: AsRef<str>,
     {
         let leader_value: String = leader_value.as_ref().into();
-
-        let leader_ident = leader_value.clone();
-        let (tx, mut rx) = broadcast::channel(100);
-        let _handle = common_runtime::spawn_global(async move {
-            loop {
-                match rx.recv().await {
-                    Ok(msg) => match msg {
-                        LeaderChangeMessage::Elected(key) => {
-                            info!(
-                                "[{leader_ident}] is elected as leader: {:?}, lease: {}",
-                                String::from_utf8_lossy(key.name()),
-                                key.lease_id()
-                            );
-                        }
-                        LeaderChangeMessage::StepDown(key) => {
-                            warn!(
-                                "[{leader_ident}] is stepping down: {:?}, lease: {}",
-                                String::from_utf8_lossy(key.name()),
-                                key.lease_id()
-                            );
-                        }
-                    },
-                    Err(RecvError::Lagged(_)) => {
-                        warn!("Log printing is too slow or leader changed too fast!");
-                    }
-                    Err(RecvError::Closed) => break,
-                }
-            }
-        });
-
+        let tx = listen_leader_change(leader_value.clone());
         Ok(Arc::new(Self {
             leader_value,
             client,