feat: Failover follower to leader

src/common/src/util/addr.rs

@@ -16,9 +16,16 @@ use std::net::SocketAddr;
 use std::str::FromStr;
 
 use risingwave_pb::common::HostAddress as ProstHostAddress;
+use risingwave_pb::meta::MetaLeaderInfo;
 
 use crate::error::{internal_error, Result};
 
+pub fn leader_info_to_host_addr(mli: MetaLeaderInfo) -> HostAddr {
+    mli.node_address
+        .parse::<HostAddr>()
+        .expect("invalid leader addr")
+}
+
 /// General host address and port.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct HostAddr {
@@ -31,7 +38,6 @@ impl std::fmt::Display for HostAddr {
         write!(f, "{}:{}", self.host, self.port)
     }
 }
-
 impl From<SocketAddr> for HostAddr {
     fn from(addr: SocketAddr) -> Self {
         HostAddr {

src/meta/src/hummock/manager/mod.rs

@@ -503,6 +503,7 @@ where
                 return Err(Error::InvalidContext(context_id));
             }
         }
+
         trx.check_equal(
             META_CF_NAME.to_owned(),
             META_LEADER_KEY.as_bytes().to_vec(),

src/meta/src/rpc/elections.rs

@@ -21,7 +21,6 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use prost::Message;
 use rand::rngs::StdRng;
 use rand::{Rng, SeedableRng};
-use risingwave_common::util::addr::HostAddr;
 use risingwave_pb::meta::{MetaLeaderInfo, MetaLeaseInfo};
 use tokio::sync::oneshot::Sender;
 use tokio::sync::watch::Receiver;
@@ -48,26 +47,17 @@ struct ElectionResult {
     pub is_leader: bool,
 }
 
-impl ElectionResult {
-    pub fn get_leader_addr(&self) -> HostAddr {
-        self.meta_leader_info
-            .node_address
-            .parse::<HostAddr>()
-            .expect("invalid leader addr")
-    }
-}
-
 /// Runs for election in an attempt to become leader
 ///
-/// ## Returns
-/// Returns `ElectionResult`, containing infos about the node who won the election or
-/// `MetaError` if the election ran into an error
-///
 /// ## Arguments
 /// `meta_store`: The meta store which holds the lease, deciding about the election result
 /// `addr`: Address of the node that runs for election
 /// `lease_time_sec`: Amount of seconds that this lease will be valid
 /// `next_lease_id`: If the node wins, the lease used until the next election will have this id
+///
+/// ## Returns
+/// Returns `ElectionResult`, containing infos about the node who won the election or
+/// `MetaError` if the election ran into an error
 async fn campaign<S: MetaStore>(
     meta_store: &Arc<S>,
     addr: &String,
@@ -175,15 +165,16 @@ async fn campaign<S: MetaStore>(
 
 /// Try to renew/acquire the leader lease
 ///
-/// ## Returns
-/// True if node was leader and was able to renew/acquire the lease.
-/// False if node was follower and thus could not renew/acquire lease.
-/// `MetaError` if operation ran into an error
 ///
 /// ## Arguments
 /// `leader_info`: Info of the node that trie
 /// `lease_time_sec`: Time in seconds that the lease is valid
-/// `meta_store`: Store which holds the lease
+/// `meta_store`: Store which holds the lease#
+///
+/// ## Returns
+/// True if node was leader and was able to renew/acquire the lease.
+/// False if node was follower and thus could not renew/acquire lease.
+/// `MetaError` if operation ran into an error
 async fn renew_lease<S: MetaStore>(
     leader_info: &MetaLeaderInfo,
     lease_time_sec: u64,
@@ -220,7 +211,7 @@ async fn renew_lease<S: MetaStore>(
 
 /// Retrieve infos about the current leader
 ///
-/// ## Attributes:
+/// ## Arguments:
 /// `meta_store`: The store holding information about the leader
 ///
 /// ## Returns
@@ -299,7 +290,7 @@ pub async fn run_elections<S: MetaStore>(
     MetaLeaderInfo,
     JoinHandle<()>,
     Sender<()>,
-    Receiver<(HostAddr, bool)>,
+    Receiver<(MetaLeaderInfo, bool)>,
 )> {
     // Randomize interval to reduce mitigate likelihood of simultaneous requests
     let mut rng: StdRng = SeedableRng::from_entropy();
@@ -317,14 +308,10 @@ pub async fn run_elections<S: MetaStore>(
             gen_rand_lease_id(addr.as_str()),
         )
         .await;
-        let (leader_addr, initial_leader, is_initial_leader) = match election_result {
+        let (initial_leader, is_initial_leader) = match election_result {
             Ok(elect_result) => {
                 tracing::info!("initial election finished");
-                (
-                    elect_result.get_leader_addr(),
-                    elect_result.meta_leader_info,
-                    elect_result.is_leader,
-                )
+                (elect_result.meta_leader_info, elect_result.is_leader)
             }
             Err(_) => {
                 tracing::info!("initial election failed. Repeating election");
@@ -344,7 +331,8 @@ pub async fn run_elections<S: MetaStore>(
 
         // define all follow up elections and terms in handle
         let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel();
-        let (leader_tx, leader_rx) = tokio::sync::watch::channel((leader_addr, is_initial_leader));
+        let (leader_tx, leader_rx) =
+            tokio::sync::watch::channel((initial_leader.clone(), is_initial_leader));
         let handle = tokio::spawn(async move {
             // runs all followup elections
             let mut ticker = tokio::time::interval(
@@ -355,12 +343,10 @@ pub async fn run_elections<S: MetaStore>(
 
             let mut is_leader = is_initial_leader;
             let mut leader_info = initial_leader.clone();
-            let n_addr = initial_leader.node_address.as_str();
-            let mut leader_addr = n_addr.parse::<HostAddr>().unwrap();
             'election: loop {
                 // Do not elect new leader directly after running the initial election
                 if !initial_election {
-                    let (leader_addr_, leader_info_, is_leader_) = match campaign(
+                    let (leader_info_, is_leader_) = match campaign(
                         &meta_store,
                         &addr,
                         lease_time_sec,
@@ -375,11 +361,7 @@ pub async fn run_elections<S: MetaStore>(
                         }
                         Ok(elect_result) => {
                             tracing::info!("election finished");
-                            (
-                                elect_result.get_leader_addr(),
-                                elect_result.meta_leader_info,
-                                elect_result.is_leader,
-                            )
+                            (elect_result.meta_leader_info, elect_result.is_leader)
                         }
                     };
 
@@ -392,12 +374,11 @@ pub async fn run_elections<S: MetaStore>(
                     }
                     leader_info = leader_info_;
                     is_leader = is_leader_;
-                    leader_addr = leader_addr_;
                 }
                 initial_election = false;
 
                 // signal to observers if there is a change in leadership
-                leader_tx.send((leader_addr.clone(), is_leader)).unwrap();
+                leader_tx.send((leader_info.clone(), is_leader)).unwrap();
 
                 // election done. Enter the term of the current leader
                 // Leader stays in power until leader crashes

src/meta/src/rpc/follower_svc.rs

@@ -0,0 +1,57 @@
+// Copyright 2022 Singularity Data
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use risingwave_pb::health::health_server::HealthServer;
+use tokio::sync::oneshot::Receiver as OneReceiver;
+use tokio::sync::watch::Receiver as WatchReceiver;
+
+use super::intercept::MetricsMiddlewareLayer;
+use super::server::AddressInfo;
+use super::service::health_service::HealthServiceImpl;
+use crate::rpc::metrics::MetaMetrics;
+
+/// Starts all services needed for the meta follower node
+pub async fn start_follower_srv(
+    mut svc_shutdown_rx: WatchReceiver<()>,
+    follower_shutdown_rx: OneReceiver<()>,
+    address_info: AddressInfo,
+) {
+    let health_srv = HealthServiceImpl::new();
+    tonic::transport::Server::builder()
+        .layer(MetricsMiddlewareLayer::new(Arc::new(MetaMetrics::new())))
+        .add_service(HealthServer::new(health_srv))
+        .serve_with_shutdown(address_info.listen_addr, async move {
+            tokio::select! {
+                _ = tokio::signal::ctrl_c() => {},
+                // shutdown service if all services should be shut down
+                res = svc_shutdown_rx.changed() =>  {
+                    match res {
+                        Ok(_) => tracing::info!("Shutting down services"),
+                        Err(_) => tracing::error!("Service shutdown sender dropped")
+                    }
+                },
+                // shutdown service if follower becomes leader
+                res = follower_shutdown_rx =>  {
+                    match res {
+                        Ok(_) => tracing::info!("Shutting down follower services"),
+                        Err(_) => tracing::error!("Follower service shutdown sender dropped")
+                    }
+                },
+            }
+        })
+        .await
+        .unwrap();
+}