diff --git a/crates/fetcher/src/emit.rs b/crates/fetcher/src/emit.rs index 5ce88186..85bb98b3 100644 --- a/crates/fetcher/src/emit.rs +++ b/crates/fetcher/src/emit.rs @@ -184,6 +184,11 @@ impl Emitter { // lease is released only once every outcome is known. let mut pending: Vec = 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}" + ); } } diff --git a/crates/fetcher/src/main.rs b/crates/fetcher/src/main.rs index cca896b1..5ef0be1b 100644 --- a/crates/fetcher/src/main.rs +++ b/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 { + 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] diff --git a/crates/fetcher/src/metrics/mod.rs b/crates/fetcher/src/metrics/mod.rs index 627fb5ef..3366d6d9 100644 --- a/crates/fetcher/src/metrics/mod.rs +++ b/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();