futher changes..

Author: lalitb

opentelemetry-sdk/Cargo.toml

@@ -44,7 +44,7 @@ pprof = { version = "0.13", features = ["flamegraph", "criterion"] }
 default = ["trace", "metrics", "logs", "internal-logs"]
 trace = ["opentelemetry/trace", "rand", "percent-encoding"]
 jaeger_remote_sampler = ["trace", "opentelemetry-http", "http", "serde", "serde_json", "url"]
-logs = ["opentelemetry/logs", "async-trait", "serde_json"]
+logs = ["opentelemetry/logs", "serde_json"]
 spec_unstable_logs_enabled = ["logs", "opentelemetry/spec_unstable_logs_enabled"]
 metrics = ["opentelemetry/metrics", "glob", "async-trait"]
 testing = ["opentelemetry/testing", "trace", "metrics", "logs", "rt-async-std", "rt-tokio", "rt-tokio-current-thread", "tokio/macros", "tokio/rt-multi-thread"]

opentelemetry-sdk/src/logs/log_emitter.rs

@@ -188,7 +188,7 @@ impl Builder {
     /// The `LogExporter` that this provider should use.
     pub fn with_simple_exporter<T: LogExporter + 'static>(self, exporter: T) -> Self {
         let mut processors = self.processors;
-        processors.push(Box::new(SimpleLogProcessor::new(Box::new(exporter))));
+        processors.push(Box::new(SimpleLogProcessor::new(exporter)));
 
         Builder { processors, ..self }
     }

opentelemetry-sdk/src/logs/log_processor.rs

@@ -552,7 +552,6 @@ mod tests {
         testing::logs::InMemoryLogExporter,
         Resource,
     };
-    use async_trait::async_trait;
     use opentelemetry::logs::AnyValue;
     use opentelemetry::logs::LogRecord as _;
     use opentelemetry::logs::{Logger, LoggerProvider as _};
@@ -567,12 +566,11 @@ mod tests {
         resource: Arc<Mutex<Option<Resource>>>,
     }
 
-    #[async_trait]
     impl LogExporter for MockLogExporter {
         fn export(
             &mut self,
             _batch: LogBatch<'_>,
-        ) -> impl std::future::Future<Output = LogResult<()>> + Send {
+        ) -> impl std::future::Future<Output = LogResult<()>> + Send + '_ {
             async { Ok(()) }
         }
 
@@ -1071,12 +1069,11 @@ mod tests {
             batch: LogBatch<'_>,
         ) -> impl std::future::Future<Output = LogResult<()>> + Send {
             // Simulate minimal dependency on tokio by sleeping asynchronously for a short duration
-            let export_count = Arc::clone(&self.export_count);
             async move {
                 tokio::time::sleep(Duration::from_millis(50)).await;
 
                 for _ in batch.iter() {
-                    export_count.fetch_add(1, Ordering::Acquire);
+                    self.export_count.fetch_add(1, Ordering::Acquire);
                 }
                 Ok(())
             }

opentelemetry-sdk/src/logs/mod.rs

@@ -49,7 +49,7 @@ mod tests {
         let exporter: InMemoryLogExporter = InMemoryLogExporter::default();
         let logger_provider = LoggerProvider::builder()
             .with_resource(resource.clone())
-            .with_log_processor(SimpleLogProcessor::new(Box::new(exporter.clone())))
+            .with_log_processor(SimpleLogProcessor::new(exporter.clone()))
             .build();
 
         // Act

opentelemetry-sdk/src/testing/logs/in_memory_exporter.rs

@@ -2,7 +2,6 @@ use crate::export::logs::{LogBatch, LogExporter};
 use crate::logs::LogRecord;
 use crate::logs::{LogError, LogResult};
 use crate::Resource;
-use async_trait::async_trait;
 use opentelemetry::InstrumentationScope;
 use std::borrow::Cow;
 use std::sync::{Arc, Mutex};
@@ -182,18 +181,22 @@ impl InMemoryLogExporter {
     }
 }
 
-#[async_trait]
 impl LogExporter for InMemoryLogExporter {
-    async fn export(&mut self, batch: LogBatch<'_>) -> LogResult<()> {
-        let mut logs_guard = self.logs.lock().map_err(LogError::from)?;
-        for (log_record, instrumentation) in batch.iter() {
-            let owned_log = OwnedLogData {
-                record: (*log_record).clone(),
-                instrumentation: (*instrumentation).clone(),
-            };
-            logs_guard.push(owned_log);
+    fn export(
+        &mut self,
+        batch: LogBatch<'_>,
+    ) -> impl std::future::Future<Output = LogResult<()>> + Send {
+        async move {
+            let mut logs_guard = self.logs.lock().map_err(LogError::from)?;
+            for (log_record, instrumentation) in batch.iter() {
+                let owned_log = OwnedLogData {
+                    record: (*log_record).clone(),
+                    instrumentation: (*instrumentation).clone(),
+                };
+                logs_guard.push(owned_log);
+            }
+            Ok(())
         }
-        Ok(())
     }
 
     fn shutdown(&mut self) {

opentelemetry-stdout/src/logs/exporter.rs

@@ -5,6 +5,7 @@ use opentelemetry_sdk::export::logs::LogBatch;
 use opentelemetry_sdk::logs::LogResult;
 use opentelemetry_sdk::Resource;
 use std::sync::atomic;
+use std::sync::{Arc, Mutex};
 
 /// An OpenTelemetry exporter that writes Logs to stdout on export.
 pub struct LogExporter {
@@ -29,30 +30,39 @@ impl fmt::Debug for LogExporter {
     }
 }
 
-#[async_trait]
 impl opentelemetry_sdk::export::logs::LogExporter for LogExporter {
     /// Export spans to stdout
-    async fn export(&mut self, batch: LogBatch<'_>) -> LogResult<()> {
-        if self.is_shutdown.load(atomic::Ordering::SeqCst) {
-            return Err("exporter is shut down".into());
-        } else {
-            println!("Logs");
-            if self.resource_emitted {
-                print_logs(batch);
+    fn export(
+        &mut self,
+        batch: LogBatch<'_>,
+    ) -> impl std::future::Future<Output = LogResult<()>> + Send {
+        let is_shutdown = self.is_shutdown.load(atomic::Ordering::SeqCst);
+        let resource_emitted_arc = Arc::new(Mutex::new(self.resource_emitted));
+        let resource_emitted_arc_clone = Arc::clone(&resource_emitted_arc);
+        let resource = self.resource.clone();
+        async move {
+            if is_shutdown {
+                Err("exporter is shut down".into())
             } else {
-                self.resource_emitted = true;
-                println!("Resource");
-                if let Some(schema_url) = self.resource.schema_url() {
-                    println!("\t Resource SchemaUrl: {:?}", schema_url);
+                println!("Logs");
+                let mut resource_emitted_guard = resource_emitted_arc_clone.lock().unwrap();
+                if *resource_emitted_guard {
+                    print_logs(batch);
+                } else {
+                    println!("Resource");
+                    if let Some(schema_url) = resource.schema_url() {
+                        println!("\t Resource SchemaUrl: {:?}", schema_url);
+                    }
+                    resource.iter().for_each(|(k, v)| {
+                        println!("\t ->  {}={:?}", k, v);
+                    });
+
+                    print_logs(batch);
+                    *resource_emitted_guard = true;
                 }
-                self.resource.iter().for_each(|(k, v)| {
-                    println!("\t ->  {}={:?}", k, v);
-                });
 
-                print_logs(batch);
+                Ok(())
             }
-
-            Ok(())
         }
     }