Add Exponential histogram (#1267)

Author: jtescher

opentelemetry-proto/CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - Implement tonic metrics proto transformations (#1184)
 - Move proto for zPage to tonic [#1214](https://github.com/open-telemetry/opentelemetry-rust/pull/1214)
+- Support exponential histograms (#1267)
 
 ### Changed
 

opentelemetry-proto/src/transform/metrics.rs

@@ -10,23 +10,28 @@ pub mod tonic {
 
     use opentelemetry::{global, metrics::MetricsError, Key, Value};
     use opentelemetry_sdk::metrics::data::{
-        self, Exemplar as SdkExemplar, Gauge as SdkGauge, Histogram as SdkHistogram,
-        Metric as SdkMetric, ScopeMetrics as SdkScopeMetrics, Sum as SdkSum, Temporality,
+        self, Exemplar as SdkExemplar, ExponentialHistogram as SdkExponentialHistogram,
+        Gauge as SdkGauge, Histogram as SdkHistogram, Metric as SdkMetric,
+        ScopeMetrics as SdkScopeMetrics, Sum as SdkSum, Temporality,
     };
     use opentelemetry_sdk::Resource as SdkResource;
 
     use crate::proto::tonic::{
         collector::metrics::v1::ExportMetricsServiceRequest,
         common::v1::KeyValue,
         metrics::v1::{
-            exemplar, exemplar::Value as TonicExemplarValue, metric::Data as TonicMetricData,
-            number_data_point, number_data_point::Value as TonicDataPointValue,
+            exemplar, exemplar::Value as TonicExemplarValue,
+            exponential_histogram_data_point::Buckets as TonicBuckets,
+            metric::Data as TonicMetricData, number_data_point,
+            number_data_point::Value as TonicDataPointValue,
             AggregationTemporality as TonicTemporality, AggregationTemporality,
-            DataPointFlags as TonicDataPointFlags, Exemplar as TonicExemplar, Gauge as TonicGauge,
-            Histogram as TonicHistogram, HistogramDataPoint as TonicHistogramDataPoint,
-            Metric as TonicMetric, NumberDataPoint as TonicNumberDataPoint,
-            ResourceMetrics as TonicResourceMetrics, ScopeMetrics as TonicScopeMetrics,
-            Sum as TonicSum,
+            DataPointFlags as TonicDataPointFlags, Exemplar as TonicExemplar,
+            ExponentialHistogram as TonicExponentialHistogram,
+            ExponentialHistogramDataPoint as TonicExponentialHistogramDataPoint,
+            Gauge as TonicGauge, Histogram as TonicHistogram,
+            HistogramDataPoint as TonicHistogramDataPoint, Metric as TonicMetric,
+            NumberDataPoint as TonicNumberDataPoint, ResourceMetrics as TonicResourceMetrics,
+            ScopeMetrics as TonicScopeMetrics, Sum as TonicSum,
         },
         resource::v1::Resource as TonicResource,
     };
@@ -159,6 +164,12 @@ pub mod tonic {
                 Ok(TonicMetricData::Histogram(hist.into()))
             } else if let Some(hist) = data.downcast_ref::<SdkHistogram<f64>>() {
                 Ok(TonicMetricData::Histogram(hist.into()))
+            } else if let Some(hist) = data.downcast_ref::<SdkExponentialHistogram<i64>>() {
+                Ok(TonicMetricData::ExponentialHistogram(hist.into()))
+            } else if let Some(hist) = data.downcast_ref::<SdkExponentialHistogram<u64>>() {
+                Ok(TonicMetricData::ExponentialHistogram(hist.into()))
+            } else if let Some(hist) = data.downcast_ref::<SdkExponentialHistogram<f64>>() {
+                Ok(TonicMetricData::ExponentialHistogram(hist.into()))
             } else if let Some(sum) = data.downcast_ref::<SdkSum<u64>>() {
                 Ok(TonicMetricData::Sum(sum.into()))
             } else if let Some(sum) = data.downcast_ref::<SdkSum<i64>>() {
@@ -229,6 +240,43 @@ pub mod tonic {
         }
     }
 
+    impl<T> From<&SdkExponentialHistogram<T>> for TonicExponentialHistogram
+    where
+        T: Numeric,
+    {
+        fn from(hist: &SdkExponentialHistogram<T>) -> Self {
+            TonicExponentialHistogram {
+                data_points: hist
+                    .data_points
+                    .iter()
+                    .map(|dp| TonicExponentialHistogramDataPoint {
+                        attributes: dp.attributes.iter().map(Into::into).collect(),
+                        start_time_unix_nano: to_nanos(dp.start_time),
+                        time_unix_nano: to_nanos(dp.time),
+                        count: dp.count as u64,
+                        sum: Some(dp.sum.into_f64()),
+                        scale: dp.scale.into(),
+                        zero_count: dp.zero_count,
+                        positive: Some(TonicBuckets {
+                            offset: dp.positive_bucket.offset,
+                            bucket_counts: dp.positive_bucket.counts.clone(),
+                        }),
+                        negative: Some(TonicBuckets {
+                            offset: dp.negative_bucket.offset,
+                            bucket_counts: dp.negative_bucket.counts.clone(),
+                        }),
+                        flags: TonicDataPointFlags::default() as u32,
+                        exemplars: dp.exemplars.iter().map(Into::into).collect(),
+                        min: dp.min.map(Numeric::into_f64),
+                        max: dp.max.map(Numeric::into_f64),
+                        zero_threshold: dp.zero_threshold,
+                    })
+                    .collect(),
+                aggregation_temporality: TonicTemporality::from(hist.temporality).into(),
+            }
+        }
+    }
+
     impl<T> From<&SdkSum<T>> for TonicSum
     where
         T: fmt::Debug + Into<TonicExemplarValue> + Into<TonicDataPointValue> + Copy,
@@ -302,19 +350,24 @@ pub mod grpcio {
 
     use opentelemetry::{global, metrics::MetricsError, Key, Value};
     use opentelemetry_sdk::metrics::data::{
-        self, Exemplar as SdkExemplar, Gauge as SdkGauge, Histogram as SdkHistogram,
-        Metric as SdkMetric, ScopeMetrics as SdkScopeMetrics, Sum as SdkSum, Temporality,
+        self, Exemplar as SdkExemplar, ExponentialHistogram as SdkExponentialHistogram,
+        Gauge as SdkGauge, Histogram as SdkHistogram, Metric as SdkMetric,
+        ScopeMetrics as SdkScopeMetrics, Sum as SdkSum, Temporality,
     };
     use opentelemetry_sdk::Resource as SdkResource;
 
     use crate::proto::grpcio::{
         collector::metrics::v1::ExportMetricsServiceRequest,
         common::v1::KeyValue,
         metrics::v1::{
-            exemplar, exemplar::Value as GrpcioExemplarValue, metric::Data as GrpcioMetricData,
-            number_data_point, number_data_point::Value as GrpcioDataPointValue,
+            exemplar, exemplar::Value as GrpcioExemplarValue,
+            exponential_histogram_data_point::Buckets as GrpcioBuckets,
+            metric::Data as GrpcioMetricData, number_data_point,
+            number_data_point::Value as GrpcioDataPointValue,
             AggregationTemporality as GrpcioTemporality, AggregationTemporality,
             DataPointFlags as GrpcioDataPointFlags, Exemplar as GrpcioExemplar,
+            ExponentialHistogram as GrpcioExponentialHistogram,
+            ExponentialHistogramDataPoint as GrpcioExponentialHistogramDataPoint,
             Gauge as GrpcioGauge, Histogram as GrpcioHistogram,
             HistogramDataPoint as GrpcioHistogramDataPoint, Metric as GrpcioMetric,
             NumberDataPoint as GrpcioNumberDataPoint, ResourceMetrics as GrpcioResourceMetrics,
@@ -451,6 +504,12 @@ pub mod grpcio {
                 Ok(GrpcioMetricData::Histogram(hist.into()))
             } else if let Some(hist) = data.downcast_ref::<SdkHistogram<f64>>() {
                 Ok(GrpcioMetricData::Histogram(hist.into()))
+            } else if let Some(hist) = data.downcast_ref::<SdkExponentialHistogram<i64>>() {
+                Ok(GrpcioMetricData::ExponentialHistogram(hist.into()))
+            } else if let Some(hist) = data.downcast_ref::<SdkExponentialHistogram<u64>>() {
+                Ok(GrpcioMetricData::ExponentialHistogram(hist.into()))
+            } else if let Some(hist) = data.downcast_ref::<SdkExponentialHistogram<f64>>() {
+                Ok(GrpcioMetricData::ExponentialHistogram(hist.into()))
             } else if let Some(sum) = data.downcast_ref::<SdkSum<u64>>() {
                 Ok(GrpcioMetricData::Sum(sum.into()))
             } else if let Some(sum) = data.downcast_ref::<SdkSum<i64>>() {
@@ -521,6 +580,43 @@ pub mod grpcio {
         }
     }
 
+    impl<T> From<&SdkExponentialHistogram<T>> for GrpcioExponentialHistogram
+    where
+        T: Numeric,
+    {
+        fn from(hist: &SdkExponentialHistogram<T>) -> Self {
+            GrpcioExponentialHistogram {
+                data_points: hist
+                    .data_points
+                    .iter()
+                    .map(|dp| GrpcioExponentialHistogramDataPoint {
+                        attributes: dp.attributes.iter().map(Into::into).collect(),
+                        start_time_unix_nano: to_nanos(dp.start_time),
+                        time_unix_nano: to_nanos(dp.time),
+                        count: dp.count as u64,
+                        sum: Some(dp.sum.into_f64()),
+                        scale: dp.scale.into(),
+                        zero_count: dp.zero_count,
+                        positive: Some(GrpcioBuckets {
+                            offset: dp.positive_bucket.offset,
+                            bucket_counts: dp.positive_bucket.counts.clone(),
+                        }),
+                        negative: Some(GrpcioBuckets {
+                            offset: dp.negative_bucket.offset,
+                            bucket_counts: dp.negative_bucket.counts.clone(),
+                        }),
+                        flags: GrpcioDataPointFlags::default() as u32,
+                        exemplars: dp.exemplars.iter().map(Into::into).collect(),
+                        min: dp.min.map(Numeric::into_f64),
+                        max: dp.max.map(Numeric::into_f64),
+                        zero_threshold: dp.zero_threshold,
+                    })
+                    .collect(),
+                aggregation_temporality: GrpcioTemporality::from(hist.temporality).into(),
+            }
+        }
+    }
+
     impl<T> From<&SdkSum<T>> for GrpcioSum
     where
         T: fmt::Debug + Into<GrpcioExemplarValue> + Into<GrpcioDataPointValue> + Copy,

opentelemetry-sdk/CHANGELOG.md

@@ -7,6 +7,7 @@
 - Log warning if two instruments have the same name with different (#1266)
   casing
 - Log warning if view is created with empty criteria (#1266)
+- Add exponential histogram support (#1267)
 
 ### Changed
 

opentelemetry-sdk/src/attributes/set.rs

@@ -104,7 +104,7 @@ impl Eq for HashKeyValue {}
 ///
 /// This must implement [Hash], [PartialEq], and [Eq] so it may be used as
 /// HashMap keys and other de-duplication methods.
-#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+#[derive(Clone, Default, Debug, Hash, PartialEq, Eq)]
 pub struct AttributeSet(BTreeSet<HashKeyValue>);
 
 impl From<&[KeyValue]> for AttributeSet {

opentelemetry-sdk/src/metrics/aggregation.rs

@@ -1,5 +1,6 @@
 use std::fmt;
 
+use crate::metrics::internal::{EXPO_MAX_SCALE, EXPO_MIN_SCALE};
 use opentelemetry::metrics::{MetricsError, Result};
 
 /// The way recorded measurements are summarized.
@@ -62,6 +63,32 @@ pub enum Aggregation {
         /// instances.
         record_min_max: bool,
     },
+
+    /// An aggregation that summarizes a set of measurements as a histogram with
+    /// bucket widths that grow exponentially.
+    Base2ExponentialHistogram {
+        /// The maximum number of buckets to use for the histogram.
+        max_size: u32,
+
+        /// The maximum resolution scale to use for the histogram.
+        ///
+        /// The maximum value is `20`, in which case the maximum number of buckets
+        /// that can fit within the range of a signed 32-bit integer index could be
+        /// used.
+        ///
+        /// The minimum value is `-10` in which case only two buckets will be used.
+        max_scale: i8,
+
+        /// Indicates whether to not record the min and max of the distribution.
+        ///
+        /// By default, these values are recorded.
+        ///
+        /// It is generally not valuable to record min and max for cumulative data
+        /// as they will represent the entire life of the instrument instead of just
+        /// the current collection cycle, you can opt out by setting this value to
+        /// `false`
+        record_min_max: bool,
+    },
 }
 
 impl fmt::Display for Aggregation {
@@ -73,6 +100,7 @@ impl fmt::Display for Aggregation {
             Aggregation::Sum => "Sum",
             Aggregation::LastValue => "LastValue",
             Aggregation::ExplicitBucketHistogram { .. } => "ExplicitBucketHistogram",
+            Aggregation::Base2ExponentialHistogram { .. } => "Base2ExponentialHistogram",
         };
 
         f.write_str(name)
@@ -97,6 +125,22 @@ impl Aggregation {
                     }
                 }
 
+                Ok(())
+            }
+            Aggregation::Base2ExponentialHistogram { max_scale, .. } => {
+                if *max_scale > EXPO_MAX_SCALE {
+                    return Err(MetricsError::Config(format!(
+                        "aggregation: exponential histogram: max scale ({}) is greater than 20",
+                        max_scale,
+                    )));
+                }
+                if *max_scale < -EXPO_MIN_SCALE {
+                    return Err(MetricsError::Config(format!(
+                        "aggregation: exponential histogram: max scale ({}) is less than -10",
+                        max_scale,
+                    )));
+                }
+
                 Ok(())
             }
         }

opentelemetry-sdk/src/metrics/data/mod.rs

@@ -186,6 +186,88 @@ impl<T: Copy> Clone for HistogramDataPoint<T> {
     }
 }
 
+/// The histogram of all measurements of values from an instrument.
+#[derive(Debug)]
+pub struct ExponentialHistogram<T> {
+    /// The individual aggregated measurements with unique attributes.
+    pub data_points: Vec<ExponentialHistogramDataPoint<T>>,
+
+    /// Describes if the aggregation is reported as the change from the last report
+    /// time, or the cumulative changes since a fixed start time.
+    pub temporality: Temporality,
+}
+
+impl<T: fmt::Debug + Send + Sync + 'static> Aggregation for ExponentialHistogram<T> {
+    fn as_any(&self) -> &dyn any::Any {
+        self
+    }
+    fn as_mut(&mut self) -> &mut dyn any::Any {
+        self
+    }
+}
+
+/// A single exponential histogram data point in a time series.
+#[derive(Debug)]
+pub struct ExponentialHistogramDataPoint<T> {
+    /// The set of key value pairs that uniquely identify the time series.
+    pub attributes: AttributeSet,
+    /// When the time series was started.
+    pub start_time: SystemTime,
+    /// The time when the time series was recorded.
+    pub time: SystemTime,
+
+    /// The number of updates this histogram has been calculated with.
+    pub count: usize,
+    /// The minimum value recorded.
+    pub min: Option<T>,
+    /// The maximum value recorded.
+    pub max: Option<T>,
+    /// The sum of the values recorded.
+    pub sum: T,
+
+    /// Describes the resolution of the histogram.
+    ///
+    /// Boundaries are located at powers of the base, where:
+    ///
+    ///   base = 2 ^ (2 ^ -scale)
+    pub scale: i8,
+
+    /// The number of values whose absolute value is less than or equal to
+    /// `zero_threshold`.
+    ///
+    /// When `zero_threshold` is `0`, this is the number of values that cannot be
+    /// expressed using the standard exponential formula as well as values that have
+    /// been rounded to zero.
+    pub zero_count: u64,
+
+    /// The range of positive value bucket counts.
+    pub positive_bucket: ExponentialBucket,
+    /// The range of negative value bucket counts.
+    pub negative_bucket: ExponentialBucket,
+
+    /// The width of the zero region.
+    ///
+    /// Where the zero region is defined as the closed interval
+    /// [-zero_threshold, zero_threshold].
+    pub zero_threshold: f64,
+
+    /// The sampled exemplars collected during the time series.
+    pub exemplars: Vec<Exemplar<T>>,
+}
+
+/// A set of bucket counts, encoded in a contiguous array of counts.
+#[derive(Debug, PartialEq)]
+pub struct ExponentialBucket {
+    /// The bucket index of the first entry in the `counts` vec.
+    pub offset: i32,
+
+    /// A vec where `counts[i]` carries the count of the bucket at index `offset + i`.
+    ///
+    /// `counts[i]` is the count of values greater than base^(offset+i) and less than
+    /// or equal to base^(offset+i+1).
+    pub counts: Vec<u64>,
+}
+
 /// A measurement sampled from a time series providing a typical example.
 #[derive(Debug)]
 pub struct Exemplar<T> {

opentelemetry-sdk/src/metrics/instrument.rs

@@ -226,7 +226,7 @@ pub(crate) struct InstrumentId {
     pub(crate) description: Cow<'static, str>,
     /// Defines the functional group of the instrument.
     pub(crate) kind: InstrumentKind,
-    /// the unit of measurement recorded.
+    /// The unit of measurement recorded.
     pub(crate) unit: Unit,
     /// Number is the underlying data type of the instrument.
     pub(crate) number: Cow<'static, str>,