Merge branch 'main' into cijothomas/test-for-observablecounter

Author: cijothomas

examples/metrics-advanced/src/main.rs

@@ -80,13 +80,12 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     // Record measurements using the histogram instrument.
     histogram.record(
         10.5,
-        [
+        &[
             KeyValue::new("mykey1", "myvalue1"),
             KeyValue::new("mykey2", "myvalue2"),
             KeyValue::new("mykey3", "myvalue3"),
             KeyValue::new("mykey4", "myvalue4"),
-        ]
-        .as_ref(),
+        ],
     );
 
     // Example 2 - Drop unwanted attributes using view.
@@ -98,13 +97,12 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     // attribute.
     counter.add(
         10,
-        [
+        &[
             KeyValue::new("mykey1", "myvalue1"),
             KeyValue::new("mykey2", "myvalue2"),
             KeyValue::new("mykey3", "myvalue3"),
             KeyValue::new("mykey4", "myvalue4"),
-        ]
-        .as_ref(),
+        ],
     );
 
     // Example 3 - Change Aggregation configuration using View.
@@ -122,35 +120,32 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     // the change of boundaries.
     histogram2.record(
         1.5,
-        [
+        &[
             KeyValue::new("mykey1", "myvalue1"),
             KeyValue::new("mykey2", "myvalue2"),
             KeyValue::new("mykey3", "myvalue3"),
             KeyValue::new("mykey4", "myvalue4"),
-        ]
-        .as_ref(),
+        ],
     );
 
     histogram2.record(
         1.2,
-        [
+        &[
             KeyValue::new("mykey1", "myvalue1"),
             KeyValue::new("mykey2", "myvalue2"),
             KeyValue::new("mykey3", "myvalue3"),
             KeyValue::new("mykey4", "myvalue4"),
-        ]
-        .as_ref(),
+        ],
     );
 
     histogram2.record(
         1.23,
-        [
+        &[
             KeyValue::new("mykey1", "myvalue1"),
             KeyValue::new("mykey2", "myvalue2"),
             KeyValue::new("mykey3", "myvalue3"),
             KeyValue::new("mykey4", "myvalue4"),
-        ]
-        .as_ref(),
+        ],
     );
 
     // Metrics are exported by default every 30 seconds when using stdout exporter,

examples/metrics-basic/src/main.rs

@@ -34,11 +34,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     // Record measurements using the Counter instrument.
     counter.add(
         10,
-        [
+        &[
             KeyValue::new("mykey1", "myvalue1"),
             KeyValue::new("mykey2", "myvalue2"),
-        ]
-        .as_ref(),
+        ],
     );
 
     // Create a ObservableCounter instrument and register a callback that reports the measurement.
@@ -52,11 +51,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
         observer.observe_u64(
             &observable_counter,
             100,
-            [
+            &[
                 KeyValue::new("mykey1", "myvalue1"),
                 KeyValue::new("mykey2", "myvalue2"),
-            ]
-            .as_ref(),
+            ],
         )
     })?;
 
@@ -66,11 +64,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     // Record measurements using the UpCounter instrument.
     updown_counter.add(
         -10,
-        [
+        &[
             KeyValue::new("mykey1", "myvalue1"),
             KeyValue::new("mykey2", "myvalue2"),
-        ]
-        .as_ref(),
+        ],
     );
 
     // Create a Observable UpDownCounter instrument and register a callback that reports the measurement.
@@ -84,11 +81,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
         observer.observe_i64(
             &observable_up_down_counter,
             100,
-            [
+            &[
                 KeyValue::new("mykey1", "myvalue1"),
                 KeyValue::new("mykey2", "myvalue2"),
-            ]
-            .as_ref(),
+            ],
         )
     })?;
 
@@ -101,11 +97,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     // Record measurements using the histogram instrument.
     histogram.record(
         10.5,
-        [
+        &[
             KeyValue::new("mykey1", "myvalue1"),
             KeyValue::new("mykey2", "myvalue2"),
-        ]
-        .as_ref(),
+        ],
     );
 
     // Note that there is no ObservableHistogram instrument.
@@ -122,11 +117,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
 
         gauge.record(
             1.0,
-            [
+            &[
                 KeyValue::new("mykey1", "myvalue1"),
                 KeyValue::new("mykey2", "myvalue2"),
-            ]
-            .as_ref(),
+            ],
         );
     }
 
@@ -142,11 +136,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
         observer.observe_f64(
             &observable_gauge,
             1.0,
-            [
+            &[
                 KeyValue::new("mykey1", "myvalue1"),
                 KeyValue::new("mykey2", "myvalue2"),
-            ]
-            .as_ref(),
+            ],
         )
     })?;
 

opentelemetry-otlp/tests/integration_test/src/asserter.rs

@@ -19,7 +19,7 @@ impl TraceAsserter {
         self.assert_resource_span_eq(&self.results, &self.expected);
     }
 
-    fn assert_resource_span_eq(&self, results: &Vec<ResourceSpans>, expected: &Vec<ResourceSpans>) {
+    fn assert_resource_span_eq(&self, results: &[ResourceSpans], expected: &[ResourceSpans]) {
         let mut results_spans = Vec::new();
         let mut expected_spans = Vec::new();
 

opentelemetry-sdk/CHANGELOG.md

@@ -46,6 +46,10 @@
   - `LoggerProviderInner` is no longer `pub (crate)`
   - `Logger.provider()` now returns `&LoggerProvider` instead of an `Option<LoggerProvider>`
 
+### Fixed
+
+- [#1481](https://github.com/open-telemetry/opentelemetry-rust/pull/1481) Fix error message caused by race condition when using PeriodicReader
+
 ## v0.21.2
 
 ### Fixed

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -1,5 +1,5 @@
 use std::{
-    env, fmt,
+    env, fmt, mem,
     sync::{Arc, Mutex, Weak},
     time::Duration,
 };
@@ -127,37 +127,39 @@ where
     /// Create a [PeriodicReader] with the given config.
     pub fn build(self) -> PeriodicReader {
         let (message_sender, message_receiver) = mpsc::channel(256);
-        let ticker = self
-            .runtime
-            .interval(self.interval)
-            .map(|_| Message::Export);
 
-        let messages = Box::pin(stream::select(message_receiver, ticker));
-        let reader = PeriodicReader {
+        let worker = move |reader: &PeriodicReader| {
+            let ticker = self
+                .runtime
+                .interval(self.interval)
+                .map(|_| Message::Export);
+
+            let messages = Box::pin(stream::select(message_receiver, ticker));
+
+            let runtime = self.runtime.clone();
+            self.runtime.spawn(Box::pin(
+                PeriodicReaderWorker {
+                    reader: reader.clone(),
+                    timeout: self.timeout,
+                    runtime,
+                    rm: ResourceMetrics {
+                        resource: Resource::empty(),
+                        scope_metrics: Vec::new(),
+                    },
+                }
+                .run(messages),
+            ));
+        };
+
+        PeriodicReader {
             exporter: Arc::new(self.exporter),
             inner: Arc::new(Mutex::new(PeriodicReaderInner {
                 message_sender,
-                sdk_producer: None,
                 is_shutdown: false,
                 external_producers: self.producers,
+                sdk_producer_or_worker: ProducerOrWorker::Worker(Box::new(worker)),
             })),
-        };
-
-        let runtime = self.runtime.clone();
-        self.runtime.spawn(Box::pin(
-            PeriodicReaderWorker {
-                reader: reader.clone(),
-                timeout: self.timeout,
-                runtime,
-                rm: ResourceMetrics {
-                    resource: Resource::empty(),
-                    scope_metrics: Vec::new(),
-                },
-            }
-            .run(messages),
-        ));
-
-        reader
+        }
     }
 }
 
@@ -223,9 +225,9 @@ impl fmt::Debug for PeriodicReader {
 
 struct PeriodicReaderInner {
     message_sender: mpsc::Sender<Message>,
-    sdk_producer: Option<Weak<dyn SdkProducer>>,
     is_shutdown: bool,
     external_producers: Vec<Box<dyn MetricProducer>>,
+    sdk_producer_or_worker: ProducerOrWorker,
 }
 
 #[derive(Debug)]
@@ -235,6 +237,11 @@ enum Message {
     Shutdown(oneshot::Sender<Result<()>>),
 }
 
+enum ProducerOrWorker {
+    Producer(Weak<dyn SdkProducer>),
+    Worker(Box<dyn FnOnce(&PeriodicReader) + Send + Sync>),
+}
+
 struct PeriodicReaderWorker<RT: Runtime> {
     reader: PeriodicReader,
     timeout: Duration,
@@ -311,14 +318,19 @@ impl MetricReader for PeriodicReader {
             Err(_) => return,
         };
 
-        // Only register once. If producer is already set, do nothing.
-        if inner.sdk_producer.is_none() {
-            inner.sdk_producer = Some(pipeline);
-        } else {
-            global::handle_error(MetricsError::Other(
-                "duplicate meter registration, did not register manual reader".into(),
-            ))
-        }
+        let worker = match &mut inner.sdk_producer_or_worker {
+            ProducerOrWorker::Producer(_) => {
+                // Only register once. If producer is already set, do nothing.
+                global::handle_error(MetricsError::Other(
+                    "duplicate meter registration, did not register manual reader".into(),
+                ));
+                return;
+            }
+            ProducerOrWorker::Worker(w) => mem::replace(w, Box::new(|_| {})),
+        };
+
+        inner.sdk_producer_or_worker = ProducerOrWorker::Producer(pipeline);
+        worker(self);
     }
 
     fn collect(&self, rm: &mut ResourceMetrics) -> Result<()> {
@@ -327,14 +339,14 @@ impl MetricReader for PeriodicReader {
             return Err(MetricsError::Other("reader is shut down".into()));
         }
 
-        match &inner.sdk_producer.as_ref().and_then(|w| w.upgrade()) {
-            Some(producer) => producer.produce(rm)?,
-            None => {
-                return Err(MetricsError::Other(
-                    "reader is shut down or not registered".into(),
-                ))
-            }
-        };
+        if let Some(producer) = match &inner.sdk_producer_or_worker {
+            ProducerOrWorker::Producer(sdk_producer) => sdk_producer.upgrade(),
+            ProducerOrWorker::Worker(_) => None,
+        } {
+            producer.produce(rm)?;
+        } else {
+            return Err(MetricsError::Other("reader is not registered".into()));
+        }
 
         let mut errs = vec![];
         for producer in &inner.external_producers {
@@ -392,3 +404,25 @@ impl MetricReader for PeriodicReader {
         shutdown_result
     }
 }
+
+#[cfg(all(test, feature = "testing"))]
+mod tests {
+    use super::PeriodicReader;
+    use crate::{
+        metrics::data::ResourceMetrics, metrics::reader::MetricReader, runtime,
+        testing::metrics::InMemoryMetricsExporter, Resource,
+    };
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn unregistered_collect() {
+        let exporter = InMemoryMetricsExporter::default();
+        let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
+
+        let mut rm = ResourceMetrics {
+            resource: Resource::empty(),
+            scope_metrics: Vec::new(),
+        };
+        let result = reader.collect(&mut rm);
+        result.expect_err("error expected when reader is not registered");
+    }
+}