chore(antithesis): split intake HTTP modules

test/antithesis/intake/src/http.rs

@@ -1,77 +1,25 @@
 //! Axum HTTP surface for the intake.
 //!
-//! The `/api/v2/series` route stacks measurement middleware ahead of the
-//! decompression layer so Pyld05 (compressed size), Pyld06 (uncompressed size),
-//! and Pyld22 (content-length) can read both the on-the-wire and decompressed
-//! body lengths, recorded as request extensions before `RequestDecompressionLayer`
-//! consumes the encoding headers.
+//! This module composes the intake router while submodules keep protocol groups
+//! and middleware separate.
 
 use std::sync::Arc;
 
-use axum::{
-    body::Body,
-    extract::{DefaultBodyLimit, Request},
-    http::StatusCode,
-    middleware::{from_fn, Next},
-    response::{IntoResponse, Response},
-    routing::post,
-    Router,
-};
-use headers::{ContentEncoding, ContentLength, HeaderMapExt};
-use tower::ServiceBuilder;
-use tower_http::decompression::RequestDecompressionLayer;
+use axum::{http::StatusCode, Router};
 
-use crate::intake;
+mod datadog;
+pub(crate) mod middleware;
+mod state;
 
-/// Memory backstop on the compressed body buffered before decompression, above any Pyld05 spec limit
-const MAX_COMPRESSED_BODY_BYTES: usize = 64 * 1024 * 1024;
+use self::state::AppState;
 
-/// Wire measurements recorded before decompression, attached as a request extension for Pyld05/Pyld06/Pyld22
-#[derive(Clone, Copy, Debug)]
-pub(crate) struct Measurements {
-    /// Compressed, on-the-wire body length, read before decompression.
-    pub(crate) compressed_len: u64,
-    /// Whether the request entered the decompression path.
-    pub(crate) decompression_applied: bool,
-    /// The declared `Content-Length`, or `None` when the header was absent.
-    pub(crate) declared_content_length: Option<u64>,
-}
+/// 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
+/// Build the intake router, `/api/v2/series` for payload assertions, others return 200 OK.
 pub fn build_router(hostname: Arc<str>) -> Router {
-    // 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(intake::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)
+        .merge(datadog::routes())
         .fallback(|| async { StatusCode::OK })
-        .with_state(hostname)
-}
-
-/// Buffer the body and record compressed size, encoding, and content-length before decompression
-async fn measure_compressed_size(req: Request, next: Next) -> Response {
-    let (parts, body) = req.into_parts();
-    let Ok(bytes) = axum::body::to_bytes(body, MAX_COMPRESSED_BODY_BYTES).await else {
-        return StatusCode::PAYLOAD_TOO_LARGE.into_response();
-    };
-    let len = bytes.len() as u64;
-    let applied = parts
-        .headers
-        .typed_get::<ContentEncoding>()
-        .is_some_and(|enc| enc.contains("deflate") || enc.contains("gzip") || enc.contains("zstd"));
-    let declared = parts.headers.typed_get::<ContentLength>().map(|cl| cl.0);
-    let mut req = Request::from_parts(parts, Body::from(bytes));
-    req.extensions_mut().insert(Measurements {
-        compressed_len: len,
-        decompression_applied: applied,
-        declared_content_length: declared,
-    });
-    next.run(req).await
+        .with_state(AppState { hostname })
 }

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

@@ -0,0 +1,26 @@
+//! Datadog-compatible intake routes.
+
+use axum::{extract::DefaultBodyLimit, middleware::from_fn, routing::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 metrics;
+
+/// 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().route("/api/v2/series", series)
+}

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

@@ -1,10 +1,9 @@
 //! `/api/v2/series` handler and validation pipeline.
 //!
-//! `handle_series` fires every payload property's assertion, walks the envelope,
-//! byte-size, and decode checks in order, then returns the first failure status or
-//! `202 Accepted`.
+//! `handle_series` fires every payload property's assertion. It walks the
+//! envelope, byte-size, and decode checks in order. It returns the first failure
+//! status, or `202 Accepted` when every check holds.
 
-use std::sync::Arc;
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use axum::{
@@ -15,7 +14,8 @@ use axum::{
 };
 use tracing::{debug, error};
 
-use crate::http::Measurements;
+use crate::http::middleware::Measurements;
+use crate::http::state::AppState;
 use crate::properties::payload::{bytes, envelope};
 use crate::series_observation::SeriesObservation;
 
@@ -56,9 +56,7 @@ impl IntoResponse for SeriesError {
 }
 
 /// Handler for `POST /api/v2/series`.
-pub(crate) async fn handle_series(
-    State(expected_hostname): State<Arc<str>>, request: Request,
-) -> Result<StatusCode, SeriesError> {
+pub(crate) async fn handle_series(State(state): State<AppState>, request: Request) -> Result<StatusCode, SeriesError> {
     // Pyld21 bounds points' timestamps against the intake wall clock at request receipt
     let now_secs = now_epoch_secs()?;
     let (parts, body) = request.into_parts();
@@ -96,7 +94,7 @@ pub(crate) async fn handle_series(
     let (observation, decode_ok) = SeriesObservation::decode(&body_bytes, decompression_applied);
 
     if let Some(observation) = observation.as_ref() {
-        observation.assert_payload_properties(now_secs, &expected_hostname);
+        observation.assert_payload_properties(now_secs, &state.hostname);
         debug!(
             bytes = body_bytes.len(),
             series = observation.series_len(),

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

@@ -0,0 +1,50 @@
+//! HTTP middleware for intake request measurement.
+//!
+//! The `/api/v2/series` route stacks measurement middleware ahead of the
+//! decompression layer so Pyld05 (compressed size), Pyld06 (uncompressed size),
+//! and Pyld22 (content-length) can read both the on-the-wire and decompressed
+//! body lengths, recorded as request extensions before `RequestDecompressionLayer`
+//! consumes the encoding headers.
+
+use axum::{
+    body::Body,
+    extract::Request,
+    http::StatusCode,
+    middleware::Next,
+    response::{IntoResponse, Response},
+};
+use headers::{ContentEncoding, ContentLength, HeaderMapExt};
+
+use super::MAX_COMPRESSED_BODY_BYTES;
+
+/// Wire measurements recorded before decompression, attached as a request extension for Pyld05/Pyld06/Pyld22.
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct Measurements {
+    /// Compressed, on-the-wire body length, read before decompression.
+    pub(crate) compressed_len: u64,
+    /// Whether the request entered the decompression path.
+    pub(crate) decompression_applied: bool,
+    /// The declared `Content-Length`, or `None` when the header was absent.
+    pub(crate) declared_content_length: Option<u64>,
+}
+
+/// Buffer the body and record compressed size, encoding, and content-length before decompression.
+pub(crate) async fn measure_compressed_size(req: Request, next: Next) -> Response {
+    let (parts, body) = req.into_parts();
+    let Ok(bytes) = axum::body::to_bytes(body, MAX_COMPRESSED_BODY_BYTES).await else {
+        return StatusCode::PAYLOAD_TOO_LARGE.into_response();
+    };
+    let len = bytes.len() as u64;
+    let applied = parts
+        .headers
+        .typed_get::<ContentEncoding>()
+        .is_some_and(|enc| enc.contains("deflate") || enc.contains("gzip") || enc.contains("zstd"));
+    let declared = parts.headers.typed_get::<ContentLength>().map(|cl| cl.0);
+    let mut req = Request::from_parts(parts, Body::from(bytes));
+    req.extensions_mut().insert(Measurements {
+        compressed_len: len,
+        decompression_applied: applied,
+        declared_content_length: declared,
+    });
+    next.run(req).await
+}

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

@@ -0,0 +1,10 @@
+//! Shared state carried by each HTTP router.
+
+use std::sync::Arc;
+
+/// Shared application state for one target's HTTP router.
+#[derive(Clone, Debug)]
+pub(crate) struct AppState {
+    /// Configured Agent hostname. Pyld17 checks each series host against it.
+    pub(crate) hostname: Arc<str>,
+}

test/antithesis/intake/src/lib.rs

@@ -37,6 +37,5 @@
 
 pub mod http;
 
-mod intake;
 mod properties;
 mod series_observation;