feat(metrics): wire §6.3 barrier counter into all OTLP ingest paths

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<agg_id, count>` 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<u64, u64>, 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<IngestState>,
@@ -359,6 +383,7 @@ async fn route_otlp_to_precompute(
     let mut by_group: HashMap<GroupKey, Vec<SampleTuple>> = HashMap::new();
     let mut raw_matched = 0usize;
     let mut raw_unmatched = 0usize;
+    let mut raw_barrier_drops: HashMap<u64, u64> = 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<WorkerMessage> = by_group
         .into_iter()
@@ -421,6 +448,7 @@ async fn route_otlp_to_precompute(
     let mut sketch_messages: Vec<WorkerMessage> = Vec::new();
     let mut sketch_matched = 0usize;
     let mut sketch_unmatched = 0usize;
+    let mut sketch_barrier_drops: HashMap<u64, u64> = 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<u64, u64> = 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

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;
+    }
 }