fix: [Geneva uploader] Optimize upload path by using Bytes to eliminate unnecessary clones (specifically for retry scenario)

opentelemetry-exporter-geneva/geneva-uploader/Cargo.toml

@@ -12,12 +12,13 @@ license = "Apache-2.0"
 [dependencies]
 opentelemetry-proto = {workspace = true, default-features = false, features = ["logs", "trace", "gen-tonic-messages"]}
 base64 = "0.22"
+bytes = "1.5"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0", features = ["raw_value"] }
 uuid = { version = "1.0", features = ["v4"] }
 # TODO - support both native-tls and rustls
 reqwest = { version = "0.12", features = ["native-tls", "native-tls-alpn"], default-features = false}
-native-tls = "0.2" 
+native-tls = "0.2"
 thiserror = "2.0"
 chrono = "0.4"
 url = "2.2"

opentelemetry-exporter-geneva/geneva-uploader/src/client.rs

@@ -4,6 +4,7 @@ use crate::config_service::client::{AuthMethod, GenevaConfigClient, GenevaConfig
 // ManagedIdentitySelector removed; no re-export needed.
 use crate::ingestion_service::uploader::{GenevaUploader, GenevaUploaderConfig};
 use crate::payload_encoder::otlp_encoder::OtlpEncoder;
+use bytes::Bytes;
 use opentelemetry_proto::tonic::logs::v1::ResourceLogs;
 use opentelemetry_proto::tonic::trace::v1::ResourceSpans;
 use std::sync::Arc;
@@ -13,7 +14,7 @@ use std::sync::Arc;
 #[derive(Debug, Clone)]
 pub struct EncodedBatch {
     pub event_name: String,
-    pub data: Vec<u8>,
+    pub data: Bytes,
     pub metadata: crate::payload_encoder::central_blob::BatchMetadata,
 }
 

opentelemetry-exporter-geneva/geneva-uploader/src/ingestion_service/mod.rs

@@ -9,12 +9,13 @@ mod tests {
             AuthMethod, GenevaConfigClient, GenevaConfigClientConfig, GenevaUploader,
             GenevaUploaderConfig,
         };
+        use bytes::Bytes;
         use std::env;
         use std::fs;
         use std::sync::Arc;
 
         pub struct TestUploadContext {
-            pub data: Vec<u8>,
+            pub data: Bytes,
             pub uploader: GenevaUploader,
             pub event_name: String,
         }
@@ -78,7 +79,7 @@ mod tests {
             let event_name = "Log".to_string();
 
             TestUploadContext {
-                data,
+                data: Bytes::from(data),
                 uploader,
                 event_name,
             }

opentelemetry-exporter-geneva/geneva-uploader/src/ingestion_service/uploader.rs

@@ -1,5 +1,6 @@
 use crate::config_service::client::{GenevaConfigClient, GenevaConfigClientError};
 use crate::payload_encoder::central_blob::BatchMetadata;
+use bytes::Bytes;
 use reqwest::{header, Client};
 use serde::Deserialize;
 use serde_json::Value;
@@ -210,7 +211,7 @@ impl GenevaUploader {
     #[allow(dead_code)]
     pub(crate) async fn upload(
         &self,
-        data: Vec<u8>,
+        data: Bytes,
         event_name: &str,
         metadata: &BatchMetadata,
     ) -> Result<IngestionResponse> {

opentelemetry-exporter-geneva/geneva-uploader/src/payload_encoder/otlp_encoder.rs

@@ -4,6 +4,7 @@ use crate::payload_encoder::central_blob::{
     BatchMetadata, CentralBlob, CentralEventEntry, CentralSchemaEntry,
 };
 use crate::payload_encoder::lz4_chunked_compression::lz4_chunked_compression;
+use bytes::Bytes;
 use chrono::{TimeZone, Utc};
 use opentelemetry_proto::tonic::common::v1::any_value::Value;
 use opentelemetry_proto::tonic::logs::v1::LogRecord;
@@ -164,7 +165,7 @@ impl OtlpEncoder {
                 .map_err(|e| format!("compression failed: {e}"))?;
             blobs.push(EncodedBatch {
                 event_name: batch_event_name.to_string(),
-                data: compressed,
+                data: Bytes::from(compressed),
                 metadata: batch_data.metadata,
             });
         }
@@ -280,7 +281,7 @@ impl OtlpEncoder {
 
         Ok(vec![EncodedBatch {
             event_name: EVENT_NAME.to_string(),
-            data: compressed,
+            data: Bytes::from(compressed),
             metadata: batch_metadata,
         }])
     }