feat: add remote dynamic filter metrics (#8309)

* feat: add remote dyn filter metrics

Signed-off-by: discord9 <discord9@163.com>

* fix: unify remote dyn filter payload budget

Signed-off-by: discord9 <discord9@163.com>

* style: import remote dyn filter metrics

Signed-off-by: discord9 <discord9@163.com>

* fix: separate remote dyn filter unregister metrics

Signed-off-by: discord9 <discord9@163.com>

* fix: rename remote dyn filter update outcome metric

Signed-off-by: discord9 <discord9@163.com>

---------

Signed-off-by: discord9 <discord9@163.com>

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.
 ///

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,
         )
     }
 

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();
 }

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<RemoteDynFilterId, RegisteredDynFilter>;
 
@@ -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<u8>,
@@ -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);
             }
         }

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)]

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(
         &current,
-        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<dyn datafusion::physical_plan::PhysicalExpr>,
-            test_dyn_filter_with_snapshot_payload("pod", 1, 40 * 1024)
+            test_dyn_filter_with_snapshot_payload("pod", 1, oversized_total_snapshot_bytes)
                 as Arc<dyn datafusion::physical_plan::PhysicalExpr>,
         ];
 

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(
-        &current,
-        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(&current, 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;
+            }
         }
     }
 

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.