feat(geneva uploader): Add rowCount parameter to ingestion URL (#484)

Author: lalitb

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

@@ -16,6 +16,7 @@ pub struct EncodedBatch {
     pub event_name: String,
     pub data: Vec<u8>,
     pub metadata: crate::payload_encoder::central_blob::BatchMetadata,
+    pub row_count: usize,
 }
 
 /// Configuration for GenevaClient (user-facing)
@@ -212,7 +213,12 @@ impl GenevaClient {
         );
 
         self.uploader
-            .upload(batch.data.clone(), &batch.event_name, &batch.metadata)
+            .upload(
+                batch.data.clone(),
+                &batch.event_name,
+                &batch.metadata,
+                batch.row_count,
+            )
             .await
             .map(|_| {
                 debug!(

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

@@ -128,7 +128,7 @@ mod tests {
 
         let response = ctx
             .uploader
-            .upload(ctx.data, &ctx.event_name, &metadata)
+            .upload(ctx.data, &ctx.event_name, &metadata, 1)
             .await
             .expect("Upload failed");
 
@@ -195,7 +195,7 @@ mod tests {
 
         let _ = ctx
             .uploader
-            .upload(ctx.data.clone(), &ctx.event_name, &warmup_metadata)
+            .upload(ctx.data.clone(), &ctx.event_name, &warmup_metadata, 1)
             .await
             .expect("Warm-up upload failed");
         let warmup_elapsed = start_warmup.elapsed();
@@ -221,7 +221,7 @@ mod tests {
                 };
 
                 let resp = uploader
-                    .upload(data, &event_name, &metadata)
+                    .upload(data, &event_name, &metadata, 1)
                     .await
                     .unwrap_or_else(|_| panic!("Upload {i} failed"));
                 let elapsed = start.elapsed();

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

@@ -163,6 +163,7 @@ impl GenevaUploader {
         data_size: usize,
         event_name: &str,
         metadata: &BatchMetadata,
+        row_count: usize,
     ) -> Result<String> {
         // Get already formatted schema IDs and format timestamps using BatchMetadata methods
         let schema_ids = &metadata.schema_ids;
@@ -181,7 +182,7 @@ impl GenevaUploader {
 
         // Create the query string
         let mut query = String::with_capacity(512); // Preallocate enough space for the query string (decided based on expected size)
-        write!(&mut query, "api/v1/ingestion/ingest?endpoint={}&moniker={}&namespace={}&event={}&version={}&sourceUniqueId={}&sourceIdentity={}&startTime={}&endTime={}&format=centralbond/lz4hc&dataSize={}&minLevel={}&schemaIds={}",
+        write!(&mut query, "api/v1/ingestion/ingest?endpoint={}&moniker={}&namespace={}&event={}&version={}&sourceUniqueId={}&sourceIdentity={}&startTime={}&endTime={}&format=centralbond/lz4hc&dataSize={}&minLevel={}&schemaIds={}&rowCount={}",
             encoded_monitoring_endpoint,
             moniker,
             self.config.namespace,
@@ -193,7 +194,8 @@ impl GenevaUploader {
             end_time_str,
             data_size,
             2,
-            schema_ids
+            schema_ids,
+            row_count
         ).map_err(|e| GenevaUploaderError::InternalError(format!("Failed to write query string: {e}")))?;
         Ok(query)
     }
@@ -205,6 +207,7 @@ impl GenevaUploader {
     /// * `event_name` - Name of the event
     /// * `event_version` - Version of the event
     /// * `metadata` - Batch metadata containing timestamps and schema information
+    /// * `row_count` - Number of rows/events in the batch
     ///
     /// # Returns
     /// * `Result<IngestionResponse>` - The response containing the ticket ID or an error
@@ -214,6 +217,7 @@ impl GenevaUploader {
         data: Vec<u8>,
         event_name: &str,
         metadata: &BatchMetadata,
+        row_count: usize,
     ) -> Result<IngestionResponse> {
         debug!(
             name: "uploader.upload",
@@ -233,6 +237,7 @@ impl GenevaUploader {
             data_size,
             event_name,
             metadata,
+            row_count,
         )?;
         let full_url = format!(
             "{}/{}",

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

@@ -190,6 +190,7 @@ impl OtlpEncoder {
                 event_name: batch_event_name.to_string(),
                 data: compressed,
                 metadata: batch_data.metadata,
+                row_count: events_count,
             });
         }
         Ok(blobs)
@@ -292,7 +293,6 @@ impl OtlpEncoder {
 
         let schemas_count = schemas.len();
         let events_count = events.len();
-
         let blob = CentralBlob {
             version: 1,
             format: 2,
@@ -327,6 +327,7 @@ impl OtlpEncoder {
             event_name: EVENT_NAME.to_string(),
             data: compressed,
             metadata: batch_metadata,
+            row_count: events_count,
         }])
     }
 
@@ -1282,4 +1283,59 @@ mod tests {
         assert!(single_result.starts_with('['));
         assert!(single_result.ends_with(']'));
     }
+
+    #[test]
+    fn test_row_count_in_encoded_batch() {
+        let encoder = OtlpEncoder::new();
+
+        // Test with logs
+        let logs = [
+            LogRecord {
+                observed_time_unix_nano: 1_700_000_000_000_000_000,
+                event_name: "test_event".to_string(),
+                severity_number: 9,
+                ..Default::default()
+            },
+            LogRecord {
+                observed_time_unix_nano: 1_700_000_001_000_000_000,
+                event_name: "test_event".to_string(),
+                severity_number: 10,
+                ..Default::default()
+            },
+            LogRecord {
+                observed_time_unix_nano: 1_700_000_002_000_000_000,
+                event_name: "test_event".to_string(),
+                severity_number: 11,
+                ..Default::default()
+            },
+        ];
+
+        let result = encoder
+            .encode_log_batch(logs.iter(), "namespace=test")
+            .unwrap();
+
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[0].row_count, 3);
+
+        // Test with spans
+        let spans = [
+            Span {
+                start_time_unix_nano: 1_700_000_000_000_000_000,
+                end_time_unix_nano: 1_700_000_001_000_000_000,
+                ..Default::default()
+            },
+            Span {
+                start_time_unix_nano: 1_700_000_002_000_000_000,
+                end_time_unix_nano: 1_700_000_003_000_000_000,
+                ..Default::default()
+            },
+        ];
+
+        let span_result = encoder
+            .encode_span_batch(spans.iter(), "namespace=test")
+            .unwrap();
+
+        assert_eq!(span_result.len(), 1);
+        assert_eq!(span_result[0].row_count, 2);
+    }
 }