chore(datadog_agent source): reuse code from util/http/encoding.rs (#24071)

* chore(datadog_agent source): reuse code from src/sources/util/http/encoding.rs

* update comment

Author: pront

Cargo.toml

@@ -643,7 +643,7 @@ sources-aws_ecs_metrics = ["sources-utils-http-client"]
 sources-aws_kinesis_firehose = ["dep:base64"]
 sources-aws_s3 = ["aws-core", "dep:aws-sdk-sqs", "dep:aws-sdk-s3", "dep:async-compression", "sources-aws_sqs", "tokio-util/io"]
 sources-aws_sqs = ["aws-core", "dep:aws-sdk-sqs"]
-sources-datadog_agent = ["sources-utils-http-error", "protobuf-build", "dep:prost"]
+sources-datadog_agent = ["sources-utils-http-encoding", "protobuf-build", "dep:prost"]
 sources-demo_logs = ["dep:fakedata"]
 sources-dnstap = ["sources-utils-net-tcp", "dep:base64", "dep:hickory-proto", "dep:dnsmsg-parser", "dep:dnstap-parser", "protobuf-build", "dep:prost"]
 sources-docker_logs = ["docker"]

src/sources/datadog_agent/mod.rs

@@ -57,10 +57,10 @@ use crate::{
     },
     event::Event,
     http::{KeepaliveConfig, MaxConnectionAgeLayer, build_http_trace_layer},
-    internal_events::{HttpBytesReceived, HttpDecompressError, StreamClosedError},
+    internal_events::{HttpBytesReceived, StreamClosedError},
     schema,
     serde::{bool_or_struct, default_decoding, default_framing_message_based},
-    sources::{self},
+    sources::{self, util::http::emit_decompress_error},
     tls::{MaybeTlsSettings, TlsEnableableConfig},
 };
 
@@ -475,20 +475,20 @@ impl DatadogAgentSource {
                         let mut decoded = Vec::new();
                         MultiGzDecoder::new(body.reader())
                             .read_to_end(&mut decoded)
-                            .map_err(|error| handle_decode_error(encoding, error))?;
+                            .map_err(|error| emit_decompress_error(encoding, error))?;
                         decoded.into()
                     }
                     "zstd" => {
                         let mut decoded = Vec::new();
                         zstd::stream::copy_decode(body.reader(), &mut decoded)
-                            .map_err(|error| handle_decode_error(encoding, error))?;
+                            .map_err(|error| emit_decompress_error(encoding, error))?;
                         decoded.into()
                     }
                     "deflate" | "x-deflate" => {
                         let mut decoded = Vec::new();
                         ZlibDecoder::new(body.reader())
                             .read_to_end(&mut decoded)
-                            .map_err(|error| handle_decode_error(encoding, error))?;
+                            .map_err(|error| emit_decompress_error(encoding, error))?;
                         decoded.into()
                     }
                     encoding => {
@@ -548,17 +548,6 @@ pub(crate) async fn handle_request(
     }
 }
 
-fn handle_decode_error(encoding: &str, error: impl std::error::Error) -> ErrorMessage {
-    emit!(HttpDecompressError {
-        encoding,
-        error: &error
-    });
-    ErrorMessage::new(
-        StatusCode::UNPROCESSABLE_ENTITY,
-        format!("Failed decompressing payload with {encoding} decoder."),
-    )
-}
-
 // https://github.com/DataDog/datadog-agent/blob/a33248c2bc125920a9577af1e16f12298875a4ad/pkg/logs/processor/json.go#L23-L49
 #[derive(Clone, Debug, Deserialize, Serialize)]
 #[serde(deny_unknown_fields)]

src/sources/opentelemetry/http.rs

@@ -39,7 +39,7 @@ use crate::{
     sources::{
         http_server::HttpConfigParamKind,
         opentelemetry::config::{LOGS, METRICS, OpentelemetryConfig, TRACES},
-        util::{add_headers, decode},
+        util::{add_headers, decompress_body},
     },
     tls::MaybeTlsSettings,
 };
@@ -214,9 +214,9 @@ fn build_warp_log_filter(
     deserializer: Option<OtlpDeserializer>,
 ) -> BoxedFilter<(Response,)> {
     let make_events = move |encoding_header: Option<String>, headers: HeaderMap, body: Bytes| {
-        decode(encoding_header.as_deref(), body)
+        decompress_body(encoding_header.as_deref(), body)
             .inspect_err(|err| {
-                // Other status codes are already handled by `sources::util::decode` (tech debt).
+                // Other status codes are already handled by `sources::util::decompress_body` (tech debt).
                 if err.status_code() == StatusCode::UNSUPPORTED_MEDIA_TYPE {
                     emit!(HttpBadRequest::new(
                         err.status_code().as_u16(),
@@ -254,9 +254,9 @@ fn build_warp_metrics_filter(
     deserializer: Option<OtlpDeserializer>,
 ) -> BoxedFilter<(Response,)> {
     let make_events = move |encoding_header: Option<String>, _headers: HeaderMap, body: Bytes| {
-        decode(encoding_header.as_deref(), body)
+        decompress_body(encoding_header.as_deref(), body)
             .inspect_err(|err| {
-                // Other status codes are already handled by `sources::util::decode` (tech debt).
+                // Other status codes are already handled by `sources::util::decompress_body` (tech debt).
                 if err.status_code() == StatusCode::UNSUPPORTED_MEDIA_TYPE {
                     emit!(HttpBadRequest::new(
                         err.status_code().as_u16(),
@@ -290,9 +290,9 @@ fn build_warp_trace_filter(
     deserializer: Option<OtlpDeserializer>,
 ) -> BoxedFilter<(Response,)> {
     let make_events = move |encoding_header: Option<String>, _headers: HeaderMap, body: Bytes| {
-        decode(encoding_header.as_deref(), body)
+        decompress_body(encoding_header.as_deref(), body)
             .inspect_err(|err| {
-                // Other status codes are already handled by `sources::util::decode` (tech debt).
+                // Other status codes are already handled by `sources::util::decompress_body` (tech debt).
                 if err.status_code() == StatusCode::UNSUPPORTED_MEDIA_TYPE {
                     emit!(HttpBadRequest::new(
                         err.status_code().as_u16(),

src/sources/prometheus/remote_write.rs

@@ -20,7 +20,7 @@ use crate::{
     serde::bool_or_struct,
     sources::{
         self,
-        util::{HttpSource, decode, http::HttpMethod},
+        util::{HttpSource, decompress_body, http::HttpMethod},
     },
     tls::TlsEnableableConfig,
 };
@@ -186,7 +186,7 @@ impl RemoteWriteSource {
 impl HttpSource for RemoteWriteSource {
     fn decode(&self, encoding_header: Option<&str>, body: Bytes) -> Result<Bytes, ErrorMessage> {
         // Default to snappy decoding the request body.
-        decode(encoding_header.or(Some("snappy")), body)
+        decompress_body(encoding_header.or(Some("snappy")), body)
     }
 
     fn build_events(

src/sources/util/http/encoding.rs

@@ -10,7 +10,7 @@ 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> {
+pub fn decompress_body(header: Option<&str>, mut body: Bytes) -> Result<Bytes, ErrorMessage> {
     if let Some(encodings) = header {
         for encoding in encodings.rsplit(',').map(str::trim) {
             body = match encoding {
@@ -19,22 +19,22 @@ pub fn decode(header: Option<&str>, mut body: Bytes) -> Result<Bytes, ErrorMessa
                     let mut decoded = Vec::new();
                     MultiGzDecoder::new(body.reader())
                         .read_to_end(&mut decoded)
-                        .map_err(|error| handle_decode_error(encoding, error))?;
+                        .map_err(|error| emit_decompress_error(encoding, error))?;
                     decoded.into()
                 }
                 "deflate" => {
                     let mut decoded = Vec::new();
                     ZlibDecoder::new(body.reader())
                         .read_to_end(&mut decoded)
-                        .map_err(|error| handle_decode_error(encoding, error))?;
+                        .map_err(|error| emit_decompress_error(encoding, error))?;
                     decoded.into()
                 }
                 "snappy" => SnappyDecoder::new()
                     .decompress_vec(&body)
-                    .map_err(|error| handle_decode_error(encoding, error))?
+                    .map_err(|error| emit_decompress_error(encoding, error))?
                     .into(),
                 "zstd" => zstd::decode_all(body.reader())
-                    .map_err(|error| handle_decode_error(encoding, error))?
+                    .map_err(|error| emit_decompress_error(encoding, error))?
                     .into(),
                 encoding => {
                     return Err(ErrorMessage::new(
@@ -49,7 +49,7 @@ pub fn decode(header: Option<&str>, mut body: Bytes) -> Result<Bytes, ErrorMessa
     Ok(body)
 }
 
-fn handle_decode_error(encoding: &str, error: impl std::error::Error) -> ErrorMessage {
+pub fn emit_decompress_error(encoding: &str, error: impl std::error::Error) -> ErrorMessage {
     emit!(HttpDecompressError {
         encoding,
         error: &error

src/sources/util/http/mod.rs

@@ -17,7 +17,7 @@ mod prelude;
 mod query;
 
 #[cfg(feature = "sources-utils-http-encoding")]
-pub use encoding::decode;
+pub use encoding::{decompress_body, emit_decompress_error};
 #[cfg(feature = "sources-utils-http-headers")]
 pub use headers::add_headers;
 pub use method::HttpMethod;

src/sources/util/http/prelude.rs

@@ -21,7 +21,7 @@ use warp::{
     reject::Rejection,
 };
 
-use super::encoding::decode;
+use super::encoding::decompress_body;
 use crate::{
     SourceSender,
     common::http::{ErrorMessage, server_auth::HttpServerAuthConfig},
@@ -57,7 +57,7 @@ pub trait HttpSource: Clone + Send + Sync + 'static {
     ) -> Result<Vec<Event>, ErrorMessage>;
 
     fn decode(&self, encoding_header: Option<&str>, body: Bytes) -> Result<Bytes, ErrorMessage> {
-        decode(encoding_header, body)
+        decompress_body(encoding_header, body)
     }
 
     #[allow(clippy::too_many_arguments)]

src/sources/util/mod.rs

@@ -68,7 +68,7 @@ pub use self::http::add_query_parameters;
     feature = "sources-prometheus-remote-write",
     feature = "sources-utils-http-encoding"
 ))]
-pub use self::http::decode;
+pub use self::http::decompress_body;
 #[cfg(any(
     feature = "sources-aws_sqs",
     feature = "sources-gcp_pubsub",