Make PeriodicReader with own Thread the default (#2403)

Author: cijothomas

examples/metrics-advanced/src/main.rs

@@ -4,7 +4,7 @@ use opentelemetry::KeyValue;
 use opentelemetry_sdk::metrics::{
     Aggregation, Instrument, PeriodicReader, SdkMeterProvider, Stream, Temporality,
 };
-use opentelemetry_sdk::{runtime, Resource};
+use opentelemetry_sdk::Resource;
 use std::error::Error;
 
 fn init_meter_provider() -> opentelemetry_sdk::metrics::SdkMeterProvider {
@@ -49,7 +49,7 @@ fn init_meter_provider() -> opentelemetry_sdk::metrics::SdkMeterProvider {
         .with_temporality(Temporality::Delta)
         .build();
 
-    let reader = PeriodicReader::builder(exporter, runtime::Tokio).build();
+    let reader = PeriodicReader::builder(exporter).build();
     let provider = SdkMeterProvider::builder()
         .with_reader(reader)
         .with_resource(Resource::new([KeyValue::new(

examples/metrics-basic/src/main.rs

@@ -1,6 +1,6 @@
 use opentelemetry::{global, KeyValue};
 use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
-use opentelemetry_sdk::{runtime, Resource};
+use opentelemetry_sdk::Resource;
 use std::error::Error;
 use std::vec;
 
@@ -9,7 +9,7 @@ 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, runtime::Tokio).build();
+    let reader = PeriodicReader::builder(exporter).build();
     let provider = SdkMeterProvider::builder()
         .with_reader(reader)
         .with_resource(Resource::new([KeyValue::new(

opentelemetry-otlp/examples/basic-otlp-http/Cargo.toml

@@ -9,12 +9,13 @@ publish = false
 default = ["reqwest"]
 reqwest = ["opentelemetry-otlp/reqwest-client"]
 hyper = ["opentelemetry-otlp/hyper-client"]
+experimental_metrics_periodicreader_with_async_runtime = ["opentelemetry_sdk/experimental_metrics_periodicreader_with_async_runtime"]
 
 
 [dependencies]
 once_cell = { workspace = true }
 opentelemetry = { path = "../../../opentelemetry" }
-opentelemetry_sdk = { path = "../../../opentelemetry-sdk", features = ["rt-tokio", "metrics", "logs"]}
+opentelemetry_sdk = { path = "../../../opentelemetry-sdk", features = ["rt-tokio", "metrics", "logs", "experimental_metrics_periodicreader_with_async_runtime"]}
 opentelemetry-http = { path = "../../../opentelemetry-http", optional = true, default-features = false}
 opentelemetry-otlp = { path = "../..", features = ["http-proto", "http-json", "logs"] , default-features = false}
 opentelemetry-appender-tracing = { path = "../../../opentelemetry-appender-tracing", default-features = false}

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

@@ -10,7 +10,7 @@ use opentelemetry_otlp::WithExportConfig;
 use opentelemetry_otlp::{LogExporter, MetricExporter, Protocol, SpanExporter};
 use opentelemetry_sdk::{
     logs::LoggerProvider,
-    metrics::{MetricError, PeriodicReader, SdkMeterProvider},
+    metrics::{MetricError, SdkMeterProvider},
     runtime,
     trace::{self as sdktrace, TracerProvider},
 };
@@ -63,8 +63,19 @@ fn init_metrics() -> Result<opentelemetry_sdk::metrics::SdkMeterProvider, Metric
         .with_endpoint("http://localhost:4318/v1/metrics")
         .build()?;
 
+    #[cfg(feature = "experimental_metrics_periodicreader_with_async_runtime")]
+    let reader =
+        opentelemetry_sdk::metrics::periodic_reader_with_async_runtime::PeriodicReader::builder(
+            exporter,
+            runtime::Tokio,
+        )
+        .build();
+    // TODO: This does not work today. See https://github.com/open-telemetry/opentelemetry-rust/issues/2400
+    #[cfg(not(feature = "experimental_metrics_periodicreader_with_async_runtime"))]
+    let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(exporter).build();
+
     Ok(SdkMeterProvider::builder()
-        .with_reader(PeriodicReader::builder(exporter, runtime::Tokio).build())
+        .with_reader(reader)
         .with_resource(RESOURCE.clone())
         .build())
 }

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

@@ -34,8 +34,7 @@ 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, runtime::Tokio).build();
+    let reader = PeriodicReader::builder(exporter).build();
 
     Ok(SdkMeterProvider::builder()
         .with_reader(reader)

opentelemetry-otlp/src/lib.rs

@@ -179,7 +179,7 @@
 //!        .build()
 //!        .unwrap();
 //!
-//!    let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(exporter, opentelemetry_sdk::runtime::Tokio)
+//!    let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(exporter)
 //!        .with_interval(std::time::Duration::from_secs(3))
 //!         .with_timeout(Duration::from_secs(10))
 //!        .build();

opentelemetry-sdk/CHANGELOG.md

@@ -28,6 +28,52 @@
      Migration Guidance: 
         - These methods were intended for log appenders. Keep the clone of the provider handle, instead of depending on above methods.
 
+- *Breaking* - `PeriodicReader` Updates
+
+   `PeriodicReader` no longer requires an async runtime by default. Instead, it
+   now creates its own background thread for execution. This change allows
+   metrics to be used in environments without async runtimes.
+
+   For users who prefer the previous behavior of relying on a specific
+   `Runtime`, they can do so by enabling the feature flag
+   **`experimental_metrics_periodicreader_with_async_runtime`**.
+
+   Migration Guide:
+
+ 1. *Default Implementation, requires no async runtime* (**Recommended**) The
+    new default implementation does not require a runtime argument. Replace the
+    builder method accordingly:
+    - *Before:* 
+      ```rust
+      let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(exporter, runtime::Tokio).build();
+      ```
+    - *After:*
+      ```rust
+      let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(exporter).build();
+      ```
+
+ 2. *Async Runtime Support*
+    If your application cannot spin up new threads or you prefer using async
+    runtimes, enable the
+    "experimental_metrics_periodicreader_with_async_runtime" feature flag and
+    adjust code as below.  
+
+    - *Before:*
+      ```rust
+      let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(exporter, runtime::Tokio).build();
+      ```
+
+    - *After:*
+      ```rust
+      let reader = opentelemetry_sdk::metrics::periodic_reader_with_async_runtime::PeriodicReader::builder(exporter, runtime::Tokio).build();
+      ```      
+
+    *Requirements:*
+    - Enable the feature flag:
+      `experimental_metrics_periodicreader_with_async_runtime`.  
+    - Continue enabling one of the async runtime feature flags: `rt-tokio`,
+      `rt-tokio-current-thread`, or `rt-async-std`.
+
 ## 0.27.1
 
 Released 2024-Nov-27

opentelemetry-sdk/Cargo.toml

@@ -54,6 +54,7 @@ rt-tokio-current-thread = ["tokio", "tokio-stream"]
 rt-async-std = ["async-std"]
 internal-logs = ["tracing"]
 experimental_metrics_periodic_reader_no_runtime = ["metrics"]
+experimental_metrics_periodicreader_with_async_runtime = ["metrics"]
 spec_unstable_metrics_views = ["metrics"]
 
 [[bench]]

opentelemetry-sdk/src/metrics/mod.rs

@@ -50,8 +50,9 @@ pub(crate) mod meter;
 mod meter_provider;
 pub(crate) mod noop;
 pub(crate) mod periodic_reader;
-#[cfg(feature = "experimental_metrics_periodic_reader_no_runtime")]
-pub(crate) mod periodic_reader_with_own_thread;
+#[cfg(feature = "experimental_metrics_periodicreader_with_async_runtime")]
+/// Module for periodic reader with async runtime.
+pub mod periodic_reader_with_async_runtime;
 pub(crate) mod pipeline;
 pub mod reader;
 pub(crate) mod view;
@@ -61,8 +62,6 @@ pub use error::{MetricError, MetricResult};
 pub use manual_reader::*;
 pub use meter_provider::*;
 pub use periodic_reader::*;
-#[cfg(feature = "experimental_metrics_periodic_reader_no_runtime")]
-pub use periodic_reader_with_own_thread::*;
 pub use pipeline::Pipeline;
 
 pub use instrument::InstrumentKind;
@@ -107,8 +106,8 @@ mod tests {
     use self::data::{HistogramDataPoint, ScopeMetrics, SumDataPoint};
     use super::*;
     use crate::metrics::data::ResourceMetrics;
+    use crate::testing::metrics::InMemoryMetricExporter;
     use crate::testing::metrics::InMemoryMetricExporterBuilder;
-    use crate::{runtime, testing::metrics::InMemoryMetricExporter};
     use data::GaugeDataPoint;
     use opentelemetry::metrics::{Counter, Meter, UpDownCounter};
     use opentelemetry::InstrumentationScope;
@@ -513,7 +512,7 @@ mod tests {
         }
 
         let exporter = InMemoryMetricExporter::default();
-        let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
+        let reader = PeriodicReader::builder(exporter.clone()).build();
         let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
 
         // Test Meter creation in 2 ways, both with empty string as meter name
@@ -529,7 +528,7 @@ mod tests {
     async fn counter_duplicate_instrument_merge() {
         // Arrange
         let exporter = InMemoryMetricExporter::default();
-        let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
+        let reader = PeriodicReader::builder(exporter.clone()).build();
         let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
 
         // Act
@@ -580,7 +579,7 @@ mod tests {
     async fn counter_duplicate_instrument_different_meter_no_merge() {
         // Arrange
         let exporter = InMemoryMetricExporter::default();
-        let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
+        let reader = PeriodicReader::builder(exporter.clone()).build();
         let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
 
         // Act
@@ -669,7 +668,7 @@ mod tests {
     async fn instrumentation_scope_identity_test() {
         // Arrange
         let exporter = InMemoryMetricExporter::default();
-        let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
+        let reader = PeriodicReader::builder(exporter.clone()).build();
         let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
 
         // Act
@@ -753,7 +752,7 @@ mod tests {
 
         // Arrange
         let exporter = InMemoryMetricExporter::default();
-        let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
+        let reader = PeriodicReader::builder(exporter.clone()).build();
         let criteria = Instrument::new().name("test_histogram");
         let stream_invalid_aggregation = Stream::new()
             .aggregation(Aggregation::ExplicitBucketHistogram {
@@ -803,7 +802,7 @@ mod tests {
 
         // Arrange
         let exporter = InMemoryMetricExporter::default();
-        let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
+        let reader = PeriodicReader::builder(exporter.clone()).build();
         let criteria = Instrument::new().name("my_observable_counter");
         // View drops all attributes.
         let stream_invalid_aggregation = Stream::new().allowed_attribute_keys(vec![]);
@@ -878,7 +877,7 @@ mod tests {
 
         // Arrange
         let exporter = InMemoryMetricExporter::default();
-        let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
+        let reader = PeriodicReader::builder(exporter.clone()).build();
         let criteria = Instrument::new().name("my_counter");
         // View drops all attributes.
         let stream_invalid_aggregation = Stream::new().allowed_attribute_keys(vec![]);
@@ -2433,7 +2432,7 @@ mod tests {
             let exporter = InMemoryMetricExporterBuilder::new().with_temporality(temporality);
 
             let exporter = exporter.build();
-            let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
+            let reader = PeriodicReader::builder(exporter.clone()).build();
             let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
 
             TestContext {