feat: [Geneva Exporter] - Add Serializer

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

@@ -7,6 +7,5 @@ rust-version = "1.75.0"
 
 [dependencies]
 
-
 [lints]
-workspace = true
+workspace = true

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

@@ -17,10 +17,13 @@ native-tls = "0.2"
 thiserror = "2.0"
 chrono = "0.4"
 url = "2.2"
+md5 = "0.7.0"
+hex = "0.4"
 lz4_flex = { version = ">=0.11.0, <0.11.4", features = ["safe-encode"], default-features = false }
 
 [features]
 self_signed_certs = [] # Empty by default for security
+mock_auth = [] # Disabled by default. Not to be enabled in the prod release. 
 default = ["self_signed_certs"] # TODO - remove this feature before release
 
 [dev-dependencies]
@@ -34,4 +37,4 @@ num_cpus = "1.16"
 lz4_flex = { version = "0.11" }
 
 [lints]
-workspace = true
+workspace = true

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

@@ -0,0 +1,106 @@
+//! High-level GenevaClient for user code. Wraps config_service and ingestion_service.
+
+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 opentelemetry_proto::tonic::logs::v1::ResourceLogs;
+use std::sync::Arc;
+
+/// Configuration for GenevaClient (user-facing)
+#[derive(Clone, Debug)]
+pub struct GenevaClientConfig {
+    pub endpoint: String,
+    pub environment: String,
+    pub account: String,
+    pub namespace: String,
+    pub region: String,
+    pub config_major_version: u32,
+    pub auth_method: AuthMethod,
+    pub tenant: String,
+    pub role_name: String,
+    pub role_instance: String,
+    // Add event name/version here if constant, or per-upload if you want them per call.
+}
+
+/// Main user-facing client for Geneva ingestion.
+#[derive(Clone)]
+pub struct GenevaClient {
+    uploader: Arc<GenevaUploader>,
+    encoder: OtlpEncoder,
+    metadata: String,
+}
+
+impl GenevaClient {
+    /// Construct a new client with minimal configuration. Fetches and caches ingestion info as needed.
+    pub async fn new(cfg: GenevaClientConfig) -> Result<Self, String> {
+        // Build config client config
+        let config_client_config = GenevaConfigClientConfig {
+            endpoint: cfg.endpoint,
+            environment: cfg.environment.clone(),
+            account: cfg.account,
+            namespace: cfg.namespace.clone(),
+            region: cfg.region,
+            config_major_version: cfg.config_major_version,
+            auth_method: cfg.auth_method,
+        };
+        let config_client = Arc::new(
+            GenevaConfigClient::new(config_client_config)
+                .map_err(|e| format!("GenevaConfigClient init failed: {e}"))?,
+        );
+
+        let source_identity = format!(
+            "Tenant={}/Role={}/RoleInstance={}",
+            cfg.tenant, cfg.role_name, cfg.role_instance
+        );
+
+        let schema_ids =
+            "c1ce0ecea020359624c493bbe97f9e80;0da22cabbee419e000541a5eda732eb3".to_string(); // TODO - find the actual value to be populated
+
+        // Uploader config
+        let uploader_config = GenevaUploaderConfig {
+            namespace: cfg.namespace.clone(),
+            source_identity,
+            environment: cfg.environment,
+            schema_ids,
+        };
+
+        let uploader = GenevaUploader::from_config_client(config_client, uploader_config)
+            .await
+            .map_err(|e| format!("GenevaUploader init failed: {e}"))?;
+        let metadata = format!(
+            "namespace={}/eventVersion={}/tenant={}/role={}/roleinstance={}",
+            cfg.namespace,
+            "Ver1v0", // You can replace this with a cfg field if version should be dynamic
+            cfg.tenant,
+            cfg.role_name,
+            cfg.role_instance,
+        );
+        Ok(Self {
+            uploader: Arc::new(uploader),
+            encoder: OtlpEncoder::new(),
+            metadata,
+        })
+    }
+
+    /// Upload OTLP logs (as ResourceLogs).
+    pub async fn upload_logs(&self, logs: &[ResourceLogs]) -> Result<(), String> {
+        let log_iter = logs
+            .iter()
+            .flat_map(|resource_log| resource_log.scope_logs.iter())
+            .flat_map(|scope_log| scope_log.log_records.iter());
+        let blobs = self.encoder.encode_log_batch(log_iter, &self.metadata);
+        for (_schema_id, event_name, encoded_blob, _row_count) in blobs {
+            // TODO - log encoded_blob for debugging
+            let compressed_blob = lz4_chunked_compression(&encoded_blob)
+                .map_err(|e| format!("LZ4 compression failed: {e}"))?;
+            // TODO - log compressed_blob for debugging
+            let event_version = "Ver2v0"; // TODO - find the actual value to be populated
+            self.uploader
+                .upload(compressed_blob, &event_name, event_version)
+                .await
+                .map_err(|e| format!("Geneva upload failed: {e}"))?;
+        }
+        Ok(())
+    }
+}

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

@@ -46,7 +46,7 @@
 /// ```
 #[allow(dead_code)]
 #[derive(Clone, Debug)]
-pub(crate) enum AuthMethod {
+pub enum AuthMethod {
     /// Certificate-based authentication
     ///
     /// # Arguments
@@ -57,6 +57,8 @@
     ///
     /// Note(TODO): This is not yet implemented.
     ManagedIdentity,
+    #[cfg(feature = "mock_auth")]
+    MockAuth, // No authentication, used for testing purposes
 }
 
 #[derive(Debug, Error)]
@@ -249,6 +251,12 @@
                     "Managed Identity authentication is not implemented yet".into(),
                 ));
             }
+            #[cfg(feature = "mock_auth")]
+            AuthMethod::MockAuth => {
+                // Mock authentication for testing purposes, no actual auth needed
+                // Just use the default client builder
+                eprintln!("WARNING: Using MockAuth for GenevaConfigClient. This should only be used in tests!");
+            }
         }
 
         let agent_identity = "GenevaUploader";
@@ -432,7 +440,7 @@
             .send()
             .await
             .map_err(GenevaConfigClientError::Http)?;
-
+        // Check if the response is successful
         let status = response.status();
         let body = response.text().await?;
 

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

@@ -139,7 +139,6 @@ impl GenevaUploader {
         // TODO - Maintain this as url-encoded in config service to avoid conversion here
         let encoded_monitoring_endpoint: String =
             byte_serialize(monitoring_endpoint.as_bytes()).collect();
-
         let encoded_source_identity: String =
             byte_serialize(self.config.source_identity.as_bytes()).collect();
 
@@ -195,7 +194,6 @@ impl GenevaUploader {
             auth_info.endpoint.trim_end_matches('/'),
             upload_uri
         );
-
         // Send the upload request
         let response = self
             .http_client
@@ -208,13 +206,13 @@ impl GenevaUploader {
             .send()
             .await
             .map_err(GenevaUploaderError::Http)?;
-
         let status = response.status();
         let body = response.text().await.map_err(GenevaUploaderError::Http)?;
 
         if status == reqwest::StatusCode::ACCEPTED {
             let ingest_response: IngestionResponse =
                 serde_json::from_str(&body).map_err(GenevaUploaderError::SerdeJson)?;
+
             Ok(ingest_response)
         } else {
             Err(GenevaUploaderError::UploadFailed {

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

@@ -1,18 +1,18 @@
 mod config_service;
-pub mod ingestion_service;
+mod ingestion_service;
 mod payload_encoder;
 
-mod uploader;
+pub mod client;
 
 #[allow(unused_imports)]
 pub(crate) use config_service::client::{
-    AuthMethod, GenevaConfigClient, GenevaConfigClientConfig, GenevaConfigClientError,
-    IngestionGatewayInfo,
+    GenevaConfigClient, GenevaConfigClientConfig, GenevaConfigClientError, IngestionGatewayInfo,
 };
 
 #[allow(unused_imports)]
 pub(crate) use ingestion_service::uploader::{
     GenevaUploader, GenevaUploaderConfig, GenevaUploaderError, IngestionResponse, Result,
 };
 
-pub use uploader::{create_uploader, GenevaUploader as Uploader};
+pub use client::{GenevaClient, GenevaClientConfig};
+pub use config_service::client::AuthMethod;