From b9f157e73bba4df73a8f38f7aa1b52c0a87c2c46 Mon Sep 17 00:00:00 2001 From: Yoenn Burban Date: Tue, 11 Aug 2026 12:09:33 +0200 Subject: [PATCH] fix(internal_metrics source): emit increments for counters and histograms `internal_metrics` emitted every metric as absolute. Sinks that require incremental metrics discard the first observation of an absolute series to avoid reporting an accumulated counter as one enormous delta, and then filter histograms whose delta is empty. A histogram observed only within a single scrape interval therefore established a baseline, produced zero deltas forever after, and was never sent. Counters lost one scrape interval. The internal registry starts at zero within the process, so the first value observed for a series is already a correct increment and nothing needs to be discarded. Track the previous absolute value per series in the source and emit counters and histograms as increments. Gauges stay absolute, as does `internal_metrics_cardinality_total`, which is declared a counter but reports the current series count and so is not monotonic. The source also scrapes once while being built, so adding it to an already-running Vector, or reloading the configuration, reports the change since that point rather than the whole accumulated registry. Series that expire from the registry are dropped from the tracked state via a per-scrape generation stamp, bounding the map. Component validation is adjusted as a consequence: `sum_counters` overwrote its accumulator for absolute metrics instead of adding, so multi-series counters silently reported one arbitrary series rather than their total. It now always sums. This makes `component_errors_total` correctly report that `http_server` and `splunk_hec` record two errors for a single malformed request, so test cases can declare `errors_per_failure`. --- .../internal_metrics_increments.breaking.md | 19 ++ lib/vector-core/src/metrics/mod.rs | 2 +- src/components/validation/runner/mod.rs | 4 +- src/components/validation/test_case.rs | 11 + .../validators/component_spec/mod.rs | 13 +- src/sources/internal_metrics/delta.rs | 231 ++++++++++++++++++ .../mod.rs} | 132 +++++++++- .../components/sources/http_server.yaml | 3 + .../components/sources/splunk_hec.yaml | 3 + 9 files changed, 404 insertions(+), 14 deletions(-) create mode 100644 changelog.d/internal_metrics_increments.breaking.md create mode 100644 src/sources/internal_metrics/delta.rs rename src/sources/{internal_metrics.rs => internal_metrics/mod.rs} (71%) diff --git a/changelog.d/internal_metrics_increments.breaking.md b/changelog.d/internal_metrics_increments.breaking.md new file mode 100644 index 0000000000000..b93257b7338b8 --- /dev/null +++ b/changelog.d/internal_metrics_increments.breaking.md @@ -0,0 +1,19 @@ +# `internal_metrics` emits incremental counters and histograms {#internal-metrics-increments} + +## Summary + +The `internal_metrics` source now emits counters and histograms as `Incremental` metrics holding the +change since the previous scrape, rather than as `Absolute` metrics holding the value accumulated +since Vector started. Gauges are unaffected, as is `internal_metrics_cardinality_total`. + +This fixes internal histograms being dropped entirely when all of their observations fall within a +single scrape interval. The source also scrapes once while being built, so adding it to an +already-running Vector reports the change since that point rather than the whole registry. + +## Migration + +Sinks are unaffected. Adjust any transform that reads the metric kind of internal counters or +histograms, such as a `remap` gating on the kind or a consumer of `metric_to_log` output reading the +`kind` field: these now see `incremental` where they previously saw `absolute`. + +authors: gwenaskell diff --git a/lib/vector-core/src/metrics/mod.rs b/lib/vector-core/src/metrics/mod.rs index 0e56d28112159..534941074d94c 100644 --- a/lib/vector-core/src/metrics/mod.rs +++ b/lib/vector-core/src/metrics/mod.rs @@ -48,7 +48,7 @@ const CARDINALITY_KEY_NAME: &str = "internal_metrics_cardinality"; static CARDINALITY_KEY: Key = Key::from_static_name(CARDINALITY_KEY_NAME); // Older deprecated counter key name -const CARDINALITY_COUNTER_KEY_NAME: &str = "internal_metrics_cardinality_total"; +pub const CARDINALITY_COUNTER_KEY_NAME: &str = "internal_metrics_cardinality_total"; static CARDINALITY_COUNTER_KEY: Key = Key::from_static_name(CARDINALITY_COUNTER_KEY_NAME); /// Controller allows capturing metric snapshots. diff --git a/src/components/validation/runner/mod.rs b/src/components/validation/runner/mod.rs index a9d8aee743f9d..af053f3ceaf57 100644 --- a/src/components/validation/runner/mod.rs +++ b/src/components/validation/runner/mod.rs @@ -326,6 +326,7 @@ impl Runner { maybe_runner_encoder.as_ref().cloned(), self.configuration.component_type, self.configuration.log_namespace(), + test_case.errors_per_failure, ); // the number of events we expect to receive from the output. @@ -565,6 +566,7 @@ fn spawn_input_driver( mut maybe_encoder: Option>, component_type: ComponentType, log_namespace: LogNamespace, + errors_per_failure: u64, ) -> JoinHandle<()> { let input_runner_metrics = Arc::clone(runner_metrics); @@ -601,7 +603,7 @@ fn spawn_input_driver( // account for failure case if failure_case { - input_runner_metrics.errors_total += 1; + input_runner_metrics.errors_total += errors_per_failure; // TODO: this assumption may need to be made configurable at some point if component_type == ComponentType::Sink { input_runner_metrics.discarded_events_total += 1; diff --git a/src/components/validation/test_case.rs b/src/components/validation/test_case.rs index a281f6e5883f0..7a3ccba1a413d 100644 --- a/src/components/validation/test_case.rs +++ b/src/components/validation/test_case.rs @@ -29,4 +29,15 @@ pub struct TestCase { pub config_name: Option, pub expectation: TestCaseExpectation, pub events: Vec, + + /// How many `component_errors_total` increments the component records for each failing event. + /// + /// Defaults to one. Some components record several distinct errors for a single failure, such + /// as a decoding error followed by the rejected request it causes. + #[serde(default = "default_errors_per_failure")] + pub errors_per_failure: u64, +} + +const fn default_errors_per_failure() -> u64 { + 1 } diff --git a/src/components/validation/validators/component_spec/mod.rs b/src/components/validation/validators/component_spec/mod.rs index c9ebddb0ef948..add2f9f14c973 100644 --- a/src/components/validation/validators/component_spec/mod.rs +++ b/src/components/validation/validators/component_spec/mod.rs @@ -1,4 +1,4 @@ -use vector_lib::event::{Event, Metric, MetricKind}; +use vector_lib::event::{Event, Metric}; use super::{ComponentMetricType, Validator}; use crate::components::validation::{ @@ -243,15 +243,12 @@ fn sum_counters( let mut sum: f64 = 0.0; let mut errs = Vec::new(); + // The `internal_metrics` source collecting this telemetry emits counters incrementally, so + // every observation is a distinct contribution. Summing also correctly accumulates metrics + // split over several series, such as `component_errors_total` broken down by `error_type`. for m in metrics { match m.value() { - vector_lib::event::MetricValue::Counter { value } => { - if let MetricKind::Absolute = m.data().kind { - sum = *value; - } else { - sum += *value; - } - } + vector_lib::event::MetricValue::Counter { value } => sum += *value, _ => errs.push(format!("{metric_name}: metric value is not a counter",)), } } diff --git a/src/sources/internal_metrics/delta.rs b/src/sources/internal_metrics/delta.rs new file mode 100644 index 0000000000000..2495373a73154 --- /dev/null +++ b/src/sources/internal_metrics/delta.rs @@ -0,0 +1,231 @@ +use std::collections::HashMap; + +use vector_lib::{ + event::{Metric, MetricKind, MetricValue, metric::MetricSeries}, + metrics::CARDINALITY_COUNTER_KEY_NAME, +}; + +/// Previous absolute value of a series, stamped with the scrape that last observed it. +struct Entry { + value: MetricValue, + generation: u64, +} + +/// Converts the absolute values captured from the metrics registry into increments. +/// +/// Registry handles are created at zero within the process, so the first value observed for a +/// series *is* its increment. That differs from externally scraped metrics, where sinks must +/// discard the first observation to avoid emitting an accumulated counter as one huge delta. +#[derive(Default)] +pub(super) struct DeltaState { + seen: HashMap, + generation: u64, +} + +impl DeltaState { + /// Records the given metrics as the baseline without emitting anything. + pub(super) fn seed(&mut self, mut metrics: Vec) { + self.convert(&mut metrics); + } + + /// Rewrites counters and histograms in place into increments over the previous scrape. + /// + /// Gauges are already meaningful as absolute values and pass through untouched. + pub(super) fn convert(&mut self, metrics: &mut [Metric]) { + self.generation = self.generation.wrapping_add(1); + let generation = self.generation; + + for metric in metrics { + if !is_cumulative(metric) { + continue; + } + + let absolute = metric.value().clone(); + match self.seen.get_mut(metric.series()) { + Some(entry) => { + if !metric.value_mut().subtract(&entry.value) { + // `subtract` leaves the value untouched when it detects a reset, so it is + // already the increment from a series that restarted at zero. + debug!( + message = "Internal metric series reset, reporting its full value.", + series = ?metric.series(), + ); + } + entry.value = absolute; + entry.generation = generation; + } + None => { + self.seen.insert( + metric.series().clone(), + Entry { + value: absolute, + generation, + }, + ); + } + } + metric.data_mut().kind = MetricKind::Incremental; + } + + // Forget series that expired from the registry, bounding the map. + self.seen.retain(|_, entry| entry.generation == generation); + } +} + +/// Whether a captured metric accumulates over the process lifetime and so needs differencing. +fn is_cumulative(metric: &Metric) -> bool { + // The cardinality counter is declared a counter but reports the current series count, so it is + // not monotonic and has to stay absolute. + if metric.name() == CARDINALITY_COUNTER_KEY_NAME { + return false; + } + matches!( + metric.value(), + MetricValue::Counter { .. } | MetricValue::AggregatedHistogram { .. } + ) +} + +#[cfg(test)] +mod tests { + use vector_lib::event::metric::Bucket; + + use super::*; + + fn counter(name: &str, value: f64) -> Metric { + Metric::new(name, MetricKind::Absolute, MetricValue::Counter { value }) + } + + fn histogram(count: u64, sum: f64) -> Metric { + Metric::new( + "histo", + MetricKind::Absolute, + MetricValue::AggregatedHistogram { + buckets: vec![Bucket { + upper_limit: 1.0, + count, + }], + count, + sum, + }, + ) + } + + fn convert_one(state: &mut DeltaState, metric: Metric) -> Metric { + let mut metrics = [metric]; + state.convert(&mut metrics); + let [metric] = metrics; + metric + } + + #[test] + fn first_sighting_emits_full_value_as_increment() { + let mut state = DeltaState::default(); + let metric = convert_one(&mut state, counter("a", 7.0)); + + assert_eq!(metric.kind(), MetricKind::Incremental); + assert_eq!(metric.value(), &MetricValue::Counter { value: 7.0 }); + } + + #[test] + fn subsequent_sightings_emit_the_delta() { + let mut state = DeltaState::default(); + convert_one(&mut state, counter("a", 7.0)); + + let metric = convert_one(&mut state, counter("a", 10.0)); + assert_eq!(metric.value(), &MetricValue::Counter { value: 3.0 }); + + let metric = convert_one(&mut state, counter("a", 10.5)); + assert_eq!(metric.value(), &MetricValue::Counter { value: 0.5 }); + } + + #[test] + fn idle_counter_emits_zero_delta() { + let mut state = DeltaState::default(); + convert_one(&mut state, counter("a", 7.0)); + + let metric = convert_one(&mut state, counter("a", 7.0)); + assert_eq!(metric.value(), &MetricValue::Counter { value: 0.0 }); + } + + #[test] + fn counter_reset_emits_full_value() { + let mut state = DeltaState::default(); + convert_one(&mut state, counter("a", 100.0)); + + // The registry restarted this series from zero, so 2.0 is the whole increment. + let metric = convert_one(&mut state, counter("a", 2.0)); + assert_eq!(metric.value(), &MetricValue::Counter { value: 2.0 }); + + // The reset value became the new baseline. + let metric = convert_one(&mut state, counter("a", 5.0)); + assert_eq!(metric.value(), &MetricValue::Counter { value: 3.0 }); + } + + #[test] + fn histogram_buckets_are_differenced() { + let mut state = DeltaState::default(); + convert_one(&mut state, histogram(2, 11.0)); + + let metric = convert_one(&mut state, histogram(5, 30.0)); + assert_eq!(metric.kind(), MetricKind::Incremental); + assert_eq!( + metric.value(), + &MetricValue::AggregatedHistogram { + buckets: vec![Bucket { + upper_limit: 1.0, + count: 3 + }], + count: 3, + sum: 19.0, + } + ); + } + + #[test] + fn gauges_pass_through_as_absolute() { + let mut state = DeltaState::default(); + let gauge = Metric::new("g", MetricKind::Absolute, MetricValue::Gauge { value: 2.0 }); + + let metric = convert_one(&mut state, gauge.clone()); + assert_eq!(metric.kind(), MetricKind::Absolute); + assert_eq!(metric.value(), &MetricValue::Gauge { value: 2.0 }); + + // No state is retained, so a later lower value is still reported verbatim. + let metric = convert_one(&mut state, gauge); + assert_eq!(metric.value(), &MetricValue::Gauge { value: 2.0 }); + } + + #[test] + fn cardinality_counter_passes_through_as_absolute() { + let mut state = DeltaState::default(); + convert_one(&mut state, counter(CARDINALITY_COUNTER_KEY_NAME, 10.0)); + + let metric = convert_one(&mut state, counter(CARDINALITY_COUNTER_KEY_NAME, 4.0)); + assert_eq!(metric.kind(), MetricKind::Absolute); + assert_eq!(metric.value(), &MetricValue::Counter { value: 4.0 }); + } + + #[test] + fn seed_suppresses_the_baseline() { + let mut state = DeltaState::default(); + state.seed(vec![counter("a", 100.0)]); + + let metric = convert_one(&mut state, counter("a", 103.0)); + assert_eq!(metric.value(), &MetricValue::Counter { value: 3.0 }); + } + + #[test] + fn expired_series_are_forgotten() { + let mut state = DeltaState::default(); + convert_one(&mut state, counter("a", 100.0)); + assert_eq!(state.seen.len(), 1); + + // A scrape without `a` means the registry expired it. + convert_one(&mut state, counter("b", 1.0)); + assert_eq!(state.seen.len(), 1); + + // So `a` re-registering counts from zero again. + let metric = convert_one(&mut state, counter("a", 4.0)); + assert_eq!(metric.value(), &MetricValue::Counter { value: 4.0 }); + } +} diff --git a/src/sources/internal_metrics.rs b/src/sources/internal_metrics/mod.rs similarity index 71% rename from src/sources/internal_metrics.rs rename to src/sources/internal_metrics/mod.rs index 76dfab8849997..487f61195615f 100644 --- a/src/sources/internal_metrics.rs +++ b/src/sources/internal_metrics/mod.rs @@ -20,6 +20,10 @@ use crate::{ shutdown::ShutdownSignal, }; +mod delta; + +use delta::DeltaState; + /// Configuration for the `internal_metrics` source. #[serde_as] #[configurable_component(source( @@ -113,12 +117,21 @@ impl SourceConfig for InternalMetricsConfig { .as_deref() .and_then(|tag| (!tag.is_empty()).then(|| tag.to_owned())); + let controller = Controller::get()?; + + // Scrape now so that a source built after Vector has already been running does not report + // the whole accumulated registry as one enormous first increment. Also covers reloads, + // which rebuild sources. + let mut deltas = DeltaState::default(); + deltas.seed(controller.capture_metrics()); + Ok(Box::pin( InternalMetrics { namespace, host_key, pid_key, - controller: Controller::get()?, + controller, + deltas, interval, out: cx.out, shutdown: cx.shutdown, @@ -141,6 +154,7 @@ struct InternalMetrics<'a> { host_key: OptionalValuePath, pid_key: Option, controller: &'a Controller, + deltas: DeltaState, interval: time::Duration, out: SourceSender, shutdown: ShutdownSignal, @@ -156,7 +170,9 @@ impl InternalMetrics<'_> { let hostname = crate::get_hostname(); let pid = std::process::id().to_string(); - let metrics = self.controller.capture_metrics(); + let mut metrics = self.controller.capture_metrics(); + self.deltas.convert(&mut metrics); + let count = metrics.len(); let byte_size = metrics.size_of(); let json_size = metrics.estimated_json_encoded_size_of(); @@ -208,14 +224,45 @@ mod tests { use crate::{ event::{ Event, - metric::{Metric, MetricValue}, + metric::{Metric, MetricKind, MetricValue}, }, test_util::{ - self, + self, collect_ready, components::{SOURCE_TAGS, run_and_assert_source_compliance}, }, }; + /// Builds and runs the source for a single scrape, letting the caller record metrics in + /// between so they land after the build-time seed. The default 1s interval fires once + /// immediately, so exactly one scrape is collected. + async fn scrape_once(record: impl FnOnce()) -> Vec { + test_util::trace_init(); + + let (tx, rx) = SourceSender::new_test(); + let source = InternalMetricsConfig::default() + .build(SourceContext::new_test(tx, None)) + .await + .unwrap(); + + record(); + + tokio::spawn(source); + time::sleep(time::Duration::from_millis(100)).await; + + collect_ready(rx) + .await + .into_iter() + .map(Event::into_metric) + .collect() + } + + fn find<'a>(metrics: &'a [Metric], name: &str) -> &'a Metric { + metrics + .iter() + .find(|metric| metric.name() == name) + .unwrap_or_else(|| panic!("{name} not emitted, got {metrics:?}")) + } + #[test] fn generate_config() { test_util::test_generate_config::(); @@ -348,6 +395,83 @@ mod tests { assert!(metric.tag_value("pid").is_none()); } + /// The scenario from OPA-5040: every observation lands inside a single scrape interval, which + /// previously left the sink with only a baseline and nothing to emit. + #[tokio::test] + async fn emits_increments_for_counters_and_histograms() { + let counter_name = CounterName::iter().next().unwrap(); + let histogram_name = HistogramName::iter().next().unwrap(); + + let metrics = scrape_once(|| { + counter!(counter_name).increment(3); + histogram!(histogram_name).record(5.0); + histogram!(histogram_name).record(6.0); + }) + .await; + + let counter = find(&metrics, counter_name.as_str()); + assert_eq!(counter.kind(), MetricKind::Incremental); + assert_eq!(counter.value(), &MetricValue::Counter { value: 3.0 }); + + let histogram = find(&metrics, histogram_name.as_str()); + assert_eq!(histogram.kind(), MetricKind::Incremental); + match histogram.value() { + MetricValue::AggregatedHistogram { count, sum, .. } => { + assert_eq!(*count, 2); + assert_eq!(*sum, 11.0); + } + value => panic!("wrong type: {value:?}"), + } + } + + #[tokio::test] + async fn gauges_stay_absolute() { + let gauge_name = GaugeName::iter().next().unwrap(); + + let metrics = scrape_once(|| gauge!(gauge_name).set(2.0)).await; + + let gauge = find(&metrics, gauge_name.as_str()); + assert_eq!(gauge.kind(), MetricKind::Absolute); + assert_eq!(gauge.value(), &MetricValue::Gauge { value: 2.0 }); + + // Declared a counter but non-monotonic, so it is left absolute too. + let cardinality = find(&metrics, vector_lib::metrics::CARDINALITY_COUNTER_KEY_NAME); + assert_eq!(cardinality.kind(), MetricKind::Absolute); + } + + /// The build-time scrape keeps a source added to an already-running Vector from reporting the + /// whole accumulated registry as its first increment. + #[tokio::test] + async fn build_time_scrape_excludes_prior_activity() { + let counter_name = CounterName::iter().next().unwrap(); + + test_util::trace_init(); + counter!(counter_name).increment(10); + + let (tx, rx) = SourceSender::new_test(); + let source = InternalMetricsConfig::default() + .build(SourceContext::new_test(tx, None)) + .await + .unwrap(); + + counter!(counter_name).increment(5); + + tokio::spawn(source); + time::sleep(time::Duration::from_millis(100)).await; + + let metrics = collect_ready(rx) + .await + .into_iter() + .map(Event::into_metric) + .collect::>(); + + // 5, not the 15 now held by the registry. + assert_eq!( + find(&metrics, counter_name.as_str()).value(), + &MetricValue::Counter { value: 5.0 } + ); + } + #[tokio::test] async fn namespace() { let namespace = "totally_custom"; diff --git a/tests/validation/components/sources/http_server.yaml b/tests/validation/components/sources/http_server.yaml index 5e1d8bf644038..a87bcc7418899 100644 --- a/tests/validation/components/sources/http_server.yaml +++ b/tests/validation/components/sources/http_server.yaml @@ -6,6 +6,9 @@ - log: simple message 3 - name: sad path expectation: partial_success + # A single malformed request is recorded both as a decoding error and as the + # rejected request it causes, so two errors are counted for one failing event. + errors_per_failure: 2 events: - log: simple message 1 - log: simple message 2 diff --git a/tests/validation/components/sources/splunk_hec.yaml b/tests/validation/components/sources/splunk_hec.yaml index 2aa081aa4b58c..6bed03471c44e 100644 --- a/tests/validation/components/sources/splunk_hec.yaml +++ b/tests/validation/components/sources/splunk_hec.yaml @@ -9,6 +9,9 @@ event: simple message 3 - name: sad path expectation: partial_success + # A single malformed request is recorded both as a decoding error and as the + # rejected request it causes, so two errors are counted for one failing event. + errors_per_failure: 2 events: - log_builder: event: simple message 1