Make PeriodicReader with own Thread the default

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

@@ -14,7 +14,7 @@ hyper = ["opentelemetry-otlp/hyper-client"]
 [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/src/main.rs

@@ -35,7 +35,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-sdk/CHANGELOG.md

@@ -28,6 +28,45 @@
      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 thread or you prefer using async
+    runtimes, enable the "experimental_metrics_periodicreader_with_async_runtime" feature flag.  
+
+    - *Code requires no change*
+      ```rust
+      let reader = opentelemetry_sdk::metrics::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

@@ -53,6 +53,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

@@ -49,9 +49,10 @@ pub(crate) mod manual_reader;
 pub(crate) mod meter;
 mod meter_provider;
 pub(crate) mod noop;
+#[cfg(not(feature = "experimental_metrics_periodicreader_with_async_runtime"))]
 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")]
+pub(crate) mod periodic_reader_with_async_runtime;
 pub(crate) mod pipeline;
 pub mod reader;
 pub(crate) mod view;
@@ -60,9 +61,10 @@ pub use aggregation::*;
 pub use error::{MetricError, MetricResult};
 pub use manual_reader::*;
 pub use meter_provider::*;
+#[cfg(not(feature = "experimental_metrics_periodicreader_with_async_runtime"))]
 pub use periodic_reader::*;
-#[cfg(feature = "experimental_metrics_periodic_reader_no_runtime")]
-pub use periodic_reader_with_own_thread::*;
+#[cfg(feature = "experimental_metrics_periodicreader_with_async_runtime")]
+pub use periodic_reader_with_async_runtime::*;
 pub use pipeline::Pipeline;
 
 pub use instrument::InstrumentKind;
@@ -513,7 +515,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 +531,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 +582,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 +671,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 +755,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 +805,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 +880,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![]);