Use dedicated thread for Metrics PeriodicReader

examples/metrics-advanced/src/main.rs

@@ -4,7 +4,7 @@ use opentelemetry::KeyValue;
 use opentelemetry_sdk::metrics::{
     Aggregation, Instrument, PeriodicReader, SdkMeterProvider, Stream,
 };
-use opentelemetry_sdk::{runtime, Resource};
+use opentelemetry_sdk::Resource;
 use std::error::Error;
 
 fn init_meter_provider() -> opentelemetry_sdk::metrics::SdkMeterProvider {
@@ -45,7 +45,7 @@ fn init_meter_provider() -> opentelemetry_sdk::metrics::SdkMeterProvider {
     };
 
     let exporter = opentelemetry_stdout::MetricsExporterBuilder::default().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/Cargo.toml

@@ -11,6 +11,7 @@ opentelemetry_sdk = { path = "../../opentelemetry-sdk", features = ["metrics", "
 opentelemetry-stdout = { path = "../../opentelemetry-stdout", features = ["metrics"]}
 tokio = { workspace = true, features = ["full"] }
 serde_json = { workspace = true }
+tracing-subscriber = { version = "0.3.18", features = ["env-filter","registry", "std"]}
 
 [features]
 default = ["otel_unstable"]

examples/metrics-basic/src/main.rs

@@ -1,13 +1,18 @@
 use opentelemetry::global;
 use opentelemetry::KeyValue;
 use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
-use opentelemetry_sdk::{runtime, Resource};
+use opentelemetry_sdk::Resource;
 use std::error::Error;
 use std::vec;
+use tracing_subscriber::filter::LevelFilter;
+use tracing_subscriber::fmt;
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
+use tracing_subscriber::Layer;
 
 fn init_meter_provider() -> opentelemetry_sdk::metrics::SdkMeterProvider {
     let exporter = opentelemetry_stdout::MetricsExporterBuilder::default().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(
@@ -21,6 +26,14 @@ fn init_meter_provider() -> opentelemetry_sdk::metrics::SdkMeterProvider {
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
+    tracing_subscriber::registry()
+        .with(
+            fmt::layer()
+                .with_thread_names(true)
+                .with_filter(LevelFilter::DEBUG),
+        )
+        .init();
+
     // Initialize the MeterProvider with the stdout Exporter.
     let meter_provider = init_meter_provider();
 

examples/self-diagnostics/src/main.rs

@@ -108,7 +108,7 @@ fn init_logger_provider() -> opentelemetry_sdk::logs::LoggerProvider {
 
 fn init_meter_provider() -> opentelemetry_sdk::metrics::SdkMeterProvider {
     let provider = opentelemetry_otlp::new_pipeline()
-        .metrics(opentelemetry_sdk::runtime::Tokio)
+        .metrics()
         .with_period(std::time::Duration::from_secs(1))
         .with_exporter(
             opentelemetry_otlp::new_exporter()

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

@@ -61,7 +61,7 @@ fn init_tracer_provider() -> Result<sdktrace::TracerProvider, TraceError> {
 
 fn init_metrics() -> Result<opentelemetry_sdk::metrics::SdkMeterProvider, MetricsError> {
     opentelemetry_otlp::new_pipeline()
-        .metrics(opentelemetry_sdk::runtime::Tokio)
+        .metrics()
         .with_exporter(
             http_exporter()
                 .with_protocol(Protocol::HttpBinary) //can be changed to `Protocol::HttpJson` to export in JSON format

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

@@ -41,7 +41,7 @@ fn init_metrics() -> Result<opentelemetry_sdk::metrics::SdkMeterProvider, Metric
         ..ExportConfig::default()
     };
     opentelemetry_otlp::new_pipeline()
-        .metrics(runtime::Tokio)
+        .metrics()
         .with_exporter(
             opentelemetry_otlp::new_exporter()
                 .tonic()

opentelemetry-otlp/src/lib.rs

@@ -174,7 +174,7 @@
 //!     };
 //!
 //!     let meter = opentelemetry_otlp::new_pipeline()
-//!         .metrics(opentelemetry_sdk::runtime::Tokio)
+//!         .metrics()
 //!         .with_exporter(
 //!             opentelemetry_otlp::new_exporter()
 //!                 .tonic()

opentelemetry-otlp/src/metric.rs

@@ -17,11 +17,13 @@
         reader::{DefaultTemporalitySelector, TemporalitySelector},
         InstrumentKind, PeriodicReader, SdkMeterProvider,
     },
-    runtime::Runtime,
     Resource,
 };
-use std::fmt::{Debug, Formatter};
 use std::time;
+use std::{
+    fmt::{Debug, Formatter},
+    time::Duration,
+};
 
 #[cfg(feature = "http-proto")]
 use crate::exporter::http::HttpExporterBuilder;
@@ -41,12 +43,8 @@
 pub const OTEL_EXPORTER_OTLP_METRICS_HEADERS: &str = "OTEL_EXPORTER_OTLP_METRICS_HEADERS";
 impl OtlpPipeline {
     /// Create a OTLP metrics pipeline.
-    pub fn metrics<RT>(self, rt: RT) -> OtlpMetricPipeline<RT, NoExporterConfig>
-    where
-        RT: Runtime,
-    {
+    pub fn metrics(self) -> OtlpMetricPipeline<NoExporterConfig> {
         OtlpMetricPipeline {
-            rt,
             temporality_selector: None,
             exporter_pipeline: NoExporterConfig(()),
             resource: None,
@@ -117,19 +115,15 @@
 ///
 /// Note that currently the OTLP metrics exporter only supports tonic as it's grpc layer and tokio as
 /// runtime.
-pub struct OtlpMetricPipeline<RT, EB> {
-    rt: RT,
+pub struct OtlpMetricPipeline<EB> {
     temporality_selector: Option<Box<dyn TemporalitySelector>>,
     exporter_pipeline: EB,
     resource: Option<Resource>,
     period: Option<time::Duration>,
     timeout: Option<time::Duration>,
 }
 
-impl<RT, EB> OtlpMetricPipeline<RT, EB>
-where
-    RT: Runtime,
-{
+impl<EB> OtlpMetricPipeline<EB> {
     /// Build with resource key value pairs.
     pub fn with_resource(self, resource: Resource) -> Self {
         OtlpMetricPipeline {
@@ -173,18 +167,14 @@
     }
 }
 
-impl<RT> OtlpMetricPipeline<RT, NoExporterConfig>
-where
-    RT: Runtime,
-{
+impl OtlpMetricPipeline<NoExporterConfig> {
     /// Build with the exporter
     pub fn with_exporter<B: Into<MetricsExporterBuilder>>(
         self,
         pipeline: B,
-    ) -> OtlpMetricPipeline<RT, MetricsExporterBuilder> {
+    ) -> OtlpMetricPipeline<MetricsExporterBuilder> {
         OtlpMetricPipeline {
             exporter_pipeline: pipeline.into(),
-            rt: self.rt,
             temporality_selector: self.temporality_selector,
             resource: self.resource,
             period: self.period,
@@ -193,18 +183,15 @@
     }
 }
 
-impl<RT> OtlpMetricPipeline<RT, MetricsExporterBuilder>
-where
-    RT: Runtime,
-{
+impl OtlpMetricPipeline<MetricsExporterBuilder> {
     /// Build MeterProvider
     pub fn build(self) -> Result<SdkMeterProvider> {
         let exporter = self.exporter_pipeline.build_metrics_exporter(
             self.temporality_selector
                 .unwrap_or_else(|| Box::new(DefaultTemporalitySelector::new())),
         )?;
 
-        let mut builder = PeriodicReader::builder(exporter, self.rt);
+        let mut builder = PeriodicReader::builder(exporter);
 
         if let Some(period) = self.period {
             builder = builder.with_interval(period);
@@ -226,7 +213,7 @@
     }
 }
 
-impl<RT, EB: Debug> Debug for OtlpMetricPipeline<RT, EB> {
+impl<EB: Debug> Debug for OtlpMetricPipeline<EB> {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         f.debug_struct("OtlpMetricPipeline")
             .field("exporter_pipeline", &self.exporter_pipeline)
@@ -293,7 +280,8 @@
 
 #[async_trait]
 impl PushMetricsExporter for MetricsExporter {
-    async fn export(&self, metrics: &mut ResourceMetrics) -> Result<()> {
+    async fn export(&self, metrics: &mut ResourceMetrics, _timeout: Duration) -> Result<()> {
+        //TODO: Pass timeout to client
         self.client.export(metrics).await
     }
 

opentelemetry-sdk/src/metrics/exporter.rs

@@ -1,4 +1,6 @@
 //! Interfaces for exporting metrics
+use std::time::Duration;
+
 use async_trait::async_trait;
 
 use opentelemetry::metrics::Result;
@@ -16,7 +18,7 @@ pub trait PushMetricsExporter: TemporalitySelector + Send + Sync + 'static {
     /// implement any retry logic. All errors returned by this function are
     /// considered unrecoverable and will be reported to a configured error
     /// Handler.
-    async fn export(&self, metrics: &mut ResourceMetrics) -> Result<()>;
+    async fn export(&self, metrics: &mut ResourceMetrics, timeout: Duration) -> Result<()>;
 
     /// Flushes any metric data held by an exporter.
     async fn force_flush(&self) -> Result<()>;

opentelemetry-sdk/src/metrics/mod.rs

@@ -131,8 +131,8 @@
     use super::*;
     use crate::metrics::data::{ResourceMetrics, Temporality};
     use crate::metrics::reader::TemporalitySelector;
+    use crate::testing::metrics::InMemoryMetricsExporter;
     use crate::testing::metrics::InMemoryMetricsExporterBuilder;
-    use crate::{runtime, testing::metrics::InMemoryMetricsExporter};
     use opentelemetry::metrics::{Counter, Meter, UpDownCounter};
     use opentelemetry::{metrics::MeterProvider as _, KeyValue};
     use rand::{rngs, Rng, SeedableRng};
@@ -491,7 +491,7 @@
     async fn counter_duplicate_instrument_merge() {
         // Arrange
         let exporter = InMemoryMetricsExporter::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
@@ -542,7 +542,7 @@
     async fn counter_duplicate_instrument_different_meter_no_merge() {
         // Arrange
         let exporter = InMemoryMetricsExporter::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
@@ -631,7 +631,7 @@
     async fn instrumentation_scope_identity_test() {
         // Arrange
         let exporter = InMemoryMetricsExporter::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
@@ -714,7 +714,7 @@
 
         // Arrange
         let exporter = InMemoryMetricsExporter::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 {
@@ -764,7 +764,7 @@
 
         // Arrange
         let exporter = InMemoryMetricsExporter::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![]);
@@ -839,7 +839,7 @@
 
         // Arrange
         let exporter = InMemoryMetricsExporter::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![]);
@@ -1250,6 +1250,58 @@
         asynchronous_instruments_cumulative_with_gap_in_measurements_helper("gauge");
     }
 
+    async fn some_async_function() -> u64 {
+        std::thread::sleep(std::time::Duration::from_millis(10));
+        1
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn async_inside_observable_callback_from_tokio_multi_with_one_worker() {
+        // Run this test with stdout enabled to see output.
+        // cargo test async_inside_observable_callbacks --features=testing -- --nocapture
+        // Arrange
+        async_inside_observable_callback_helper();
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn async_inside_observable_callback_from_tokio_multi_with_two_worker() {
+        // Run this test with stdout enabled to see output.
+        // cargo test async_inside_observable_callbacks --features=testing -- --nocapture
+        // Arrange
+        async_inside_observable_callback_helper();
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn async_inside_observable_callback_from_tokio_current_thread() {
+        // Run this test with stdout enabled to see output.
+        // cargo test async_inside_observable_callbacks --features=testing -- --nocapture
+        // Arrange
+        async_inside_observable_callback_helper();
+    }
+
+    fn async_inside_observable_callback_helper() {
+        // Run this test with stdout enabled to see output.
+        // cargo test async_inside_observable_callbacks --features=testing -- --nocapture
+        // Arrange
+        let mut test_context = TestContext::new(Temporality::Delta);
+        let _obs_gauge = test_context
+            .meter()
+            .u64_observable_gauge("my_observable_gauge")
+            .with_callback(|observer| {
+                // call async function from here
+                let value = futures_executor::block_on(some_async_function());
+                observer.observe(value, &[]);
+            })
+            .init();
+
+        test_context.flush_metrics();
+        let last_value =
+            test_context.get_aggregation::<data::Gauge<u64>>("my_observable_gauge", None);
+        assert_eq!(last_value.data_points.len(), 1);
+        let data_point = &last_value.data_points[0];
+        assert_eq!(data_point.value, 1);
+    }
+
     fn asynchronous_instruments_cumulative_with_gap_in_measurements_helper(
         instrument_name: &'static str,
     ) {
@@ -2373,7 +2425,7 @@
             exporter = exporter.with_temporality_selector(TestTemporalitySelector(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 {