enhancement(antithesis): accept Datadog intake endpoints

test/antithesis/intake/Cargo.toml

@@ -22,6 +22,7 @@ datadog-protos = { workspace = true }
 headers = { workspace = true }
 mime = { workspace = true }
 protobuf = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 tokio = { workspace = true, features = [
   "macros",
@@ -35,6 +36,7 @@ tower-http = { workspace = true, features = [
   "decompression-deflate",
   "decompression-gzip",
   "decompression-zstd",
+  "limit",
 ] }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true, features = [

test/antithesis/intake/src/http.rs

@@ -16,7 +16,11 @@ use self::state::AppState;
 /// Memory backstop on the compressed body buffered before decompression. Sits above any Pyld05 spec limit.
 const MAX_COMPRESSED_BODY_BYTES: usize = 64 * 1024 * 1024;
 
-/// Build the intake router, `/api/v2/series` for payload assertions, others return 200 OK.
+/// Caps the decompressed body a handler buffers. Exceeds every Pyld06 spec limit.
+const MAX_DECOMPRESSED_BODY_BYTES: usize = 64 * 1024 * 1024;
+
+/// Build the intake router. `/api/v2/series` fires payload assertions. Datadog endpoints answer
+/// 202. A malformed body gets 400. An oversized body gets 413. Unmatched paths answer 200.
 pub fn build_router(hostname: Arc<str>) -> Router {
     Router::new()
         .merge(datadog::routes())

test/antithesis/intake/src/http/datadog.rs

@@ -1,26 +1,68 @@
-//! Datadog-compatible intake routes.
+//! Datadog-compatible HTTP intake routes.
+//!
+//! This module owns the public routes that Datadog Agent and ADP send to:
+//!
+//! - `POST /api/v2/series`: accepts metric series payloads and records payload
+//!   shape assertions.
+//! - `POST /api/beta/sketches`: accepts distribution sketch payloads.
+//! - `POST /api/v1/events_batch`: accepts protobuf event batches.
+//! - `POST /api/v1/events`: accepts JSON event intake payloads and rejects
+//!   malformed bodies.
+//! - `POST /intake/`: accepts the shared JSON intake endpoint and ignores
+//!   non-event bodies.
+//! - `POST /api/v1/check_run`: accepts service check payloads.
+//! - `GET /api/v1/validate`: accepts Datadog Agent connectivity validation.
 
-use axum::{extract::DefaultBodyLimit, middleware::from_fn, routing::post, Router};
+use axum::{
+    extract::DefaultBodyLimit,
+    http::StatusCode,
+    middleware::from_fn,
+    routing::{get, post},
+    Router,
+};
 use tower::ServiceBuilder;
-use tower_http::decompression::RequestDecompressionLayer;
+use tower_http::{decompression::RequestDecompressionLayer, limit::RequestBodyLimitLayer};
 
-use self::metrics::handle_series;
 use super::middleware::measure_compressed_size;
 use super::state::AppState;
+use super::MAX_COMPRESSED_BODY_BYTES;
 
+mod events;
 mod metrics;
+mod service_checks;
 
 /// Build Datadog-compatible intake routes.
 pub(crate) fn routes() -> Router<AppState> {
-    // Pyld01-Pyld06 and Pyld22 need the compressed body and raw headers, so the series
-    // route runs `measure_compressed_size` outermost, then decompresses, then
-    // lifts the body limit (the middleware's own cap is the backstop).
-    let series = post(handle_series).layer(
-        ServiceBuilder::new()
-            .layer(from_fn(measure_compressed_size))
-            .layer(RequestDecompressionLayer::new().pass_through_unaccepted(true))
-            .layer(DefaultBodyLimit::disable()),
-    );
+    Router::new()
+        .merge(series_route())
+        .merge(decoded_payload_routes())
+        .route("/api/v1/validate", get(|| async { StatusCode::OK }))
+}
+
+/// The `/api/v2/series` route. Pyld01-Pyld06 and Pyld22 need the compressed body and raw headers.
+/// `measure_compressed_size` runs outermost, then decompression. The route lifts the default body
+/// limit. The middleware cap is the backstop.
+fn series_route() -> Router<AppState> {
+    let layers = ServiceBuilder::new()
+        .layer(from_fn(measure_compressed_size))
+        .layer(RequestDecompressionLayer::new().pass_through_unaccepted(true))
+        .layer(DefaultBodyLimit::disable());
+    Router::new().route("/api/v2/series", post(metrics::handle_series).layer(layers))
+}
 
-    Router::new().route("/api/v2/series", series)
+/// Routes that decompress and parse a body without recording `Measurements`. One shared stack
+/// caps the wire body with `RequestBodyLimitLayer` as it streams. Decompression follows. Each
+/// handler caps the decompressed body with `to_bytes`.
+fn decoded_payload_routes() -> Router<AppState> {
+    Router::new()
+        .route("/api/beta/sketches", post(metrics::handle_sketches))
+        .route("/api/v1/events_batch", post(events::handle_events_batch))
+        .route("/api/v1/events", post(events::handle_events_v1))
+        .route("/intake/", post(events::handle_intake))
+        .route("/api/v1/check_run", post(service_checks::handle_check_run_v1))
+        .layer(
+            ServiceBuilder::new()
+                .layer(RequestBodyLimitLayer::new(MAX_COMPRESSED_BODY_BYTES))
+                .layer(RequestDecompressionLayer::new().pass_through_unaccepted(true)),
+        )
 }

test/antithesis/intake/src/http/datadog/events.rs

@@ -0,0 +1,121 @@
+//! Event intake handlers.
+
+use std::collections::HashMap;
+
+use axum::{
+    body::{to_bytes, Body},
+    http::StatusCode,
+};
+use datadog_protos::events::EventsPayload;
+use protobuf::Message;
+use serde::Deserialize;
+use tracing::{debug, error};
+
+use crate::http::MAX_DECOMPRESSED_BODY_BYTES;
+
+/// Handler for `POST /api/v1/events_batch`.
+pub(crate) async fn handle_events_batch(body: Body) -> StatusCode {
+    let body = match to_bytes(body, MAX_DECOMPRESSED_BODY_BYTES).await {
+        Ok(body) => body,
+        Err(e) => {
+            error!(error = %e, cap = MAX_DECOMPRESSED_BODY_BYTES, "Rejected events batch body at the decompressed cap.");
+            return StatusCode::PAYLOAD_TOO_LARGE;
+        }
+    };
+    match EventsPayload::parse_from_bytes(&body) {
+        Ok(_) => StatusCode::ACCEPTED,
+        Err(e) => {
+            error!(error = %e, "failed to parse events batch payload");
+            StatusCode::BAD_REQUEST
+        }
+    }
+}
+
+/// Handler for `POST /api/v1/events`.
+pub(crate) async fn handle_events_v1(body: Body) -> StatusCode {
+    let body = match to_bytes(body, MAX_DECOMPRESSED_BODY_BYTES).await {
+        Ok(body) => body,
+        Err(e) => {
+            error!(error = %e, cap = MAX_DECOMPRESSED_BODY_BYTES, "Rejected events body at the decompressed cap.");
+            return StatusCode::PAYLOAD_TOO_LARGE;
+        }
+    };
+    record_intake_events(&body, true)
+}
+
+/// Handler for `POST /intake/`.
+pub(crate) async fn handle_intake(body: Body) -> StatusCode {
+    let body = match to_bytes(body, MAX_DECOMPRESSED_BODY_BYTES).await {
+        Ok(body) => body,
+        Err(e) => {
+            error!(error = %e, cap = MAX_DECOMPRESSED_BODY_BYTES, "Rejected intake body at the decompressed cap.");
+            return StatusCode::PAYLOAD_TOO_LARGE;
+        }
+    };
+    record_intake_events(&body, false)
+}
+
+fn record_intake_events(body: &[u8], strict: bool) -> StatusCode {
+    let payload = match serde_json::from_slice::<IntakePayload>(body) {
+        Ok(payload) => payload,
+        Err(e) if strict => {
+            error!(error = %e, "failed to parse events intake payload");
+            return StatusCode::BAD_REQUEST;
+        }
+        Err(e) => {
+            debug!(error = %e, "intake payload did not contain events");
+            return StatusCode::OK;
+        }
+    };
+    payload.touch();
+    if strict {
+        StatusCode::ACCEPTED
+    } else {
+        StatusCode::OK
+    }
+}
+
+#[derive(Deserialize)]
+struct IntakePayload {
+    events: Option<HashMap<String, Vec<IntakeEvent>>>,
+}
+
+impl IntakePayload {
+    fn touch(self) {
+        let Some(events_by_source) = self.events else {
+            return;
+        };
+        for events in events_by_source.into_values() {
+            for event in events {
+                event.touch();
+            }
+        }
+    }
+}
+
+#[derive(Deserialize)]
+struct IntakeEvent {
+    msg_title: Option<String>,
+    msg_text: Option<String>,
+    alert_type: Option<String>,
+    aggregation_key: Option<String>,
+    host: Option<String>,
+    priority: Option<String>,
+    tags: Option<Vec<String>>,
+    timestamp: Option<i64>,
+}
+
+impl IntakeEvent {
+    fn touch(self) {
+        let _ = (
+            self.msg_title,
+            self.msg_text,
+            self.alert_type,
+            self.aggregation_key,
+            self.host,
+            self.priority,
+            self.tags,
+            self.timestamp,
+        );
+    }
+}

test/antithesis/intake/src/http/datadog/metrics.rs

@@ -7,21 +7,21 @@
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use axum::{
-    body::to_bytes,
+    body::{to_bytes, Body},
     extract::{Request, State},
     http::StatusCode,
     response::{IntoResponse, Response},
 };
+use datadog_protos::metrics::SketchPayload;
+use protobuf::Message;
 use tracing::{debug, error};
 
 use crate::http::middleware::Measurements;
 use crate::http::state::AppState;
+use crate::http::MAX_DECOMPRESSED_BODY_BYTES;
 use crate::properties::payload::{bytes, envelope};
 use crate::series_observation::SeriesObservation;
 
-/// Memory backstop on the decompressed body buffered in the handler, above the Pyld06 5 MiB spec limit
-const MAX_DECOMPRESSED_BODY_BYTES: usize = 64 * 1024 * 1024;
-
 /// Reasons `handle_series` cannot evaluate a request.
 #[derive(Debug)]
 pub(crate) enum SeriesError {
@@ -113,6 +113,24 @@ pub(crate) async fn handle_series(State(state): State<AppState>, request: Reques
     Ok(failure.unwrap_or(StatusCode::ACCEPTED))
 }
 
+/// Handler for `POST /api/beta/sketches`.
+pub(crate) async fn handle_sketches(body: Body) -> StatusCode {
+    let body = match to_bytes(body, MAX_DECOMPRESSED_BODY_BYTES).await {
+        Ok(body) => body,
+        Err(e) => {
+            error!(error = %e, cap = MAX_DECOMPRESSED_BODY_BYTES, "Rejected sketches body at the decompressed cap.");
+            return StatusCode::PAYLOAD_TOO_LARGE;
+        }
+    };
+    match SketchPayload::parse_from_bytes(&body) {
+        Ok(_) => StatusCode::ACCEPTED,
+        Err(e) => {
+            error!(error = %e, "failed to parse sketch payload");
+            StatusCode::BAD_REQUEST
+        }
+    }
+}
+
 /// Return the first failed status check, in the given pipeline order, or `None`
 /// when every check holds.
 fn first_status_failure(checks: &[(bool, StatusCode)]) -> Option<StatusCode> {

test/antithesis/intake/src/http/datadog/service_checks.rs

@@ -0,0 +1,57 @@
+//! Service check intake handlers.
+
+use axum::{
+    body::{to_bytes, Body},
+    http::StatusCode,
+};
+use serde::Deserialize;
+use tracing::error;
+
+use crate::http::MAX_DECOMPRESSED_BODY_BYTES;
+
+/// Handler for `POST /api/v1/check_run`.
+pub(crate) async fn handle_check_run_v1(body: Body) -> StatusCode {
+    let body = match to_bytes(body, MAX_DECOMPRESSED_BODY_BYTES).await {
+        Ok(body) => body,
+        Err(e) => {
+            error!(error = %e, cap = MAX_DECOMPRESSED_BODY_BYTES, "Rejected check_run body at the decompressed cap.");
+            return StatusCode::PAYLOAD_TOO_LARGE;
+        }
+    };
+    let items = match serde_json::from_slice::<Vec<CheckRunItem>>(&body) {
+        Ok(items) => items,
+        Err(e) => {
+            error!(error = %e, "failed to parse check_run payload");
+            return StatusCode::BAD_REQUEST;
+        }
+    };
+    for item in items {
+        item.touch();
+    }
+    StatusCode::ACCEPTED
+}
+
+#[derive(Deserialize)]
+struct CheckRunItem {
+    #[serde(rename = "check")]
+    name: String,
+    status: u8,
+    #[serde(rename = "host_name")]
+    hostname: Option<String>,
+    message: Option<String>,
+    tags: Option<Vec<String>>,
+    timestamp: Option<u64>,
+}
+
+impl CheckRunItem {
+    fn touch(self) {
+        let _ = (
+            self.name,
+            self.status,
+            self.hostname,
+            self.message,
+            self.tags,
+            self.timestamp,
+        );
+    }
+}