Merge branch 'main' into addlink

Author: lalitb

opentelemetry-sdk/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## vNext
 
+- Fix SimpleSpanProcessor to be consistent with log counterpart. Also removed
+  dependency on crossbeam-channel.
+  [1612](https://github.com/open-telemetry/opentelemetry-rust/pull/1612/files)
+
 ## v0.22.1
 
 ### Fixed

opentelemetry-sdk/Cargo.toml

@@ -14,7 +14,6 @@ opentelemetry = { version = "0.22", path = "../opentelemetry/" }
 opentelemetry-http = { version = "0.11", path = "../opentelemetry-http", optional = true }
 async-std = { workspace = true, features = ["unstable"], optional = true }
 async-trait = { workspace = true, optional = true }
-crossbeam-channel = { version = "0.5", optional = true }
 futures-channel = "0.3"
 futures-executor = { workspace = true }
 futures-util = { workspace = true, features = ["std", "sink", "async-await-macro"] }
@@ -45,9 +44,9 @@ pprof = { version = "0.13", features = ["flamegraph", "criterion"] }
 
 [features]
 default = ["trace"]
-trace = ["opentelemetry/trace", "crossbeam-channel", "rand", "async-trait", "percent-encoding"]
+trace = ["opentelemetry/trace", "rand", "async-trait", "percent-encoding"]
 jaeger_remote_sampler = ["trace", "opentelemetry-http", "http", "serde", "serde_json", "url"]
-logs = ["opentelemetry/logs", "crossbeam-channel", "async-trait", "serde_json"]
+logs = ["opentelemetry/logs", "async-trait", "serde_json"]
 logs_level_enabled = ["logs", "opentelemetry/logs_level_enabled"]
 metrics = ["opentelemetry/metrics", "glob", "async-trait"]
 testing = ["opentelemetry/testing", "trace", "metrics", "logs", "rt-async-std", "rt-tokio", "rt-tokio-current-thread", "tokio/macros", "tokio/rt-multi-thread"]

opentelemetry-sdk/src/logs/log_processor.rs

@@ -52,12 +52,10 @@ pub trait LogProcessor: Send + Sync + Debug {
     fn event_enabled(&self, level: Severity, target: &str, name: &str) -> bool;
 }
 
-/// A [`LogProcessor`] that exports synchronously when logs are emitted.
-///
-/// # Examples
-///
-/// Note that the simple processor exports synchronously every time a log is
-/// emitted. If you find this limiting, consider the batch processor instead.
+/// A [LogProcessor] that passes logs to the configured `LogExporter`, as soon
+/// as they are emitted, without any batching. This is typically useful for
+/// debugging and testing. For scenarios requiring higher
+/// performance/throughput, consider using [BatchLogProcessor].
 #[derive(Debug)]
 pub struct SimpleLogProcessor {
     exporter: Mutex<Box<dyn LogExporter>>,

opentelemetry-sdk/src/metrics/mod.rs

@@ -60,6 +60,7 @@ pub use view::*;
 
 #[cfg(all(test, feature = "testing"))]
 mod tests {
+    use self::data::ScopeMetrics;
     use super::*;
     use crate::metrics::data::{ResourceMetrics, Temporality};
     use crate::metrics::reader::TemporalitySelector;
@@ -70,6 +71,7 @@ mod tests {
         metrics::{MeterProvider as _, Unit},
         KeyValue,
     };
+    use std::borrow::Cow;
 
     // "multi_thread" tokio flavor must be used else flush won't
     // be able to make progress!
@@ -214,6 +216,179 @@ mod tests {
         assert_eq!(datapoint.value, 15);
     }
 
+    // "multi_thread" tokio flavor must be used else flush won't
+    // be able to make progress!
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn counter_duplicate_instrument_different_meter_no_merge() {
+        // Arrange
+        let exporter = InMemoryMetricsExporter::default();
+        let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
+        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
+
+        // Act
+        let meter1 = meter_provider.meter("test.meter1");
+        let meter2 = meter_provider.meter("test.meter2");
+        let counter1 = meter1
+            .u64_counter("my_counter")
+            .with_unit(Unit::new("my_unit"))
+            .with_description("my_description")
+            .init();
+
+        let counter2 = meter2
+            .u64_counter("my_counter")
+            .with_unit(Unit::new("my_unit"))
+            .with_description("my_description")
+            .init();
+
+        let attribute = vec![KeyValue::new("key1", "value1")];
+        counter1.add(10, &attribute);
+        counter2.add(5, &attribute);
+
+        meter_provider.force_flush().unwrap();
+
+        // Assert
+        let resource_metrics = exporter
+            .get_finished_metrics()
+            .expect("metrics are expected to be exported.");
+        assert!(
+            resource_metrics[0].scope_metrics.len() == 2,
+            "There should be 2 separate scope"
+        );
+        assert!(
+            resource_metrics[0].scope_metrics[0].metrics.len() == 1,
+            "There should be single metric for the scope"
+        );
+        assert!(
+            resource_metrics[0].scope_metrics[1].metrics.len() == 1,
+            "There should be single metric for the scope"
+        );
+
+        let scope1 = find_scope_metric(&resource_metrics[0].scope_metrics, "test.meter1");
+        let scope2 = find_scope_metric(&resource_metrics[0].scope_metrics, "test.meter2");
+
+        if let Some(scope1) = scope1 {
+            let metric1 = &scope1.metrics[0];
+            assert_eq!(metric1.name, "my_counter");
+            assert_eq!(metric1.unit.as_str(), "my_unit");
+            assert_eq!(metric1.description, "my_description");
+            let sum1 = metric1
+                .data
+                .as_any()
+                .downcast_ref::<data::Sum<u64>>()
+                .expect("Sum aggregation expected for Counter instruments by default");
+
+            // Expecting 1 time-series.
+            assert_eq!(sum1.data_points.len(), 1);
+
+            let datapoint1 = &sum1.data_points[0];
+            assert_eq!(datapoint1.value, 10);
+        } else {
+            panic!("No MetricScope found for 'test.meter1'");
+        }
+
+        if let Some(scope2) = scope2 {
+            let metric2 = &scope2.metrics[0];
+            assert_eq!(metric2.name, "my_counter");
+            assert_eq!(metric2.unit.as_str(), "my_unit");
+            assert_eq!(metric2.description, "my_description");
+            let sum2 = metric2
+                .data
+                .as_any()
+                .downcast_ref::<data::Sum<u64>>()
+                .expect("Sum aggregation expected for Counter instruments by default");
+
+            // Expecting 1 time-series.
+            assert_eq!(sum2.data_points.len(), 1);
+
+            let datapoint2 = &sum2.data_points[0];
+            assert_eq!(datapoint2.value, 5);
+        } else {
+            panic!("No MetricScope found for 'test.meter2'");
+        }
+    }
+
+    // "multi_thread" tokio flavor must be used else flush won't
+    // be able to make progress!
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn instrumentation_scope_identity_test() {
+        // Arrange
+        let exporter = InMemoryMetricsExporter::default();
+        let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
+        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
+
+        // Act
+        // Meters are identical except for scope attributes, but scope attributes are not an identifying property.
+        // Hence there should be a single metric stream output for this test.
+        let meter1 = meter_provider.versioned_meter(
+            "test.meter",
+            Some("v0.1.0"),
+            Some("schema_url"),
+            Some(vec![KeyValue::new("key", "value1")]),
+        );
+        let meter2 = meter_provider.versioned_meter(
+            "test.meter",
+            Some("v0.1.0"),
+            Some("schema_url"),
+            Some(vec![KeyValue::new("key", "value2")]),
+        );
+        let counter1 = meter1
+            .u64_counter("my_counter")
+            .with_unit(Unit::new("my_unit"))
+            .with_description("my_description")
+            .init();
+
+        let counter2 = meter2
+            .u64_counter("my_counter")
+            .with_unit(Unit::new("my_unit"))
+            .with_description("my_description")
+            .init();
+
+        let attribute = vec![KeyValue::new("key1", "value1")];
+        counter1.add(10, &attribute);
+        counter2.add(5, &attribute);
+
+        meter_provider.force_flush().unwrap();
+
+        // Assert
+        let resource_metrics = exporter
+            .get_finished_metrics()
+            .expect("metrics are expected to be exported.");
+        println!("resource_metrics: {:?}", resource_metrics);
+        assert!(
+            resource_metrics[0].scope_metrics.len() == 1,
+            "There should be a single scope as the meters are identical"
+        );
+        assert!(
+            resource_metrics[0].scope_metrics[0].metrics.len() == 1,
+            "There should be single metric for the scope as instruments are identical"
+        );
+
+        let scope = &resource_metrics[0].scope_metrics[0].scope;
+        assert_eq!(scope.name, "test.meter");
+        assert_eq!(scope.version, Some(Cow::Borrowed("v0.1.0")));
+        assert_eq!(scope.schema_url, Some(Cow::Borrowed("schema_url")));
+
+        // This is validating current behavior, but it is not guaranteed to be the case in the future,
+        // as this is a user error and SDK reserves right to change this behavior.
+        assert_eq!(scope.attributes, vec![KeyValue::new("key", "value1")]);
+
+        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
+        assert_eq!(metric.name, "my_counter");
+        assert_eq!(metric.unit.as_str(), "my_unit");
+        assert_eq!(metric.description, "my_description");
+        let sum = metric
+            .data
+            .as_any()
+            .downcast_ref::<data::Sum<u64>>()
+            .expect("Sum aggregation expected for Counter instruments by default");
+
+        // Expecting 1 time-series.
+        assert_eq!(sum.data_points.len(), 1);
+
+        let datapoint = &sum.data_points[0];
+        assert_eq!(datapoint.value, 15);
+    }
+
     // "multi_thread" tokio flavor must be used else flush won't
     // be able to make progress!
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
@@ -692,6 +867,15 @@ mod tests {
         assert!(resource_metrics.is_empty(), "No metrics should be exported as no new measurements were recorded since last collect.");
     }
 
+    fn find_scope_metric<'a>(
+        metrics: &'a [ScopeMetrics],
+        name: &'a str,
+    ) -> Option<&'a ScopeMetrics> {
+        metrics
+            .iter()
+            .find(|&scope_metric| scope_metric.scope.name == name)
+    }
+
     struct DeltaTemporalitySelector();
     impl TemporalitySelector for DeltaTemporalitySelector {
         fn temporality(&self, _kind: InstrumentKind) -> Temporality {

opentelemetry-sdk/src/testing/trace/span_exporters.rs

@@ -7,13 +7,15 @@ use crate::{
     InstrumentationLibrary,
 };
 use async_trait::async_trait;
-use crossbeam_channel::{unbounded, Receiver, SendError, Sender};
 use futures_util::future::BoxFuture;
 pub use opentelemetry::testing::trace::TestSpan;
 use opentelemetry::trace::{
     SpanContext, SpanId, SpanKind, Status, TraceFlags, TraceId, TraceState,
 };
-use std::fmt::{Display, Formatter};
+use std::{
+    fmt::{Display, Formatter},
+    sync::{Arc, Mutex},
+};
 
 pub fn new_test_export_span_data() -> SpanData {
     let config = Config::default();
@@ -40,42 +42,47 @@ pub fn new_test_export_span_data() -> SpanData {
     }
 }
 
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct TestSpanExporter {
-    tx_export: Sender<SpanData>,
-    tx_shutdown: Sender<()>,
+    pub export_called: Arc<Mutex<bool>>,
+    pub shutdown_called: Arc<Mutex<bool>>,
+}
+
+impl Default for TestSpanExporter {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl TestSpanExporter {
+    pub fn new() -> Self {
+        TestSpanExporter {
+            export_called: Arc::new(Mutex::new(false)),
+            shutdown_called: Arc::new(Mutex::new(false)),
+        }
+    }
+
+    pub fn is_export_called(&self) -> bool {
+        *self.export_called.lock().unwrap()
+    }
+
+    pub fn is_shutdown_called(&self) -> bool {
+        *self.shutdown_called.lock().unwrap()
+    }
 }
 
 #[async_trait]
 impl SpanExporter for TestSpanExporter {
-    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
-        for span_data in batch {
-            if let Err(err) = self
-                .tx_export
-                .send(span_data)
-                .map_err::<TestExportError, _>(Into::into)
-            {
-                return Box::pin(std::future::ready(Err(Into::into(err))));
-            }
-        }
+    fn export(&mut self, _batch: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
+        *self.export_called.lock().unwrap() = true;
         Box::pin(std::future::ready(Ok(())))
     }
 
     fn shutdown(&mut self) {
-        let _ = self.tx_shutdown.send(()); // ignore error
+        *self.shutdown_called.lock().unwrap() = true;
     }
 }
 
-pub fn new_test_exporter() -> (TestSpanExporter, Receiver<SpanData>, Receiver<()>) {
-    let (tx_export, rx_export) = unbounded();
-    let (tx_shutdown, rx_shutdown) = unbounded();
-    let exporter = TestSpanExporter {
-        tx_export,
-        tx_shutdown,
-    };
-    (exporter, rx_export, rx_shutdown)
-}
-
 #[derive(Debug)]
 pub struct TokioSpanExporter {
     tx_export: tokio::sync::mpsc::UnboundedSender<SpanData>,
@@ -139,12 +146,6 @@ impl<T> From<tokio::sync::mpsc::error::SendError<T>> for TestExportError {
     }
 }
 
-impl<T> From<crossbeam_channel::SendError<T>> for TestExportError {
-    fn from(err: SendError<T>) -> Self {
-        TestExportError(err.to_string())
-    }
-}
-
 /// A no-op instance of an [`SpanExporter`].
 ///
 /// [`SpanExporter`]: crate::export::trace::SpanExporter