chore(sinks): log reason for non-retryable errors

changelog.d/http_sink_retry_error_body.fix.md

@@ -0,0 +1,3 @@
+HTTP-based sinks that use `http_response_retry_logic` (including the `opentelemetry`/`http` sink, `keep`, `honeycomb`, GCP Stackdriver metrics/logs, and `prometheus_remote_write`) now include a truncated (up to 1KB) response body in the "Not retriable; dropping the request" error log for non-retriable responses (e.g. `400 Bad Request`). Previously only the HTTP status code's canonical reason phrase was logged, hiding the actual error returned by the destination.
+
+authors: peter-ehikhuemen_ddog

lib/codecs/src/encoding/format/otlp.rs

@@ -255,6 +255,46 @@ mod tests {
         assert_eq!(metric.clone(), round_trip_metric(metric));
     }
 
+    #[test]
+    fn dump_two_metrics_concatenated_for_manual_probe() {
+        let mut serializer = OtlpSerializer::new().unwrap();
+
+        let m1 = Metric::new(
+            "cpu_usage",
+            MetricKind::Absolute,
+            MetricValue::Gauge { value: 12.5 },
+        )
+        .with_timestamp(Some(Utc::now()));
+        let m2 = Metric::new(
+            "mem_usage",
+            MetricKind::Absolute,
+            MetricValue::Gauge { value: 55.0 },
+        )
+        .with_timestamp(Some(Utc::now()));
+
+        let mut buf1 = BytesMut::new();
+        serializer
+            .encode(Event::Metric(m1), &mut buf1)
+            .expect("encode m1");
+        let mut buf2 = BytesMut::new();
+        serializer
+            .encode(Event::Metric(m2), &mut buf2)
+            .expect("encode m2");
+
+        let mut combined = Vec::new();
+        combined.extend_from_slice(&buf1);
+        combined.extend_from_slice(&buf2);
+
+        std::fs::write("/tmp/two_metrics_concat.bin", &combined).unwrap();
+        std::fs::write("/tmp/one_metric.bin", &buf1).unwrap();
+        eprintln!(
+            "buf1 len={} buf2 len={} combined len={}",
+            buf1.len(),
+            buf2.len(),
+            combined.len()
+        );
+    }
+
     #[test]
     fn unsupported_metric_values_return_err() {
         let mut serializer = OtlpSerializer::new().unwrap();

src/sinks/util/http.rs

@@ -945,18 +945,63 @@ impl DriverResponse for HttpResponse {
     }
 }
 
+/// Maximum number of response body bytes to include in a non-retriable error's reason.
+const MAX_ERROR_BODY_BYTES: usize = 1024;
+
+/// A `RetryLogic` for use with `HttpResponse` that, unlike `HttpStatusRetryLogic`, has
+/// access to the response body so it can include it in the reason when a request is not
+/// retriable (the response body is otherwise discarded before it reaches the driver's
+/// error log).
+#[derive(Debug, Clone)]
+pub struct HttpResponseRetryLogic<Request> {
+    request: PhantomData<Request>,
+    retry_strategy: RetryStrategy,
+}
+
+impl<Request: Clone + Send + Sync + 'static> RetryLogic for HttpResponseRetryLogic<Request> {
+    type Error = HttpError;
+    type Request = Request;
+    type Response = HttpResponse;
+
+    fn is_retriable_error(&self, error: &Self::Error) -> bool {
+        if self.retry_strategy == RetryStrategy::None {
+            false
+        } else {
+            error.is_retriable()
+        }
+    }
+
+    fn is_retriable_timeout(&self) -> bool {
+        self.retry_strategy != RetryStrategy::None
+    }
+
+    fn should_retry_response(&self, response: &Self::Response) -> RetryAction<Self::Request> {
+        let status = response.http_response.status();
+        match self.retry_strategy.retry_action(status) {
+            RetryAction::DontRetry(reason) => {
+                let body = response.http_response.body();
+                let truncated = &body[..body.len().min(MAX_ERROR_BODY_BYTES)];
+                RetryAction::DontRetry(
+                    format!(
+                        "{reason}, response body: {}",
+                        String::from_utf8_lossy(truncated)
+                    )
+                    .into(),
+                )
+            }
+            action => action,
+        }
+    }
+}
+
 /// Creates a `RetryLogic` for use with `HttpResponse`.
-pub fn http_response_retry_logic<Request: Clone + Send + Sync + 'static>(
+pub const fn http_response_retry_logic<Request: Clone + Send + Sync + 'static>(
     retry_strategy: RetryStrategy,
-) -> HttpStatusRetryLogic<
-    impl Fn(&HttpResponse) -> StatusCode + Clone + Send + Sync + 'static,
-    Request,
-    HttpResponse,
-> {
-    HttpStatusRetryLogic::new(
-        |req: &HttpResponse| req.http_response.status(),
+) -> HttpResponseRetryLogic<Request> {
+    HttpResponseRetryLogic {
+        request: PhantomData,
         retry_strategy,
-    )
+    }
 }
 
 /// Uses the estimated json encoded size to determine batch sizing.
@@ -1187,6 +1232,61 @@ mod test {
         );
     }
 
+    fn http_response_with_body(status: u16, body: impl Into<Bytes>) -> HttpResponse {
+        HttpResponse {
+            http_response: Response::builder()
+                .status(status)
+                .body(body.into())
+                .unwrap(),
+            events_byte_size: GroupedCountByteSize::new_untagged(),
+            raw_byte_size: 0,
+        }
+    }
+
+    #[test]
+    fn http_response_retry_logic_includes_body_for_non_retriable_responses() {
+        let logic = http_response_retry_logic::<()>(RetryStrategy::Default);
+        let response = http_response_with_body(400, "out of order sample");
+
+        match logic.should_retry_response(&response) {
+            RetryAction::DontRetry(reason) => {
+                assert!(reason.contains("out of order sample"), "reason: {reason}");
+            }
+            _ => panic!("expected DontRetry"),
+        }
+    }
+
+    #[test]
+    fn http_response_retry_logic_omits_body_for_retriable_responses() {
+        let logic = http_response_retry_logic::<()>(RetryStrategy::Default);
+        let response = http_response_with_body(500, "internal error detail");
+
+        match logic.should_retry_response(&response) {
+            RetryAction::Retry(reason) => {
+                assert!(
+                    !reason.contains("internal error detail"),
+                    "reason: {reason}"
+                );
+            }
+            _ => panic!("expected Retry"),
+        }
+    }
+
+    #[test]
+    fn http_response_retry_logic_truncates_long_bodies() {
+        let logic = http_response_retry_logic::<()>(RetryStrategy::Default);
+        let body = "x".repeat(MAX_ERROR_BODY_BYTES * 2);
+        let response = http_response_with_body(400, body);
+
+        match logic.should_retry_response(&response) {
+            RetryAction::DontRetry(reason) => {
+                let x_count = reason.chars().filter(|&c| c == 'x').count();
+                assert_eq!(x_count, MAX_ERROR_BODY_BYTES);
+            }
+            _ => panic!("expected DontRetry"),
+        }
+    }
+
     #[tokio::test]
     async fn util_http_it_makes_http_requests() {
         let (_guard, addr) = next_addr();