Merge branch 'main' into otel-sdk-remove-oncecell

Author: lalitb

README.md

@@ -24,15 +24,19 @@ observability tools.
 
 ## Project Status
 
+The table below summarizes the overall status of each component. Some components
+include unstable features, which are documented in their respective crate
+documentation.
+
 | Signal/Component      | Overall Status     |
 | --------------------  | ------------------ |
 | Logs-API              | RC*                |
 | Logs-SDK              | Beta               |
 | Logs-OTLP Exporter    | Beta               |
 | Logs-Appender-Tracing | Beta               |
 | Metrics-API           | RC                 |
-| Metrics-SDK           | Beta              |
-| Metrics-OTLP Exporter | Beta              |
+| Metrics-SDK           | Beta               |
+| Metrics-OTLP Exporter | Beta               |
 | Traces-API            | Beta               |
 | Traces-SDK            | Beta               |
 | Traces-OTLP Exporter  | Beta               |

examples/self-diagnostics/src/main.rs

@@ -1,6 +1,7 @@
 use opentelemetry::global;
 use opentelemetry::KeyValue;
 use opentelemetry_sdk::metrics::PeriodicReader;
+use opentelemetry_sdk::Resource;
 use std::error::Error;
 use tracing::info;
 use tracing_subscriber::fmt;
@@ -13,6 +14,10 @@ fn init_meter_provider() -> opentelemetry_sdk::metrics::SdkMeterProvider {
     let reader = PeriodicReader::builder(exporter, opentelemetry_sdk::runtime::Tokio).build();
 
     let provider = opentelemetry_sdk::metrics::SdkMeterProvider::builder()
+        .with_resource(Resource::new([KeyValue::new(
+            "service.name",
+            "self-diagnostics-example",
+        )]))
         .with_reader(reader)
         .build();
 
@@ -26,7 +31,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     // OpenTelemetry uses `tracing` crate for its internal logging. Unless a
     // tracing subscriber is set, the logs will be discarded. In this example,
     // we configure a `tracing` subscriber to:
-    // 1. Print logs of level INFO or higher to stdout using tracing's fmt layer.
+    // 1. Print logs of level DEBUG or higher to stdout using tracing's fmt layer.
     // 2. Filter logs from OpenTelemetry's dependencies (like tonic, hyper,
     // reqwest etc. which are commonly used by the OTLP exporter) to only print
     // ERROR-level logs. This filtering helps reduce repetitive log messages
@@ -39,7 +44,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     // Hence, one may use "add_directive("opentelemetry=off".parse().unwrap())"
     // to turn off all logs from OpenTelemetry.
 
-    let filter = EnvFilter::new("info")
+    let filter = EnvFilter::new("debug")
         .add_directive("hyper=error".parse().unwrap())
         .add_directive("tonic=error".parse().unwrap())
         .add_directive("h2=error".parse().unwrap())
@@ -54,11 +59,14 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     info!("Starting self-diagnostics example");
 
     let meter = global::meter("example");
-    // Create a counter using an invalid name to trigger
-    // internal log about the same.
-    let counter = meter.u64_counter("my_counter with_space").build();
+    let counter = meter.u64_counter("my_counter").build();
     counter.add(10, &[KeyValue::new("key", "value")]);
 
+    let _observable_counter = meter
+        .u64_observable_counter("my_observable_counter")
+        .with_callback(|observer| observer.observe(10, &[KeyValue::new("key", "value")]))
+        .build();
+
     meter_provider.shutdown()?;
     info!("Shutdown complete. Bye!");
     Ok(())

opentelemetry-sdk/CHANGELOG.md

@@ -25,6 +25,12 @@
   - Bug fix: Empty Logger names are retained as-is instead of replacing with
     "rust.opentelemetry.io/sdk/logger"
     [#2316](https://github.com/open-telemetry/opentelemetry-rust/pull/2316)
+  
+  - `Logger::provider`: This method is deprecated as of version `0.27.1`. To be removed in `0.28.0`.
+  - `Logger::instrumentation_scope`: This method is deprecated as of version `0.27.1`. To be removed in `0.28.0`
+     Migration Guidance: 
+        - These methods are intended for log appenders. Keep the clone of the provider handle, instead of depending on above methods.
+
 
 ## 0.27.0
 

opentelemetry-sdk/src/logs/log_emitter.rs

@@ -242,11 +242,19 @@ impl Logger {
         Logger { scope, provider }
     }
 
+    #[deprecated(
+        since = "0.27.1",
+        note = "This method was intended for appender developers, but has no defined use-case in typical workflows. It is deprecated and will be removed in the next major release."
+    )]
     /// LoggerProvider associated with this logger.
     pub fn provider(&self) -> &LoggerProvider {
         &self.provider
     }
 
+    #[deprecated(
+        since = "0.27.1",
+        note = "This method was intended for appender developers, but has no defined use-case in typical workflows. It is deprecated and will be removed in the next major release."
+    )]
     /// Instrumentation scope of this logger.
     pub fn instrumentation_scope(&self) -> &InstrumentationScope {
         &self.scope
@@ -262,7 +270,7 @@ impl opentelemetry::logs::Logger for Logger {
 
     /// Emit a `LogRecord`.
     fn emit(&self, mut record: Self::LogRecord) {
-        let provider = self.provider();
+        let provider = &self.provider;
         let processors = provider.log_processors();
 
         //let mut log_record = record;
@@ -281,22 +289,17 @@ impl opentelemetry::logs::Logger for Logger {
         }
 
         for p in processors {
-            p.emit(&mut record, self.instrumentation_scope());
+            p.emit(&mut record, &self.scope);
         }
     }
 
     #[cfg(feature = "spec_unstable_logs_enabled")]
     fn event_enabled(&self, level: Severity, target: &str) -> bool {
-        let provider = self.provider();
+        let provider = &self.provider;
 
         let mut enabled = false;
         for processor in provider.log_processors() {
-            enabled = enabled
-                || processor.event_enabled(
-                    level,
-                    target,
-                    self.instrumentation_scope().name().as_ref(),
-                );
+            enabled = enabled || processor.event_enabled(level, target, self.scope.name().as_ref());
         }
         enabled
     }
@@ -729,14 +732,14 @@ mod tests {
             emitted_logs[0].clone().record.body,
             Some(AnyValue::String("Testing empty logger name".into()))
         );
-        assert_eq!(logger.instrumentation_scope().name(), "");
+        assert_eq!(logger.scope.name(), "");
 
         // Assert the second log created through the scope
         assert_eq!(
             emitted_logs[1].clone().record.body,
             Some(AnyValue::String("Testing empty logger scope name".into()))
         );
-        assert_eq!(scoped_logger.instrumentation_scope().name(), "");
+        assert_eq!(scoped_logger.scope.name(), "");
     }
 
     #[derive(Debug)]

opentelemetry-sdk/src/logs/mod.rs

@@ -108,6 +108,7 @@ mod tests {
     }
 
     #[test]
+    #[allow(deprecated)]
     fn logger_attributes() {
         let provider = LoggerProvider::builder().build();
         let scope = InstrumentationScope::builder("test_logger")

opentelemetry-sdk/src/metrics/internal/aggregate.rs

@@ -12,11 +12,11 @@ use super::{
     precomputed_sum::PrecomputedSum, sum::Sum, Number,
 };
 
-const STREAM_CARDINALITY_LIMIT: u32 = 2000;
+pub(crate) const STREAM_CARDINALITY_LIMIT: usize = 2000;
 
 /// Checks whether aggregator has hit cardinality limit for metric streams
 pub(crate) fn is_under_cardinality_limit(size: usize) -> bool {
-    size < STREAM_CARDINALITY_LIMIT as usize
+    size < STREAM_CARDINALITY_LIMIT
 }
 
 /// Receives measurements to be aggregated.

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

@@ -7,12 +7,12 @@ mod sum;
 
 use core::fmt;
 use std::collections::{HashMap, HashSet};
-use std::mem::take;
+use std::mem::swap;
 use std::ops::{Add, AddAssign, DerefMut, Sub};
 use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, AtomicUsize, Ordering};
 use std::sync::{Arc, RwLock};
 
-use aggregate::is_under_cardinality_limit;
+use aggregate::{is_under_cardinality_limit, STREAM_CARDINALITY_LIMIT};
 pub(crate) use aggregate::{AggregateBuilder, ComputeAggregation, Measure};
 pub(crate) use exponential_histogram::{EXPO_MAX_SCALE, EXPO_MIN_SCALE};
 use opentelemetry::{otel_warn, KeyValue};
@@ -56,6 +56,11 @@ where
 {
     /// Trackers store the values associated with different attribute sets.
     trackers: RwLock<HashMap<Vec<KeyValue>, Arc<A>>>,
+
+    /// Used by collect exclusively. The data type must match the one used in
+    /// `trackers` to allow mem::swap.
+    trackers_for_collect: RwLock<HashMap<Vec<KeyValue>, Arc<A>>>,
+
     /// Number of different attribute set stored in the `trackers` map.
     count: AtomicUsize,
     /// Indicates whether a value with no attributes has been stored.
@@ -72,7 +77,10 @@ where
 {
     fn new(config: A::InitConfig) -> Self {
         ValueMap {
-            trackers: RwLock::new(HashMap::new()),
+            trackers: RwLock::new(HashMap::with_capacity(1 + STREAM_CARDINALITY_LIMIT)),
+            // TODO: For cumulative, this is not required, so avoid this
+            // pre-allocation.
+            trackers_for_collect: RwLock::new(HashMap::with_capacity(1 + STREAM_CARDINALITY_LIMIT)),
             has_no_attribute_value: AtomicBool::new(false),
             no_attribute_tracker: A::create(&config),
             count: AtomicUsize::new(0),
@@ -175,19 +183,23 @@ where
             ));
         }
 
-        let trackers = match self.trackers.write() {
-            Ok(mut trackers) => {
+        if let Ok(mut trackers_collect) = self.trackers_for_collect.write() {
+            if let Ok(mut trackers_current) = self.trackers.write() {
+                swap(trackers_collect.deref_mut(), trackers_current.deref_mut());
                 self.count.store(0, Ordering::SeqCst);
-                take(trackers.deref_mut())
+            } else {
+                otel_warn!(name: "MeterProvider.InternalError", message = "Metric collection failed. Report this issue in OpenTelemetry repo.", details ="ValueMap trackers lock poisoned");
+                return;
             }
-            Err(_) => todo!(),
-        };
 
-        let mut seen = HashSet::new();
-        for (attrs, tracker) in trackers.into_iter() {
-            if seen.insert(Arc::as_ptr(&tracker)) {
-                dest.push(map_fn(attrs, tracker.clone_and_reset(&self.config)));
+            let mut seen = HashSet::new();
+            for (attrs, tracker) in trackers_collect.drain() {
+                if seen.insert(Arc::as_ptr(&tracker)) {
+                    dest.push(map_fn(attrs, tracker.clone_and_reset(&self.config)));
+                }
             }
+        } else {
+            otel_warn!(name: "MeterProvider.InternalError", message = "Metric collection failed. Report this issue in OpenTelemetry repo.", details ="ValueMap trackers for collect lock poisoned");
         }
     }
 }

opentelemetry-sdk/src/metrics/meter_provider.rs

@@ -174,6 +174,10 @@ impl MeterProvider for SdkMeterProvider {
 
     fn meter_with_scope(&self, scope: InstrumentationScope) -> Meter {
         if self.inner.is_shutdown.load(Ordering::Relaxed) {
+            otel_debug!(
+                name: "MeterProvider.NoOpMeterReturned",
+                meter_name = scope.name(),
+            );
             return Meter::new(Arc::new(NoopMeter::new()));
         }
 
@@ -182,14 +186,26 @@ impl MeterProvider for SdkMeterProvider {
         };
 
         if let Ok(mut meters) = self.inner.meters.lock() {
-            let meter = meters
-                .entry(scope)
-                .or_insert_with_key(|scope| {
-                    Arc::new(SdkMeter::new(scope.clone(), self.inner.pipes.clone()))
-                })
-                .clone();
-            Meter::new(meter)
+            if let Some(existing_meter) = meters.get(&scope) {
+                otel_debug!(
+                    name: "MeterProvider.ExistingMeterReturned",
+                    meter_name = scope.name(),
+                );
+                Meter::new(existing_meter.clone())
+            } else {
+                let new_meter = Arc::new(SdkMeter::new(scope.clone(), self.inner.pipes.clone()));
+                meters.insert(scope.clone(), new_meter.clone());
+                otel_debug!(
+                    name: "MeterProvider.NewMeterCreated",
+                    meter_name = scope.name(),
+                );
+                Meter::new(new_meter)
+            }
         } else {
+            otel_debug!(
+                name: "MeterProvider.NoOpMeterReturned",
+                meter_name = scope.name(),
+            );
             Meter::new(Arc::new(NoopMeter::new()))
         }
     }
@@ -242,6 +258,11 @@ impl MeterProviderBuilder {
     /// Construct a new [MeterProvider] with this configuration.
 
     pub fn build(self) -> SdkMeterProvider {
+        otel_debug!(
+            name: "MeterProvider.Building",
+            builder = format!("{:?}", &self),
+        );
+
         let meter_provider = SdkMeterProvider {
             inner: Arc::new(SdkMeterProviderInner {
                 pipes: Arc::new(Pipelines::new(

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -132,6 +132,7 @@ where
             name: "PeriodicReader.BuildCompleted",
             message = "Periodic reader built.",
             interval_in_secs = self.interval.as_secs(),
+            temporality = format!("{:?}", self.exporter.temporality()),
         );
 
         PeriodicReader {

opentelemetry-sdk/src/metrics/pipeline.rs

@@ -72,6 +72,10 @@ impl Pipeline {
     /// unique values.
     fn add_sync(&self, scope: InstrumentationScope, i_sync: InstrumentSync) {
         let _ = self.inner.lock().map(|mut inner| {
+            otel_debug!(
+                name : "InstrumentCreated",
+                instrument_name = i_sync.name.as_ref(),
+            );
             inner.aggregations.entry(scope).or_default().push(i_sync);
         });
     }
@@ -99,6 +103,10 @@ impl SdkProducer for Pipeline {
     /// Returns aggregated metrics from a single collection.
     fn produce(&self, rm: &mut ResourceMetrics) -> MetricResult<()> {
         let inner = self.inner.lock()?;
+        otel_debug!(
+            name: "MeterProviderInvokingObservableCallbacks",
+            count =  inner.callbacks.len(),
+        );
         for cb in &inner.callbacks {
             // TODO consider parallel callbacks.
             cb();