feat: add region migration endpoint

src/meta-srv/src/metasrv.rs

@@ -48,6 +48,7 @@ use crate::error::{
 use crate::failure_detector::PhiAccrualFailureDetectorOptions;
 use crate::handler::HeartbeatHandlerGroup;
 use crate::lock::DistLockRef;
+use crate::procedure::region_migration::manager::RegionMigrationManagerRef;
 use crate::pubsub::{PublishRef, SubscribeManagerRef};
 use crate::selector::{Selector, SelectorType};
 use crate::service::mailbox::MailboxRef;
@@ -249,6 +250,7 @@ pub struct MetaSrv {
     table_metadata_manager: TableMetadataManagerRef,
     memory_region_keeper: MemoryRegionKeeperRef,
     greptimedb_telemetry_task: Arc<GreptimeDBTelemetryTask>,
+    region_migration_manager: RegionMigrationManagerRef,
 
     plugins: Plugins,
 }
@@ -411,6 +413,10 @@ impl MetaSrv {
         &self.memory_region_keeper
     }
 
+    pub fn region_migration_manager(&self) -> &RegionMigrationManagerRef {
+        &self.region_migration_manager
+    }
+
     pub fn publish(&self) -> Option<PublishRef> {
         self.plugins.get::<PublishRef>()
     }

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

@@ -57,6 +57,8 @@ use crate::metasrv::{
     ElectionRef, MetaSrv, MetaSrvOptions, MetasrvInfo, SelectorContext, SelectorRef, TABLE_ID_SEQ,
 };
 use crate::procedure::region_failover::RegionFailoverManager;
+use crate::procedure::region_migration::manager::RegionMigrationManager;
+use crate::procedure::region_migration::ContextFactoryImpl;
 use crate::pubsub::PublishRef;
 use crate::selector::lease_based::LeaseBasedSelector;
 use crate::service::mailbox::MailboxRef;
@@ -236,6 +238,17 @@ impl MetaSrvBuilder {
             &opening_region_keeper,
         )?;
 
+        let region_migration_manager = Arc::new(RegionMigrationManager::new(
+            procedure_manager.clone(),
+            ContextFactoryImpl::new(
+                table_metadata_manager.clone(),
+                opening_region_keeper.clone(),
+                mailbox.clone(),
+                options.server_addr.clone(),
+            ),
+        ));
+        region_migration_manager.try_start()?;
+
         let handler_group = match handler_group {
             Some(handler_group) => handler_group,
             None => {
@@ -323,6 +336,7 @@ impl MetaSrvBuilder {
             .await,
             plugins: plugins.unwrap_or_else(Plugins::default),
             memory_region_keeper: opening_region_keeper,
+            region_migration_manager,
         })
     }
 }

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

@@ -135,6 +135,24 @@ pub struct ContextFactoryImpl {
     server_addr: String,
 }
 
+impl ContextFactoryImpl {
+    /// Returns an [ContextFactoryImpl].
+    pub fn new(
+        table_metadata_manager: TableMetadataManagerRef,
+        opening_region_keeper: MemoryRegionKeeperRef,
+        mailbox: MailboxRef,
+        server_addr: String,
+    ) -> Self {
+        Self {
+            volatile_ctx: VolatileContext::default(),
+            table_metadata_manager,
+            opening_region_keeper,
+            mailbox,
+            server_addr,
+        }
+    }
+}
+
 impl ContextFactory for ContextFactoryImpl {
     fn new_context(self, persistent_ctx: PersistentContext) -> Context {
         Context {

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

@@ -31,8 +31,10 @@ use crate::procedure::region_migration::{
     ContextFactoryImpl, PersistentContext, RegionMigrationProcedure,
 };
 
+pub type RegionMigrationManagerRef = Arc<RegionMigrationManager>;
+
 /// Manager of region migration procedure.
-pub(crate) struct RegionMigrationManager {
+pub struct RegionMigrationManager {
     procedure_manager: ProcedureManagerRef,
     running_procedures: Arc<RwLock<HashMap<RegionId, RegionMigrationProcedureTask>>>,
     context_factory: ContextFactoryImpl,

src/meta-srv/src/service/admin.rs

@@ -93,6 +93,11 @@ pub fn make_admin_service(meta_srv: MetaSrv) -> Admin {
         .route("/route", handler.clone())
         .route("/route/help", handler);
 
+    let handler = region_migration::SubmitRegionMigrationTaskHandler {
+        region_migration_manager: meta_srv.region_migration_manager().clone(),
+    };
+    let router = router.route("/region-migration", handler);
+
     let router = Router::nest("/admin", router);
 
     Admin::new(router)

src/meta-srv/src/service/admin/region_migration.rs

@@ -25,14 +25,11 @@ use tonic::codegen::http;
 
 use super::HttpHandler;
 use crate::error::{self, Error, Result};
-
-pub trait PeerLookup: Send + Sync {
-    fn peer(&self, peer_id: u64) -> Option<Peer>;
-}
+use crate::procedure::region_migration::manager::RegionMigrationManagerRef;
 
 /// The handler of submitting migration task.
 pub struct SubmitRegionMigrationTaskHandler {
-    // TODO(weny): waits for https://github.com/GreptimeTeam/greptimedb/pull/3014
+    pub region_migration_manager: RegionMigrationManagerRef,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]