diff --git a/changelog.d/http_sink_retry_error_body.fix.md b/changelog.d/http_sink_retry_error_body.fix.md new file mode 100644 index 0000000000000..caa07d65e2930 --- /dev/null +++ b/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 diff --git a/lib/codecs/src/encoding/format/otlp.rs b/lib/codecs/src/encoding/format/otlp.rs index 90f9b5e379a45..a8ff052bff174 100644 --- a/lib/codecs/src/encoding/format/otlp.rs +++ b/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(); diff --git a/src/sinks/util/http.rs b/src/sinks/util/http.rs index 89225f4077741..ed4b48adb28f0 100644 --- a/src/sinks/util/http.rs +++ b/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: PhantomData, + retry_strategy: RetryStrategy, +} + +impl RetryLogic for HttpResponseRetryLogic { + 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 { + 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( +pub const fn http_response_retry_logic( retry_strategy: RetryStrategy, -) -> HttpStatusRetryLogic< - impl Fn(&HttpResponse) -> StatusCode + Clone + Send + Sync + 'static, - Request, - HttpResponse, -> { - HttpStatusRetryLogic::new( - |req: &HttpResponse| req.http_response.status(), +) -> HttpResponseRetryLogic { + 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) -> 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();