From b72465e67931912a7d5f9c50fe62b39eb972fcfd Mon Sep 17 00:00:00 2001 From: discord9 Date: Thu, 25 Jun 2026 18:04:33 +0800 Subject: [PATCH] feat: add remote dynamic filter metrics (#8309) * feat: add remote dyn filter metrics Signed-off-by: discord9 * fix: unify remote dyn filter payload budget Signed-off-by: discord9 * style: import remote dyn filter metrics Signed-off-by: discord9 * fix: separate remote dyn filter unregister metrics Signed-off-by: discord9 * fix: rename remote dyn filter update outcome metric Signed-off-by: discord9 --------- Signed-off-by: discord9 --- src/common/query/src/request.rs | 5 +- .../request/initial_remote_dyn_filter_reg.rs | 11 ++- src/datanode/src/metrics.rs | 42 +++++++++ .../src/region_server/registrations.rs | 86 +++++++++++++++++-- .../src/region_server/remote_dyn_filter.rs | 5 +- src/query/src/dist_plan/dyn_filter_bridge.rs | 12 +-- .../dist_plan/remote_dyn_filter_registry.rs | 66 ++++++++++---- src/query/src/metrics.rs | 27 ++++++ 8 files changed, 210 insertions(+), 44 deletions(-) diff --git a/src/common/query/src/request.rs b/src/common/query/src/request.rs index 1a2a58b8ca8..6dc62b0a89f 100644 --- a/src/common/query/src/request.rs +++ b/src/common/query/src/request.rs @@ -36,13 +36,14 @@ use store_api::storage::RegionId; /// Current wire-format version for remote dynamic filter payload updates. pub use self::initial_remote_dyn_filter_reg::{ - INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY, - INITIAL_REMOTE_DYN_FILTER_REGS_MAX_TOTAL_PROTO_BYTES, InitialDynFilterReg, + INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY, InitialDynFilterReg, InitialDynFilterRegs, InitialDynFilterSnapshot, }; use crate::error::{DynFilterPayloadTooLargeSnafu, Error as CommonQueryError}; pub const DYN_FILTER_PROTOCOL_VERSION: u32 = 1; +/// Shared raw protobuf byte budget for remote dynamic filter payloads. +pub const REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES: usize = 512 * 1024; /// Serialized predicate payload for remote dynamic filter updates. /// diff --git a/src/common/query/src/request/initial_remote_dyn_filter_reg.rs b/src/common/query/src/request/initial_remote_dyn_filter_reg.rs index 4d73a5eef29..a4a00d5face 100644 --- a/src/common/query/src/request/initial_remote_dyn_filter_reg.rs +++ b/src/common/query/src/request/initial_remote_dyn_filter_reg.rs @@ -22,17 +22,16 @@ use datafusion_common::Result as DataFusionResult; use serde::{Deserialize, Serialize}; use crate::request::{ - DynFilterPayload, decode_physical_expr_from_bytes, encode_physical_expr_to_bytes, + DynFilterPayload, REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, decode_physical_expr_from_bytes, + encode_physical_expr_to_bytes, }; pub const INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY: &str = "initial_remote_dyn_filter_registrations"; pub const INITIAL_REMOTE_DYN_FILTER_REGS_MAX_COUNT: usize = 64; -/// Raw encoded registration byte budget for initial remote dynamic filter registrations. -/// -/// Counts proto payload bytes before JSON/base64 expansion, not the final extension size. -pub const INITIAL_REMOTE_DYN_FILTER_REGS_MAX_TOTAL_PROTO_BYTES: usize = 64 * 1024; +// Initial registration validation uses the shared remote dynamic filter payload budget. +// It counts raw protobuf payload bytes before JSON/base64 expansion, not the final extension size. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct InitialDynFilterRegs { #[serde(rename = "registrations")] @@ -58,7 +57,7 @@ impl InitialDynFilterRegs { pub fn validate_default_bounds(&self) -> Result<(), String> { self.validate_bounds( INITIAL_REMOTE_DYN_FILTER_REGS_MAX_COUNT, - INITIAL_REMOTE_DYN_FILTER_REGS_MAX_TOTAL_PROTO_BYTES, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, ) } diff --git a/src/datanode/src/metrics.rs b/src/datanode/src/metrics.rs index 4e763f5858b..74da5b153f4 100644 --- a/src/datanode/src/metrics.rs +++ b/src/datanode/src/metrics.rs @@ -91,4 +91,46 @@ lazy_static! { &[REGION_REQUEST_TYPE] ) .unwrap(); + + /// Remote dynamic filter update processing outcomes, labeled with the outcome. + pub static ref REMOTE_DYN_FILTER_UPDATE_OUTCOME_TOTAL: IntCounterVec = register_int_counter_vec!( + "greptime_datanode_remote_dyn_filter_update_outcome_total", + "remote dynamic filter update processing outcomes", + &["result"] + ) + .unwrap(); + + /// Remote dynamic filter update drops, labeled with drop reason. + pub static ref REMOTE_DYN_FILTER_UPDATE_DROP_TOTAL: IntCounterVec = register_int_counter_vec!( + "greptime_datanode_remote_dyn_filter_update_drop_total", + "remote dynamic filter update drops by reason", + &["reason"] + ) + .unwrap(); + + /// Remote dynamic filter unregister outcomes, labeled with the outcome. + pub static ref REMOTE_DYN_FILTER_UNREGISTER_TOTAL: IntCounterVec = register_int_counter_vec!( + "greptime_datanode_remote_dyn_filter_unregister_total", + "remote dynamic filter unregister outcomes", + &["result"] + ) + .unwrap(); + + /// Current count of remote dynamic filter registry entries. + pub static ref REMOTE_DYN_FILTER_REGISTRY_ENTRIES: IntGauge = register_int_gauge!( + "greptime_datanode_remote_dyn_filter_registry_entries", + "current remote dynamic filter registry entry count", + ) + .unwrap(); + + /// Remote dynamic filter accepted update payload bytes received. + pub static ref REMOTE_DYN_FILTER_PAYLOAD_BYTES: Histogram = register_histogram!( + "greptime_datanode_remote_dyn_filter_payload_bytes", + "remote dynamic filter accepted update payload bytes received", + vec![ + 128.0, 256.0, 512.0, 1024.0, 2048.0, 4096.0, 8192.0, 16384.0, 32768.0, + 65536.0, 131072.0, 262144.0, 524288.0, + ] + ) + .unwrap(); } diff --git a/src/datanode/src/region_server/registrations.rs b/src/datanode/src/region_server/registrations.rs index 92a1bb813c1..2eb86a9ff70 100644 --- a/src/datanode/src/region_server/registrations.rs +++ b/src/datanode/src/region_server/registrations.rs @@ -19,7 +19,7 @@ use std::sync::{Arc, Mutex, OnceLock}; use common_query::request::{ DynFilterPayload, INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY, InitialDynFilterReg, - InitialDynFilterRegs, + InitialDynFilterRegs, REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, }; use common_telemetry::warn; use dashmap::DashMap; @@ -32,7 +32,11 @@ use session::context::QueryContextRef; use session::query_id::QueryId; use store_api::storage::RegionId; -pub(super) const REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES: usize = 64 * 1024; +use crate::metrics::{ + REMOTE_DYN_FILTER_PAYLOAD_BYTES, REMOTE_DYN_FILTER_REGISTRY_ENTRIES, + REMOTE_DYN_FILTER_UNREGISTER_TOTAL, REMOTE_DYN_FILTER_UPDATE_DROP_TOTAL, + REMOTE_DYN_FILTER_UPDATE_OUTCOME_TOTAL, +}; type QueryRemoteDynFilterRegs = HashMap; @@ -112,6 +116,49 @@ pub(super) enum RemoteDynFilterUpdateOutcome { DecodeFailed, } +impl RemoteDynFilterUpdateOutcome { + pub fn as_str(&self) -> &'static str { + match self { + Self::MissingRegistration => "missing_registration", + Self::Buffered => "buffered", + Self::Applied => "applied", + Self::Idempotent => "idempotent", + Self::Stale => "stale", + Self::AlreadyComplete => "already_complete", + Self::PayloadTooLarge => "payload_too_large", + Self::DecodeFailed => "decode_failed", + } + } + + fn is_drop(&self) -> bool { + matches!( + self, + Self::MissingRegistration + | Self::Stale + | Self::AlreadyComplete + | Self::PayloadTooLarge + | Self::DecodeFailed + ) + } +} + +fn record_update_outcome(outcome: RemoteDynFilterUpdateOutcome) { + REMOTE_DYN_FILTER_UPDATE_OUTCOME_TOTAL + .with_label_values(&[outcome.as_str()]) + .inc(); + if outcome.is_drop() { + REMOTE_DYN_FILTER_UPDATE_DROP_TOTAL + .with_label_values(&[outcome.as_str()]) + .inc(); + } +} + +fn record_unregister_outcome(outcome: RemoteDynFilterUpdateOutcome) { + REMOTE_DYN_FILTER_UNREGISTER_TOTAL + .with_label_values(&[outcome.as_str()]) + .inc(); +} + #[derive(Debug, Clone)] struct PendingDynFilterUpdate { payload: Vec, @@ -455,6 +502,7 @@ pub(super) fn register_initial_dyn_filter_regs( PendingDynFilterUpdate::from_initial_reg(reg), region_id, )); + REMOTE_DYN_FILTER_REGISTRY_ENTRIES.inc(); registered_filter_ids.push(filter_id); } } @@ -501,15 +549,21 @@ pub(super) fn apply_remote_dyn_filter_update( payload.len(), REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES ); - return RemoteDynFilterUpdateOutcome::PayloadTooLarge; + let outcome = RemoteDynFilterUpdateOutcome::PayloadTooLarge; + record_update_outcome(outcome); + return outcome; } + REMOTE_DYN_FILTER_PAYLOAD_BYTES.observe(payload.len() as f64); + let Some(query_regs) = regs_by_query.get_query(query_id) else { warn!( "Ignored remote dynamic filter update without query registration, query_id: {}, filter_id: {}", query_id, filter_id ); - return RemoteDynFilterUpdateOutcome::MissingRegistration; + let outcome = RemoteDynFilterUpdateOutcome::MissingRegistration; + record_update_outcome(outcome); + return outcome; }; let mut query_regs = query_regs.lock().unwrap(); @@ -518,10 +572,15 @@ pub(super) fn apply_remote_dyn_filter_update( "Ignored remote dynamic filter update without filter registration, query_id: {}, filter_id: {}", query_id, filter_id ); - return RemoteDynFilterUpdateOutcome::MissingRegistration; + let outcome = RemoteDynFilterUpdateOutcome::MissingRegistration; + record_update_outcome(outcome); + return outcome; }; - registered.apply_or_buffer_update(payload, generation, is_complete) + let outcome = registered.apply_or_buffer_update(payload, generation, is_complete); + record_update_outcome(outcome); + + outcome } pub(super) fn unregister_remote_dyn_filter( @@ -534,7 +593,9 @@ pub(super) fn unregister_remote_dyn_filter( "Ignored remote dynamic filter unregister without query registration, query_id: {}, filter_id: {}", query_id, filter_id ); - return RemoteDynFilterUpdateOutcome::MissingRegistration; + let outcome = RemoteDynFilterUpdateOutcome::MissingRegistration; + record_unregister_outcome(outcome); + return outcome; }; let (registered, should_remove_query) = { @@ -544,8 +605,11 @@ pub(super) fn unregister_remote_dyn_filter( "Ignored remote dynamic filter unregister without filter registration, query_id: {}, filter_id: {}", query_id, filter_id ); - return RemoteDynFilterUpdateOutcome::MissingRegistration; + let outcome = RemoteDynFilterUpdateOutcome::MissingRegistration; + record_unregister_outcome(outcome); + return outcome; }; + REMOTE_DYN_FILTER_REGISTRY_ENTRIES.dec(); let should_remove_query = locked.is_empty(); (registered, should_remove_query) }; @@ -555,7 +619,10 @@ pub(super) fn unregister_remote_dyn_filter( regs_by_query.remove_query_if_empty(query_id, &query_regs); } - RemoteDynFilterUpdateOutcome::Applied + let outcome = RemoteDynFilterUpdateOutcome::Applied; + record_unregister_outcome(outcome); + + outcome } pub(super) fn remove_initial_dyn_filter_regs( @@ -583,6 +650,7 @@ pub(super) fn remove_initial_dyn_filter_regs( .unwrap_or(false); if should_remove_filter && let Some(registered) = locked.remove(filter_id) { + REMOTE_DYN_FILTER_REGISTRY_ENTRIES.dec(); removed_filters.push(registered); } } diff --git a/src/datanode/src/region_server/remote_dyn_filter.rs b/src/datanode/src/region_server/remote_dyn_filter.rs index f750cb896cd..1317dc05021 100644 --- a/src/datanode/src/region_server/remote_dyn_filter.rs +++ b/src/datanode/src/region_server/remote_dyn_filter.rs @@ -352,6 +352,7 @@ mod tests { use common_query::request::{ DynFilterPayload, INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY, InitialDynFilterReg, InitialDynFilterRegs, InitialDynFilterSnapshot, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, }; use common_recordbatch::RecordBatches; use datafusion::arrow::datatypes::Schema as ArrowSchema; @@ -370,9 +371,7 @@ mod tests { use store_api::storage::RegionId; use super::*; - use crate::region_server::registrations::{ - REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, RemoteDynFilterRegistry, - }; + use crate::region_server::registrations::RemoteDynFilterRegistry; use crate::tests::mock_region_server; #[derive(Debug, Clone)] diff --git a/src/query/src/dist_plan/dyn_filter_bridge.rs b/src/query/src/dist_plan/dyn_filter_bridge.rs index 546aa207782..2937924d5e7 100644 --- a/src/query/src/dist_plan/dyn_filter_bridge.rs +++ b/src/query/src/dist_plan/dyn_filter_bridge.rs @@ -16,9 +16,8 @@ use std::any::Any; use std::sync::Arc; use common_query::request::{ - DynFilterPayload, INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY, - INITIAL_REMOTE_DYN_FILTER_REGS_MAX_TOTAL_PROTO_BYTES, InitialDynFilterReg, - InitialDynFilterRegs, InitialDynFilterSnapshot, + DynFilterPayload, INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY, InitialDynFilterReg, + InitialDynFilterRegs, InitialDynFilterSnapshot, REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, }; use datafusion_common::Result; use datafusion_physical_expr::PhysicalExpr; @@ -169,7 +168,7 @@ fn initial_snapshot( let payload = match DynFilterPayload::from_datafusion_expr( ¤t, - INITIAL_REMOTE_DYN_FILTER_REGS_MAX_TOTAL_PROTO_BYTES, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, ) { Ok(payload) => payload, Err(error) => { @@ -485,10 +484,11 @@ mod tests { #[test] fn capture_remote_dyn_filters_for_pushdown_rejects_oversized_snapshots() { + let oversized_total_snapshot_bytes = REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES * 3 / 5; let parent_filters = vec![ - test_dyn_filter_with_snapshot_payload("host", 0, 40 * 1024) + test_dyn_filter_with_snapshot_payload("host", 0, oversized_total_snapshot_bytes) as Arc, - test_dyn_filter_with_snapshot_payload("pod", 1, 40 * 1024) + test_dyn_filter_with_snapshot_payload("pod", 1, oversized_total_snapshot_bytes) as Arc, ]; diff --git a/src/query/src/dist_plan/remote_dyn_filter_registry.rs b/src/query/src/dist_plan/remote_dyn_filter_registry.rs index 3d0bda9132b..fdb9d668045 100644 --- a/src/query/src/dist_plan/remote_dyn_filter_registry.rs +++ b/src/query/src/dist_plan/remote_dyn_filter_registry.rs @@ -18,7 +18,7 @@ use std::sync::{Arc, Mutex, RwLock, Weak}; use std::time::Duration; use api::v1::region::{RemoteDynFilterUnregister, RemoteDynFilterUpdate}; -use common_query::request::DynFilterPayload; +use common_query::request::{DynFilterPayload, REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES}; use common_runtime::spawn_global; use common_telemetry::{debug, warn}; use datafusion_physical_expr::PhysicalExpr; @@ -28,9 +28,12 @@ use store_api::storage::RegionId; use tokio::sync::{Notify, watch}; use crate::dist_plan::FilterId; +use crate::metrics::{ + REMOTE_DYN_FILTER_ENCODE_TOTAL, REMOTE_DYN_FILTER_PAYLOAD_BYTES, + REMOTE_DYN_FILTER_UPDATE_RPC_TOTAL, +}; use crate::region_query::RegionQueryHandlerRef; -const REMOTE_DYN_FILTER_UPDATE_PAYLOAD_MAX_BYTES: usize = 64 * 1024; const REMOTE_DYN_FILTER_RECONCILE_INTERVAL: Duration = Duration::from_secs(1); /// Bound best-effort RDF control RPCs so one bad subscriber cannot stall fanout. const REMOTE_DYN_FILTER_CONTROL_RPC_TIMEOUT: Duration = Duration::from_secs(10); @@ -387,20 +390,31 @@ async fn fanout_snapshot_for_query( let _ = entry.mark_generation_sent(generation); } - let payload = match DynFilterPayload::from_datafusion_expr( - ¤t, - REMOTE_DYN_FILTER_UPDATE_PAYLOAD_MAX_BYTES, - ) { - Ok(DynFilterPayload::Datafusion(payload)) => payload, - Ok(_) => { - warn!("Ignored unsupported remote dynamic filter producer payload"); - return true; - } - Err(error) => { - warn!(error; "Failed to encode remote dynamic filter producer snapshot"); - return true; - } - }; + let payload = + match DynFilterPayload::from_datafusion_expr(¤t, REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES) + { + Ok(DynFilterPayload::Datafusion(payload)) => { + REMOTE_DYN_FILTER_ENCODE_TOTAL + .with_label_values(&["success"]) + .inc(); + REMOTE_DYN_FILTER_PAYLOAD_BYTES.observe(payload.len() as f64); + payload + } + Ok(_) => { + REMOTE_DYN_FILTER_ENCODE_TOTAL + .with_label_values(&["unsupported"]) + .inc(); + warn!("Ignored unsupported remote dynamic filter producer payload"); + return true; + } + Err(error) => { + REMOTE_DYN_FILTER_ENCODE_TOTAL + .with_label_values(&["error"]) + .inc(); + warn!(error; "Failed to encode remote dynamic filter producer snapshot"); + return true; + } + }; fanout_update_for_query( query_id, @@ -456,6 +470,9 @@ async fn fanout_update_for_query( { ControlRpcResult::Ok(result) => { if let Err(error) = result { + REMOTE_DYN_FILTER_UPDATE_RPC_TOTAL + .with_label_values(&["error"]) + .inc(); warn!( error; "Failed to fan out remote dynamic filter update, query_id={}, filter_id={}, region_id={}", @@ -463,10 +480,23 @@ async fn fanout_update_for_query( filter_id, subscriber.region_id() ); + } else { + REMOTE_DYN_FILTER_UPDATE_RPC_TOTAL + .with_label_values(&["success"]) + .inc(); } } - ControlRpcResult::TimedOut => {} - ControlRpcResult::LifecycleClosed => return false, + ControlRpcResult::TimedOut => { + REMOTE_DYN_FILTER_UPDATE_RPC_TOTAL + .with_label_values(&["timeout"]) + .inc(); + } + ControlRpcResult::LifecycleClosed => { + REMOTE_DYN_FILTER_UPDATE_RPC_TOTAL + .with_label_values(&["cancelled"]) + .inc(); + return false; + } } } diff --git a/src/query/src/metrics.rs b/src/query/src/metrics.rs index bee436660c2..3eec4ef4c61 100644 --- a/src/query/src/metrics.rs +++ b/src/query/src/metrics.rs @@ -99,6 +99,33 @@ lazy_static! { "total number of query memory allocations rejected" ) .unwrap(); + + /// Remote dynamic filter fanout RPC results, labeled with status. + pub static ref REMOTE_DYN_FILTER_UPDATE_RPC_TOTAL: IntCounterVec = register_int_counter_vec!( + "greptime_query_remote_dyn_filter_update_rpc_total", + "remote dynamic filter fanout RPC results", + &["status"] + ) + .unwrap(); + + /// Remote dynamic filter fanout payload bytes. + pub static ref REMOTE_DYN_FILTER_PAYLOAD_BYTES: Histogram = register_histogram!( + "greptime_query_remote_dyn_filter_payload_bytes", + "remote dynamic filter fanout payload bytes", + vec![ + 128.0, 256.0, 512.0, 1024.0, 2048.0, 4096.0, 8192.0, 16384.0, 32768.0, + 65536.0, 131072.0, 262144.0, 524288.0, + ] + ) + .unwrap(); + + /// Remote dynamic filter encode results, labeled with result. + pub static ref REMOTE_DYN_FILTER_ENCODE_TOTAL: IntCounterVec = register_int_counter_vec!( + "greptime_query_remote_dyn_filter_encode_total", + "remote dynamic filter encode results", + &["result"] + ) + .unwrap(); } /// A stream to call the callback once a RecordBatch stream is done.