feat(opentelemetry-sdk): add HistogramAggregation resolution

Author: michaelbeaumont

opentelemetry-sdk/src/metrics/exporter.rs

@@ -5,7 +5,7 @@ use std::time::Duration;
 
 use crate::metrics::data::ResourceMetrics;
 
-use super::Temporality;
+use super::{HistogramAggregation, Temporality};
 
 /// Exporter handles the delivery of metric data to external receivers.
 ///
@@ -37,4 +37,7 @@ pub trait PushMetricExporter: Send + Sync + 'static {
 
     /// Access the [Temporality] of the MetricExporter.
     fn temporality(&self) -> Temporality;
+
+    /// The default aggregation to use for histogram instruments.
+    fn default_histogram_aggregation(&self) -> HistogramAggregation;
 }

opentelemetry-sdk/src/metrics/in_memory_exporter.rs

@@ -3,7 +3,7 @@ use crate::metrics::data::{
     ExponentialHistogram, Gauge, Histogram, MetricData, ResourceMetrics, Sum,
 };
 use crate::metrics::exporter::PushMetricExporter;
-use crate::metrics::Temporality;
+use crate::metrics::{HistogramAggregation, Temporality};
 use crate::InMemoryExporterError;
 use std::collections::VecDeque;
 use std::fmt;
@@ -261,4 +261,8 @@ impl PushMetricExporter for InMemoryMetricExporter {
     fn temporality(&self) -> Temporality {
         self.temporality
     }
+
+    fn default_histogram_aggregation(&self) -> HistogramAggregation {
+        HistogramAggregation::ExplicitBucketHistogram
+    }
 }

opentelemetry-sdk/src/metrics/mod.rs

@@ -120,6 +120,35 @@ impl FromStr for Temporality {
     }
 }
 
+/// The default histogram aggregation selection for a [`exporter::PushMetricExporter`].
+///
+/// This controls which aggregation type is used for histogram instruments
+/// when no explicit aggregation is configured via a view.
+///
+/// Corresponds to the `OTEL_EXPORTER_OTLP_METRICS_DEFAULT_HISTOGRAM_AGGREGATION`
+/// environment variable.
+#[derive(Debug, Clone, Copy, PartialEq, Default)]
+pub enum HistogramAggregation {
+    /// Use explicit bucket histogram aggregation with default boundaries.
+    #[default]
+    ExplicitBucketHistogram,
+
+    /// Use base2 exponential bucket histogram aggregation.
+    Base2ExponentialBucketHistogram,
+}
+
+impl FromStr for HistogramAggregation {
+    type Err = ();
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s.to_lowercase().as_str() {
+            "explicit_bucket_histogram" => Ok(Self::ExplicitBucketHistogram),
+            "base2_exponential_bucket_histogram" => Ok(Self::Base2ExponentialBucketHistogram),
+            _ => Err(()),
+        }
+    }
+}
+
 #[cfg(all(test, feature = "testing"))]
 mod tests {
     use self::data::{HistogramDataPoint, MetricData, ScopeMetrics, SumDataPoint};

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -18,7 +18,7 @@ use crate::{
 
 use super::{
     data::ResourceMetrics, instrument::InstrumentKind, pipeline::Pipeline, reader::MetricReader,
-    Temporality,
+    HistogramAggregation, Temporality,
 };
 
 const DEFAULT_INTERVAL: Duration = Duration::from_secs(60);
@@ -363,6 +363,10 @@ impl<E: PushMetricExporter> PeriodicReaderInner<E> {
         self.exporter.temporality()
     }
 
+    fn default_histogram_aggregation(&self) -> HistogramAggregation {
+        self.exporter.default_histogram_aggregation()
+    }
+
     fn collect(&self, rm: &mut ResourceMetrics) -> OTelSdkResult {
         let producer = self.producer.lock().expect("lock poisoned");
         if let Some(p) = producer.as_ref() {
@@ -506,6 +510,10 @@ impl<E: PushMetricExporter> MetricReader for PeriodicReader<E> {
     fn temporality(&self, kind: InstrumentKind) -> Temporality {
         kind.temporality_preference(self.inner.temporality(kind))
     }
+
+    fn default_histogram_aggregation(&self) -> HistogramAggregation {
+        self.inner.default_histogram_aggregation()
+    }
 }
 
 #[cfg(all(test, feature = "testing"))]
@@ -515,7 +523,7 @@ mod tests {
         error::{OTelSdkError, OTelSdkResult},
         metrics::{
             data::ResourceMetrics, exporter::PushMetricExporter, reader::MetricReader,
-            InMemoryMetricExporter, SdkMeterProvider, Temporality,
+            HistogramAggregation, InMemoryMetricExporter, SdkMeterProvider, Temporality,
         },
         Resource,
     };
@@ -574,6 +582,10 @@ mod tests {
         fn temporality(&self) -> Temporality {
             Temporality::Cumulative
         }
+
+        fn default_histogram_aggregation(&self) -> HistogramAggregation {
+            HistogramAggregation::ExplicitBucketHistogram
+        }
     }
 
     #[derive(Debug, Clone, Default)]
@@ -602,6 +614,10 @@ mod tests {
         fn temporality(&self) -> Temporality {
             Temporality::Cumulative
         }
+
+        fn default_histogram_aggregation(&self) -> HistogramAggregation {
+            HistogramAggregation::ExplicitBucketHistogram
+        }
     }
 
     #[test]

opentelemetry-sdk/src/metrics/periodic_reader_with_async_runtime.rs

@@ -22,6 +22,7 @@ use crate::{
 
 use super::{
     data::ResourceMetrics, instrument::InstrumentKind, pipeline::Pipeline, reader::MetricReader,
+    HistogramAggregation,
 };
 
 const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
@@ -437,6 +438,10 @@ impl<E: PushMetricExporter> MetricReader for PeriodicReader<E> {
     fn temporality(&self, kind: InstrumentKind) -> super::Temporality {
         kind.temporality_preference(self.exporter.temporality())
     }
+
+    fn default_histogram_aggregation(&self) -> HistogramAggregation {
+        self.exporter.default_histogram_aggregation()
+    }
 }
 
 #[cfg(all(test, feature = "testing"))]

opentelemetry-sdk/src/metrics/pipeline.rs

@@ -23,7 +23,7 @@ use crate::{
 
 use self::internal::AggregateFns;
 
-use super::{aggregation::Aggregation, Temporality};
+use super::{aggregation::Aggregation, HistogramAggregation, Temporality};
 
 /// Connects all of the instruments created by a meter provider to a [MetricReader].
 ///
@@ -378,14 +378,15 @@ where
         // TODO: Create a separate pub (crate) Stream struct for the pipeline,
         // as Stream will not have any optional fields as None at this point and
         // new struct can better reflect this.
+        let histogram_agg = self.pipeline.reader.default_histogram_aggregation();
         let mut agg = stream
             .aggregation
             .take()
-            .unwrap_or_else(|| default_aggregation_selector(kind));
+            .unwrap_or_else(|| default_aggregation_selector(kind, histogram_agg));
 
         // Apply default if stream or reader aggregation returns default
         if matches!(agg, aggregation::Aggregation::Default) {
-            agg = default_aggregation_selector(kind);
+            agg = default_aggregation_selector(kind, histogram_agg);
         }
 
         if let Err(err) = is_aggregator_compatible(&kind, &agg) {
@@ -421,7 +422,8 @@ where
                 filter,
                 cardinality_limit,
             );
-            let AggregateFns { measure, collect } = match aggregate_fn(b, &agg, kind) {
+            let AggregateFns { measure, collect } = match aggregate_fn(b, &agg, kind, histogram_agg)
+            {
                 Ok(Some(inst)) => inst,
                 other => return other.map(|fs| fs.map(|inst| inst.measure)), // Drop aggregator or error
             };
@@ -497,23 +499,35 @@ where
 /// * Observable UpDownCounter ⇨ Sum
 /// * Gauge ⇨ LastValue
 /// * Observable Gauge ⇨ LastValue
-/// * Histogram ⇨ ExplicitBucketHistogram
+/// * Histogram ⇨ determined by `histogram_agg`
 ///
 /// [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 {
+fn default_aggregation_selector(
+    kind: InstrumentKind,
+    histogram_agg: HistogramAggregation,
+) -> Aggregation {
     match kind {
         InstrumentKind::Counter
         | InstrumentKind::UpDownCounter
         | InstrumentKind::ObservableCounter
         | 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 => match histogram_agg {
+            HistogramAggregation::ExplicitBucketHistogram => 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,
+            },
+            HistogramAggregation::Base2ExponentialBucketHistogram => {
+                Aggregation::Base2ExponentialHistogram {
+                    max_size: 160,
+                    max_scale: 20,
+                    record_min_max: true,
+                }
+            }
         },
     }
 }
@@ -525,9 +539,15 @@ fn aggregate_fn<T: Number>(
     b: AggregateBuilder<T>,
     agg: &aggregation::Aggregation,
     kind: InstrumentKind,
+    histogram_agg: HistogramAggregation,
 ) -> MetricResult<Option<AggregateFns<T>>> {
     match agg {
-        Aggregation::Default => aggregate_fn(b, &default_aggregation_selector(kind), kind),
+        Aggregation::Default => aggregate_fn(
+            b,
+            &default_aggregation_selector(kind, histogram_agg),
+            kind,
+            histogram_agg,
+        ),
         Aggregation::Drop => Ok(None),
         Aggregation::LastValue => {
             match kind {

opentelemetry-sdk/src/metrics/reader.rs

@@ -3,7 +3,10 @@ use crate::error::OTelSdkResult;
 use std::time::Duration;
 use std::{fmt, sync::Weak};
 
-use super::{data::ResourceMetrics, instrument::InstrumentKind, pipeline::Pipeline, Temporality};
+use super::{
+    data::ResourceMetrics, instrument::InstrumentKind, pipeline::Pipeline, HistogramAggregation,
+    Temporality,
+};
 
 /// The interface used between the SDK and an exporter.
 ///
@@ -58,6 +61,14 @@ pub trait MetricReader: fmt::Debug + Send + Sync + 'static {
     ///
     /// If not configured, the Cumulative temporality SHOULD be used.
     fn temporality(&self, kind: InstrumentKind) -> Temporality;
+
+    /// The default histogram aggregation.
+    /// This SHOULD be obtained from the exporter.
+    ///
+    /// If not configured, [`HistogramAggregation::ExplicitBucketHistogram`] is used.
+    fn default_histogram_aggregation(&self) -> HistogramAggregation {
+        HistogramAggregation::default()
+    }
 }
 
 /// Produces metrics for a [MetricReader].