Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions opentelemetry-sdk/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@
- `id_generator`, `should_sample`

[3227]: https://github.com/open-telemetry/opentelemetry-rust/pull/3227
- Fixed Sum and PrecomputedSum not to accept negative values if monotonic
[#3260](https://github.com/open-telemetry/opentelemetry-rust/pull/3260)

## 0.31.0

Expand Down
102 changes: 101 additions & 1 deletion opentelemetry-sdk/src/metrics/internal/precomputed_sum.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
use opentelemetry::KeyValue;
use opentelemetry::{otel_warn, KeyValue};

use crate::metrics::data::{self, AggregatedMetrics, MetricData, SumDataPoint};
use crate::metrics::Temporality;
Expand Down Expand Up @@ -132,6 +132,16 @@ where
T: Number,
{
fn call(&self, measurement: T, attrs: &[KeyValue]) {
// Validate monotonic counter increment is non-negative
if self.monotonic && measurement < T::default() {
otel_warn!(
name: "ObservableCounter.NegativeValue",
message = "Observable counters are monotonic and can only accept non-negative values. This measurement will be dropped.",
value = format!("{:?}", measurement)
);
return;
}

self.filter.apply(attrs, |filtered| {
self.value_map.measure(measurement, filtered);
})
Expand All @@ -151,3 +161,93 @@ where
(len, new.map(T::make_aggregated_metrics))
}
}

#[cfg(test)]
mod tests {
use super::*;

#[test]
fn precomputed_sum_monotonic_rejects_negative_values() {
let sum = PrecomputedSum::<f64>::new(
Temporality::Cumulative,
AttributeSetFilter::new(None),
true, // monotonic = true
100,
);

Measure::call(&sum, 5.0, &[]);
Measure::call(&sum, -3.0, &[]); // Should be dropped

let (_len, metrics) = sum.cumulative(None);
assert!(metrics.is_some(), "Should have metrics");
let metrics = metrics.unwrap();

assert!(
matches!(metrics, MetricData::Sum(_)),
"Expected Sum metric data"
);
let MetricData::Sum(sum_data) = metrics else {
unreachable!()
};

assert_eq!(sum_data.data_points.len(), 1);
// Only the positive value should be recorded
assert_eq!(sum_data.data_points[0].value, 5.0);
}

#[test]
fn precomputed_sum_non_monotonic_accepts_negative_values() {
let sum = PrecomputedSum::<f64>::new(
Temporality::Cumulative,
AttributeSetFilter::new(None),
false, // monotonic = false
100,
);

Measure::call(&sum, 5.0, &[]);
Measure::call(&sum, -3.0, &[]); // Should be accepted

let (_len, metrics) = sum.cumulative(None);
assert!(metrics.is_some(), "Should have metrics");
let metrics = metrics.unwrap();

assert!(
matches!(metrics, MetricData::Sum(_)),
"Expected Sum metric data"
);
let MetricData::Sum(sum_data) = metrics else {
unreachable!()
};

// Both values should be recorded (precomputed sum overwrites, so last value wins)
assert_eq!(sum_data.data_points.len(), 1);
assert_eq!(sum_data.data_points[0].value, -3.0);
}

#[test]
fn precomputed_sum_monotonic_accepts_zero() {
let sum = PrecomputedSum::<f64>::new(
Temporality::Cumulative,
AttributeSetFilter::new(None),
true, // monotonic = true
100,
);

Measure::call(&sum, 0.0, &[]);

let (_len, metrics) = sum.cumulative(None);
assert!(metrics.is_some(), "Should have metrics");
let metrics = metrics.unwrap();

assert!(
matches!(metrics, MetricData::Sum(_)),
"Expected Sum metric data"
);
let MetricData::Sum(sum_data) = metrics else {
unreachable!()
};

assert_eq!(sum_data.data_points.len(), 1);
assert_eq!(sum_data.data_points[0].value, 0.0);
}
}
130 changes: 129 additions & 1 deletion opentelemetry-sdk/src/metrics/internal/sum.rs
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
use crate::metrics::data::{self, AggregatedMetrics, MetricData, SumDataPoint};
use crate::metrics::Temporality;
use opentelemetry::KeyValue;
use opentelemetry::{otel_warn, KeyValue};

use super::aggregate::{AggregateTimeInitiator, AttributeSetFilter};
use super::{Aggregator, AtomicTracker, ComputeAggregation, Measure, Number};
Expand Down Expand Up @@ -149,6 +149,16 @@ where
T: Number,
{
fn call(&self, measurement: T, attrs: &[KeyValue]) {
// Validate monotonic counter increment is non-negative
if self.monotonic && measurement < T::default() {
otel_warn!(
name: "Counter.NegativeValue",
message = "Counters are monotonic and can only accept non-negative values. This measurement will be dropped.",
value = format!("{:?}", measurement)
);
return;
}

self.filter.apply(attrs, |filtered| {
self.value_map.measure(measurement, filtered);
})
Expand All @@ -168,3 +178,121 @@ where
(len, new.map(T::make_aggregated_metrics))
}
}

#[cfg(test)]
mod tests {
use super::*;

#[test]
fn sum_monotonic_rejects_negative_values() {
let sum = Sum::<f64>::new(
Temporality::Cumulative,
AttributeSetFilter::new(None),
true, // monotonic = true
100,
);

Measure::call(&sum, 5.0, &[]);
Measure::call(&sum, -3.0, &[]);

let (_len, metrics) = sum.cumulative(None);
assert!(metrics.is_some(), "Should have metrics");
let metrics = metrics.unwrap();

assert!(
matches!(metrics, MetricData::Sum(_)),
"Expected Sum metric data"
);
let MetricData::Sum(sum_data) = metrics else {
unreachable!()
};

assert_eq!(sum_data.data_points.len(), 1);
assert_eq!(sum_data.data_points[0].value, 5.0);
}

#[test]
fn sum_non_monotonic_accepts_negative_values() {
// Create a non-monotonic sum (up-down counter)
let sum = Sum::<f64>::new(
Temporality::Cumulative,
AttributeSetFilter::new(None),
false, // monotonic = false
100,
);

Measure::call(&sum, 5.0, &[]);
Measure::call(&sum, -3.0, &[]);

let (_len, metrics) = sum.cumulative(None);
assert!(metrics.is_some(), "Should have metrics");
let metrics = metrics.unwrap();

assert!(
matches!(metrics, MetricData::Sum(_)),
"Expected Sum metric data"
);
let MetricData::Sum(sum_data) = metrics else {
unreachable!()
};

assert_eq!(sum_data.data_points.len(), 1);
// Both values should be summed: 5.0 + (-3.0) = 2.0
assert_eq!(sum_data.data_points[0].value, 2.0);
}

#[test]
fn sum_monotonic_accepts_zero() {
let sum = Sum::<f64>::new(
Temporality::Cumulative,
AttributeSetFilter::new(None),
true,
100,
);

Measure::call(&sum, 0.0, &[]);

let (_len, metrics) = sum.cumulative(None);
assert!(metrics.is_some(), "Should have metrics");
let metrics = metrics.unwrap();

assert!(
matches!(metrics, MetricData::Sum(_)),
"Expected Sum metric data"
);
let MetricData::Sum(sum_data) = metrics else {
unreachable!()
};

assert_eq!(sum_data.data_points.len(), 1);
assert_eq!(sum_data.data_points[0].value, 0.0);
}

#[test]
fn sum_monotonic_rejects_negative_i64() {
let sum = Sum::<i64>::new(
Temporality::Cumulative,
AttributeSetFilter::new(None),
true, // monotonic = true
100,
);

Measure::call(&sum, 10, &[]);
Measure::call(&sum, -5, &[]);

let (_len, metrics) = sum.cumulative(None);
assert!(metrics.is_some(), "Should have metrics");
let metrics = metrics.unwrap();

assert!(
matches!(metrics, MetricData::Sum(_)),
"Expected Sum metric data"
);
let MetricData::Sum(sum_data) = metrics else {
unreachable!()
};

assert_eq!(sum_data.data_points.len(), 1);
assert_eq!(sum_data.data_points[0].value, 10);
}
}
4 changes: 4 additions & 0 deletions opentelemetry/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,10 @@
- `trace_id`, `span_id`, `end_time`, `status`, `sampling_result`
- `with_trace_id`, `with_span_id`, `with_end_time`, `with_status`, `with_sampling_result`
- **Added** `#[must_use]` attribute to `opentelemetry::metrics::AsyncInstrumentBuilder` to add compile time warning when `.build()` is not called on observable instrument builders, preventing silent failures where callbacks are never registered and metrics are never reported.
- **Documentation** Enhanced documentation for `Counter` and `ObservableCounter` to clarify monotonic behavior:
- Updated `Counter` and `ObservableCounter` struct documentation to explicitly state they are monotonic instruments that only accept non-negative values
- Enhanced `Counter::add()` method documentation to specify that negative values violate the monotonic contract and will be dropped by the SDK
- Updated `InstrumentProvider` trait method documentation for counter creation methods to clarify monotonic behavior

[3227]: https://github.com/open-telemetry/opentelemetry-rust/pull/3227

Expand Down
30 changes: 28 additions & 2 deletions opentelemetry/src/metrics/instruments/counter.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,14 @@ use std::sync::Arc;

use super::SyncInstrument;

/// An instrument that records increasing values.
/// A monotonic instrument that records increasing values.
///
/// Counters are used to measure values that only increase over time, such as the number
/// of requests received, bytes sent, or errors encountered. Only non-negative values
/// should be recorded. Negative values violate the monotonic property and will be
/// dropped by the SDK with a warning.
///
/// # Cloning
///
/// [`Counter`] can be cloned to create multiple handles to the same instrument. If a [`Counter`] needs to be shared,
/// users are recommended to clone the [`Counter`] instead of creating duplicate [`Counter`]s for the same metric. Creating
Expand All @@ -29,12 +36,31 @@ impl<T> Counter<T> {
}

/// Records an increment to the counter.
///
/// # Arguments
///
/// * `value` - A non-negative value to add to the counter. According to the
/// OpenTelemetry specification, counters are monotonic instruments that record
/// increasing values. Passing a negative value violates this contract.
///
/// * `attributes` - A set of key-value pairs that describe the measurement context.
///
/// # Behavior with negative values
///
/// The API does not validate the value, but the SDK implementation will log a warning
/// and drop negative values to maintain the monotonic property of counters. Applications
/// should ensure only non-negative values are passed to this method.
pub fn add(&self, value: T, attributes: &[KeyValue]) {
self.0.measure(value, attributes)
}
}

/// An async instrument that records increasing values.
/// A monotonic asynchronous instrument that records increasing values.
///
/// Observable counters are used to measure values that only increase over time and are
/// observed via callbacks, such as process CPU time or total memory usage. Only non-negative
/// values should be recorded. Negative values violate the monotonic property and will be
/// dropped by the SDK with a warning.
#[derive(Clone)]
#[non_exhaustive]
pub struct ObservableCounter<T> {
Expand Down
16 changes: 12 additions & 4 deletions opentelemetry/src/metrics/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -18,25 +18,33 @@ pub use noop::NoopMeterProvider;

/// SDK implemented trait for creating instruments
pub trait InstrumentProvider {
/// creates an instrument for recording increasing values.
/// Creates a monotonic counter for recording increasing values.
///
/// Counters only accept non-negative values. Negative values will be dropped by the SDK.
fn u64_counter(&self, _builder: InstrumentBuilder<'_, Counter<u64>>) -> Counter<u64> {
Counter::new(Arc::new(noop::NoopSyncInstrument::new()))
}

/// creates an instrument for recording increasing values.
/// Creates a monotonic counter for recording increasing values.
///
/// Counters only accept non-negative values. Negative values will be dropped by the SDK.
fn f64_counter(&self, _builder: InstrumentBuilder<'_, Counter<f64>>) -> Counter<f64> {
Counter::new(Arc::new(noop::NoopSyncInstrument::new()))
}

/// creates an instrument for recording increasing values via callback.
/// Creates a monotonic observable counter for recording increasing values via callback.
///
/// Observable counters only accept non-negative values. Negative values will be dropped by the SDK.
fn u64_observable_counter(
&self,
_builder: AsyncInstrumentBuilder<'_, ObservableCounter<u64>, u64>,
) -> ObservableCounter<u64> {
ObservableCounter::new()
}

/// creates an instrument for recording increasing values via callback.
/// Creates a monotonic observable counter for recording increasing values via callback.
///
/// Observable counters only accept non-negative values. Negative values will be dropped by the SDK.
fn f64_observable_counter(
&self,
_builder: AsyncInstrumentBuilder<'_, ObservableCounter<f64>, f64>,
Expand Down