fix(metrics): report pressure and processed counts

crates/fetcher/src/emit.rs

@@ -184,6 +184,11 @@ impl Emitter {
         // lease is released only once every outcome is known.
         let mut pending: Vec<Outbound> = std::mem::take(&mut batch.rows);
         let total = pending.len();
+        // A refused row failed enrichment, so only the rest were processed.
+        let processed = pending
+            .iter()
+            .filter(|row| row.tenancy != Tenancy::Refused)
+            .count();
         let mut backoff = BACKPRESSURE_BACKOFF;
         for attempt in 0..=retries {
             if pending.is_empty() {
@@ -209,6 +214,8 @@ impl Emitter {
             )));
         }
         drop(batch);
+        // Counted only once the batch is through, so an aborted tick counts its rows on the re-fetch alone, as received does
+        self.metrics.add_records_processed(processed as u64);
         if usage.stamped > 0 {
             metrics::counter!(USAGE_INGESTED_BYTES_TOTAL, "org_id" => org.to_owned())
                 .increment(usage.stamped);
@@ -568,6 +575,31 @@ mod tests {
         assert!(lines.contains(UNSTAMPABLE), "{lines}");
     }
 
+    /// Processed counts the records enrichment let through, and a row it
+    /// refused is not one of them.
+    #[tokio::test]
+    async fn processed_counts_the_rows_enrichment_let_through() {
+        let emitter = emitter_over("org_id: acme\n", &MemoryConfig::default());
+        let mut batcher = Batcher::new(AccumulateConfig::default(), Arc::new(NoLease));
+        batcher.push(
+            Outbound::new("fixture_land", r#"{"_org_id":"victim"}"#).with_tenancy(Tenancy::Refused),
+            None,
+        );
+        batcher.push(Outbound::new("fixture_land", r#"{"a":1}"#), None);
+        batcher.push(
+            Outbound::new("fixture_land", "plain text").with_tenancy(Tenancy::Unstamped),
+            None,
+        );
+
+        emitter.emit(batcher.take()).await.expect("sent");
+
+        let output = emitter.metrics.render();
+        assert!(
+            output.contains("dfe_records_processed_total 2\n"),
+            "{output}"
+        );
+    }
+
     /// A flush that aborts counts nothing, even for the records it sent before
     /// the abort: the tick re-fetches and sends them again.
     #[tokio::test]
@@ -592,5 +624,10 @@ mod tests {
             recorder.counter("usage_ingested_bytes_total", &[("org_id", "default")]),
             None
         );
+        let output = emitter.metrics.render();
+        assert!(
+            output.contains("dfe_records_processed_total 0\n"),
+            "{output}"
+        );
     }
 }

crates/fetcher/src/main.rs

@@ -354,15 +354,7 @@ async fn run_fetcher_service(
     // with fetcher-specific weights for buffer depth, transport errors, and memory.
     let scaling_pressure = Arc::new(scalo::scaling::ScalingPressure::new(
         config.scaling.clone(),
-        vec![
-            ScalingComponent::new(
-                "buffer_depth",
-                0.40,
-                config.buffer.memory_limit.max(1) as f64,
-            ),
-            ScalingComponent::new("transport_errors", 0.30, 100.0),
-            ScalingComponent::new("memory", 0.30, 1.0),
-        ],
+        scaling_components(),
     ));
 
     // Start config hot-reload
@@ -431,7 +423,7 @@ async fn run_fetcher_service(
                         let limit = mg.limit_bytes();
                         let circuit_open = !state.output_healthy();
                         // Canonical weighted pressure engine (served to KEDA).
-                        scaling.set_component("buffer_depth", used as f64);
+                        scaling.set_component("buffer_depth", buffer_depth(used, limit));
                         scaling.set_memory(used, limit);
                         scaling.set_circuit_open(circuit_open);
                         // Keep the per-tick scaling gauges current for KEDA's
@@ -794,6 +786,23 @@ async fn run_fetcher_service(
     Ok(())
 }
 
+/// The fetcher's scaling components, each read as a 0-1 fraction of its saturation.
+fn scaling_components() -> Vec<ScalingComponent> {
+    vec![
+        ScalingComponent::new("buffer_depth", 0.40, 1.0),
+        ScalingComponent::new("transport_errors", 0.30, 100.0),
+        ScalingComponent::new("memory", 0.30, 1.0),
+    ]
+}
+
+/// The share of the memory guard's limit the buffered bytes hold, 0 when it has no limit.
+fn buffer_depth(used: u64, limit: u64) -> f64 {
+    if limit == 0 {
+        return 0.0;
+    }
+    used as f64 / limit as f64
+}
+
 /// The pressure latch framework drivers gate on: the memory guard as the one
 /// HARD source under the configured hysteresis, or `None` when the brake is
 /// off.
@@ -879,6 +888,33 @@ mod tests {
         resolved
     }
 
+    /// An idle fetcher reads near 0 and a buffer that fills the guard's limit
+    /// reads the component's full 40, whatever `buffer.memory_limit` is.
+    #[test]
+    fn buffer_depth_pressure_follows_the_guard_limit() {
+        let cases: [(&str, u64, u64, f64); 5] = [
+            (
+                "idle",
+                14_225_408,
+                684_510_412,
+                40.0 * 14_225_408.0 / 684_510_412.0,
+            ),
+            ("half full", 50, 100, 20.0),
+            ("full", 100, 100, 40.0),
+            ("over the limit", 500, 100, 40.0),
+            ("no limit", 14_225_408, 0, 0.0),
+        ];
+        for (name, used, limit, expected) in cases {
+            let pressure = scalo::scaling::ScalingPressure::new(
+                scalo::scaling::ScalingPressureConfig::default(),
+                scaling_components(),
+            );
+            pressure.set_component("buffer_depth", buffer_depth(used, limit));
+            let got = pressure.calculate();
+            assert!((got - expected).abs() < 1e-9, "{name}: {got} != {expected}");
+        }
+    }
+
     /// `metrics-manifest` builds a bare manager and calls `register_metrics`,
     /// so the catalogue holds the fetcher's metrics only if the app overrides it.
     #[test]

crates/fetcher/src/metrics/mod.rs

@@ -224,6 +224,7 @@ pub struct Metrics {
     // Pipeline / delivery
     pipeline_ready: AtomicU64,          // gauge: 1=ready, 0=backpressured
     records_delivered_total: AtomicU64, // counter
+    records_processed_total: AtomicU64, // counter
 
     // Gauges
     active_fetches: AtomicU64,
@@ -279,6 +280,7 @@ impl Metrics {
             dead_letters_dropped_total: AtomicU64::new(0),
             pipeline_ready: AtomicU64::new(1),
             records_delivered_total: AtomicU64::new(0),
+            records_processed_total: AtomicU64::new(0),
             active_fetches: AtomicU64::new(0),
             active_extractors: AtomicU64::new(0),
             memory_used_bytes: AtomicU64::new(0),
@@ -852,6 +854,19 @@ impl Metrics {
         }
     }
 
+    /// Add records that made it through processing (enrichment) and went on to the output.
+    ///
+    /// ServiceMetrics has no processed counter, so the dual-emit counts the
+    /// bare `records_processed_total` series scalo registers.
+    #[inline]
+    pub fn add_records_processed(&self, count: u64) {
+        self.records_processed_total
+            .fetch_add(count, Ordering::Relaxed);
+        if self.dfe.is_some() {
+            metrics::counter!("records_processed_total").increment(count);
+        }
+    }
+
     // ==========================================================================
     // Gauges
     // ==========================================================================
@@ -1012,6 +1027,13 @@ impl Metrics {
             self.records_delivered_total.load(Ordering::Relaxed)
         ));
 
+        output.push_str("# HELP dfe_records_processed_total Total records processed\n");
+        output.push_str("# TYPE dfe_records_processed_total counter\n");
+        output.push_str(&format!(
+            "dfe_records_processed_total {}\n",
+            self.records_processed_total.load(Ordering::Relaxed)
+        ));
+
         // Extractor counters
         output.push_str("# HELP dfe_fetcher_extractor_runs_total Total extractor runs\n");
         output.push_str("# TYPE dfe_fetcher_extractor_runs_total counter\n");
@@ -1453,6 +1475,25 @@ mod tests {
         );
     }
 
+    /// ServiceMetrics has no processed counter, so the fetcher counts the
+    /// bare series itself, and the local total reads the same.
+    #[test]
+    fn processed_records_reach_records_processed_total() {
+        let manager = MetricsManager::with_config(scalo::metrics::MetricsConfig::offline(""));
+        let mut output = String::new();
+        let hits = counted("records_processed_total", || {
+            let m = Metrics::with_dfe(&manager);
+            m.add_records_processed(3);
+            m.add_records_processed(2);
+            output = m.render();
+        });
+        assert_eq!(hits, 5);
+        assert!(
+            output.contains("dfe_records_processed_total 5\n"),
+            "{output}"
+        );
+    }
+
     #[test]
     fn test_inc_records_filtered() {
         let metrics = Metrics::new();