diff --git a/asap-query-engine/src/drivers/ingest/otel.rs b/asap-query-engine/src/drivers/ingest/otel.rs index 7413bfd8..bff659d4 100644 --- a/asap-query-engine/src/drivers/ingest/otel.rs +++ b/asap-query-engine/src/drivers/ingest/otel.rs @@ -338,6 +338,30 @@ fn process_otlp_request(request: &ExportMetricsServiceRequest, transport: &str) /// Metrics whose name does not match any aggregation in the streaming /// config are dropped with a debug log — the precompute engine only /// maintains state for configured metrics. +/// +/// Flush a per-driver `HashMap` of §6.3 write-barrier +/// drops into `IngestState::record_barrier_drop`, emitting a single +/// debug log summarising the batch. Called from every OTLP routing +/// function after its inner loop finishes, so a query against the +/// `/metrics` endpoint sees a unified `samples_blocked_by_schema_barrier` +/// counter regardless of which OTLP variant the DataCollector is +/// shipping. +fn flush_barrier_drops(state: &IngestState, drops: &HashMap, driver_tag: &'static str) { + if drops.is_empty() { + return; + } + let total: u64 = drops.values().sum(); + for (agg_id, count) in drops { + state.record_barrier_drop(*agg_id, *count); + } + debug!( + driver = driver_tag, + total_dropped = total, + by_agg_id = ?drops, + "§6.3 write barrier dropped OTLP samples (agg is retired/expired)" + ); +} + async fn route_otlp_to_precompute( request: &ExportMetricsServiceRequest, ingest_state: &Arc, @@ -359,6 +383,7 @@ async fn route_otlp_to_precompute( let mut by_group: HashMap> = HashMap::new(); let mut raw_matched = 0usize; let mut raw_unmatched = 0usize; + let mut raw_barrier_drops: HashMap = HashMap::new(); for point in &points { let series_key = format_series_key(&point.name, &point.labels); @@ -373,6 +398,7 @@ async fn route_otlp_to_precompute( } // §6.3 write-side schema barrier — see ingest_handler.rs. if !ingest_state.schemas.is_writable(config.aggregation_id) { + *raw_barrier_drops.entry(config.aggregation_id).or_default() += 1; continue; } let group_key = IngestState::extract_group_key_for(&series_key, config); @@ -388,6 +414,7 @@ async fn route_otlp_to_precompute( raw_unmatched += 1; } } + flush_barrier_drops(ingest_state, &raw_barrier_drops, "otlp-raw"); let raw_messages: Vec = by_group .into_iter() @@ -421,6 +448,7 @@ async fn route_otlp_to_precompute( let mut sketch_messages: Vec = Vec::new(); let mut sketch_matched = 0usize; let mut sketch_unmatched = 0usize; + let mut sketch_barrier_drops: HashMap = HashMap::new(); for point in &sketch_payloads { let series_key = format_series_key(&point.name, &point.labels); let ts_ms = (point.timestamp_nanos / 1_000_000) as i64; @@ -435,6 +463,9 @@ async fn route_otlp_to_precompute( } // §6.3 write-side schema barrier — see ingest_handler.rs. if !ingest_state.schemas.is_writable(config.aggregation_id) { + *sketch_barrier_drops + .entry(config.aggregation_id) + .or_default() += 1; continue; } let group_key = IngestState::extract_group_key_for(&series_key, config); @@ -476,6 +507,7 @@ async fn route_otlp_to_precompute( sketch_unmatched += 1; } } + flush_barrier_drops(ingest_state, &sketch_barrier_drops, "otlp-sketch-envelope"); if !sketch_messages.is_empty() { if let Err(e) = ingest_state @@ -525,6 +557,7 @@ async fn route_modified_otlp_sketches_to_precompute( let mut routed = 0usize; let mut decoded_failed = 0usize; let mut unconfigured = 0usize; + let mut barrier_drops: HashMap = HashMap::new(); for resource_metrics in &request.resource_metrics { let resource_attrs = resource_metrics @@ -644,6 +677,7 @@ async fn route_modified_otlp_sketches_to_precompute( } // §6.3 write-side schema barrier — see ingest_handler.rs. if !ingest_state.schemas.is_writable(config.aggregation_id) { + *barrier_drops.entry(config.aggregation_id).or_default() += 1; continue; } let group_key = IngestState::extract_group_key_for(&series_key, config); @@ -666,6 +700,8 @@ async fn route_modified_otlp_sketches_to_precompute( } } + flush_barrier_drops(ingest_state, &barrier_drops, "otlp-modified-proto"); + if !messages.is_empty() { if let Err(e) = ingest_state .router diff --git a/asap-query-engine/src/precompute_engine/ingest_handler.rs b/asap-query-engine/src/precompute_engine/ingest_handler.rs index bbb638f4..87fc0ba3 100644 --- a/asap-query-engine/src/precompute_engine/ingest_handler.rs +++ b/asap-query-engine/src/precompute_engine/ingest_handler.rs @@ -64,6 +64,22 @@ impl IngestState { pub fn extract_group_key_for(series_key: &str, config: &AggregationConfig) -> String { extract_group_key(series_key, config) } + + /// Record a §6.3 write-side barrier drop. Updates the in-process + /// `samples_blocked_by_schema_barrier` atomic and the Prometheus + /// `queryengine_ingest_samples_blocked_by_schema_barrier_total` + /// counter, both keyed by `agg_id`. Called by every ingest path + /// (Prometheus remote-write, VictoriaMetrics remote-write, + /// OTLP raw points, OTLP sketch envelopes, OTLP modified-proto + /// sketches) so the drop rate observable on `/metrics` is a + /// single number regardless of which driver is active. + pub fn record_barrier_drop(&self, agg_id: u64, count: u64) { + self.samples_blocked_by_schema_barrier + .fetch_add(count, std::sync::atomic::Ordering::Relaxed); + crate::stores::sketch_db::metrics::SAMPLES_BLOCKED_BY_SCHEMA_BARRIER + .with_label_values(&[&agg_id.to_string()]) + .inc_by(count as f64); + } } /// Extract the group key (grouping label values joined by semicolons) @@ -179,16 +195,8 @@ pub(crate) async fn route_decoded_samples( if !dropped_by_barrier.is_empty() { let total_dropped: u64 = dropped_by_barrier.values().sum(); - 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); + state.record_barrier_drop(*agg_id, *count); } debug!( total_dropped, @@ -453,4 +461,42 @@ mod tests { drop(state); let _ = drain.await; } + + /// `record_barrier_drop` is the single entry point every ingest + /// driver (Prometheus remote-write, VictoriaMetrics remote-write, + /// OTLP raw / sketch / modified-proto) funnels through, so + /// verify both sides of the contract: the in-process atomic AND + /// the Prometheus `CounterVec` move together, keyed by agg_id. + #[tokio::test] + async fn record_barrier_drop_advances_atomic_and_prom_counter() { + let (state, drain) = setup_state(4242, "metric_helper_test").await; + let label = "4242"; + let prom_baseline = crate::stores::sketch_db::metrics::SAMPLES_BLOCKED_BY_SCHEMA_BARRIER + .with_label_values(&[label]) + .get(); + let atomic_baseline = state + .samples_blocked_by_schema_barrier + .load(Ordering::Relaxed); + + state.record_barrier_drop(4242, 7); + + let prom_after = crate::stores::sketch_db::metrics::SAMPLES_BLOCKED_BY_SCHEMA_BARRIER + .with_label_values(&[label]) + .get(); + let atomic_after = state + .samples_blocked_by_schema_barrier + .load(Ordering::Relaxed); + + assert!( + (prom_after - prom_baseline - 7.0).abs() < f64::EPSILON, + "prom counter must advance by 7 via the helper" + ); + assert_eq!( + atomic_after - atomic_baseline, + 7, + "atomic counter must advance by 7 via the helper" + ); + drop(state); + let _ = drain.await; + } }