Merge branch 'main' into current-thread-int-test-logs

Author: lalitb

examples/metrics-advanced/src/main.rs

@@ -49,14 +49,12 @@ fn init_meter_provider() -> opentelemetry_sdk::metrics::SdkMeterProvider {
         .with_temporality(Temporality::Delta)
         .build();
 
-    let reader = PeriodicReader::builder(exporter).build();
-
     let resource = Resource::builder()
         .with_service_name("metrics-advanced-example")
         .build();
 
     let provider = SdkMeterProvider::builder()
-        .with_reader(reader)
+        .with_periodic_exporter(exporter)
         .with_resource(resource)
         .with_view(my_view_rename_and_unit)
         .with_view(my_view_drop_attributes)

examples/metrics-basic/src/main.rs

@@ -10,9 +10,8 @@ fn init_meter_provider() -> opentelemetry_sdk::metrics::SdkMeterProvider {
         // Build exporter using Delta Temporality (Defaults to Temporality::Cumulative)
         // .with_temporality(opentelemetry_sdk::metrics::Temporality::Delta)
         .build();
-    let reader = PeriodicReader::builder(exporter).build();
     let provider = SdkMeterProvider::builder()
-        .with_reader(reader)
+        .with_periodic_exporter(exporter)
         .with_resource(
             Resource::builder()
                 .with_service_name("metrics-basic-example")

opentelemetry-otlp/examples/basic-otlp-http/src/main.rs

@@ -61,10 +61,8 @@ fn init_metrics() -> Result<opentelemetry_sdk::metrics::SdkMeterProvider, Metric
         .with_endpoint("http://localhost:4318/v1/metrics")
         .build()?;
 
-    let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(exporter).build();
-
     Ok(SdkMeterProvider::builder()
-        .with_reader(reader)
+        .with_periodic_exporter(exporter)
         .with_resource(RESOURCE.clone())
         .build())
 }

opentelemetry-otlp/examples/basic-otlp/src/main.rs

@@ -33,10 +33,9 @@ fn init_traces() -> Result<sdktrace::TracerProvider, TraceError> {
 
 fn init_metrics() -> Result<opentelemetry_sdk::metrics::SdkMeterProvider, MetricError> {
     let exporter = MetricExporter::builder().with_tonic().build()?;
-    let reader = PeriodicReader::builder(exporter).build();
 
     Ok(SdkMeterProvider::builder()
-        .with_reader(reader)
+        .with_periodic_exporter(exporter)
         .with_resource(RESOURCE.clone())
         .build())
 }

opentelemetry-sdk/CHANGELOG.md

@@ -405,14 +405,33 @@ Released 2024-Nov-27
      Migration Guidance:
         - These methods are intended for log appenders. Keep the clone of the provider handle, instead of depending on above methods.
 
-
   - **Bug Fix:** Validates the `with_boundaries` bucket boundaries used in
     Histograms. The boundaries provided by the user must not contain `f64::NAN`,
     `f64::INFINITY` or `f64::NEG_INFINITY` and must be sorted in strictly
     increasing order, and contain no duplicates. Instruments will not record
     measurements if the boundaries are invalid.
     [#2351](https://github.com/open-telemetry/opentelemetry-rust/pull/2351)
 
+- Added `with_periodic_exporter` method to `MeterProviderBuilder`, allowing
+  users to easily attach an exporter with a PeriodicReader for automatic metric
+  export. Retained with_reader() for advanced use cases where a custom
+  MetricReader configuration is needed.
+  [2597](https://github.com/open-telemetry/opentelemetry-rust/pull/2597)
+  Example Usage:
+
+  ```rust
+  SdkMeterProvider::builder()
+      .with_periodic_exporter(exporter)
+      .build();
+  ```
+
+  Using a custom PeriodicReader (advanced use case):
+
+  let reader = PeriodicReader::builder(exporter).build();
+  SdkMeterProvider::builder()
+      .with_reader(reader)
+      .build();
+
 ## 0.27.0
 
 Released 2024-Nov-11

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/meter_provider.rs

@@ -19,7 +19,8 @@ use crate::{
 };
 
 use super::{
-    meter::SdkMeter, noop::NoopMeter, pipeline::Pipelines, reader::MetricReader, view::View,
+    exporter::PushMetricExporter, meter::SdkMeter, noop::NoopMeter, pipeline::Pipelines,
+    reader::MetricReader, view::View, PeriodicReader,
 };
 
 /// Handles the creation and coordination of [Meter]s.
@@ -244,14 +245,37 @@ impl MeterProviderBuilder {
     }
 
     /// Associates a [MetricReader] with a [MeterProvider].
+    /// [`MeterProviderBuilder::with_periodic_exporter()] can be used to add a PeriodicReader which is
+    /// the most common use case.
     ///
-    /// By default, if this option is not used, the [MeterProvider] will perform no
-    /// operations; no data will be exported without a [MetricReader].
+    /// A [MeterProvider] will export no metrics without [MetricReader]
+    /// added.
     pub fn with_reader<T: MetricReader>(mut self, reader: T) -> Self {
         self.readers.push(Box::new(reader));
         self
     }
 
+    /// Adds a [`PushMetricExporter`] to the [`MeterProvider`] and configures it
+    /// to export metrics at **fixed** intervals (60 seconds) using a
+    /// [`PeriodicReader`].
+    ///
+    /// To customize the export interval, set the
+    /// **"OTEL_METRIC_EXPORT_INTERVAL"** environment variable (in
+    /// milliseconds).
+    ///
+    /// Most users should use this method to attach an exporter. Advanced users
+    /// who need finer control over the export process can use
+    /// [`crate::metrics::PeriodicReaderBuilder`] to configure a custom reader and attach it
+    /// using [`MeterProviderBuilder::with_reader()`].
+    pub fn with_periodic_exporter<T>(mut self, exporter: T) -> Self
+    where
+        T: PushMetricExporter,
+    {
+        let reader = PeriodicReader::builder(exporter).build();
+        self.readers.push(Box::new(reader));
+        self
+    }
+
     #[cfg(feature = "spec_unstable_metrics_views")]
     /// Associates a [View] with a [MeterProvider].
     ///

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
 ///
@@ -175,10 +200,36 @@ impl PeriodicReader {
                             otel_debug!(
                                 name: "PeriodReaderThreadExportingDueToFlush"
                             );
-                            if let Err(_e) = cloned_reader.collect_and_export(timeout) {
-                                response_sender.send(false).unwrap();
-                            } else {
-                                response_sender.send(true).unwrap();
+
+                            let export_result = cloned_reader.collect_and_export(timeout);
+                            otel_debug!(
+                                name: "PeriodReaderInvokedExport",
+                                export_result = format!("{:?}", export_result)
+                            );
+
+                            // If response_sender is disconnected, we can't send
+                            // the result back. This occurs when the thread that
+                            // initiated flush gave up due to timeout.
+                            // Gracefully handle that with internal logs. The
+                            // internal errors are of Info level, as this is
+                            // useful for user to know whether the flush was
+                            // successful or not, when flush() itself merely
+                            // tells that it timed out.
+
+                            if export_result.is_err() {
+                                if response_sender.send(false).is_err() {
+                                    otel_info!(
+                                        name: "PeriodReader.Flush.ResponseSendError",
+                                        message = "PeriodicReader's flush has failed, but unable to send this info back to caller. 
+                                        This occurs when the caller has timed out waiting for the response. If you see this occuring frequently, consider increasing the flush timeout."
+                                    );
+                                }
+                            } else if response_sender.send(true).is_err() {
+                                otel_info!(
+                                    name: "PeriodReader.Flush.ResponseSendError",
+                                    message = "PeriodicReader's flush has completed successfully, but unable to send this info back to caller. 
+                                    This occurs when the caller has timed out waiting for the response. If you see this occuring frequently, consider increasing the flush timeout."
+                                );
                             }
 
                             // Adjust the remaining interval after the flush
@@ -207,15 +258,39 @@ impl PeriodicReader {
                             // Perform final export and break out of loop and exit the thread
                             otel_debug!(name: "PeriodReaderThreadExportingDueToShutdown");
                             let export_result = cloned_reader.collect_and_export(timeout);
+                            otel_debug!(
+                                name: "PeriodReaderInvokedExport",
+                                export_result = format!("{:?}", export_result)
+                            );
                             let shutdown_result = exporter_arc.shutdown();
                             otel_debug!(
                                 name: "PeriodReaderInvokedExporterShutdown",
                                 shutdown_result = format!("{:?}", shutdown_result)
                             );
+
+                            // If response_sender is disconnected, we can't send
+                            // the result back. This occurs when the thread that
+                            // initiated shutdown gave up due to timeout.
+                            // Gracefully handle that with internal logs and
+                            // continue with shutdown (i.e exit thread) The
+                            // internal errors are of Info level, as this is
+                            // useful for user to know whether the shutdown was
+                            // successful or not, when shutdown() itself merely
+                            // tells that it timed out.
                             if export_result.is_err() || shutdown_result.is_err() {
-                                response_sender.send(false).unwrap();
-                            } else {
-                                response_sender.send(true).unwrap();
+                                if response_sender.send(false).is_err() {
+                                    otel_info!(
+                                        name: "PeriodReaderThreadShutdown.ResponseSendError",
+                                        message = "PeriodicReader's shutdown has failed, but unable to send this info back to caller. 
+                                        This occurs when the caller has timed out waiting for the response. If you see this occuring frequently, consider increasing the shutdown timeout."
+                                    );
+                                }
+                            } else if response_sender.send(true).is_err() {
+                                otel_info!(
+                                    name: "PeriodReaderThreadShutdown.ResponseSendError",
+                                    message = "PeriodicReader completed its shutdown, but unable to send this info back to caller. 
+                                    This occurs when the caller has timed out waiting for the response. If you see this occuring frequently, consider increasing the shutdown timeout."
+                                );
                             }
 
                             otel_debug!(
@@ -230,11 +305,11 @@ impl PeriodicReader {
                                 name: "PeriodReaderThreadExportingDueToTimer"
                             );
 
-                            if let Err(_e) = cloned_reader.collect_and_export(timeout) {
-                                otel_debug!(
-                                    name: "PeriodReaderThreadExportingDueToTimerFailed"
-                                );
-                            }
+                            let export_result = cloned_reader.collect_and_export(timeout);
+                            otel_debug!(
+                                name: "PeriodReaderInvokedExport",
+                                export_result = format!("{:?}", export_result)
+                            );
 
                             let time_taken_for_export = export_start.elapsed();
                             if time_taken_for_export > interval {
@@ -365,17 +440,7 @@ impl PeriodicReaderInner {
 
         // Relying on futures executor to execute async call.
         // TODO: Pass timeout to exporter
-        let exporter_result = futures_executor::block_on(self.exporter.export(&mut rm));
-        #[allow(clippy::question_mark)]
-        if let Err(e) = exporter_result {
-            otel_warn!(
-                name: "PeriodReaderExportError",
-                error = format!("{:?}", e)
-            );
-            return Err(e);
-        }
-
-        Ok(())
+        futures_executor::block_on(self.exporter.export(&mut rm))
     }
 
     fn force_flush(&self) -> MetricResult<()> {

opentelemetry-sdk/src/trace/provider.rs

@@ -134,7 +134,8 @@ impl Drop for TracerProviderInner {
             let _ = self.shutdown(); // errors are handled within shutdown
         } else {
             otel_debug!(
-                name: "TracerProvider.Drop.AlreadyShutdown"
+                name: "TracerProvider.Drop.AlreadyShutdown",
+                message = "TracerProvider was already shut down; drop will not attempt shutdown again."
             );
         }
     }