Merge branch 'main' into cijothomas/fixperiodicreaderpanic

Author: cijothomas

opentelemetry-otlp/CHANGELOG.md

@@ -13,6 +13,15 @@
 - Remove unnecessarily public trait `opentelemetry_otlp::metrics::MetricsClient`
   and `MetricExporter::new(..)` method. Use
   `MetricExporter::builder()...build()` to obtain `MetricExporter`.
+- The HTTP clients (reqwest, reqwest-blocking, hyper) now support the
+timeout internal configured in below order
+  - Signal specific env variable `OTEL_EXPORTER_OTLP_TRACES_TIMEOUT`, 
+  `OTEL_EXPORTER_OTLP_LOGS_TIMEOUT` or `OTEL_EXPORTER_OTLP_TIMEOUT`.
+  - `OTEL_EXPORTER_OTLP_TIMEOUT` env variable.
+  - `with_http().with_timeout()` API method of
+`LogExporterBuilder` and `SpanExporterBuilder` and `MetricsExporterBuilder`.
+  - The default interval of 10sec is used if none is configured.
+
 
 ## 0.27.0
 

opentelemetry-otlp/src/exporter/http/mod.rs

@@ -44,15 +44,7 @@ mod trace;
 use opentelemetry_http::hyper::HyperClient;
 
 /// Configuration of the http transport
-#[derive(Debug)]
-#[cfg_attr(
-    all(
-        not(feature = "reqwest-client"),
-        not(feature = "reqwest-blocking-client"),
-        not(feature = "hyper-client")
-    ),
-    derive(Default)
-)]
+#[derive(Debug, Default)]
 pub struct HttpConfig {
     /// Select the HTTP client
     client: Option<Arc<dyn HttpClient>>,
@@ -61,44 +53,6 @@ pub struct HttpConfig {
     headers: Option<HashMap<String, String>>,
 }
 
-#[cfg(any(
-    feature = "reqwest-blocking-client",
-    feature = "reqwest-client",
-    feature = "hyper-client"
-))]
-impl Default for HttpConfig {
-    fn default() -> Self {
-        #[cfg(feature = "reqwest-blocking-client")]
-        let default_client = std::thread::spawn(|| {
-            Some(Arc::new(reqwest::blocking::Client::new()) as Arc<dyn HttpClient>)
-        })
-        .join()
-        .expect("creating reqwest::blocking::Client on a new thread not to fail");
-        #[cfg(all(not(feature = "reqwest-blocking-client"), feature = "reqwest-client"))]
-        let default_client = Some(Arc::new(reqwest::Client::new()) as Arc<dyn HttpClient>);
-        #[cfg(all(
-            not(feature = "reqwest-client"),
-            not(feature = "reqwest-blocking-client"),
-            feature = "hyper-client"
-        ))]
-        // TODO - support configuring custom connector and executor
-        let default_client = Some(Arc::new(HyperClient::with_default_connector(
-            Duration::from_secs(10),
-            None,
-        )) as Arc<dyn HttpClient>);
-        #[cfg(all(
-            not(feature = "reqwest-client"),
-            not(feature = "reqwest-blocking-client"),
-            not(feature = "hyper-client")
-        ))]
-        let default_client = None;
-        HttpConfig {
-            client: default_client,
-            headers: None,
-        }
-    }
-}
-
 /// Configuration for the OTLP HTTP exporter.
 ///
 /// ## Examples
@@ -171,11 +125,56 @@ impl HttpExporterBuilder {
             },
             None => self.exporter_config.timeout,
         };
-        let http_client = self
-            .http_config
-            .client
-            .take()
-            .ok_or(crate::Error::NoHttpClient)?;
+
+        #[allow(unused_mut)] // TODO - clippy thinks mut is not needed, but it is
+        let mut http_client = self.http_config.client.take();
+
+        if http_client.is_none() {
+            #[cfg(all(
+                not(feature = "reqwest-client"),
+                not(feature = "reqwest-blocking-client"),
+                feature = "hyper-client"
+            ))]
+            {
+                // TODO - support configuring custom connector and executor
+                http_client = Some(Arc::new(HyperClient::with_default_connector(timeout, None))
+                    as Arc<dyn HttpClient>);
+            }
+            #[cfg(all(
+                not(feature = "hyper-client"),
+                not(feature = "reqwest-blocking-client"),
+                feature = "reqwest-client"
+            ))]
+            {
+                http_client = Some(Arc::new(
+                    reqwest::Client::builder()
+                        .timeout(timeout)
+                        .build()
+                        .unwrap_or_default(),
+                ) as Arc<dyn HttpClient>);
+            }
+            #[cfg(all(
+                not(feature = "hyper-client"),
+                not(feature = "reqwest-client"),
+                feature = "reqwest-blocking-client"
+            ))]
+            {
+                let timeout_clone = timeout;
+                http_client = Some(Arc::new(
+                    std::thread::spawn(move || {
+                        reqwest::blocking::Client::builder()
+                            .timeout(timeout_clone)
+                            .build()
+                            .unwrap_or_else(|_| reqwest::blocking::Client::new())
+                    })
+                    .join()
+                    .unwrap(), // Unwrap thread result
+                ) as Arc<dyn HttpClient>);
+            }
+        }
+
+        let http_client = http_client.ok_or(crate::Error::NoHttpClient)?;
+
         #[allow(clippy::mutable_key_type)] // http headers are not mutated
         let mut headers: HashMap<HeaderName, HeaderValue> = self
             .http_config

opentelemetry-sdk/CHANGELOG.md

@@ -304,6 +304,39 @@ limit.
   `opentelemetry_sdk::trace::{InMemorySpanExporter, InMemorySpanExporterBuilder};`
   `opentelemetry_sdk::metrics::{InMemoryMetricExporter, InMemoryMetricExporterBuilder};`
 
+- *Breaking*: The `BatchLogProcessor` 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 Logs exporter, the export timeout can be configured using:
+- The environment variables `OTEL_EXPORTER_OTLP_TIMEOUT` or `OTEL_EXPORTER_OTLP_LOGS_TIMEOUT`.
+- The opentelemetry_otlp API, via `.with_tonic().with_timeout()` or `.with_http().with_timeout()`.
+Before:
+```rust
+let processor = BatchLogProcessor::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 = BatchLogProcessor::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/src/logs/log_processor.rs

@@ -57,8 +57,10 @@ pub(crate) const OTEL_BLRP_SCHEDULE_DELAY: &str = "OTEL_BLRP_SCHEDULE_DELAY";
 /// Default delay interval between two consecutive exports.
 pub(crate) const OTEL_BLRP_SCHEDULE_DELAY_DEFAULT: u64 = 1_000;
 /// Maximum allowed time to export data.
+#[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
 pub(crate) const OTEL_BLRP_EXPORT_TIMEOUT: &str = "OTEL_BLRP_EXPORT_TIMEOUT";
 /// Default maximum allowed time to export data.
+#[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
 pub(crate) const OTEL_BLRP_EXPORT_TIMEOUT_DEFAULT: u64 = 30_000;
 /// Maximum queue size.
 pub(crate) const OTEL_BLRP_MAX_QUEUE_SIZE: &str = "OTEL_BLRP_MAX_QUEUE_SIZE";
@@ -229,7 +231,6 @@ type LogsData = Box<(LogRecord, InstrumentationScope)>;
 /// - 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.
-///     - **Export timeout**: Maximum duration allowed for an export operation.
 ///     - **Scheduled delay**: Frequency at which the batch is exported.
 ///
 /// When using this processor with the OTLP Exporter, the following exporter
@@ -255,7 +256,6 @@ type LogsData = Box<(LogRecord, InstrumentationScope)>;
 ///             .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))
 ///             .build(),
 ///     )
 ///     .build();
@@ -525,12 +525,7 @@ impl BatchLogProcessor {
                         let count_of_logs = logs.len(); // Count of logs that will be exported
                         total_exported_logs += count_of_logs;
 
-                        result = export_with_timeout_sync(
-                            config.max_export_timeout,
-                            exporter,
-                            logs,
-                            last_export_time,
-                        ); // This method clears the logs vec after exporting
+                        result = export_batch_sync(exporter, logs, last_export_time); // This method clears the logs vec after exporting
 
                         current_batch_size.fetch_sub(count_of_logs, Ordering::Relaxed);
                     }
@@ -656,8 +651,7 @@ impl BatchLogProcessor {
 }
 
 #[allow(clippy::vec_box)]
-fn export_with_timeout_sync<E>(
-    _: Duration, // TODO, enforcing timeout in exporter.
+fn export_batch_sync<E>(
     exporter: &E,
     batch: &mut Vec<Box<(LogRecord, InstrumentationScope)>>,
     last_export_time: &mut Instant,
@@ -733,6 +727,7 @@ pub struct BatchConfig {
     pub(crate) max_export_batch_size: usize,
 
     /// The maximum duration to export a batch of data.
+    #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
     pub(crate) max_export_timeout: Duration,
 }
 
@@ -748,6 +743,7 @@ pub struct BatchConfigBuilder {
     max_queue_size: usize,
     scheduled_delay: Duration,
     max_export_batch_size: usize,
+    #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
     max_export_timeout: Duration,
 }
 
@@ -764,6 +760,7 @@ impl Default for BatchConfigBuilder {
             max_queue_size: OTEL_BLRP_MAX_QUEUE_SIZE_DEFAULT,
             scheduled_delay: Duration::from_millis(OTEL_BLRP_SCHEDULE_DELAY_DEFAULT),
             max_export_batch_size: OTEL_BLRP_MAX_EXPORT_BATCH_SIZE_DEFAULT,
+            #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
             max_export_timeout: Duration::from_millis(OTEL_BLRP_EXPORT_TIMEOUT_DEFAULT),
         }
         .init_from_env_vars()
@@ -791,6 +788,7 @@ impl BatchConfigBuilder {
     /// Set max_export_timeout for [`BatchConfigBuilder`].
     /// It's the maximum duration to export a batch of data.
     /// The default value is 30000 milliseconds.
+    #[cfg(feature = "experimental_logs_batch_log_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
@@ -816,6 +814,7 @@ impl BatchConfigBuilder {
         BatchConfig {
             max_queue_size: self.max_queue_size,
             scheduled_delay: self.scheduled_delay,
+            #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
             max_export_timeout: self.max_export_timeout,
             max_export_batch_size,
         }
@@ -843,6 +842,7 @@ impl BatchConfigBuilder {
             self.scheduled_delay = Duration::from_millis(scheduled_delay);
         }
 
+        #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
         if let Some(max_export_timeout) = env::var(OTEL_BLRP_EXPORT_TIMEOUT)
             .ok()
             .and_then(|s| u64::from_str(&s).ok())
@@ -946,6 +946,7 @@ mod tests {
             config.scheduled_delay,
             Duration::from_millis(OTEL_BLRP_SCHEDULE_DELAY_DEFAULT)
         );
+        #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
         assert_eq!(
             config.max_export_timeout,
             Duration::from_millis(OTEL_BLRP_EXPORT_TIMEOUT_DEFAULT)
@@ -969,6 +970,7 @@ mod tests {
         let config = temp_env::with_vars(env_vars, BatchConfig::default);
 
         assert_eq!(config.scheduled_delay, Duration::from_millis(2000));
+        #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
         assert_eq!(config.max_export_timeout, Duration::from_millis(60000));
         assert_eq!(config.max_queue_size, 4096);
         assert_eq!(config.max_export_batch_size, 1024);
@@ -989,6 +991,7 @@ mod tests {
             config.scheduled_delay,
             Duration::from_millis(OTEL_BLRP_SCHEDULE_DELAY_DEFAULT)
         );
+        #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
         assert_eq!(
             config.max_export_timeout,
             Duration::from_millis(OTEL_BLRP_EXPORT_TIMEOUT_DEFAULT)
@@ -997,15 +1000,18 @@ mod tests {
 
     #[test]
     fn test_batch_config_with_fields() {
-        let batch = BatchConfigBuilder::default()
+        let batch_builder = BatchConfigBuilder::default()
             .with_max_export_batch_size(1)
             .with_scheduled_delay(Duration::from_millis(2))
-            .with_max_export_timeout(Duration::from_millis(3))
-            .with_max_queue_size(4)
-            .build();
+            .with_max_queue_size(4);
+
+        #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
+        let batch_builder = batch_builder.with_max_export_timeout(Duration::from_millis(3));
+        let batch = batch_builder.build();
 
         assert_eq!(batch.max_export_batch_size, 1);
         assert_eq!(batch.scheduled_delay, Duration::from_millis(2));
+        #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
         assert_eq!(batch.max_export_timeout, Duration::from_millis(3));
         assert_eq!(batch.max_queue_size, 4);
     }
@@ -1029,6 +1035,8 @@ mod tests {
                 builder.config.max_queue_size,
                 OTEL_BLRP_MAX_QUEUE_SIZE_DEFAULT
             );
+
+            #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
             assert_eq!(
                 builder.config.max_export_timeout,
                 Duration::from_millis(2046)
@@ -1049,7 +1057,6 @@ mod tests {
         let expected = BatchConfigBuilder::default()
             .with_max_export_batch_size(1)
             .with_scheduled_delay(Duration::from_millis(2))
-            .with_max_export_timeout(Duration::from_millis(3))
             .with_max_queue_size(4)
             .build();
 
@@ -1059,7 +1066,6 @@ mod tests {
         let actual = &builder.config;
         assert_eq!(actual.max_export_batch_size, 1);
         assert_eq!(actual.scheduled_delay, Duration::from_millis(2));
-        assert_eq!(actual.max_export_timeout, Duration::from_millis(3));
         assert_eq!(actual.max_queue_size, 4);
     }