enhancement(antithesis): accept Datadog intake endpoints

Author: blt

Cargo.lock

@@ -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",

test/antithesis/intake/Cargo.toml

@@ -1,26 +1,92 @@
-//! 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 self::metrics::handle_series;
 use super::middleware::measure_compressed_size;
 use super::state::AppState;
 
+mod events;
 mod metrics;
+mod service_checks;
 
 /// Build Datadog-compatible intake routes.
 pub(crate) fn routes() -> Router<AppState> {
+    Router::new()
+        .merge(metric_routes())
+        .merge(event_routes())
+        .merge(service_check_routes())
+        .route("/api/v1/validate", get(|| async { StatusCode::OK }))
+}
+
+fn metric_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(
+    let series = post(metrics::handle_series).layer(
         ServiceBuilder::new()
             .layer(from_fn(measure_compressed_size))
             .layer(RequestDecompressionLayer::new().pass_through_unaccepted(true))
             .layer(DefaultBodyLimit::disable()),
     );
 
-    Router::new().route("/api/v2/series", series)
+    let decoded_payload_route = ServiceBuilder::new()
+        .layer(RequestDecompressionLayer::new().pass_through_unaccepted(true))
+        .layer(DefaultBodyLimit::disable());
+
+    Router::new().route("/api/v2/series", series).route(
+        "/api/beta/sketches",
+        post(metrics::handle_sketches).layer(decoded_payload_route),
+    )
+}
+
+fn event_routes() -> Router<AppState> {
+    let decoded_payload_route = ServiceBuilder::new()
+        .layer(RequestDecompressionLayer::new().pass_through_unaccepted(true))
+        .layer(DefaultBodyLimit::disable());
+
+    Router::new()
+        .route(
+            "/api/v1/events_batch",
+            post(events::handle_events_batch).layer(decoded_payload_route.clone()),
+        )
+        .route(
+            "/api/v1/events",
+            post(events::handle_events_v1).layer(decoded_payload_route.clone()),
+        )
+        .route(
+            "/intake/",
+            post(events::handle_intake).layer(decoded_payload_route.clone()),
+        )
+}
+
+fn service_check_routes() -> Router<AppState> {
+    let decoded_payload_route = ServiceBuilder::new()
+        .layer(RequestDecompressionLayer::new().pass_through_unaccepted(true))
+        .layer(DefaultBodyLimit::disable());
+
+    Router::new().route(
+        "/api/v1/check_run",
+        post(service_checks::handle_check_run_v1).layer(decoded_payload_route),
+    )
 }

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

@@ -0,0 +1,95 @@
+//! Event intake handlers.
+
+use std::collections::HashMap;
+
+use axum::{body::Bytes, http::StatusCode};
+use datadog_protos::events::EventsPayload;
+use protobuf::Message;
+use serde::Deserialize;
+use tracing::{debug, error};
+
+/// Handler for `POST /api/v1/events_batch`.
+pub(crate) async fn handle_events_batch(body: Bytes) -> StatusCode {
+    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: Bytes) -> StatusCode {
+    record_intake_events(&body, true)
+}
+
+/// Handler for `POST /intake/`.
+pub(crate) async fn handle_intake(body: Bytes) -> StatusCode {
+    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/events.rs

@@ -7,11 +7,13 @@
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use axum::{
-    body::to_bytes,
+    body::{to_bytes, Bytes},
     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;
@@ -113,6 +115,17 @@ 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: Bytes) -> StatusCode {
+    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/metrics.rs

@@ -0,0 +1,45 @@
+//! Service check intake handlers.
+
+use axum::{body::Bytes, http::StatusCode};
+use serde::Deserialize;
+use tracing::error;
+
+/// Handler for `POST /api/v1/check_run`.
+pub(crate) async fn handle_check_run_v1(body: Bytes) -> StatusCode {
+    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,
+        );
+    }
+}