Merge branch 'main' into preallocate-attribute-mem

Author: lalitb

opentelemetry-sdk/CHANGELOG.md

@@ -3,6 +3,7 @@
 ## vNext
 
 - `opentelemetry_sdk::logs::record::LogRecord` and `opentelemetry_sdk::logs::record::TraceContext` derive from `PartialEq` to facilitate Unit Testing.
+- Fixed an issue causing a panic during shutdown when using the `TokioCurrentThread` tracing batch processor.
 
 ## v0.24.1
 

opentelemetry-sdk/benches/log.rs

@@ -6,27 +6,41 @@
 use std::collections::HashMap;
 use std::time::SystemTime;
 
-use async_trait::async_trait;
 use criterion::{criterion_group, criterion_main, Criterion};
 
 use opentelemetry::logs::{
     AnyValue, LogRecord as _, LogResult, Logger as _, LoggerProvider as _, Severity,
 };
 use opentelemetry::trace::Tracer;
 use opentelemetry::trace::TracerProvider as _;
-use opentelemetry::{InstrumentationLibrary, Key};
-use opentelemetry_sdk::export::logs::{LogData, LogExporter};
-use opentelemetry_sdk::logs::{Logger, LoggerProvider};
+use opentelemetry::Key;
+use opentelemetry_sdk::export::logs::LogData;
+use opentelemetry_sdk::logs::{Logger, LoggerProvider, LogProcessor};
 use opentelemetry_sdk::trace;
 use opentelemetry_sdk::trace::{Sampler, TracerProvider};
 
 #[derive(Debug)]
-struct VoidExporter;
+struct NoopProcessor;
 
-#[async_trait]
-impl LogExporter for VoidExporter {
-    async fn export<'a>(&mut self, _batch: Vec<std::borrow::Cow<'a, LogData>>) -> LogResult<()> {
-        LogResult::Ok(())
+impl LogProcessor for NoopProcessor {
+    fn emit(&self, _data: &mut LogData) {}
+
+    fn force_flush(&self) -> LogResult<()> {
+        Ok(())
+    }
+
+    fn shutdown(&self) -> LogResult<()> {
+        Ok(())
+    }
+
+    #[cfg(feature = "logs_level_enabled")]
+    fn event_enabled(
+        &self,
+        _level: opentelemetry::logs::Severity,
+        _target: &str,
+        _name: &str,
+    ) -> bool {
+        true
     }
 }
 
@@ -35,7 +49,7 @@ fn log_benchmark_group<F: Fn(&Logger)>(c: &mut Criterion, name: &str, f: F) {
 
     group.bench_function("no-context", |b| {
         let provider = LoggerProvider::builder()
-            .with_simple_exporter(VoidExporter)
+            .with_log_processor(NoopProcessor {})
             .build();
 
         let logger = provider.logger("no-context");
@@ -45,7 +59,7 @@ fn log_benchmark_group<F: Fn(&Logger)>(c: &mut Criterion, name: &str, f: F) {
 
     group.bench_function("with-context", |b| {
         let provider = LoggerProvider::builder()
-            .with_simple_exporter(VoidExporter)
+            .with_log_processor(NoopProcessor {})
             .build();
 
         let logger = provider.logger("with-context");

opentelemetry-sdk/src/trace/mod.rs

@@ -4,7 +4,7 @@
 //!
 //! * The [`Tracer`] struct which performs all tracing operations.
 //! * The [`Span`] struct with is a mutable object storing information about the
-//! current operation execution.
+//!   current operation execution.
 //! * The [`TracerProvider`] struct which configures and produces [`Tracer`]s.
 mod config;
 mod events;

opentelemetry-sdk/src/trace/span_processor.rs

@@ -449,22 +449,28 @@ impl<R: RuntimeChannel> BatchSpanProcessor<R> {
     pub(crate) fn new(exporter: Box<dyn SpanExporter>, config: BatchConfig, runtime: R) -> Self {
         let (message_sender, message_receiver) =
             runtime.batch_message_channel(config.max_queue_size);
-        let ticker = runtime
-            .interval(config.scheduled_delay)
-            .map(|_| BatchMessage::Flush(None));
-        let timeout_runtime = runtime.clone();
-
-        let messages = Box::pin(stream::select(message_receiver, ticker));
-        let processor = BatchSpanProcessorInternal {
-            spans: Vec::new(),
-            export_tasks: FuturesUnordered::new(),
-            runtime: timeout_runtime,
-            config,
-            exporter,
-        };
 
+        let inner_runtime = runtime.clone();
         // Spawn worker process via user-defined spawn function.
-        runtime.spawn(Box::pin(processor.run(messages)));
+        runtime.spawn(Box::pin(async move {
+            // Timer will take a reference to the current runtime, so its important we do this within the
+            // runtime.spawn()
+            let ticker = inner_runtime
+                .interval(config.scheduled_delay)
+                .map(|_| BatchMessage::Flush(None));
+            let timeout_runtime = inner_runtime.clone();
+
+            let messages = Box::pin(stream::select(message_receiver, ticker));
+            let processor = BatchSpanProcessorInternal {
+                spans: Vec::new(),
+                export_tasks: FuturesUnordered::new(),
+                runtime: timeout_runtime,
+                config,
+                exporter,
+            };
+
+            processor.run(messages).await
+        }));
 
         // Return batch processor with link to worker
         BatchSpanProcessor { message_sender }

opentelemetry-zipkin/src/propagator/mod.rs

@@ -3,13 +3,13 @@
 //! The `B3Propagator` facilitates `SpanContext` propagation using
 //! B3 Headers. This propagator supports both version of B3 headers,
 //!  1. Single Header:
-//!    b3: {trace_id}-{span_id}-{sampling_state}-{parent_span_id}
+//!     b3: {trace_id}-{span_id}-{sampling_state}-{parent_span_id}
 //!  2. Multiple Headers:
-//!    X-B3-TraceId: {trace_id}
-//!    X-B3-ParentSpanId: {parent_span_id}
-//!    X-B3-SpanId: {span_id}
-//!    X-B3-Sampled: {sampling_state}
-//!    X-B3-Flags: {debug_flag}
+//!     X-B3-TraceId: {trace_id}
+//!     X-B3-ParentSpanId: {parent_span_id}
+//!     X-B3-SpanId: {span_id}
+//!     X-B3-Sampled: {sampling_state}
+//!     X-B3-Flags: {debug_flag}
 //!
 //! If `inject_encoding` is set to `B3Encoding::SingleHeader` then `b3` header is used to inject
 //! and extract. Otherwise, separate headers are used to inject and extract.