Antithesis intake

This commit introduces an intake API for antithesis tests. It's a new
implementation compared to the pre-existing intake in the project as
this intake has different concerns. However, my hope is to eventually
merge the two together once this one is done baking, or supplant the
other depending on where we get to.

Author: blt

Cargo.lock

@@ -33,6 +33,7 @@ members = [
   "lib/saluki-tls",
   "lib/stringtheory",
   "test/antithesis/harness",
+  "test/antithesis/intake",
 ]
 resolver = "2"
 

Cargo.toml

@@ -4,7 +4,7 @@
 #
 # Build context is the repository root. Three named targets:
 #   - adp      : agent-data-plane built WITH Antithesis coverage instrumentation + SDK (the SUT)
-#   - intake   : datadog-intake mock Datadog intake (dependency)
+#   - intake   : antithesis-intake mock Datadog intake + W-property assertions (dependency)
 #   - workload : DogStatsD driver + test templates + setup-complete (the client)
 #
 # ADP is built native x86_64-unknown-linux-gnu (glibc), so no musl cross-compile headers are needed.
@@ -67,7 +67,7 @@ RUN --mount=type=cache,target=/adp/target,id=antithesis-adp-target \
     echo "Instrumentation symbols present."
 
 # ---------------------------------------------------------------------------
-# Build the correctness tools (datadog-intake) and the test-command binaries, uninstrumented.
+# Build the antithesis-intake mock and the test-command binaries, uninstrumented.
 # These are supporting harness components, not the SUT, so they need no coverage instrumentation.
 # ---------------------------------------------------------------------------
 FROM build-base AS tools-builder
@@ -77,14 +77,13 @@ RUN --mount=type=cache,target=/tools/target,id=antithesis-tools-target \
     --mount=type=cache,target=/root/.cargo/registry,id=cargo-registry \
     --mount=type=cache,target=/root/.cargo/git,id=cargo-git \
     cargo build --release \
-        --bin datadog-intake \
+        --bin antithesis-intake \
         --bin parallel_driver_send_dogstatsd --bin parallel_driver_sketchburst \
-        --bin finally_verify_delivery --bin eventually_adp_alive \
+        --bin eventually_adp_alive \
         --bin first_sample_config && \
-    cp /tools/target/release/datadog-intake /usr/local/bin/datadog-intake && \
+    cp /tools/target/release/antithesis-intake /usr/local/bin/antithesis-intake && \
     cp /tools/target/release/parallel_driver_send_dogstatsd /usr/local/bin/parallel_driver_send_dogstatsd && \
     cp /tools/target/release/parallel_driver_sketchburst /usr/local/bin/parallel_driver_sketchburst && \
-    cp /tools/target/release/finally_verify_delivery /usr/local/bin/finally_verify_delivery && \
     cp /tools/target/release/eventually_adp_alive /usr/local/bin/eventually_adp_alive && \
     cp /tools/target/release/first_sample_config /usr/local/bin/first_sample_config
 
@@ -118,12 +117,15 @@ ENTRYPOINT ["/entrypoint.sh"]
 CMD ["run"]
 
 # ---------------------------------------------------------------------------
-# Runtime: datadog-intake (mock Datadog intake dependency).
+# Runtime: antithesis-intake (mock Datadog intake + W-property assertions).
 # ---------------------------------------------------------------------------
 FROM ${APP_IMAGE} AS intake
 ENV NO_COLOR=1
-COPY --from=tools-builder /usr/local/bin/datadog-intake /usr/local/bin/datadog-intake
-ENTRYPOINT ["/usr/local/bin/datadog-intake"]
+# W17 resolves each series' host resource against this hostname. Keep it in sync
+# with `hostname:` in adp/datadog.yaml and the DD_HOSTNAME default in the harness.
+ENV DD_HOSTNAME=antithesis-adp
+COPY --from=tools-builder /usr/local/bin/antithesis-intake /usr/local/bin/antithesis-intake
+ENTRYPOINT ["/usr/local/bin/antithesis-intake"]
 
 # ---------------------------------------------------------------------------
 # Runtime: workload client (DogStatsD driver + test templates).
@@ -141,7 +143,6 @@ COPY test/antithesis/deploy/workload/test/ /opt/antithesis/test/
 COPY --from=tools-builder --chmod=755 /usr/local/bin/first_sample_config /opt/antithesis/test/v1/main/first_sample_config
 COPY --from=tools-builder --chmod=755 /usr/local/bin/parallel_driver_send_dogstatsd /opt/antithesis/test/v1/main/parallel_driver_send_dogstatsd
 COPY --from=tools-builder --chmod=755 /usr/local/bin/parallel_driver_sketchburst /opt/antithesis/test/v1/main/parallel_driver_sketchburst
-COPY --from=tools-builder --chmod=755 /usr/local/bin/finally_verify_delivery /opt/antithesis/test/v1/main/finally_verify_delivery
 COPY --from=tools-builder --chmod=755 /usr/local/bin/eventually_adp_alive /opt/antithesis/test/v1/main/eventually_adp_alive
 COPY --chmod=755 test/antithesis/deploy/workload/entrypoint.sh /entrypoint.sh
 ENTRYPOINT ["/entrypoint.sh"]

test/antithesis/deploy/Dockerfile

@@ -14,7 +14,7 @@ services:
     environment:
       NO_COLOR: "1"
     healthcheck:
-      # datadog-intake serves HTTP on :2049. /dev/tcp avoids needing curl in the image.
+      # antithesis-intake serves HTTP on :2049. /dev/tcp avoids needing curl in the image.
       test: ["CMD-SHELL", "bash -c 'exec 3<>/dev/tcp/localhost/2049'"]
       interval: 2s
       timeout: 2s

test/antithesis/deploy/docker-compose.yaml

@@ -0,0 +1,48 @@
+[package]
+name = "antithesis-intake"
+version = "0.1.0"
+edition = { workspace = true }
+license = { workspace = true }
+repository = { workspace = true }
+publish = false
+
+[[bin]]
+name = "antithesis-intake"
+path = "src/bin/intake.rs"
+
+[lints]
+workspace = true
+
+[dependencies]
+antithesis_sdk = { workspace = true, features = ["full"] }
+anyhow = { workspace = true, features = ["std"] }
+axum = { workspace = true, features = ["http1", "json", "tokio", "tracing"] }
+clap = { workspace = true, features = ["derive", "env", "error-context", "help", "std", "usage"] }
+datadog-protos = { workspace = true }
+headers = "0.4"
+mime = "0.3"
+protobuf = { workspace = true }
+serde_json = { workspace = true }
+tokio = { workspace = true, features = [
+  "macros",
+  "net",
+  "rt",
+  "rt-multi-thread",
+  "signal",
+] }
+tower = { workspace = true }
+tower-http = { workspace = true, features = [
+  "decompression-deflate",
+  "decompression-gzip",
+  "decompression-zstd",
+] }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true, features = [
+  "ansi",
+  "env-filter",
+  "fmt",
+  "local-time",
+  "registry",
+  "std",
+  "tracing-log",
+] }

test/antithesis/harness/src/bin/finally_verify_delivery.rs

@@ -0,0 +1,57 @@
+# A property asserting intake for Datadog Agent-like artifacts
+
+This intake asserts structural and aggregation properties on payloads from
+Datadog Agent-like programs and is intended to mimic the constraints set by
+Datadog Intake API. This implementation is forked from the other intake in this
+project and may be later merged back up, although the goals to make aggregation
+assertions here is invasive.
+
+### How This Project Works
+
+This document is the 'specification' for an abstract DogStatsD Agent. We assert
+that for any given input stream ADP emits to the intake data that is correctly
+shaped and, in a future update, that the aggregation model of ADP is accurate to
+the reference implementation of Datadog Agent DogStatsD.
+
+# Properties
+
+## Payloads
+
+The Agent emits outputs to Datadog intake endpoints as payloads. The current
+specification covers `/api/v2/series` only. The Agent also emits to
+`/api/v3/series`, which a future revision will add.
+
+In this section we define properties that hold for `/api/v2/series` payloads
+irrespective of load generation profile. Precisely, a 'payload' is an HTTP
+envelope wrap around the compressed bytes of a
+[`MetricPayload`](https://github.com/DataDog/agent-payload/blob/0a5f9ebbbe9c2a1f1e671467511f6189d3a3b443/proto/metrics/agent_payload.proto#L30-L72).
+
+Some properties reference rig-controlled parameters. `MaxTags(orgID)` and
+`MaxResources(orgID)` are per-org caps with defaults 100 and 500 respectively.
+`hostname` is the value the rig passes to the Agent via the `DD_HOSTNAME`
+environment variable. The Agent's hostname provider chain resolves this first,
+short-circuiting cloud-metadata and OS fallbacks.
+
+| Number | Category      | Name                   | Description                                                    |
+|--------|---------------|------------------------|----------------------------------------------------------------|
+| Pyld01     | Envelope      | Content-Type           | `Content-Type` in `{application/x-protobuf, application/json}` |
+| Pyld02     | Envelope      | Content-Encoding       | `Content-Encoding` in `{deflate, gzip, zstd, identity}`        |
+| Pyld03     | Envelope      | API Key                | `DD-Api-Key` header present iff non-empty key configured       |
+| Pyld05     | Bytes         | Compressed Size        | body < 500 KiB compressed                                      |
+| Pyld06     | Bytes         | Uncompressed Size      | body <= 5 MiB uncompressed                                     |
+| Pyld07     | MetricPayload | Decode                 | body decodes as v2 `MetricPayload` via `rust-protobuf`.        |
+| Pyld08     | MetricPayload | Point Count            | total points <= configured `serializer_max_series_points_per_payload` |
+| Pyld09     | MetricSeries  | Metric Non-Empty       | `MetricSeries.metric` is non-empty                             |
+| Pyld10    | MetricSeries  | Metric Length          | `len(metric) <= 350` bytes                                     |
+| Pyld11    | MetricSeries  | Metric Alphabetic      | `metric` contains at least one ASCII alphabetic char           |
+| Pyld12    | MetricSeries  | Type Enum              | `type` in `{COUNT, RATE, GAUGE}`                               |
+| Pyld13    | MetricSeries  | Tag Count              | `len(tags) <= MaxTags(orgID)`                                  |
+| Pyld14    | MetricSeries  | Tag Prefix Reserved    | no tag starts with `device:` or `dd.internal.resource:`        |
+| Pyld15    | MetricSeries  | Per-Series Point Count | `len(points) <=` configured `serializer_max_series_points_per_payload` |
+| Pyld16    | MetricSeries  | Origin Populated       | `origin.{product, category, service}` enum-valid               |
+| Pyld17    | Resource      | Host Resource Resolved | intake's `Host()` scan resolves a `(type="host", name=hostname)` resource |
+| Pyld18    | Resource      | Resource Count         | `len(resources) <= MaxResources(orgID)`                        |
+| Pyld19    | Resource      | Host Name Length       | host `name <= 255` bytes                                       |
+| Pyld20    | MetricPoint   | Value Not-NaN          | `value` is not NaN                                             |
+| Pyld21    | MetricPoint   | Timestamp Future Bound | `timestamp <= intake_now + 600s`                               |
+| Pyld22    | Bytes         | Content-Length         | `Content-Length` absent or value equals body byte count        |

test/antithesis/intake/Cargo.toml

@@ -0,0 +1,91 @@
+//! Binary entry point for the mock Datadog intake.
+
+#![deny(warnings)]
+
+use std::sync::Arc;
+
+use antithesis_intake::http::build_router;
+use antithesis_sdk::prelude::*;
+use anyhow::{Context, Result};
+use clap::Parser;
+use tokio::{net::TcpListener, sync::mpsc};
+use tracing::{error, info};
+use tracing_subscriber::{filter::LevelFilter, EnvFilter};
+
+/// Mock Datadog intake configuration, settable via environment variables
+#[derive(Parser)]
+#[command(name = "antithesis-intake")]
+struct Config {
+    /// Address the HTTP intake binds and serves on.
+    #[arg(long = "listen-addr", env = "LISTEN_ADDR", default_value = "0.0.0.0:2049")]
+    listen_addr: String,
+
+    /// Agent hostname, resolved by Pyld17 (keep in sync with `hostname` in `datadog.yaml`)
+    #[arg(long = "hostname", env = "DD_HOSTNAME", default_value = "antithesis-adp")]
+    hostname: String,
+}
+
+#[tokio::main]
+async fn main() {
+    tracing_subscriber::fmt()
+        .compact()
+        .with_env_filter(
+            EnvFilter::builder()
+                .with_default_directive(LevelFilter::INFO.into())
+                .from_env_lossy(),
+        )
+        .with_ansi(true)
+        .with_target(true)
+        .init();
+
+    antithesis_init();
+
+    if let Err(e) = serve(Config::parse()).await {
+        error!("{:?}", e);
+        std::process::exit(1);
+    }
+    info!("antithesis-intake stopped.");
+}
+
+/// Build the intake app, bind the listener, and serve until a shutdown signal.
+async fn serve(config: Config) -> Result<()> {
+    info!(hostname = %config.hostname, "Pyld17 resolves each series host against this hostname.");
+
+    let (shutdown_tx, mut shutdown_rx) = mpsc::channel(1);
+    spawn_signal_handlers(shutdown_tx).context("Failed to configure signal handlers.")?;
+
+    let listener = TcpListener::bind(&config.listen_addr)
+        .await
+        .context("Failed to bind HTTP intake listener.")?;
+    info!("antithesis-intake started: listening on {}.", config.listen_addr);
+
+    axum::serve(listener, build_router(Arc::from(config.hostname.as_str())))
+        .with_graceful_shutdown(async move { shutdown_rx.recv().await.unwrap_or(()) })
+        .await
+        .map_err(Into::into)
+}
+
+/// Spawn a task that signals shutdown on SIGINT or SIGTERM. Unix-only: this
+/// intake runs only in the Linux harness container.
+fn spawn_signal_handlers(shutdown_tx: mpsc::Sender<()>) -> Result<()> {
+    use tokio::{
+        select,
+        signal::unix::{signal, SignalKind},
+    };
+
+    let mut sigint_handler = signal(SignalKind::interrupt()).context("Failed to set up SIGINT handler.")?;
+    let mut sigterm_handler = signal(SignalKind::terminate()).context("Failed to set up SIGTERM handler.")?;
+
+    tokio::spawn(async move {
+        select! {
+            _ = sigint_handler.recv() => info!("Received SIGINT, shutting down..."),
+            _ = sigterm_handler.recv() => info!("Received SIGTERM, shutting down..."),
+        }
+
+        if let Err(e) = shutdown_tx.send(()).await {
+            error!("Failed to send shutdown signal: {:?}", e);
+        }
+    });
+
+    Ok(())
+}