changes..

Author: lalitb

opentelemetry-sdk/src/export/logs/mod.rs

@@ -81,9 +81,10 @@ pub trait LogExporter: Send + Sync + Debug {
     /// `Ok(())`) or an error (`Err(LogError)`) if the export operation failed.
     ///
     fn export<'a>(
-        &mut self,
-        batch: LogBatch<'a>,
-    ) -> impl std::future::Future<Output = LogResult<()>> + Send;
+        &'a mut self,
+        batch: &'a LogBatch<'a>,
+    ) -> impl std::future::Future<Output = LogResult<()>> + Send + 'a;
+
     /// Shuts down the exporter.
     fn shutdown(&mut self) {}
     #[cfg(feature = "spec_unstable_logs_enabled")]

opentelemetry-sdk/src/logs/log_processor.rs

@@ -108,7 +108,8 @@ impl<T: LogExporter> LogProcessor for SimpleLogProcessor<T> {
             .map_err(|_| LogError::MutexPoisoned("SimpleLogProcessor".into()))
             .and_then(|mut exporter| {
                 let log_tuple = &[(record as &LogRecord, instrumentation)];
-                futures_executor::block_on(exporter.export(LogBatch::new(log_tuple)))
+                let log_batch = LogBatch::new(log_tuple);
+                futures_executor::block_on(exporter.export(&log_batch))
             });
         // Handle errors with specific static names
         match result {
@@ -217,7 +218,6 @@ impl<R: RuntimeChannel> BatchLogProcessor<R> {
     {
         let (message_sender, message_receiver) =
             runtime.batch_message_channel(config.max_queue_size);
-        //let exporter = Arc::new(Mutex::new(exporter));
         let inner_runtime = runtime.clone();
 
         // Spawn worker process via user-defined spawn function.
@@ -340,7 +340,8 @@ where
         .iter()
         .map(|log_data| (&log_data.0, &log_data.1))
         .collect();
-    let export = exporter.export(LogBatch::new(log_vec.as_slice()));
+    let log_batch = LogBatch::new(log_vec.as_slice());
+    let export = exporter.export(&log_batch);
     let timeout = runtime.delay(time_out);
     pin_mut!(export);
     pin_mut!(timeout);
@@ -567,10 +568,10 @@ mod tests {
     }
 
     impl LogExporter for MockLogExporter {
-        fn export(
-            &mut self,
-            _batch: LogBatch<'_>,
-        ) -> impl std::future::Future<Output = LogResult<()>> + Send + '_ {
+        fn export<'a>(
+            &'a mut self,
+            _batch: &'a LogBatch<'a>,
+        ) -> impl std::future::Future<Output = LogResult<()>> + Send + 'a {
             async { Ok(()) }
         }
 
@@ -1064,10 +1065,10 @@ mod tests {
     }
 
     impl LogExporter for LogExporterThatRequiresTokio {
-        fn export(
-            &mut self,
-            batch: LogBatch<'_>,
-        ) -> impl std::future::Future<Output = LogResult<()>> + Send {
+        fn export<'a>(
+            &'a mut self,
+            batch: &'a LogBatch<'a>,
+        ) -> impl std::future::Future<Output = LogResult<()>> + Send + 'a {
             // Simulate minimal dependency on tokio by sleeping asynchronously for a short duration
             async move {
                 tokio::time::sleep(Duration::from_millis(50)).await;

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

@@ -182,10 +182,10 @@ impl InMemoryLogExporter {
 }
 
 impl LogExporter for InMemoryLogExporter {
-    fn export(
-        &mut self,
-        batch: LogBatch<'_>,
-    ) -> impl std::future::Future<Output = LogResult<()>> + Send {
+    fn export<'a>(
+        &'a mut self,
+        batch: &'a LogBatch<'a>,
+    ) -> impl std::future::Future<Output = LogResult<()>> + Send + 'a {
         async move {
             let mut logs_guard = self.logs.lock().map_err(LogError::from)?;
             for (log_record, instrumentation) in batch.iter() {

opentelemetry-stdout/src/logs/exporter.rs

@@ -32,33 +32,27 @@ impl fmt::Debug for LogExporter {
 
 impl opentelemetry_sdk::export::logs::LogExporter for LogExporter {
     /// Export spans to stdout
-    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();
+    fn export<'a>(
+        &'a mut self,
+        batch: &'a LogBatch<'a>,
+    ) -> impl std::future::Future<Output = LogResult<()>> + Send + 'a {
         async move {
-            if is_shutdown {
-                Err("exporter is shut down".into())
+            if self.is_shutdown.load(atomic::Ordering::SeqCst) {
+                return Err("exporter is shut down".into());
             } else {
                 println!("Logs");
-                let mut resource_emitted_guard = resource_emitted_arc_clone.lock().unwrap();
-                if *resource_emitted_guard {
+                if self.resource_emitted {
                     print_logs(batch);
                 } else {
+                    self.resource_emitted = true;
                     println!("Resource");
-                    if let Some(schema_url) = resource.schema_url() {
+                    if let Some(schema_url) = self.resource.schema_url() {
                         println!("\t Resource SchemaUrl: {:?}", schema_url);
                     }
-                    resource.iter().for_each(|(k, v)| {
+                    self.resource.iter().for_each(|(k, v)| {
                         println!("\t ->  {}={:?}", k, v);
                     });
-
                     print_logs(batch);
-                    *resource_emitted_guard = true;
                 }
 
                 Ok(())
@@ -75,7 +69,7 @@ impl opentelemetry_sdk::export::logs::LogExporter for LogExporter {
     }
 }
 
-fn print_logs(batch: LogBatch<'_>) {
+fn print_logs(batch: &LogBatch<'_>) {
     for (i, log) in batch.iter().enumerate() {
         println!("Log #{}", i);
         let (record, _library) = log;