APMSVLS-494: add a standalone fake intake for local APM debugging

.github/workflows/rs_ci.yml

@@ -82,10 +82,11 @@ jobs:
         run: cargo clippy --workspace --all-targets --features default
       - working-directory: bottlecap
         run: cargo clippy --workspace --all-targets --no-default-features --features fips
-      # No other job compiles the test-mode feature: it gates test-only
-      # constructors that are absent from default and fips builds.
+      # No other job compiles the test-mode or fake-intake features: they gate
+      # test-only constructors and fixtures that are absent from default and
+      # fips builds.
       - working-directory: bottlecap
-        run: cargo clippy --workspace --all-targets --features default,test-mode
+        run: cargo clippy --workspace --all-targets --features default,test-mode,fake-intake
 
   build-all:
     name: Build All
@@ -144,7 +145,9 @@ jobs:
           tool: nextest@0.9
       - uses: mozilla-actions/sccache-action@7d986dd989559c6ecdb630a3fd2557667be217ad # v0.0.9
       - working-directory: bottlecap
-        run: cargo nextest run --workspace
+        # fake-intake enables the APM / DSM integration tests and the fake-intake
+        # unit tests; default and fips feature sets skip those targets.
+        run: cargo nextest run --workspace --features fake-intake
 
   format:
     name: Format

.gitlab/templates/pipeline.yaml.tpl

@@ -63,9 +63,23 @@ cargo clippy:
     # 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
-    # No other job compiles the test-mode feature: it gates test-only
-    # constructors that are absent from default and fips builds.
-    - cargo clippy --workspace --features default,test-mode
+    # No other job compiles the test-mode or fake-intake features: they gate
+    # test-only constructors and fixtures that are absent from default and
+    # fips builds.
+    - cargo clippy --workspace --all-targets --features default,test-mode,fake-intake
+
+cargo test:
+  stage: test
+  tags: ["arch:amd64"]
+  image: ${CI_DOCKER_TARGET_IMAGE}:${CI_DOCKER_TARGET_VERSION}
+  needs: []
+  script:
+    - apt-get update && apt-get install -y --fix-missing --no-install-recommends golang-go
+    - cd bottlecap
+    # fake-intake enables the APM / DSM integration tests and the fake-intake
+    # unit tests; default and fips feature sets skip those targets.
+    - cargo install cargo-nextest --locked
+    - cargo nextest run --workspace --features fake-intake
 
 {{ range $flavor := (ds "flavors").flavors }}
 

bottlecap/Cargo.toml

@@ -117,6 +117,19 @@ flate2 = { version = "1.1", default-features = false, features = ["rust_backend"
 [[bin]]
 name = "bottlecap"
 
+[[bin]]
+name = "fake-intake"
+path = "src/bin/fake-intake/main.rs"
+required-features = ["fake-intake"]
+
+[[test]]
+name = "apm_integration_test"
+required-features = ["fake-intake"]
+
+[[test]]
+name = "dsm_integration_test"
+required-features = ["fake-intake"]
+
 [profile.dev]
 debug = true   # same as debuginfo=2 and no stripping
 strip = false
@@ -180,6 +193,10 @@ fips = [
 # Not enabled in `default` or `fips`, so the items it gates do not appear
 # in production builds.
 test-mode = []
+# Builds the standalone fake-intake binary (`src/bin/fake-intake`) and the
+# embedded `fake_intake` library module used by the APM / DSM integration
+# tests. Not enabled in `default` or `fips`.
+fake-intake = ["tokio/signal"]
 
 [lints.rust]
 unexpected_cfgs = { level = "warn", check-cfg = ['cfg(coverage,coverage_nightly)'] }

bottlecap/README.md

@@ -3,6 +3,66 @@
 ## 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 fake-intake
+
+The `fake-intake` feature builds a standalone fake 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 fake-intake --features fake-intake
+```
+
+The same feature gates the embedded `bottlecap::fake_intake` module used by the `apm_integration_test` and `dsm_integration_test` integration tests, which are skipped by default-feature test runs.
+
+### Configuration
+
+| Variable | Default | Meaning |
+|---|---|---|
+| `FAKE_INTAKE_PORT` | `8127` | Loopback listener port; `0` picks a free port (the bound address is printed at startup) |
+| `FAKE_INTAKE_FAIL_STATS_FIRST_N` | `0` | Return HTTP 500 for the first N stats request attempts |
+| `FAKE_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 stored
+- `POST /api/v0.2/traces`: protobuf `AgentPayload` (gzip, zstd, and identity accepted): `202 Accepted` when stored
+- `POST /api/v0.1/pipeline_stats`: MessagePack DSM pipeline stats: `202 Accepted` when stored
+- Malformed payloads return `400 Bad Request` (unless inside the stats failure-injection window, which returns `500`)
+
+### Failure injection semantics
+
+With `FAKE_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 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 `FAKE_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 are serialized field by field because `AgentPayload` does not implement `Serialize` in `libdd-trace-protobuf` 4.0.1. DSM dumps contain only the fields the fixture decodes. Filenames are collision-resistant and never overwrite earlier dumps.
+
+### Smoke procedure
+
+Point the test-mode trace processor (currently in the separate worktree / PR that adds the `bottlecap-test-mode` binary) at a local fake intake:
+
+```bash
+cd ../<test-mode-worktree>/bottlecap
+cargo build --bin bottlecap-test-mode --features test-mode
+
+cd ../<this-checkout>/bottlecap
+cargo run --bin fake-intake --features fake-intake
+
+# in another shell:
+DD_APM_DD_URL=http://127.0.0.1:8127 \
+DD_LAMBDA_EXTENSION_COMPUTE_STATS=true \
+DD_SERVICE=fake-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 fake-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.
+
 ## 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/fake-intake/main.rs

@@ -0,0 +1,109 @@
+// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/
+// SPDX-License-Identifier: Apache-2.0
+
+//! Standalone fake Datadog APM intake for local debugging.
+//!
+//! Runs `bottlecap::fake_intake` 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:
+//!
+//! | Variable                        | Default | Purpose                                          |
+//! |---------------------------------|---------|--------------------------------------------------|
+//! | `FAKE_INTAKE_PORT`              | `8127`  | Loopback listener port (`0` = OS-assigned)       |
+//! | `FAKE_INTAKE_FAIL_STATS_FIRST_N`| `0`     | Return HTTP 500 for the first N stats attempts   |
+//! | `FAKE_INTAKE_DUMP_DIR`          | unset   | Write one JSON envelope per decoded request here |
+
+#![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 bottlecap::fake_intake::{FakeIntake, FakeIntakeOptions};
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let options = parse_options()?;
+    let intake = FakeIntake::start_with_options(options)
+        .await
+        .context("fake-intake: failed to start")?;
+
+    println!("fake-intake: listening on {}", intake.base_url());
+    println!("fake-intake: stats   endpoint POST {}", intake.stats_url());
+    println!("fake-intake: traces  endpoint POST {}", intake.traces_url());
+    println!(
+        "fake-intake: DSM     endpoint POST {}/api/v0.1/pipeline_stats",
+        intake.base_url()
+    );
+    println!("fake-intake: waiting for SIGINT or SIGTERM to shut down");
+
+    wait_for_shutdown()
+        .await
+        .context("fake-intake: failed to wait for shutdown signal")?;
+    drop(intake);
+    println!("fake-intake: shut down");
+    Ok(())
+}
+
+fn parse_options() -> anyhow::Result<FakeIntakeOptions> {
+    let port = match std::env::var("FAKE_INTAKE_PORT") {
+        Ok(raw) => raw
+            .parse::<u16>()
+            .with_context(|| format!("fake-intake: invalid FAKE_INTAKE_PORT value '{raw}', expected a port number between 0 and 65535"))?,
+        Err(std::env::VarError::NotPresent) => 8127,
+        Err(std::env::VarError::NotUnicode(raw)) => {
+            bail!("fake-intake: FAKE_INTAKE_PORT is not valid Unicode: {}", raw.display())
+        }
+    };
+
+    let fail_stats_first_n = match std::env::var("FAKE_INTAKE_FAIL_STATS_FIRST_N") {
+        Ok(raw) => raw
+            .parse::<usize>()
+            .with_context(|| format!("fake-intake: invalid FAKE_INTAKE_FAIL_STATS_FIRST_N value '{raw}', expected a non-negative integer"))?,
+        Err(std::env::VarError::NotPresent) => 0,
+        Err(std::env::VarError::NotUnicode(raw)) => {
+            bail!("fake-intake: FAKE_INTAKE_FAIL_STATS_FIRST_N is not valid Unicode: {}", raw.display())
+        }
+    };
+
+    let dump_dir = match std::env::var("FAKE_INTAKE_DUMP_DIR") {
+        Ok(raw) => Some(PathBuf::from(raw)),
+        Err(std::env::VarError::NotPresent) => None,
+        Err(std::env::VarError::NotUnicode(raw)) => {
+            bail!(
+                "fake-intake: FAKE_INTAKE_DUMP_DIR is not valid Unicode: {}",
+                raw.display()
+            )
+        }
+    };
+
+    Ok(FakeIntakeOptions {
+        port,
+        request_summaries: true,
+        fail_stats_first_n,
+        dump_dir,
+    })
+}
+
+async fn wait_for_shutdown() -> anyhow::Result<()> {
+    use tokio::signal::unix::{SignalKind, signal};
+
+    let mut sigterm = signal(SignalKind::terminate())
+        .context("fake-intake: failed to register SIGTERM handler")?;
+    tokio::select! {
+        _ = tokio::signal::ctrl_c() => {}
+        _ = sigterm.recv() => {}
+    }
+    Ok(())
+}