initial commit

Author: lalitb

opentelemetry-appender-tracing/benches/logs.rs

@@ -33,7 +33,7 @@ struct NoopExporter {
 
 #[async_trait]
 impl LogExporter for NoopExporter {
-    async fn export(&mut self, _: Vec<LogData>) -> LogResult<()> {
+    async fn export<'a>(&mut self, _: Vec<std::borrow::Cow<'a, LogData>>) -> LogResult<()> {
         LogResult::Ok(())
     }
 
@@ -54,7 +54,7 @@ impl NoopProcessor {
 }
 
 impl LogProcessor for NoopProcessor {
-    fn emit(&self, _: LogData) {
+    fn emit(&self, _: &mut LogData) {
         // no-op
     }
 

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

@@ -9,7 +9,7 @@ use super::OtlpHttpClient;
 
 #[async_trait]
 impl LogExporter for OtlpHttpClient {
-    async fn export(&mut self, batch: Vec<LogData>) -> LogResult<()> {
+    async fn export<'a>(&mut self, batch: Vec<std::borrow::Cow<'a, LogData>>) -> LogResult<()> {
         let client = self
             .client
             .lock()
@@ -19,7 +19,12 @@ impl LogExporter for OtlpHttpClient {
                 _ => Err(LogError::Other("exporter is already shut down".into())),
             })?;
 
-        let (body, content_type) = { self.build_logs_export_body(batch, &self.resource)? };
+        //TBD :avoid cloning and work only on borrowed logdata
+        let owned_batch = batch.into_iter()
+            .map(|cow_log_data| cow_log_data.into_owned())  // Converts Cow to owned LogData
+            .collect::<Vec<LogData>>();
+
+        let (body, content_type) = { self.build_logs_export_body(owned_batch, &self.resource)? };
         let mut request = http::Request::builder()
             .method(Method::POST)
             .uri(&self.collector_endpoint)

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

@@ -52,7 +52,8 @@ impl TonicLogsClient {
 
 #[async_trait]
 impl LogExporter for TonicLogsClient {
-    async fn export(&mut self, batch: Vec<LogData>) -> LogResult<()> {
+    async fn export<'a>(&mut self, batch: Vec<std::borrow::Cow<'a, LogData>>) -> LogResult<()> {
+        // clone if batch is borrowed
         let (mut client, metadata, extensions) = match &mut self.inner {
             Some(inner) => {
                 let (m, e, _) = inner
@@ -65,9 +66,11 @@ impl LogExporter for TonicLogsClient {
             None => return Err(LogError::Other("exporter is already shut down".into())),
         };
 
+        // TBD: We can avoid cloning and work only on logdata reference
         let resource_logs = {
             batch
                 .into_iter()
+                .map(|log_data_cow| (log_data_cow.into_owned()))
                 .map(|log_data| (log_data, &self.resource))
                 .map(Into::into)
                 .collect()

opentelemetry-otlp/src/logs.rs

@@ -98,7 +98,7 @@ impl LogExporter {
 
 #[async_trait]
 impl opentelemetry_sdk::export::logs::LogExporter for LogExporter {
-    async fn export(&mut self, batch: Vec<LogData>) -> opentelemetry::logs::LogResult<()> {
+    async fn export<'a>(&mut self, batch: Vec<std::borrow::Cow<'a, LogData>>) -> opentelemetry::logs::LogResult<()> {
         self.client.export(batch).await
     }
 

opentelemetry-sdk/benches/log.rs

@@ -19,7 +19,7 @@ struct VoidExporter;
 
 #[async_trait]
 impl LogExporter for VoidExporter {
-    async fn export(&mut self, _batch: Vec<LogData>) -> LogResult<()> {
+    async fn export<'a>(&mut self, _batch: Vec<std::borrow::Cow<'a, LogData>>) -> LogResult<()> {
         LogResult::Ok(())
     }
 }

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

@@ -9,12 +9,15 @@ use opentelemetry::{
     InstrumentationLibrary,
 };
 use std::fmt::Debug;
+use std::borrow::Cow;
 
 /// `LogExporter` defines the interface that log exporters should implement.
 #[async_trait]
 pub trait LogExporter: Send + Sync + Debug {
     /// Exports a batch of [`LogData`].
-    async fn export(&mut self, batch: Vec<LogData>) -> LogResult<()>;
+    //async fn export(&mut self, batch: Vec<LogData>) -> LogResult<()>;
+    async fn export<'a>(&mut self, batch: Vec<Cow<'a, LogData>>) -> LogResult<()>;
+
     /// Shuts down the exporter.
     fn shutdown(&mut self) {}
     #[cfg(feature = "logs_level_enabled")]

opentelemetry-sdk/src/logs/log_emitter.rs

@@ -236,17 +236,19 @@ impl opentelemetry::logs::Logger for Logger {
             cx.has_active_span()
                 .then(|| TraceContext::from(cx.span().span_context()))
         });
+        let mut log_record = record;
+        if let Some(ref trace_context) = trace_context {
+            log_record.trace_context = Some(trace_context.clone());
+        }
+
+        let mut data = LogData {
+            record: log_record,
+            instrumentation: self.instrumentation_library().clone(),
+        };
 
         for p in processors {
-            let mut cloned_record = record.clone();
-            if let Some(ref trace_context) = trace_context {
-                cloned_record.trace_context = Some(trace_context.clone());
-            }
-            let data = LogData {
-                record: cloned_record,
-                instrumentation: self.instrumentation_library().clone(),
-            };
-            p.emit(data);
+            //let mut cloned_record = record.clone();
+            p.emit(&mut data);
         }
     }
 
@@ -303,7 +305,7 @@ mod tests {
     }
 
     impl LogProcessor for ShutdownTestLogProcessor {
-        fn emit(&self, _data: LogData) {
+        fn emit(&self, _data: &mut LogData) {
             self.is_shutdown
                 .lock()
                 .map(|is_shutdown| {
@@ -535,7 +537,7 @@ mod tests {
     }
 
     impl LogProcessor for LazyLogProcessor {
-        fn emit(&self, _data: LogData) {
+        fn emit(&self, _data: &mut LogData) {
             // nothing to do.
         }
 

opentelemetry-sdk/src/logs/log_processor.rs

@@ -22,6 +22,7 @@ use std::{
     sync::Arc,
     time::Duration,
 };
+use std::borrow::Cow;
 
 /// Delay interval between two consecutive exports.
 const OTEL_BLRP_SCHEDULE_DELAY: &str = "OTEL_BLRP_SCHEDULE_DELAY";
@@ -45,7 +46,7 @@ const OTEL_BLRP_MAX_EXPORT_BATCH_SIZE_DEFAULT: usize = 512;
 /// [`Logger`]: crate::logs::Logger
 pub trait LogProcessor: Send + Sync + Debug {
     /// Called when a log record is ready to processed and exported.
-    fn emit(&self, data: LogData);
+    fn emit(&self, data: &mut LogData);
     /// Force the logs lying in the cache to be exported.
     fn force_flush(&self) -> LogResult<()>;
     /// Shuts down the processor.
@@ -80,7 +81,7 @@ impl SimpleLogProcessor {
 }
 
 impl LogProcessor for SimpleLogProcessor {
-    fn emit(&self, data: LogData) {
+    fn emit(&self, data: &mut LogData) {
         // noop after shutdown
         if self.is_shutdown.load(std::sync::atomic::Ordering::Relaxed) {
             return;
@@ -90,7 +91,7 @@ impl LogProcessor for SimpleLogProcessor {
             .exporter
             .lock()
             .map_err(|_| LogError::Other("simple logprocessor mutex poison".into()))
-            .and_then(|mut exporter| futures_executor::block_on(exporter.export(vec![data])));
+            .and_then(|mut exporter| futures_executor::block_on(exporter.export(vec![Cow::Borrowed(data)])));
         if let Err(err) = result {
             global::handle_error(err);
         }
@@ -140,8 +141,8 @@ impl<R: RuntimeChannel> Debug for BatchLogProcessor<R> {
 }
 
 impl<R: RuntimeChannel> LogProcessor for BatchLogProcessor<R> {
-    fn emit(&self, data: LogData) {
-        let result = self.message_sender.try_send(BatchMessage::ExportLog(data));
+    fn emit(&self, data: &mut LogData) {
+        let result = self.message_sender.try_send(BatchMessage::ExportLog(data.clone()));
 
         if let Err(err) = result {
             global::handle_error(LogError::Other(err.into()));
@@ -201,7 +202,7 @@ impl<R: RuntimeChannel> BatchLogProcessor<R> {
                 match message {
                     // Log has finished, add to buffer of pending logs.
                     BatchMessage::ExportLog(log) => {
-                        logs.push(log);
+                        logs.push(Cow::Owned(log));
 
                         if logs.len() == config.max_export_batch_size {
                             let result = export_with_timeout(
@@ -285,11 +286,11 @@ impl<R: RuntimeChannel> BatchLogProcessor<R> {
     }
 }
 
-async fn export_with_timeout<R, E>(
+async fn export_with_timeout<'a, R, E>(
     time_out: Duration,
     exporter: &mut E,
     runtime: &R,
-    batch: Vec<LogData>,
+    batch: Vec<Cow<'a, LogData>>,
 ) -> ExportResult
 where
     R: RuntimeChannel,
@@ -496,6 +497,7 @@ mod tests {
         OTEL_BLRP_MAX_QUEUE_SIZE, OTEL_BLRP_SCHEDULE_DELAY,
     };
     use crate::testing::logs::InMemoryLogsExporterBuilder;
+    use std::borrow::Cow;
     use crate::{
         export::logs::{LogData, LogExporter},
         logs::{
@@ -522,7 +524,7 @@ mod tests {
 
     #[async_trait]
     impl LogExporter for MockLogExporter {
-        async fn export(&mut self, _batch: Vec<LogData>) -> LogResult<()> {
+        async fn export<'a>(&mut self,  _batch: Vec<Cow<'a, LogData>>) -> LogResult<()> {
             Ok(())
         }
 
@@ -744,17 +746,15 @@ mod tests {
             BatchConfig::default(),
             runtime::Tokio,
         );
-        processor.emit(LogData {
+        let mut log_data = LogData {
             record: Default::default(),
             instrumentation: Default::default(),
-        });
+        };
+        processor.emit(&mut log_data);
         processor.force_flush().unwrap();
         processor.shutdown().unwrap();
         // todo: expect to see errors here. How should we assert this?
-        processor.emit(LogData {
-            record: Default::default(),
-            instrumentation: Default::default(),
-        });
+        processor.emit(&mut log_data);
         assert_eq!(1, exporter.get_emitted_logs().unwrap().len())
     }
 
@@ -765,10 +765,12 @@ mod tests {
             .build();
         let processor = SimpleLogProcessor::new(Box::new(exporter.clone()));
 
-        processor.emit(LogData {
+        let mut log_data = LogData{
             record: Default::default(),
             instrumentation: Default::default(),
-        });
+        };
+
+        processor.emit(&mut log_data);
 
         processor.shutdown().unwrap();
 
@@ -777,10 +779,7 @@ mod tests {
             .load(std::sync::atomic::Ordering::Relaxed);
         assert!(is_shutdown);
 
-        processor.emit(LogData {
-            record: Default::default(),
-            instrumentation: Default::default(),
-        });
+        processor.emit(&mut log_data);
 
         assert_eq!(1, exporter.get_emitted_logs().unwrap().len())
     }

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

@@ -175,11 +175,12 @@ impl InMemoryLogsExporter {
 
 #[async_trait]
 impl LogExporter for InMemoryLogsExporter {
-    async fn export(&mut self, batch: Vec<LogData>) -> LogResult<()> {
-        self.logs
-            .lock()
-            .map(|mut logs_guard| logs_guard.append(&mut batch.clone()))
-            .map_err(LogError::from)
+    async fn export<'a>(&mut self, batch: Vec<Cow<'a, LogData>>) -> LogResult<()> {
+        let mut logs_guard = self.logs.lock().map_err(LogError::from)?;
+        for log in batch.into_iter() {
+            logs_guard.push((*log).clone());
+        }
+        Ok(())
     }
     fn shutdown(&mut self) {
         if self.should_reset_on_shutdown {

opentelemetry-stdout/src/logs/exporter.rs

@@ -7,6 +7,7 @@ use opentelemetry::{
 use opentelemetry_sdk::export::logs::{ExportResult, LogData};
 use opentelemetry_sdk::Resource;
 use std::io::{stdout, Write};
+use std::borrow::Cow;
 
 type Encoder =
     Box<dyn Fn(&mut dyn Write, crate::logs::transform::LogData) -> LogResult<()> + Send + Sync>;
@@ -44,9 +45,9 @@ 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: Vec<LogData>) -> ExportResult {
+    async fn export<'a>(&mut self, batch: Vec<Cow<'a, LogData>>) -> ExportResult {
         if let Some(writer) = &mut self.writer {
-            let log_data = crate::logs::transform::LogData::from((batch, &self.resource));
+            let log_data = crate::logs::transform::LogData::from((batch.into_iter().map(Cow::into_owned).collect(), &self.resource));
             let result = (self.encoder)(writer, log_data) as LogResult<()>;
             result.and_then(|_| writer.write_all(b"\n").map_err(|e| Error(e).into()))
         } else {

stress/src/logs.rs

@@ -9,7 +9,7 @@ mod throughput;
 pub struct NoOpLogProcessor;
 
 impl LogProcessor for NoOpLogProcessor {
-    fn emit(&self, _data: opentelemetry_sdk::export::logs::LogData) {}
+    fn emit(&self, _data: &mut opentelemetry_sdk::export::logs::LogData) {}
 
     fn force_flush(&self) -> opentelemetry::logs::LogResult<()> {
         Ok(())