refactor: use downgrading the region instead of closing region (#2863)

* refactor: use downgrading the region instead of closing region * feat: enhance the tests for alive keeper * feat: add a metric to track region lease expired * chore: apply suggestions from CR * chore: enable logging for test_distributed_handle_ddl_request * refactor: simplify lease keeper * feat: add metrics for lease keeper * chore: apply suggestions from CR * chore: apply suggestions from CR * chore: apply suggestions from CR * refactor: move OpeningRegionKeeper to common_meta * feat: register operating regions to MemoryRegionKeeper
GreptimeTeam · Dec 12, 2023 · cf6bba0 · cf6bba0
1 parent 89a0d3a
commit cf6bba0
Show file tree

Hide file tree

Showing 30 changed files with 768 additions and 1,180 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/src/cmd/src/standalone.rs b/src/cmd/src/standalone.rs
@@ -23,6 +23,7 @@ use common_meta::ddl::DdlTaskExecutorRef;
 use common_meta::ddl_manager::DdlManager;
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
+use common_meta::region_keeper::MemoryRegionKeeper;
 use common_procedure::ProcedureManagerRef;
 use common_telemetry::info;
 use common_telemetry::logging::LoggingOptions;
@@ -396,6 +397,7 @@ impl StartCommand {
                 Arc::new(DummyCacheInvalidator),
                 table_metadata_manager,
                 Arc::new(StandaloneTableMetadataCreator::new(kv_backend)),
+                Arc::new(MemoryRegionKeeper::default()),
             )
             .context(InitDdlManagerSnafu)?,
         );

diff --git a/src/common/meta/Cargo.toml b/src/common/meta/Cargo.toml
@@ -24,6 +24,7 @@ common-runtime.workspace = true
 common-telemetry.workspace = true
 common-time.workspace = true
 datatypes.workspace = true
+derive_builder.workspace = true
 etcd-client.workspace = true
 futures.workspace = true
 humantime-serde.workspace = true

diff --git a/src/common/meta/src/ddl.rs b/src/common/meta/src/ddl.rs
@@ -23,6 +23,7 @@ use crate::cache_invalidator::CacheInvalidatorRef;
 use crate::datanode_manager::DatanodeManagerRef;
 use crate::error::Result;
 use crate::key::TableMetadataManagerRef;
+use crate::region_keeper::MemoryRegionKeeperRef;
 use crate::rpc::ddl::{SubmitDdlTaskRequest, SubmitDdlTaskResponse};
 use crate::rpc::router::RegionRoute;
 
@@ -70,4 +71,5 @@ pub struct DdlContext {
     pub datanode_manager: DatanodeManagerRef,
     pub cache_invalidator: CacheInvalidatorRef,
     pub table_metadata_manager: TableMetadataManagerRef,
+    pub memory_region_keeper: MemoryRegionKeeperRef,
 }
diff --git a/src/common/meta/src/ddl/create_table.rs b/src/common/meta/src/ddl/create_table.rs
@@ -18,7 +18,10 @@ use api::v1::region::{
 };
 use api::v1::{ColumnDef, SemanticType};
 use async_trait::async_trait;
-use common_procedure::error::{FromJsonSnafu, Result as ProcedureResult, ToJsonSnafu};
+use common_error::ext::BoxedError;
+use common_procedure::error::{
+    ExternalSnafu, FromJsonSnafu, Result as ProcedureResult, ToJsonSnafu,
+};
 use common_procedure::{Context as ProcedureContext, LockKey, Procedure, Status};
 use common_telemetry::info;
 use common_telemetry::tracing_context::TracingContext;
@@ -35,8 +38,11 @@ use crate::ddl::DdlContext;
 use crate::error::{self, Result};
 use crate::key::table_name::TableNameKey;
 use crate::metrics;
+use crate::region_keeper::OperatingRegionGuard;
 use crate::rpc::ddl::CreateTableTask;
-use crate::rpc::router::{find_leader_regions, find_leaders, RegionRoute};
+use crate::rpc::router::{
+    find_leader_regions, find_leaders, operating_leader_regions, RegionRoute,
+};
 
 pub struct CreateTableProcedure {
     pub context: DdlContext,
@@ -60,10 +66,18 @@ impl CreateTableProcedure {
 
     pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult<Self> {
         let data = serde_json::from_str(json).context(FromJsonSnafu)?;
-        Ok(CreateTableProcedure {
-            context,
-            creator: TableCreator { data },
-        })
+
+        let mut creator = TableCreator {
+            data,
+            opening_regions: vec![],
+        };
+
+        creator
+            .register_opening_regions(&context)
+            .map_err(BoxedError::new)
+            .context(ExternalSnafu)?;
+
+        Ok(CreateTableProcedure { context, creator })
     }
 
     pub fn table_info(&self) -> &RawTableInfo {
@@ -169,6 +183,9 @@ impl CreateTableProcedure {
     }
 
     pub async fn on_datanode_create_regions(&mut self) -> Result<Status> {
+        // Registers opening regions
+        self.creator.register_opening_regions(&self.context)?;
+
         let create_table_data = &self.creator.data;
         let region_routes = &create_table_data.region_routes;
 
@@ -226,7 +243,9 @@ impl CreateTableProcedure {
 
         self.creator.data.state = CreateTableState::CreateMetadata;
 
-        Ok(Status::executing(true))
+        // Ensures the procedures after the crash start from the `DatanodeCreateRegions` stage.
+        // TODO(weny): Add more tests.
+        Ok(Status::executing(false))
     }
 
     async fn on_create_metadata(&self) -> Result<Status> {
@@ -282,7 +301,10 @@ impl Procedure for CreateTableProcedure {
 }
 
 pub struct TableCreator {
+    /// The serializable data.
     pub data: CreateTableData,
+    /// The guards of opening.
+    pub opening_regions: Vec<OperatingRegionGuard>,
 }
 
 impl TableCreator {
@@ -294,8 +316,36 @@ impl TableCreator {
                 task,
                 region_routes,
             },
+            opening_regions: vec![],
         }
     }
+
+    /// Register opening regions if doesn't exist.
+    pub fn register_opening_regions(&mut self, context: &DdlContext) -> Result<()> {
+        let region_routes = &self.data.region_routes;
+
+        let opening_regions = operating_leader_regions(region_routes);
+
+        if self.opening_regions.len() == opening_regions.len() {
+            return Ok(());
+        }
+
+        let mut opening_region_guards = Vec::with_capacity(opening_regions.len());
+
+        for (region_id, datanode_id) in opening_regions {
+            let guard = context
+                .memory_region_keeper
+                .register(datanode_id, region_id)
+                .context(error::RegionOperatingRaceSnafu {
+                    region_id,
+                    peer_id: datanode_id,
+                })?;
+            opening_region_guards.push(guard);
+        }
+
+        self.opening_regions = opening_region_guards;
+        Ok(())
+    }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, AsRefStr)]

diff --git a/src/common/meta/src/ddl/drop_table.rs b/src/common/meta/src/ddl/drop_table.rs
@@ -26,7 +26,7 @@ use common_telemetry::tracing_context::TracingContext;
 use common_telemetry::{debug, info};
 use futures::future::join_all;
 use serde::{Deserialize, Serialize};
-use snafu::{ensure, ResultExt};
+use snafu::{ensure, OptionExt, ResultExt};
 use store_api::storage::RegionId;
 use strum::AsRefStr;
 use table::engine::TableReference;
@@ -42,12 +42,19 @@ use crate::key::table_name::TableNameKey;
 use crate::key::table_route::TableRouteValue;
 use crate::key::DeserializedValueWithBytes;
 use crate::metrics;
+use crate::region_keeper::OperatingRegionGuard;
 use crate::rpc::ddl::DropTableTask;
-use crate::rpc::router::{find_leader_regions, find_leaders, RegionRoute};
+use crate::rpc::router::{
+    find_leader_regions, find_leaders, operating_leader_regions, RegionRoute,
+};
 
 pub struct DropTableProcedure {
+    /// The context of procedure runtime.
     pub context: DdlContext,
+    /// The serializable data.
     pub data: DropTableData,
+    /// The guards of opening regions.
+    pub dropping_regions: Vec<OperatingRegionGuard>,
 }
 
 #[allow(dead_code)]
@@ -64,12 +71,17 @@ impl DropTableProcedure {
         Self {
             context,
             data: DropTableData::new(cluster_id, task, table_route_value, table_info_value),
+            dropping_regions: vec![],
         }
     }
 
     pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult<Self> {
         let data = serde_json::from_str(json).context(FromJsonSnafu)?;
-        Ok(Self { context, data })
+        Ok(Self {
+            context,
+            data,
+            dropping_regions: vec![],
+        })
     }
 
     async fn on_prepare(&mut self) -> Result<Status> {
@@ -102,8 +114,42 @@ impl DropTableProcedure {
         Ok(Status::executing(true))
     }
 
+    /// Register dropping regions if doesn't exist.
+    fn register_dropping_regions(&mut self) -> Result<()> {
+        let region_routes = self.data.region_routes();
+
+        let dropping_regions = operating_leader_regions(region_routes);
+
+        if self.dropping_regions.len() == dropping_regions.len() {
+            return Ok(());
+        }
+
+        let mut dropping_region_guards = Vec::with_capacity(dropping_regions.len());
+
+        for (region_id, datanode_id) in dropping_regions {
+            let guard = self
+                .context
+                .memory_region_keeper
+                .register(datanode_id, region_id)
+                .context(error::RegionOperatingRaceSnafu {
+                    region_id,
+                    peer_id: datanode_id,
+                })?;
+            dropping_region_guards.push(guard);
+        }
+
+        self.dropping_regions = dropping_region_guards;
+        Ok(())
+    }
+
     /// Removes the table metadata.
     async fn on_remove_metadata(&mut self) -> Result<Status> {
+        // NOTES: If the meta server is crashed after the `RemoveMetadata`,
+        // Corresponding regions of this table on the Datanode will be closed automatically.
+        // Then any future dropping operation will fail.
+
+        // TODO(weny): Considers introducing a RegionStatus to indicate the region is dropping.
+
         let table_metadata_manager = &self.context.table_metadata_manager;
         let table_info_value = &self.data.table_info_value;
         let table_route_value = &self.data.table_route_value;

diff --git a/src/common/meta/src/ddl_manager.rs b/src/common/meta/src/ddl_manager.rs
@@ -37,6 +37,7 @@ use crate::key::table_info::TableInfoValue;
 use crate::key::table_name::TableNameKey;
 use crate::key::table_route::TableRouteValue;
 use crate::key::{DeserializedValueWithBytes, TableMetadataManagerRef};
+use crate::region_keeper::MemoryRegionKeeperRef;
 use crate::rpc::ddl::DdlTask::{AlterTable, CreateTable, DropTable, TruncateTable};
 use crate::rpc::ddl::{
     AlterTableTask, CreateTableTask, DropTableTask, SubmitDdlTaskRequest, SubmitDdlTaskResponse,
@@ -52,6 +53,7 @@ pub struct DdlManager {
     cache_invalidator: CacheInvalidatorRef,
     table_metadata_manager: TableMetadataManagerRef,
     table_meta_allocator: TableMetadataAllocatorRef,
+    memory_region_keeper: MemoryRegionKeeperRef,
 }
 
 impl DdlManager {
@@ -62,13 +64,15 @@ impl DdlManager {
         cache_invalidator: CacheInvalidatorRef,
         table_metadata_manager: TableMetadataManagerRef,
         table_meta_allocator: TableMetadataAllocatorRef,
+        memory_region_keeper: MemoryRegionKeeperRef,
     ) -> Result<Self> {
         let manager = Self {
             procedure_manager,
             datanode_manager: datanode_clients,
             cache_invalidator,
             table_metadata_manager,
             table_meta_allocator,
+            memory_region_keeper,
         };
         manager.register_loaders()?;
         Ok(manager)
@@ -85,6 +89,7 @@ impl DdlManager {
             datanode_manager: self.datanode_manager.clone(),
             cache_invalidator: self.cache_invalidator.clone(),
             table_metadata_manager: self.table_metadata_manager.clone(),
+            memory_region_keeper: self.memory_region_keeper.clone(),
         }
     }
 
@@ -446,6 +451,7 @@ mod tests {
     use crate::key::TableMetadataManager;
     use crate::kv_backend::memory::MemoryKvBackend;
     use crate::peer::Peer;
+    use crate::region_keeper::MemoryRegionKeeper;
     use crate::rpc::router::RegionRoute;
     use crate::state_store::KvStateStore;
 
@@ -488,6 +494,7 @@ mod tests {
             Arc::new(DummyCacheInvalidator),
             table_metadata_manager,
             Arc::new(DummyTableMetadataAllocator),
+            Arc::new(MemoryRegionKeeper::default()),
         );
 
         let expected_loaders = vec![

diff --git a/src/common/meta/src/error.rs b/src/common/meta/src/error.rs
@@ -19,10 +19,11 @@ use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
 use serde_json::error::Error as JsonError;
 use snafu::{Location, Snafu};
-use store_api::storage::RegionNumber;
+use store_api::storage::{RegionId, RegionNumber};
 use table::metadata::TableId;
 
 use crate::peer::Peer;
+use crate::DatanodeId;
 
 #[derive(Snafu)]
 #[snafu(visibility(pub))]
@@ -31,6 +32,17 @@ pub enum Error {
     #[snafu(display("Empty key is not allowed"))]
     EmptyKey { location: Location },
 
+    #[snafu(display(
+        "Another procedure is operating the region: {} on peer: {}",
+        region_id,
+        peer_id
+    ))]
+    RegionOperatingRace {
+        location: Location,
+        peer_id: DatanodeId,
+        region_id: RegionId,
+    },
+
     #[snafu(display("Invalid result with a txn response: {}", err_msg))]
     InvalidTxnResult { err_msg: String, location: Location },
 
@@ -291,7 +303,16 @@ impl ErrorExt for Error {
             | SequenceOutOfRange { .. }
             | UnexpectedSequenceValue { .. }
             | InvalidHeartbeatResponse { .. }
-            | InvalidTxnResult { .. } => StatusCode::Unexpected,
+            | InvalidTxnResult { .. }
+            | EncodeJson { .. }
+            | DecodeJson { .. }
+            | PayloadNotExist { .. }
+            | ConvertRawKey { .. }
+            | DecodeProto { .. }
+            | BuildTableMeta { .. }
+            | TableRouteNotFound { .. }
+            | ConvertRawTableInfo { .. }
+            | RegionOperatingRace { .. } => StatusCode::Unexpected,
 
             SendMessage { .. }
             | GetKvCache { .. }
@@ -306,15 +327,6 @@ impl ErrorExt for Error {
             TableNotFound { .. } => StatusCode::TableNotFound,
             TableAlreadyExists { .. } => StatusCode::TableAlreadyExists,
 
-            EncodeJson { .. }
-            | DecodeJson { .. }
-            | PayloadNotExist { .. }
-            | ConvertRawKey { .. }
-            | DecodeProto { .. }
-            | BuildTableMeta { .. }
-            | TableRouteNotFound { .. }
-            | ConvertRawTableInfo { .. } => StatusCode::Unexpected,
-
             SubmitProcedure { source, .. } | WaitProcedure { source, .. } => source.status_code(),
             RegisterProcedureLoader { source, .. } => source.status_code(),
             External { source, .. } => source.status_code(),