diff --git a/config/config.md b/config/config.md index a8d12c3557..2d85ddd717 100644 --- a/config/config.md +++ b/config/config.md @@ -249,7 +249,7 @@ | `pipeline.cache_ttl` | String | `10s` | Time to live of the local pipeline cache. Default is `10s`. | | `event_recorder` | -- | -- | Configuration options for the event recorder. | | `event_recorder.ttl` | String | `90d` | TTL for the events table that will be used to store the events. Default is `90d`. | -| `event_recorder.event_types` | Array | -- | Event types to record. Current available event types: `create_database`,
`alter_database`, `drop_database`, `create_flow`, `drop_flow`,
`create_table`, `create_logical_tables`, `alter_table`, `alter_logical_tables`,
`drop_table`, `undrop_table`, `purge_dropped_table`, `truncate_table`,
`create_view`, `drop_view`, `admin_function`, `reconcile_table`,
`reconcile_logical_tables`.
When omitted, all current and future event types are recorded.
Set to an empty array to disable event recording. | +| `event_recorder.event_types` | Array | -- | Event types to record. Current available event types: `create_database`,
`alter_database`, `drop_database`, `create_flow`, `drop_flow`,
`create_table`, `create_logical_tables`, `alter_table`, `alter_logical_tables`,
`drop_table`, `undrop_table`, `purge_dropped_table`, `truncate_table`,
`create_view`, `drop_view`, `admin_function`, `reconcile_catalog`,
`reconcile_database`, `reconcile_table`, `reconcile_logical_tables`.
When omitted, all current and future event types are recorded.
Set to an empty array to disable event recording. | | `memory` | -- | -- | The memory options. | | `memory.enable_heap_profiling` | Bool | `true` | Whether to enable heap profiling activation during startup.
When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable
is set to "prof:true,prof_active:false". The official image adds this env variable.
Default is true. | @@ -477,7 +477,7 @@ | `wal.create_topic_timeout` | String | `30s` | The timeout for creating a Kafka topic.
**It's only used when the provider is `kafka`**. | | `event_recorder` | -- | -- | Configuration options for the event recorder. | | `event_recorder.ttl` | String | `90d` | TTL for the events table that will be used to store the events. Default is `90d`. | -| `event_recorder.event_types` | Array | -- | Event types to record. Current available event types: `region_migration`,
`create_database`, `alter_database`, `drop_database`, `create_flow`,
`drop_flow`, `create_table`, `create_logical_tables`, `alter_table`,
`alter_logical_tables`, `drop_table`, `undrop_table`, `purge_dropped_table`,
`truncate_table`, `create_view`, `drop_view`, `repartition`,
`repartition_group`, `wal_prune`, `batch_gc`, `reconcile_table`,
`reconcile_logical_tables`.
When omitted, all current and future event types are recorded.
Set to an empty array to disable event recording. | +| `event_recorder.event_types` | Array | -- | Event types to record. Current available event types: `region_migration`,
`create_database`, `alter_database`, `drop_database`, `create_flow`,
`drop_flow`, `create_table`, `create_logical_tables`, `alter_table`,
`alter_logical_tables`, `drop_table`, `undrop_table`, `purge_dropped_table`,
`truncate_table`, `create_view`, `drop_view`, `repartition`,
`repartition_group`, `wal_prune`, `batch_gc`, `reconcile_catalog`,
`reconcile_database`, `reconcile_table`, `reconcile_logical_tables`.
When omitted, all current and future event types are recorded.
Set to an empty array to disable event recording. | | `stats_persistence` | -- | -- | Configuration options for the stats persistence. | | `stats_persistence.ttl` | String | `0s` | TTL for the stats table that will be used to store the stats.
Set to `0s` to disable stats persistence.
Default is `0s`.
If you want to enable stats persistence, set the TTL to a value greater than 0.
It is recommended to set a small value, e.g., `3h`. | | `stats_persistence.interval` | String | `10m` | The interval to persist the stats. Default is `10m`.
The minimum value is `10m`, if the value is less than `10m`, it will be overridden to `10m`. | diff --git a/config/metasrv.example.toml b/config/metasrv.example.toml index 514993a4ca..b923c5a549 100644 --- a/config/metasrv.example.toml +++ b/config/metasrv.example.toml @@ -328,8 +328,8 @@ ttl = "90d" ## `drop_flow`, `create_table`, `create_logical_tables`, `alter_table`, ## `alter_logical_tables`, `drop_table`, `undrop_table`, `purge_dropped_table`, ## `truncate_table`, `create_view`, `drop_view`, `repartition`, -## `repartition_group`, `wal_prune`, `batch_gc`, `reconcile_table`, -## `reconcile_logical_tables`. +## `repartition_group`, `wal_prune`, `batch_gc`, `reconcile_catalog`, +## `reconcile_database`, `reconcile_table`, `reconcile_logical_tables`. ## When omitted, all current and future event types are recorded. ## Set to an empty array to disable event recording. #+ event_types = ["region_migration"] diff --git a/config/standalone.example.toml b/config/standalone.example.toml index b05cf0177d..0bef079324 100644 --- a/config/standalone.example.toml +++ b/config/standalone.example.toml @@ -973,8 +973,8 @@ ttl = "90d" ## `alter_database`, `drop_database`, `create_flow`, `drop_flow`, ## `create_table`, `create_logical_tables`, `alter_table`, `alter_logical_tables`, ## `drop_table`, `undrop_table`, `purge_dropped_table`, `truncate_table`, -## `create_view`, `drop_view`, `admin_function`, `reconcile_table`, -## `reconcile_logical_tables`. +## `create_view`, `drop_view`, `admin_function`, `reconcile_catalog`, +## `reconcile_database`, `reconcile_table`, `reconcile_logical_tables`. ## When omitted, all current and future event types are recorded. ## Set to an empty array to disable event recording. #+ event_types = ["create_database"] diff --git a/src/common/meta/src/reconciliation/event.rs b/src/common/meta/src/reconciliation/event.rs index cff4a0db19..06629bc48c 100644 --- a/src/common/meta/src/reconciliation/event.rs +++ b/src/common/meta/src/reconciliation/event.rs @@ -28,6 +28,10 @@ use store_api::storage::TableId; use crate::reconciliation::ResolveStrategy; +/// Stable event type stored for catalog reconciliation procedures. +pub(crate) const RECONCILE_CATALOG_EVENT_TYPE: &str = "reconcile_catalog"; +/// Stable event type stored for database reconciliation procedures. +pub(crate) const RECONCILE_DATABASE_EVENT_TYPE: &str = "reconcile_database"; /// Stable event type stored for logical table reconciliation procedures. pub(crate) const RECONCILE_LOGICAL_TABLES_EVENT_TYPE: &str = "reconcile_logical_tables"; /// Stable event type stored for physical table reconciliation procedures. @@ -45,6 +49,23 @@ pub(crate) struct ReconciliationLocator { } impl ReconciliationLocator { + /// Creates a locator for a catalog. + pub(crate) fn catalog(catalog_name: &str) -> Self { + Self { + catalog_name: Some(catalog_name.to_string()), + ..Default::default() + } + } + + /// Creates a locator for a database. + pub(crate) fn database(catalog_name: &str, schema_name: &str) -> Self { + Self { + catalog_name: Some(catalog_name.to_string()), + schema_name: Some(schema_name.to_string()), + ..Default::default() + } + } + /// Creates a locator for a physical table with its fully qualified name and ID. pub(crate) fn physical_table( catalog_name: &str, @@ -101,6 +122,229 @@ impl ReconciliationLocator { } } +#[derive(Debug, Serialize)] +#[serde(untagged)] +enum ReconcileCatalogPayload { + Submitted(CatalogSubmittedPayload), + Result(CatalogResultPayload), +} + +#[derive(Debug, Serialize)] +struct CatalogSubmittedPayload { + version: u8, + resolve_strategy: &'static str, + fail_fast: bool, + parallelism: usize, +} + +#[derive(Debug, Serialize)] +struct CatalogResultPayload { + version: u8, + complete: bool, + processed_database_count: usize, + succeeded_database_count: usize, + failed_database_count: usize, +} + +/// Event representation for catalog reconciliation. +#[derive(Debug)] +pub(crate) struct ReconcileCatalogEvent { + locator: ReconciliationLocator, + payload: Option, +} + +impl ReconcileCatalogEvent { + /// Builds the bounded intent event emitted when catalog reconciliation is submitted. + pub(crate) fn submitted( + locator: ReconciliationLocator, + resolve_strategy: ResolveStrategy, + fail_fast: bool, + parallelism: usize, + ) -> Self { + Self { + locator, + payload: Some(ReconcileCatalogPayload::Submitted( + CatalogSubmittedPayload { + version: PAYLOAD_VERSION, + resolve_strategy: resolve_strategy_name(resolve_strategy), + fail_fast, + parallelism, + }, + )), + } + } + + /// Builds a terminal event from the existing volatile reconciliation metrics. + /// + /// Metrics are best-effort observations from the current process and reset on recovery. + pub(crate) fn result( + locator: ReconciliationLocator, + complete: bool, + succeeded_database_count: usize, + failed_database_count: usize, + ) -> Self { + Self { + locator, + payload: Some(ReconcileCatalogPayload::Result(CatalogResultPayload { + version: PAYLOAD_VERSION, + complete, + processed_database_count: succeeded_database_count + failed_database_count, + succeeded_database_count, + failed_database_count, + })), + } + } + + /// Builds a catalog lifecycle event whose reconciliation payload is null. + pub(crate) fn lifecycle(locator: ReconciliationLocator) -> Self { + Self { + locator, + payload: None, + } + } +} + +impl Event for ReconcileCatalogEvent { + fn event_type(&self) -> &str { + RECONCILE_CATALOG_EVENT_TYPE + } + + fn json_payload(&self) -> Result { + match &self.payload { + Some(payload) => serde_json::to_value(payload).context(SerializeEventSnafu), + None => Ok(serde_json::Value::Null), + } + } + + fn extra_schema(&self) -> Vec { + ReconciliationLocator::schema() + } + + fn extra_rows(&self) -> Result> { + Ok(vec![self.locator.row()]) + } + + fn as_any(&self) -> &dyn Any { + self + } +} + +#[derive(Debug, Serialize)] +#[serde(untagged)] +enum ReconcileDatabasePayload { + Submitted(DatabaseSubmittedPayload), + Result(DatabaseResultPayload), +} + +#[derive(Debug, Serialize)] +struct DatabaseSubmittedPayload { + version: u8, + resolve_strategy: &'static str, + fail_fast: bool, + parallelism: usize, + is_subprocedure: bool, +} + +#[derive(Debug, Serialize)] +struct DatabaseResultPayload { + version: u8, + complete: bool, + processed_table_count: usize, + succeeded_table_count: usize, + failed_table_count: usize, + succeeded_subprocedure_count: usize, + failed_subprocedure_count: usize, +} + +/// Event representation for database reconciliation. +#[derive(Debug)] +pub(crate) struct ReconcileDatabaseEvent { + locator: ReconciliationLocator, + payload: Option, +} + +impl ReconcileDatabaseEvent { + /// Builds the bounded intent event emitted when database reconciliation is submitted. + pub(crate) fn submitted( + locator: ReconciliationLocator, + resolve_strategy: ResolveStrategy, + fail_fast: bool, + parallelism: usize, + is_subprocedure: bool, + ) -> Self { + Self { + locator, + payload: Some(ReconcileDatabasePayload::Submitted( + DatabaseSubmittedPayload { + version: PAYLOAD_VERSION, + resolve_strategy: resolve_strategy_name(resolve_strategy), + fail_fast, + parallelism, + is_subprocedure, + }, + )), + } + } + + /// Builds a terminal event from the existing volatile reconciliation metrics. + /// + /// Metrics are best-effort observations from the current process and reset on recovery. + pub(crate) fn result( + locator: ReconciliationLocator, + complete: bool, + succeeded_table_count: usize, + failed_table_count: usize, + succeeded_subprocedure_count: usize, + failed_subprocedure_count: usize, + ) -> Self { + Self { + locator, + payload: Some(ReconcileDatabasePayload::Result(DatabaseResultPayload { + version: PAYLOAD_VERSION, + complete, + processed_table_count: succeeded_table_count + failed_table_count, + succeeded_table_count, + failed_table_count, + succeeded_subprocedure_count, + failed_subprocedure_count, + })), + } + } + + /// Builds a database lifecycle event whose reconciliation payload is null. + pub(crate) fn lifecycle(locator: ReconciliationLocator) -> Self { + Self { + locator, + payload: None, + } + } +} + +impl Event for ReconcileDatabaseEvent { + fn event_type(&self) -> &str { + RECONCILE_DATABASE_EVENT_TYPE + } + + fn json_payload(&self) -> Result { + match &self.payload { + Some(payload) => serde_json::to_value(payload).context(SerializeEventSnafu), + None => Ok(serde_json::Value::Null), + } + } + + fn extra_schema(&self) -> Vec { + ReconciliationLocator::schema() + } + + fn extra_rows(&self) -> Result> { + Ok(vec![self.locator.row()]) + } + + fn as_any(&self) -> &dyn Any { + self + } +} + #[derive(Debug, Serialize)] #[serde(untagged)] enum ReconcileTablePayload { @@ -355,12 +599,14 @@ mod tests { #[test] fn reconciliation_events_use_the_shared_locator_contract() { + let catalog = ReconcileCatalogEvent::lifecycle(ReconciliationLocator::catalog("greptime")); + assert_eq!(catalog.event_type(), RECONCILE_CATALOG_EVENT_TYPE); let table = ReconcileTableEvent::table_lifecycle(ReconciliationLocator::physical_table( "greptime", "public", "metrics", 42, )); assert_eq!(table.event_type(), RECONCILE_TABLE_EVENT_TYPE); assert_eq!( - table + catalog .extra_schema() .into_iter() .map(|column| { @@ -375,22 +621,22 @@ mod tests { ( "catalog_name".to_string(), ColumnDataType::String, - SemanticType::Field + SemanticType::Field, ), ( "schema_name".to_string(), ColumnDataType::String, - SemanticType::Field + SemanticType::Field, ), ( "table_name".to_string(), ColumnDataType::String, - SemanticType::Field + SemanticType::Field, ), ( "table_id".to_string(), ColumnDataType::Uint32, - SemanticType::Field + SemanticType::Field, ), ( "physical_table_id".to_string(), @@ -399,6 +645,40 @@ mod tests { ), ] ); + assert_eq!( + catalog.extra_rows().unwrap(), + vec![Row { + values: vec![ + ValueData::StringValue("greptime".to_string()).into(), + Value::default(), + Value::default(), + Value::default(), + Value::default(), + ], + }] + ); + assert_eq!(catalog.json_payload().unwrap(), serde_json::Value::Null); + + let database = ReconcileDatabaseEvent::lifecycle(ReconciliationLocator::database( + "greptime", "public", + )); + assert_eq!(database.event_type(), RECONCILE_DATABASE_EVENT_TYPE); + assert_eq!(database.extra_schema(), catalog.extra_schema()); + assert_eq!( + database.extra_rows().unwrap(), + vec![Row { + values: vec![ + ValueData::StringValue("greptime".to_string()).into(), + ValueData::StringValue("public".to_string()).into(), + Value::default(), + Value::default(), + Value::default(), + ], + }] + ); + assert_eq!(database.json_payload().unwrap(), serde_json::Value::Null); + + assert_eq!(table.extra_schema(), catalog.extra_schema()); assert_eq!( table.extra_rows().unwrap(), vec![Row { @@ -458,6 +738,22 @@ mod tests { (ResolveStrategy::UseMetasrv, "use_metasrv"), (ResolveStrategy::AbortOnConflict, "abort_on_conflict"), ] { + let catalog = ReconcileCatalogEvent::submitted( + ReconciliationLocator::catalog("greptime"), + strategy, + true, + 16, + ); + assert_eq!( + catalog.json_payload().unwrap(), + json!({ + "version": 1, + "resolve_strategy": expected, + "fail_fast": true, + "parallelism": 16, + }) + ); + let table = ReconcileTableEvent::table_submitted( ReconciliationLocator::physical_table("greptime", "public", "metrics", 42), strategy, @@ -473,6 +769,24 @@ mod tests { ); } + let database = ReconcileDatabaseEvent::submitted( + ReconciliationLocator::database("greptime", "public"), + ResolveStrategy::UseMetasrv, + false, + 64, + true, + ); + assert_eq!( + database.json_payload().unwrap(), + json!({ + "version": 1, + "resolve_strategy": "use_metasrv", + "fail_fast": false, + "parallelism": 64, + "is_subprocedure": true, + }) + ); + let logical_tables = ReconcileLogicalTablesEvent::submitted( vec![ ReconciliationLocator::logical_table("greptime", "public", "cpu", 43, 42), @@ -492,6 +806,40 @@ mod tests { #[test] fn terminal_payloads_distinguish_complete_and_partial_results() { + let catalog = + ReconcileCatalogEvent::result(ReconciliationLocator::catalog("greptime"), true, 3, 1); + assert_eq!( + catalog.json_payload().unwrap(), + json!({ + "version": 1, + "complete": true, + "processed_database_count": 4, + "succeeded_database_count": 3, + "failed_database_count": 1, + }) + ); + + let database = ReconcileDatabaseEvent::result( + ReconciliationLocator::database("greptime", "public"), + false, + 5, + 2, + 4, + 1, + ); + assert_eq!( + database.json_payload().unwrap(), + json!({ + "version": 1, + "complete": false, + "processed_table_count": 7, + "succeeded_table_count": 5, + "failed_table_count": 2, + "succeeded_subprocedure_count": 4, + "failed_subprocedure_count": 1, + }) + ); + let table = ReconcileTableEvent::table_result( ReconciliationLocator::physical_table("greptime", "public", "metrics", 42), false, diff --git a/src/common/meta/src/reconciliation/reconcile_catalog.rs b/src/common/meta/src/reconciliation/reconcile_catalog.rs index d3c2e526dd..e28e81d323 100644 --- a/src/common/meta/src/reconciliation/reconcile_catalog.rs +++ b/src/common/meta/src/reconciliation/reconcile_catalog.rs @@ -18,8 +18,8 @@ use std::time::Instant; use common_procedure::error::FromJsonSnafu; use common_procedure::{ - Context as ProcedureContext, Error as ProcedureError, LockKey, Procedure, - Result as ProcedureResult, Status, + Context as ProcedureContext, Error as ProcedureError, EventContext, EventTrigger, LockKey, + Procedure, Result as ProcedureResult, Status, }; use futures::stream::BoxStream; use serde::{Deserialize, Serialize}; @@ -31,6 +31,9 @@ use crate::key::TableMetadataManagerRef; use crate::lock_key::CatalogLock; use crate::metrics; use crate::node_manager::NodeManagerRef; +use crate::reconciliation::event::{ + RECONCILE_CATALOG_EVENT_TYPE, ReconcileCatalogEvent, ReconciliationLocator, +}; use crate::reconciliation::reconcile_catalog::start::ReconcileCatalogStart; use crate::reconciliation::reconcile_table::resolve_column_metadata::ResolveStrategy; use crate::reconciliation::utils::{ @@ -216,6 +219,43 @@ impl Procedure for ReconcileCatalogProcedure { LockKey::new(vec![CatalogLock::Write(catalog).into()]) } + + fn event(&self, ctx: &EventContext<'_>) -> Option> { + if !ctx.event_type_filter.allows(RECONCILE_CATALOG_EVENT_TYPE) { + return None; + } + + let persistent_ctx = &self.context.persistent_ctx; + let locator = ReconciliationLocator::catalog(&persistent_ctx.catalog); + let event = match ctx.trigger { + EventTrigger::Submitted => ReconcileCatalogEvent::submitted( + locator, + persistent_ctx.resolve_strategy, + persistent_ctx.fast_fail, + persistent_ctx.parallelism, + ), + EventTrigger::Succeeded => self.result_event(locator, true), + EventTrigger::Failed | EventTrigger::Poisoned => self.result_event(locator, false), + _ => ReconcileCatalogEvent::lifecycle(locator), + }; + Some(Box::new(event)) + } +} + +impl ReconcileCatalogProcedure { + fn result_event( + &self, + locator: ReconciliationLocator, + complete: bool, + ) -> ReconcileCatalogEvent { + let metrics = &self.context.volatile_ctx.metrics; + ReconcileCatalogEvent::result( + locator, + complete, + metrics.succeeded_databases, + metrics.failed_databases, + ) + } } #[async_trait::async_trait] @@ -235,3 +275,217 @@ pub(crate) trait State: Sync + Send + Debug { fn as_any(&self) -> &dyn Any; } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use common_event_recorder::{EventTypeFilter, EventTypeFilterRef}; + use common_procedure::{ + ChildSubmissionOutcome, EventContext, EventTrigger, Procedure, ProcedureId, ProcedureState, + RetryPhase, + }; + use serde_json::{Value, json}; + + use super::*; + use crate::reconciliation::event::RECONCILE_DATABASE_EVENT_TYPE; + use crate::test_util::{MockDatanodeManager, new_ddl_context}; + + struct CatalogEventHarness { + procedure_id: ProcedureId, + lifecycle_state: ProcedureState, + event_type_filter: EventTypeFilterRef, + } + + impl CatalogEventHarness { + fn all() -> Self { + Self { + procedure_id: ProcedureId::random(), + lifecycle_state: ProcedureState::Running, + event_type_filter: Arc::new(EventTypeFilter::All), + } + } + + fn selected(event_types: impl IntoIterator) -> Self { + Self { + event_type_filter: Arc::new(EventTypeFilter::Only( + event_types.into_iter().map(str::to_string).collect(), + )), + ..Self::all() + } + } + + fn event( + &self, + procedure: &dyn Procedure, + trigger: EventTrigger, + ) -> Option> { + procedure.event(&EventContext { + procedure_id: self.procedure_id, + lifecycle_state: &self.lifecycle_state, + trigger, + event_type_filter: self.event_type_filter.clone(), + event_context: None, + }) + } + } + + #[test] + fn catalog_submitted_event_reports_intent() { + let submitted = CatalogEventHarness::all() + .event(&test_procedure(), EventTrigger::Submitted) + .unwrap(); + assert_eq!(submitted.event_type(), RECONCILE_CATALOG_EVENT_TYPE); + assert_eq!( + submitted.json_payload().unwrap(), + json!({ + "version": 1, + "resolve_strategy": "use_latest", + "fail_fast": false, + "parallelism": 8, + }) + ); + } + + #[test] + fn catalog_non_terminal_lifecycle_events_have_null_payloads() { + let events = CatalogEventHarness::all(); + let mut procedure = test_procedure(); + procedure.context.volatile_ctx.metrics = populated_metrics(); + + for trigger in [ + EventTrigger::Recovered, + EventTrigger::ChildSubmitted { + procedure_id: ProcedureId::random(), + outcome: ChildSubmissionOutcome::Accepted, + }, + EventTrigger::Retrying { + phase: RetryPhase::Execute, + attempt: 1, + }, + EventTrigger::RollingBack, + ] { + assert_eq!( + events + .event(&procedure, trigger) + .unwrap() + .json_payload() + .unwrap(), + Value::Null + ); + } + } + + #[test] + fn catalog_terminal_events_report_existing_metrics() { + let events = CatalogEventHarness::all(); + let mut procedure = test_procedure(); + procedure.context.volatile_ctx.metrics = populated_metrics(); + + for (trigger, complete) in [ + (EventTrigger::Succeeded, true), + (EventTrigger::Failed, false), + (EventTrigger::Poisoned, false), + ] { + assert_eq!( + events + .event(&procedure, trigger) + .unwrap() + .json_payload() + .unwrap(), + json!({ + "version": 1, + "complete": complete, + "processed_database_count": 4, + "succeeded_database_count": 3, + "failed_database_count": 1, + }) + ); + } + } + + #[test] + fn catalog_event_filtering_uses_the_catalog_event_type() { + let procedure = test_procedure(); + assert!( + CatalogEventHarness::selected([RECONCILE_CATALOG_EVENT_TYPE]) + .event(&procedure, EventTrigger::Submitted) + .is_some() + ); + assert!( + CatalogEventHarness::selected([RECONCILE_DATABASE_EVENT_TYPE]) + .event(&procedure, EventTrigger::Submitted) + .is_none() + ); + assert!( + CatalogEventHarness::selected([]) + .event(&procedure, EventTrigger::Submitted) + .is_none() + ); + } + + #[test] + fn catalog_recovery_preserves_locator_and_resets_metrics() { + let events = CatalogEventHarness::all(); + let mut procedure = test_procedure(); + procedure.context.volatile_ctx.metrics = populated_metrics(); + let original_dump = procedure.dump().unwrap(); + procedure.context.volatile_ctx.metrics = ReconcileCatalogMetrics::default(); + assert_eq!(procedure.dump().unwrap(), original_dump); + + let loaded = ReconcileCatalogProcedure::from_json(test_context(), &original_dump).unwrap(); + assert_eq!(loaded.dump().unwrap(), original_dump); + assert_eq!( + events + .event(&loaded, EventTrigger::Recovered) + .unwrap() + .extra_rows() + .unwrap(), + events + .event(&procedure, EventTrigger::Submitted) + .unwrap() + .extra_rows() + .unwrap(), + ); + assert_eq!( + events + .event(&loaded, EventTrigger::Succeeded) + .unwrap() + .json_payload() + .unwrap(), + json!({ + "version": 1, + "complete": true, + "processed_database_count": 0, + "succeeded_database_count": 0, + "failed_database_count": 0, + }) + ); + } + + fn populated_metrics() -> ReconcileCatalogMetrics { + ReconcileCatalogMetrics { + succeeded_databases: 3, + failed_databases: 1, + } + } + + fn test_procedure() -> ReconcileCatalogProcedure { + ReconcileCatalogProcedure::new( + test_context(), + "greptime".to_string(), + false, + ResolveStrategy::UseLatest, + 8, + ) + } + + fn test_context() -> Context { + let ddl_context = new_ddl_context(Arc::new(MockDatanodeManager::new(()))); + Context { + node_manager: ddl_context.node_manager, + table_metadata_manager: ddl_context.table_metadata_manager, + cache_invalidator: ddl_context.cache_invalidator, + } + } +} diff --git a/src/common/meta/src/reconciliation/reconcile_database.rs b/src/common/meta/src/reconciliation/reconcile_database.rs index 7d4262877c..f5e2fe9f9a 100644 --- a/src/common/meta/src/reconciliation/reconcile_database.rs +++ b/src/common/meta/src/reconciliation/reconcile_database.rs @@ -25,8 +25,8 @@ use std::time::Instant; use async_trait::async_trait; use common_procedure::error::{FromJsonSnafu, ToJsonSnafu}; use common_procedure::{ - Context as ProcedureContext, Error as ProcedureError, LockKey, Procedure, - Result as ProcedureResult, Status, + Context as ProcedureContext, Error as ProcedureError, EventContext, EventTrigger, LockKey, + Procedure, Result as ProcedureResult, Status, }; use futures::stream::BoxStream; use serde::{Deserialize, Serialize}; @@ -41,6 +41,9 @@ use crate::key::table_name::TableNameValue; use crate::lock_key::{CatalogLock, SchemaLock}; use crate::metrics; use crate::node_manager::NodeManagerRef; +use crate::reconciliation::event::{ + RECONCILE_DATABASE_EVENT_TYPE, ReconcileDatabaseEvent, ReconciliationLocator, +}; use crate::reconciliation::reconcile_database::start::ReconcileDatabaseStart; use crate::reconciliation::reconcile_table::resolve_column_metadata::ResolveStrategy; use crate::reconciliation::utils::{ @@ -264,6 +267,47 @@ impl Procedure for ReconcileDatabaseProcedure { SchemaLock::write(catalog, schema).into(), ]) } + + fn event(&self, ctx: &EventContext<'_>) -> Option> { + if !ctx.event_type_filter.allows(RECONCILE_DATABASE_EVENT_TYPE) { + return None; + } + + let persistent_ctx = &self.context.persistent_ctx; + let locator = + ReconciliationLocator::database(&persistent_ctx.catalog, &persistent_ctx.schema); + let event = match ctx.trigger { + EventTrigger::Submitted => ReconcileDatabaseEvent::submitted( + locator, + persistent_ctx.resolve_strategy, + persistent_ctx.fail_fast, + persistent_ctx.parallelism, + persistent_ctx.is_subprocedure, + ), + EventTrigger::Succeeded => self.result_event(locator, true), + EventTrigger::Failed | EventTrigger::Poisoned => self.result_event(locator, false), + _ => ReconcileDatabaseEvent::lifecycle(locator), + }; + Some(Box::new(event)) + } +} + +impl ReconcileDatabaseProcedure { + fn result_event( + &self, + locator: ReconciliationLocator, + complete: bool, + ) -> ReconcileDatabaseEvent { + let metrics = self.context.metrics(); + ReconcileDatabaseEvent::result( + locator, + complete, + metrics.succeeded_tables, + metrics.failed_tables, + metrics.succeeded_procedures, + metrics.failed_procedures, + ) + } } #[async_trait::async_trait] @@ -283,3 +327,229 @@ pub(crate) trait State: Sync + Send + Debug { fn as_any(&self) -> &dyn Any; } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use common_event_recorder::{EventTypeFilter, EventTypeFilterRef}; + use common_procedure::{ + ChildSubmissionOutcome, EventContext, EventTrigger, Procedure, ProcedureId, ProcedureState, + RetryPhase, + }; + use serde_json::{Value, json}; + + use super::*; + use crate::reconciliation::event::RECONCILE_CATALOG_EVENT_TYPE; + use crate::test_util::{MockDatanodeManager, new_ddl_context}; + + struct DatabaseEventHarness { + procedure_id: ProcedureId, + lifecycle_state: ProcedureState, + event_type_filter: EventTypeFilterRef, + } + + impl DatabaseEventHarness { + fn all() -> Self { + Self { + procedure_id: ProcedureId::random(), + lifecycle_state: ProcedureState::Running, + event_type_filter: Arc::new(EventTypeFilter::All), + } + } + + fn selected(event_types: impl IntoIterator) -> Self { + Self { + event_type_filter: Arc::new(EventTypeFilter::Only( + event_types.into_iter().map(str::to_string).collect(), + )), + ..Self::all() + } + } + + fn event( + &self, + procedure: &dyn Procedure, + trigger: EventTrigger, + ) -> Option> { + procedure.event(&EventContext { + procedure_id: self.procedure_id, + lifecycle_state: &self.lifecycle_state, + trigger, + event_type_filter: self.event_type_filter.clone(), + event_context: None, + }) + } + } + + #[test] + fn database_submitted_events_cover_root_and_child_intent() { + let events = DatabaseEventHarness::all(); + let root = test_procedure(false); + let child = test_procedure(true); + for (procedure, is_subprocedure) in [(&root, false), (&child, true)] { + let submitted = events.event(procedure, EventTrigger::Submitted).unwrap(); + assert_eq!(submitted.event_type(), RECONCILE_DATABASE_EVENT_TYPE); + assert_eq!( + submitted.json_payload().unwrap(), + json!({ + "version": 1, + "resolve_strategy": "use_metasrv", + "fail_fast": false, + "parallelism": 8, + "is_subprocedure": is_subprocedure, + }) + ); + } + } + + #[test] + fn database_non_terminal_lifecycle_events_have_null_payloads() { + let events = DatabaseEventHarness::all(); + let mut procedure = test_procedure(true); + procedure.context.volatile_ctx.metrics = populated_metrics(); + + for trigger in [ + EventTrigger::Recovered, + EventTrigger::ChildSubmitted { + procedure_id: ProcedureId::random(), + outcome: ChildSubmissionOutcome::Accepted, + }, + EventTrigger::Retrying { + phase: RetryPhase::Execute, + attempt: 2, + }, + EventTrigger::RollingBack, + ] { + assert_eq!( + events + .event(&procedure, trigger) + .unwrap() + .json_payload() + .unwrap(), + Value::Null + ); + } + } + + #[test] + fn database_terminal_events_report_existing_metrics() { + let events = DatabaseEventHarness::all(); + let mut procedure = test_procedure(true); + procedure.context.volatile_ctx.metrics = populated_metrics(); + + for (trigger, complete) in [ + (EventTrigger::Succeeded, true), + (EventTrigger::Failed, false), + (EventTrigger::Poisoned, false), + ] { + assert_eq!( + events + .event(&procedure, trigger) + .unwrap() + .json_payload() + .unwrap(), + json!({ + "version": 1, + "complete": complete, + "processed_table_count": 8, + "succeeded_table_count": 5, + "failed_table_count": 3, + "succeeded_subprocedure_count": 3, + "failed_subprocedure_count": 2, + }) + ); + } + } + + #[test] + fn database_event_filtering_uses_the_database_event_type() { + let procedure = test_procedure(false); + assert!( + DatabaseEventHarness::selected([RECONCILE_DATABASE_EVENT_TYPE]) + .event(&procedure, EventTrigger::Submitted) + .is_some() + ); + assert!( + DatabaseEventHarness::selected([RECONCILE_CATALOG_EVENT_TYPE]) + .event(&procedure, EventTrigger::Submitted) + .is_none() + ); + assert!( + DatabaseEventHarness::selected([]) + .event(&procedure, EventTrigger::Submitted) + .is_none() + ); + } + + #[test] + fn database_recovery_preserves_locator_and_resets_metrics() { + let events = DatabaseEventHarness::all(); + let mut procedure = test_procedure(false); + procedure.context.volatile_ctx.metrics = populated_metrics(); + let original_dump = procedure.dump().unwrap(); + procedure.context.volatile_ctx.metrics = ReconcileDatabaseMetrics::default(); + assert_eq!(procedure.dump().unwrap(), original_dump); + + let loaded = ReconcileDatabaseProcedure::from_json(test_context(), &original_dump).unwrap(); + assert_eq!(loaded.dump().unwrap(), original_dump); + assert_eq!( + events + .event(&loaded, EventTrigger::Recovered) + .unwrap() + .extra_rows() + .unwrap(), + events + .event(&procedure, EventTrigger::Submitted) + .unwrap() + .extra_rows() + .unwrap(), + ); + assert_eq!( + events + .event(&loaded, EventTrigger::Succeeded) + .unwrap() + .json_payload() + .unwrap(), + json!({ + "version": 1, + "complete": true, + "processed_table_count": 0, + "succeeded_table_count": 0, + "failed_table_count": 0, + "succeeded_subprocedure_count": 0, + "failed_subprocedure_count": 0, + }) + ); + } + + fn populated_metrics() -> ReconcileDatabaseMetrics { + ReconcileDatabaseMetrics { + succeeded_tables: 5, + failed_tables: 3, + succeeded_procedures: 3, + failed_procedures: 2, + } + } + + fn test_procedure(is_subprocedure: bool) -> ReconcileDatabaseProcedure { + ReconcileDatabaseProcedure::new( + test_context(), + "greptime".to_string(), + "public".to_string(), + false, + 8, + ResolveStrategy::UseMetasrv, + is_subprocedure, + ) + } + + fn test_context() -> Context { + let ddl_context = new_ddl_context(Arc::new(MockDatanodeManager::new(()))); + Context { + node_manager: ddl_context.node_manager, + table_metadata_manager: ddl_context.table_metadata_manager, + cache_invalidator: ddl_context.cache_invalidator, + } + } +} diff --git a/tests-integration/tests/reconciliation_event.rs b/tests-integration/tests/reconciliation_event.rs index ebd12f2af9..19eaea6ff2 100644 --- a/tests-integration/tests/reconciliation_event.rs +++ b/tests-integration/tests/reconciliation_event.rs @@ -27,6 +27,216 @@ const EVENTS_TABLE: &str = "greptime_private.events"; const CATALOG: &str = "greptime"; const DATABASE: &str = "reconciliation_event_database"; +#[tokio::test(flavor = "multi_thread")] +async fn test_catalog_and_database_reconciliation_events() { + common_telemetry::init_default_ut_logging(); + + let cluster = GreptimeDbClusterBuilder::new("catalog_database_reconciliation_events") + .await + .with_datanodes(1) + .build(true) + .await; + let frontend = cluster.fe_instance().clone(); + run_sql(frontend.as_ref(), &format!("CREATE DATABASE {DATABASE}")).await; + run_sql( + frontend.as_ref(), + &format!("CREATE TABLE {DATABASE}.metrics (ts TIMESTAMP TIME INDEX)"), + ) + .await; + + let catalog_procedure_id = cluster + .metasrv + .reconciliation_manager() + .reconcile_catalog(CATALOG.to_string(), ResolveStrategy::UseLatest, 1) + .await + .unwrap(); + let mut procedure_watcher = cluster + .metasrv + .procedure_manager() + .procedure_watcher(catalog_procedure_id) + .unwrap(); + watcher::wait(&mut procedure_watcher).await.unwrap(); + let catalog_procedure_id = catalog_procedure_id.to_string(); + + assert_eventually_eq( + &frontend, + &format!( + "SELECT count(*) AS event_count, \ + json_get_int(payload, 'version') AS version, \ + json_get_string(payload, 'resolve_strategy') AS resolve_strategy, \ + json_get_bool(payload, 'fail_fast') AS fail_fast, \ + json_get_int(payload, 'parallelism') AS parallelism \ + FROM {EVENTS_TABLE} \ + WHERE type = 'reconcile_catalog' AND procedure_id = '{catalog_procedure_id}' \ + AND json_get_string(procedure_trigger, 'type') = 'Submitted' \ + AND catalog_name = '{CATALOG}' AND schema_name IS NULL AND table_name IS NULL \ + AND table_id IS NULL AND physical_table_id IS NULL \ + GROUP BY json_get_int(payload, 'version'), \ + json_get_string(payload, 'resolve_strategy'), \ + json_get_bool(payload, 'fail_fast'), \ + json_get_int(payload, 'parallelism')" + ), + "\ ++-------------+---------+------------------+-----------+-------------+ +| event_count | version | resolve_strategy | fail_fast | parallelism | ++-------------+---------+------------------+-----------+-------------+ +| 1 | 1 | use_latest | false | 1 | ++-------------+---------+------------------+-----------+-------------+", + ) + .await; + + let catalog_result_payload = find_eventually_string( + &frontend, + &format!( + "SELECT json_to_string(payload) AS payload FROM {EVENTS_TABLE} \ + WHERE type = 'reconcile_catalog' AND procedure_id = '{catalog_procedure_id}' \ + AND json_get_string(procedure_trigger, 'type') = 'Succeeded' \ + AND catalog_name = '{CATALOG}' AND schema_name IS NULL AND table_name IS NULL \ + AND table_id IS NULL AND physical_table_id IS NULL LIMIT 1" + ), + "payload", + ) + .await; + let catalog_result: serde_json::Value = serde_json::from_str(&catalog_result_payload).unwrap(); + assert_eq!(catalog_result["version"], serde_json::json!(1)); + assert_eq!(catalog_result["complete"], serde_json::json!(true)); + let processed_database_count = catalog_result["processed_database_count"].as_u64().unwrap(); + let succeeded_database_count = catalog_result["succeeded_database_count"].as_u64().unwrap(); + let failed_database_count = catalog_result["failed_database_count"].as_u64().unwrap(); + assert_eq!( + processed_database_count, + succeeded_database_count + failed_database_count + ); + assert!(processed_database_count > 0); + assert_eq!(failed_database_count, 0); + + let database_procedure_id = find_eventually_string( + &frontend, + &format!( + "SELECT procedure_id FROM {EVENTS_TABLE} \ + WHERE type = 'reconcile_database' AND catalog_name = '{CATALOG}' \ + AND schema_name = '{DATABASE}' \ + AND json_get_string(procedure_trigger, 'type') = 'Submitted' LIMIT 1" + ), + "procedure_id", + ) + .await; + + assert_eventually_eq( + &frontend, + &format!( + "SELECT parent.procedure_id AS parent_procedure_id, \ + child.procedure_id AS child_procedure_id, \ + parent.type AS parent_event_type, \ + child.type AS child_event_type, \ + parent.catalog_name AS parent_catalog_name, \ + parent.schema_name AS parent_schema_name, \ + child.catalog_name AS child_catalog_name, \ + child.schema_name AS child_schema_name \ + FROM {EVENTS_TABLE} AS parent \ + JOIN {EVENTS_TABLE} AS child \ + ON json_get_string(parent.procedure_trigger, 'procedure_id') = child.procedure_id \ + WHERE parent.procedure_id = '{catalog_procedure_id}' \ + AND json_get_string(parent.procedure_trigger, 'type') = 'ChildSubmitted' \ + AND child.procedure_id = '{database_procedure_id}' \ + AND json_get_string(child.procedure_trigger, 'type') = 'Submitted'" + ), + &format!( + "\ ++--------------------------------------+--------------------------------------+-------------------+--------------------+---------------------+--------------------+--------------------+-------------------------------+ +| parent_procedure_id | child_procedure_id | parent_event_type | child_event_type | parent_catalog_name | parent_schema_name | child_catalog_name | child_schema_name | ++--------------------------------------+--------------------------------------+-------------------+--------------------+---------------------+--------------------+--------------------+-------------------------------+ +| {catalog_procedure_id} | {database_procedure_id} | reconcile_catalog | reconcile_database | greptime | | greptime | reconciliation_event_database | ++--------------------------------------+--------------------------------------+-------------------+--------------------+---------------------+--------------------+--------------------+-------------------------------+" + ), + ) + .await; + + assert_eventually_eq( + &frontend, + &format!( + "SELECT count(*) AS event_count, \ + json_get_string(procedure_trigger, 'outcome') AS outcome \ + FROM {EVENTS_TABLE} \ + WHERE type = 'reconcile_catalog' AND procedure_id = '{catalog_procedure_id}' \ + AND json_get_string(procedure_trigger, 'type') = 'ChildSubmitted' \ + AND json_get_string(procedure_trigger, 'procedure_id') = '{database_procedure_id}' \ + AND catalog_name = '{CATALOG}' AND schema_name IS NULL \ + AND json_is_null(payload) \ + GROUP BY json_get_string(procedure_trigger, 'outcome')" + ), + "\ ++-------------+----------+ +| event_count | outcome | ++-------------+----------+ +| 1 | Accepted | ++-------------+----------+", + ) + .await; + + assert_eventually_eq( + &frontend, + &format!( + "SELECT count(*) AS event_count, \ + json_get_int(payload, 'version') AS version, \ + json_get_string(payload, 'resolve_strategy') AS resolve_strategy, \ + json_get_bool(payload, 'fail_fast') AS fail_fast, \ + json_get_int(payload, 'parallelism') AS parallelism, \ + json_get_bool(payload, 'is_subprocedure') AS is_subprocedure \ + FROM {EVENTS_TABLE} \ + WHERE type = 'reconcile_database' AND procedure_id = '{database_procedure_id}' \ + AND json_get_string(procedure_trigger, 'type') = 'Submitted' \ + AND catalog_name = '{CATALOG}' AND schema_name = '{DATABASE}' \ + AND table_name IS NULL AND table_id IS NULL AND physical_table_id IS NULL \ + GROUP BY json_get_int(payload, 'version'), \ + json_get_string(payload, 'resolve_strategy'), \ + json_get_bool(payload, 'fail_fast'), \ + json_get_int(payload, 'parallelism'), \ + json_get_bool(payload, 'is_subprocedure')" + ), + "\ ++-------------+---------+------------------+-----------+-------------+-----------------+ +| event_count | version | resolve_strategy | fail_fast | parallelism | is_subprocedure | ++-------------+---------+------------------+-----------+-------------+-----------------+ +| 1 | 1 | use_latest | false | 1 | true | ++-------------+---------+------------------+-----------+-------------+-----------------+", + ) + .await; + + assert_eventually_eq( + &frontend, + &format!( + "SELECT count(*) AS event_count, \ + json_get_int(payload, 'version') AS version, \ + json_get_bool(payload, 'complete') AS complete, \ + json_get_int(payload, 'processed_table_count') AS processed_count, \ + json_get_int(payload, 'succeeded_table_count') AS succeeded_count, \ + json_get_int(payload, 'failed_table_count') AS failed_count, \ + json_get_int(payload, 'succeeded_subprocedure_count') AS succeeded_child_count, \ + json_get_int(payload, 'failed_subprocedure_count') AS failed_child_count \ + FROM {EVENTS_TABLE} \ + WHERE type = 'reconcile_database' AND procedure_id = '{database_procedure_id}' \ + AND json_get_string(procedure_trigger, 'type') = 'Succeeded' \ + AND catalog_name = '{CATALOG}' AND schema_name = '{DATABASE}' \ + AND table_name IS NULL AND table_id IS NULL AND physical_table_id IS NULL \ + GROUP BY json_get_int(payload, 'version'), \ + json_get_bool(payload, 'complete'), \ + json_get_int(payload, 'processed_table_count'), \ + json_get_int(payload, 'succeeded_table_count'), \ + json_get_int(payload, 'failed_table_count'), \ + json_get_int(payload, 'succeeded_subprocedure_count'), \ + json_get_int(payload, 'failed_subprocedure_count')" + ), + "\ ++-------------+---------+----------+-----------------+-----------------+--------------+-----------------------+--------------------+ +| event_count | version | complete | processed_count | succeeded_count | failed_count | succeeded_child_count | failed_child_count | ++-------------+---------+----------+-----------------+-----------------+--------------+-----------------------+--------------------+ +| 1 | 1 | true | 1 | 1 | 0 | 1 | 0 | ++-------------+---------+----------+-----------------+-----------------+--------------+-----------------------+--------------------+", + ) + .await; +} + #[tokio::test(flavor = "multi_thread")] async fn test_table_reconciliation_events() { common_telemetry::init_default_ut_logging();