Minor fixes to PeriodicReader and examples (#2409)

Author: cijothomas

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

@@ -15,7 +15,7 @@ experimental_metrics_periodicreader_with_async_runtime = ["opentelemetry_sdk/exp
 [dependencies]
 once_cell = { workspace = true }
 opentelemetry = { path = "../../../opentelemetry" }
-opentelemetry_sdk = { path = "../../../opentelemetry-sdk", features = ["rt-tokio", "metrics", "logs", "experimental_metrics_periodicreader_with_async_runtime"]}
+opentelemetry_sdk = { path = "../../../opentelemetry-sdk", features = ["rt-tokio", "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

@@ -82,12 +82,6 @@ fn init_metrics() -> Result<opentelemetry_sdk::metrics::SdkMeterProvider, Metric
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
-    let tracer_provider = init_traces()?;
-    global::set_tracer_provider(tracer_provider.clone());
-
-    let meter_provider = init_metrics()?;
-    global::set_meter_provider(meter_provider.clone());
-
     let logger_provider = init_logs()?;
 
     // Create a new OpenTelemetryTracingBridge using the above LoggerProvider.
@@ -126,6 +120,12 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
         .with(fmt_layer)
         .init();
 
+    let tracer_provider = init_traces()?;
+    global::set_tracer_provider(tracer_provider.clone());
+
+    let meter_provider = init_metrics()?;
+    global::set_meter_provider(meter_provider.clone());
+
     let common_scope_attributes = vec![KeyValue::new("scope-key", "scope-value")];
     let scope = InstrumentationScope::builder("basic")
         .with_version("1.0")

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

@@ -7,9 +7,9 @@ publish = false
 
 [dependencies]
 once_cell = { workspace = true }
-opentelemetry = { path = "../../../opentelemetry", features = ["metrics", "logs"] }
-opentelemetry_sdk = { path = "../../../opentelemetry-sdk", features = ["rt-tokio", "logs"] }
-opentelemetry-otlp = { path = "../../../opentelemetry-otlp", features = ["tonic", "metrics", "logs"] }
+opentelemetry = { path = "../../../opentelemetry" }
+opentelemetry_sdk = { path = "../../../opentelemetry-sdk", features = ["rt-tokio"] }
+opentelemetry-otlp = { path = "../../../opentelemetry-otlp" }
 opentelemetry-semantic-conventions = { path = "../../../opentelemetry-semantic-conventions" }
 tokio = { version = "1.0", features = ["full"] }
 opentelemetry-appender-tracing = { path = "../../../opentelemetry-appender-tracing", default-features = false}

opentelemetry-otlp/examples/basic-otlp/README.md

@@ -14,8 +14,16 @@ The example employs a `BatchExporter` for logs and traces, which is the
 recommended approach when using OTLP exporters. While it can be modified to use
 a `SimpleExporter`, this requires the main method to be a `tokio::main` function
 since the `tonic` client requires a Tokio runtime. If you prefer not to use
-`tokio::main`, then the `init_logs`, `init_traces`, and `init_metrics` functions
-must be executed within a Tokio runtime. Below is an example:
+`tokio::main`, then the `init_logs` and `init_traces` functions must be executed
+within a Tokio runtime. 
+
+This examples uses the default `PeriodicReader` for metrics, which uses own
+thread for background processing/exporting. Since the `tonic` client requires a
+Tokio runtime, the main method must be a `tokio::main` function. If you prefer not
+to use `tokio::main`, then the `init_metrics` function must be executed within a
+Tokio runtime.
+
+Below is an example on how to use non `tokio::main`:
 
 ```rust
 fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {

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

@@ -56,12 +56,6 @@ fn init_logs() -> Result<opentelemetry_sdk::logs::LoggerProvider, LogError> {
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
-    let tracer_provider = init_traces()?;
-    global::set_tracer_provider(tracer_provider.clone());
-
-    let meter_provider = init_metrics()?;
-    global::set_meter_provider(meter_provider.clone());
-
     let logger_provider = init_logs()?;
 
     // Create a new OpenTelemetryTracingBridge using the above LoggerProvider.
@@ -100,6 +94,12 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
         .with(fmt_layer)
         .init();
 
+    let tracer_provider = init_traces()?;
+    global::set_tracer_provider(tracer_provider.clone());
+
+    let meter_provider = init_metrics()?;
+    global::set_meter_provider(meter_provider.clone());
+
     let common_scope_attributes = vec![KeyValue::new("scope-key", "scope-value")];
     let scope = InstrumentationScope::builder("basic")
         .with_version("1.0")

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -175,12 +175,12 @@ impl PeriodicReader {
                 let mut remaining_interval = interval;
                 otel_info!(
                     name: "PeriodReaderThreadStarted",
-                    interval = interval.as_secs(),
-                    timeout = timeout.as_secs()
+                    interval_in_millisecs = interval.as_millis(),
+                    timeout_in_millisecs = timeout.as_millis()
                 );
                 loop {
                     otel_debug!(
-                        name: "PeriodReaderThreadLoopAlive", message = "Next export will happen after interval, unless flush or shutdown is triggered.", interval = remaining_interval.as_millis()
+                        name: "PeriodReaderThreadLoopAlive", message = "Next export will happen after interval, unless flush or shutdown is triggered.", interval_in_millisecs = remaining_interval.as_millis()
                     );
                     match message_receiver.recv_timeout(remaining_interval) {
                         Ok(Message::Flush(response_sender)) => {
@@ -228,22 +228,19 @@ impl PeriodicReader {
                         Err(mpsc::RecvTimeoutError::Timeout) => {
                             let export_start = Instant::now();
                             otel_debug!(
-                                name: "PeriodReaderThreadExportingDueToTimer",
-                                event_name = "PeriodReaderThreadExportingDueToTimer"
+                                name: "PeriodReaderThreadExportingDueToTimer"
                             );
 
                             if let Err(_e) = cloned_reader.collect_and_export(timeout) {
                                 otel_debug!(
-                                    name: "PeriodReaderThreadExportingDueToTimerFailed",
-                                    event_name = "PeriodReaderThreadExportingDueToTimerFailed"
+                                    name: "PeriodReaderThreadExportingDueToTimerFailed"
                                 );
                             }
 
                             let time_taken_for_export = export_start.elapsed();
                             if time_taken_for_export > interval {
                                 otel_debug!(
-                                    name: "PeriodReaderThreadExportTookLongerThanInterval",
-                                    event_name = "PeriodReaderThreadExportTookLongerThanInterval"
+                                    name: "PeriodReaderThreadExportTookLongerThanInterval"
                                 );
                                 // if export took longer than interval, do the
                                 // next export immediately.
@@ -265,8 +262,7 @@ impl PeriodicReader {
                     }
                 }
                 otel_info!(
-                    name: "PeriodReaderThreadStopped",
-                    event_name = "PeriodReaderThreadStopped"
+                    name: "PeriodReaderThreadStopped"
                 );
             });
 
@@ -275,7 +271,6 @@ impl PeriodicReader {
         if let Err(e) = result_thread_creation {
             otel_error!(
                 name: "PeriodReaderThreadStartError",
-                event_name =  "PeriodReaderThreadStartError",
                 error = format!("{:?}", e)
             );
         }
@@ -339,7 +334,6 @@ impl PeriodicReaderInner {
         if let Err(e) = collect_result {
             otel_warn!(
                 name: "PeriodReaderCollectError",
-                event_name = "PeriodReaderCollectError",
                 error = format!("{:?}", e)
             );
             return Err(e);
@@ -350,6 +344,11 @@ impl PeriodicReaderInner {
             return Ok(());
         }
 
+        let metrics_count = rm.scope_metrics.iter().fold(0, |count, scope_metrics| {
+            count + scope_metrics.metrics.len()
+        });
+        otel_debug!(name: "PeriodicReaderMetricsCollected", count = metrics_count);
+
         // TODO: substract the time taken for collect from the timeout. collect
         // involves observable callbacks too, which are user defined and can
         // take arbitrary time.
@@ -362,7 +361,6 @@ impl PeriodicReaderInner {
         if let Err(e) = exporter_result {
             otel_warn!(
                 name: "PeriodReaderExportError",
-                event_name = "PeriodReaderExportError",
                 error = format!("{:?}", e)
             );
             return Err(e);
@@ -385,7 +383,6 @@ impl PeriodicReaderInner {
             Err(e) => {
                 otel_debug!(
                     name: "PeriodReaderForceFlushError",
-                    event_name = "PeriodReaderForceFlushError",
                     error = format!("{:?}", e)
                 );
                 return Err(MetricError::Other(e.to_string()));
@@ -420,7 +417,6 @@ impl PeriodicReaderInner {
             Err(e) => {
                 otel_debug!(
                     name: "PeriodReaderShutdownError",
-                    event_name = "PeriodReaderShutdownError",
                     error = format!("{:?}", e)
                 );
                 return Err(MetricError::Other(e.to_string()));
@@ -504,7 +500,7 @@ mod tests {
     };
 
     // use below command to run all tests
-    // cargo test metrics::periodic_reader_with_own_thread::tests --features=testing,experimental_metrics_periodic_reader_no_runtime -- --nocapture
+    // cargo test metrics::periodic_reader::tests --features=testing,spec_unstable_metrics_views -- --nocapture
 
     #[derive(Debug, Clone)]
     struct MetricExporterThatFailsOnlyOnFirst {