diff --git a/Cargo.lock b/Cargo.lock index b18a6b9b8e..53ab970c47 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2379,6 +2379,7 @@ dependencies = [ "store-api", "tokio", "tokio-util", + "toml 0.8.23", ] [[package]] @@ -14004,6 +14005,7 @@ dependencies = [ "common-base", "common-config", "common-error", + "common-event-recorder", "common-macro", "common-memory-manager", "common-meta", @@ -14029,6 +14031,7 @@ dependencies = [ "store-api", "table", "tokio", + "toml 0.8.23", "url", ] diff --git a/config/config.md b/config/config.md index af4e5f61f8..92d0e77ed1 100644 --- a/config/config.md +++ b/config/config.md @@ -227,6 +227,9 @@ | `slow_query.sample_ratio` | Float | Unset | The sampling ratio of slow query log. The value should be in the range of (0, 1]. | | `tracing` | -- | -- | The tracing options. Only effect when compiled with `tokio-console` feature. | | `tracing.tokio_console_addr` | String | Unset | The tokio console address. | +| `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 type: `region_migration`.
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. | @@ -433,6 +436,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 type: `region_migration`.
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 33f3378cbc..6b50c36a35 100644 --- a/config/metasrv.example.toml +++ b/config/metasrv.example.toml @@ -306,6 +306,10 @@ create_topic_timeout = "30s" [event_recorder] ## TTL for the events table that will be used to store the events. Default is `90d`. ttl = "90d" +## Event types to record. Current available event type: `region_migration`. +## When omitted, all current and future event types are recorded. +## Set to an empty array to disable event recording. +#+ event_types = ["region_migration"] ## Configuration options for the stats persistence. [stats_persistence] diff --git a/config/standalone.example.toml b/config/standalone.example.toml index 1758fd6086..40db11b0d0 100644 --- a/config/standalone.example.toml +++ b/config/standalone.example.toml @@ -889,6 +889,15 @@ default_ratio = 1.0 ## @toml2docs:none-default #+ tokio_console_addr = "127.0.0.1" +## Configuration options for the event recorder. +[event_recorder] +## TTL for the events table that will be used to store the events. Default is `90d`. +ttl = "90d" +## Event types to record. Current available event type: `region_migration`. +## When omitted, all current and future event types are recorded. +## Set to an empty array to disable event recording. +#+ event_types = ["region_migration"] + ## The memory options. [memory] ## Whether to enable heap profiling activation during startup. diff --git a/src/cmd/src/standalone.rs b/src/cmd/src/standalone.rs index 2834a10d0c..5690c2d0bc 100644 --- a/src/cmd/src/standalone.rs +++ b/src/cmd/src/standalone.rs @@ -402,7 +402,7 @@ impl StartCommand { .metadata_kv_backend_creator .create(metadata_dir, &opts) .await?; - let procedure_manager = + let (procedure_manager, event_recorder_handle) = standalone::build_procedure_manager(kv_backend.clone(), opts.procedure); plugins::setup_standalone_plugins(&mut plugins, &plugin_opts, &opts, kv_backend.clone()) @@ -608,6 +608,8 @@ impl StartCommand { .context(error::StartFrontendSnafu)?; let fe_instance = Arc::new(fe_instance); + event_recorder_handle.install(fe_instance.event_recorder()); + // set the frontend client for flownode let grpc_handler = fe_instance.clone() as Arc; let weak_grpc_handler = Arc::downgrade(&grpc_handler); diff --git a/src/cmd/tests/load_config_test.rs b/src/cmd/tests/load_config_test.rs index 78c10eb14d..e07eadf9f4 100644 --- a/src/cmd/tests/load_config_test.rs +++ b/src/cmd/tests/load_config_test.rs @@ -411,6 +411,30 @@ fn test_load_heartbeat_env_vars_from_env() { }); } +#[test] +fn test_load_event_types_from_env() { + let env_prefix = "EVENT_TYPES_UT"; + let env_key = [env_prefix, "EVENT_RECORDER", "EVENT_TYPES"].join(ENV_VAR_SEP); + + temp_env::with_var(env_key, Some("region_migration"), || { + for event_types in [ + GreptimeOptions::::load_layered_options(None, env_prefix) + .unwrap() + .component + .event_recorder + .event_types, + GreptimeOptions::::load_layered_options(None, env_prefix) + .unwrap() + .component + .event_recorder + .event_types, + ] { + assert!(event_types.allows("region_migration")); + assert!(!event_types.allows("other_event")); + } + }); +} + #[test] fn test_load_metric_config_with_removed_sparse_primary_key_encoding() { // The `sparse_primary_key_encoding` option was removed from the metric diff --git a/src/common/event-recorder/Cargo.toml b/src/common/event-recorder/Cargo.toml index 33997115e4..7990cd3d18 100644 --- a/src/common/event-recorder/Cargo.toml +++ b/src/common/event-recorder/Cargo.toml @@ -4,6 +4,9 @@ version.workspace = true edition.workspace = true license.workspace = true +[features] +testing = [] + [dependencies] api.workspace = true async-trait.workspace = true @@ -23,5 +26,8 @@ store-api.workspace = true tokio.workspace = true tokio-util.workspace = true +[dev-dependencies] +toml.workspace = true + [lints] workspace = true diff --git a/src/common/event-recorder/src/event_table.rs b/src/common/event-recorder/src/event_table.rs new file mode 100644 index 0000000000..dd1fe6246c --- /dev/null +++ b/src/common/event-recorder/src/event_table.rs @@ -0,0 +1,236 @@ +// Copyright 2023 Greptime Team +// +// 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 api::v1::column_data_type_extension::TypeExt; +use api::v1::value::ValueData; +use api::v1::{ + ColumnDataType, ColumnDataTypeExtension, ColumnSchema, JsonTypeExtension, SemanticType, Value, +}; + +/// A canonical column in the shared event table. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct EventTableColumn { + name: &'static str, + datatype: ColumnDataType, + semantic_type: SemanticType, + json_binary: bool, +} + +impl EventTableColumn { + const fn new( + name: &'static str, + datatype: ColumnDataType, + semantic_type: SemanticType, + ) -> Self { + Self { + name, + datatype, + semantic_type, + json_binary: false, + } + } + + const fn json_binary( + name: &'static str, + datatype: ColumnDataType, + semantic_type: SemanticType, + ) -> Self { + Self { + name, + datatype, + semantic_type, + json_binary: true, + } + } + + /// Returns the canonical column name. + pub const fn name(&self) -> &'static str { + self.name + } + + /// Builds the canonical API schema for this column. + pub fn column_schema(&self) -> ColumnSchema { + ColumnSchema { + column_name: self.name.to_string(), + datatype: self.datatype.into(), + semantic_type: self.semantic_type.into(), + datatype_extension: self.json_binary.then(|| ColumnDataTypeExtension { + type_ext: Some(TypeExt::JsonType(JsonTypeExtension::JsonBinary.into())), + }), + ..Default::default() + } + } +} + +/// The canonical event type column. +pub const TYPE_COLUMN: EventTableColumn = + EventTableColumn::new("type", ColumnDataType::String, SemanticType::Tag); +/// The canonical event payload column. +pub const PAYLOAD_COLUMN: EventTableColumn = + EventTableColumn::json_binary("payload", ColumnDataType::Binary, SemanticType::Field); +/// The canonical event timestamp column. +pub const TIMESTAMP_COLUMN: EventTableColumn = EventTableColumn::new( + "timestamp", + ColumnDataType::TimestampNanosecond, + SemanticType::Timestamp, +); +/// The canonical procedure identifier envelope column. +pub const PROCEDURE_ID_COLUMN: EventTableColumn = + EventTableColumn::new("procedure_id", ColumnDataType::String, SemanticType::Field); +/// The canonical procedure state envelope column. +pub const PROCEDURE_STATE_COLUMN: EventTableColumn = EventTableColumn::new( + "procedure_state", + ColumnDataType::String, + SemanticType::Field, +); +/// The canonical procedure error envelope column. +pub const PROCEDURE_ERROR_COLUMN: EventTableColumn = EventTableColumn::new( + "procedure_error", + ColumnDataType::String, + SemanticType::Field, +); +/// The canonical procedure trigger envelope column. +pub const PROCEDURE_TRIGGER_COLUMN: EventTableColumn = EventTableColumn::new( + "procedure_trigger", + ColumnDataType::String, + SemanticType::Field, +); +/// The canonical catalog name dimension. +pub const CATALOG_NAME_COLUMN: EventTableColumn = + EventTableColumn::new("catalog_name", ColumnDataType::String, SemanticType::Field); +/// The canonical schema name dimension. +pub const SCHEMA_NAME_COLUMN: EventTableColumn = + EventTableColumn::new("schema_name", ColumnDataType::String, SemanticType::Field); + +/// Builds API schemas from canonical event-table columns while preserving their order. +pub fn column_schemas<'a>( + columns: impl IntoIterator, +) -> Vec { + columns + .into_iter() + .map(EventTableColumn::column_schema) + .collect() +} + +/// Builds the canonical base schema for every recorded event. +pub fn base_column_schemas() -> Vec { + column_schemas([&TYPE_COLUMN, &PAYLOAD_COLUMN, &TIMESTAMP_COLUMN]) +} + +/// Builds the canonical procedure event envelope schema. +pub fn procedure_event_column_schemas() -> Vec { + column_schemas([ + &PROCEDURE_ID_COLUMN, + &PROCEDURE_STATE_COLUMN, + &PROCEDURE_ERROR_COLUMN, + &PROCEDURE_TRIGGER_COLUMN, + ]) +} + +/// Builds an API value from an optional typed value. +pub fn nullable_value(value: Option) -> Value { + Value { value_data: value } +} + +/// Builds a nullable API string value. +pub fn nullable_string(value: Option) -> Value +where + T: AsRef, +{ + nullable_value(value.map(|value| ValueData::StringValue(value.as_ref().to_string()))) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn base_schema_preserves_names_types_semantics_extensions_and_order() { + assert_eq!( + base_column_schemas(), + vec![ + ColumnSchema { + column_name: "type".to_string(), + datatype: ColumnDataType::String.into(), + semantic_type: SemanticType::Tag.into(), + ..Default::default() + }, + ColumnSchema { + column_name: "payload".to_string(), + datatype: ColumnDataType::Binary.into(), + semantic_type: SemanticType::Field.into(), + datatype_extension: Some(ColumnDataTypeExtension { + type_ext: Some(TypeExt::JsonType(JsonTypeExtension::JsonBinary.into())), + }), + ..Default::default() + }, + ColumnSchema { + column_name: "timestamp".to_string(), + datatype: ColumnDataType::TimestampNanosecond.into(), + semantic_type: SemanticType::Timestamp.into(), + ..Default::default() + }, + ] + ); + } + + #[test] + fn procedure_envelope_schema_preserves_names_types_semantics_and_order() { + assert_eq!( + procedure_event_column_schemas(), + [ + "procedure_id", + "procedure_state", + "procedure_error", + "procedure_trigger", + ] + .map(|column_name| ColumnSchema { + column_name: column_name.to_string(), + datatype: ColumnDataType::String.into(), + semantic_type: SemanticType::Field.into(), + ..Default::default() + }) + ); + } + + #[test] + fn shared_dimension_schema_preserves_names_types_semantics_and_order() { + assert_eq!( + column_schemas([&CATALOG_NAME_COLUMN, &SCHEMA_NAME_COLUMN]), + ["catalog_name", "schema_name"].map(|column_name| ColumnSchema { + column_name: column_name.to_string(), + datatype: ColumnDataType::String.into(), + semantic_type: SemanticType::Field.into(), + ..Default::default() + }) + ); + } + + #[test] + fn nullable_values_preserve_types_and_nulls() { + assert_eq!( + nullable_string(Some("catalog")), + Value { + value_data: Some(ValueData::StringValue("catalog".to_string())) + } + ); + assert_eq!(nullable_value(None), Value { value_data: None }); + assert_eq!( + nullable_value(Some(ValueData::BoolValue(true))), + Value { + value_data: Some(ValueData::BoolValue(true)) + } + ); + } +} diff --git a/src/common/event-recorder/src/lib.rs b/src/common/event-recorder/src/lib.rs index 292f3fee8c..3e62482498 100644 --- a/src/common/event-recorder/src/lib.rs +++ b/src/common/event-recorder/src/lib.rs @@ -15,6 +15,10 @@ #![feature(duration_constructors)] pub mod error; +pub mod event_table; pub mod recorder; +#[cfg(any(test, feature = "testing"))] +pub mod testing; + pub use recorder::*; diff --git a/src/common/event-recorder/src/recorder.rs b/src/common/event-recorder/src/recorder.rs index eb014c638f..d7da4eca9b 100644 --- a/src/common/event-recorder/src/recorder.rs +++ b/src/common/event-recorder/src/recorder.rs @@ -13,17 +13,13 @@ // limitations under the License. use std::any::Any; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::fmt::Debug; use std::sync::Arc; use std::time::Duration; -use api::v1::column_data_type_extension::TypeExt; use api::v1::value::ValueData; -use api::v1::{ - ColumnDataType, ColumnDataTypeExtension, ColumnSchema, JsonTypeExtension, Row, - RowInsertRequest, RowInsertRequests, Rows, SemanticType, -}; +use api::v1::{ColumnSchema, Row, RowInsertRequest, RowInsertRequests, Rows}; use async_trait::async_trait; use backon::{BackoffBuilder, ExponentialBuilder}; use common_telemetry::{debug, error, info, warn}; @@ -38,20 +34,75 @@ use tokio::time::sleep; use tokio_util::sync::CancellationToken; use crate::error::{MismatchedSchemaSnafu, Result}; +use crate::event_table::{PAYLOAD_COLUMN, TIMESTAMP_COLUMN, TYPE_COLUMN, base_column_schemas}; /// The default table name for storing the events. pub const DEFAULT_EVENTS_TABLE_NAME: &str = "events"; /// The column name for the event type. -pub const EVENTS_TABLE_TYPE_COLUMN_NAME: &str = "type"; +pub const EVENTS_TABLE_TYPE_COLUMN_NAME: &str = TYPE_COLUMN.name(); /// The column name for the event payload. -pub const EVENTS_TABLE_PAYLOAD_COLUMN_NAME: &str = "payload"; +pub const EVENTS_TABLE_PAYLOAD_COLUMN_NAME: &str = PAYLOAD_COLUMN.name(); /// The column name for the event timestamp. -pub const EVENTS_TABLE_TIMESTAMP_COLUMN_NAME: &str = "timestamp"; +pub const EVENTS_TABLE_TIMESTAMP_COLUMN_NAME: &str = TIMESTAMP_COLUMN.name(); /// EventRecorderRef is the reference to the event recorder. pub type EventRecorderRef = Arc; +/// A shared event-type filter used by event producers and recorders. +pub type EventTypeFilterRef = Arc; + +/// Restricts the event types that are recorded. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub enum EventTypeFilter { + /// Records all current and future event types. + #[default] + All, + /// Records only the event types in the set. + Only(HashSet), +} + +impl EventTypeFilter { + /// Returns whether the filter retains `event_type`. + pub fn allows(&self, event_type: &str) -> bool { + match self { + Self::All => true, + Self::Only(event_types) => event_types.contains(event_type), + } + } +} + +fn deserialize_event_types<'de, D>( + deserializer: D, +) -> std::result::Result +where + D: serde::Deserializer<'de>, +{ + HashSet::::deserialize(deserializer) + .map(|event_types| Arc::new(EventTypeFilter::Only(event_types))) +} + +fn serialize_event_types( + event_types: &EventTypeFilterRef, + serializer: S, +) -> std::result::Result +where + S: serde::Serializer, +{ + match event_types.as_ref() { + EventTypeFilter::All => serializer.serialize_none(), + EventTypeFilter::Only(event_types) => { + let mut event_types = event_types.iter().collect::>(); + event_types.sort_unstable(); + event_types.serialize(serializer) + } + } +} + +fn event_type_filter_is_all(event_types: &EventTypeFilterRef) -> bool { + matches!(event_types.as_ref(), EventTypeFilter::All) +} + /// The time interval for flushing batched events to the event handler. pub const DEFAULT_FLUSH_INTERVAL_SECONDS: Duration = Duration::from_secs(5); /// The default TTL(90 days) for the events table. @@ -131,31 +182,10 @@ pub fn build_row_inserts_request(events: &[&Box]) -> Result = Vec::with_capacity(3 + event.extra_schema().len()); - schema.extend(vec![ - ColumnSchema { - column_name: EVENTS_TABLE_TYPE_COLUMN_NAME.to_string(), - datatype: ColumnDataType::String.into(), - semantic_type: SemanticType::Tag.into(), - ..Default::default() - }, - ColumnSchema { - column_name: EVENTS_TABLE_PAYLOAD_COLUMN_NAME.to_string(), - datatype: ColumnDataType::Binary as i32, - semantic_type: SemanticType::Field as i32, - datatype_extension: Some(ColumnDataTypeExtension { - type_ext: Some(TypeExt::JsonType(JsonTypeExtension::JsonBinary.into())), - }), - ..Default::default() - }, - ColumnSchema { - column_name: EVENTS_TABLE_TIMESTAMP_COLUMN_NAME.to_string(), - datatype: ColumnDataType::TimestampNanosecond.into(), - semantic_type: SemanticType::Timestamp.into(), - ..Default::default() - }, - ]); - schema.extend(event.extra_schema()); + let extra_schema = event.extra_schema(); + let mut schema: Vec = Vec::with_capacity(3 + extra_schema.len()); + schema.extend(base_column_schemas()); + schema.extend(extra_schema); let mut rows: Vec = Vec::with_capacity(events.len()); for event in events { @@ -202,6 +232,9 @@ pub trait EventRecorder: Send + Sync + Debug + 'static { /// Records an event for persistence and processing by [EventHandler]. fn record(&self, event: Box); + /// Returns the event types accepted by this recorder. + fn event_type_filter(&self) -> EventTypeFilterRef; + /// Cancels the event recorder. fn close(&self); } @@ -246,14 +279,27 @@ pub trait EventHandler: Send + Sync + 'static { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct EventRecorderOptions { /// TTL for the events table that will be used to store the events. - #[serde(with = "humantime_serde")] + #[serde(default = "default_events_table_ttl", with = "humantime_serde")] pub ttl: Duration, + /// Event types that the recorder retains. When omitted, all event types are retained. + #[serde( + default, + deserialize_with = "deserialize_event_types", + serialize_with = "serialize_event_types", + skip_serializing_if = "event_type_filter_is_all" + )] + pub event_types: EventTypeFilterRef, +} + +fn default_events_table_ttl() -> Duration { + DEFAULT_EVENTS_TABLE_TTL } impl Default for EventRecorderOptions { fn default() -> Self { Self { ttl: DEFAULT_EVENTS_TABLE_TTL, + event_types: Arc::new(EventTypeFilter::All), } } } @@ -263,6 +309,8 @@ impl Default for EventRecorderOptions { pub struct EventRecorderImpl { // The channel to send the events to the background processor. tx: Sender>, + // The event types this recorder accepts before sending to the background processor. + event_types: EventTypeFilterRef, // The cancel token to cancel the background processor. cancel_token: CancellationToken, // The background processor to process the events. @@ -271,11 +319,20 @@ pub struct EventRecorderImpl { impl EventRecorderImpl { pub fn new(event_handler: Box) -> Self { + Self::with_event_type_filter(event_handler, Arc::new(EventTypeFilter::All)) + } + + /// Creates an event recorder with an event-type filter. + pub fn with_event_type_filter( + event_handler: Box, + event_types: EventTypeFilterRef, + ) -> Self { let (tx, rx) = channel(DEFAULT_CHANNEL_SIZE); let cancel_token = CancellationToken::new(); let mut recorder = Self { tx, + event_types, handle: None, cancel_token: cancel_token.clone(), }; @@ -302,11 +359,19 @@ impl EventRecorderImpl { impl EventRecorder for EventRecorderImpl { // Accepts an event and send it to the background handler. fn record(&self, event: Box) { + if !self.event_types.allows(event.event_type()) { + return; + } + if let Err(e) = self.tx.try_send(event) { error!("Failed to send event to the background processor: {}", e); } } + fn event_type_filter(&self) -> EventTypeFilterRef { + self.event_types.clone() + } + // Closes the event recorder. It will stop the background processor and flush the buffer. fn close(&self) { self.cancel_token.cancel(); @@ -438,6 +503,9 @@ impl EventProcessor { #[cfg(test)] mod tests { + use std::collections::HashSet; + use std::sync::atomic::{AtomicUsize, Ordering}; + use serde_json::json; use super::*; @@ -479,6 +547,84 @@ mod tests { } } + #[test] + fn test_event_type_filter_defaults_to_all() { + let options = toml::from_str::("ttl = '90d'").unwrap(); + + assert!(options.event_types.allows("slow_query")); + assert!(options.event_types.allows("future_event")); + } + + #[test] + fn test_event_type_filter_deserializes_explicit_empty_array() { + let options = + toml::from_str::("ttl = '90d'\nevent_types = []").unwrap(); + + assert_eq!( + options.event_types.as_ref(), + &EventTypeFilter::Only(HashSet::new()) + ); + } + + #[test] + fn test_event_recorder_options_default_ttl_for_partial_table() { + let options = toml::from_str::("event_types = []").unwrap(); + + assert_eq!(DEFAULT_EVENTS_TABLE_TTL, options.ttl); + assert_eq!( + options.event_types.as_ref(), + &EventTypeFilter::Only(HashSet::new()) + ); + } + + #[test] + fn test_event_type_filter_deserializes_selected_types() { + let options = toml::from_str::( + "ttl = '90d'\nevent_types = ['create_database']", + ) + .unwrap(); + + assert!(options.event_types.allows("create_database")); + assert!(!options.event_types.allows("drop_database")); + } + + struct CountingEventHandler { + count: Arc, + } + + #[async_trait] + impl EventHandler for CountingEventHandler { + async fn handle(&self, _events: &[Box]) -> Result<()> { + self.count.fetch_add(1, Ordering::Relaxed); + Ok(()) + } + } + + #[tokio::test] + async fn test_event_recorder_rejects_filtered_event_before_queueing() { + let count = Arc::new(AtomicUsize::new(0)); + let event_type_filter = Arc::new(EventTypeFilter::Only(HashSet::new())); + let mut event_recorder = EventRecorderImpl::with_event_type_filter( + Box::new(CountingEventHandler { + count: count.clone(), + }), + event_type_filter.clone(), + ); + + assert!(Arc::ptr_eq( + &event_type_filter, + &event_recorder.event_type_filter() + )); + + event_recorder.record(Box::new(TestEvent {})); + event_recorder.close(); + + if let Some(handle) = event_recorder.handle.take() { + assert!(handle.await.is_ok()); + } + assert_eq!(count.load(Ordering::Relaxed), 0); + } + #[tokio::test] async fn test_event_recorder() { let mut event_recorder = EventRecorderImpl::new(Box::new(TestEventHandlerImpl {})); diff --git a/src/common/event-recorder/src/testing.rs b/src/common/event-recorder/src/testing.rs new file mode 100644 index 0000000000..345c8b56bc --- /dev/null +++ b/src/common/event-recorder/src/testing.rs @@ -0,0 +1,77 @@ +// Copyright 2023 Greptime Team +// +// 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 api::v1::{ColumnSchema, Row}; + +use crate::Event; + +/// Asserts the type, additional schema, and additional rows of an event. +pub fn assert_event_contract( + event: &E, + expected_event_type: &str, + expected_schema: &[ColumnSchema], + expected_rows: &[Row], +) where + E: Event + ?Sized, +{ + assert_eq!(event.event_type(), expected_event_type); + assert_eq!(event.extra_schema(), expected_schema); + assert_eq!(event.extra_rows().expect("event rows"), expected_rows); +} + +#[cfg(test)] +mod tests { + use std::any::Any; + + use api::v1::Row; + use api::v1::value::ValueData; + + use super::*; + use crate::error::Result; + use crate::event_table::{CATALOG_NAME_COLUMN, column_schemas, nullable_string}; + + #[derive(Debug)] + struct TestEvent; + + impl Event for TestEvent { + fn event_type(&self) -> &str { + "test_event_table" + } + + fn extra_schema(&self) -> Vec { + column_schemas([&CATALOG_NAME_COLUMN]) + } + + fn extra_rows(&self) -> Result> { + Ok(vec![Row { + values: vec![nullable_string(Some("value"))], + }]) + } + + fn as_any(&self) -> &dyn Any { + self + } + } + + #[test] + fn event_contract_assertions_cover_type_schema_and_rows() { + let event = TestEvent; + let schema = vec![CATALOG_NAME_COLUMN.column_schema()]; + let rows = vec![Row { + values: vec![ValueData::StringValue("value".to_string()).into()], + }]; + + assert_event_contract(&event, "test_event_table", &schema, &rows); + } +} diff --git a/src/common/procedure/src/event.rs b/src/common/procedure/src/event.rs index 9de3bc389c..e7a5269847 100644 --- a/src/common/procedure/src/event.rs +++ b/src/common/procedure/src/event.rs @@ -15,17 +15,21 @@ use std::any::Any; use api::v1::value::ValueData; -use api::v1::{ColumnDataType, ColumnSchema, Row, SemanticType}; +use api::v1::{ColumnSchema, Row}; use common_event_recorder::Event; use common_event_recorder::error::Result; +use common_event_recorder::event_table::{ + PROCEDURE_ERROR_COLUMN, PROCEDURE_ID_COLUMN, PROCEDURE_STATE_COLUMN, PROCEDURE_TRIGGER_COLUMN, + procedure_event_column_schemas, +}; use common_time::timestamp::{TimeUnit, Timestamp}; use crate::{EventTrigger, ProcedureId, ProcedureState}; -pub const EVENTS_TABLE_PROCEDURE_ID_COLUMN_NAME: &str = "procedure_id"; -pub const EVENTS_TABLE_PROCEDURE_STATE_COLUMN_NAME: &str = "procedure_state"; -pub const EVENTS_TABLE_PROCEDURE_ERROR_COLUMN_NAME: &str = "procedure_error"; -pub const EVENTS_TABLE_PROCEDURE_TRIGGER_COLUMN_NAME: &str = "procedure_trigger"; +pub const EVENTS_TABLE_PROCEDURE_ID_COLUMN_NAME: &str = PROCEDURE_ID_COLUMN.name(); +pub const EVENTS_TABLE_PROCEDURE_STATE_COLUMN_NAME: &str = PROCEDURE_STATE_COLUMN.name(); +pub const EVENTS_TABLE_PROCEDURE_ERROR_COLUMN_NAME: &str = PROCEDURE_ERROR_COLUMN.name(); +pub const EVENTS_TABLE_PROCEDURE_TRIGGER_COLUMN_NAME: &str = PROCEDURE_TRIGGER_COLUMN.name(); /// `ProcedureEvent` represents an event emitted by a procedure during its execution lifecycle. #[derive(Debug)] @@ -73,32 +77,7 @@ impl Event for ProcedureEvent { } fn extra_schema(&self) -> Vec { - let mut schema = vec![ - ColumnSchema { - column_name: EVENTS_TABLE_PROCEDURE_ID_COLUMN_NAME.to_string(), - datatype: ColumnDataType::String.into(), - semantic_type: SemanticType::Field.into(), - ..Default::default() - }, - ColumnSchema { - column_name: EVENTS_TABLE_PROCEDURE_STATE_COLUMN_NAME.to_string(), - datatype: ColumnDataType::String.into(), - semantic_type: SemanticType::Field.into(), - ..Default::default() - }, - ColumnSchema { - column_name: EVENTS_TABLE_PROCEDURE_ERROR_COLUMN_NAME.to_string(), - datatype: ColumnDataType::String.into(), - semantic_type: SemanticType::Field.into(), - ..Default::default() - }, - ColumnSchema { - column_name: EVENTS_TABLE_PROCEDURE_TRIGGER_COLUMN_NAME.to_string(), - datatype: ColumnDataType::String.into(), - semantic_type: SemanticType::Field.into(), - ..Default::default() - }, - ]; + let mut schema = procedure_event_column_schemas(); schema.append(&mut self.internal_event.extra_schema()); schema } @@ -140,11 +119,15 @@ impl Event for ProcedureEvent { #[cfg(test)] mod tests { + use std::sync::Arc; + use api::v1::value::ValueData; - use api::v1::{ColumnDataType, ColumnSchema, Row, SemanticType}; + use api::v1::{ColumnDataType, ColumnSchema, Row, SemanticType, Value}; + use common_error::mock::MockError; + use common_error::status_code::StatusCode; use common_event_recorder::Event; - use crate::{EventTrigger, ProcedureEvent, ProcedureId, ProcedureState}; + use crate::{Error, EventTrigger, ProcedureEvent, ProcedureId, ProcedureState}; #[derive(Debug)] struct TestEvent; @@ -171,6 +154,9 @@ mod tests { Row { values: vec![ValueData::StringValue("test_event2".to_string()).into()], }, + Row { + values: vec![Value { value_data: None }], + }, ]) } @@ -180,7 +166,79 @@ mod tests { } #[test] - fn test_procedure_event_extra_rows() { + fn procedure_event_extra_rows_preserve_envelope_values_and_internal_nulls() { + let procedure_id = ProcedureId::parse_str("00000000-0000-0000-0000-000000000001").unwrap(); + let procedure_event = ProcedureEvent::new( + procedure_id, + Box::new(TestEvent {}), + ProcedureState::Running, + EventTrigger::Submitted, + ); + + let procedure_event_extra_rows = procedure_event.extra_rows().unwrap(); + assert_eq!( + procedure_event_extra_rows, + vec![ + Row { + values: vec![ + ValueData::StringValue(procedure_id.to_string()).into(), + ValueData::StringValue("Running".to_string()).into(), + ValueData::StringValue(String::new()).into(), + ValueData::StringValue("Submitted".to_string()).into(), + ValueData::StringValue("test_event1".to_string()).into(), + ], + }, + Row { + values: vec![ + ValueData::StringValue(procedure_id.to_string()).into(), + ValueData::StringValue("Running".to_string()).into(), + ValueData::StringValue(String::new()).into(), + ValueData::StringValue("Submitted".to_string()).into(), + ValueData::StringValue("test_event2".to_string()).into(), + ], + }, + Row { + values: vec![ + ValueData::StringValue(procedure_id.to_string()).into(), + ValueData::StringValue("Running".to_string()).into(), + ValueData::StringValue(String::new()).into(), + ValueData::StringValue("Submitted".to_string()).into(), + Value { value_data: None }, + ], + }, + ] + ); + } + + #[test] + fn procedure_event_extra_rows_include_error_for_failed_state() { + let error = Arc::new(Error::external(MockError::new(StatusCode::Unexpected))); + let procedure_event = ProcedureEvent::new( + ProcedureId::parse_str("00000000-0000-0000-0000-000000000001").unwrap(), + Box::new(TestEvent {}), + ProcedureState::failed(error.clone()), + EventTrigger::Failed, + ); + + let procedure_event_extra_rows = procedure_event.extra_rows().unwrap(); + + assert_eq!(procedure_event_extra_rows.len(), 3); + assert_eq!( + procedure_event_extra_rows[0].values[1], + ValueData::StringValue("Failed".to_string()).into() + ); + assert_eq!( + procedure_event_extra_rows[0].values[2], + ValueData::StringValue(format!("{error:?}")).into() + ); + assert_eq!( + procedure_event_extra_rows[0].values[3], + ValueData::StringValue("Failed".to_string()).into() + ); + } + + #[test] + fn test_procedure_event_extra_schema() { let procedure_event = ProcedureEvent::new( ProcedureId::random(), Box::new(TestEvent {}), @@ -188,21 +246,21 @@ mod tests { EventTrigger::Submitted, ); - let procedure_event_extra_rows = procedure_event.extra_rows().unwrap(); - assert_eq!(procedure_event_extra_rows.len(), 2); - assert_eq!(procedure_event_extra_rows[0].values.len(), 5); assert_eq!( - procedure_event_extra_rows[0].values[3], - ValueData::StringValue("Submitted".to_string()).into() - ); - assert_eq!( - procedure_event_extra_rows[0].values[4], - ValueData::StringValue("test_event1".to_string()).into() - ); - assert_eq!(procedure_event_extra_rows[1].values.len(), 5); - assert_eq!( - procedure_event_extra_rows[1].values[4], - ValueData::StringValue("test_event2".to_string()).into() + procedure_event.extra_schema(), + [ + "procedure_id", + "procedure_state", + "procedure_error", + "procedure_trigger", + "test_event_column", + ] + .map(|column_name| ColumnSchema { + column_name: column_name.to_string(), + datatype: ColumnDataType::String.into(), + semantic_type: SemanticType::Field.into(), + ..Default::default() + }) ); } diff --git a/src/common/procedure/src/local.rs b/src/common/procedure/src/local.rs index c704b6f8c0..f70065e01e 100644 --- a/src/common/procedure/src/local.rs +++ b/src/common/procedure/src/local.rs @@ -631,6 +631,26 @@ impl Default for ManagerConfig { type PauseAwareRef = Arc; +struct EventRecorderConfig { + recorder: Option, +} + +/// A delayed configuration handle for procedure lifecycle event recording. +#[derive(Clone)] +pub struct EventRecorderHandle(Arc>); + +impl EventRecorderHandle { + fn new(recorder: Option) -> Self { + Self(Arc::new(Mutex::new(EventRecorderConfig { recorder }))) + } + + /// Installs the recorder used by subsequently submitted procedures. + pub fn install(&self, recorder: EventRecorderRef) { + let mut config = self.0.lock().unwrap(); + config.recorder = Some(recorder); + } +} + #[async_trait] pub trait PauseAware: Send + Sync { /// Returns true if the procedure manager is paused. @@ -647,7 +667,7 @@ pub struct LocalManager { remove_outdated_meta_task: TokioMutex>>, config: ManagerConfig, pause_aware: Option, - event_recorder: Option, + event_recorder: EventRecorderHandle, } impl LocalManager { @@ -669,10 +689,15 @@ impl LocalManager { remove_outdated_meta_task: TokioMutex::new(None), config, pause_aware, - event_recorder, + event_recorder: EventRecorderHandle::new(event_recorder), } } + /// Returns the handle used to configure procedure lifecycle event recording. + pub fn event_recorder_handle(&self) -> EventRecorderHandle { + self.event_recorder.clone() + } + /// Build remove outedated meta task pub fn build_remove_outdated_meta_task(&self) -> RepeatedTask { RepeatedTask::new( @@ -703,6 +728,7 @@ impl LocalManager { procedure.poison_keys(), procedure.type_name(), )); + let event_recorder = self.event_recorder.0.lock().unwrap(); let runner = Runner { meta: meta.clone(), procedure, @@ -713,7 +739,7 @@ impl LocalManager { .with_max_times(self.max_retry_times), store: self.procedure_store.clone(), rolling_back: false, - event_recorder: self.event_recorder.clone(), + event_recorder: event_recorder.recorder.clone(), execute_retry_attempt: 0, rollback_retry_attempt: 0, }; @@ -996,12 +1022,13 @@ pub(crate) mod test_util { #[cfg(test)] mod tests { use std::assert_matches; + use std::collections::HashSet; use std::sync::Mutex; use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering}; use common_error::mock::MockError; use common_error::status_code::StatusCode; - use common_event_recorder::{Event, EventRecorder}; + use common_event_recorder::{Event, EventRecorder, EventTypeFilter, EventTypeFilterRef}; use common_test_util::temp_dir::create_temp_dir; use tokio::sync::oneshot; use tokio::time::timeout; @@ -1017,12 +1044,20 @@ mod tests { ManagerContext::new(poison_manager) } - #[derive(Debug, Default)] + #[derive(Debug)] struct CapturingEventRecorder { events: Mutex>>, + event_type_filter: EventTypeFilterRef, } impl CapturingEventRecorder { + fn with_event_type_filter(event_type_filter: EventTypeFilterRef) -> Self { + Self { + events: Mutex::new(vec![]), + event_type_filter, + } + } + fn triggers(&self) -> Vec { self.events .lock() @@ -1045,9 +1080,19 @@ mod tests { self.events.lock().unwrap().push(event); } + fn event_type_filter(&self) -> EventTypeFilterRef { + self.event_type_filter.clone() + } + fn close(&self) {} } + impl Default for CapturingEventRecorder { + fn default() -> Self { + Self::with_event_type_filter(Arc::new(EventTypeFilter::All)) + } + } + #[derive(Debug)] struct TestProcedureEvent; @@ -1234,6 +1279,72 @@ mod tests { } } + struct FilterCapturingProcedure { + captured_filter: Arc>>, + } + + #[async_trait] + impl Procedure for FilterCapturingProcedure { + fn type_name(&self) -> &str { + "FilterCapturingProcedure" + } + + async fn execute(&mut self, _: &Context) -> Result { + Ok(Status::done()) + } + + fn dump(&self) -> Result { + Ok(String::new()) + } + + fn lock_key(&self) -> LockKey { + LockKey::default() + } + + fn event(&self, ctx: &EventContext<'_>) -> Option> { + *self.captured_filter.lock().unwrap() = Some(ctx.event_type_filter.clone()); + Some(Box::new(TestProcedureEvent)) + } + } + + #[tokio::test] + async fn test_event_filter_is_shared_with_procedure_context() { + let dir = create_temp_dir("shared_event_filter"); + let state_store = Arc::new(ObjectStateStore::new(test_util::new_object_store(&dir))); + let poison_manager = Arc::new(InMemoryPoisonStore::new()); + let event_type_filter = Arc::new(EventTypeFilter::Only(HashSet::from([String::from( + "test_procedure", + )]))); + let event_recorder = Arc::new(CapturingEventRecorder::with_event_type_filter( + event_type_filter.clone(), + )); + let captured_filter = Arc::new(Mutex::new(None)); + let manager = LocalManager::new( + ManagerConfig::default(), + state_store, + poison_manager, + None, + None, + ); + manager.event_recorder_handle().install(event_recorder); + manager.manager_ctx.start(); + + manager + .submit(ProcedureWithId { + id: ProcedureId::random(), + procedure: Box::new(FilterCapturingProcedure { + captured_filter: captured_filter.clone(), + }), + }) + .await + .unwrap(); + + let captured_filter = captured_filter.lock().unwrap().clone().unwrap(); + assert!(Arc::ptr_eq(&event_type_filter, &captured_filter)); + assert!(captured_filter.allows("test_procedure")); + assert!(!captured_filter.allows("other_procedure")); + } + #[tokio::test] async fn test_fresh_submission_emits_submitted_event() { let dir = create_temp_dir("fresh_submission_event"); @@ -1245,8 +1356,11 @@ mod tests { state_store, poison_manager, None, - Some(event_recorder.clone()), + None, ); + manager + .event_recorder_handle() + .install(event_recorder.clone()); manager.manager_ctx.start(); manager @@ -1275,8 +1389,11 @@ mod tests { state_store, poison_manager, None, - Some(event_recorder.clone()), + None, ); + manager + .event_recorder_handle() + .install(event_recorder.clone()); manager.manager_ctx.start(); manager .register_loader("ProcedureToLoad", ProcedureToLoad::loader()) @@ -1308,6 +1425,28 @@ mod tests { assert!(!event_recorder.triggers().contains(&EventTrigger::Submitted)); } + #[tokio::test] + async fn test_event_recorder_handle_installs_after_start() { + let dir = create_temp_dir("set_event_recorder_after_start"); + let state_store = Arc::new(ObjectStateStore::new(test_util::new_object_store(&dir))); + let poison_manager = Arc::new(InMemoryPoisonStore::new()); + let manager = LocalManager::new( + ManagerConfig::default(), + state_store, + poison_manager, + None, + None, + ); + + manager.start().await.unwrap(); + + manager + .event_recorder_handle() + .install(Arc::new(CapturingEventRecorder::default())); + + manager.stop().await.unwrap(); + } + #[derive(Debug)] struct BlockingProcedure { started_tx: Option>, diff --git a/src/common/procedure/src/local/runner.rs b/src/common/procedure/src/local/runner.rs index dff7990f82..3c5cacb329 100644 --- a/src/common/procedure/src/local/runner.rs +++ b/src/common/procedure/src/local/runner.rs @@ -776,6 +776,7 @@ impl Runner { procedure_id: self.meta.id, lifecycle_state: &state, trigger: trigger.clone(), + event_type_filter: recorder.event_type_filter(), }; if let Some(event) = self.procedure.event(&context) { recorder.record(Box::new(crate::event::ProcedureEvent::new( diff --git a/src/common/procedure/src/procedure.rs b/src/common/procedure/src/procedure.rs index 6d07b160b6..fac945453b 100644 --- a/src/common/procedure/src/procedure.rs +++ b/src/common/procedure/src/procedure.rs @@ -19,7 +19,7 @@ use std::str::FromStr; use std::sync::Arc; use async_trait::async_trait; -use common_event_recorder::Event; +use common_event_recorder::{Event, EventTypeFilterRef}; use serde::{Deserialize, Serialize}; use smallvec::{SmallVec, smallvec}; use snafu::{ResultExt, Snafu}; @@ -250,6 +250,8 @@ pub struct EventContext<'a> { pub lifecycle_state: &'a ProcedureState, /// Lifecycle action that caused the event hook to be called. pub trigger: EventTrigger, + /// Event types retained by the configured recorder. + pub event_type_filter: EventTypeFilterRef, } /// Lifecycle action that causes the framework to invoke [`Procedure::event`]. @@ -756,6 +758,7 @@ mod tests { procedure_id: ProcedureId::random(), lifecycle_state: &state, trigger: EventTrigger::Succeeded, + event_type_filter: Arc::new(common_event_recorder::EventTypeFilter::All), }; assert!(DefaultEventProcedure.event(&context).is_none()); diff --git a/src/frontend/src/events.rs b/src/frontend/src/events.rs index 75e4dc15db..859d45c098 100644 --- a/src/frontend/src/events.rs +++ b/src/frontend/src/events.rs @@ -23,55 +23,28 @@ use common_event_recorder::{ DEFAULT_COMPACTION_TIME_WINDOW, Event, EventHandler, build_row_inserts_request, group_events_by_type, }; -use common_frontend::slow_query_event::SLOW_QUERY_EVENT_TYPE; -use datafusion::common::HashMap; use operator::statement::{InserterImpl, StatementExecutorRef}; use snafu::ResultExt; /// EventHandlerImpl is the default event handler implementation in frontend. pub struct EventHandlerImpl { - default_inserter: Box, - /// The inserters for the event types. - inserters: HashMap>, + inserter: Box, } impl EventHandlerImpl { /// Create a new EventHandlerImpl. - pub fn new( - statement_executor: StatementExecutorRef, - slow_query_ttl: Duration, - global_ttl: Duration, - ) -> Self { + pub fn new(statement_executor: StatementExecutorRef, ttl: Duration) -> Self { Self { - inserters: HashMap::from([( - SLOW_QUERY_EVENT_TYPE.to_string(), - Box::new(InserterImpl::new( - statement_executor.clone(), - Some(InsertOptions { - ttl: slow_query_ttl, - append_mode: true, - twcs_compaction_time_window: Some(DEFAULT_COMPACTION_TIME_WINDOW), - }), - )) as _, - )]), - default_inserter: Box::new(InserterImpl::new( - statement_executor.clone(), + inserter: Box::new(InserterImpl::new( + statement_executor, Some(InsertOptions { - ttl: global_ttl, + ttl, append_mode: true, twcs_compaction_time_window: Some(DEFAULT_COMPACTION_TIME_WINDOW), }), )), } } - - fn inserter(&self, event_type: &str) -> &dyn Inserter { - let Some(inserter) = self.inserters.get(event_type) else { - return self.default_inserter.as_ref(); - }; - - inserter.as_ref() - } } const DEFAULT_CONTEXT: Context = Context { @@ -84,11 +57,10 @@ impl EventHandler for EventHandlerImpl { async fn handle(&self, events: &[Box]) -> Result<()> { let event_groups = group_events_by_type(events); - for (event_type, events) in event_groups { + for (_, events) in event_groups { let requests = build_row_inserts_request(&events)?; - let inserter = self.inserter(event_type); - inserter + self.inserter .insert_rows(&DEFAULT_CONTEXT, requests) .await .map_err(BoxedError::new) diff --git a/src/frontend/src/instance.rs b/src/frontend/src/instance.rs index a49b0d60b2..07fce9feb6 100644 --- a/src/frontend/src/instance.rs +++ b/src/frontend/src/instance.rs @@ -128,7 +128,8 @@ pub struct Instance { inserter: InserterRef, deleter: DeleterRef, table_metadata_manager: TableMetadataManagerRef, - event_recorder: Option, + event_recorder: EventRecorderRef, + slow_query_recorder: EventRecorderRef, process_manager: ProcessManagerRef, slow_query_options: SlowQueryOptions, influxdb_default_merge_mode: InfluxdbMergeMode, @@ -175,6 +176,11 @@ impl Instance { &self.process_manager } + /// Returns the event recorder configured for this frontend instance. + pub fn event_recorder(&self) -> EventRecorderRef { + self.event_recorder.clone() + } + pub fn node_manager(&self) -> &NodeManagerRef { self.inserter.node_manager() } @@ -246,16 +252,14 @@ impl Instance { return None; } - self.event_recorder.clone().map(|event_recorder| { - SlowQueryTimer::new( - CatalogQueryStatement::Sql(stmt.clone()), - schema_name, - self.slow_query_options.threshold, - self.slow_query_options.sample_ratio, - self.slow_query_options.record_type, - event_recorder, - ) - }) + Some(SlowQueryTimer::new( + CatalogQueryStatement::Sql(stmt.clone()), + schema_name, + self.slow_query_options.threshold, + self.slow_query_options.sample_ratio, + self.slow_query_options.record_type, + self.slow_query_recorder.clone(), + )) } async fn query_statement(&self, stmt: Statement, query_ctx: QueryContextRef) -> Result { @@ -798,20 +802,16 @@ impl Instance { let catalog_name = query_ctx.current_catalog().to_string(); let schema_name = query_ctx.current_schema(); let slow_query_timer = if plan_is_readonly { - self.slow_query_options - .enable - .then(|| self.event_recorder.clone()) - .flatten() - .map(|event_recorder| { - SlowQueryTimer::new( - CatalogQueryStatement::Plan(query.clone()), - schema_name.clone(), - self.slow_query_options.threshold, - self.slow_query_options.sample_ratio, - self.slow_query_options.record_type, - event_recorder, - ) - }) + self.slow_query_options.enable.then(|| { + SlowQueryTimer::new( + CatalogQueryStatement::Plan(query.clone()), + schema_name.clone(), + self.slow_query_options.threshold, + self.slow_query_options.sample_ratio, + self.slow_query_options.record_type, + self.slow_query_recorder.clone(), + ) + }) } else { None }; @@ -1185,21 +1185,16 @@ impl PrometheusHandler for Instance { }; let raw_query = query_statement.to_string(); - let slow_query_timer = self - .slow_query_options - .enable - .then(|| self.event_recorder.clone()) - .flatten() - .map(|event_recorder| { - SlowQueryTimer::new( - query_statement, - query_ctx.current_schema(), - self.slow_query_options.threshold, - self.slow_query_options.sample_ratio, - self.slow_query_options.record_type, - event_recorder, - ) - }); + let slow_query_timer = self.slow_query_options.enable.then(|| { + SlowQueryTimer::new( + query_statement, + query_ctx.current_schema(), + self.slow_query_options.threshold, + self.slow_query_options.sample_ratio, + self.slow_query_options.record_type, + self.slow_query_recorder.clone(), + ) + }); let ticket = self.process_manager.register_query( query_ctx.current_catalog().to_string(), @@ -2262,6 +2257,17 @@ mod tests { }) } + #[tokio::test] + async fn test_event_recorder_is_exposed() -> TestResult<()> { + let instance = + test_instance_with_tables(test_table(1024, "source")?, test_table(1025, "target")?) + .await?; + + let _event_recorder = instance.event_recorder(); + + Ok(()) + } + #[tokio::test] async fn test_target_independent_checker_skips_target_resolution() -> TestResult<()> { let physical_table = "physical_metric"; diff --git a/src/frontend/src/instance/builder.rs b/src/frontend/src/instance/builder.rs index d197ec5292..32395efd2d 100644 --- a/src/frontend/src/instance/builder.rs +++ b/src/frontend/src/instance/builder.rs @@ -304,11 +304,16 @@ impl FrontendBuilder { plugins.insert::(statement_executor.clone()); - let event_recorder = Arc::new(EventRecorderImpl::new(Box::new(EventHandlerImpl::new( - statement_executor.clone(), - self.options.slow_query.ttl, - self.options.event_recorder.ttl, - )))); + let slow_query_recorder = Arc::new(EventRecorderImpl::new(Box::new( + EventHandlerImpl::new(statement_executor.clone(), self.options.slow_query.ttl), + ))); + let event_recorder = Arc::new(EventRecorderImpl::with_event_type_filter( + Box::new(EventHandlerImpl::new( + statement_executor.clone(), + self.options.event_recorder.ttl, + )), + self.options.event_recorder.event_types.clone(), + )); Ok(Instance { frontend_peer_addr, @@ -320,7 +325,8 @@ impl FrontendBuilder { inserter, deleter, table_metadata_manager, - event_recorder: Some(event_recorder), + event_recorder, + slow_query_recorder, process_manager, otlp_metrics_table_legacy_cache: DashMap::new(), slow_query_options: self.options.slow_query.clone(), diff --git a/src/meta-srv/src/metasrv.rs b/src/meta-srv/src/metasrv.rs index 7480b8cb50..bbf6c64ca5 100644 --- a/src/meta-srv/src/metasrv.rs +++ b/src/meta-srv/src/metasrv.rs @@ -408,7 +408,11 @@ impl Default for MetasrvOptions { impl Configurable for MetasrvOptions { fn env_list_keys() -> Option<&'static [&'static str]> { - Some(&["wal.broker_endpoints", "store_addrs"]) + Some(&[ + "wal.broker_endpoints", + "store_addrs", + "event_recorder.event_types", + ]) } } @@ -942,6 +946,9 @@ impl Metasrv { #[cfg(test)] mod tests { + use common_event_recorder::EventTypeFilter; + + use super::*; use crate::metasrv::MetasrvNodeInfo; #[test] @@ -953,4 +960,20 @@ mod tests { assert_eq!(node_info.git_commit, "1234567890"); assert_eq!(node_info.start_time_ms, 1715145600); } + + #[test] + fn test_metasrv_event_recorder_options_preserve_event_type_filter_semantics() { + let all = MetasrvOptions::default().event_recorder; + let none: EventRecorderOptions = toml::from_str("ttl = '90d'\nevent_types = []").unwrap(); + let selected: EventRecorderOptions = + toml::from_str("ttl = '90d'\nevent_types = ['create_database']").unwrap(); + + assert!(all.event_types.allows("future_event")); + assert_eq!( + none.event_types.as_ref(), + &EventTypeFilter::Only(Default::default()) + ); + assert!(selected.event_types.allows("create_database")); + assert!(!selected.event_types.allows("drop_database")); + } } diff --git a/src/meta-srv/src/metasrv/builder.rs b/src/meta-srv/src/metasrv/builder.rs index dd5e318b94..64e99f540a 100644 --- a/src/meta-srv/src/metasrv/builder.rs +++ b/src/meta-srv/src/metasrv/builder.rs @@ -247,9 +247,10 @@ impl MetasrvBuilder { }), )); // Builds the event recorder to record important events and persist them as the system table. - let event_recorder = Arc::new(EventRecorderImpl::new(Box::new(EventHandlerImpl::new( - event_inserter, - )))); + let event_recorder = Arc::new(EventRecorderImpl::with_event_type_filter( + Box::new(EventHandlerImpl::new(event_inserter)), + options.event_recorder.event_types.clone(), + )); let selector = selector.unwrap_or_else(|| Arc::new(LeaseBasedSelector)); let pushers = Pushers::default(); diff --git a/src/meta-srv/src/procedure/region_migration.rs b/src/meta-srv/src/procedure/region_migration.rs index 8dd6aac271..2b58d08136 100644 --- a/src/meta-srv/src/procedure/region_migration.rs +++ b/src/meta-srv/src/procedure/region_migration.rs @@ -64,7 +64,7 @@ use tokio::time::Instant; use self::migration_start::RegionMigrationStart; use crate::error::{self, Result}; -use crate::events::region_migration_event::RegionMigrationEvent; +use crate::events::region_migration_event::{REGION_MIGRATION_EVENT_TYPE, RegionMigrationEvent}; use crate::metrics::{ METRIC_META_REGION_MIGRATION_ERROR, METRIC_META_REGION_MIGRATION_EXECUTE, METRIC_META_REGION_MIGRATION_STAGE_ELAPSED, @@ -964,7 +964,11 @@ impl Procedure for RegionMigrationProcedure { LockKey::new(self.context.persistent_ctx.lock_key()) } - fn event(&self, _ctx: &EventContext<'_>) -> Option> { + fn event(&self, ctx: &EventContext<'_>) -> Option> { + if !ctx.event_type_filter.allows(REGION_MIGRATION_EVENT_TYPE) { + return None; + } + Some(Box::new(RegionMigrationEvent::from_persistent_ctx( &self.context.persistent_ctx, ))) @@ -1051,6 +1055,7 @@ mod tests { procedure_id: common_procedure::ProcedureId::random(), lifecycle_state: &state, trigger, + event_type_filter: Arc::new(common_event_recorder::EventTypeFilter::All), }) .unwrap(); assert_eq!(event.event_type(), "region_migration"); diff --git a/src/standalone/Cargo.toml b/src/standalone/Cargo.toml index ac7d412c82..16479991f4 100644 --- a/src/standalone/Cargo.toml +++ b/src/standalone/Cargo.toml @@ -14,6 +14,7 @@ client.workspace = true common-base.workspace = true common-config.workspace = true common-error.workspace = true +common-event-recorder.workspace = true common-macro.workspace = true common-memory-manager.workspace = true common-meta.workspace = true @@ -40,3 +41,6 @@ store-api.workspace = true table.workspace = true tokio.workspace = true url.workspace = true + +[dev-dependencies] +toml.workspace = true diff --git a/src/standalone/src/options.rs b/src/standalone/src/options.rs index bf1000fd58..637715013a 100644 --- a/src/standalone/src/options.rs +++ b/src/standalone/src/options.rs @@ -14,6 +14,7 @@ use common_base::readable_size::ReadableSize; use common_config::{Configurable, KvBackendConfig}; +use common_event_recorder::EventRecorderOptions; use common_memory_manager::OnExhaustedPolicy; use common_options::memory::MemoryOptions; use common_telemetry::logging::{LoggingOptions, SlowQueryOptions, TracingOptions}; @@ -71,6 +72,8 @@ pub struct StandaloneOptions { pub slow_query: SlowQueryOptions, pub query: QueryOptions, pub memory: MemoryOptions, + /// The event recorder options. + pub event_recorder: EventRecorderOptions, /// Environment variable keys to read and report in heartbeat messages. pub heartbeat_env_vars: Vec, } @@ -109,6 +112,7 @@ impl Default for StandaloneOptions { slow_query: SlowQueryOptions::default(), query: QueryOptions::default(), memory: MemoryOptions::default(), + event_recorder: EventRecorderOptions::default(), heartbeat_env_vars: vec![], } } @@ -116,7 +120,11 @@ impl Default for StandaloneOptions { impl Configurable for StandaloneOptions { fn env_list_keys() -> Option<&'static [&'static str]> { - Some(&["heartbeat_env_vars", "wal.broker_endpoints"]) + Some(&[ + "heartbeat_env_vars", + "wal.broker_endpoints", + "event_recorder.event_types", + ]) } } @@ -150,6 +158,7 @@ impl StandaloneOptions { logging: cloned_opts.logging, user_provider: cloned_opts.user_provider, slow_query: cloned_opts.slow_query, + event_recorder: cloned_opts.event_recorder, heartbeat_env_vars: cloned_opts.heartbeat_env_vars.clone(), ..Default::default() } @@ -183,3 +192,39 @@ impl StandaloneOptions { } } } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use common_event_recorder::EventTypeFilter; + + use super::*; + + #[test] + fn test_event_recorder_event_types_preserve_filter_semantics() { + let all: StandaloneOptions = toml::from_str("").unwrap(); + let none: StandaloneOptions = toml::from_str("[event_recorder]\nevent_types = []").unwrap(); + let selected: StandaloneOptions = + toml::from_str("[event_recorder]\nevent_types = ['create_database']").unwrap(); + + assert!(all.event_recorder.event_types.allows("future_event")); + assert_eq!( + none.event_recorder.event_types.as_ref(), + &EventTypeFilter::Only(Default::default()) + ); + assert!( + selected + .event_recorder + .event_types + .allows("create_database") + ); + assert!(!selected.event_recorder.event_types.allows("drop_database")); + + let frontend_options = selected.frontend_options(); + assert!(Arc::ptr_eq( + &selected.event_recorder.event_types, + &frontend_options.event_recorder.event_types, + )); + } +} diff --git a/src/standalone/src/procedure.rs b/src/standalone/src/procedure.rs index 853221e698..a232fba950 100644 --- a/src/standalone/src/procedure.rs +++ b/src/standalone/src/procedure.rs @@ -21,7 +21,7 @@ use common_meta::ddl_manager::{RepartitionProcedureFactory, RepartitionSource}; use common_meta::key::runtime_switch::RuntimeSwitchManager; use common_meta::kv_backend::KvBackendRef; use common_meta::state_store::KvStateStore; -use common_procedure::local::{LocalManager, ManagerConfig}; +use common_procedure::local::{EventRecorderHandle, LocalManager, ManagerConfig}; use common_procedure::options::ProcedureConfig; use common_procedure::{BoxedProcedure, ProcedureManagerRef}; use store_api::storage::TableId; @@ -33,7 +33,7 @@ use crate::error::NoSupportRepartitionProcedureSnafu; pub fn build_procedure_manager( kv_backend: KvBackendRef, procedure_config: ProcedureConfig, -) -> ProcedureManagerRef { +) -> (ProcedureManagerRef, EventRecorderHandle) { let kv_state_store = Arc::new(KvStateStore::new(kv_backend.clone())); let manager_config = ManagerConfig { @@ -43,13 +43,16 @@ pub fn build_procedure_manager( ..Default::default() }; let runtime_switch_manager = Arc::new(RuntimeSwitchManager::new(kv_backend)); - Arc::new(LocalManager::new( + let procedure_manager = LocalManager::new( manager_config, kv_state_store.clone(), kv_state_store, Some(runtime_switch_manager), None, - )) + ); + let event_recorder_handle = procedure_manager.event_recorder_handle(); + + (Arc::new(procedure_manager), event_recorder_handle) } /// No-op implementation of [`RepartitionProcedureFactory`] for standalone mode. diff --git a/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs index b4d645126a..2aff75dac5 100644 --- a/tests-integration/src/standalone.rs +++ b/tests-integration/src/standalone.rs @@ -40,6 +40,7 @@ use common_meta::region_registry::LeaderRegionRegistry; use common_meta::sequence::SequenceBuilder; use common_meta::wal_provider::build_wal_provider; use common_procedure::ProcedureManagerRef; +use common_procedure::local::EventRecorderHandle; use common_procedure::options::ProcedureConfig; use common_telemetry::logging::SlowQueryOptions; use common_wal::config::{DatanodeWalConfig, MetasrvWalConfig}; @@ -64,6 +65,7 @@ pub struct GreptimeDbStandalone { // Used in rebuild. pub kv_backend: KvBackendRef, pub procedure_manager: ProcedureManagerRef, + pub event_recorder_handle: EventRecorderHandle, } impl GreptimeDbStandalone { @@ -157,6 +159,7 @@ impl GreptimeDbStandaloneBuilder { guard: TestGuard, opts: StandaloneOptions, procedure_manager: ProcedureManagerRef, + event_recorder_handle: EventRecorderHandle, register_procedure_loaders: bool, ) -> GreptimeDbStandalone { let plugins = self.plugin.clone().unwrap_or_default(); @@ -281,6 +284,8 @@ impl GreptimeDbStandaloneBuilder { .unwrap(); let instance = Arc::new(instance); + event_recorder_handle.install(instance.event_recorder()); + // set the frontend client for flownode let grpc_handler = instance.clone() as Arc; let weak_grpc_handler = Arc::downgrade(&grpc_handler); @@ -324,6 +329,7 @@ impl GreptimeDbStandaloneBuilder { guard, kv_backend, procedure_manager, + event_recorder_handle, } } @@ -347,7 +353,7 @@ impl GreptimeDbStandaloneBuilder { kv_backend_config, ) .unwrap(); - let procedure_manager = + let (procedure_manager, event_recorder_handle) = standalone::build_procedure_manager(kv_backend.clone(), procedure_config); let standalone_opts = StandaloneOptions { @@ -361,7 +367,14 @@ impl GreptimeDbStandaloneBuilder { ..StandaloneOptions::default() }; - self.build_with(kv_backend, guard, standalone_opts, procedure_manager, true) - .await + self.build_with( + kv_backend, + guard, + standalone_opts, + procedure_manager, + event_recorder_handle, + true, + ) + .await } } diff --git a/tests-integration/src/tests/test_util.rs b/tests-integration/src/tests/test_util.rs index 9cb0ed5315..473cb3ce2b 100644 --- a/tests-integration/src/tests/test_util.rs +++ b/tests-integration/src/tests/test_util.rs @@ -142,11 +142,19 @@ impl MockInstanceBuilder { guard, kv_backend, procedure_manager, + event_recorder_handle, .. } = instance; MockInstanceImpl::Standalone( builder - .build_with(kv_backend, guard, opts, procedure_manager, false) + .build_with( + kv_backend, + guard, + opts, + procedure_manager, + event_recorder_handle, + false, + ) .await, ) } diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 99ef6bc59e..e460e11449 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -2271,6 +2271,9 @@ allow_query_fallback = false [memory] enable_heap_profiling = true + +[event_recorder] +ttl = "2months 29days 2h 52m 48s" "#, ) .trim()