enhancement(antithesis): Introduce rig intake API (#1826)

## Summary

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. It will rapidly diverge from the
other intake, being focused on allowing driver and check claims when
running under Antithesis.

## Change Type
- [ ] Bug fix
- [ ] New feature
- [x] Non-functional (chore, refactoring, docs)
- [ ] Performance

Author: blt

Cargo.lock

@@ -33,6 +33,7 @@ members = [
   "lib/saluki-tls",
   "lib/stringtheory",
   "test/antithesis/harness",
+  "test/antithesis/intake",
 ]
 resolver = "2"
 
@@ -235,6 +236,7 @@ lru-slab = { version = "0.1.2", default-features = false }
 hickory-resolver = { version = "0.26", default-features = false }
 antithesis-instrumentation = { version = "0.1" }
 antithesis_sdk = { git = "https://github.com/antithesishq/antithesis-sdk-rust", rev = "78c9db56771f0f6b01cb5404765c5a96852c159c", default-features = false } # 0.2.8 is pinned to rand 0.8, rev version allows us to select workspace rand
+mime = "0.3"
 
 [patch.crates-io]
 # Forked version of `hyper-http-proxy` that removes an unused dependency on `rustls-native-certs`, which transitively depends

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 = { workspace = true }
+mime = { workspace = true }
+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 and non-empty                      |
+| 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        |