feat(sinks): log response body when HTTP response failed

Author: keuin

Cargo.lock

@@ -438,6 +438,7 @@ heim = { git = "https://github.com/vectordotdev/heim.git", branch = "update-nix"
 mlua = { version = "0.10.5", default-features = false, features = ["lua54", "send", "vendored", "macros"], optional = true }
 sysinfo = "0.37.2"
 byteorder = "1.5.0"
+brotli = "8.0.2"
 
 [target.'cfg(windows)'.dependencies]
 windows-service = "0.8.0"

Cargo.toml

@@ -0,0 +1,3 @@
+Added HTTP response body previews to error logs. When an HTTP sink request fails, Vector will now attempt to decompress (gzip, zstd, br, deflate) and log the first 1024 characters of the response body to help troubleshooting.
+
+authors: Keuin

changelog.d/display_sink_response_error_message.enhancement.md

@@ -2,11 +2,14 @@
 use aws_credential_types::provider::SharedCredentialsProvider;
 #[cfg(feature = "aws-core")]
 use aws_types::region::Region;
+use brotli::Decompressor as BrotliDecoder;
 use bytes::{Buf, Bytes};
+use flate2::read::GzDecoder;
 use futures::{Sink, future::BoxFuture};
 use headers::HeaderName;
 use http::{HeaderValue, Request, Response, StatusCode, header};
 use http_body::Body as _;
+use zstd::stream::read::Decoder as ZstdDecoder;
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct OrderedHeaderName(HeaderName);
@@ -38,21 +41,24 @@ impl PartialOrd for OrderedHeaderName {
         Some(self.cmp(other))
     }
 }
+use flate2::bufread::ZlibDecoder;
+use http::header::{CONTENT_ENCODING, CONTENT_TYPE};
+use hyper::Body;
+use pin_project::pin_project;
+use snafu::{ResultExt, Snafu};
+use std::io::Read;
 use std::{
     collections::BTreeMap,
     fmt,
     future::Future,
     hash::Hash,
+    io,
     marker::PhantomData,
     pin::Pin,
     sync::Arc,
     task::{Context, Poll, ready},
     time::Duration,
 };
-
-use hyper::Body;
-use pin_project::pin_project;
-use snafu::{ResultExt, Snafu};
 use tower::{Service, ServiceBuilder};
 use tower_http::decompression::DecompressionLayer;
 use vector_lib::{
@@ -580,15 +586,89 @@ impl<Req: Clone + Send + Sync + 'static> RetryLogic for HttpRetryLogic<Req> {
             StatusCode::NOT_IMPLEMENTED => {
                 RetryAction::DontRetry("endpoint not implemented".into())
             }
-            _ if status.is_server_error() => RetryAction::Retry(
-                format!("{}: {}", status, String::from_utf8_lossy(response.body())).into(),
-            ),
             _ if status.is_success() => RetryAction::Successful,
-            _ => RetryAction::DontRetry(format!("response status: {status}").into()),
+            _ => {
+                let body_preview = get_response_preview(response)
+                    .unwrap_or_else(|err| format!("cannot peek: {err:?}"));
+                if status.is_server_error() {
+                    RetryAction::Retry(format!("{status}: {body_preview}").into())
+                } else {
+                    RetryAction::DontRetry(
+                        format!("response status: {status}, body: {body_preview}").into(),
+                    )
+                }
+            }
         }
     }
 }
 
+#[derive(Debug, Snafu)]
+pub enum ResponsePreviewError {
+    #[snafu(display("Cannot preview a binary response content: {}", content_type))]
+    BinaryContent { content_type: String },
+    #[snafu(display("Unknown encoding of content in HTTP response: {}", content_encoding))]
+    UnknownEncoding { content_encoding: String },
+    #[snafu(display("Error reading data: {:?}", err))]
+    IOError { err: io::Error },
+}
+
+/// Try to decompress and read the first 1024 bytes from the HTTP response body.
+fn get_response_preview(response: &Response<Bytes>) -> crate::Result<String> {
+    const BROTLI_INTERNAL_BUFFER_SIZE: usize = 4096;
+    const PEEK_UTF8_CHARACTERS: usize = 1024;
+    const UNKNOWN_CONTENT_TYPE: &str = "unspecified";
+
+    // skip binary data
+    let content_type = response
+        .headers()
+        .get(CONTENT_TYPE)
+        .and_then(|v| v.to_str().ok())
+        .unwrap_or(UNKNOWN_CONTENT_TYPE);
+    if content_type.starts_with("image/")
+        || content_type.starts_with("video/")
+        || content_type.starts_with("audio/")
+        || content_type == "application/octet-stream"
+    {
+        return Err(Box::new(ResponsePreviewError::BinaryContent {
+            content_type: content_type.to_string(),
+        }));
+    }
+
+    let body_bytes = response.body().as_ref();
+    let content_encoding = response
+        .headers()
+        .get(CONTENT_ENCODING)
+        .and_then(|v| v.to_str().ok())
+        .unwrap_or_else(|| "unknown");
+
+    // handle different compression methods in HTTP response
+    let mut reader: Box<dyn Read> = match content_encoding {
+        "gzip" => Box::new(GzDecoder::new(body_bytes)),
+        "deflate" => Box::new(ZlibDecoder::new(body_bytes)),
+        "br" => Box::new(BrotliDecoder::new(body_bytes, BROTLI_INTERNAL_BUFFER_SIZE)),
+        "zstd" => Box::new(
+            ZstdDecoder::new(body_bytes).unwrap_or_else(|_| ZstdDecoder::new(body_bytes).unwrap()),
+        ),
+        // unspecified or identity encoding, treat as utf-8 directly
+        UNKNOWN_CONTENT_TYPE | "identity" => Box::new(body_bytes),
+        encoding => {
+            return Err(Box::new(ResponsePreviewError::UnknownEncoding {
+                content_encoding: encoding.to_string(),
+            }));
+        }
+    };
+
+    // one utf-8 char takes up to 4 bytes, read at most that number of bytes for utf-8 decoding
+    let mut buf = [0u8; PEEK_UTF8_CHARACTERS * 4];
+    match reader.read(&mut buf) {
+        Ok(cnt) => Ok(String::from_utf8_lossy(&buf[..cnt])
+            .chars()
+            .take(PEEK_UTF8_CHARACTERS)
+            .collect()),
+        Err(why) => Err(Box::new(ResponsePreviewError::IOError { err: why })),
+    }
+}
+
 /// A more generic version of `HttpRetryLogic` that accepts anything that can be converted
 /// to a status code
 #[derive(Debug)]