feat(OTLP): add zstd and gzip compression support to HTTP OTLP exporter

opentelemetry-otlp/Cargo.toml

@@ -43,13 +43,18 @@ serde = { workspace = true, features = ["derive"], optional = true }
 thiserror = { workspace = true }
 serde_json = { workspace = true, optional = true }
 
+# compression dependencies
+flate2 = { version = "1.1.2", optional = true }
+zstd = { version = "0.13", optional = true }
+
 [dev-dependencies]
 tokio-stream = { workspace = true, features = ["net"] }
 opentelemetry_sdk = { features = ["trace", "testing"], path = "../opentelemetry-sdk" }
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
 futures-util = { workspace = true }
 temp-env = { workspace = true }
 tonic = { workspace = true, features = ["router", "server"] }
+async-trait = { workspace = true }
 
 [features]
 # telemetry pillars and functions
@@ -67,6 +72,10 @@ default = ["http-proto", "reqwest-blocking-client", "trace", "metrics", "logs",
 grpc-tonic = ["tonic", "prost", "http", "tokio", "opentelemetry-proto/gen-tonic"]
 gzip-tonic = ["tonic/gzip"]
 zstd-tonic = ["tonic/zstd"]
+
+# http compression
+gzip-http = ["flate2"]
+zstd-http = ["zstd"]
 tls = ["tonic/tls-ring"]
 tls-roots = ["tls", "tonic/tls-native-roots"]
 tls-webpki-roots = ["tls", "tonic/tls-webpki-roots"]

opentelemetry-otlp/examples/basic-otlp-http/Cargo.toml

@@ -15,6 +15,8 @@ bench = false
 [features]
 default = ["reqwest-blocking"]
 reqwest-blocking = ["opentelemetry-otlp/reqwest-blocking-client"]
+gzip = ["opentelemetry-otlp/gzip-http"]
+zstd = ["opentelemetry-otlp/zstd-http"]
 
 [dependencies]
 opentelemetry = { path = "../../../opentelemetry" }

opentelemetry-otlp/examples/basic-otlp-http/src/main.rs

@@ -5,6 +5,8 @@ use opentelemetry::{
 };
 use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
 use opentelemetry_otlp::WithExportConfig;
+#[cfg(any(feature = "gzip", feature = "zstd"))]
+use opentelemetry_otlp::{Compression, WithHttpConfig};
 use opentelemetry_otlp::{LogExporter, MetricExporter, Protocol, SpanExporter};
 use opentelemetry_sdk::Resource;
 use opentelemetry_sdk::{
@@ -27,9 +29,29 @@ fn get_resource() -> Resource {
 }
 
 fn init_logs() -> SdkLoggerProvider {
-    let exporter = LogExporter::builder()
+    #[cfg(any(feature = "gzip", feature = "zstd"))]
+    let mut exporter_builder = LogExporter::builder()
         .with_http()
-        .with_protocol(Protocol::HttpBinary)
+        .with_protocol(Protocol::HttpBinary);
+
+    #[cfg(not(any(feature = "gzip", feature = "zstd")))]
+    let exporter_builder = LogExporter::builder()
+        .with_http()
+        .with_protocol(Protocol::HttpBinary);
+
+    #[cfg(feature = "gzip")]
+    {
+        exporter_builder = exporter_builder.with_compression(Compression::Gzip);
+        println!("Using gzip compression for logs");
+    }
+
+    #[cfg(all(feature = "zstd", not(feature = "gzip")))]
+    {
+        exporter_builder = exporter_builder.with_compression(Compression::Zstd);
+        println!("Using zstd compression for logs");
+    }
+
+    let exporter = exporter_builder
         .build()
         .expect("Failed to create log exporter");
 
@@ -40,9 +62,29 @@ fn init_logs() -> SdkLoggerProvider {
 }
 
 fn init_traces() -> SdkTracerProvider {
-    let exporter = SpanExporter::builder()
+    #[cfg(any(feature = "gzip", feature = "zstd"))]
+    let mut exporter_builder = SpanExporter::builder()
+        .with_http()
+        .with_protocol(Protocol::HttpBinary); //can be changed to `Protocol::HttpJson` to export in JSON format
+
+    #[cfg(not(any(feature = "gzip", feature = "zstd")))]
+    let exporter_builder = SpanExporter::builder()
         .with_http()
-        .with_protocol(Protocol::HttpBinary) //can be changed to `Protocol::HttpJson` to export in JSON format
+        .with_protocol(Protocol::HttpBinary); //can be changed to `Protocol::HttpJson` to export in JSON format
+
+    #[cfg(feature = "gzip")]
+    {
+        exporter_builder = exporter_builder.with_compression(Compression::Gzip);
+        println!("Using gzip compression for traces");
+    }
+
+    #[cfg(all(feature = "zstd", not(feature = "gzip")))]
+    {
+        exporter_builder = exporter_builder.with_compression(Compression::Zstd);
+        println!("Using zstd compression for traces");
+    }
+
+    let exporter = exporter_builder
         .build()
         .expect("Failed to create trace exporter");
 
@@ -53,9 +95,29 @@ fn init_traces() -> SdkTracerProvider {
 }
 
 fn init_metrics() -> SdkMeterProvider {
-    let exporter = MetricExporter::builder()
+    #[cfg(any(feature = "gzip", feature = "zstd"))]
+    let mut exporter_builder = MetricExporter::builder()
+        .with_http()
+        .with_protocol(Protocol::HttpBinary); //can be changed to `Protocol::HttpJson` to export in JSON format
+
+    #[cfg(not(any(feature = "gzip", feature = "zstd")))]
+    let exporter_builder = MetricExporter::builder()
         .with_http()
-        .with_protocol(Protocol::HttpBinary) //can be changed to `Protocol::HttpJson` to export in JSON format
+        .with_protocol(Protocol::HttpBinary); //can be changed to `Protocol::HttpJson` to export in JSON format
+
+    #[cfg(feature = "gzip")]
+    {
+        exporter_builder = exporter_builder.with_compression(Compression::Gzip);
+        println!("Using gzip compression for metrics");
+    }
+
+    #[cfg(all(feature = "zstd", not(feature = "gzip")))]
+    {
+        exporter_builder = exporter_builder.with_compression(Compression::Zstd);
+        println!("Using zstd compression for metrics");
+    }
+
+    let exporter = exporter_builder
         .build()
         .expect("Failed to create metric exporter");
 

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,21 @@ impl MetricsClient for OtlpHttpClient {
                 _ => Err(OTelSdkError::AlreadyShutdown),
             })?;
 
-        let (body, content_type) = self.build_metrics_export_body(metrics).ok_or_else(|| {
-            OTelSdkError::InternalFailure("Failed to serialize metrics".to_string())
-        })?;
-        let mut request = http::Request::builder()
+        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_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:?}")))?;