feat: Implement gzip compression

Author: scottgerring

opentelemetry-otlp/src/exporter/http/logs.rs

@@ -14,14 +14,20 @@ impl LogExporter for OtlpHttpClient {
             .clone()
             .ok_or(OTelSdkError::AlreadyShutdown)?;
 
-        let (body, content_type) = self
+        let (body, content_type, content_encoding) = self
             .build_logs_export_body(batch)
             .map_err(OTelSdkError::InternalFailure)?;
 
-        let mut request = http::Request::builder()
+        let mut request_builder = http::Request::builder()
             .method(Method::POST)
             .uri(&self.collector_endpoint)
-            .header(CONTENT_TYPE, content_type)
+            .header(CONTENT_TYPE, content_type);
+            
+        if let Some(encoding) = content_encoding {
+            request_builder = request_builder.header("Content-Encoding", encoding);
+        }
+
+        let mut request = request_builder
             .body(body.into())
             .map_err(|e| OTelSdkError::InternalFailure(e.to_string()))?;
 

opentelemetry-otlp/src/exporter/http/metrics.rs

@@ -19,13 +19,20 @@ impl MetricsClient for OtlpHttpClient {
                 _ => Err(OTelSdkError::AlreadyShutdown),
             })?;
 
-        let (body, content_type) = self.build_metrics_export_body(metrics).ok_or_else(|| {
+        let (body, content_type, content_encoding) = self.build_metrics_export_body(metrics).ok_or_else(|| {
             OTelSdkError::InternalFailure("Failed to serialize metrics".to_string())
         })?;
-        let mut request = http::Request::builder()
+        
+        let mut request_builder = http::Request::builder()
             .method(Method::POST)
             .uri(&self.collector_endpoint)
-            .header(CONTENT_TYPE, content_type)
+            .header(CONTENT_TYPE, content_type);
+            
+        if let Some(encoding) = content_encoding {
+            request_builder = request_builder.header("Content-Encoding", encoding);
+        }
+
+        let mut request = request_builder
             .body(body.into())
             .map_err(|e| OTelSdkError::InternalFailure(format!("{e:?}")))?;
 

opentelemetry-otlp/src/exporter/http/mod.rs

@@ -4,7 +4,6 @@ use super::{
 };
 use crate::{ExportConfig, Protocol, OTEL_EXPORTER_OTLP_ENDPOINT, OTEL_EXPORTER_OTLP_HEADERS};
 use http::{HeaderName, HeaderValue, Uri};
-#[cfg(feature = "http-json")]
 use opentelemetry::otel_debug;
 use opentelemetry_http::HttpClient;
 use opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema;
@@ -278,14 +277,37 @@ pub(crate) struct OtlpHttpClient {
     headers: HashMap<HeaderName, HeaderValue>,
     protocol: Protocol,
     _timeout: Duration,
-    #[allow(dead_code)] // TODO: Remove when compression implementation is added
     compression: Option<crate::Compression>,
     #[allow(dead_code)]
     // <allow dead> would be removed once we support set_resource for metrics and traces.
     resource: opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema,
 }
 
 impl OtlpHttpClient {
+    /// Compress data using gzip if compression is enabled and the feature is available
+    fn compress_body(&self, body: Vec<u8>) -> Result<(Vec<u8>, Option<&'static str>), String> {
+        match self.compression {
+            #[cfg(feature = "gzip-http")]
+            Some(crate::Compression::Gzip) => {
+                use std::io::Write;
+                use flate2::{write::GzEncoder, Compression};
+                
+                let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
+                encoder.write_all(&body).map_err(|e| e.to_string())?;
+                let compressed = encoder.finish().map_err(|e| e.to_string())?;
+                Ok((compressed, Some("gzip")))
+            }
+            #[cfg(not(feature = "gzip-http"))]
+            Some(crate::Compression::Gzip) => {
+                Err("gzip compression requested but gzip-http feature not enabled".to_string())
+            }
+            Some(crate::Compression::Zstd) => {
+                Err("zstd compression not implemented yet".to_string())
+            }
+            None => Ok((body, None)),
+        }
+    }
+
     #[allow(clippy::mutable_key_type)] // http headers are not mutated
     fn new(
         client: Arc<dyn HttpClient>,
@@ -310,59 +332,73 @@ impl OtlpHttpClient {
     fn build_trace_export_body(
         &self,
         spans: Vec<SpanData>,
-    ) -> Result<(Vec<u8>, &'static str), String> {
+    ) -> Result<(Vec<u8>, &'static str, Option<&'static str>), String> {
         use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
         let resource_spans = group_spans_by_resource_and_scope(spans, &self.resource);
 
         let req = ExportTraceServiceRequest { resource_spans };
-        match self.protocol {
+        let (body, content_type) = match self.protocol {
             #[cfg(feature = "http-json")]
             Protocol::HttpJson => match serde_json::to_string_pretty(&req) {
-                Ok(json) => Ok((json.into_bytes(), "application/json")),
-                Err(e) => Err(e.to_string()),
+                Ok(json) => (json.into_bytes(), "application/json"),
+                Err(e) => return Err(e.to_string()),
             },
-            _ => Ok((req.encode_to_vec(), "application/x-protobuf")),
-        }
+            _ => (req.encode_to_vec(), "application/x-protobuf"),
+        };
+        
+        let (compressed_body, content_encoding) = self.compress_body(body)?;
+        Ok((compressed_body, content_type, content_encoding))
     }
 
     #[cfg(feature = "logs")]
     fn build_logs_export_body(
         &self,
         logs: LogBatch<'_>,
-    ) -> Result<(Vec<u8>, &'static str), String> {
+    ) -> Result<(Vec<u8>, &'static str, Option<&'static str>), String> {
         use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
         let resource_logs = group_logs_by_resource_and_scope(logs, &self.resource);
         let req = ExportLogsServiceRequest { resource_logs };
 
-        match self.protocol {
+        let (body, content_type) = match self.protocol {
             #[cfg(feature = "http-json")]
             Protocol::HttpJson => match serde_json::to_string_pretty(&req) {
-                Ok(json) => Ok((json.into(), "application/json")),
-                Err(e) => Err(e.to_string()),
+                Ok(json) => (json.into_bytes(), "application/json"),
+                Err(e) => return Err(e.to_string()),
             },
-            _ => Ok((req.encode_to_vec(), "application/x-protobuf")),
-        }
+            _ => (req.encode_to_vec(), "application/x-protobuf"),
+        };
+        
+        let (compressed_body, content_encoding) = self.compress_body(body)?;
+        Ok((compressed_body, content_type, content_encoding))
     }
 
     #[cfg(feature = "metrics")]
     fn build_metrics_export_body(
         &self,
         metrics: &ResourceMetrics,
-    ) -> Option<(Vec<u8>, &'static str)> {
+    ) -> Option<(Vec<u8>, &'static str, Option<&'static str>)> {
         use opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequest;
 
         let req: ExportMetricsServiceRequest = metrics.into();
 
-        match self.protocol {
+        let (body, content_type) = match self.protocol {
             #[cfg(feature = "http-json")]
             Protocol::HttpJson => match serde_json::to_string_pretty(&req) {
-                Ok(json) => Some((json.into(), "application/json")),
+                Ok(json) => (json.into_bytes(), "application/json"),
                 Err(e) => {
                     otel_debug!(name: "JsonSerializationFaied", error = e.to_string());
-                    None
+                    return None;
                 }
             },
-            _ => Some((req.encode_to_vec(), "application/x-protobuf")),
+            _ => (req.encode_to_vec(), "application/x-protobuf"),
+        };
+        
+        match self.compress_body(body) {
+            Ok((compressed_body, content_encoding)) => Some((compressed_body, content_type, content_encoding)),
+            Err(e) => {
+                otel_debug!(name: "CompressionFailed", error = e);
+                None
+            }
         }
     }
 }

opentelemetry-otlp/src/exporter/http/trace.rs

@@ -22,17 +22,21 @@ impl SpanExporter for OtlpHttpClient {
             Err(err) => return Err(err),
         };
 
-        let (body, content_type) = match self.build_trace_export_body(batch) {
-            Ok(body) => body,
+        let (body, content_type, content_encoding) = match self.build_trace_export_body(batch) {
+            Ok(result) => result,
             Err(e) => return Err(OTelSdkError::InternalFailure(e.to_string())),
         };
 
-        let mut request = match http::Request::builder()
+        let mut request_builder = http::Request::builder()
             .method(Method::POST)
             .uri(&self.collector_endpoint)
-            .header(CONTENT_TYPE, content_type)
-            .body(body.into())
-        {
+            .header(CONTENT_TYPE, content_type);
+            
+        if let Some(encoding) = content_encoding {
+            request_builder = request_builder.header("Content-Encoding", encoding);
+        }
+
+        let mut request = match request_builder.body(body.into()) {
             Ok(req) => req,
             Err(e) => return Err(OTelSdkError::InternalFailure(e.to_string())),
         };