From ee714d26432c06f184ba430787d1c1749fb1d71e Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Mon, 20 Apr 2026 10:57:56 -0400 Subject: [PATCH] =?UTF-8?q?feat(metrics):=20expose=20=C2=A76.3=20barrier?= =?UTF-8?q?=20drops=20as=20Prometheus=20counter?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up from #45: the barrier counter was only visible via the in-process AtomicU64 + debug log, so production deployments running at INFO had no graphable signal for silent drops. Changes: - New `stores::sketch_db::metrics` module with a `CounterVec` `queryengine_ingest_samples_blocked_by_schema_barrier_total`, keyed by `agg_id`, registered through the global `prometheus::default_registry()` so the existing `/metrics` handler (`handle_metrics` in drivers/query/servers/http.rs) scrapes it automatically. - Per-agg increment right alongside the atomic bump in `route_decoded_samples`, fed by the same per-batch `dropped_by_barrier` tally so the counter and the log stay consistent. - New unit test `barrier_prom_counter_increments_per_agg_label` that keys on a unique agg_id (9001) to get a deterministic baseline in the process-global registry. Why label by agg_id (not a scalar counter): lets operators alert on "drops on a specific agg while the registry still reports that agg Active" — the silent-drop regression the counter is meant to catch. 729 lib tests pass (+1), clippy clean, fmt clean. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../src/precompute_engine/ingest_handler.rs | 44 +++++++++++++++++++ .../src/stores/sketch_db/metrics.rs | 26 +++++++++++ asap-query-engine/src/stores/sketch_db/mod.rs | 1 + 3 files changed, 71 insertions(+) create mode 100644 asap-query-engine/src/stores/sketch_db/metrics.rs diff --git a/asap-query-engine/src/precompute_engine/ingest_handler.rs b/asap-query-engine/src/precompute_engine/ingest_handler.rs index ff7eea9d..bbb638f4 100644 --- a/asap-query-engine/src/precompute_engine/ingest_handler.rs +++ b/asap-query-engine/src/precompute_engine/ingest_handler.rs @@ -182,6 +182,14 @@ pub(crate) async fn route_decoded_samples( state .samples_blocked_by_schema_barrier .fetch_add(total_dropped, std::sync::atomic::Ordering::Relaxed); + // Also bump the Prometheus counter (per-agg label) so the + // drop rate is scrapable from /metrics without enabling debug + // logs in production. + for (agg_id, count) in &dropped_by_barrier { + crate::stores::sketch_db::metrics::SAMPLES_BLOCKED_BY_SCHEMA_BARRIER + .with_label_values(&[&agg_id.to_string()]) + .inc_by(*count as f64); + } debug!( total_dropped, by_agg_id = ?dropped_by_barrier, @@ -409,4 +417,40 @@ mod tests { drop(state); let _ = drain.await; } + + /// Prometheus CounterVec is process-global via `lazy_static`, so + /// we key on a unique agg_id (9001) to get a fresh baseline that + /// no other test in the process has touched. + #[tokio::test] + async fn barrier_prom_counter_increments_per_agg_label() { + let (state, drain) = setup_state(9001, "metric_prom_test").await; + let label = "9001"; + let baseline = crate::stores::sketch_db::metrics::SAMPLES_BLOCKED_BY_SCHEMA_BARRIER + .with_label_values(&[label]) + .get(); + + assert!(state.schemas.force_expire(9001).is_some()); + + let _ = route_decoded_samples( + &state, + vec![ + sample("metric_prom_test", 100, 1.0), + sample("metric_prom_test", 200, 2.0), + sample("metric_prom_test", 300, 3.0), + sample("metric_prom_test", 400, 4.0), + ], + std::time::Instant::now(), + ) + .await; + + let after = crate::stores::sketch_db::metrics::SAMPLES_BLOCKED_BY_SCHEMA_BARRIER + .with_label_values(&[label]) + .get(); + assert!( + (after - baseline - 4.0).abs() < f64::EPSILON, + "prom counter for agg_id={label} should have advanced by 4; baseline={baseline}, after={after}" + ); + drop(state); + let _ = drain.await; + } } diff --git a/asap-query-engine/src/stores/sketch_db/metrics.rs b/asap-query-engine/src/stores/sketch_db/metrics.rs new file mode 100644 index 00000000..c03031f0 --- /dev/null +++ b/asap-query-engine/src/stores/sketch_db/metrics.rs @@ -0,0 +1,26 @@ +//! Prometheus metrics exposed on `/metrics` for the sketch DB. +//! +//! Scope: counters / gauges that turn otherwise-silent ingest or +//! eviction behaviour into graphable signals. Registered via +//! `lazy_static` + `prometheus::register_counter_vec!`, so a single +//! `prometheus::gather()` at the HTTP handler picks them up +//! automatically. + +use lazy_static::lazy_static; +use prometheus::{register_counter_vec, CounterVec}; + +lazy_static! { + /// §6.3 write-side barrier drops, per `agg_id`. Incremented when + /// the ingest path matches a sample's metric to an agg config + /// whose `AggSchema` is Retired or Expired, so the sample is + /// dropped instead of being routed to a worker. A non-zero rate + /// with no Retired/Expired schemas in the registry means the + /// reconcile / hot-reload path is lagging — silent drop becomes + /// a graphable alert condition. + pub static ref SAMPLES_BLOCKED_BY_SCHEMA_BARRIER: CounterVec = register_counter_vec!( + "queryengine_ingest_samples_blocked_by_schema_barrier_total", + "Ingest samples dropped by the §6.3 write-side schema barrier, keyed by agg_id", + &["agg_id"] + ) + .unwrap(); +} diff --git a/asap-query-engine/src/stores/sketch_db/mod.rs b/asap-query-engine/src/stores/sketch_db/mod.rs index 21ab51f8..0adc2e79 100644 --- a/asap-query-engine/src/stores/sketch_db/mod.rs +++ b/asap-query-engine/src/stores/sketch_db/mod.rs @@ -34,6 +34,7 @@ pub mod backfill_processor; pub mod backfill_service; pub mod backfill_window_builder; pub mod backfill_worker; +pub mod metrics; pub mod prometheus_reader; pub mod raw_sample_reader; pub mod schema;