Add non monotonic counter support to prometheus (#385)

Author: jtescher

examples/basic-otlp/src/main.rs

@@ -61,7 +61,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
 
     let value_recorder_two = meter.f64_value_recorder("ex.com.two").init();
 
-    let _correlations =
+    let _baggage =
         Context::current_with_baggage(vec![FOO_KEY.string("foo1"), BAR_KEY.string("bar1")])
             .attach();
 

examples/basic/src/main.rs

@@ -67,7 +67,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
 
     let value_recorder_two = meter.f64_value_recorder("ex.com.two").init();
 
-    let _correlations =
+    let _baggage =
         Context::current_with_baggage(vec![FOO_KEY.string("foo1"), BAR_KEY.string("bar1")])
             .attach();
 

opentelemetry-prometheus/src/lib.rs

@@ -66,9 +66,9 @@
 
 use opentelemetry::global;
 use opentelemetry::sdk::{
-    export::metrics::{CheckpointSet, ExportKind, Histogram, Record, Sum},
+    export::metrics::{CheckpointSet, ExportKindSelector, Histogram, LastValue, Record, Sum},
     metrics::{
-        aggregators::{HistogramAggregator, SumAggregator},
+        aggregators::{HistogramAggregator, LastValueAggregator, SumAggregator},
         controllers,
         selectors::simple::Selector,
         PullController,
@@ -77,9 +77,7 @@ use opentelemetry::sdk::{
 };
 use opentelemetry::{
     labels,
-    metrics::{
-        registry::RegistryMeterProvider, Descriptor, InstrumentKind, MetricsError, NumberKind,
-    },
+    metrics::{registry::RegistryMeterProvider, MetricsError, NumberKind},
     KeyValue,
 };
 use std::collections::HashMap;
@@ -92,7 +90,8 @@ use sanitize::sanitize;
 
 /// Cache disabled by default.
 const DEFAULT_CACHE_PERIOD: Duration = Duration::from_secs(0);
-const EXPORT_KIND: ExportKind = ExportKind::Cumulative;
+
+const EXPORT_KIND_SELECTOR: ExportKindSelector = ExportKindSelector::Cumulative;
 
 /// Create a new prometheus exporter builder.
 pub fn exporter() -> ExporterBuilder {
@@ -181,7 +180,7 @@ impl ExporterBuilder {
             .default_histogram_boundaries
             .unwrap_or_else(|| vec![0.5, 0.9, 0.99]);
         let selector = Box::new(Selector::Histogram(default_histogram_boundaries.clone()));
-        let mut controller_builder = controllers::pull(selector, Box::new(EXPORT_KIND))
+        let mut controller_builder = controllers::pull(selector, Box::new(EXPORT_KIND_SELECTOR))
             .with_cache_period(self.cache_period.unwrap_or(DEFAULT_CACHE_PERIOD))
             .with_memory(true);
         if let Some(resource) = self.resource {
@@ -256,17 +255,6 @@ impl PrometheusExporter {
             .map_err(Into::into)
             .map(|locked| locked.provider())
     }
-
-    /// Determine the export kind this exporter should use for a given instrument
-    /// and descriptor.
-    pub fn export_kind_for(&self, _descriptor: &Descriptor, _kind: &InstrumentKind) -> ExportKind {
-        // NOTE: Summary values should use Delta aggregation, then be
-        // combined into a sliding window, see the TODO below.
-        // NOTE: Prometheus also supports a "GaugeDelta" exposition format,
-        // which is expressed as a delta histogram.  Need to understand if this
-        // should be a default behavior for ValueRecorder/ValueObserver.
-        EXPORT_KIND
-    }
 }
 
 #[derive(Debug)]
@@ -296,9 +284,10 @@ impl prometheus::core::Collector for Collector {
                 return metrics;
             }
 
-            if let Err(err) = controller.try_for_each(&EXPORT_KIND, &mut |record| {
+            if let Err(err) = controller.try_for_each(&EXPORT_KIND_SELECTOR, &mut |record| {
                 let agg = record.aggregator().ok_or(MetricsError::NoDataCollected)?;
                 let number_kind = record.descriptor().number_kind();
+                let instrument_kind = record.descriptor().instrument_kind();
 
                 let mut label_keys = Vec::new();
                 let mut label_values = Vec::new();
@@ -309,7 +298,15 @@ impl prometheus::core::Collector for Collector {
                 if let Some(hist) = agg.as_any().downcast_ref::<HistogramAggregator>() {
                     metrics.push(build_histogram(hist, number_kind, desc, label_values)?);
                 } else if let Some(sum) = agg.as_any().downcast_ref::<SumAggregator>() {
-                    metrics.push(build_counter(sum, number_kind, desc, label_values)?);
+                    let counter = if instrument_kind.monotonic() {
+                        build_monotonic_counter(sum, number_kind, desc, label_values)?
+                    } else {
+                        build_non_monotonic_counter(sum, number_kind, desc, label_values)?
+                    };
+
+                    metrics.push(counter);
+                } else if let Some(last) = agg.as_any().downcast_ref::<LastValueAggregator>() {
+                    metrics.push(build_last_value(last, number_kind, desc, label_values)?);
                 }
 
                 Ok(())
@@ -324,7 +321,59 @@ impl prometheus::core::Collector for Collector {
     }
 }
 
-fn build_counter(
+fn build_last_value(
+    lv: &LastValueAggregator,
+    kind: &NumberKind,
+    desc: prometheus::core::Desc,
+    labels: Vec<KeyValue>,
+) -> Result<prometheus::proto::MetricFamily, MetricsError> {
+    let (last_value, _) = lv.last_value()?;
+
+    let mut g = prometheus::proto::Gauge::default();
+    g.set_value(last_value.to_f64(kind));
+
+    let mut m = prometheus::proto::Metric::default();
+    m.set_label(protobuf::RepeatedField::from_vec(
+        labels.into_iter().map(build_label_pair).collect(),
+    ));
+    m.set_gauge(g);
+
+    let mut mf = prometheus::proto::MetricFamily::default();
+    mf.set_name(desc.fq_name);
+    mf.set_help(desc.help);
+    mf.set_field_type(prometheus::proto::MetricType::GAUGE);
+    mf.set_metric(protobuf::RepeatedField::from_vec(vec![m]));
+
+    Ok(mf)
+}
+
+fn build_non_monotonic_counter(
+    sum: &SumAggregator,
+    kind: &NumberKind,
+    desc: prometheus::core::Desc,
+    labels: Vec<KeyValue>,
+) -> Result<prometheus::proto::MetricFamily, MetricsError> {
+    let sum = sum.sum()?;
+
+    let mut g = prometheus::proto::Gauge::default();
+    g.set_value(sum.to_f64(kind));
+
+    let mut m = prometheus::proto::Metric::default();
+    m.set_label(protobuf::RepeatedField::from_vec(
+        labels.into_iter().map(build_label_pair).collect(),
+    ));
+    m.set_gauge(g);
+
+    let mut mf = prometheus::proto::MetricFamily::default();
+    mf.set_name(desc.fq_name);
+    mf.set_help(desc.help);
+    mf.set_field_type(prometheus::proto::MetricType::GAUGE);
+    mf.set_metric(protobuf::RepeatedField::from_vec(vec![m]));
+
+    Ok(mf)
+}
+
+fn build_monotonic_counter(
     sum: &SumAggregator,
     kind: &NumberKind,
     desc: prometheus::core::Desc,

opentelemetry-prometheus/tests/integration_test.rs

@@ -1,5 +1,8 @@
 use opentelemetry::sdk::Resource;
-use opentelemetry::{metrics::MeterProvider, KeyValue};
+use opentelemetry::{
+    metrics::{MeterProvider, ObserverResult},
+    KeyValue,
+};
 use opentelemetry_prometheus::PrometheusExporter;
 use prometheus::{Encoder, TextEncoder};
 
@@ -12,6 +15,7 @@ fn test_add() {
 
     let meter = exporter.provider().unwrap().meter("test");
 
+    let up_down_counter = meter.f64_up_down_counter("updowncounter").init();
     let counter = meter.f64_counter("counter").init();
     let value_recorder = meter.f64_value_recorder("value_recorder").init();
 
@@ -22,18 +26,32 @@ fn test_add() {
     counter.add(10.0, &labels);
     counter.add(5.3, &labels);
 
-    expected.push("counter{A=\"B\",C=\"D\",R=\"V\"} 15.3");
+    expected.push(r#"counter{A="B",C="D",R="V"} 15.3"#);
+
+    let cb_labels = labels.clone();
+    let _observer = meter
+        .i64_value_observer("intobserver", move |result: ObserverResult<i64>| {
+            result.observe(1, cb_labels.as_ref())
+        })
+        .init();
+
+    expected.push(r#"intobserver{A="B",C="D",R="V"} 1"#);
 
     value_recorder.record(-0.6, &labels);
     value_recorder.record(-0.4, &labels);
     value_recorder.record(0.6, &labels);
     value_recorder.record(20.0, &labels);
 
-    expected.push("value_recorder_bucket{A=\"B\",C=\"D\",R=\"V\",le=\"+Inf\"} 4");
-    expected.push("value_recorder_bucket{A=\"B\",C=\"D\",R=\"V\",le=\"-0.5\"} 1");
-    expected.push("value_recorder_bucket{A=\"B\",C=\"D\",R=\"V\",le=\"1\"} 3");
-    expected.push("value_recorder_count{A=\"B\",C=\"D\",R=\"V\"} 4");
-    expected.push("value_recorder_sum{A=\"B\",C=\"D\",R=\"V\"} 19.6");
+    expected.push(r#"value_recorder_bucket{A="B",C="D",R="V",le="+Inf"} 4"#);
+    expected.push(r#"value_recorder_bucket{A="B",C="D",R="V",le="-0.5"} 1"#);
+    expected.push(r#"value_recorder_bucket{A="B",C="D",R="V",le="1"} 3"#);
+    expected.push(r#"value_recorder_count{A="B",C="D",R="V"} 4"#);
+    expected.push(r#"value_recorder_sum{A="B",C="D",R="V"} 19.6"#);
+
+    up_down_counter.add(10.0, &labels);
+    up_down_counter.add(-3.2, &labels);
+
+    expected.push(r#"updowncounter{A="B",C="D",R="V"} 6.8"#);
 
     compare_export(&exporter, expected)
 }

opentelemetry/src/api/metrics/async_instrument.rs

@@ -69,8 +69,8 @@ where
         }
     }
 
-    /// Observe captures a single integer value from the associated instrument
-    /// callback, with the given labels.
+    /// Observe captures a single value from the associated instrument callback,
+    /// with the given labels.
     pub fn observe(&self, value: T, labels: &[KeyValue]) {
         (self.f)(
             labels,

opentelemetry/src/api/metrics/kind.rs

@@ -51,13 +51,12 @@ impl InstrumentKind {
 
     /// Whether this kind of instrument groups its inputs (as opposed to adding).
     pub fn grouping(&self) -> bool {
-        matches!(
-            self,
-            InstrumentKind::Counter
-                | InstrumentKind::UpDownCounter
-                | InstrumentKind::SumObserver
-                | InstrumentKind::UpDownSumObserver
-        )
+        !self.adding()
+    }
+
+    /// Whether this kind of instrument exposes a non-decreasing sum.
+    pub fn monotonic(&self) -> bool {
+        matches!(self, InstrumentKind::Counter | InstrumentKind::SumObserver)
     }
 
     /// Whether this kind of instrument receives precomputed sums.

opentelemetry/src/api/mod.rs

@@ -1,6 +1,8 @@
 #[cfg(feature = "metrics")]
 #[cfg_attr(docsrs, doc(cfg(feature = "metrics")))]
 use crate::api::metrics::MetricsError;
+#[cfg(feature = "trace")]
+#[cfg_attr(docsrs, doc(cfg(feature = "trace")))]
 use crate::api::trace::TraceError;
 use std::sync::PoisonError;
 

opentelemetry/src/global/error_handler.rs

@@ -21,7 +21,7 @@ pub fn handle_error<T: Into<Error>>(err: T) {
             #[cfg(feature = "trace")]
             #[cfg_attr(docsrs, doc(cfg(feature = "trace")))]
             Error::Trace(err) => eprintln!("OpenTelemetry trace error occurred {:?}", err),
-            Error::Other(err_msg) => println!("OpenTelemetry error occurred {}", err_msg),
+            Error::Other(err_msg) => eprintln!("OpenTelemetry error occurred {}", err_msg),
         },
     }
 }

opentelemetry/src/sdk/export/metrics/mod.rs

@@ -168,7 +168,7 @@ pub trait Subtractor {
 /// Exporter handles presentation of the checkpoint of aggregate metrics. This
 /// is the final stage of a metrics export pipeline, where metric data are
 /// formatted for a specific system.
-pub trait Exporter: ExportKindSelector {
+pub trait Exporter: ExportKindFor {
     /// Export is called immediately after completing a collection pass in the SDK.
     ///
     /// The CheckpointSet interface refers to the Processor that just completed
@@ -179,7 +179,7 @@ pub trait Exporter: ExportKindSelector {
 /// ExportKindSelector is a sub-interface of Exporter used to indicate
 /// whether the Processor should compute Delta or Cumulative
 /// Aggregations.
-pub trait ExportKindSelector: fmt::Debug {
+pub trait ExportKindFor: fmt::Debug {
     /// Determines the correct `ExportKind` that should be used when exporting data
     /// for the given metric instrument.
     fn export_kind_for(&self, descriptor: &Descriptor) -> ExportKind;
@@ -202,7 +202,7 @@ pub trait CheckpointSet: fmt::Debug {
     /// return the error to the caller.
     fn try_for_each(
         &mut self,
-        export_selector: &dyn ExportKindSelector,
+        export_selector: &dyn ExportKindFor,
         f: &mut dyn FnMut(&Record<'_>) -> Result<()>,
     ) -> Result<()>;
 }
@@ -373,15 +373,22 @@ impl<'a> Accumulation<'a> {
 #[derive(Clone, Debug)]
 pub enum ExportKind {
     /// Indicates that the `Exporter` expects a cumulative `Aggregation`.
-    Cumulative = 1, // e.g., Prometheus
+    Cumulative = 1,
 
     /// Indicates that the `Exporter` expects a delta `Aggregation`.
-    Delta = 2, // e.g., StatsD
+    Delta = 2,
+}
 
-    /// Indicates that the `Exporter` expects either a cumulative or a delta
-    /// `Aggregation`, whichever does not require maintaining state for the
-    /// given instrument.
-    PassThrough = 4, // e.g., OTLP
+/// Strategies for selecting which export kind is used for an instrument.
+#[derive(Debug)]
+pub enum ExportKindSelector {
+    /// A selector that always returns [`ExportKind::Cumulative`].
+    Cumulative,
+    /// A selector that always returns [`ExportKind::Delta`].
+    Delta,
+    /// A selector that returns cumulative or delta based on a given instrument
+    /// kind.
+    Stateless,
 }
 
 impl ExportKind {
@@ -409,8 +416,18 @@ impl ExportKind {
     }
 }
 
-impl ExportKindSelector for ExportKind {
-    fn export_kind_for(&self, _descriptor: &Descriptor) -> ExportKind {
-        self.clone()
+impl ExportKindFor for ExportKindSelector {
+    fn export_kind_for(&self, descriptor: &Descriptor) -> ExportKind {
+        match self {
+            ExportKindSelector::Cumulative => ExportKind::Cumulative,
+            ExportKindSelector::Delta => ExportKind::Delta,
+            ExportKindSelector::Stateless => {
+                if descriptor.instrument_kind().precomputed_sum() {
+                    ExportKind::Cumulative
+                } else {
+                    ExportKind::Delta
+                }
+            }
+        }
     }
 }

opentelemetry/src/sdk/export/metrics/stdout.rs

@@ -2,8 +2,8 @@
 use crate::global;
 use crate::sdk::{
     export::metrics::{
-        CheckpointSet, Count, ExportKind, ExportKindSelector, Exporter, LastValue, Max, Min,
-        Quantile, Sum,
+        CheckpointSet, Count, ExportKind, ExportKindFor, ExportKindSelector, Exporter, LastValue,
+        Max, Min, Quantile, Sum,
     },
     metrics::{
         aggregators::{
@@ -235,16 +235,16 @@ where
     }
 }
 
-impl<W> ExportKindSelector for StdoutExporter<W>
+impl<W> ExportKindFor for StdoutExporter<W>
 where
     W: fmt::Debug + io::Write,
 {
-    fn export_kind_for(&self, _descriptor: &Descriptor) -> ExportKind {
-        ExportKind::PassThrough
+    fn export_kind_for(&self, descriptor: &Descriptor) -> ExportKind {
+        ExportKindSelector::Stateless.export_kind_for(descriptor)
     }
 }
 
-/// A formatter for user-defined batch serilization.
+/// A formatter for user-defined batch serialization.
 pub struct Formatter(Box<dyn Fn(ExportBatch) -> Result<String> + Send + Sync>);
 impl fmt::Debug for Formatter {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@@ -361,7 +361,7 @@ where
         let (spawn, interval, exporter) = self.try_build()?;
         let mut push_builder = controllers::push(
             simple::Selector::Exact,
-            ExportKind::PassThrough,
+            ExportKindSelector::Stateless,
             exporter,
             spawn,
             interval,