Merge branch 'main' into cijothomas/fixperiodicreaderpanic

Author: cijothomas

opentelemetry-prometheus/examples/hyper.rs

@@ -15,7 +15,7 @@ use opentelemetry_sdk::metrics::SdkMeterProvider;
 use prometheus::{Encoder, Registry, TextEncoder};
 use std::net::SocketAddr;
 use std::sync::Arc;
-use std::time::SystemTime;
+use opentelemetry::time::now;
 use tokio::net::TcpListener;
 
 static HANDLER_ALL: Lazy<[KeyValue; 1]> = Lazy::new(|| [KeyValue::new("handler", "all")]);
@@ -25,7 +25,7 @@ async fn serve_req(
     state: Arc<AppState>,
 ) -> Result<Response<Full<Bytes>>, hyper::Error> {
     println!("Receiving request at path {}", req.uri());
-    let request_start = SystemTime::now();
+    let request_start = now();
 
     state.http_counter.add(1, HANDLER_ALL.as_ref());
 

opentelemetry-sdk/CHANGELOG.md

@@ -337,6 +337,39 @@ let processor = BatchLogProcessor::builder(exporter)
     .build();
 ```
 
+- *Breaking*: The `BatchSpanProcessor` no longer supports configuration of `max_export_timeout` 
+   or the `OTEL_BLRP_EXPORT_TIMEOUT` environment variable. Timeout handling is now the 
+   responsibility of the exporter.
+   For example, in the OTLP Span exporter, the export timeout can be configured using:
+   - The environment variables `OTEL_EXPORTER_OTLP_TIMEOUT` or `OTEL_EXPORTER_OTLP_TRACES_TIMEOUT`.
+   - The opentelemetry_otlp API, via `.with_tonic().with_timeout()` or `.with_http().with_timeout()`.
+  Before:
+```rust
+let processor = BatchSpanProcessor::builder(exporter)
+    .with_batch_config(
+        BatchConfigBuilder::default()
+            .with_max_queue_size(2048)
+            .with_max_export_batch_size(512)
+            .with_scheduled_delay(Duration::from_secs(5))
+            .with_max_export_timeout(Duration::from_secs(30)) // Previously configurable
+            .build(),
+    )
+    .build();
+```
+
+After:
+```rust
+let processor = BatchSpanProcessor::builder(exporter)
+    .with_batch_config(
+        BatchConfigBuilder::default()
+            .with_max_queue_size(2048)
+            .with_max_export_batch_size(512)
+            .with_scheduled_delay(Duration::from_secs(5)) // No `max_export_timeout`
+            .build(),
+    )
+    .build();
+```
+
 ## 0.27.1
 
 Released 2024-Nov-27

opentelemetry-sdk/benches/batch_span_processor.rs

@@ -1,4 +1,5 @@
 use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
+use opentelemetry::time::now;
 use opentelemetry::trace::{
     SpanContext, SpanId, SpanKind, Status, TraceFlags, TraceId, TraceState,
 };
@@ -8,7 +9,6 @@ use opentelemetry_sdk::trace::{
     BatchConfigBuilder, BatchSpanProcessor, SpanEvents, SpanLinks, SpanProcessor,
 };
 use std::sync::Arc;
-use std::time::SystemTime;
 use tokio::runtime::Runtime;
 
 fn get_span_data() -> Vec<SpanData> {
@@ -24,8 +24,8 @@ fn get_span_data() -> Vec<SpanData> {
             parent_span_id: SpanId::from_u64(12),
             span_kind: SpanKind::Client,
             name: Default::default(),
-            start_time: SystemTime::now(),
-            end_time: SystemTime::now(),
+            start_time: now(),
+            end_time: now(),
             attributes: Vec::new(),
             dropped_attributes_count: 0,
             events: SpanEvents::default(),

opentelemetry-sdk/benches/log.rs

@@ -15,8 +15,8 @@ RAM: 64.0 GB
 | Logging_Comparable_To_Appender | 87 ns       |
 */
 
+use opentelemetry::time::now;
 use std::collections::HashMap;
-use std::time::SystemTime;
 
 use criterion::{criterion_group, criterion_main, Criterion};
 
@@ -111,7 +111,7 @@ fn logging_comparable_to_appender(c: &mut Criterion) {
     c.bench_function("Logging_Comparable_To_Appender", |b| {
         b.iter(|| {
             let mut log_record = logger.create_log_record();
-            let now = SystemTime::now();
+            let now = now();
             log_record.set_observed_timestamp(now);
             log_record.set_target("my-target".to_string());
             log_record.set_event_name("CheckoutFailed");
@@ -253,7 +253,7 @@ fn criterion_benchmark(c: &mut Criterion) {
         logger.emit(log_record);
     });
 
-    let now = SystemTime::now();
+    let now = now();
     log_benchmark_group(c, "full-log", |logger| {
         let mut log_record = logger.create_log_record();
         log_record.set_body("full log".into());

opentelemetry-sdk/benches/log_exporter.rs

@@ -10,8 +10,8 @@
     | LogExporterWithoutFuture       | 92 ns      |
 */
 
+use opentelemetry::time::now;
 use std::sync::Mutex;
-use std::time::SystemTime;
 
 use async_trait::async_trait;
 use criterion::{criterion_group, criterion_main, Criterion};
@@ -126,7 +126,7 @@ fn exporter_with_future(c: &mut Criterion) {
     c.bench_function("LogExporterWithFuture", |b| {
         b.iter(|| {
             let mut log_record = logger.create_log_record();
-            let now = SystemTime::now();
+            let now = now();
             log_record.set_observed_timestamp(now);
             log_record.set_target("my-target".to_string());
             log_record.set_event_name("CheckoutFailed");
@@ -152,7 +152,7 @@ fn exporter_without_future(c: &mut Criterion) {
     c.bench_function("LogExporterWithoutFuture", |b| {
         b.iter(|| {
             let mut log_record = logger.create_log_record();
-            let now = SystemTime::now();
+            let now = now();
             log_record.set_observed_timestamp(now);
             log_record.set_target("my-target".to_string());
             log_record.set_event_name("CheckoutFailed");

opentelemetry-sdk/benches/log_processor.rs

@@ -11,10 +11,10 @@
     | log_clone_and_send_to_channel_processor     | 403 ns      |
 */
 
+use opentelemetry::time::now;
 use std::{
     sync::{Arc, Mutex},
     thread::sleep,
-    time::SystemTime,
 };
 
 use criterion::{criterion_group, criterion_main, Criterion};
@@ -29,7 +29,7 @@ use opentelemetry_sdk::logs::{LogProcessor, LogRecord, LogResult, Logger, Logger
 
 fn create_log_record(logger: &Logger) -> LogRecord {
     let mut log_record = logger.create_log_record();
-    let now = SystemTime::now();
+    let now = now();
     log_record.set_observed_timestamp(now);
     log_record.set_target("my-target".to_string());
     log_record.set_event_name("CheckoutFailed");

opentelemetry-sdk/src/logs/log_processor.rs

@@ -228,19 +228,33 @@ type LogsData = Box<(LogRecord, InstrumentationScope)>;
 /// individually. It uses a **dedicated background thread** to manage and export logs
 /// asynchronously, ensuring that the application's main execution flow is not blocked.
 ///
-/// - This processor supports the following configurations:
-///     - **Queue size**: Maximum number of log records that can be buffered.
-///     - **Batch size**: Maximum number of log records to include in a single export.
-///     - **Scheduled delay**: Frequency at which the batch is exported.
+/// This processor supports the following configurations:
+/// - **Queue size**: Maximum number of log records that can be buffered.
+/// - **Batch size**: Maximum number of log records to include in a single export.
+/// - **Scheduled delay**: Frequency at which the batch is exported.
 ///
 /// When using this processor with the OTLP Exporter, the following exporter
 /// features are supported:
-/// - `grpc-tonic`: This requires `MeterProvider` to be created within a tokio
-///   runtime.
+/// - `grpc-tonic`: Requires `LoggerProvider` to be created within a tokio runtime.
 /// - `reqwest-blocking-client`: Works with a regular `main` or `tokio::main`.
 ///
 /// In other words, other clients like `reqwest` and `hyper` are not supported.
 ///
+/// `BatchLogProcessor` buffers logs in memory and exports them in batches. An
+/// export is triggered when `max_export_batch_size` is reached or every
+/// `scheduled_delay` milliseconds. Users can explicitly trigger an export using
+/// the `force_flush` method. Shutdown also triggers an export of all buffered
+/// logs and is recommended to be called before the application exits to ensure
+/// all buffered logs are exported.
+///
+/// **Warning**: When using tokio's current-thread runtime, `shutdown()`, which
+/// is a blocking call ,should not be called from your main thread. This can
+/// cause deadlock. Instead, call `shutdown()` from a separate thread or use
+/// tokio's `spawn_blocking`.
+///
+/// [`shutdown()`]: crate::logs::LoggerProvider::shutdown
+/// [`force_flush()`]: crate::logs::LoggerProvider::force_flush
+///
 /// ### Using a BatchLogProcessor:
 ///
 /// ```rust

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -91,25 +91,50 @@ where
     }
 }
 
-/// A [MetricReader] that continuously collects and exports metrics at a set
-/// interval.
+/// A `MetricReader` that periodically collects and exports metrics at a configurable interval.
 ///
-/// By default, `PeriodicReader` will collect and export metrics every 60
-/// seconds. The export time is not counted towards the interval between
-/// attempts. `PeriodicReader` itself does not enforce a timeout. Instead, the
-/// timeout is passed on to the configured exporter for each export attempt.
+/// By default, [`PeriodicReader`] collects and exports metrics every **60 seconds**.
+/// The time taken for export is **not** included in the interval. Use [`PeriodicReaderBuilder`]
+/// to customize the interval.
 ///
-/// `PeriodicReader` spawns a background thread to handle the periodic
-/// collection and export of metrics. The background thread will continue to run
-/// until `shutdown()` is called.
+/// [`PeriodicReader`] spawns a background thread to handle metric collection and export.
+/// This thread remains active until [`shutdown()`] is called.
 ///
-/// When using this reader with the OTLP Exporter, the following exporter
-/// features are supported:
-/// - `grpc-tonic`: This requires `MeterProvider` to be created within a tokio
-///   runtime.
-/// - `reqwest-blocking-client`: Works with a regular `main` or `tokio::main`.
+/// ## Collection Process
+/// "Collection" refers to gathering aggregated metrics from the SDK's internal storage.
+/// During this phase, callbacks from observable instruments are also triggered.
 ///
-/// In other words, other clients like `reqwest` and `hyper` are not supported.
+/// [`PeriodicReader`] does **not** enforce a timeout for collection. If an
+/// observable callback takes too long, it may delay the next collection cycle.
+/// If a callback never returns, it **will stall** all metric collection (and exports)
+/// indefinitely.
+///
+/// ## Exporter Compatibility
+/// When used with the [`OTLP Exporter`](https://docs.rs/opentelemetry-otlp), the following
+/// transport options are supported:
+///
+/// - **`grpc-tonic`**: Requires [`MeterProvider`] to be initialized within a `tokio` runtime.
+/// - **`reqwest-blocking-client`**: Works with both a standard (`main`) function and `tokio::main`.
+///
+/// [`PeriodicReader`] does **not** enforce a timeout for exports either. Instead,
+/// the configured exporter is responsible for enforcing timeouts. If an export operation
+/// never returns, [`PeriodicReader`] will **stop exporting new metrics**, stalling
+/// metric collection.
+///
+/// ## Manual Export & Shutdown
+/// Users can manually trigger an export via [`force_flush()`]. Calling [`shutdown()`]
+/// exports any remaining metrics and should be done before application exit to ensure
+/// all data is sent.
+///
+/// **Warning**: If using **tokio’s current-thread runtime**, calling [`shutdown()`]
+/// from the main thread may cause a deadlock. To prevent this, call [`shutdown()`]
+/// from a separate thread or use tokio's `spawn_blocking`.
+///
+/// [`PeriodicReader`]: crate::metrics::PeriodicReader
+/// [`PeriodicReaderBuilder`]: crate::metrics::PeriodicReaderBuilder
+/// [`MeterProvider`]: crate::metrics::SdkMeterProvider
+/// [`shutdown()`]: crate::metrics::SdkMeterProvider::shutdown
+/// [`force_flush()`]: crate::metrics::SdkMeterProvider::force_flush
 ///
 /// # Example
 ///

opentelemetry-sdk/src/trace/span_processor.rs

@@ -217,7 +217,6 @@ use crate::trace::ExportResult;
 ///                 .with_max_queue_size(1024) // Buffer up to 1024 spans.
 ///                 .with_max_export_batch_size(256) // Export in batches of up to 256 spans.
 ///                 .with_scheduled_delay(Duration::from_secs(5)) // Export every 5 seconds.
-///                 .with_max_export_timeout(Duration::from_secs(10)) // Timeout after 10 seconds.
 ///                 .build(),
 ///         )
 ///         .build();
@@ -253,7 +252,38 @@ enum BatchMessage {
     SetResource(Arc<Resource>),
 }
 
-/// A batch span processor with a dedicated background thread.
+/// The `BatchSpanProcessor` collects finished spans in a buffer and exports them
+/// in batches to the configured `SpanExporter`. This processor is ideal for
+/// high-throughput environments, as it minimizes the overhead of exporting spans
+/// individually. It uses a **dedicated background thread** to manage and export spans
+/// asynchronously, ensuring that the application's main execution flow is not blocked.
+///
+/// This processor supports the following configurations:
+/// - **Queue size**: Maximum number of spans that can be buffered.
+/// - **Batch size**: Maximum number of spans to include in a single export.
+/// - **Scheduled delay**: Frequency at which the batch is exported.
+///
+/// When using this processor with the OTLP Exporter, the following exporter
+/// features are supported:
+/// - `grpc-tonic`: Requires `TracerProvider` to be created within a tokio runtime.
+/// - `reqwest-blocking-client`: Works with a regular `main` or `tokio::main`.
+///
+/// In other words, other clients like `reqwest` and `hyper` are not supported.
+///
+/// `BatchSpanProcessor` buffers spans in memory and exports them in batches. An
+/// export is triggered when `max_export_batch_size` is reached or every
+/// `scheduled_delay` milliseconds. Users can explicitly trigger an export using
+/// the `force_flush` method. Shutdown also triggers an export of all buffered
+/// spans and is recommended to be called before the application exits to ensure
+/// all buffered spans are exported.
+///
+/// **Warning**: When using tokio's current-thread runtime, `shutdown()`, which
+/// is a blocking call ,should not be called from your main thread. This can
+/// cause deadlock. Instead, call `shutdown()` from a separate thread or use
+/// tokio's `spawn_blocking`.
+///
+/// [`shutdown()`]: crate::trace::TracerProvider::shutdown
+/// [`force_flush()`]: crate::trace::TracerProvider::force_flush
 #[derive(Debug)]
 pub struct BatchSpanProcessor {
     span_sender: SyncSender<SpanData>, // Data channel to store spans
@@ -443,20 +473,14 @@ impl BatchSpanProcessor {
         }
 
         let count_of_spans = spans.len(); // Count of spans that will be exported
-        let result = Self::export_with_timeout_sync(
-            config.max_export_timeout,
-            exporter,
-            spans,
-            last_export_time,
-        ); // This method clears the spans vec after exporting
+        let result = Self::export_batch_sync(exporter, spans, last_export_time); // This method clears the spans vec after exporting
 
         current_batch_size.fetch_sub(count_of_spans, Ordering::Relaxed);
         result
     }
 
     #[allow(clippy::vec_box)]
-    fn export_with_timeout_sync<E>(
-        _: Duration, // TODO, enforcing timeout in exporter.
+    fn export_batch_sync<E>(
         exporter: &mut E,
         batch: &mut Vec<SpanData>,
         last_export_time: &mut Instant,
@@ -740,6 +764,7 @@ impl BatchConfigBuilder {
     /// Set max_export_timeout for [`BatchConfigBuilder`].
     /// It's the maximum duration to export a batch of data.
     /// The The default value is 30000 milliseconds.
+    #[cfg(feature = "experimental_trace_batch_span_processor_with_async_runtime")]
     pub fn with_max_export_timeout(mut self, max_export_timeout: Duration) -> Self {
         self.max_export_timeout = max_export_timeout;
         self
@@ -960,10 +985,11 @@ mod tests {
         let batch = BatchConfigBuilder::default()
             .with_max_export_batch_size(10)
             .with_scheduled_delay(Duration::from_millis(10))
-            .with_max_export_timeout(Duration::from_millis(10))
             .with_max_queue_size(10);
         #[cfg(feature = "experimental_trace_batch_span_processor_with_async_runtime")]
         let batch = batch.with_max_concurrent_exports(10);
+        #[cfg(feature = "experimental_trace_batch_span_processor_with_async_runtime")]
+        let batch = batch.with_max_export_timeout(Duration::from_millis(10));
         let batch = batch.build();
         assert_eq!(batch.max_export_batch_size, 10);
         assert_eq!(batch.scheduled_delay, Duration::from_millis(10));
@@ -1037,7 +1063,6 @@ mod tests {
             .with_max_queue_size(10)
             .with_max_export_batch_size(10)
             .with_scheduled_delay(Duration::from_secs(5))
-            .with_max_export_timeout(Duration::from_secs(2))
             .build();
         let processor = BatchSpanProcessor::new(exporter, config);
 
@@ -1060,7 +1085,6 @@ mod tests {
             .with_max_queue_size(10)
             .with_max_export_batch_size(10)
             .with_scheduled_delay(Duration::from_secs(5))
-            .with_max_export_timeout(Duration::from_secs(2))
             .build();
         let processor = BatchSpanProcessor::new(exporter, config);
 
@@ -1090,7 +1114,6 @@ mod tests {
             .with_max_queue_size(10)
             .with_max_export_batch_size(10)
             .with_scheduled_delay(Duration::from_secs(5))
-            .with_max_export_timeout(Duration::from_secs(2))
             .build();
         let processor = BatchSpanProcessor::new(exporter, config);
 
@@ -1126,7 +1149,6 @@ mod tests {
         let config = BatchConfigBuilder::default()
             .with_max_queue_size(2) // Small queue size to test span dropping
             .with_scheduled_delay(Duration::from_secs(5))
-            .with_max_export_timeout(Duration::from_secs(2))
             .build();
         let processor = BatchSpanProcessor::new(exporter, config);