Merge branch 'main' into batch-span-processor-dedicated-thread

Author: lalitb

examples/tracing-grpc/src/client.rs

@@ -22,7 +22,7 @@ fn init_tracer() -> sdktrace::TracerProvider {
 
 struct MetadataMap<'a>(&'a mut tonic::metadata::MetadataMap);
 
-impl<'a> Injector for MetadataMap<'a> {
+impl Injector for MetadataMap<'_> {
     /// Set a key and value in the MetadataMap.  Does nothing if the key or value are not valid inputs
     fn set(&mut self, key: &str, value: String) {
         if let Ok(key) = tonic::metadata::MetadataKey::from_bytes(key.as_bytes()) {

examples/tracing-grpc/src/server.rs

@@ -29,7 +29,7 @@ pub mod hello_world {
 
 struct MetadataMap<'a>(&'a tonic::metadata::MetadataMap);
 
-impl<'a> Extractor for MetadataMap<'a> {
+impl Extractor for MetadataMap<'_> {
     /// Get a value for a key from the MetadataMap.  If the value can't be converted to &str, returns None
     fn get(&self, key: &str) -> Option<&str> {
         self.0.get(key).and_then(|metadata| metadata.to_str().ok())

opentelemetry-appender-log/src/lib.rs

@@ -239,7 +239,7 @@ mod any_value {
     pub(crate) fn serialize(value: log::kv::Value) -> Option<AnyValue> {
         struct ValueVisitor(Option<AnyValue>);
 
-        impl<'kvs> log::kv::VisitValue<'kvs> for ValueVisitor {
+        impl log::kv::VisitValue<'_> for ValueVisitor {
             fn visit_any(&mut self, value: log::kv::Value) -> Result<(), log::kv::Error> {
                 self.0 = Some(AnyValue::String(StringValue::from(value.to_string())));
 

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

@@ -6,11 +6,9 @@ license = "Apache-2.0"
 publish = false
 
 [features]
-default = ["reqwest", "experimental_metrics_periodicreader_with_async_runtime"]
-reqwest = ["opentelemetry-otlp/reqwest-client"]
+default = ["reqwest-blocking"]
+reqwest-blocking = ["opentelemetry-otlp/reqwest-blocking-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-otlp/examples/basic-otlp-http/README.md

@@ -13,11 +13,9 @@ applications, these filters should be adjusted appropriately.
 
 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 enabling feature flag `reqwest-blocking-client` and
-making the `main()` a normal main and *not* `tokio::main`
+a `SimpleExporter`, this requires making the main function a regular main and
+*not* tokio main.
 
-// TODO: Metrics does not work with non tokio main when using `reqwest-blocking-client` today, fix that when switching
-// default to use own thread.
 // TODO: Document `hyper` feature flag when using SimpleProcessor.
 
 ## Usage

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

@@ -11,7 +11,6 @@ use opentelemetry_otlp::{LogExporter, MetricExporter, Protocol, SpanExporter};
 use opentelemetry_sdk::{
     logs::LoggerProvider,
     metrics::{MetricError, SdkMeterProvider},
-    runtime,
     trace::{self as sdktrace, TracerProvider},
 };
 use opentelemetry_sdk::{
@@ -50,7 +49,9 @@ fn init_traces() -> Result<sdktrace::TracerProvider, TraceError> {
         .build()?;
 
     Ok(TracerProvider::builder()
-        .with_batch_exporter(exporter, runtime::Tokio)
+        // TODO: Enable BatchExporter after
+        // https://github.com/open-telemetry/opentelemetry-rust/pull/2456
+        .with_simple_exporter(exporter)
         .with_resource(RESOURCE.clone())
         .build())
 }
@@ -62,15 +63,6 @@ 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()
@@ -79,8 +71,10 @@ fn init_metrics() -> Result<opentelemetry_sdk::metrics::SdkMeterProvider, Metric
         .build())
 }
 
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
+// #[tokio::main]
+// TODO: Re-enable tokio::main, if needed, after
+// https://github.com/open-telemetry/opentelemetry-rust/pull/2456
+fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     let logger_provider = init_logs()?;
 
     // Create a new OpenTelemetryTracingBridge using the above LoggerProvider.

opentelemetry-otlp/tests/integration_test/src/logs_asserter.rs

@@ -1,3 +1,4 @@
+use anyhow::Result;
 use opentelemetry_proto::tonic::logs::v1::{LogRecord, LogsData, ResourceLogs};
 use std::fs::File;
 
@@ -96,9 +97,9 @@ impl std::fmt::Debug for LogRecordWrapper {
 }
 
 // read a file contains ResourceSpans in json format
-pub fn read_logs_from_json(file: File) -> Vec<ResourceLogs> {
+pub fn read_logs_from_json(file: File) -> Result<Vec<ResourceLogs>> {
     let reader = std::io::BufReader::new(file);
 
-    let log_data: LogsData = serde_json::from_reader(reader).unwrap();
-    log_data.resource_logs
+    let log_data: LogsData = serde_json::from_reader(reader)?;
+    Ok(log_data.resource_logs)
 }

opentelemetry-otlp/tests/integration_test/tests/logs.rs

@@ -38,19 +38,33 @@ fn init_logs() -> Result<sdklogs::LoggerProvider> {
 mod logtests {
     use super::*;
     use integration_test_runner::logs_asserter::{read_logs_from_json, LogsAsserter};
+    use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
     use std::{fs::File, time::Duration};
+    use tracing::info;
+    use tracing_subscriber::layer::SubscriberExt;
+
     #[test]
     #[should_panic(expected = "assertion `left == right` failed: body does not match")]
     pub fn test_assert_logs_eq_failure() {
-        let left = read_logs_from_json(File::open("./expected/logs.json").unwrap());
-        let right = read_logs_from_json(File::open("./expected/failed_logs.json").unwrap());
+        let left = read_logs_from_json(
+            File::open("./expected/logs.json").expect("failed to open expected file"),
+        )
+        .expect("Failed to read logs from expected file");
+
+        let right = read_logs_from_json(
+            File::open("./expected/failed_logs.json")
+                .expect("failed to open expected failed log file"),
+        )
+        .expect("Failed to read logs from expected failed log file");
         LogsAsserter::new(right, left).assert();
     }
 
     #[test]
-    pub fn test_assert_logs_eq() {
-        let logs = read_logs_from_json(File::open("./expected/logs.json").unwrap());
+    pub fn test_assert_logs_eq() -> Result<()> {
+        let logs = read_logs_from_json(File::open("./expected/logs.json")?)?;
         LogsAsserter::new(logs.clone(), logs).assert();
+
+        Ok(())
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
@@ -84,15 +98,44 @@ mod logtests {
 
         Ok(())
     }
+
+    #[test]
+    #[cfg(any(feature = "tonic-client", feature = "reqwest-blocking-client"))]
+    pub fn logs_batch_non_tokio_main() -> Result<()> {
+        // Initialize the logger provider inside a tokio runtime
+        // as this allows tonic client to capture the runtime,
+        // but actual export occurs from the dedicated std::thread
+        // created by BatchLogProcessor.
+        let rt = tokio::runtime::Runtime::new()?;
+        let logger_provider = rt.block_on(async {
+            // While we're here setup our collector container too, as this needs tokio to run
+            test_utils::start_collector_container().await?;
+            init_logs()
+        })?;
+
+        info!("LoggerProvider created");
+        let layer = OpenTelemetryTracingBridge::new(&logger_provider);
+        let subscriber = tracing_subscriber::registry().with(layer);
+        {
+            let _guard = tracing::subscriber::set_default(subscriber);
+            info!(target: "my-target", "hello from {}. My price is {}.", "banana", 2.99);
+        }
+        let _ = logger_provider.shutdown();
+        // tokio::time::sleep(Duration::from_secs(10)).await;
+        assert_logs_results(test_utils::LOGS_FILE, "expected/logs.json");
+
+        Ok(())
+    }
 }
 
-pub fn assert_logs_results(result: &str, expected: &str) {
-    let left = read_logs_from_json(File::open(expected).unwrap());
-    let right = read_logs_from_json(File::open(result).unwrap());
+pub fn assert_logs_results(result: &str, expected: &str) -> Result<()> {
+    let left = read_logs_from_json(File::open(expected)?)?;
+    let right = read_logs_from_json(File::open(result)?)?;
 
     LogsAsserter::new(left, right).assert();
 
-    assert!(File::open(result).unwrap().metadata().unwrap().size() > 0)
+    assert!(File::open(result).unwrap().metadata().unwrap().size() > 0);
+    Ok(())
 }
 
 ///

opentelemetry-sdk/CHANGELOG.md

@@ -92,7 +92,10 @@
       `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`.
-      
+
+    Feature flag "experimental_metrics_periodic_reader_no_runtime" is removed as the PeriodicReader
+    offered under that feature flag is now the default.
+
   - Bump msrv to 1.75.0.
 
 - *Breaking* : [#2314](https://github.com/open-telemetry/opentelemetry-rust/pull/2314)

opentelemetry-sdk/Cargo.toml

@@ -53,7 +53,6 @@ rt-tokio = ["tokio", "tokio-stream"]
 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"]
 experimental_logs_batch_log_processor_with_async_runtime = ["logs"]