Merge branch 'main' into appender-tracing-fix-trace-context

Author: cijothomas

opentelemetry-appender-tracing/benches/logs.rs

@@ -16,8 +16,8 @@
 use criterion::{criterion_group, criterion_main, Criterion};
 use opentelemetry::InstrumentationScope;
 use opentelemetry_appender_tracing::layer as tracing_layer;
-use opentelemetry_sdk::export::logs::{LogBatch, LogExporter};
 use opentelemetry_sdk::logs::LogResult;
+use opentelemetry_sdk::logs::{LogBatch, LogExporter};
 use opentelemetry_sdk::logs::{LogProcessor, LogRecord, LoggerProvider};
 use opentelemetry_sdk::Resource;
 use pprof::criterion::{Output, PProfProfiler};

opentelemetry-appender-tracing/src/layer.rs

@@ -233,7 +233,7 @@ mod tests {
     use opentelemetry::trace::TracerProvider as _;
     use opentelemetry::trace::{TraceContextExt, TraceFlags, Tracer};
     use opentelemetry::{logs::AnyValue, Key};
-    use opentelemetry_sdk::export::logs::{LogBatch, LogExporter};
+    use opentelemetry_sdk::logs::{LogBatch, LogExporter};
     use opentelemetry_sdk::logs::{LogRecord, LogResult, LoggerProvider};
     use opentelemetry_sdk::testing::logs::InMemoryLogExporter;
     use opentelemetry_sdk::trace::{Sampler, TracerProvider};

opentelemetry-otlp/src/exporter/http/logs.rs

@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use http::{header::CONTENT_TYPE, Method};
 use opentelemetry::otel_debug;
-use opentelemetry_sdk::export::logs::{LogBatch, LogExporter};
+use opentelemetry_sdk::logs::{LogBatch, LogExporter};
 use opentelemetry_sdk::logs::{LogError, LogResult};
 
 use super::OtlpHttpClient;

opentelemetry-otlp/src/exporter/http/mod.rs

@@ -13,10 +13,10 @@ use opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema;
 use opentelemetry_proto::transform::logs::tonic::group_logs_by_resource_and_scope;
 #[cfg(feature = "trace")]
 use opentelemetry_proto::transform::trace::tonic::group_spans_by_resource_and_scope;
-#[cfg(feature = "logs")]
-use opentelemetry_sdk::export::logs::LogBatch;
 #[cfg(feature = "trace")]
 use opentelemetry_sdk::export::trace::SpanData;
+#[cfg(feature = "logs")]
+use opentelemetry_sdk::logs::LogBatch;
 use prost::Message;
 use std::collections::HashMap;
 use std::env;

opentelemetry-otlp/src/exporter/tonic/logs.rs

@@ -3,7 +3,7 @@ use opentelemetry::otel_debug;
 use opentelemetry_proto::tonic::collector::logs::v1::{
     logs_service_client::LogsServiceClient, ExportLogsServiceRequest,
 };
-use opentelemetry_sdk::export::logs::{LogBatch, LogExporter};
+use opentelemetry_sdk::logs::{LogBatch, LogExporter};
 use opentelemetry_sdk::logs::{LogError, LogResult};
 use tonic::{codegen::CompressionEncoding, service::Interceptor, transport::Channel, Request};
 

opentelemetry-otlp/src/logs.rs

@@ -8,7 +8,7 @@ use std::fmt::Debug;
 
 use opentelemetry_sdk::logs::LogResult;
 
-use opentelemetry_sdk::export::logs::LogBatch;
+use opentelemetry_sdk::logs::LogBatch;
 
 use crate::{HasExportConfig, NoExporterBuilderSet};
 
@@ -140,7 +140,7 @@ impl LogExporter {
     }
 }
 
-impl opentelemetry_sdk::export::logs::LogExporter for LogExporter {
+impl opentelemetry_sdk::logs::LogExporter for LogExporter {
     #[allow(clippy::manual_async_fn)]
     fn export(
         &self,

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

@@ -20,8 +20,7 @@
 
 use anyhow::Result;
 use opentelemetry::{otel_debug, otel_info};
-use std::fs;
-use std::fs::File;
+use std::fs::{self, File, OpenOptions};
 use std::os::unix::fs::PermissionsExt;
 use std::sync::{Arc, Mutex, Once, OnceLock};
 use testcontainers::core::wait::HttpWaitStrategy;
@@ -125,6 +124,17 @@ fn upsert_empty_file(path: &str) -> File {
     file
 }
 
+/// Cleans up file specificed as argument by truncating its content.
+///
+/// This function is meant to cleanup the generated json file before a test starts,
+/// preventing entries from previous tests from interfering with the current test's results.
+pub fn cleanup_file(file_path: &str) {
+    let _ = OpenOptions::new()
+        .write(true)
+        .truncate(true)
+        .open(file_path); // ignore result, as file may not exist
+}
+
 ///
 /// Shuts down our collector container. This should be run as part of each test
 /// suite shutting down!

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

@@ -2,12 +2,14 @@
 
 use anyhow::Result;
 use ctor::dtor;
+use integration_test_runner::logs_asserter::{read_logs_from_json, LogsAsserter};
 use integration_test_runner::test_utils;
 use opentelemetry_otlp::LogExporter;
 use opentelemetry_sdk::logs::LoggerProvider;
 use opentelemetry_sdk::{logs as sdklogs, Resource};
 use std::fs::File;
 use std::io::Read;
+use std::os::unix::fs::MetadataExt;
 
 fn init_logs(is_simple: bool) -> Result<sdklogs::LoggerProvider> {
     let exporter_builder = LogExporter::builder();
@@ -88,26 +90,26 @@ mod logtests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     #[cfg(any(feature = "tonic-client", feature = "reqwest-blocking-client"))]
     pub async fn logs_batch_tokio_multi_thread() -> Result<()> {
-        logs_batch_tokio_helper().await
+        logs_tokio_helper(false).await
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     #[cfg(any(feature = "tonic-client", feature = "reqwest-blocking-client"))]
     pub async fn logs_batch_tokio_multi_with_one_worker() -> Result<()> {
-        logs_batch_tokio_helper().await
+        logs_tokio_helper(false).await
     }
 
     #[tokio::test(flavor = "current_thread")]
     #[cfg(any(feature = "tonic-client", feature = "reqwest-blocking-client"))]
     pub async fn logs_batch_tokio_current() -> Result<()> {
-        logs_batch_tokio_helper().await
+        logs_tokio_helper(false).await
     }
 
-    async fn logs_batch_tokio_helper() -> Result<()> {
-        use crate::{assert_logs_results, init_logs};
+    async fn logs_tokio_helper(is_simple: bool) -> Result<()> {
+        use crate::{assert_logs_results_contains, init_logs};
         test_utils::start_collector_container().await?;
 
-        let logger_provider = init_logs(false).unwrap();
+        let logger_provider = init_logs(is_simple).unwrap();
         let layer = OpenTelemetryTracingBridge::new(&logger_provider);
         let subscriber = tracing_subscriber::registry().with(layer);
         // generate a random uuid and store it to expected guid
@@ -119,58 +121,37 @@ mod logtests {
 
         let _ = logger_provider.shutdown();
         tokio::time::sleep(Duration::from_secs(5)).await;
-        assert_logs_results(test_utils::LOGS_FILE, expected_uuid.as_str())?;
+        assert_logs_results_contains(test_utils::LOGS_FILE, expected_uuid.as_str())?;
         Ok(())
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     #[cfg(any(feature = "tonic-client", feature = "reqwest-client"))]
     pub async fn logs_simple_tokio_multi_thread() -> Result<()> {
-        logs_simple_tokio_helper().await
+        logs_tokio_helper(true).await
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     #[cfg(any(feature = "tonic-client", feature = "reqwest-client"))]
     pub async fn logs_simple_tokio_multi_with_one_worker() -> Result<()> {
-        logs_simple_tokio_helper().await
+        logs_tokio_helper(true).await
     }
 
     // Ignored, to be investigated
     #[ignore]
     #[tokio::test(flavor = "current_thread")]
     #[cfg(any(feature = "tonic-client", feature = "reqwest-client"))]
     pub async fn logs_simple_tokio_current() -> Result<()> {
-        logs_simple_tokio_helper().await
-    }
-
-    async fn logs_simple_tokio_helper() -> Result<()> {
-        use crate::{assert_logs_results, init_logs};
-        test_utils::start_collector_container().await?;
-
-        let logger_provider = init_logs(true).unwrap();
-        let layer = OpenTelemetryTracingBridge::new(&logger_provider);
-        let subscriber = tracing_subscriber::registry().with(layer);
-        info!("Tracing initialized");
-        // generate a random uuid and store it to expected guid
-        let expected_uuid = Uuid::new_v4().to_string();
-        {
-            let _guard = tracing::subscriber::set_default(subscriber);
-            info!(target: "my-target",  uuid = expected_uuid, "hello from {}. My price is {}.", "banana", 2.99);
-        }
-
-        let _ = logger_provider.shutdown();
-        tokio::time::sleep(Duration::from_secs(5)).await;
-        assert_logs_results(test_utils::LOGS_FILE, expected_uuid.as_str())?;
-        Ok(())
+        logs_tokio_helper(true).await
     }
 
     #[test]
     #[cfg(any(feature = "tonic-client", feature = "reqwest-blocking-client"))]
     pub fn logs_batch_non_tokio_main() -> Result<()> {
-        logs_batch_non_tokio_helper()
+        logs_non_tokio_helper(false)
     }
 
-    fn logs_batch_non_tokio_helper() -> Result<()> {
+    fn logs_non_tokio_helper(is_simple: bool) -> 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
@@ -179,7 +160,7 @@ mod logtests {
         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(false)
+            init_logs(is_simple)
         })?;
         let layer = layer::OpenTelemetryTracingBridge::new(&logger_provider);
         let subscriber = tracing_subscriber::registry().with(layer);
@@ -192,43 +173,18 @@ mod logtests {
 
         let _ = logger_provider.shutdown();
         std::thread::sleep(Duration::from_secs(5));
-        assert_logs_results(test_utils::LOGS_FILE, expected_uuid.as_str())?;
+        assert_logs_results_contains(test_utils::LOGS_FILE, expected_uuid.as_str())?;
         Ok(())
     }
 
     #[test]
     #[cfg(any(feature = "tonic-client", feature = "reqwest-blocking-client"))]
     pub fn logs_simple_non_tokio_main() -> Result<()> {
-        logs_simple_non_tokio_helper()
-    }
-
-    fn logs_simple_non_tokio_helper() -> Result<()> {
-        // Initialize the logger provider inside a tokio runtime
-        // as this allows tonic client to capture the runtime,
-        // but actual export occurs from the main non-tokio thread.
-        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(true)
-        })?;
-        let layer = layer::OpenTelemetryTracingBridge::new(&logger_provider);
-        let subscriber = tracing_subscriber::registry().with(layer);
-        // generate a random uuid and store it to expected guid
-        let expected_uuid = Uuid::new_v4().to_string();
-        {
-            let _guard = tracing::subscriber::set_default(subscriber);
-            info!(target: "my-target",  uuid = expected_uuid, "hello from {}. My price is {}.", "banana", 2.99);
-        }
-
-        let _ = logger_provider.shutdown();
-        std::thread::sleep(Duration::from_secs(5));
-        assert_logs_results(test_utils::LOGS_FILE, expected_uuid.as_str())?;
-        Ok(())
+        logs_non_tokio_helper(true)
     }
 }
 
-pub fn assert_logs_results(result: &str, expected_content: &str) -> Result<()> {
+pub fn assert_logs_results_contains(result: &str, expected_content: &str) -> Result<()> {
     let file = File::open(result)?;
     let mut contents = String::new();
     let mut reader = std::io::BufReader::new(&file);
@@ -237,6 +193,16 @@ pub fn assert_logs_results(result: &str, expected_content: &str) -> Result<()> {
     Ok(())
 }
 
+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);
+    Ok(())
+}
+
 ///
 /// Make sure we stop the collector container, otherwise it will sit around hogging our
 /// ports and subsequent test runs will fail.

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

@@ -0,0 +1,64 @@
+#![cfg(unix)]
+
+use anyhow::Result;
+use ctor::dtor;
+use integration_test_runner::logs_asserter::{read_logs_from_json, LogsAsserter};
+use integration_test_runner::test_utils;
+use opentelemetry_appender_tracing::layer;
+use opentelemetry_otlp::LogExporter;
+use opentelemetry_sdk::logs::LoggerProvider;
+use opentelemetry_sdk::Resource;
+use std::fs::File;
+use std::os::unix::fs::MetadataExt;
+use tracing::info;
+use tracing_subscriber::layer::SubscriberExt;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[cfg(feature = "tonic-client")]
+pub async fn test_logs() -> Result<()> {
+    test_utils::start_collector_container().await?;
+    test_utils::cleanup_file("./actual/logs.json"); // Ensure logs.json is empty before the test
+    let exporter_builder = LogExporter::builder().with_tonic();
+    let exporter = exporter_builder.build()?;
+    let mut logger_provider_builder = LoggerProvider::builder();
+    logger_provider_builder = logger_provider_builder.with_batch_exporter(exporter);
+    let logger_provider = logger_provider_builder
+        .with_resource(
+            Resource::builder_empty()
+                .with_service_name("logs-integration-test")
+                .build(),
+        )
+        .build();
+    let layer = 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(std::time::Duration::from_secs(5)).await;
+    assert_logs_results(test_utils::LOGS_FILE, "expected/logs.json")?;
+    Ok(())
+}
+
+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);
+    Ok(())
+}
+
+///
+/// Make sure we stop the collector container, otherwise it will sit around hogging our
+/// ports and subsequent test runs will fail.
+///
+#[dtor]
+fn shutdown() {
+    println!("metrics::shutdown");
+    test_utils::stop_collector_container();
+}

opentelemetry-proto/src/transform/logs.rs

@@ -12,7 +12,7 @@ pub mod tonic {
         transform::common::{to_nanos, tonic::ResourceAttributesWithSchema},
     };
     use opentelemetry::logs::{AnyValue as LogsAnyValue, Severity};
-    use opentelemetry_sdk::export::logs::LogBatch;
+    use opentelemetry_sdk::logs::LogBatch;
     use std::borrow::Cow;
     use std::collections::HashMap;
 
@@ -222,7 +222,7 @@ mod tests {
     use crate::transform::common::tonic::ResourceAttributesWithSchema;
     use opentelemetry::logs::LogRecord as _;
     use opentelemetry::InstrumentationScope;
-    use opentelemetry_sdk::{export::logs::LogBatch, logs::LogRecord, Resource};
+    use opentelemetry_sdk::{logs::LogBatch, logs::LogRecord, Resource};
     use std::time::SystemTime;
 
     fn create_test_log_data(