Remove MetricProducer (#2105)

Author: cijothomas

opentelemetry-sdk/CHANGELOG.md

@@ -3,6 +3,9 @@
 ## vNext
 
 - Update `async-std` dependency version to 1.13
+- *Breaking* - Remove support for `MetricProducer` which allowed metrics from
+  external sources to be sent through OpenTelemetry.
+  [#2105](https://github.com/open-telemetry/opentelemetry-rust/pull/2105)  
 
 ## v0.25.0
 

opentelemetry-sdk/src/metrics/manual_reader.rs

@@ -12,9 +12,7 @@ use super::{
     data::{ResourceMetrics, Temporality},
     instrument::InstrumentKind,
     pipeline::Pipeline,
-    reader::{
-        DefaultTemporalitySelector, MetricProducer, MetricReader, SdkProducer, TemporalitySelector,
-    },
+    reader::{DefaultTemporalitySelector, MetricReader, SdkProducer, TemporalitySelector},
 };
 
 /// A simple [MetricReader] that allows an application to read metrics on demand.
@@ -51,7 +49,6 @@ impl fmt::Debug for ManualReader {
 struct ManualReaderInner {
     sdk_producer: Option<Weak<dyn SdkProducer>>,
     is_shutdown: bool,
-    external_producers: Vec<Box<dyn MetricProducer>>,
 }
 
 impl ManualReader {
@@ -61,15 +58,11 @@ impl ManualReader {
     }
 
     /// A [MetricReader] which is directly called to collect metrics.
-    pub(crate) fn new(
-        temporality_selector: Box<dyn TemporalitySelector>,
-        producers: Vec<Box<dyn MetricProducer>>,
-    ) -> Self {
+    pub(crate) fn new(temporality_selector: Box<dyn TemporalitySelector>) -> Self {
         ManualReader {
             inner: Box::new(Mutex::new(ManualReaderInner {
                 sdk_producer: None,
                 is_shutdown: false,
-                external_producers: producers,
             })),
             temporality_selector,
         }
@@ -98,7 +91,7 @@ impl MetricReader for ManualReader {
         });
     }
 
-    /// Gathers all metrics from the SDK and other [MetricProducer]s, calling any
+    /// Gathers all metrics from the SDK, calling any
     /// callbacks necessary and returning the results.
     ///
     /// Returns an error if called after shutdown.
@@ -113,19 +106,7 @@ impl MetricReader for ManualReader {
             }
         };
 
-        let mut errs = vec![];
-        for producer in &inner.external_producers {
-            match producer.produce() {
-                Ok(metrics) => rm.scope_metrics.push(metrics),
-                Err(err) => errs.push(err),
-            }
-        }
-
-        if errs.is_empty() {
-            Ok(())
-        } else {
-            Err(MetricsError::Other(format!("{:?}", errs)))
-        }
+        Ok(())
     }
 
     /// ForceFlush is a no-op, it always returns nil.
@@ -140,7 +121,6 @@ impl MetricReader for ManualReader {
         // Any future call to collect will now return an error.
         inner.sdk_producer = None;
         inner.is_shutdown = true;
-        inner.external_producers = Vec::new();
 
         Ok(())
     }
@@ -149,7 +129,6 @@ impl MetricReader for ManualReader {
 /// Configuration for a [ManualReader]
 pub struct ManualReaderBuilder {
     temporality_selector: Box<dyn TemporalitySelector>,
-    producers: Vec<Box<dyn MetricProducer>>,
 }
 
 impl fmt::Debug for ManualReaderBuilder {
@@ -162,7 +141,6 @@ impl Default for ManualReaderBuilder {
     fn default() -> Self {
         ManualReaderBuilder {
             temporality_selector: Box::new(DefaultTemporalitySelector { _private: () }),
-            producers: vec![],
         }
     }
 }
@@ -184,17 +162,8 @@ impl ManualReaderBuilder {
         self
     }
 
-    /// Registers a 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.producers.push(Box::new(producer));
-        self
-    }
-
     /// Create a new [ManualReader] from this configuration.
     pub fn build(self) -> ManualReader {
-        ManualReader::new(self.temporality_selector, self.producers)
+        ManualReader::new(self.temporality_selector)
     }
 }

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -18,10 +18,7 @@ use opentelemetry::{
 
 use crate::runtime::Runtime;
 use crate::{
-    metrics::{
-        exporter::PushMetricsExporter,
-        reader::{MetricProducer, SdkProducer},
-    },
+    metrics::{exporter::PushMetricsExporter, reader::SdkProducer},
     Resource,
 };
 
@@ -57,7 +54,6 @@ pub struct PeriodicReaderBuilder<E, RT> {
     interval: Duration,
     timeout: Duration,
     exporter: E,
-    producers: Vec<Box<dyn MetricProducer>>,
     runtime: RT,
 }
 
@@ -79,7 +75,6 @@ where
         PeriodicReaderBuilder {
             interval,
             timeout,
-            producers: vec![],
             exporter,
             runtime,
         }
@@ -114,15 +109,6 @@ where
         self
     }
 
-    /// Registers a 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.producers.push(Box::new(producer));
-        self
-    }
-
     /// Create a [PeriodicReader] with the given config.
     pub fn build(self) -> PeriodicReader {
         let (message_sender, message_receiver) = mpsc::channel(256);
@@ -156,7 +142,6 @@ where
             inner: Arc::new(Mutex::new(PeriodicReaderInner {
                 message_sender,
                 is_shutdown: false,
-                external_producers: self.producers,
                 sdk_producer_or_worker: ProducerOrWorker::Worker(Box::new(worker)),
             })),
         }
@@ -226,7 +211,6 @@ impl fmt::Debug for PeriodicReader {
 struct PeriodicReaderInner {
     message_sender: mpsc::Sender<Message>,
     is_shutdown: bool,
-    external_producers: Vec<Box<dyn MetricProducer>>,
     sdk_producer_or_worker: ProducerOrWorker,
 }
 
@@ -342,19 +326,7 @@ impl MetricReader for PeriodicReader {
             return Err(MetricsError::Other("reader is not registered".into()));
         }
 
-        let mut errs = vec![];
-        for producer in &inner.external_producers {
-            match producer.produce() {
-                Ok(metrics) => rm.scope_metrics.push(metrics),
-                Err(err) => errs.push(err),
-            }
-        }
-
-        if errs.is_empty() {
-            Ok(())
-        } else {
-            Err(MetricsError::Other(format!("{:?}", errs)))
-        }
+        Ok(())
     }
 
     fn force_flush(&self) -> Result<()> {

opentelemetry-sdk/src/metrics/reader.rs

@@ -4,7 +4,7 @@ use std::{fmt, sync::Weak};
 use opentelemetry::metrics::Result;
 
 use super::{
-    data::{ResourceMetrics, ScopeMetrics, Temporality},
+    data::{ResourceMetrics, Temporality},
     instrument::InstrumentKind,
     pipeline::Pipeline,
 };
@@ -59,12 +59,6 @@ pub(crate) trait SdkProducer: fmt::Debug + Send + Sync {
     fn produce(&self, rm: &mut ResourceMetrics) -> Result<()>;
 }
 
-/// Produces metrics for a [MetricReader] from an external source.
-pub trait MetricProducer: fmt::Debug + Send + Sync {
-    /// Returns aggregated metrics from an external source.
-    fn produce(&self) -> Result<ScopeMetrics>;
-}
-
 /// An interface for selecting the temporality for an [InstrumentKind].
 pub trait TemporalitySelector: Send + Sync {
     /// Selects the temporality to use based on the [InstrumentKind].