rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-09-12 00:12:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(event-recorder): configure lifecycle event recording (#8648)

Browse Source 
* refactor(event-recorder): centralize event table helpers

Signed-off-by: WenyXu <wenymedia@gmail.com>

* feat(procedure): wire lifecycle event recorder

Signed-off-by: WenyXu <wenymedia@gmail.com>

* feat(event-recorder): filter events by type

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix(event-recorder): derive event type filter default

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix(event-recorder): decouple frontend filtering

Signed-off-by: WenyXu <wenymedia@gmail.com>

* chore: remove docs

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix(event-recorder): complete configuration support

Signed-off-by: WenyXu <wenymedia@gmail.com>

* refactor(event-recorder): centralize filter ownership

Signed-off-by: WenyXu <wenymedia@gmail.com>

* test(config): update event recorder snapshot

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix(frontend): decouple slow query event recorder

Signed-off-by: WenyXu <wenymedia@gmail.com>

* chore: apply suggestions

Signed-off-by: WenyXu <wenymedia@gmail.com>

---------

Signed-off-by: WenyXu <wenymedia@gmail.com>
This commit is contained in:
Weny Xu
2026-07-27 13:05:56 +00:00
committed by GitHub
parent 3e67c67607
commit 7344d47756
27 changed files with 993 additions and 188 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+3
View File
@@ -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",
]
config/config.md
+4
View File
@@ -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`.<br/>When omitted, all current and future event types are recorded.<br/>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.<br/>When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable<br/>is set to "prof:true,prof_active:false". The official image adds this env variable.<br/>Default is true. |
@@ -433,6 +436,7 @@
| `wal.create_topic_timeout` | String | `30s` | The timeout for creating a Kafka topic.<br/>**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`.<br/>When omitted, all current and future event types are recorded.<br/>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.<br/>Set to `0s` to disable stats persistence.<br/>Default is `0s`.<br/>If you want to enable stats persistence, set the TTL to a value greater than 0.<br/>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`.<br/>The minimum value is `10m`, if the value is less than `10m`, it will be overridden to `10m`. |
config/metasrv.example.toml
+4
View File
@@ -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]
config/standalone.example.toml
+9
View File
@@ -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.
src/cmd/src/standalone.rs
+3 -1
View File
@@ -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<dyn GrpcQueryHandlerWithBoxedError>;
let weak_grpc_handler = Arc::downgrade(&grpc_handler);
src/cmd/tests/load_config_test.rs
+24
View File
@@ -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::<MetasrvOptions>::load_layered_options(None, env_prefix)
.unwrap()
.component
.event_recorder
.event_types,
GreptimeOptions::<StandaloneOptions>::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
src/common/event-recorder/Cargo.toml
+6
View File
@@ -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
src/common/event-recorder/src/event_table.rs
+236
View File
@@ -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<Item = &'a EventTableColumn>,
) -> Vec<ColumnSchema> {
columns
.into_iter()
.map(EventTableColumn::column_schema)
.collect()
}
/// Builds the canonical base schema for every recorded event.
pub fn base_column_schemas() -> Vec<ColumnSchema> {
column_schemas([&TYPE_COLUMN, &PAYLOAD_COLUMN, &TIMESTAMP_COLUMN])
}
/// Builds the canonical procedure event envelope schema.
pub fn procedure_event_column_schemas() -> Vec<ColumnSchema> {
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<ValueData>) -> Value {
Value { value_data: value }
}
/// Builds a nullable API string value.
pub fn nullable_string<T>(value: Option<T>) -> Value
where
T: AsRef<str>,
{
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))
}
);
}
}
src/common/event-recorder/src/lib.rs
+4
View File
@@ -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::*;
src/common/event-recorder/src/recorder.rs
+181 -35
View File
@@ -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<dyn EventRecorder>;
/// A shared event-type filter used by event producers and recorders.
pub type EventTypeFilterRef = Arc<EventTypeFilter>;
/// 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<String>),
}
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<EventTypeFilterRef, D::Error>
where
D: serde::Deserializer<'de>,
{
HashSet::<String>::deserialize(deserializer)
.map(|event_types| Arc::new(EventTypeFilter::Only(event_types)))
}
fn serialize_event_types<S>(
event_types: &EventTypeFilterRef,
serializer: S,
) -> std::result::Result<S::Ok, S::Error>
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::<Vec<_>>();
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<dyn Event>]) -> Result<RowInsert
// We already validated the events, so it's safe to get the first event to build the schema for the RowInsertRequest.
let event = &events[0];
let mut schema: Vec<ColumnSchema> = 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<ColumnSchema> = Vec::with_capacity(3 + extra_schema.len());
schema.extend(base_column_schemas());
schema.extend(extra_schema);
let mut rows: Vec<Row> = 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<dyn Event>);
/// 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<Box<dyn Event>>,
// 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<dyn EventHandler>) -> 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<dyn EventHandler>,
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<dyn Event>) {
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::<EventRecorderOptions>("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::<EventRecorderOptions>("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::<EventRecorderOptions>("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::<EventRecorderOptions>(
"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<AtomicUsize>,
}
#[async_trait]
impl EventHandler for CountingEventHandler {
async fn handle(&self, _events: &[Box<dyn Event>]) -> 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 {}));
src/common/event-recorder/src/testing.rs
+77
View File
@@ -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<E>(
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<ColumnSchema> {
column_schemas([&CATALOG_NAME_COLUMN])
}
fn extra_rows(&self) -> Result<Vec<Row>> {
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);
}
}
src/common/procedure/src/event.rs
+106 -48
View File
@@ -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<ColumnSchema> {
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()
})
);
}
src/common/procedure/src/local.rs
+146 -7
View File
@@ -631,6 +631,26 @@ impl Default for ManagerConfig {
type PauseAwareRef = Arc<dyn PauseAware>;
struct EventRecorderConfig {
recorder: Option<EventRecorderRef>,
}
/// A delayed configuration handle for procedure lifecycle event recording.
#[derive(Clone)]
pub struct EventRecorderHandle(Arc<Mutex<EventRecorderConfig>>);
impl EventRecorderHandle {
fn new(recorder: Option<EventRecorderRef>) -> 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<Option<RepeatedTask<Error>>>,
config: ManagerConfig,
pause_aware: Option<PauseAwareRef>,
event_recorder: Option<EventRecorderRef>,
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<Error> {
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<Vec<Box<dyn Event>>>,
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<EventTrigger> {
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<Mutex<Option<EventTypeFilterRef>>>,
}
#[async_trait]
impl Procedure for FilterCapturingProcedure {
fn type_name(&self) -> &str {
"FilterCapturingProcedure"
}
async fn execute(&mut self, _: &Context) -> Result<Status> {
Ok(Status::done())
}
fn dump(&self) -> Result<String> {
Ok(String::new())
}
fn lock_key(&self) -> LockKey {
LockKey::default()
}
fn event(&self, ctx: &EventContext<'_>) -> Option<Box<dyn Event>> {
*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<oneshot::Sender<()>>,
src/common/procedure/src/local/runner.rs
+1
View File
@@ -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(
src/common/procedure/src/procedure.rs
+4 -1
View File
@@ -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());
src/frontend/src/events.rs
+7 -35
View File
@@ -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<dyn Inserter>,
/// The inserters for the event types.
inserters: HashMap<String, Box<dyn Inserter>>,
inserter: Box<dyn Inserter>,
}
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<dyn Event>]) -> 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)
src/frontend/src/instance.rs
+46 -40
View File
@@ -128,7 +128,8 @@ pub struct Instance {
inserter: InserterRef,
deleter: DeleterRef,
table_metadata_manager: TableMetadataManagerRef,
event_recorder: Option<EventRecorderRef>,
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<Output> {
@@ -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";
src/frontend/src/instance/builder.rs
+12 -6
View File
@@ -304,11 +304,16 @@ impl FrontendBuilder {
plugins.insert::<StatementExecutorRef>(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(),
src/meta-srv/src/metasrv.rs
+24 -1
View File
@@ -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"));
}
}
src/meta-srv/src/metasrv/builder.rs
+4 -3
View File
@@ -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();
src/meta-srv/src/procedure/region_migration.rs
+7 -2
View File
@@ -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<Box<dyn Event>> {
fn event(&self, ctx: &EventContext<'_>) -> Option<Box<dyn Event>> {
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");
src/standalone/Cargo.toml
+4
View File
@@ -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
src/standalone/src/options.rs
+46 -1
View File
@@ -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<String>,
}
@@ -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,
));
}
}
src/standalone/src/procedure.rs
+7 -4
View File
@@ -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.
tests-integration/src/standalone.rs
+16 -3
View File
@@ -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<dyn GrpcQueryHandlerWithBoxedError>;
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
}
}
tests-integration/src/tests/test_util.rs
+9 -1
View File
@@ -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,
)
}
tests-integration/tests/http.rs
+3
View File
@@ -2271,6 +2271,9 @@ allow_query_fallback = false
[memory]
enable_heap_profiling = true
[event_recorder]
ttl = "2months 29days 2h 52m 48s"
"#,
)
.trim()