fix(opentelemetry source): return 4xx instead of 500 for rejected HTTP requests

changelog.d/26456_opentelemetry_http_rejection_status.fix.md

@@ -0,0 +1,3 @@
+The `opentelemetry` source's OTLP/HTTP endpoints now return `415 Unsupported Media Type` when the `Content-Type` is missing or is not `application/x-protobuf`. Other rejected requests, such as a request to an unknown path or with a method other than `POST`, now return the applicable `4xx` status code instead of `500 Internal Server Error`.
+
+authors: Andrew-Hinson thomasqueirozb

src/sources/opentelemetry/http.rs

@@ -2,8 +2,8 @@ use std::{convert::Infallible, net::SocketAddr, time::Duration};
 
 use bytes::Bytes;
 use futures_util::FutureExt;
-use http::StatusCode;
-use hyper::{Server, service::make_service_fn};
+use http::{StatusCode, header::CONTENT_TYPE};
+use hyper::{Server, body::HttpBody as _, service::make_service_fn};
 use prost::Message;
 use snafu::Snafu;
 use tokio::net::TcpStream;
@@ -92,7 +92,11 @@ pub(crate) async fn run_http_server(
 
     let span = Span::current();
     // Admission wraps the Warp service, so rejected requests cannot reach `capped_body`.
-    let admitted = request_control.layer(warp::service(routes));
+    let admitted = request_control.layer(
+        ServiceBuilder::new()
+            .and_then(encode_error_as_status)
+            .service(warp::service(routes)),
+    );
     let make_svc = make_service_fn(move |conn: &MaybeTlsIncomingStream<TcpStream>| {
         let svc = ServiceBuilder::new()
             .layer(build_http_trace_layer(span.clone()))
@@ -482,25 +486,80 @@ fn acknowledgement_failure_response(status: AcknowledgementFailure) -> Response
     warp::reply::with_status(response, StatusCode::INTERNAL_SERVER_ERROR).into_response()
 }
 
-async fn handle_rejection(err: Rejection) -> Result<impl Reply, std::convert::Infallible> {
-    if let Some(err_msg) = err.find::<ErrorMessage>() {
-        let reply = protobuf(Status {
-            code: 2, // UNKNOWN - OTLP doesn't require use of status.code, but we can't encode a None here
-            message: err_msg.message().into(),
-            ..Default::default()
-        });
-
-        Ok(warp::reply::with_status(reply, err_msg.status_code()))
+/// Maps rejections that need a different status code than warp's default.
+///
+/// Every other rejection is returned to warp, which chooses its status code. The resulting
+/// response is then re-encoded by [`encode_error_as_status`].
+async fn handle_rejection(err: Rejection) -> Result<Response, Rejection> {
+    let (message, status) = if let Some(err_msg) = err.find::<ErrorMessage>() {
+        (err_msg.message().into(), err_msg.status_code())
+    } else if is_content_type_rejection(&err) {
+        // The route only matches `Content-Type: application/x-protobuf`, so any other (or a
+        // missing) content type is a client error rather than an internal one. warp would
+        // report this as `400 Bad Request`, so it is handled here.
+        (
+            "Unsupported content type; this endpoint requires `application/x-protobuf`.".into(),
+            StatusCode::UNSUPPORTED_MEDIA_TYPE,
+        )
+    } else if err.find::<ApiError>().is_some() {
+        (format!("{err:?}"), StatusCode::INTERNAL_SERVER_ERROR)
     } else {
-        let reply = protobuf(Status {
-            code: 2, // UNKNOWN - OTLP doesn't require use of status.code, but we can't encode a None here
-            message: format!("{err:?}"),
-            ..Default::default()
-        });
-
-        Ok(warp::reply::with_status(
-            reply,
-            StatusCode::INTERNAL_SERVER_ERROR,
-        ))
+        return Err(err);
+    };
+
+    Ok(status_response(message, status))
+}
+
+/// Returns true if the request was rejected because of its `Content-Type` header.
+///
+/// `warp::header::exact_ignore_case` rejects with `InvalidHeader` both when the header has a
+/// different value and when it is absent; `MissingHeader` is checked as well for robustness.
+fn is_content_type_rejection(err: &Rejection) -> bool {
+    let is_content_type = |name: &str| name.eq_ignore_ascii_case(CONTENT_TYPE.as_str());
+    err.find::<warp::reject::InvalidHeader>()
+        .is_some_and(|e| is_content_type(e.name()))
+        || err
+            .find::<warp::reject::MissingHeader>()
+            .is_some_and(|e| is_content_type(e.name()))
+}
+
+/// Replaces the body of error responses generated by warp with a protobuf-encoded `Status`.
+///
+/// OTLP/HTTP requires a `Status` body for all 4xx and 5xx responses, but warp replies to the
+/// rejections that `handle_rejection` returns with a plain-text body. The status code chosen by
+/// warp is kept, and its message is used as the `Status` message.
+async fn encode_error_as_status(response: Response) -> Result<Response, Infallible> {
+    let status = response.status();
+    let is_protobuf = response
+        .headers()
+        .get(CONTENT_TYPE)
+        .is_some_and(|value| value == "application/x-protobuf");
+    if is_protobuf || !(status.is_client_error() || status.is_server_error()) {
+        return Ok(response);
     }
+
+    // warp's rejection replies have short, fully buffered bodies, and `404 Not Found` has none.
+    let body = response
+        .into_body()
+        .collect()
+        .await
+        .map(|collected| collected.to_bytes())
+        .unwrap_or_default();
+    let message = if body.is_empty() {
+        status.canonical_reason().unwrap_or_default().to_owned()
+    } else {
+        String::from_utf8_lossy(&body).into_owned()
+    };
+
+    Ok(status_response(message, status))
+}
+
+fn status_response(message: String, status: StatusCode) -> Response {
+    let reply = protobuf(Status {
+        code: 2, // UNKNOWN - OTLP doesn't require use of status.code, but we can't encode a None here
+        message,
+        ..Default::default()
+    });
+
+    warp::reply::with_status(reply, status).into_response()
 }

src/sources/opentelemetry/tests.rs

@@ -2001,6 +2001,82 @@ async fn http_logs_use_otlp_decoding_emits_metric() {
     }
 }
 
+/// Sends a request to the OTLP/HTTP endpoint and returns the status code with the decoded
+/// `google.rpc.Status` message from the response body.
+async fn send_http_request(request: reqwest::RequestBuilder) -> (reqwest::StatusCode, String) {
+    let response = request.send().await.expect("Failed to send request.");
+    let status = response.status();
+    let body = response
+        .bytes()
+        .await
+        .expect("Failed to read response body.");
+    let message = super::status::Status::decode(body)
+        .expect("Response body should be a protobuf `Status`.")
+        .message;
+    (status, message)
+}
+
+#[tokio::test]
+async fn http_rejections_return_client_error_status_codes() {
+    let env = build_otlp_test_env(LOGS, None).await;
+    let http_addr = env.config.http.address;
+    test_util::wait_for_tcp(http_addr).await;
+    let client = reqwest::Client::new();
+    let logs_url = format!("http://{http_addr}/v1/logs");
+
+    // JSON is not supported, so the request should be rejected as an unsupported media type.
+    let (status, message) = send_http_request(
+        client
+            .post(&logs_url)
+            .header("Content-Type", "application/json")
+            .body(r#"{"resourceLogs":[]}"#),
+    )
+    .await;
+    assert_eq!(status, reqwest::StatusCode::UNSUPPORTED_MEDIA_TYPE);
+    assert!(message.contains("application/x-protobuf"), "{message}");
+
+    // A missing content type is treated the same as an unsupported one.
+    let (status, _) = send_http_request(client.post(&logs_url)).await;
+    assert_eq!(status, reqwest::StatusCode::UNSUPPORTED_MEDIA_TYPE);
+
+    // Other rejections keep the status code chosen by warp, with a protobuf `Status` body.
+    // Only `POST` is routed.
+    let (status, _) = send_http_request(client.get(&logs_url)).await;
+    assert_eq!(status, reqwest::StatusCode::METHOD_NOT_ALLOWED);
+
+    // Unknown paths are not found.
+    let (status, _) = send_http_request(
+        client
+            .post(format!("http://{http_addr}/v1/unknown"))
+            .header("Content-Type", "application/x-protobuf"),
+    )
+    .await;
+    assert_eq!(status, reqwest::StatusCode::NOT_FOUND);
+
+    // A header value that can't be parsed is a bad request, and warp's message is kept.
+    let invalid_encoding = reqwest::header::HeaderValue::from_bytes(b"gzip\xff")
+        .expect("Header value should be valid.");
+    let (status, message) = send_http_request(
+        client
+            .post(&logs_url)
+            .header("Content-Type", "application/x-protobuf")
+            .header("Content-Encoding", invalid_encoding),
+    )
+    .await;
+    assert_eq!(status, reqwest::StatusCode::BAD_REQUEST);
+    assert!(message.contains("content-encoding"), "{message}");
+
+    // The supported content type is still accepted.
+    let response = client
+        .post(&logs_url)
+        .header("Content-Type", "application/x-protobuf")
+        .body(ExportLogsServiceRequest::default().encode_to_vec())
+        .send()
+        .await
+        .expect("Failed to send request.");
+    assert_eq!(response.status(), reqwest::StatusCode::OK);
+}
+
 #[cfg(test)]
 mod otlp_decoding_config_tests {
     use indoc::indoc;