fix(opentelemetry source): fix HTTP not decompressing payloads (#24068)

* fix(opentelemetry source): fix HTTP not decompressing payloads

* update test modules

* changelog

* handle and emit bad request errors

* avoid double counting compression type error

* revert renaming operation

* clippy fix

Author: pront

changelog.d/opentelemetry_source_http_decompression.fix.md

@@ -0,0 +1,4 @@
+Fixed a `opentelemetry` source bug where HTTP payloads were not decompressed according to the request headers.
+This only applied when `use_otlp_decoding` (recently added) was set to `true`.
+
+authors: pront

src/sources/opentelemetry/http.rs

@@ -34,7 +34,7 @@ use crate::{
     common::http::ErrorMessage,
     event::Event,
     http::{KeepaliveConfig, MaxConnectionAgeLayer, build_http_trace_layer},
-    internal_events::{EventsReceived, StreamClosedError},
+    internal_events::{EventsReceived, HttpBadRequest, StreamClosedError},
     shutdown::ShutdownSignal,
     sources::{
         http_server::HttpConfigParamKind,
@@ -144,6 +144,26 @@ fn enrich_events(
     );
 }
 
+fn emit_decode_error(error: impl std::fmt::Display) -> ErrorMessage {
+    let message = format!("Could not decode request: {error}");
+    emit!(HttpBadRequest::new(
+        StatusCode::BAD_REQUEST.as_u16(),
+        &message
+    ));
+    ErrorMessage::new(StatusCode::BAD_REQUEST, message)
+}
+
+fn parse_with_deserializer(
+    deserializer: &OtlpDeserializer,
+    body: Bytes,
+    log_namespace: LogNamespace,
+) -> Result<Vec<Event>, ErrorMessage> {
+    deserializer
+        .parse(body, log_namespace)
+        .map(|r| r.into_vec())
+        .map_err(emit_decode_error)
+}
+
 fn build_ingest_filter<Resp, F>(
     telemetry_type: &'static str,
     acknowledgements: bool,
@@ -194,21 +214,29 @@ fn build_warp_log_filter(
     deserializer: Option<OtlpDeserializer>,
 ) -> BoxedFilter<(Response,)> {
     let make_events = move |encoding_header: Option<String>, headers: HeaderMap, body: Bytes| {
-        if let Some(d) = deserializer.as_ref() {
-            d.parse(body, log_namespace)
-                .map(|r| r.into_vec())
-                .map_err(|e| ErrorMessage::new(StatusCode::BAD_REQUEST, e.to_string()))
-        } else {
-            decode(encoding_header.as_deref(), body)
-                .and_then(|body| {
-                    bytes_received.emit(ByteSize(body.len()));
-                    decode_log_body(body, log_namespace, &events_received)
-                })
-                .map(|mut events| {
-                    enrich_events(&mut events, &headers_cfg, &headers, log_namespace);
-                    events
-                })
-        }
+        decode(encoding_header.as_deref(), body)
+            .inspect_err(|err| {
+                // Other status codes are already handled by `sources::util::decode` (tech debt).
+                if err.status_code() == StatusCode::UNSUPPORTED_MEDIA_TYPE {
+                    emit!(HttpBadRequest::new(
+                        err.status_code().as_u16(),
+                        err.message()
+                    ));
+                }
+            })
+            .and_then(|decoded_body| {
+                bytes_received.emit(ByteSize(decoded_body.len()));
+                if let Some(d) = deserializer.as_ref() {
+                    parse_with_deserializer(d, decoded_body, log_namespace)
+                } else {
+                    decode_log_body(decoded_body, log_namespace, &events_received).map(
+                        |mut events| {
+                            enrich_events(&mut events, &headers_cfg, &headers, log_namespace);
+                            events
+                        },
+                    )
+                }
+            })
     };
 
     build_ingest_filter::<ExportLogsServiceResponse, _>(
@@ -226,16 +254,24 @@ fn build_warp_metrics_filter(
     deserializer: Option<OtlpDeserializer>,
 ) -> BoxedFilter<(Response,)> {
     let make_events = move |encoding_header: Option<String>, _headers: HeaderMap, body: Bytes| {
-        if let Some(d) = deserializer.as_ref() {
-            d.parse(body, LogNamespace::default())
-                .map(|r| r.into_vec())
-                .map_err(|e| ErrorMessage::new(StatusCode::BAD_REQUEST, e.to_string()))
-        } else {
-            decode(encoding_header.as_deref(), body).and_then(|body| {
-                bytes_received.emit(ByteSize(body.len()));
-                decode_metrics_body(body, &events_received)
+        decode(encoding_header.as_deref(), body)
+            .inspect_err(|err| {
+                // Other status codes are already handled by `sources::util::decode` (tech debt).
+                if err.status_code() == StatusCode::UNSUPPORTED_MEDIA_TYPE {
+                    emit!(HttpBadRequest::new(
+                        err.status_code().as_u16(),
+                        err.message()
+                    ));
+                }
+            })
+            .and_then(|decoded_body| {
+                bytes_received.emit(ByteSize(decoded_body.len()));
+                if let Some(d) = deserializer.as_ref() {
+                    parse_with_deserializer(d, decoded_body, LogNamespace::default())
+                } else {
+                    decode_metrics_body(decoded_body, &events_received)
+                }
             })
-        }
     };
 
     build_ingest_filter::<ExportMetricsServiceResponse, _>(
@@ -254,16 +290,24 @@ fn build_warp_trace_filter(
     deserializer: Option<OtlpDeserializer>,
 ) -> BoxedFilter<(Response,)> {
     let make_events = move |encoding_header: Option<String>, _headers: HeaderMap, body: Bytes| {
-        if let Some(d) = deserializer.as_ref() {
-            d.parse(body, LogNamespace::default())
-                .map(|r| r.into_vec())
-                .map_err(|e| ErrorMessage::new(StatusCode::BAD_REQUEST, e.to_string()))
-        } else {
-            decode(encoding_header.as_deref(), body).and_then(|body| {
-                bytes_received.emit(ByteSize(body.len()));
-                decode_trace_body(body, &events_received)
+        decode(encoding_header.as_deref(), body)
+            .inspect_err(|err| {
+                // Other status codes are already handled by `sources::util::decode` (tech debt).
+                if err.status_code() == StatusCode::UNSUPPORTED_MEDIA_TYPE {
+                    emit!(HttpBadRequest::new(
+                        err.status_code().as_u16(),
+                        err.message()
+                    ));
+                }
+            })
+            .and_then(|decoded_body| {
+                bytes_received.emit(ByteSize(decoded_body.len()));
+                if let Some(d) = deserializer.as_ref() {
+                    parse_with_deserializer(d, decoded_body, LogNamespace::default())
+                } else {
+                    decode_trace_body(decoded_body, &events_received)
+                }
             })
-        }
     };
 
     build_ingest_filter::<ExportTraceServiceResponse, _>(
@@ -278,12 +322,7 @@ fn decode_trace_body(
     body: Bytes,
     events_received: &Registered<EventsReceived>,
 ) -> Result<Vec<Event>, ErrorMessage> {
-    let request = ExportTraceServiceRequest::decode(body).map_err(|error| {
-        ErrorMessage::new(
-            StatusCode::BAD_REQUEST,
-            format!("Could not decode request: {error}"),
-        )
-    })?;
+    let request = ExportTraceServiceRequest::decode(body).map_err(emit_decode_error)?;
 
     let events: Vec<Event> = request
         .resource_spans
@@ -304,12 +343,7 @@ fn decode_log_body(
     log_namespace: LogNamespace,
     events_received: &Registered<EventsReceived>,
 ) -> Result<Vec<Event>, ErrorMessage> {
-    let request = ExportLogsServiceRequest::decode(body).map_err(|error| {
-        ErrorMessage::new(
-            StatusCode::BAD_REQUEST,
-            format!("Could not decode request: {error}"),
-        )
-    })?;
+    let request = ExportLogsServiceRequest::decode(body).map_err(emit_decode_error)?;
 
     let events: Vec<Event> = request
         .resource_logs
@@ -329,12 +363,7 @@ fn decode_metrics_body(
     body: Bytes,
     events_received: &Registered<EventsReceived>,
 ) -> Result<Vec<Event>, ErrorMessage> {
-    let request = ExportMetricsServiceRequest::decode(body).map_err(|error| {
-        ErrorMessage::new(
-            StatusCode::BAD_REQUEST,
-            format!("Could not decode request: {error}"),
-        )
-    })?;
+    let request = ExportMetricsServiceRequest::decode(body).map_err(emit_decode_error)?;
 
     let events: Vec<Event> = request
         .resource_metrics

src/sources/util/http/encoding.rs

@@ -7,6 +7,9 @@ use warp::http::StatusCode;
 
 use crate::{common::http::ErrorMessage, internal_events::HttpDecompressError};
 
+/// Decompresses the body based on the Content-Encoding header.
+///
+/// Supports gzip, deflate, snappy, zstd, and identity (no compression).
 pub fn decode(header: Option<&str>, mut body: Bytes) -> Result<Bytes, ErrorMessage> {
     if let Some(encodings) = header {
         for encoding in encodings.rsplit(',').map(str::trim) {

tests/data/e2e/opentelemetry/logs/collector-source.yaml

@@ -9,18 +9,22 @@ processors:
   batch: { }
 
 exporters:
-  otlp:
+  otlp/grpc:
     endpoint: vector:4317
     tls:
       insecure: true
+  otlphttp/vector:
+    endpoint: http://vector:4318
+    tls:
+      insecure: true
   debug: { }
 
 service:
   pipelines:
     logs:
       receivers: [ otlp ]
       processors: [ batch ]
-      exporters: [ debug, otlp ]
+      exporters: [ debug, otlp/grpc, otlphttp/vector ]
 
     metrics:
       receivers: [ otlp ]

tests/e2e/opentelemetry/logs/mod.rs

@@ -7,7 +7,7 @@ use vector_lib::opentelemetry::proto::common::v1::any_value::Value as AnyValueEn
 use vector_lib::opentelemetry::proto::{DESCRIPTOR_BYTES, LOGS_REQUEST_MESSAGE_TYPE};
 use vrl::value::Value as VrlValue;
 
-const EXPECTED_LOG_COUNT: usize = 100;
+const EXPECTED_LOG_COUNT: usize = 200; // 100 via gRPC + 100 via HTTP
 
 fn read_file_helper(filename: &str) -> Result<String, io::Error> {
     let local_path = Path::new("/output/opentelemetry-logs").join(filename);
@@ -186,15 +186,14 @@ fn vector_sink_otel_sink_logs_match() {
     let vector_request = parse_export_logs_request(&vector_content)
         .expect("Failed to parse vector logs as ExportLogsServiceRequest");
 
-    // Count total log records in collector output
+    // Count total log records
     let collector_log_count = collector_request
         .resource_logs
         .iter()
         .flat_map(|rl| &rl.scope_logs)
         .flat_map(|sl| &sl.log_records)
         .count();
 
-    // Count total log records in vector output
     let vector_log_count = vector_request
         .resource_logs
         .iter()
@@ -220,9 +219,10 @@ fn vector_sink_otel_sink_logs_match() {
     assert_log_records_static_fields(&collector_request);
     assert_log_records_static_fields(&vector_request);
 
-    // Compare the requests - this compares the full protobuf structure
+    // Both collector and Vector receive 200 logs total (100 via gRPC + 100 via HTTP).
+    // Compare them directly to verify the entire pipeline works correctly.
     assert_eq!(
         collector_request, vector_request,
-        "ExportLogsServiceRequest mismatch between collector and vector"
+        "Collector and Vector log requests should match"
     );
 }