Re-use memory via aggregate functions (#1266)

This simplifies memory reuse on subsequent collection cycles.

Also small ergonomics changes:

* Move `MetricsProducer` config to builders to match other config
* Remove the need for separate `Delta*` and `Cumulative*` aggregate
  types
* Return error earlier if readers are shut down
* Log warning if two instruments have the same name with different
  casing
* Log warning if view is created with empty criteria

Author: jtescher

opentelemetry-prometheus/src/config.rs

@@ -1,7 +1,10 @@
 use core::fmt;
 use once_cell::sync::OnceCell;
 use opentelemetry::metrics::{MetricsError, Result};
-use opentelemetry_sdk::metrics::{reader::AggregationSelector, ManualReaderBuilder};
+use opentelemetry_sdk::metrics::{
+    reader::{AggregationSelector, MetricProducer},
+    ManualReaderBuilder,
+};
 use std::sync::{Arc, Mutex};
 
 use crate::{Collector, PrometheusExporter};
@@ -14,8 +17,8 @@ pub struct ExporterBuilder {
     without_units: bool,
     without_counter_suffixes: bool,
     namespace: Option<String>,
-    aggregation: Option<Box<dyn AggregationSelector>>,
     disable_scope_info: bool,
+    reader: ManualReaderBuilder,
 }
 
 impl fmt::Debug for ExporterBuilder {
@@ -26,7 +29,6 @@ impl fmt::Debug for ExporterBuilder {
             .field("without_units", &self.without_units)
             .field("without_counter_suffixes", &self.without_counter_suffixes)
             .field("namespace", &self.namespace)
-            .field("aggregation", &self.aggregation.is_some())
             .field("disable_scope_info", &self.disable_scope_info)
             .finish()
     }
@@ -108,17 +110,22 @@ impl ExporterBuilder {
     ///
     /// [DefaultAggregationSelector]: opentelemetry_sdk::metrics::reader::DefaultAggregationSelector
     pub fn with_aggregation_selector(mut self, agg: impl AggregationSelector + 'static) -> Self {
-        self.aggregation = Some(Box::new(agg));
+        self.reader = self.reader.with_aggregation_selector(agg);
+        self
+    }
+
+    /// Registers an external [MetricProducer] with this reader.
+    ///
+    /// The producer is used as a source of aggregated metric data which is
+    /// incorporated into metrics collected from the SDK.
+    pub fn with_producer(mut self, producer: impl MetricProducer + 'static) -> Self {
+        self.reader = self.reader.with_producer(producer);
         self
     }
 
     /// Creates a new [PrometheusExporter] from this configuration.
     pub fn build(self) -> Result<PrometheusExporter> {
-        let mut reader = ManualReaderBuilder::new();
-        if let Some(selector) = self.aggregation {
-            reader = reader.with_aggregation_selector(selector)
-        }
-        let reader = Arc::new(reader.build());
+        let reader = Arc::new(self.reader.build());
 
         let collector = Collector {
             reader: Arc::clone(&reader),

opentelemetry-prometheus/src/lib.rs

@@ -102,7 +102,7 @@ use opentelemetry::{
 use opentelemetry_sdk::{
     metrics::{
         data::{self, ResourceMetrics, Temporality},
-        reader::{AggregationSelector, MetricProducer, MetricReader, TemporalitySelector},
+        reader::{AggregationSelector, MetricReader, TemporalitySelector},
         Aggregation, InstrumentKind, ManualReader, Pipeline,
     },
     Resource, Scope,
@@ -166,10 +166,6 @@ impl MetricReader for PrometheusExporter {
         self.reader.register_pipeline(pipeline)
     }
 
-    fn register_producer(&self, producer: Box<dyn MetricProducer>) {
-        self.reader.register_producer(producer)
-    }
-
     fn collect(&self, rm: &mut ResourceMetrics) -> Result<()> {
         self.reader.collect(rm)
     }

opentelemetry-sdk/CHANGELOG.md

@@ -2,6 +2,12 @@
 
 ## vNext
 
+### Added
+
+- Log warning if two instruments have the same name with different (#1266)
+  casing
+- Log warning if view is created with empty criteria (#1266)
+
 ### Changed
 
 - Default Resource (the one used when no other Resource is explicitly provided) now includes `TelemetryResourceDetector`,
@@ -14,6 +20,8 @@
 - Add in memory span exporter [#1216](https://github.com/open-telemetry/opentelemetry-rust/pull/1216)
 - Add in memory log exporter [#1231](https://github.com/open-telemetry/opentelemetry-rust/pull/1231)
 - Add `Sync` bound to the `SpanExporter` and `LogExporter` traits [#1240](https://github.com/open-telemetry/opentelemetry-rust/pull/1240)
+- Move `MetricsProducer` config to builders to match other config (#1266)
+- Return error earlier if readers are shut down (#1266)
 
 ### Removed
 

opentelemetry-sdk/benches/metric.rs

@@ -10,7 +10,7 @@ use opentelemetry_sdk::{
     metrics::{
         data::{ResourceMetrics, Temporality},
         new_view,
-        reader::{AggregationSelector, MetricProducer, MetricReader, TemporalitySelector},
+        reader::{AggregationSelector, MetricReader, TemporalitySelector},
         Aggregation, Instrument, InstrumentKind, ManualReader, MeterProvider, Pipeline, Stream,
         View,
     },
@@ -37,10 +37,6 @@ impl MetricReader for SharedReader {
         self.0.register_pipeline(pipeline)
     }
 
-    fn register_producer(&self, producer: Box<dyn MetricProducer>) {
-        self.0.register_producer(producer)
-    }
-
     fn collect(&self, rm: &mut ResourceMetrics) -> Result<()> {
         self.0.collect(rm)
     }

opentelemetry-sdk/src/metrics/aggregation.rs

@@ -29,7 +29,7 @@ pub enum Aggregation {
     /// An aggregation that summarizes a set of measurements as the last one made.
     LastValue,
 
-    /// An aggregation that summarizes a set of measurements as an histogram with
+    /// An aggregation that summarizes a set of measurements as a histogram with
     /// explicitly defined buckets.
     ExplicitBucketHistogram {
         /// The increasing bucket boundary values.

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

@@ -51,6 +51,8 @@ pub struct Metric {
 pub trait Aggregation: fmt::Debug + any::Any + Send + Sync {
     /// Support downcasting
     fn as_any(&self) -> &dyn any::Any;
+    /// Support downcasting during aggregation
+    fn as_mut(&mut self) -> &mut dyn any::Any;
 }
 
 /// A measurement of the current value of an instrument.
@@ -64,6 +66,9 @@ impl<T: fmt::Debug + Send + Sync + 'static> Aggregation for Gauge<T> {
     fn as_any(&self) -> &dyn any::Any {
         self
     }
+    fn as_mut(&mut self) -> &mut dyn any::Any {
+        self
+    }
 }
 
 /// Represents the sum of all measurements of values from an instrument.
@@ -82,6 +87,9 @@ impl<T: fmt::Debug + Send + Sync + 'static> Aggregation for Sum<T> {
     fn as_any(&self) -> &dyn any::Any {
         self
     }
+    fn as_mut(&mut self) -> &mut dyn any::Any {
+        self
+    }
 }
 
 /// DataPoint is a single data point in a time series.
@@ -126,6 +134,9 @@ impl<T: fmt::Debug + Send + Sync + 'static> Aggregation for Histogram<T> {
     fn as_any(&self) -> &dyn any::Any {
         self
     }
+    fn as_mut(&mut self) -> &mut dyn any::Any {
+        self
+    }
 }
 
 /// A single histogram data point in a time series.

opentelemetry-sdk/src/metrics/instrument.rs

@@ -10,11 +10,10 @@ use opentelemetry::{
 use crate::{
     attributes::AttributeSet,
     instrumentation::Scope,
-    metrics::data::Temporality,
-    metrics::{aggregation::Aggregation, internal::Aggregator},
+    metrics::{aggregation::Aggregation, internal::Measure},
 };
 
-pub(crate) const EMPTY_AGG_MSG: &str = "no aggregators for observable instrument";
+pub(crate) const EMPTY_MEASURE_MSG: &str = "no aggregators for observable instrument";
 
 /// The identifier of a group of instruments that all perform the same function.
 #[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)]
@@ -218,52 +217,61 @@ impl Stream {
     }
 }
 
-/// the identifying properties of a stream.
+/// The identifying properties of an instrument.
 #[derive(Debug, PartialEq, Eq, Hash)]
-pub(crate) struct StreamId {
-    /// The human-readable identifier of the stream.
+pub(crate) struct InstrumentId {
+    /// The human-readable identifier of the instrument.
     pub(crate) name: Cow<'static, str>,
     /// Describes the purpose of the data.
     pub(crate) description: Cow<'static, str>,
+    /// Defines the functional group of the instrument.
+    pub(crate) kind: InstrumentKind,
     /// the unit of measurement recorded.
     pub(crate) unit: Unit,
-    /// The stream uses for an instrument.
-    pub(crate) aggregation: String,
-    /// Monotonic is the monotonicity of an instruments data type. This field is
-    /// not used for all data types, so a zero value needs to be understood in the
-    /// context of Aggregation.
-    pub(crate) monotonic: bool,
-    /// Temporality is the temporality of a stream's data type. This field is
-    /// not used by some data types.
-    pub(crate) temporality: Option<Temporality>,
-    /// Number is the number type of the stream.
+    /// Number is the underlying data type of the instrument.
     pub(crate) number: Cow<'static, str>,
 }
 
+impl InstrumentId {
+    /// Instrument names are considered case-insensitive ASCII.
+    ///
+    /// Standardize the instrument name to always be lowercase so it can be compared
+    /// via hash.
+    ///
+    /// See [naming syntax] for full requirements.
+    ///
+    /// [naming syntax]: https://github.com/open-telemetry/opentelemetry-specification/blob/v1.21.0/specification/metrics/api.md#instrument-name-syntax
+    pub(crate) fn normalize(&mut self) {
+        if self.name.chars().any(|c| c.is_ascii_uppercase()) {
+            self.name = self.name.to_ascii_lowercase().into();
+        }
+    }
+}
+
 pub(crate) struct InstrumentImpl<T> {
-    pub(crate) aggregators: Vec<Arc<dyn Aggregator<T>>>,
+    pub(crate) measures: Vec<Arc<dyn Measure<T>>>,
 }
 
-impl<T: Copy> SyncCounter<T> for InstrumentImpl<T> {
+impl<T: Copy + 'static> SyncCounter<T> for InstrumentImpl<T> {
     fn add(&self, val: T, attrs: &[KeyValue]) {
-        for agg in &self.aggregators {
-            agg.aggregate(val, AttributeSet::from(attrs))
+        for measure in &self.measures {
+            measure.call(val, AttributeSet::from(attrs))
         }
     }
 }
 
-impl<T: Copy> SyncUpDownCounter<T> for InstrumentImpl<T> {
+impl<T: Copy + 'static> SyncUpDownCounter<T> for InstrumentImpl<T> {
     fn add(&self, val: T, attrs: &[KeyValue]) {
-        for agg in &self.aggregators {
-            agg.aggregate(val, AttributeSet::from(attrs))
+        for measure in &self.measures {
+            measure.call(val, AttributeSet::from(attrs))
         }
     }
 }
 
-impl<T: Copy> SyncHistogram<T> for InstrumentImpl<T> {
+impl<T: Copy + 'static> SyncHistogram<T> for InstrumentImpl<T> {
     fn record(&self, val: T, attrs: &[KeyValue]) {
-        for agg in &self.aggregators {
-            agg.aggregate(val, AttributeSet::from(attrs))
+        for measure in &self.measures {
+            measure.call(val, AttributeSet::from(attrs))
         }
     }
 }
@@ -306,7 +314,7 @@ impl<T> Eq for ObservableId<T> {}
 #[derive(Clone)]
 pub(crate) struct Observable<T> {
     pub(crate) id: ObservableId<T>,
-    aggregators: Vec<Arc<dyn Aggregator<T>>>,
+    measures: Vec<Arc<dyn Measure<T>>>,
 }
 
 impl<T> Observable<T> {
@@ -316,7 +324,7 @@ impl<T> Observable<T> {
         name: Cow<'static, str>,
         description: Cow<'static, str>,
         unit: Unit,
-        aggregators: Vec<Arc<dyn Aggregator<T>>>,
+        measures: Vec<Arc<dyn Measure<T>>>,
     ) -> Self {
         Self {
             id: ObservableId {
@@ -329,7 +337,7 @@ impl<T> Observable<T> {
                 },
                 _marker: marker::PhantomData,
             },
-            aggregators,
+            measures,
         }
     }
 
@@ -340,8 +348,8 @@ impl<T> Observable<T> {
     /// any aggregators. Also, an error is returned if scope defines a Meter other
     /// than the observable it was created by.
     pub(crate) fn registerable(&self, scope: &Scope) -> Result<()> {
-        if self.aggregators.is_empty() {
-            return Err(MetricsError::Other(EMPTY_AGG_MSG.into()));
+        if self.measures.is_empty() {
+            return Err(MetricsError::Other(EMPTY_MEASURE_MSG.into()));
         }
         if &self.id.inner.scope != scope {
             return Err(MetricsError::Other(format!(
@@ -356,8 +364,8 @@ impl<T> Observable<T> {
 
 impl<T: Copy + Send + Sync + 'static> AsyncInstrument<T> for Observable<T> {
     fn observe(&self, measurement: T, attrs: &[KeyValue]) {
-        for agg in &self.aggregators {
-            agg.aggregate(measurement, AttributeSet::from(attrs))
+        for measure in &self.measures {
+            measure.call(measurement, AttributeSet::from(attrs))
         }
     }