feat: observable histogram

Author: thomaseizinger

opentelemetry-sdk/src/metrics/instrument.rs

@@ -26,6 +26,8 @@ pub enum InstrumentKind {
     /// A group of instruments that record a distribution of values synchronously with
     /// the code path they are measuring.
     Histogram,
+    /// A group of instruments that record a distribution of values in an asynchronously callback.
+    ObservableHistogram,
     /// A group of instruments that record increasing values in an asynchronous
     /// callback.
     ObservableCounter,
@@ -51,6 +53,7 @@ impl InstrumentKind {
             Temporality::Delta => match self {
                 Self::Counter
                 | Self::Histogram
+                | Self::ObservableHistogram
                 | Self::ObservableCounter
                 | Self::Gauge
                 | Self::ObservableGauge => Temporality::Delta,
@@ -60,7 +63,8 @@ impl InstrumentKind {
             },
             Temporality::LowMemory => match self {
                 Self::Counter | InstrumentKind::Histogram => Temporality::Delta,
-                Self::ObservableCounter
+                Self::ObservableHistogram
+                | Self::ObservableCounter
                 | Self::Gauge
                 | Self::ObservableGauge
                 | Self::UpDownCounter

opentelemetry-sdk/src/metrics/meter.rs

@@ -4,9 +4,9 @@ use std::{borrow::Cow, sync::Arc};
 
 use opentelemetry::{
     metrics::{
-        AsyncInstrumentBuilder, Counter, Gauge, Histogram, HistogramBuilder, InstrumentBuilder,
-        InstrumentProvider, ObservableCounter, ObservableGauge, ObservableUpDownCounter,
-        UpDownCounter,
+        AsyncHistogramBuilder, AsyncInstrumentBuilder, Counter, Gauge, Histogram, HistogramBuilder,
+        InstrumentBuilder, InstrumentProvider, ObservableCounter, ObservableGauge,
+        ObservableHistogram, ObservableUpDownCounter, UpDownCounter,
     },
     otel_error, InstrumentationScope,
 };
@@ -124,7 +124,7 @@ impl SdkMeter {
         let validation_result = validate_instrument_config(builder.name.as_ref(), &builder.unit);
         if let Err(err) = validation_result {
             otel_error!(
-                name: "InstrumentCreationFailed", 
+                name: "InstrumentCreationFailed",
                 meter_name = self.scope.name(),
                 instrument_name = builder.name.as_ref(),
                 message = "Callbacks for this ObservableCounter will not be invoked.",
@@ -183,7 +183,7 @@ impl SdkMeter {
         let validation_result = validate_instrument_config(builder.name.as_ref(), &builder.unit);
         if let Err(err) = validation_result {
             otel_error!(
-                name: "InstrumentCreationFailed", 
+                name: "InstrumentCreationFailed",
                 meter_name = self.scope.name(),
                 instrument_name = builder.name.as_ref(),
                 message = "Callbacks for this ObservableUpDownCounter will not be invoked.",
@@ -242,7 +242,7 @@ impl SdkMeter {
         let validation_result = validate_instrument_config(builder.name.as_ref(), &builder.unit);
         if let Err(err) = validation_result {
             otel_error!(
-                name: "InstrumentCreationFailed", 
+                name: "InstrumentCreationFailed",
                 meter_name = self.scope.name(),
                 instrument_name = builder.name.as_ref(),
                 message = "Callbacks for this ObservableGauge will not be invoked.",
@@ -290,6 +290,65 @@ impl SdkMeter {
         }
     }
 
+    fn create_observable_histogram<T>(
+        &self,
+        builder: AsyncHistogramBuilder<'_, ObservableHistogram<T>, T>,
+        resolver: &InstrumentResolver<'_, T>,
+    ) -> ObservableHistogram<T>
+    where
+        T: Number,
+    {
+        let validation_result = validate_instrument_config(builder.name.as_ref(), &builder.unit);
+        if let Err(err) = validation_result {
+            otel_error!(
+                name: "InstrumentCreationFailed",
+                meter_name = self.scope.name(),
+                instrument_name = builder.name.as_ref(),
+                message = "Callbacks for this ObservableHistogram will not be invoked.",
+                reason = format!("{}", err));
+            return ObservableHistogram::new();
+        }
+
+        match resolver.measures(
+            InstrumentKind::ObservableHistogram,
+            builder.name.clone(),
+            builder.description,
+            builder.unit,
+            None,
+        ) {
+            Ok(ms) => {
+                if ms.is_empty() {
+                    otel_error!(
+                        name: "InstrumentCreationFailed",
+                        meter_name = self.scope.name(),
+                        instrument_name = builder.name.as_ref(),
+                        message = "Callbacks for this ObservableHistogram will not be invoked. Check View Configuration."
+                    );
+                    return ObservableHistogram::new();
+                }
+
+                let observable = Arc::new(Observable::new(ms));
+
+                for callback in builder.callbacks {
+                    let cb_inst = Arc::clone(&observable);
+                    self.pipes
+                        .register_callback(move || callback(cb_inst.as_ref()));
+                }
+
+                ObservableHistogram::new()
+            }
+            Err(err) => {
+                otel_error!(
+                    name: "InstrumentCreationFailed",
+                    meter_name = self.scope.name(),
+                    instrument_name = builder.name.as_ref(),
+                    message = "Callbacks for this ObservableHistogram will not be invoked.",
+                    reason = format!("{}", err));
+                ObservableHistogram::new()
+            }
+        }
+    }
+
     fn create_updown_counter<T>(
         &self,
         builder: InstrumentBuilder<'_, UpDownCounter<T>>,
@@ -548,6 +607,22 @@ impl InstrumentProvider for SdkMeter {
         let resolver = InstrumentResolver::new(self, &self.u64_resolver);
         self.create_histogram(builder, &resolver)
     }
+
+    fn f64_observable_histogram(
+        &self,
+        builder: AsyncHistogramBuilder<'_, ObservableHistogram<f64>, f64>,
+    ) -> ObservableHistogram<f64> {
+        let resolver = InstrumentResolver::new(self, &self.f64_resolver);
+        self.create_observable_histogram(builder, &resolver)
+    }
+
+    fn u64_observable_histogram(
+        &self,
+        builder: AsyncHistogramBuilder<'_, ObservableHistogram<u64>, u64>,
+    ) -> ObservableHistogram<u64> {
+        let resolver = InstrumentResolver::new(self, &self.u64_resolver);
+        self.create_observable_histogram(builder, &resolver)
+    }
 }
 
 fn validate_instrument_config(name: &str, unit: &Option<Cow<'static, str>>) -> MetricResult<()> {

opentelemetry-sdk/src/metrics/pipeline.rs

@@ -482,6 +482,7 @@ where
 /// * Gauge ⇨ LastValue
 /// * Observable Gauge ⇨ LastValue
 /// * Histogram ⇨ ExplicitBucketHistogram
+/// * Observable Histogram ⇨ ExplicitBucketHistogram
 ///
 /// [the spec]: https://github.com/open-telemetry/opentelemetry-specification/blob/v1.19.0/specification/metrics/sdk.md#default-aggregation
 fn default_aggregation_selector(kind: InstrumentKind) -> Aggregation {
@@ -492,13 +493,15 @@ fn default_aggregation_selector(kind: InstrumentKind) -> Aggregation {
         | InstrumentKind::ObservableUpDownCounter => Aggregation::Sum,
         InstrumentKind::Gauge => Aggregation::LastValue,
         InstrumentKind::ObservableGauge => Aggregation::LastValue,
-        InstrumentKind::Histogram => Aggregation::ExplicitBucketHistogram {
-            boundaries: vec![
-                0.0, 5.0, 10.0, 25.0, 50.0, 75.0, 100.0, 250.0, 500.0, 750.0, 1000.0, 2500.0,
-                5000.0, 7500.0, 10000.0,
-            ],
-            record_min_max: true,
-        },
+        InstrumentKind::Histogram | InstrumentKind::ObservableHistogram => {
+            Aggregation::ExplicitBucketHistogram {
+                boundaries: vec![
+                    0.0, 5.0, 10.0, 25.0, 50.0, 75.0, 100.0, 250.0, 500.0, 750.0, 1000.0, 2500.0,
+                    5000.0, 7500.0, 10000.0,
+                ],
+                record_min_max: true,
+            }
+        }
     }
 }
 
@@ -529,6 +532,7 @@ fn aggregate_fn<T: Number>(
                 // MetricReader.Collect MUST only receive data points with measurements recorded since the previous collection
                 InstrumentKind::ObservableCounter => b.precomputed_sum(true),
                 InstrumentKind::ObservableUpDownCounter => b.precomputed_sum(false),
+                InstrumentKind::ObservableHistogram => b.precomputed_sum(false),
                 InstrumentKind::Counter | InstrumentKind::Histogram => b.sum(true),
                 _ => b.sum(false),
             };
@@ -583,6 +587,7 @@ fn aggregate_fn<T: Number>(
 /// | Counter                  | ✓    |           | ✓   | ✓         | ✓                     |
 /// | UpDownCounter            | ✓    |           | ✓   | ✓         | ✓                     |
 /// | Histogram                | ✓    |           | ✓   | ✓         | ✓                     |
+/// | Observable Histogram     | ✓    |           | ✓   | ✓         | ✓                     |
 /// | Observable Counter       | ✓    |           | ✓   | ✓         | ✓                     |
 /// | Observable UpDownCounter | ✓    |           | ✓   | ✓         | ✓                     |
 /// | Gauge                    | ✓    | ✓         |     | ✓         | ✓                     |
@@ -601,6 +606,7 @@ fn is_aggregator_compatible(
                     | InstrumentKind::UpDownCounter
                     | InstrumentKind::Gauge
                     | InstrumentKind::Histogram
+                    | InstrumentKind::ObservableHistogram
                     | InstrumentKind::ObservableCounter
                     | InstrumentKind::ObservableUpDownCounter
                     | InstrumentKind::ObservableGauge
@@ -615,6 +621,7 @@ fn is_aggregator_compatible(
                 | InstrumentKind::ObservableUpDownCounter
                 | InstrumentKind::Counter
                 | InstrumentKind::Histogram
+                | InstrumentKind::ObservableHistogram
                 | InstrumentKind::UpDownCounter => Ok(()),
                 _ => {
                     // TODO: review need for aggregation check after

opentelemetry/src/metrics/instruments/histogram.rs

@@ -33,3 +33,32 @@ impl<T> Histogram<T> {
         self.0.measure(value, attributes)
     }
 }
+
+/// An async instrument that records a distribution of values.
+#[derive(Clone)]
+#[non_exhaustive]
+pub struct ObservableHistogram<T> {
+    _marker: std::marker::PhantomData<T>,
+}
+
+impl<T> fmt::Debug for ObservableHistogram<T>
+where
+    T: fmt::Debug,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_fmt(format_args!(
+            "ObservableHistogram<{}>",
+            std::any::type_name::<T>()
+        ))
+    }
+}
+
+impl<T> ObservableHistogram<T> {
+    /// Create a new gauge
+    #[allow(clippy::new_without_default)]
+    pub fn new() -> Self {
+        ObservableHistogram {
+            _marker: std::marker::PhantomData,
+        }
+    }
+}