[APMSVLS-499] test infra: use shared datadog-mock-intake crate and add standalone mock-intake binary

.github/workflows/rs_ci.yml

@@ -84,9 +84,9 @@ jobs:
         run: cargo clippy --workspace --all-targets --features default
       - working-directory: bottlecap
         run: cargo clippy --workspace --all-targets --no-default-features --features fips
-      # Also lint the feature-gated test-mode binary.
+      # Also lint the feature-gated test-mode and mock-intake binaries.
       - working-directory: bottlecap
-        run: cargo clippy --workspace --all-targets --features default,test-mode
+        run: cargo clippy --workspace --all-targets --features default,test-mode,mock-intake
 
   build-all:
     name: Build All
@@ -149,6 +149,9 @@ jobs:
       - name: Test the test-mode binary
         working-directory: bottlecap
         run: cargo nextest run --features test-mode --bin bottlecap-test-mode
+      - name: Test the mock-intake binary
+        working-directory: bottlecap
+        run: cargo nextest run --features mock-intake --bin mock-intake
 
   format:
     name: Format

.gitlab/templates/pipeline.yaml.tpl

@@ -61,11 +61,10 @@ cargo clippy:
     - apt-get update && apt-get install -y --fix-missing --no-install-recommends golang-go
     - cd bottlecap
     # We need to do these separately because the fips feature is incompatible with the default feature.
-    - cargo clippy --workspace --features default
-    - cargo clippy --workspace --no-default-features --features fips
-    # The test-mode feature gates the bottlecap-test-mode binary via
-    # required-features, so no other job compiles it.
-    - cargo clippy --workspace --features default,test-mode
+    - cargo clippy --workspace --all-targets --features default
+    - cargo clippy --workspace --all-targets --no-default-features --features fips
+    # Also lint the feature-gated test-mode and mock-intake binaries.
+    - cargo clippy --workspace --all-targets --features default,test-mode,mock-intake
 
 {{ range $flavor := (ds "flavors").flavors }}
 

bottlecap/Cargo.toml

@@ -94,6 +94,10 @@ datadog-protos = { version = "0.1.0", default-features = false, git = "https://g
 libdd-capabilities = "4.0.1"
 libdd-common = { version = "10.0.0", default-features = false }
 libdd-trace-protobuf = "5.1.1"
+# Shared mock APM intake crate from serverless-components, linked only by the
+# standalone mock-intake binary behind the `mock-intake` feature. The APM /
+# DSM integration tests consume it as an ungated dev-dependency below.
+datadog-mock-intake = { git = "https://github.com/DataDog/serverless-components", rev = "fc89cb128a217385c2135a28fea158fc17d0ca40", optional = true }
 libdd-trace-utils = { version = "16.0.0", default-features = false, features = ["mini_agent"] }
 libdd-trace-normalization = "4.3.1"
 libdd-trace-obfuscation = { version = "12.0.0", default-features = false }
@@ -119,11 +123,11 @@ tempfile = "3.20"
 # tls_cert_file mitigation). Already present transitively via reqwest's
 # hyper-rustls, so promoting it to a dev-dep adds nothing to the shipped binary.
 tokio-rustls = { version = "0.26", default-features = false, features = ["logging", "tls12"] }
-# fake-intake test harness: decode msgpack+gzip stats payloads on arrival
-rmp-serde = { version = "1.3.1", default-features = false }
-flate2 = { version = "1.1", default-features = false, features = ["rust_backend"] }
 # fault-intake test harness: decode the `datadog-protos` metrics payloads
 protobuf = { version = "3.7", default-features = false }
+# Shared mock APM intake for the APM / DSM integration tests. Ungated so the
+# tests run in every feature set.
+datadog-mock-intake = { git = "https://github.com/DataDog/serverless-components", rev = "fc89cb128a217385c2135a28fea158fc17d0ca40" }
 
 [build-dependencies]
 # No external dependencies needed for the build script
@@ -135,6 +139,10 @@ name = "bottlecap"
 name = "bottlecap-test-mode"
 required-features = ["test-mode"]
 
+[[bin]]
+name = "mock-intake"
+required-features = ["mock-intake"]
+
 [profile.dev]
 debug = true   # same as debuginfo=2 and no stripping
 strip = false
@@ -201,6 +209,11 @@ fips = [
 # base dependency so the shipped extension does not carry the signal driver
 # for a binary only the test-mode build produces.
 test-mode = ["tokio/signal"]
+# Builds the standalone mock-intake binary (`src/bin/mock-intake`). The
+# APM / DSM integration tests do not require this feature: they use
+# `datadog-mock-intake` as a dev-dependency and run in every test build.
+# Not enabled in `default` or `fips`.
+mock-intake = ["dep:datadog-mock-intake", "tokio/signal"]
 
 [lints.rust]
 unexpected_cfgs = { level = "warn", check-cfg = ['cfg(coverage,coverage_nightly)'] }

bottlecap/LICENSE-3rdparty.csv

@@ -43,6 +43,7 @@ ctor,https://github.com/mmastrac/rust-ctor,Apache-2.0 OR MIT,Matt Mastracci <mat
 datadog-agent-config,https://github.com/DataDog/serverless-components,Apache-2.0,The datadog-agent-config Authors
 datadog-agent-trace-sampler,https://github.com/DataDog/serverless-components,Apache-2.0,The datadog-agent-trace-sampler Authors
 datadog-fips,https://github.com/DataDog/serverless-components,Apache-2.0,The datadog-fips Authors
+datadog-mock-intake,https://github.com/DataDog/serverless-components,Apache-2.0,The datadog-mock-intake Authors
 datadog-opentelemetry,https://github.com/DataDog/dd-trace-rs/tree/main/datadog-opentelemetry,Apache-2.0,Datadog Inc. <info@datadoghq.com>
 datadog-protos,https://github.com/DataDog/saluki,Apache-2.0,The datadog-protos Authors
 datadog-serverless-logging,https://github.com/DataDog/serverless-components,Apache-2.0,The datadog-serverless-logging Authors

bottlecap/README.md

@@ -3,6 +3,19 @@
 ## Development
 Use the `/scripts/build_bottlecap_layer.sh` and either publish it as a layer and test in Lambda or copy the binary into a container image and test there. Ask AJ or see the internal wiki for more.
 
+## Local APM debugging with mock-intake
+
+The `mock-intake` feature builds a standalone mock Datadog APM intake: a small HTTP server that accepts the same APM endpoints the extension flushes to, decodes the payloads, and reports them locally. It is gated behind a feature and is never part of `default` or `fips` production builds.
+
+```bash
+cd bottlecap
+cargo build --bin mock-intake --features mock-intake
+```
+
+The binary wraps the shared `datadog-mock-intake` crate from the `serverless-components` repository. The same crate backs the `apm_integration_test` and `dsm_integration_test` integration tests, which run in every test build via a dev-dependency.
+
+For configuration, endpoints, failure injection, request summaries, JSON dumps, and a smoke procedure, see [mock-intake/README.md](src/bin/mock-intake/README.md).
+
 ## Flush Strategies
 
 Bottlecap supports several flush strategies that control when and how observability data (metrics, logs, traces) is sent to Datadog. The strategy is configured via the `DD_SERVERLESS_FLUSH_STRATEGY` environment variable.

bottlecap/src/bin/mock-intake/README.md

@@ -0,0 +1,66 @@
+# mock-intake
+
+A standalone mock Datadog APM intake for local APM debugging: a small HTTP server that accepts the same APM endpoints the extension flushes to, decodes the payloads, and reports them locally.
+
+Built behind the `mock-intake` feature, which is never part of `default` or `fips` production builds:
+
+```bash
+cd bottlecap
+cargo build --bin mock-intake --features mock-intake
+```
+
+The binary wraps the shared `datadog-mock-intake` crate from the `serverless-components` repository. The standalone binary disables in-memory captures; request summaries and JSON dumps remain available. The same crate backs the `apm_integration_test` and `dsm_integration_test` integration tests, which retain captures by default and run in every test build via a dev-dependency.
+
+## Configuration
+
+| Variable | Default | Meaning |
+|---|---|---|
+| `MOCK_INTAKE_PORT` | `8127` | Loopback listener port; `0` picks a free port (the bound address is printed at startup) |
+| `MOCK_INTAKE_FAIL_STATS_FIRST_N` | `0` | Return HTTP 500 for the first N stats request attempts |
+| `MOCK_INTAKE_DUMP_DIR` | unset | Write one JSON envelope per successfully decoded request into this directory |
+
+Malformed configuration values cause a clear error and a nonzero exit. Bind failures and dump-directory creation failures are also reported instead of being ignored.
+
+## Endpoints and response codes
+
+- `POST /api/v0.2/stats`: MessagePack `StatsPayload` (gzip, zstd, and identity accepted): `202 Accepted` when decoded (unless failure injection rejects the attempt)
+- `POST /api/v0.2/traces`: protobuf `AgentPayload` (gzip, zstd, and identity accepted): `202 Accepted` when decoded
+- `POST /api/v0.1/pipeline_stats`: MessagePack DSM pipeline stats: `202 Accepted` when decoded
+- Malformed payloads return `400 Bad Request` (unless inside the stats failure-injection window, which returns `500`)
+- Request bodies above the 2 MiB wire-body limit return `413 Payload Too Large` without being decoded
+
+## Failure injection semantics
+
+With `MOCK_INTAKE_FAIL_STATS_FIRST_N=N`, the first N stats *attempts* return 500 and their payloads are not captured by the intake. Attempts are counted per server instance, atomically, so concurrent requests cannot exceed the limit. Rejected attempts are still decoded, summarized, and dumped, so a rejected attempt and its successful retry are each visible with their own status. Trace and DSM requests do not consume the rejection budget.
+
+## Request summaries and stats grouping
+
+Each handled request produces one summary line on stderr (prefix `mock-intake:`) with the request identity, endpoint, content encoding (`identity` when absent), response status, and decoded payload count. For stats requests, hits are grouped by the full aggregation key: client service/env/version plus origin context (hostname, container, tags, git/image metadata, process tags, aggregation mode) and grouped dimensions (service, name, resource, type, DB type, HTTP status, gRPC status, synthetics, span kind, trace-root flag, HTTP method/endpoint, service source, peer tags, deprecated span-derived primary tags, and additional metric tags). Measurements (hits, errors, duration, sketches) and delivery metadata are excluded from the key, so matching dimensions combine across time buckets. Tag lists are sorted in the summary without modifying captured payloads.
+
+## JSON dumps
+
+With `MOCK_INTAKE_DUMP_DIR` set, each successfully decoded request attempt (including rejected stats attempts) writes one JSON file containing the request identity, endpoint, encoding, response status, and decoded payload. Trace payloads use explicit field-by-field JSON serialization rather than a derived `Serialize` implementation. DSM dumps contain only the fields the crate decodes. Filenames are collision-resistant and never overwrite earlier dumps.
+
+## Smoke procedure
+
+Point the test-mode trace processor at a local mock intake. Build both binaries from this checkout, starting at the repository root:
+
+```bash
+cd bottlecap
+cargo build --bin bottlecap-test-mode --bin mock-intake --features test-mode,mock-intake
+dump_dir=$(mktemp -d)
+printf 'JSON dumps: %s\n' "$dump_dir"
+MOCK_INTAKE_DUMP_DIR="$dump_dir" ./target/debug/mock-intake
+```
+
+In another shell, start at the same checkout's repository root:
+
+```bash
+cd bottlecap
+DD_APM_DD_URL=http://127.0.0.1:8127 \
+DD_LAMBDA_EXTENSION_COMPUTE_STATS=true \
+DD_SERVICE=mock-intake-smoke DD_ENV=local DD_VERSION=smoke \
+./target/debug/bottlecap-test-mode
+```
+
+Unset `DD_SERVERLESS_FLUSH_STRATEGY` so flushing is manual, and make sure no ambient proxy intercepts localhost. Once `GET http://127.0.0.1:8126/info` succeeds, POST a MessagePack trace payload to `/v0.4/traces` (with `X-Datadog-Trace-Count`), then `POST /flush`. The `mock-intake:` summary lines should show the trace request and a stats request whose grouped hits match your spans, and the JSON dumps should contain both payloads.

bottlecap/src/bin/mock-intake/main.rs

@@ -0,0 +1,115 @@
+// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/
+// SPDX-License-Identifier: Apache-2.0
+
+//! Standalone mock Datadog APM intake for local debugging.
+//!
+//! Runs the shared `datadog-mock-intake` crate with request summaries enabled
+//! and accepts the same APM endpoints the Lambda extension flushes to. Point a
+//! tracer or the `bottlecap-test-mode` trace processor at it and inspect
+//! decoded payloads and JSON dumps locally. See `bottlecap/README.md` for usage.
+//!
+//! Environment variables are documented in `README.md` next to this file.
+
+#![deny(clippy::all)]
+#![deny(clippy::pedantic)]
+#![deny(clippy::unwrap_used)]
+#![deny(unused_extern_crates)]
+#![deny(unused_allocation)]
+#![deny(unused_assignments)]
+#![deny(unused_comparisons)]
+#![deny(unreachable_pub)]
+#![deny(missing_copy_implementations)]
+#![deny(missing_debug_implementations)]
+
+use std::path::PathBuf;
+
+use anyhow::{Context, bail};
+use datadog_mock_intake::{MockIntake, MockIntakeOptions};
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let options = parse_options()?;
+    let intake = MockIntake::start_with_options(options)
+        .await
+        .context("mock-intake: failed to start")?;
+
+    println!("mock-intake: listening on {}", intake.base_url());
+    println!("mock-intake: stats   endpoint POST {}", intake.stats_url());
+    println!("mock-intake: traces  endpoint POST {}", intake.traces_url());
+    println!(
+        "mock-intake: DSM     endpoint POST {}",
+        intake.pipeline_stats_url()
+    );
+    println!("mock-intake: waiting for SIGINT or SIGTERM to shut down");
+
+    wait_for_shutdown()
+        .await
+        .context("mock-intake: failed to wait for shutdown signal")?;
+    drop(intake);
+    println!("mock-intake: shut down");
+    Ok(())
+}
+
+/// Parse an environment variable via `FromStr`, falling back to `default` when
+/// unset. `expected` describes the accepted format for the error message.
+fn parse_env_or<T: std::str::FromStr>(name: &str, default: T, expected: &str) -> anyhow::Result<T> {
+    match std::env::var(name) {
+        Ok(raw) => raw.parse::<T>().map_err(|_| {
+            anyhow::anyhow!("mock-intake: invalid {name} value '{raw}', expected {expected}")
+        }),
+        Err(std::env::VarError::NotPresent) => Ok(default),
+        Err(std::env::VarError::NotUnicode(raw)) => bail!(
+            "mock-intake: {name} is not valid Unicode: {}",
+            raw.display()
+        ),
+    }
+}
+
+fn parse_options() -> anyhow::Result<MockIntakeOptions> {
+    let port = parse_env_or(
+        "MOCK_INTAKE_PORT",
+        8127u16,
+        "a port number between 0 and 65535",
+    )?;
+    let fail_stats_first_n = parse_env_or(
+        "MOCK_INTAKE_FAIL_STATS_FIRST_N",
+        0usize,
+        "a non-negative integer",
+    )?;
+
+    Ok(MockIntakeOptions {
+        port,
+        request_summaries: true,
+        fail_stats_first_n,
+        dump_dir: std::env::var_os("MOCK_INTAKE_DUMP_DIR").map(PathBuf::from),
+        disable_captures: true,
+    })
+}
+
+async fn wait_for_shutdown() -> anyhow::Result<()> {
+    use tokio::signal::unix::{SignalKind, signal};
+
+    let mut sigterm = signal(SignalKind::terminate())
+        .context("mock-intake: failed to register SIGTERM handler")?;
+    tokio::select! {
+        result = tokio::signal::ctrl_c() => result?,
+        _ = sigterm.recv() => {}
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+#[cfg_attr(coverage_nightly, coverage(off))]
+mod tests {
+    use super::parse_options;
+
+    #[test]
+    fn standalone_disables_captures() {
+        figment::Jail::expect_with(|jail| {
+            jail.clear_env();
+            let options = parse_options().expect("default options should parse");
+            assert!(options.disable_captures);
+            Ok(())
+        });
+    }
+}