feat: [Geneva Exporter] Split Client API into batching and uploading (#398)

Author: lalitb

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

@@ -178,8 +178,9 @@ mod benchmarks {
                         .collect();
 
                     b.iter(|| {
-                        let res =
-                            encoder.encode_log_batch(black_box(logs.iter()), black_box(metadata));
+                        let res = encoder
+                            .encode_log_batch(black_box(logs.iter()), black_box(metadata))
+                            .unwrap();
                         black_box(res); // double sure the return value is generated
                     });
                 },
@@ -207,7 +208,9 @@ mod benchmarks {
 
                     b.iter(|| {
                         let res = black_box(
-                            encoder.encode_log_batch(black_box(logs.iter()), black_box(metadata)),
+                            encoder
+                                .encode_log_batch(black_box(logs.iter()), black_box(metadata))
+                                .unwrap(),
                         );
                         black_box(res); // double sure the return value is generated
                     });
@@ -232,7 +235,9 @@ mod benchmarks {
 
             b.iter(|| {
                 let res = black_box(
-                    encoder.encode_log_batch(black_box(logs.iter()), black_box(metadata)),
+                    encoder
+                        .encode_log_batch(black_box(logs.iter()), black_box(metadata))
+                        .unwrap(),
                 );
                 black_box(res);
             });

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

@@ -2,12 +2,19 @@
 
 use crate::config_service::client::{AuthMethod, GenevaConfigClient, GenevaConfigClientConfig};
 use crate::ingestion_service::uploader::{GenevaUploader, GenevaUploaderConfig};
-use crate::payload_encoder::lz4_chunked_compression::lz4_chunked_compression;
 use crate::payload_encoder::otlp_encoder::OtlpEncoder;
-use futures::stream::{self, StreamExt};
 use opentelemetry_proto::tonic::logs::v1::ResourceLogs;
 use std::sync::Arc;
 
+/// Public batch type (already LZ4 chunked compressed).
+/// Produced by `OtlpEncoder::encode_log_batch` and returned to callers.
+#[derive(Debug, Clone)]
+pub struct EncodedBatch {
+    pub event_name: String,
+    pub data: Vec<u8>,
+    pub metadata: crate::payload_encoder::central_blob::BatchMetadata,
+}
+
 /// Configuration for GenevaClient (user-facing)
 #[derive(Clone, Debug)]
 pub struct GenevaClientConfig {
@@ -21,8 +28,6 @@ pub struct GenevaClientConfig {
     pub tenant: String,
     pub role_name: String,
     pub role_instance: String,
-    /// Maximum number of concurrent uploads. If None, defaults to number of CPU cores.
-    pub max_concurrent_uploads: Option<usize>,
     // Add event name/version here if constant, or per-upload if you want them per call.
 }
 
@@ -32,7 +37,6 @@ pub struct GenevaClient {
     uploader: Arc<GenevaUploader>,
     encoder: OtlpEncoder,
     metadata: String,
-    max_concurrent_uploads: usize,
 }
 
 impl GenevaClient {
@@ -78,57 +82,35 @@ impl GenevaClient {
         let uploader = GenevaUploader::from_config_client(config_client, uploader_config)
             .await
             .map_err(|e| format!("GenevaUploader init failed: {e}"))?;
-        let max_concurrent_uploads = cfg.max_concurrent_uploads.unwrap_or_else(|| {
-            // TODO - Use a more sophisticated method to determine concurrency if needed
-            // currently using number of CPU cores
-            std::thread::available_parallelism()
-                .map(|p| p.get())
-                .unwrap_or(4)
-        });
         Ok(Self {
             uploader: Arc::new(uploader),
             encoder: OtlpEncoder::new(),
             metadata,
-            max_concurrent_uploads,
         })
     }
 
-    /// Upload OTLP logs (as ResourceLogs).
-    pub async fn upload_logs(&self, logs: &[ResourceLogs]) -> Result<(), String> {
+    /// Encode OTLP logs into LZ4 chunked compressed batches.
+    pub fn encode_and_compress_logs(
+        &self,
+        logs: &[ResourceLogs],
+    ) -> Result<Vec<EncodedBatch>, String> {
         let log_iter = logs
             .iter()
             .flat_map(|resource_log| resource_log.scope_logs.iter())
             .flat_map(|scope_log| scope_log.log_records.iter());
-        // TODO: Investigate using tokio::spawn_blocking for event encoding to avoid blocking
-        // the async executor thread for CPU-intensive work.
-        let blobs = self.encoder.encode_log_batch(log_iter, &self.metadata);
 
-        // create an iterator that yields futures for each upload
-        let upload_futures = blobs.into_iter().map(|batch| {
-            async move {
-                // TODO: Investigate using tokio::spawn_blocking for LZ4 compression to avoid blocking
-                // the async executor thread for CPU-intensive work.
-                let compressed_blob = lz4_chunked_compression(&batch.data).map_err(|e| {
-                    format!("LZ4 compression failed: {e} Event: {}", batch.event_name)
-                })?;
-                self.uploader
-                    .upload(compressed_blob, &batch.event_name, &batch.metadata)
-                    .await
-                    .map(|_| ())
-                    .map_err(|e| format!("Geneva upload failed: {e} Event: {}", batch.event_name))
-            }
-        });
-        // Execute uploads concurrently with configurable concurrency
-        let errors: Vec<String> = stream::iter(upload_futures)
-            .buffer_unordered(self.max_concurrent_uploads)
-            .filter_map(|result| async move { result.err() })
-            .collect()
-            .await;
+        self.encoder
+            .encode_log_batch(log_iter, &self.metadata)
+            .map_err(|e| format!("Compression failed: {e}"))
+    }
 
-        // Return error if any uploads failed
-        if !errors.is_empty() {
-            return Err(format!("Upload failures: {}", errors.join("; ")));
-        }
-        Ok(())
+    /// Upload a single compressed batch.
+    /// This allows for granular control over uploads, including custom retry logic for individual batches.
+    pub async fn upload_batch(&self, batch: &EncodedBatch) -> Result<(), String> {
+        self.uploader
+            .upload(batch.data.clone(), &batch.event_name, &batch.metadata)
+            .await
+            .map(|_| ())
+            .map_err(|e| format!("Geneva upload failed: {e} Event: {}", batch.event_name))
     }
 }

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

@@ -88,12 +88,13 @@ impl From<reqwest::Error> for GenevaUploaderError {
 
 pub(crate) type Result<T> = std::result::Result<T, GenevaUploaderError>;
 
-#[allow(dead_code)]
 /// Response from the ingestion API when submitting data
 #[derive(Debug, Clone, Deserialize)]
 pub(crate) struct IngestionResponse {
+    #[allow(dead_code)]
     pub(crate) ticket: String,
     #[serde(flatten)]
+    #[allow(dead_code)]
     pub(crate) extra: HashMap<String, Value>,
 }
 

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

@@ -1,6 +1,6 @@
 mod config_service;
 mod ingestion_service;
-pub mod payload_encoder;
+mod payload_encoder;
 
 pub mod client;
 
@@ -14,8 +14,9 @@ pub(crate) use config_service::client::{
 
 #[allow(unused_imports)]
 pub(crate) use ingestion_service::uploader::{
-    GenevaUploader, GenevaUploaderConfig, GenevaUploaderError, IngestionResponse, Result,
+    GenevaUploader, GenevaUploaderConfig, GenevaUploaderError, Result,
 };
 
+pub use client::EncodedBatch;
 pub use client::{GenevaClient, GenevaClientConfig};
 pub use config_service::client::AuthMethod;

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

@@ -52,17 +52,6 @@ impl BatchMetadata {
     }
 }
 
-/// Represents an encoded batch with all necessary metadata
-#[derive(Debug, Clone)]
-pub(crate) struct EncodedBatch {
-    /// The event name for this batch
-    pub(crate) event_name: String,
-    /// The encoded binary data
-    pub(crate) data: Vec<u8>,
-    /// Batch metadata containing timestamps and schema information
-    pub(crate) metadata: BatchMetadata,
-}
-
 /// Helper to encode UTF-8 Rust str to UTF-16LE bytes
 /// TODO - consider avoiding temporary allocation, by passing a mutable buffer
 #[allow(dead_code)]

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

@@ -1,7 +1,9 @@
+use crate::client::EncodedBatch;
 use crate::payload_encoder::bond_encoder::{BondDataType, BondEncodedSchema, BondWriter, FieldDef};
 use crate::payload_encoder::central_blob::{
-    BatchMetadata, CentralBlob, CentralEventEntry, CentralSchemaEntry, EncodedBatch,
+    BatchMetadata, CentralBlob, CentralEventEntry, CentralSchemaEntry,
 };
+use crate::payload_encoder::lz4_chunked_compression::lz4_chunked_compression;
 use chrono::{TimeZone, Utc};
 use opentelemetry_proto::tonic::common::v1::any_value::Value;
 use opentelemetry_proto::tonic::logs::v1::LogRecord;
@@ -29,8 +31,14 @@ impl OtlpEncoder {
         OtlpEncoder {}
     }
 
-    /// Encode a batch of logs into a vector of (event_name, bytes, schema_ids, start_time_nanos, end_time_nanos)
-    pub(crate) fn encode_log_batch<'a, I>(&self, logs: I, metadata: &str) -> Vec<EncodedBatch>
+    /// Encode a batch of logs into a vector of (event_name, compressed_bytes, schema_ids, start_time_nanos, end_time_nanos)
+    /// The returned `data` field contains LZ4 chunked compressed bytes.
+    /// On compression failure, the error is returned (no logging, no fallback).
+    pub(crate) fn encode_log_batch<'a, I>(
+        &self,
+        logs: I,
+        metadata: &str,
+    ) -> Result<Vec<EncodedBatch>, String>
     where
         I: IntoIterator<Item = &'a opentelemetry_proto::tonic::logs::v1::LogRecord>,
     {
@@ -143,14 +151,16 @@ impl OtlpEncoder {
                 schemas: batch_data.schemas,
                 events: batch_data.events,
             };
-            let bytes = blob.to_bytes();
+            let uncompressed = blob.to_bytes();
+            let compressed = lz4_chunked_compression(&uncompressed)
+                .map_err(|e| format!("compression failed: {e}"))?;
             blobs.push(EncodedBatch {
                 event_name: batch_event_name,
-                data: bytes,
+                data: compressed,
                 metadata: batch_data.metadata,
             });
         }
-        blobs
+        Ok(blobs)
     }
 
     /// Determine fields and calculate schema ID in a single pass for optimal performance
@@ -391,7 +401,7 @@ mod tests {
         });
 
         let metadata = "namespace=testNamespace/eventVersion=Ver1v0";
-        let result = encoder.encode_log_batch([log].iter(), metadata);
+        let result = encoder.encode_log_batch([log].iter(), metadata).unwrap();
 
         assert!(!result.is_empty());
     }
@@ -438,7 +448,9 @@ mod tests {
         let metadata = "namespace=test";
 
         // Encode multiple log records with different schema structures but same event_name
-        let result = encoder.encode_log_batch([log1, log2, log3].iter(), metadata);
+        let result = encoder
+            .encode_log_batch([log1, log2, log3].iter(), metadata)
+            .unwrap();
 
         // Should create one batch (same event_name = "user_action")
         assert_eq!(result.len(), 1);
@@ -495,7 +507,7 @@ mod tests {
             ..Default::default()
         };
 
-        let result = encoder.encode_log_batch([log].iter(), "test");
+        let result = encoder.encode_log_batch([log].iter(), "test").unwrap();
 
         assert_eq!(result.len(), 1);
         assert_eq!(result[0].event_name, "test_event");
@@ -535,7 +547,9 @@ mod tests {
             }),
         });
 
-        let result = encoder.encode_log_batch([log1, log2, log3].iter(), "test");
+        let result = encoder
+            .encode_log_batch([log1, log2, log3].iter(), "test")
+            .unwrap();
 
         // All should be in one batch with same event_name
         assert_eq!(result.len(), 1);
@@ -561,7 +575,9 @@ mod tests {
             ..Default::default()
         };
 
-        let result = encoder.encode_log_batch([log1, log2].iter(), "test");
+        let result = encoder
+            .encode_log_batch([log1, log2].iter(), "test")
+            .unwrap();
 
         // Should create 2 separate batches
         assert_eq!(result.len(), 2);
@@ -584,7 +600,7 @@ mod tests {
             ..Default::default()
         };
 
-        let result = encoder.encode_log_batch([log].iter(), "test");
+        let result = encoder.encode_log_batch([log].iter(), "test").unwrap();
 
         assert_eq!(result.len(), 1);
         assert_eq!(result[0].event_name, "Log"); // Should default to "Log"
@@ -630,7 +646,9 @@ mod tests {
             }),
         });
 
-        let result = encoder.encode_log_batch([log1, log2, log3, log4].iter(), "test");
+        let result = encoder
+            .encode_log_batch([log1, log2, log3, log4].iter(), "test")
+            .unwrap();
 
         // Should create 3 batches: "user_action", "system_alert", "Log"
         assert_eq!(result.len(), 3);

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

@@ -9,6 +9,7 @@ rust-version = "1.75.0"
 opentelemetry_sdk = {workspace = true, default-features = false, features = ["logs"]}
 opentelemetry-proto = {workspace = true, default-features = false, features = ["logs"]}
 geneva-uploader = {path = "../geneva-uploader/", version = "0.1.0"}
+futures = "0.3"
 
 [dev-dependencies]
 opentelemetry-appender-tracing = {workspace = true}
@@ -19,4 +20,4 @@ tracing-subscriber = { version = "0.3.0", default-features = false, features = [
 tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
 
 [lints]
-workspace = true
+workspace = true

opentelemetry-exporter-geneva/opentelemetry-exporter-geneva/examples/basic.rs

@@ -62,7 +62,6 @@ async fn main() {
         tenant,
         role_name,
         role_instance,
-        max_concurrent_uploads: None, // Use default
     };
 
     let geneva_client = GenevaClient::new(config)