From 56e7fae000f50594cf2558b214b0c32c345e8896 Mon Sep 17 00:00:00 2001
From: Utkarsh Umesan Pillai <66651184+utpilla@users.noreply.github.com>
Date: Mon, 30 Dec 2024 23:46:06 -0800
Subject: [PATCH] Avoid vec allocation during each export for BatchLogProcessor
 - Part 2 (#2488)

---
 opentelemetry-sdk/src/export/logs/mod.rs    | 65 ++++++++++++++++++---
 opentelemetry-sdk/src/logs/log_processor.rs |  7 +--
 2 files changed, 59 insertions(+), 13 deletions(-)

diff --git a/opentelemetry-sdk/src/export/logs/mod.rs b/opentelemetry-sdk/src/export/logs/mod.rs
index 902adb54dd..4602b2f52b 100644
--- a/opentelemetry-sdk/src/export/logs/mod.rs
+++ b/opentelemetry-sdk/src/export/logs/mod.rs
@@ -18,9 +18,17 @@ use std::fmt::Debug;
 ///
 #[derive(Debug)]
 pub struct LogBatch<'a> {
-    /// The data field contains a slice of tuples, where each tuple consists of a reference to
-    /// a `LogRecord` and a reference to an `InstrumentationScope`.
-    data: &'a [(&'a LogRecord, &'a InstrumentationScope)],
+    data: LogBatchData<'a>,
+}
+
+/// The `LogBatchData` enum represents the data field of a `LogBatch`.
+/// It can either be:
+/// - A shared reference to a slice of boxed tuples, where each tuple consists of an owned `LogRecord` and an owned `InstrumentationScope`.
+/// - Or it can be a shared reference to a slice of tuples, where each tuple consists of a reference to a `LogRecord` and a reference to an `InstrumentationScope`.
+#[derive(Debug)]
+enum LogBatchData<'a> {
+    BorrowedVec(&'a [Box<(LogRecord, InstrumentationScope)>]), // Used by BatchProcessor which clones the LogRecords for its own use.
+    BorrowedSlice(&'a [(&'a LogRecord, &'a InstrumentationScope)]),
 }
 
 impl<'a> LogBatch<'a> {
@@ -39,7 +47,17 @@ impl<'a> LogBatch<'a> {
     /// Note - this is not a public function, and should not be used directly. This would be
     /// made private in the future.
     pub fn new(data: &'a [(&'a LogRecord, &'a InstrumentationScope)]) -> LogBatch<'a> {
-        LogBatch { data }
+        LogBatch {
+            data: LogBatchData::BorrowedSlice(data),
+        }
+    }
+
+    pub(crate) fn new_with_owned_data(
+        data: &'a [Box<(LogRecord, InstrumentationScope)>],
+    ) -> LogBatch<'a> {
+        LogBatch {
+            data: LogBatchData::BorrowedVec(data),
+        }
     }
 }
 
@@ -54,9 +72,42 @@ impl LogBatch<'_> {
     /// An iterator that yields references to the `LogRecord` and `InstrumentationScope` in the batch.
     ///
     pub fn iter(&self) -> impl Iterator<Item = (&LogRecord, &InstrumentationScope)> {
-        self.data
-            .iter()
-            .map(|(record, library)| (*record, *library))
+        LogBatchDataIter {
+            data: &self.data,
+            index: 0,
+        }
+    }
+}
+
+struct LogBatchDataIter<'a> {
+    data: &'a LogBatchData<'a>,
+    index: usize,
+}
+
+impl<'a> Iterator for LogBatchDataIter<'a> {
+    type Item = (&'a LogRecord, &'a InstrumentationScope);
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.data {
+            LogBatchData::BorrowedVec(data) => {
+                if self.index < data.len() {
+                    let record = &*data[self.index];
+                    self.index += 1;
+                    Some((&record.0, &record.1))
+                } else {
+                    None
+                }
+            }
+            LogBatchData::BorrowedSlice(data) => {
+                if self.index < data.len() {
+                    let record = &data[self.index];
+                    self.index += 1;
+                    Some((record.0, record.1))
+                } else {
+                    None
+                }
+            }
+        }
     }
 }
 
diff --git a/opentelemetry-sdk/src/logs/log_processor.rs b/opentelemetry-sdk/src/logs/log_processor.rs
index 61823ab854..b6284b87c6 100644
--- a/opentelemetry-sdk/src/logs/log_processor.rs
+++ b/opentelemetry-sdk/src/logs/log_processor.rs
@@ -524,12 +524,7 @@ where
         return LogResult::Ok(());
     }
 
-    let log_vec: Vec<(&LogRecord, &InstrumentationScope)> = batch
-        .iter()
-        .map(|log_data| (&log_data.0, &log_data.1))
-        .collect();
-
-    let export = exporter.export(LogBatch::new(log_vec.as_slice()));
+    let export = exporter.export(LogBatch::new_with_owned_data(batch.as_slice()));
     let export_result = futures_executor::block_on(export);
 
     // Clear the batch vec after exporting