Merge branch 'main' into ban/separate-example-names

Author: lalitb

.github/workflows/ci.yml

@@ -27,6 +27,7 @@ jobs:
           - rust: stable
             os: actuated-arm64-4cpu-16gb
     runs-on: ${{ matrix.os }}
+    continue-on-error: ${{ matrix.rust == 'beta' }}
     steps:
     - name: Free disk space
       if: ${{ matrix.os == 'ubuntu-latest'}}

examples/metrics-basic/Cargo.toml

@@ -6,12 +6,9 @@ license = "Apache-2.0"
 publish = false
 
 [dependencies]
-opentelemetry = { path = "../../opentelemetry", features = ["metrics", "otel_unstable"] }
+opentelemetry = { path = "../../opentelemetry", features = ["metrics"] }
 opentelemetry_sdk = { path = "../../opentelemetry-sdk", features = ["metrics", "rt-tokio"] }
 opentelemetry-stdout = { path = "../../opentelemetry-stdout", features = ["metrics"]}
 tokio = { workspace = true, features = ["full"] }
 serde_json = { workspace = true }
 
-[features]
-default = ["otel_unstable"]
-otel_unstable = ["opentelemetry/otel_unstable"]

examples/self-diagnostics/Cargo.toml

@@ -12,4 +12,4 @@ opentelemetry-stdout = { path = "../../opentelemetry-stdout"}
 tokio = { workspace = true, features = ["full"] }
 tracing = { workspace = true, features = ["std"]}
 tracing-core = { workspace = true }
-tracing-subscriber = { version = "0.3.18", features = ["env-filter","registry", "std"]}
+tracing-subscriber = { workspace = true, features = ["env-filter","registry", "std", "fmt"]}

opentelemetry-sdk/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## vNext
 
+## 0.27.1
+
+Released 2024-Nov-27
+
 - **DEPRECATED**:
   - `trace::Config` methods are moving onto `TracerProvider` Builder to be consistent with other signals. See https://github.com/open-telemetry/opentelemetry-rust/pull/2303 for migration guide.
     `trace::Config` is scheduled to be removed from public API in `v0.28.0`.

opentelemetry-sdk/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "opentelemetry_sdk"
-version = "0.27.0"
+version = "0.27.1"
 description = "The SDK for the OpenTelemetry metrics collection and distributed tracing framework"
 homepage = "https://github.com/open-telemetry/opentelemetry-rust"
 repository = "https://github.com/open-telemetry/opentelemetry-rust"

opentelemetry-sdk/src/logs/log_emitter.rs

@@ -59,11 +59,19 @@ impl opentelemetry::logs::LoggerProvider for LoggerProvider {
     fn logger_with_scope(&self, scope: InstrumentationScope) -> Self::Logger {
         // If the provider is shutdown, new logger will refer a no-op logger provider.
         if self.inner.is_shutdown.load(Ordering::Relaxed) {
+            otel_debug!(
+                name: "LoggerProvider.NoOpLoggerReturned",
+                logger_name = scope.name(),
+            );
             return Logger::new(scope, noop_logger_provider().clone());
         }
         if scope.name().is_empty() {
             otel_info!(name: "LoggerNameEmpty",  message = "Logger name is empty; consider providing a meaningful name. Logger will function normally and the provided name will be used as-is.");
         };
+        otel_debug!(
+            name: "LoggerProvider.NewLoggerReturned",
+            logger_name = scope.name(),
+        );
         Logger::new(scope, self.clone())
     }
 }
@@ -92,6 +100,9 @@ impl LoggerProvider {
 
     /// Shuts down this `LoggerProvider`
     pub fn shutdown(&self) -> LogResult<()> {
+        otel_debug!(
+            name: "LoggerProvider.ShutdownInvokedByUser",
+        );
         if self
             .inner
             .is_shutdown
@@ -224,6 +235,10 @@ impl Builder {
         for processor in logger_provider.log_processors() {
             processor.set_resource(logger_provider.resource());
         }
+
+        otel_debug!(
+            name: "LoggerProvider.Built",
+        );
         logger_provider
     }
 }

opentelemetry-sdk/src/logs/log_processor.rs

@@ -13,7 +13,7 @@ use futures_util::{
 use opentelemetry::logs::Severity;
 use opentelemetry::{otel_debug, otel_error, otel_warn, InstrumentationScope};
 
-use std::sync::atomic::AtomicBool;
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::{cmp::min, env, sync::Mutex};
 use std::{
     fmt::{self, Debug, Formatter},
@@ -154,6 +154,12 @@ impl LogProcessor for SimpleLogProcessor {
 /// them at a pre-configured interval.
 pub struct BatchLogProcessor<R: RuntimeChannel> {
     message_sender: R::Sender<BatchMessage>,
+
+    // Track dropped logs - we'll log this at shutdown
+    dropped_logs_count: AtomicUsize,
+
+    // Track the maximum queue size that was configured for this processor
+    max_queue_size: usize,
 }
 
 impl<R: RuntimeChannel> Debug for BatchLogProcessor<R> {
@@ -172,11 +178,13 @@ impl<R: RuntimeChannel> LogProcessor for BatchLogProcessor<R> {
         )));
 
         // TODO - Implement throttling to prevent error flooding when the queue is full or closed.
-        if let Err(err) = result {
-            otel_error!(
-                name: "BatchLogProcessor.Export.Error",
-                error = format!("{}", err)
-            );
+        if result.is_err() {
+            // Increment dropped logs count. The first time we have to drop a log,
+            // emit a warning.
+            if self.dropped_logs_count.fetch_add(1, Ordering::Relaxed) == 0 {
+                otel_warn!(name: "BatchLogProcessor.LogDroppingStarted",
+                    message = "BatchLogProcessor dropped a LogRecord due to queue full/internal errors. No further log will be emitted for further drops until Shutdown. During Shutdown time, a log will be emitted with exact count of total logs dropped.");
+            }
         }
     }
 
@@ -192,6 +200,17 @@ impl<R: RuntimeChannel> LogProcessor for BatchLogProcessor<R> {
     }
 
     fn shutdown(&self) -> LogResult<()> {
+        let dropped_logs = self.dropped_logs_count.load(Ordering::Relaxed);
+        let max_queue_size = self.max_queue_size;
+        if dropped_logs > 0 {
+            otel_warn!(
+                name: "BatchLogProcessor.LogsDropped",
+                dropped_logs_count = dropped_logs,
+                max_queue_size = max_queue_size,
+                message = "Logs were dropped due to a queue being full or other error. The count represents the total count of log records dropped in the lifetime of this BatchLogProcessor. Consider increasing the queue size and/or decrease delay between intervals."
+            );
+        }
+
         let (res_sender, res_receiver) = oneshot::channel();
         self.message_sender
             .try_send(BatchMessage::Shutdown(res_sender))
@@ -215,6 +234,7 @@ impl<R: RuntimeChannel> BatchLogProcessor<R> {
         let (message_sender, message_receiver) =
             runtime.batch_message_channel(config.max_queue_size);
         let inner_runtime = runtime.clone();
+        let max_queue_size = config.max_queue_size;
 
         // Spawn worker process via user-defined spawn function.
         runtime.spawn(Box::pin(async move {
@@ -296,8 +316,13 @@ impl<R: RuntimeChannel> BatchLogProcessor<R> {
                 }
             }
         }));
+
         // Return batch processor with link to worker
-        BatchLogProcessor { message_sender }
+        BatchLogProcessor {
+            message_sender,
+            dropped_logs_count: AtomicUsize::new(0),
+            max_queue_size,
+        }
     }
 
     /// Create a new batch processor builder

opentelemetry-sdk/src/trace/span_processor.rs

@@ -45,12 +45,13 @@ use futures_util::{
     stream::{self, FusedStream, FuturesUnordered},
     StreamExt as _,
 };
-use opentelemetry::{otel_debug, otel_error};
+use opentelemetry::{otel_debug, otel_error, otel_warn};
 use opentelemetry::{
     trace::{TraceError, TraceResult},
     Context,
 };
 use std::cmp::min;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
 use std::{env, fmt, str::FromStr, time::Duration};
 
@@ -227,6 +228,12 @@ impl SpanProcessor for SimpleSpanProcessor {
 /// [`async-std`]: https://async.rs
 pub struct BatchSpanProcessor<R: RuntimeChannel> {
     message_sender: R::Sender<BatchMessage>,
+
+    // Track dropped spans
+    dropped_spans_count: AtomicUsize,
+
+    // Track the maximum queue size that was configured for this processor
+    max_queue_size: usize,
 }
 
 impl<R: RuntimeChannel> fmt::Debug for BatchSpanProcessor<R> {
@@ -249,11 +256,14 @@ impl<R: RuntimeChannel> SpanProcessor for BatchSpanProcessor<R> {
 
         let result = self.message_sender.try_send(BatchMessage::ExportSpan(span));
 
-        if let Err(err) = result {
-            otel_debug!(
-                name: "BatchSpanProcessor.OnEnd.ExportQueueingFailed",
-                reason = format!("{:?}", TraceError::Other(err.into()))
-            );
+        // If the queue is full, and we can't buffer a span
+        if result.is_err() {
+            // Increment the number of dropped spans. If this is the first time we've had to drop,
+            // emit a warning.
+            if self.dropped_spans_count.fetch_add(1, Ordering::Relaxed) == 0 {
+                otel_warn!(name: "BatchSpanProcessor.SpanDroppingStarted",
+                    message = "Beginning to drop span messages due to full/internal errors. No further log will be emitted for further drops until Shutdown. During Shutdown time, a log will be emitted with exact count of total spans dropped.");
+            }
         }
     }
 
@@ -269,6 +279,17 @@ impl<R: RuntimeChannel> SpanProcessor for BatchSpanProcessor<R> {
     }
 
     fn shutdown(&self) -> TraceResult<()> {
+        let dropped_spans = self.dropped_spans_count.load(Ordering::Relaxed);
+        let max_queue_size = self.max_queue_size;
+        if dropped_spans > 0 {
+            otel_warn!(
+                name: "BatchSpanProcessor.Shutdown",
+                dropped_spans = dropped_spans,
+                max_queue_size = max_queue_size,
+                message = "Spans were dropped due to a full or closed queue. The count represents the total count of span records dropped in the lifetime of the BatchLogProcessor. Consider increasing the queue size and/or decrease delay between intervals."
+            );
+        }
+
         let (res_sender, res_receiver) = oneshot::channel();
         self.message_sender
             .try_send(BatchMessage::Shutdown(res_sender))
@@ -469,6 +490,8 @@ impl<R: RuntimeChannel> BatchSpanProcessor<R> {
         let (message_sender, message_receiver) =
             runtime.batch_message_channel(config.max_queue_size);
 
+        let max_queue_size = config.max_queue_size;
+
         let inner_runtime = runtime.clone();
         // Spawn worker process via user-defined spawn function.
         runtime.spawn(Box::pin(async move {
@@ -493,7 +516,11 @@ impl<R: RuntimeChannel> BatchSpanProcessor<R> {
         }));
 
         // Return batch processor with link to worker
-        BatchSpanProcessor { message_sender }
+        BatchSpanProcessor {
+            message_sender,
+            dropped_spans_count: AtomicUsize::new(0),
+            max_queue_size,
+        }
     }
 
     /// Create a new batch processor builder

opentelemetry/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## vNext
 
+## 0.27.1
+
+Released 2024-Nov-27
+
 ## 0.27.0
 
 Released 2024-Nov-11

opentelemetry/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "opentelemetry"
-version = "0.27.0"
+version = "0.27.1"
 description = "OpenTelemetry API for Rust"
 homepage = "https://github.com/open-telemetry/opentelemetry-rust"
 repository = "https://github.com/open-telemetry/opentelemetry-rust"