feat: implement automatic region failure detector registrations

config/config.md

@@ -323,6 +323,7 @@
 | `selector` | String | `round_robin` | Datanode selector type.<br/>- `round_robin` (default value)<br/>- `lease_based`<br/>- `load_based`<br/>For details, please see "https://docs.greptime.com/developer-guide/metasrv/selector". |
 | `use_memory_store` | Bool | `false` | Store data in memory. |
 | `enable_region_failover` | Bool | `false` | Whether to enable region failover.<br/>This feature is only available on GreptimeDB running on cluster mode and<br/>- Using Remote WAL<br/>- Using shared storage (e.g., s3). |
+| `region_failure_detector_initialization_delay` | String | `10m` | Delay before initializing region failure detectors.<br/><br/>This delay helps prevent premature initialization of region failure detectors in cases where<br/>cluster maintenance mode is enabled right after metasrv starts, especially when the cluster<br/>is not deployed via the recommended GreptimeDB Operator. Without this delay, early detector registration<br/>may trigger unnecessary region failovers during datanode startup. |
 | `allow_region_failover_on_local_wal` | Bool | `false` | Whether to allow region failover on local WAL.<br/>**This option is not recommended to be set to true, because it may lead to data loss during failover.** |
 | `node_max_idle_time` | String | `24hours` | Max allowed idle time before removing node info from metasrv memory. |
 | `enable_telemetry` | Bool | `true` | Whether to enable greptimedb telemetry. Enabled by default. |

config/metasrv.example.toml

@@ -43,6 +43,14 @@ use_memory_store = false
 ## - Using shared storage (e.g., s3).
 enable_region_failover = false
 
+## Delay before initializing region failure detectors.
+##
+## This delay helps prevent premature initialization of region failure detectors in cases where
+## cluster maintenance mode is enabled right after metasrv starts, especially when the cluster
+## is not deployed via the recommended GreptimeDB Operator. Without this delay, early detector registration
+## may trigger unnecessary region failovers during datanode startup.
+region_failure_detector_initialization_delay = '10m'
+
 ## Whether to allow region failover on local WAL.
 ## **This option is not recommended to be set to true, because it may lead to data loss during failover.**
 allow_region_failover_on_local_wal = false

src/common/meta/src/key/table_route.rs

@@ -48,6 +48,11 @@ impl TableRouteKey {
     pub fn new(table_id: TableId) -> Self {
         Self { table_id }
     }
+
+    /// Returns the range prefix of the table route key.
+    pub fn range_prefix() -> Vec<u8> {
+        format!("{}/", TABLE_ROUTE_PREFIX).into_bytes()
+    }
 }
 
 #[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]

src/meta-srv/src/error.rs

@@ -54,14 +54,6 @@ pub enum Error {
         peer_id: u64,
     },
 
-    #[snafu(display("Failed to lookup peer: {}", peer_id))]
-    LookupPeer {
-        #[snafu(implicit)]
-        location: Location,
-        source: common_meta::error::Error,
-        peer_id: u64,
-    },
-
     #[snafu(display("Another migration procedure is running for region: {}", region_id))]
     MigrationRunning {
         #[snafu(implicit)]
@@ -1033,7 +1025,6 @@ impl ErrorExt for Error {
             }
 
             Error::Other { source, .. } => source.status_code(),
-            Error::LookupPeer { source, .. } => source.status_code(),
             Error::NoEnoughAvailableNode { .. } => StatusCode::RuntimeResourcesExhausted,
 
             #[cfg(feature = "pg_kvbackend")]

src/meta-srv/src/metasrv.rs

@@ -110,6 +110,14 @@ pub struct MetasrvOptions {
     pub use_memory_store: bool,
     /// Whether to enable region failover.
     pub enable_region_failover: bool,
+    /// Delay before initializing region failure detectors.
+    ///
+    /// This delay helps prevent premature initialization of region failure detectors in cases where
+    /// cluster maintenance mode is enabled right after metasrv starts, especially when the cluster
+    /// is not deployed via the recommended GreptimeDB Operator. Without this delay, early detector registration
+    /// may trigger unnecessary region failovers during datanode startup.
+    #[serde(with = "humantime_serde")]
+    pub region_failure_detector_initialization_delay: Duration,
     /// Whether to allow region failover on local WAL.
     ///
     /// If it's true, the region failover will be allowed even if the local WAL is used.
@@ -219,6 +227,7 @@ impl Default for MetasrvOptions {
             selector: SelectorType::default(),
             use_memory_store: false,
             enable_region_failover: false,
+            region_failure_detector_initialization_delay: Duration::from_secs(10 * 60),
             allow_region_failover_on_local_wal: false,
             grpc: GrpcOptions {
                 bind_addr: format!("127.0.0.1:{}", DEFAULT_METASRV_ADDR_PORT),

src/meta-srv/src/metasrv/builder.rs

@@ -64,7 +64,7 @@ use crate::procedure::wal_prune::manager::{WalPruneManager, WalPruneTicker};
 use crate::procedure::wal_prune::Context as WalPruneContext;
 use crate::region::supervisor::{
     HeartbeatAcceptor, RegionFailureDetectorControl, RegionSupervisor, RegionSupervisorSelector,
-    RegionSupervisorTicker, DEFAULT_TICK_INTERVAL,
+    RegionSupervisorTicker, DEFAULT_INITIALIZATION_RETRY_PERIOD, DEFAULT_TICK_INTERVAL,
 };
 use crate::selector::lease_based::LeaseBasedSelector;
 use crate::selector::round_robin::RoundRobinSelector;
@@ -299,6 +299,8 @@ impl MetasrvBuilder {
                 Arc::new(RegionFailureDetectorControl::new(tx.clone())) as _,
                 Some(Arc::new(RegionSupervisorTicker::new(
                     DEFAULT_TICK_INTERVAL,
+                    options.region_failure_detector_initialization_delay,
+                    DEFAULT_INITIALIZATION_RETRY_PERIOD,
                     tx.clone(),
                 ))),
             )
@@ -341,6 +343,7 @@ impl MetasrvBuilder {
                 region_migration_manager.clone(),
                 maintenance_mode_manager.clone(),
                 peer_lookup_service.clone(),
+                leader_cached_kv_backend.clone(),
             );
 
             Some(RegionFailureHandler::new(

src/meta-srv/src/procedure/region_migration/manager.rs

@@ -23,7 +23,7 @@ use common_meta::key::table_route::TableRouteValue;
 use common_meta::peer::Peer;
 use common_meta::rpc::router::RegionRoute;
 use common_procedure::{watcher, ProcedureId, ProcedureManagerRef, ProcedureWithId};
-use common_telemetry::{error, info};
+use common_telemetry::{error, info, warn};
 use snafu::{ensure, OptionExt, ResultExt};
 use store_api::storage::RegionId;
 use table::table_name::TableName;
@@ -253,10 +253,12 @@ impl RegionMigrationManager {
     }
 
     /// Throws an error if `leader_peer` is not the `from_peer`.
+    ///
+    /// If `from_peer` is unknown, use the leader peer as the `from_peer`.
     fn verify_region_leader_peer(
         &self,
         region_route: &RegionRoute,
-        task: &RegionMigrationProcedureTask,
+        task: &mut RegionMigrationProcedureTask,
     ) -> Result<()> {
         let leader_peer = region_route
             .leader_peer
@@ -275,6 +277,15 @@ impl RegionMigrationManager {
             }
         );
 
+        if task.from_peer.addr.is_empty() {
+            warn!(
+                "The `from_peer` is unknown, use the leader peer({}) as the `from_peer`, region: {}",
+                leader_peer, task.region_id
+            );
+            // The peer id is the same as the leader peer id.
+            task.from_peer = leader_peer.clone();
+        }
+
         Ok(())
     }
 
@@ -300,7 +311,7 @@ impl RegionMigrationManager {
     /// Submits a new region migration procedure.
     pub async fn submit_procedure(
         &self,
-        task: RegionMigrationProcedureTask,
+        mut task: RegionMigrationProcedureTask,
     ) -> Result<Option<ProcedureId>> {
         let Some(guard) = self.insert_running_procedure(&task) else {
             return error::MigrationRunningSnafu {
@@ -333,20 +344,14 @@ impl RegionMigrationManager {
             .fail();
         }
 
-        self.verify_region_leader_peer(&region_route, &task)?;
+        self.verify_region_leader_peer(&region_route, &mut task)?;
         self.verify_region_follower_peers(&region_route, &task)?;
         let table_info = self.retrieve_table_info(region_id).await?;
         let TableName {
             catalog_name,
             schema_name,
             ..
         } = table_info.table_name();
-        METRIC_META_REGION_MIGRATION_DATANODES
-            .with_label_values(&["src", &task.from_peer.id.to_string()])
-            .inc();
-        METRIC_META_REGION_MIGRATION_DATANODES
-            .with_label_values(&["desc", &task.to_peer.id.to_string()])
-            .inc();
         let RegionMigrationProcedureTask {
             region_id,
             from_peer,
@@ -377,6 +382,12 @@ impl RegionMigrationManager {
                     return;
                 }
             };
+            METRIC_META_REGION_MIGRATION_DATANODES
+                .with_label_values(&["src", &task.from_peer.id.to_string()])
+                .inc();
+            METRIC_META_REGION_MIGRATION_DATANODES
+                .with_label_values(&["desc", &task.to_peer.id.to_string()])
+                .inc();
 
             if let Err(e) = watcher::wait(watcher).await {
                 error!(e; "Failed to wait region migration procedure {procedure_id} for {task}");