From 4ac1097ee74bd8ea48652deca2f19314e424dbd0 Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Fri, 25 Sep 2026 13:05:10 -0400 Subject: [PATCH 01/19] Use shared datadog-mock-intake crate and add standalone fake-intake binary Replace the test-local APM intake fixture with the datadog-mock-intake crate from serverless-components (pinned to its first published revision). The APM and DSM integration tests use it as an ungated dev-dependency and run in every test build; the previous fixture is deleted. Add a standalone fake-intake binary behind a new fake-intake feature for local APM debugging: it runs the shared crate with request summaries, stats failure injection (FAKE_INTAKE_FAIL_STATS_FIRST_N), and JSON dumps (FAKE_INTAKE_DUMP_DIR). CI clippy now compiles the feature so the binary stays checked. --- .github/workflows/rs_ci.yml | 4 +- .gitlab/templates/pipeline.yaml.tpl | 5 +- bottlecap/Cargo.lock | 21 ++ bottlecap/Cargo.toml | 17 ++ bottlecap/README.md | 61 +++++ bottlecap/src/bin/fake-intake/main.rs | 109 ++++++++ bottlecap/tests/apm_integration_test.rs | 32 +-- bottlecap/tests/common/fake_intake.rs | 317 ------------------------ bottlecap/tests/dsm_integration_test.rs | 9 +- 9 files changed, 232 insertions(+), 343 deletions(-) create mode 100644 bottlecap/src/bin/fake-intake/main.rs delete mode 100644 bottlecap/tests/common/fake_intake.rs diff --git a/.github/workflows/rs_ci.yml b/.github/workflows/rs_ci.yml index 00a647cb7..9d04fa662 100644 --- a/.github/workflows/rs_ci.yml +++ b/.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 fake-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,fake-intake build-all: name: Build All diff --git a/.gitlab/templates/pipeline.yaml.tpl b/.gitlab/templates/pipeline.yaml.tpl index db89d1069..64b21a33f 100644 --- a/.gitlab/templates/pipeline.yaml.tpl +++ b/.gitlab/templates/pipeline.yaml.tpl @@ -63,9 +63,8 @@ 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 - # 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 + # Also lint the feature-gated test-mode and fake-intake binaries. + - cargo clippy --workspace --all-targets --features default,test-mode,fake-intake {{ range $flavor := (ds "flavors").flavors }} diff --git a/bottlecap/Cargo.lock b/bottlecap/Cargo.lock index b9b7c1b46..f9d8c13c5 100644 --- a/bottlecap/Cargo.lock +++ b/bottlecap/Cargo.lock @@ -475,6 +475,7 @@ dependencies = [ "datadog-agent-config", "datadog-agent-trace-sampler", "datadog-fips", + "datadog-mock-intake", "datadog-opentelemetry", "datadog-protos", "datadog-serverless-logging", @@ -848,6 +849,26 @@ dependencies = [ "tracing", ] +[[package]] +name = "datadog-mock-intake" +version = "0.1.0" +source = "git+https://github.com/DataDog/serverless-components?rev=c5f424b49cc6307f4204e016ad57d9973bdba243#c5f424b49cc6307f4204e016ad57d9973bdba243" +dependencies = [ + "bytes", + "flate2", + "http-body-util", + "hyper 1.8.1", + "hyper-util", + "libdd-trace-protobuf", + "prost 0.14.3", + "rmp-serde", + "serde", + "serde_json", + "thiserror 1.0.69", + "tokio", + "zstd", +] + [[package]] name = "datadog-opentelemetry" version = "0.6.0" diff --git a/bottlecap/Cargo.toml b/bottlecap/Cargo.toml index f437d5ad4..49c1db2e5 100644 --- a/bottlecap/Cargo.toml +++ b/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 fake-intake binary behind the `fake-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 = "e462105e918904e6b1ad2ebc153ae561cebe5a34", 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 } @@ -124,6 +128,9 @@ 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 = "e462105e918904e6b1ad2ebc153ae561cebe5a34" } [build-dependencies] # No external dependencies needed for the build script @@ -135,6 +142,11 @@ name = "bottlecap" name = "bottlecap-test-mode" required-features = ["test-mode"] +[[bin]] +name = "fake-intake" +path = "src/bin/fake-intake/main.rs" +required-features = ["fake-intake"] + [profile.dev] debug = true # same as debuginfo=2 and no stripping strip = false @@ -201,6 +213,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 fake-intake binary (`src/bin/fake-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`. +fake-intake = ["dep:datadog-mock-intake", "tokio/signal"] [lints.rust] unexpected_cfgs = { level = "warn", check-cfg = ['cfg(coverage,coverage_nightly)'] } diff --git a/bottlecap/README.md b/bottlecap/README.md index 60a9db35e..5ecd6d060 100644 --- a/bottlecap/README.md +++ b/bottlecap/README.md @@ -3,6 +3,67 @@ ## 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 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. + +### 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`) +- Request bodies above the 2 MiB wire-body limit return `413 Payload Too Large` without being decoded + +### 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 (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 `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 crate 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 ..//bottlecap +cargo build --bin bottlecap-test-mode --features test-mode + +cd ..//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 `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. + ## 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. diff --git a/bottlecap/src/bin/fake-intake/main.rs b/bottlecap/src/bin/fake-intake/main.rs new file mode 100644 index 000000000..ff5a2b7d9 --- /dev/null +++ b/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 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: +//! +//! | 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 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("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 { + let port = match std::env::var("FAKE_INTAKE_PORT") { + Ok(raw) => raw + .parse::() + .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::() + .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(MockIntakeOptions { + 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(()) +} diff --git a/bottlecap/tests/apm_integration_test.rs b/bottlecap/tests/apm_integration_test.rs index d4a50d61d..c1b5bb7e9 100644 --- a/bottlecap/tests/apm_integration_test.rs +++ b/bottlecap/tests/apm_integration_test.rs @@ -9,7 +9,7 @@ //! - `StatsFlusher` → msgpack+gzip `pb::StatsPayload` on `/api/v0.2/stats` //! - `TraceFlusher` → protobuf `pb::AgentPayload` on `/api/v0.2/traces` //! -//! Each test spins up a `FakeIntake`, points the flusher at it, triggers a +//! Each test spins up a `MockIntake`, points the flusher at it, triggers a //! flush, then decodes the captured payload and asserts on concrete fields. //! This is what APMSVLS-496 phase 1 unblocks: regression coverage for //! payload-level changes that `body_contains`-style mocks can't catch. @@ -43,10 +43,7 @@ use libdd_trace_utils::trace_utils::{TracerGenericTags, TracerHeaderTags}; use libdd_trace_utils::tracer_payload::TracerPayloadCollection; use tokio::sync::Mutex; -#[path = "common/fake_intake.rs"] -mod fake_intake; - -use fake_intake::FakeIntake; +use datadog_mock_intake::MockIntake; const DD_API_KEY: &str = "my_test_key"; @@ -105,7 +102,7 @@ fn endpoint_for(url: &str, api_key: &str) -> Endpoint { #[tokio::test] async fn stats_payload_roundtrip_through_fake_intake() { - let fake_intake = FakeIntake::start().await; + let fake_intake = MockIntake::start().await; let config = test_config(); let http_client = create_client(None, None, false).expect("failed to create http client"); @@ -217,7 +214,7 @@ async fn stats_payload_roundtrip_through_fake_intake() { #[tokio::test] async fn trace_payload_roundtrip_through_fake_intake() { - let fake_intake = FakeIntake::start().await; + let fake_intake = MockIntake::start().await; let config = test_config(); let http_client = create_client(None, None, false).expect("failed to create http client"); let endpoint = endpoint_for(&fake_intake.traces_url(), DD_API_KEY); @@ -329,6 +326,11 @@ struct PipelineOutcome { /// Drives `traces` through `SendingTraceProcessor::send_processed_traces` with the given /// `lambda_extension_compute_stats` / `client_computed_stats`, then flushes both the trace /// and stats pipelines into a fresh fake-intake and returns what it captured. +async fn run_processor_pipeline_with_traces( + compute_on_extension: bool, + client_computed_stats: bool, + traces: Vec>, +) -> PipelineOutcome { async fn run_processor_pipeline_with_traces( compute_on_extension: bool, client_computed_stats: bool, @@ -346,7 +348,7 @@ async fn run_processor_pipeline_with_settings( error_rare_sample_tracer_drop: bool, traces: Vec>, ) -> PipelineOutcome { - let fake_intake = FakeIntake::start().await; + let fake_intake = MockIntake::start().await; let config = Arc::new(Config { api_key: DD_API_KEY.to_string(), @@ -1061,7 +1063,7 @@ fn make_eligible_span(span_kind: &str, peer_meta: &[(&str, &str)]) -> pb::Span { /// Wire concentrator -> aggregator -> flusher pointed at the fake intake, feed in /// `spans`, force a flush, and return the single captured `StatsPayload`. async fn flush_spans_to_fake_intake( - fake_intake: &FakeIntake, + fake_intake: &MockIntake, config: Arc, spans: &[pb::Span], ) -> pb::StatsPayload { @@ -1107,7 +1109,7 @@ async fn flush_spans_to_fake_intake( /// the gap left by the in-memory concentrator unit tests, which never serialize. #[tokio::test] async fn stats_span_kind_through_fake_intake() { - let fake_intake = FakeIntake::start().await; + let fake_intake = MockIntake::start().await; let span = make_eligible_span("server", &[]); let payload = flush_spans_to_fake_intake(&fake_intake, test_config(), &[span]).await; @@ -1135,7 +1137,7 @@ async fn stats_span_kind_through_fake_intake() { /// the concentrator -> flusher -> intake path. #[tokio::test] async fn stats_peer_tags_through_fake_intake() { - let fake_intake = FakeIntake::start().await; + let fake_intake = MockIntake::start().await; let span = make_eligible_span( "client", &[("db.instance", "i-1234"), ("db.system", "postgres")], @@ -1183,7 +1185,7 @@ fn grouped_entries(payload: &pb::StatsPayload) -> Vec<&pb::ClientGroupedStats> { /// must not be exported as additional metric tags. #[tokio::test] async fn stats_additional_metric_tags_through_fake_intake() { - let fake_intake = FakeIntake::start().await; + let fake_intake = MockIntake::start().await; let config = config_from_env(&[ ("DD_API_KEY", DD_API_KEY), ("DD_SITE", "datadoghq.com"), @@ -1242,7 +1244,7 @@ async fn stats_additional_metric_tags_through_fake_intake() { /// group. Proves the gate affects the payload, not just the parsed config. #[tokio::test] async fn stats_additional_metric_tags_gated_off_through_fake_intake() { - let fake_intake = FakeIntake::start().await; + let fake_intake = MockIntake::start().await; let config = config_from_env(&[ ("DD_API_KEY", DD_API_KEY), ("DD_SITE", "datadoghq.com"), @@ -1281,7 +1283,7 @@ async fn stats_additional_metric_tags_gated_off_through_fake_intake() { /// the total hit count intact. #[tokio::test] async fn stats_additional_metric_tags_cardinality_limit_through_fake_intake() { - let fake_intake = FakeIntake::start().await; + let fake_intake = MockIntake::start().await; let config = config_from_env(&[ ("DD_API_KEY", DD_API_KEY), ("DD_SITE", "datadoghq.com"), @@ -1331,7 +1333,7 @@ async fn stats_additional_metric_tags_cardinality_limit_through_fake_intake() { /// combination, and a change in either key producing a new group. #[tokio::test] async fn stats_additional_metric_tags_multiple_keys_through_fake_intake() { - let fake_intake = FakeIntake::start().await; + let fake_intake = MockIntake::start().await; let config = config_from_env(&[ ("DD_API_KEY", DD_API_KEY), ("DD_SITE", "datadoghq.com"), diff --git a/bottlecap/tests/common/fake_intake.rs b/bottlecap/tests/common/fake_intake.rs deleted file mode 100644 index 1963f8a9a..000000000 --- a/bottlecap/tests/common/fake_intake.rs +++ /dev/null @@ -1,317 +0,0 @@ -// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ -// SPDX-License-Identifier: Apache-2.0 - -// This helper is `#[path]`-included by several test binaries; not every binary -// exercises every endpoint/accessor, so unused items here are expected. -#![allow(dead_code)] -//! In-process fake Datadog intake for APM payload-level integration tests. -//! -//! Spawns an axum server on a random local port that accepts the same APM -//! endpoints bottlecap flushes to, decodes msgpack / protobuf payloads on -//! arrival, and stores the decoded structs. Tests then call typed query -//! methods to assert on payload contents. -//! -//! Endpoints supported: -//! -//! - `POST /api/v0.2/stats`: msgpack, gzip-compressed, `pb::StatsPayload` -//! - `POST /api/v0.2/traces`: protobuf (optionally zstd-compressed), `pb::AgentPayload` -//! - `POST /api/v0.1/pipeline_stats`: msgpack (struct-as-map), gzip-compressed, DSM pipeline stats -//! -//! Prototype for APMSVLS-494 phase 1. If the API proves out, this file gets -//! extracted into the shared `datadog/apm-agent-parity-rs` repo in phase 2. - -use std::io::Read; -use std::sync::{Arc, Mutex}; - -use axum::{ - Router, - body::Bytes, - extract::State, - http::{HeaderMap, StatusCode}, - routing::post, -}; -use libdd_trace_protobuf::pb; -use prost::Message; -use serde::Deserialize; -use tokio::net::TcpListener; -use tokio::sync::oneshot; -use tokio::task::JoinHandle; - -/// A DSM pipeline-stats payload as it lands on `/api/v0.1/pipeline_stats`. -/// Only the fields tests assert on are decoded; serde ignores the rest -/// (including the `serde_bytes` latency sketches). -#[derive(Clone, Deserialize)] -pub struct PipelineStatsPayload { - #[serde(rename = "Env")] - pub env: String, - #[serde(rename = "Service")] - pub service: String, - #[serde(rename = "TracerVersion")] - pub tracer_version: String, - #[serde(rename = "Version")] - pub version: String, - #[serde(rename = "Tags")] - pub tags: Vec, - #[serde(rename = "Stats")] - pub stats: Vec, -} - -#[derive(Clone, Deserialize)] -pub struct PipelineStatsBucket { - #[serde(rename = "Stats")] - pub stats: Vec, -} - -#[derive(Clone, Deserialize)] -pub struct PipelineStatsPoint { - #[serde(rename = "Hash")] - pub hash: u64, - #[serde(rename = "ParentHash")] - pub parent_hash: u64, - #[serde(rename = "EdgeTags")] - pub edge_tags: Vec, -} - -/// Captured, decoded APM payloads for a single test run. -#[derive(Default)] -struct Captured { - stats: Vec, - traces: Vec, - pipeline_stats: Vec, -} - -/// Shared server state. The axum handlers write to the mutexes; tests read -/// via `FakeIntake::stats_payloads()` / `trace_payloads()`. -struct SharedState { - captured: Mutex, -} - -/// A running fake-intake server. Drop shuts it down. -pub struct FakeIntake { - base_url: String, - state: Arc, - shutdown_tx: Option>, - task: Option>, -} - -impl FakeIntake { - /// Bind to `127.0.0.1` on an OS-assigned port and start serving. - pub async fn start() -> Self { - let listener = TcpListener::bind("127.0.0.1:0") - .await - .expect("fake_intake: failed to bind listener"); - let addr = listener - .local_addr() - .expect("fake_intake: failed to read bound address"); - let base_url = format!("http://{addr}"); - - let state = Arc::new(SharedState { - captured: Mutex::new(Captured::default()), - }); - - let router = Router::new() - .route("/api/v0.2/stats", post(handle_stats)) - .route("/api/v0.2/traces", post(handle_traces)) - .route("/api/v0.1/pipeline_stats", post(handle_pipeline_stats)) - .with_state(Arc::clone(&state)); - - let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); - let task = tokio::spawn(async move { - axum::serve(listener, router) - .with_graceful_shutdown(async move { - let _ = shutdown_rx.await; - }) - .await - .expect("fake_intake: axum server error"); - }); - - Self { - base_url, - state, - shutdown_tx: Some(shutdown_tx), - task: Some(task), - } - } - - /// Full URL for the stats endpoint. - #[must_use] - pub fn stats_url(&self) -> String { - format!("{}/api/v0.2/stats", self.base_url) - } - - /// Full URL for the traces endpoint. - #[must_use] - pub fn traces_url(&self) -> String { - format!("{}/api/v0.2/traces", self.base_url) - } - - /// Base URL (scheme + host + port, no path). Use as the `apm_dd_url` for - /// components that build their own endpoint path (e.g. `DsmProcessor`). - #[must_use] - pub fn base_url(&self) -> String { - self.base_url.clone() - } - - /// All DSM pipeline-stats payloads captured so far, in arrival order. - #[must_use] - pub fn pipeline_stats_payloads(&self) -> Vec { - self.state - .captured - .lock() - .expect("fake_intake: pipeline_stats mutex poisoned") - .pipeline_stats - .clone() - } - - /// All `StatsPayload`s captured so far, in arrival order. - #[must_use] - pub fn stats_payloads(&self) -> Vec { - self.state - .captured - .lock() - .expect("fake_intake: stats mutex poisoned") - .stats - .clone() - } - - /// All `AgentPayload`s captured so far, in arrival order. - #[must_use] - pub fn trace_payloads(&self) -> Vec { - self.state - .captured - .lock() - .expect("fake_intake: traces mutex poisoned") - .traces - .clone() - } -} - -impl Drop for FakeIntake { - fn drop(&mut self) { - if let Some(tx) = self.shutdown_tx.take() { - let _ = tx.send(()); - } - if let Some(task) = self.task.take() { - task.abort(); - } - } -} - -async fn handle_stats( - State(state): State>, - headers: HeaderMap, - body: Bytes, -) -> StatusCode { - let decoded = match decompress(&headers, &body) { - Ok(d) => d, - Err(e) => { - eprintln!("{e}"); - return StatusCode::BAD_REQUEST; - } - }; - match rmp_serde::from_slice::(&decoded) { - Ok(payload) => { - state - .captured - .lock() - .expect("fake_intake: stats mutex poisoned") - .stats - .push(payload); - StatusCode::ACCEPTED - } - Err(err) => { - eprintln!("fake_intake: failed to decode StatsPayload msgpack: {err}"); - StatusCode::BAD_REQUEST - } - } -} - -async fn handle_traces( - State(state): State>, - headers: HeaderMap, - body: Bytes, -) -> StatusCode { - let decoded = match decompress(&headers, &body) { - Ok(d) => d, - Err(e) => { - eprintln!("{e}"); - return StatusCode::BAD_REQUEST; - } - }; - match pb::AgentPayload::decode(decoded.as_slice()) { - Ok(payload) => { - state - .captured - .lock() - .expect("fake_intake: traces mutex poisoned") - .traces - .push(payload); - StatusCode::ACCEPTED - } - Err(err) => { - eprintln!("fake_intake: failed to decode AgentPayload protobuf: {err}"); - StatusCode::BAD_REQUEST - } - } -} - -async fn handle_pipeline_stats( - State(state): State>, - headers: HeaderMap, - body: Bytes, -) -> StatusCode { - let decoded = match decompress(&headers, &body) { - Ok(d) => d, - Err(e) => { - eprintln!("{e}"); - return StatusCode::BAD_REQUEST; - } - }; - match rmp_serde::from_slice::(&decoded) { - Ok(payload) => { - state - .captured - .lock() - .expect("fake_intake: pipeline_stats mutex poisoned") - .pipeline_stats - .push(payload); - StatusCode::ACCEPTED - } - Err(err) => { - eprintln!("fake_intake: failed to decode pipeline stats msgpack: {err}"); - StatusCode::BAD_REQUEST - } - } -} - -/// Decompress a request body based on its `Content-Encoding` header. -/// Supports `gzip` and `zstd`. An unknown or absent encoding is treated as -/// identity: the body is returned unchanged. -fn decompress(headers: &HeaderMap, body: &Bytes) -> Result, String> { - let encoding = headers - .get("content-encoding") - .and_then(|v| v.to_str().ok()) - .unwrap_or("") - .to_ascii_lowercase(); - - match encoding.as_str() { - "gzip" => { - let mut decoder = flate2::read::GzDecoder::new(body.as_ref()); - let mut out = Vec::new(); - decoder - .read_to_end(&mut out) - .map_err(|e| format!("fake_intake: gzip decode failed: {e}"))?; - Ok(out) - } - "zstd" => zstd::stream::decode_all(body.as_ref()) - .map_err(|e| format!("fake_intake: zstd decode failed: {e}")), - _ => { - if !encoding.is_empty() { - eprintln!( - "fake_intake: unrecognized Content-Encoding '{encoding}', treating as identity" - ); - } - Ok(body.to_vec()) - } - } -} diff --git a/bottlecap/tests/dsm_integration_test.rs b/bottlecap/tests/dsm_integration_test.rs index a086a8ebe..ac1d2258f 100644 --- a/bottlecap/tests/dsm_integration_test.rs +++ b/bottlecap/tests/dsm_integration_test.rs @@ -8,7 +8,7 @@ //! `DsmProcessor::record_consume` → `drain_into_proxy` (aggregate + serialize + //! gzip) → `ProxyFlusher::flush` → `POST /api/v0.1/pipeline_stats`. //! -//! The test spins up a `FakeIntake`, points the `DsmProcessor` at it via the +//! The test spins up a `MockIntake`, points the `DsmProcessor` at it via the //! `apm_dd_url` argument (mirroring how `DD_APM_DD_URL` flows through in //! production), triggers a flush, then decodes the captured msgpack payload and //! asserts on concrete fields. Unit tests stop at the `ProxyRequest` boundary; @@ -28,10 +28,7 @@ use datadog_fips::reqwest_adapter::create_reqwest_client_builder; use dogstatsd::api_key::ApiKeyFactory; use tokio::sync::Mutex; -#[path = "common/fake_intake.rs"] -mod fake_intake; - -use fake_intake::FakeIntake; +use datadog_mock_intake::MockIntake; const DD_API_KEY: &str = "my_test_key"; @@ -56,7 +53,7 @@ fn tags_provider(config: &Arc) -> Arc { #[tokio::test] async fn dsm_pipeline_stats_roundtrip_through_fake_intake() { - let fake_intake = FakeIntake::start().await; + let fake_intake = MockIntake::start().await; let config = test_config(); let http_client = create_reqwest_client_builder() .expect("failed to create reqwest client builder") From 05694a7b9cdaf3dfce3ea105d951d4f5931cae80 Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Fri, 25 Sep 2026 15:29:07 -0400 Subject: [PATCH 02/19] Rename the fake-intake binary, feature, and env vars to mock-intake The standalone binary is now mock-intake, gated by the mock-intake feature, configured with MOCK_INTAKE_PORT, MOCK_INTAKE_FAIL_STATS_FIRST_N, and MOCK_INTAKE_DUMP_DIR. This aligns the binary with the datadog-mock-intake crate it wraps, whose summaries already used the mock-intake prefix. The mock-intake crate is now a git dependency, so LICENSE-3rdparty.csv gains its first-party record. --- .github/workflows/rs_ci.yml | 4 +- .gitlab/templates/pipeline.yaml.tpl | 4 +- bottlecap/Cargo.lock | 2 +- bottlecap/Cargo.toml | 12 +++--- bottlecap/LICENSE-3rdparty.csv | 1 + bottlecap/README.md | 22 +++++----- .../bin/{fake-intake => mock-intake}/main.rs | 42 +++++++++---------- 7 files changed, 44 insertions(+), 43 deletions(-) rename bottlecap/src/bin/{fake-intake => mock-intake}/main.rs (68%) diff --git a/.github/workflows/rs_ci.yml b/.github/workflows/rs_ci.yml index 9d04fa662..1a6bbef4e 100644 --- a/.github/workflows/rs_ci.yml +++ b/.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 and fake-intake binaries. + # Also lint the feature-gated test-mode and mock-intake binaries. - working-directory: bottlecap - run: cargo clippy --workspace --all-targets --features default,test-mode,fake-intake + run: cargo clippy --workspace --all-targets --features default,test-mode,mock-intake build-all: name: Build All diff --git a/.gitlab/templates/pipeline.yaml.tpl b/.gitlab/templates/pipeline.yaml.tpl index 64b21a33f..2c3f88eb4 100644 --- a/.gitlab/templates/pipeline.yaml.tpl +++ b/.gitlab/templates/pipeline.yaml.tpl @@ -63,8 +63,8 @@ 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 - # Also lint the feature-gated test-mode and fake-intake binaries. - - cargo clippy --workspace --all-targets --features default,test-mode,fake-intake + # 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 }} diff --git a/bottlecap/Cargo.lock b/bottlecap/Cargo.lock index f9d8c13c5..9fd478ce2 100644 --- a/bottlecap/Cargo.lock +++ b/bottlecap/Cargo.lock @@ -852,7 +852,7 @@ dependencies = [ [[package]] name = "datadog-mock-intake" version = "0.1.0" -source = "git+https://github.com/DataDog/serverless-components?rev=c5f424b49cc6307f4204e016ad57d9973bdba243#c5f424b49cc6307f4204e016ad57d9973bdba243" +source = "git+https://github.com/DataDog/serverless-components?rev=37f5b32a853f82a00523051827af074662f3e8fa#37f5b32a853f82a00523051827af074662f3e8fa" dependencies = [ "bytes", "flate2", diff --git a/bottlecap/Cargo.toml b/bottlecap/Cargo.toml index 49c1db2e5..b12337400 100644 --- a/bottlecap/Cargo.toml +++ b/bottlecap/Cargo.toml @@ -95,7 +95,7 @@ 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 fake-intake binary behind the `fake-intake` feature. The APM / +# 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 = "e462105e918904e6b1ad2ebc153ae561cebe5a34", optional = true } libdd-trace-utils = { version = "16.0.0", default-features = false, features = ["mini_agent"] } @@ -143,9 +143,9 @@ name = "bottlecap-test-mode" required-features = ["test-mode"] [[bin]] -name = "fake-intake" -path = "src/bin/fake-intake/main.rs" -required-features = ["fake-intake"] +name = "mock-intake" +path = "src/bin/mock-intake/main.rs" +required-features = ["mock-intake"] [profile.dev] debug = true # same as debuginfo=2 and no stripping @@ -213,11 +213,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 fake-intake binary (`src/bin/fake-intake`). The +# 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`. -fake-intake = ["dep:datadog-mock-intake", "tokio/signal"] +mock-intake = ["dep:datadog-mock-intake", "tokio/signal"] [lints.rust] unexpected_cfgs = { level = "warn", check-cfg = ['cfg(coverage,coverage_nightly)'] } diff --git a/bottlecap/LICENSE-3rdparty.csv b/bottlecap/LICENSE-3rdparty.csv index 6682f5859..e7d2a1da0 100644 --- a/bottlecap/LICENSE-3rdparty.csv +++ b/bottlecap/LICENSE-3rdparty.csv @@ -43,6 +43,7 @@ ctor,https://github.com/mmastrac/rust-ctor,Apache-2.0 OR MIT,Matt Mastracci 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 diff --git a/bottlecap/README.md b/bottlecap/README.md index 5ecd6d060..dd5186f69 100644 --- a/bottlecap/README.md +++ b/bottlecap/README.md @@ -3,13 +3,13 @@ ## 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 +## Local APM debugging with mock-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. +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 fake-intake --features fake-intake +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. @@ -18,9 +18,9 @@ The binary wraps the shared `datadog-mock-intake` crate from the `serverless-com | 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 | +| `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. @@ -34,7 +34,7 @@ Malformed configuration values cause a clear error and a nonzero exit. Bind fail ### 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. +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 @@ -42,23 +42,23 @@ Each handled request produces one summary line on stderr (prefix `mock-intake:`) ### 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 crate decodes. Filenames are collision-resistant and never overwrite earlier 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 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 crate 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: +Point the test-mode trace processor (currently in the separate worktree / PR that adds the `bottlecap-test-mode` binary) at a local mock intake: ```bash cd ..//bottlecap cargo build --bin bottlecap-test-mode --features test-mode cd ..//bottlecap -cargo run --bin fake-intake --features fake-intake +cargo run --bin mock-intake --features mock-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 \ +DD_SERVICE=mock-intake-smoke DD_ENV=local DD_VERSION=smoke \ ./target/debug/bottlecap-test-mode ``` diff --git a/bottlecap/src/bin/fake-intake/main.rs b/bottlecap/src/bin/mock-intake/main.rs similarity index 68% rename from bottlecap/src/bin/fake-intake/main.rs rename to bottlecap/src/bin/mock-intake/main.rs index ff5a2b7d9..9bc379d4c 100644 --- a/bottlecap/src/bin/fake-intake/main.rs +++ b/bottlecap/src/bin/mock-intake/main.rs @@ -1,7 +1,7 @@ // Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 -//! Standalone fake Datadog APM intake for local debugging. +//! 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 @@ -12,9 +12,9 @@ //! //! | 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 | +//! | `MOCK_INTAKE_PORT` | `8127` | Loopback listener port (`0` = OS-assigned) | +//! | `MOCK_INTAKE_FAIL_STATS_FIRST_N`| `0` | Return HTTP 500 for the first N stats attempts | +//! | `MOCK_INTAKE_DUMP_DIR` | unset | Write one JSON envelope per decoded request here | #![deny(clippy::all)] #![deny(clippy::pedantic)] @@ -37,52 +37,52 @@ async fn main() -> anyhow::Result<()> { let options = parse_options()?; let intake = MockIntake::start_with_options(options) .await - .context("fake-intake: failed to start")?; + .context("mock-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!("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!( - "fake-intake: DSM endpoint POST {}/api/v0.1/pipeline_stats", + "mock-intake: DSM endpoint POST {}/api/v0.1/pipeline_stats", intake.base_url() ); - println!("fake-intake: waiting for SIGINT or SIGTERM to shut down"); + println!("mock-intake: waiting for SIGINT or SIGTERM to shut down"); wait_for_shutdown() .await - .context("fake-intake: failed to wait for shutdown signal")?; + .context("mock-intake: failed to wait for shutdown signal")?; drop(intake); - println!("fake-intake: shut down"); + println!("mock-intake: shut down"); Ok(()) } fn parse_options() -> anyhow::Result { - let port = match std::env::var("FAKE_INTAKE_PORT") { + let port = match std::env::var("MOCK_INTAKE_PORT") { Ok(raw) => raw .parse::() - .with_context(|| format!("fake-intake: invalid FAKE_INTAKE_PORT value '{raw}', expected a port number between 0 and 65535"))?, + .with_context(|| format!("mock-intake: invalid MOCK_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()) + bail!("mock-intake: MOCK_INTAKE_PORT is not valid Unicode: {}", raw.display()) } }; - let fail_stats_first_n = match std::env::var("FAKE_INTAKE_FAIL_STATS_FIRST_N") { + let fail_stats_first_n = match std::env::var("MOCK_INTAKE_FAIL_STATS_FIRST_N") { Ok(raw) => raw .parse::() - .with_context(|| format!("fake-intake: invalid FAKE_INTAKE_FAIL_STATS_FIRST_N value '{raw}', expected a non-negative integer"))?, + .with_context(|| format!("mock-intake: invalid MOCK_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()) + bail!("mock-intake: MOCK_INTAKE_FAIL_STATS_FIRST_N is not valid Unicode: {}", raw.display()) } }; - let dump_dir = match std::env::var("FAKE_INTAKE_DUMP_DIR") { + let dump_dir = match std::env::var("MOCK_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: {}", + "mock-intake: MOCK_INTAKE_DUMP_DIR is not valid Unicode: {}", raw.display() ) } @@ -100,7 +100,7 @@ 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")?; + .context("mock-intake: failed to register SIGTERM handler")?; tokio::select! { _ = tokio::signal::ctrl_c() => {} _ = sigterm.recv() => {} From c7d34bd2b1e111f564c7aa119d74182bfc622eef Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Tue, 29 Sep 2026 13:26:05 -0400 Subject: [PATCH 03/19] Clean up mock-intake env parsing and dev-deps Dedupe the duplicated env-var parsing in the mock-intake binary into a generic parse_env_or helper with identical error messages, and drop the rmp-serde/flate2 dev-dependencies left over from when the harness decoded stats payloads itself; the shared datadog-mock-intake crate handles that now. --- bottlecap/Cargo.toml | 3 -- bottlecap/src/bin/mock-intake/main.rs | 41 ++++++++++++++++----------- 2 files changed, 25 insertions(+), 19 deletions(-) diff --git a/bottlecap/Cargo.toml b/bottlecap/Cargo.toml index b12337400..c44226c9c 100644 --- a/bottlecap/Cargo.toml +++ b/bottlecap/Cargo.toml @@ -123,9 +123,6 @@ 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 diff --git a/bottlecap/src/bin/mock-intake/main.rs b/bottlecap/src/bin/mock-intake/main.rs index 9bc379d4c..8bc745477 100644 --- a/bottlecap/src/bin/mock-intake/main.rs +++ b/bottlecap/src/bin/mock-intake/main.rs @@ -56,26 +56,35 @@ async fn main() -> anyhow::Result<()> { Ok(()) } -fn parse_options() -> anyhow::Result { - let port = match std::env::var("MOCK_INTAKE_PORT") { - Ok(raw) => raw - .parse::() - .with_context(|| format!("mock-intake: invalid MOCK_INTAKE_PORT value '{raw}', expected a port number between 0 and 65535"))?, - Err(std::env::VarError::NotPresent) => 8127, +/// 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(name: &str, default: T, expected: &str) -> anyhow::Result { + let value = match std::env::var(name) { + Ok(raw) => raw.parse::().map_err(|_| { + anyhow::anyhow!("mock-intake: invalid {name} value '{raw}', expected {expected}") + })?, + Err(std::env::VarError::NotPresent) => default, Err(std::env::VarError::NotUnicode(raw)) => { - bail!("mock-intake: MOCK_INTAKE_PORT is not valid Unicode: {}", raw.display()) + bail!( + "mock-intake: {name} is not valid Unicode: {}", + raw.display() + ) } }; + Ok(value) +} - let fail_stats_first_n = match std::env::var("MOCK_INTAKE_FAIL_STATS_FIRST_N") { - Ok(raw) => raw - .parse::() - .with_context(|| format!("mock-intake: invalid MOCK_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!("mock-intake: MOCK_INTAKE_FAIL_STATS_FIRST_N is not valid Unicode: {}", raw.display()) - } - }; +fn parse_options() -> anyhow::Result { + 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", + )?; let dump_dir = match std::env::var("MOCK_INTAKE_DUMP_DIR") { Ok(raw) => Some(PathBuf::from(raw)), From 7a6692262a1a36895d817b4117c35a831daa6347 Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Tue, 29 Sep 2026 14:07:15 -0400 Subject: [PATCH 04/19] Point mock-intake smoke docs at test-mode PR --- bottlecap/README.md | 2 +- bottlecap/src/bin/mock-intake/main.rs | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/bottlecap/README.md b/bottlecap/README.md index dd5186f69..3298688b1 100644 --- a/bottlecap/README.md +++ b/bottlecap/README.md @@ -46,7 +46,7 @@ With `MOCK_INTAKE_DUMP_DIR` set, each successfully decoded request attempt (incl ### Smoke procedure -Point the test-mode trace processor (currently in the separate worktree / PR that adds the `bottlecap-test-mode` binary) at a local mock intake: +Point the test-mode trace processor at a local mock intake. The `bottlecap-test-mode` binary is added in [#1216](https://github.com/DataDog/datadog-lambda-extension/pull/1216) (not yet merged); build it from a checkout of that branch: ```bash cd ..//bottlecap diff --git a/bottlecap/src/bin/mock-intake/main.rs b/bottlecap/src/bin/mock-intake/main.rs index 8bc745477..8fe216394 100644 --- a/bottlecap/src/bin/mock-intake/main.rs +++ b/bottlecap/src/bin/mock-intake/main.rs @@ -5,7 +5,8 @@ //! //! 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 +//! tracer or the `bottlecap-test-mode` trace processor (added in +//! DataDog/datadog-lambda-extension#1216) at it and inspect //! decoded payloads and JSON dumps locally. See `bottlecap/README.md` for usage. //! //! Environment variables: From 657cd30272c61285a1dbc2b9b1236ebea638661a Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Tue, 29 Sep 2026 14:08:40 -0400 Subject: [PATCH 05/19] Use pipeline_stats_url() in mock-intake startup output --- bottlecap/src/bin/mock-intake/main.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/bottlecap/src/bin/mock-intake/main.rs b/bottlecap/src/bin/mock-intake/main.rs index 8fe216394..c772e12a4 100644 --- a/bottlecap/src/bin/mock-intake/main.rs +++ b/bottlecap/src/bin/mock-intake/main.rs @@ -44,8 +44,8 @@ async fn main() -> anyhow::Result<()> { println!("mock-intake: stats endpoint POST {}", intake.stats_url()); println!("mock-intake: traces endpoint POST {}", intake.traces_url()); println!( - "mock-intake: DSM endpoint POST {}/api/v0.1/pipeline_stats", - intake.base_url() + "mock-intake: DSM endpoint POST {}", + intake.pipeline_stats_url() ); println!("mock-intake: waiting for SIGINT or SIGTERM to shut down"); From ed521cfbda238d225155de2a98cfceac928f792b Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Tue, 29 Sep 2026 14:09:07 -0400 Subject: [PATCH 06/19] Accept non-UTF-8 paths for MOCK_INTAKE_DUMP_DIR --- bottlecap/src/bin/mock-intake/main.rs | 11 +---------- 1 file changed, 1 insertion(+), 10 deletions(-) diff --git a/bottlecap/src/bin/mock-intake/main.rs b/bottlecap/src/bin/mock-intake/main.rs index c772e12a4..11c843df4 100644 --- a/bottlecap/src/bin/mock-intake/main.rs +++ b/bottlecap/src/bin/mock-intake/main.rs @@ -87,16 +87,7 @@ fn parse_options() -> anyhow::Result { "a non-negative integer", )?; - let dump_dir = match std::env::var("MOCK_INTAKE_DUMP_DIR") { - Ok(raw) => Some(PathBuf::from(raw)), - Err(std::env::VarError::NotPresent) => None, - Err(std::env::VarError::NotUnicode(raw)) => { - bail!( - "mock-intake: MOCK_INTAKE_DUMP_DIR is not valid Unicode: {}", - raw.display() - ) - } - }; + let dump_dir = std::env::var_os("MOCK_INTAKE_DUMP_DIR").map(PathBuf::from); Ok(MockIntakeOptions { port, From 5bf9e7d7174a248dd3f71534ce7b43de91abc0d3 Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Tue, 29 Sep 2026 14:09:49 -0400 Subject: [PATCH 07/19] Lint all targets in every GitLab clippy run --- .gitlab/templates/pipeline.yaml.tpl | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.gitlab/templates/pipeline.yaml.tpl b/.gitlab/templates/pipeline.yaml.tpl index 2c3f88eb4..7f4f6dd57 100644 --- a/.gitlab/templates/pipeline.yaml.tpl +++ b/.gitlab/templates/pipeline.yaml.tpl @@ -61,8 +61,8 @@ 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 + - 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 From 553d1a175302f1f02574b14938572aedfa2bded3 Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Tue, 29 Sep 2026 14:10:39 -0400 Subject: [PATCH 08/19] Rename fake intake references in tests to mock intake --- bottlecap/tests/apm_integration_test.rs | 105 +++++++++++------------- bottlecap/tests/dsm_integration_test.rs | 16 ++-- 2 files changed, 58 insertions(+), 63 deletions(-) diff --git a/bottlecap/tests/apm_integration_test.rs b/bottlecap/tests/apm_integration_test.rs index c1b5bb7e9..4b45095a3 100644 --- a/bottlecap/tests/apm_integration_test.rs +++ b/bottlecap/tests/apm_integration_test.rs @@ -1,7 +1,7 @@ // Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 -//! Payload-level APM integration tests using the in-process fake-intake. +//! Payload-level APM integration tests using the in-process mock-intake. //! //! Covers the two flush paths bottlecap uses to forward APM data to the //! Datadog backend: @@ -101,8 +101,8 @@ fn endpoint_for(url: &str, api_key: &str) -> Endpoint { } #[tokio::test] -async fn stats_payload_roundtrip_through_fake_intake() { - let fake_intake = MockIntake::start().await; +async fn stats_payload_roundtrip_through_mock_intake() { + let mock_intake = MockIntake::start().await; let config = test_config(); let http_client = create_client(None, None, false).expect("failed to create http client"); @@ -123,7 +123,7 @@ async fn stats_payload_roundtrip_through_fake_intake() { aggregator, config, http_client, - fake_intake.stats_url(), + mock_intake.stats_url(), ); let client_stats = pb::ClientStatsPayload { @@ -134,13 +134,13 @@ async fn stats_payload_roundtrip_through_fake_intake() { tracer_version: "test-tracer".to_string(), runtime_id: "00000000-0000-0000-0000-000000000001".to_string(), sequence: 7, - service: "fake-intake-test-service".to_string(), + service: "mock-intake-test-service".to_string(), stats: vec![pb::ClientStatsBucket { start: 1_700_000_000_000_000_000, duration: 10_000_000_000, agent_time_shift: 0, stats: vec![pb::ClientGroupedStats { - service: "fake-intake-test-service".to_string(), + service: "mock-intake-test-service".to_string(), name: "handler".to_string(), resource: "GET /fake".to_string(), r#type: "web".to_string(), @@ -179,7 +179,7 @@ async fn stats_payload_roundtrip_through_fake_intake() { "stats send reported a retry-able failure: {failed:?}", ); - let captured = fake_intake.stats_payloads(); + let captured = mock_intake.stats_payloads(); assert_eq!(captured.len(), 1, "expected exactly one StatsPayload"); let payload = &captured[0]; @@ -194,7 +194,7 @@ async fn stats_payload_roundtrip_through_fake_intake() { assert_eq!(inner.hostname, ""); assert_eq!(inner.env, "test-env"); assert_eq!(inner.version, "1.2.3"); - assert_eq!(inner.service, "fake-intake-test-service"); + assert_eq!(inner.service, "mock-intake-test-service"); assert_eq!(inner.sequence, 7); assert_eq!(inner.stats.len(), 1); let bucket = &inner.stats[0]; @@ -213,17 +213,17 @@ async fn stats_payload_roundtrip_through_fake_intake() { } #[tokio::test] -async fn trace_payload_roundtrip_through_fake_intake() { - let fake_intake = MockIntake::start().await; +async fn trace_payload_roundtrip_through_mock_intake() { + let mock_intake = MockIntake::start().await; let config = test_config(); let http_client = create_client(None, None, false).expect("failed to create http client"); - let endpoint = endpoint_for(&fake_intake.traces_url(), DD_API_KEY); + let endpoint = endpoint_for(&mock_intake.traces_url(), DD_API_KEY); let (aggregator_service, aggregator_handle) = AggregatorService::default(); tokio::spawn(aggregator_service.run()); let span = pb::Span { - service: "fake-intake-trace-service".to_string(), + service: "mock-intake-trace-service".to_string(), name: "web.request".to_string(), resource: "GET /fake".to_string(), trace_id: 0x1111_1111_1111_1111, @@ -276,7 +276,7 @@ async fn trace_payload_roundtrip_through_fake_intake() { "trace flush reported a retry-able failure: {failed:?}", ); - let captured = fake_intake.trace_payloads(); + let captured = mock_intake.trace_payloads(); assert_eq!(captured.len(), 1, "expected exactly one AgentPayload"); let payload = &captured[0]; @@ -290,7 +290,7 @@ async fn trace_payload_roundtrip_through_fake_intake() { assert_eq!(chunk.priority, 1); assert_eq!(chunk.spans.len(), 1); let span = &chunk.spans[0]; - assert_eq!(span.service, "fake-intake-trace-service"); + assert_eq!(span.service, "mock-intake-trace-service"); assert_eq!(span.name, "web.request"); assert_eq!(span.resource, "GET /fake"); assert_eq!(span.trace_id, 0x1111_1111_1111_1111); @@ -298,9 +298,9 @@ async fn trace_payload_roundtrip_through_fake_intake() { } // --------------------------------------------------------------------------- -// APMSVLS-487 Tier 3: full fake-intake E2E through `SendingTraceProcessor`. +// APMSVLS-487 Tier 3: full mock-intake E2E through `SendingTraceProcessor`. // -// Unlike `trace_payload_roundtrip_through_fake_intake` (which inserts a hand-built +// Unlike `trace_payload_roundtrip_through_mock_intake` (which inserts a hand-built // `pb::TracerPayload` directly), these tests route a trace through // `SendingTraceProcessor::send_processed_traces` so they exercise the real // `process_traces`/`ChunkProcessor` stamping of `_dd.compute_stats` and the @@ -325,12 +325,7 @@ struct PipelineOutcome { /// Drives `traces` through `SendingTraceProcessor::send_processed_traces` with the given /// `lambda_extension_compute_stats` / `client_computed_stats`, then flushes both the trace -/// and stats pipelines into a fresh fake-intake and returns what it captured. -async fn run_processor_pipeline_with_traces( - compute_on_extension: bool, - client_computed_stats: bool, - traces: Vec>, -) -> PipelineOutcome { +/// and stats pipelines into a fresh mock-intake and returns what it captured. async fn run_processor_pipeline_with_traces( compute_on_extension: bool, client_computed_stats: bool, @@ -348,14 +343,14 @@ async fn run_processor_pipeline_with_settings( error_rare_sample_tracer_drop: bool, traces: Vec>, ) -> PipelineOutcome { - let fake_intake = MockIntake::start().await; + let mock_intake = MockIntake::start().await; let config = Arc::new(Config { api_key: DD_API_KEY.to_string(), site: "datadoghq.com".to_string(), // process_traces builds its trace endpoint directly from apm_dd_url. - apm_dd_url: fake_intake.traces_url(), - service: Some("fake-intake-trace-service".to_string()), + apm_dd_url: mock_intake.traces_url(), + service: Some("mock-intake-trace-service".to_string()), ext: bottlecap::config::LambdaConfig { lambda_extension_compute_stats: compute_on_extension, ..Default::default() @@ -436,14 +431,14 @@ async fn run_processor_pipeline_with_settings( stats_aggregator, Arc::clone(&config), http_client, - fake_intake.stats_url(), + mock_intake.stats_url(), ); let failed = stats_flusher.flush(true, None).await; assert!(failed.is_none(), "stats flush failed: {failed:?}"); PipelineOutcome { - traces: fake_intake.trace_payloads(), - stats: fake_intake.stats_payloads(), + traces: mock_intake.trace_payloads(), + stats: mock_intake.stats_payloads(), } } @@ -459,7 +454,7 @@ async fn run_processor_pipeline( ) -> PipelineOutcome { // A top-level root span so the concentrator produces stats. let mut span = pb::Span { - service: "fake-intake-trace-service".to_string(), + service: "mock-intake-trace-service".to_string(), name: "web.request".to_string(), resource: "GET /fake".to_string(), trace_id: 0x1111_1111_1111_1111, @@ -496,7 +491,7 @@ fn captured_compute_stats(traces: &[pb::AgentPayload]) -> Option { /// Single-span trace root with the given sampling priority and error flag. fn sampled_root_span(trace_id: u64, priority: f64, error: i32) -> pb::Span { pb::Span { - service: "fake-intake-trace-service".to_string(), + service: "mock-intake-trace-service".to_string(), name: "web.request".to_string(), resource: "GET /fake".to_string(), trace_id, @@ -859,7 +854,7 @@ async fn e2e_client_computed_stats_absent_meta_and_no_stats() { /// Distinct `id` values keep trace/span ids unique across invocations. fn stats_trace(id: u64, resource: &str, duration: i64, error: i32) -> Vec { let mut span = pb::Span { - service: "fake-intake-trace-service".to_string(), + service: "mock-intake-trace-service".to_string(), name: "web.request".to_string(), resource: resource.to_string(), trace_id: id, @@ -1044,7 +1039,7 @@ fn make_eligible_span(span_kind: &str, peer_meta: &[(&str, &str)]) -> pb::Span { meta.insert("span.kind".to_string(), span_kind.to_string()); pb::Span { - service: "fake-intake-stats-service".to_string(), + service: "mock-intake-stats-service".to_string(), name: "test-op".to_string(), resource: "test-resource".to_string(), trace_id: 1, @@ -1060,10 +1055,10 @@ fn make_eligible_span(span_kind: &str, peer_meta: &[(&str, &str)]) -> pb::Span { } } -/// Wire concentrator -> aggregator -> flusher pointed at the fake intake, feed in +/// Wire concentrator -> aggregator -> flusher pointed at the mock intake, feed in /// `spans`, force a flush, and return the single captured `StatsPayload`. -async fn flush_spans_to_fake_intake( - fake_intake: &MockIntake, +async fn flush_spans_to_mock_intake( + mock_intake: &MockIntake, config: Arc, spans: &[pb::Span], ) -> pb::StatsPayload { @@ -1089,7 +1084,7 @@ async fn flush_spans_to_fake_intake( aggregator, config, http_client, - fake_intake.stats_url(), + mock_intake.stats_url(), ); let failed = flusher.flush(true, None).await; @@ -1098,7 +1093,7 @@ async fn flush_spans_to_fake_intake( "stats flush reported a retry-able failure: {failed:?}", ); - let captured = fake_intake.stats_payloads(); + let captured = mock_intake.stats_payloads(); assert_eq!(captured.len(), 1, "expected exactly one StatsPayload"); captured.into_iter().next().expect("captured payload") } @@ -1108,11 +1103,11 @@ async fn flush_spans_to_fake_intake( /// at the intake as a grouped-stats entry with `span_kind="server"`. This closes /// the gap left by the in-memory concentrator unit tests, which never serialize. #[tokio::test] -async fn stats_span_kind_through_fake_intake() { - let fake_intake = MockIntake::start().await; +async fn stats_span_kind_through_mock_intake() { + let mock_intake = MockIntake::start().await; let span = make_eligible_span("server", &[]); - let payload = flush_spans_to_fake_intake(&fake_intake, test_config(), &[span]).await; + let payload = flush_spans_to_mock_intake(&mock_intake, test_config(), &[span]).await; let grouped: Vec<_> = payload .stats @@ -1136,14 +1131,14 @@ async fn stats_span_kind_through_fake_intake() { /// populated in `peer_tags`, proving peer-tags survive serialization through /// the concentrator -> flusher -> intake path. #[tokio::test] -async fn stats_peer_tags_through_fake_intake() { - let fake_intake = MockIntake::start().await; +async fn stats_peer_tags_through_mock_intake() { + let mock_intake = MockIntake::start().await; let span = make_eligible_span( "client", &[("db.instance", "i-1234"), ("db.system", "postgres")], ); - let payload = flush_spans_to_fake_intake(&fake_intake, test_config(), &[span]).await; + let payload = flush_spans_to_mock_intake(&mock_intake, test_config(), &[span]).await; let with_peer_tags: Vec<_> = payload .stats @@ -1184,8 +1179,8 @@ fn grouped_entries(payload: &pb::StatsPayload) -> Vec<&pb::ClientGroupedStats> { /// same group. Span meta keys not listed in `DD_TRACE_STATS_ADDITIONAL_TAGS` /// must not be exported as additional metric tags. #[tokio::test] -async fn stats_additional_metric_tags_through_fake_intake() { - let fake_intake = MockIntake::start().await; +async fn stats_additional_metric_tags_through_mock_intake() { + let mock_intake = MockIntake::start().await; let config = config_from_env(&[ ("DD_API_KEY", DD_API_KEY), ("DD_SITE", "datadoghq.com"), @@ -1202,7 +1197,7 @@ async fn stats_additional_metric_tags_through_fake_intake() { make_eligible_span("server", &[("tenant_id", "acme")]), ]; - let payload = flush_spans_to_fake_intake(&fake_intake, config, &spans).await; + let payload = flush_spans_to_mock_intake(&mock_intake, config, &spans).await; let grouped = grouped_entries(&payload); let with_tags: Vec<_> = grouped @@ -1243,8 +1238,8 @@ async fn stats_additional_metric_tags_through_fake_intake() { /// spans differing only in the unexported meta value must merge into a single /// group. Proves the gate affects the payload, not just the parsed config. #[tokio::test] -async fn stats_additional_metric_tags_gated_off_through_fake_intake() { - let fake_intake = MockIntake::start().await; +async fn stats_additional_metric_tags_gated_off_through_mock_intake() { + let mock_intake = MockIntake::start().await; let config = config_from_env(&[ ("DD_API_KEY", DD_API_KEY), ("DD_SITE", "datadoghq.com"), @@ -1257,7 +1252,7 @@ async fn stats_additional_metric_tags_gated_off_through_fake_intake() { make_eligible_span("server", &[("region", "eu-west-1")]), ]; - let payload = flush_spans_to_fake_intake(&fake_intake, config, &spans).await; + let payload = flush_spans_to_mock_intake(&mock_intake, config, &spans).await; let grouped = grouped_entries(&payload); assert_eq!( @@ -1282,8 +1277,8 @@ async fn stats_additional_metric_tags_gated_off_through_fake_intake() { /// collapse the second into the `tracer_blocked_value` overflow group, keeping /// the total hit count intact. #[tokio::test] -async fn stats_additional_metric_tags_cardinality_limit_through_fake_intake() { - let fake_intake = MockIntake::start().await; +async fn stats_additional_metric_tags_cardinality_limit_through_mock_intake() { + let mock_intake = MockIntake::start().await; let config = config_from_env(&[ ("DD_API_KEY", DD_API_KEY), ("DD_SITE", "datadoghq.com"), @@ -1298,7 +1293,7 @@ async fn stats_additional_metric_tags_cardinality_limit_through_fake_intake() { make_eligible_span("server", &[("region", "eu-west-1")]), ]; - let payload = flush_spans_to_fake_intake(&fake_intake, config, &spans).await; + let payload = flush_spans_to_mock_intake(&mock_intake, config, &spans).await; let grouped = grouped_entries(&payload); assert_eq!( @@ -1332,8 +1327,8 @@ async fn stats_additional_metric_tags_cardinality_limit_through_fake_intake() { /// the two values together: one group per distinct combination, repeats aggregating per /// combination, and a change in either key producing a new group. #[tokio::test] -async fn stats_additional_metric_tags_multiple_keys_through_fake_intake() { - let fake_intake = MockIntake::start().await; +async fn stats_additional_metric_tags_multiple_keys_through_mock_intake() { + let mock_intake = MockIntake::start().await; let config = config_from_env(&[ ("DD_API_KEY", DD_API_KEY), ("DD_SITE", "datadoghq.com"), @@ -1351,7 +1346,7 @@ async fn stats_additional_metric_tags_multiple_keys_through_fake_intake() { make_eligible_span("server", &[("region", "eu-west-1"), ("tenant_id", "acme")]), ]; - let payload = flush_spans_to_fake_intake(&fake_intake, config, &spans).await; + let payload = flush_spans_to_mock_intake(&mock_intake, config, &spans).await; let grouped = grouped_entries(&payload); fn tags_of(s: &pb::ClientGroupedStats) -> Vec<&str> { diff --git a/bottlecap/tests/dsm_integration_test.rs b/bottlecap/tests/dsm_integration_test.rs index ac1d2258f..597ababba 100644 --- a/bottlecap/tests/dsm_integration_test.rs +++ b/bottlecap/tests/dsm_integration_test.rs @@ -1,7 +1,7 @@ // Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 -//! Payload-level DSM integration test using the in-process fake-intake. +//! Payload-level DSM integration test using the in-process mock-intake. //! //! Covers the full Data Streams Monitoring egress path: //! @@ -52,8 +52,8 @@ fn tags_provider(config: &Arc) -> Arc { } #[tokio::test] -async fn dsm_pipeline_stats_roundtrip_through_fake_intake() { - let fake_intake = MockIntake::start().await; +async fn dsm_pipeline_stats_roundtrip_through_mock_intake() { + let mock_intake = MockIntake::start().await; let config = test_config(); let http_client = create_reqwest_client_builder() .expect("failed to create reqwest client builder") @@ -63,10 +63,10 @@ async fn dsm_pipeline_stats_roundtrip_through_fake_intake() { let proxy_aggregator = Arc::new(Mutex::new(ProxyAggregator::default())); // The DsmProcessor derives its target URL from `apm_dd_url`, appending - // `/api/v0.1/pipeline_stats`. Pointing it at the fake intake's base URL + // `/api/v0.1/pipeline_stats`. Pointing it at the mock intake's base URL // exercises the same path a custom DD_APM_DD_URL takes in production. let dsm_processor = DsmProcessor::new( - "fake-intake-dsm-service".to_string(), + "mock-intake-dsm-service".to_string(), "test-env".to_string(), "1.0".to_string(), "2.0".to_string(), @@ -74,7 +74,7 @@ async fn dsm_pipeline_stats_roundtrip_through_fake_intake() { "team:serverless".to_string(), "region:us-east-1".to_string(), ], - &fake_intake.base_url(), + &mock_intake.base_url(), Arc::clone(&proxy_aggregator), ); @@ -103,7 +103,7 @@ async fn dsm_pipeline_stats_roundtrip_through_fake_intake() { "flush reported failed requests: {failed:?}" ); - let payloads = fake_intake.pipeline_stats_payloads(); + let payloads = mock_intake.pipeline_stats_payloads(); assert_eq!( payloads.len(), 1, @@ -111,7 +111,7 @@ async fn dsm_pipeline_stats_roundtrip_through_fake_intake() { ); let payload = &payloads[0]; - assert_eq!(payload.service, "fake-intake-dsm-service"); + assert_eq!(payload.service, "mock-intake-dsm-service"); assert_eq!(payload.env, "test-env"); assert_eq!(payload.tracer_version, "1.0"); assert_eq!(payload.version, "2.0"); From 20b97428565264243d9dc5186846b8152f1ca016 Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Tue, 29 Sep 2026 14:26:40 -0400 Subject: [PATCH 09/19] Remove orphaned empty tests/common module --- bottlecap/tests/common/mod.rs | 2 -- 1 file changed, 2 deletions(-) delete mode 100644 bottlecap/tests/common/mod.rs diff --git a/bottlecap/tests/common/mod.rs b/bottlecap/tests/common/mod.rs deleted file mode 100644 index 521e0d805..000000000 --- a/bottlecap/tests/common/mod.rs +++ /dev/null @@ -1,2 +0,0 @@ -// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ -// SPDX-License-Identifier: Apache-2.0 From b0a3eb792a4d8ee9700d21889a3c159e07290dea Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Tue, 20 Oct 2026 13:23:37 -0400 Subject: [PATCH 10/19] Move mock-intake detail docs to binary README --- bottlecap/README.md | 50 +------------------- bottlecap/src/bin/mock-intake/README.md | 62 +++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 49 deletions(-) create mode 100644 bottlecap/src/bin/mock-intake/README.md diff --git a/bottlecap/README.md b/bottlecap/README.md index 3298688b1..bf0769183 100644 --- a/bottlecap/README.md +++ b/bottlecap/README.md @@ -14,55 +14,7 @@ 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. -### 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 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`) -- 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 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 crate decodes. Filenames are collision-resistant and never overwrite earlier dumps. - -### Smoke procedure - -Point the test-mode trace processor at a local mock intake. The `bottlecap-test-mode` binary is added in [#1216](https://github.com/DataDog/datadog-lambda-extension/pull/1216) (not yet merged); build it from a checkout of that branch: - -```bash -cd ..//bottlecap -cargo build --bin bottlecap-test-mode --features test-mode - -cd ..//bottlecap -cargo run --bin mock-intake --features mock-intake - -# in another shell: -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. +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 diff --git a/bottlecap/src/bin/mock-intake/README.md b/bottlecap/src/bin/mock-intake/README.md new file mode 100644 index 000000000..915787e33 --- /dev/null +++ b/bottlecap/src/bin/mock-intake/README.md @@ -0,0 +1,62 @@ +# 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 same crate backs the `apm_integration_test` and `dsm_integration_test` integration tests, which 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 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`) +- 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 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 crate decodes. Filenames are collision-resistant and never overwrite earlier dumps. + +## Smoke procedure + +Point the test-mode trace processor at a local mock intake. The `bottlecap-test-mode` binary is added in [#1216](https://github.com/DataDog/datadog-lambda-extension/pull/1216) (not yet merged); build it from a checkout of that branch: + +```bash +cd ..//bottlecap +cargo build --bin bottlecap-test-mode --features test-mode + +cd ..//bottlecap +cargo run --bin mock-intake --features mock-intake + +# in another shell: +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. From ca798c4a0a1d83ecd681227b1822df04bd13a06b Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Wed, 21 Oct 2026 17:27:41 -0400 Subject: [PATCH 11/19] Pin datadog-mock-intake to the merged serverless-components rev --- bottlecap/Cargo.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/bottlecap/Cargo.lock b/bottlecap/Cargo.lock index 9fd478ce2..de827afd0 100644 --- a/bottlecap/Cargo.lock +++ b/bottlecap/Cargo.lock @@ -852,12 +852,12 @@ dependencies = [ [[package]] name = "datadog-mock-intake" version = "0.1.0" -source = "git+https://github.com/DataDog/serverless-components?rev=37f5b32a853f82a00523051827af074662f3e8fa#37f5b32a853f82a00523051827af074662f3e8fa" +source = "git+https://github.com/DataDog/serverless-components?rev=e462105e918904e6b1ad2ebc153ae561cebe5a34#e462105e918904e6b1ad2ebc153ae561cebe5a34" dependencies = [ "bytes", "flate2", "http-body-util", - "hyper 1.8.1", + "hyper 1.12.0", "hyper-util", "libdd-trace-protobuf", "prost 0.14.3", From 10d15efdc0173a74c509c23175df1a25ba67f58d Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Thu, 22 Oct 2026 15:15:17 -0400 Subject: [PATCH 12/19] Simplify mock intake configuration and fixtures --- bottlecap/src/bin/mock-intake/main.rs | 13 +++++-------- bottlecap/tests/apm_integration_test.rs | 2 +- 2 files changed, 6 insertions(+), 9 deletions(-) diff --git a/bottlecap/src/bin/mock-intake/main.rs b/bottlecap/src/bin/mock-intake/main.rs index 11c843df4..f82253ee2 100644 --- a/bottlecap/src/bin/mock-intake/main.rs +++ b/bottlecap/src/bin/mock-intake/main.rs @@ -60,19 +60,18 @@ async fn main() -> anyhow::Result<()> { /// 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(name: &str, default: T, expected: &str) -> anyhow::Result { - let value = match std::env::var(name) { + match std::env::var(name) { Ok(raw) => raw.parse::().map_err(|_| { anyhow::anyhow!("mock-intake: invalid {name} value '{raw}', expected {expected}") - })?, - Err(std::env::VarError::NotPresent) => default, + }), + Err(std::env::VarError::NotPresent) => Ok(default), Err(std::env::VarError::NotUnicode(raw)) => { bail!( "mock-intake: {name} is not valid Unicode: {}", raw.display() ) } - }; - Ok(value) + } } fn parse_options() -> anyhow::Result { @@ -87,13 +86,11 @@ fn parse_options() -> anyhow::Result { "a non-negative integer", )?; - let dump_dir = std::env::var_os("MOCK_INTAKE_DUMP_DIR").map(PathBuf::from); - Ok(MockIntakeOptions { port, request_summaries: true, fail_stats_first_n, - dump_dir, + dump_dir: std::env::var_os("MOCK_INTAKE_DUMP_DIR").map(PathBuf::from), }) } diff --git a/bottlecap/tests/apm_integration_test.rs b/bottlecap/tests/apm_integration_test.rs index 4b45095a3..36c838204 100644 --- a/bottlecap/tests/apm_integration_test.rs +++ b/bottlecap/tests/apm_integration_test.rs @@ -600,7 +600,7 @@ async fn e2e_decision_maker_promoted_into_chunk_tags() { let trace_id = 0x3333_3333_3333_3333; let mut span = pb::Span { - service: "fake-intake-trace-service".to_string(), + service: "mock-intake-trace-service".to_string(), name: "web.request".to_string(), resource: "GET /fake".to_string(), trace_id, From 190509d1ecb8c93966b8dea285ea6929a03118cd Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Wed, 4 Nov 2026 17:51:21 -0500 Subject: [PATCH 13/19] Disable captures in standalone mock intake --- bottlecap/Cargo.lock | 2 +- bottlecap/Cargo.toml | 4 ++-- bottlecap/src/bin/mock-intake/README.md | 8 ++++---- bottlecap/src/bin/mock-intake/main.rs | 17 +++++++++++++++++ 4 files changed, 24 insertions(+), 7 deletions(-) diff --git a/bottlecap/Cargo.lock b/bottlecap/Cargo.lock index de827afd0..63dfb7b78 100644 --- a/bottlecap/Cargo.lock +++ b/bottlecap/Cargo.lock @@ -852,7 +852,7 @@ dependencies = [ [[package]] name = "datadog-mock-intake" version = "0.1.0" -source = "git+https://github.com/DataDog/serverless-components?rev=e462105e918904e6b1ad2ebc153ae561cebe5a34#e462105e918904e6b1ad2ebc153ae561cebe5a34" +source = "git+https://github.com/DataDog/serverless-components?rev=fc89cb128a217385c2135a28fea158fc17d0ca40#fc89cb128a217385c2135a28fea158fc17d0ca40" dependencies = [ "bytes", "flate2", diff --git a/bottlecap/Cargo.toml b/bottlecap/Cargo.toml index c44226c9c..18478e7da 100644 --- a/bottlecap/Cargo.toml +++ b/bottlecap/Cargo.toml @@ -97,7 +97,7 @@ 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 = "e462105e918904e6b1ad2ebc153ae561cebe5a34", optional = true } +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 } @@ -127,7 +127,7 @@ tokio-rustls = { version = "0.26", default-features = false, features = ["loggin 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 = "e462105e918904e6b1ad2ebc153ae561cebe5a34" } +datadog-mock-intake = { git = "https://github.com/DataDog/serverless-components", rev = "fc89cb128a217385c2135a28fea158fc17d0ca40" } [build-dependencies] # No external dependencies needed for the build script diff --git a/bottlecap/src/bin/mock-intake/README.md b/bottlecap/src/bin/mock-intake/README.md index 915787e33..91116f4fd 100644 --- a/bottlecap/src/bin/mock-intake/README.md +++ b/bottlecap/src/bin/mock-intake/README.md @@ -9,7 +9,7 @@ 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. +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 @@ -23,9 +23,9 @@ Malformed configuration values cause a clear error and a nonzero exit. Bind fail ## 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 +- `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 diff --git a/bottlecap/src/bin/mock-intake/main.rs b/bottlecap/src/bin/mock-intake/main.rs index f82253ee2..d233afe2e 100644 --- a/bottlecap/src/bin/mock-intake/main.rs +++ b/bottlecap/src/bin/mock-intake/main.rs @@ -91,6 +91,7 @@ fn parse_options() -> anyhow::Result { request_summaries: true, fail_stats_first_n, dump_dir: std::env::var_os("MOCK_INTAKE_DUMP_DIR").map(PathBuf::from), + disable_captures: true, }) } @@ -105,3 +106,19 @@ async fn wait_for_shutdown() -> anyhow::Result<()> { } 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(()) + }); + } +} From 51cff75d7e67969029aa035125f423386190e72b Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Wed, 4 Nov 2026 17:55:13 -0500 Subject: [PATCH 14/19] Propagate mock intake signal errors --- bottlecap/src/bin/mock-intake/main.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bottlecap/src/bin/mock-intake/main.rs b/bottlecap/src/bin/mock-intake/main.rs index d233afe2e..a6a60f8ca 100644 --- a/bottlecap/src/bin/mock-intake/main.rs +++ b/bottlecap/src/bin/mock-intake/main.rs @@ -101,7 +101,7 @@ async fn wait_for_shutdown() -> anyhow::Result<()> { let mut sigterm = signal(SignalKind::terminate()) .context("mock-intake: failed to register SIGTERM handler")?; tokio::select! { - _ = tokio::signal::ctrl_c() => {} + result = tokio::signal::ctrl_c() => result?, _ = sigterm.recv() => {} } Ok(()) From 0f1c67910abca5603241a90cd50ac047296ca90a Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Wed, 4 Nov 2026 18:20:54 -0500 Subject: [PATCH 15/19] Build smoke test binaries from one checkout --- bottlecap/src/bin/mock-intake/README.md | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/bottlecap/src/bin/mock-intake/README.md b/bottlecap/src/bin/mock-intake/README.md index 91116f4fd..acb461e78 100644 --- a/bottlecap/src/bin/mock-intake/README.md +++ b/bottlecap/src/bin/mock-intake/README.md @@ -43,16 +43,18 @@ With `MOCK_INTAKE_DUMP_DIR` set, each successfully decoded request attempt (incl ## Smoke procedure -Point the test-mode trace processor at a local mock intake. The `bottlecap-test-mode` binary is added in [#1216](https://github.com/DataDog/datadog-lambda-extension/pull/1216) (not yet merged); build it from a checkout of that branch: +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 --features test-mode +cd bottlecap +cargo build --bin bottlecap-test-mode --bin mock-intake --features test-mode,mock-intake +./target/debug/mock-intake +``` -cd ..//bottlecap -cargo run --bin mock-intake --features mock-intake +In another shell, start at the same checkout's repository root: -# in another shell: +```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 \ From 076ef97f0ae29630462aa4bab004563d3fce6fa9 Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Wed, 4 Nov 2026 18:47:19 -0500 Subject: [PATCH 16/19] Enable JSON dumps in the smoke procedure --- bottlecap/src/bin/mock-intake/README.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/bottlecap/src/bin/mock-intake/README.md b/bottlecap/src/bin/mock-intake/README.md index acb461e78..5b26edbbf 100644 --- a/bottlecap/src/bin/mock-intake/README.md +++ b/bottlecap/src/bin/mock-intake/README.md @@ -48,7 +48,9 @@ Point the test-mode trace processor at a local mock intake. Build both binaries ```bash cd bottlecap cargo build --bin bottlecap-test-mode --bin mock-intake --features test-mode,mock-intake -./target/debug/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: From d763a387cf7c4a5b5dc034fe91a9976d8e8eb225 Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Wed, 4 Nov 2026 18:56:29 -0500 Subject: [PATCH 17/19] Simplify mock intake setup and documentation --- bottlecap/Cargo.toml | 1 - bottlecap/src/bin/mock-intake/README.md | 2 +- bottlecap/src/bin/mock-intake/main.rs | 13 +++++-------- 3 files changed, 6 insertions(+), 10 deletions(-) diff --git a/bottlecap/Cargo.toml b/bottlecap/Cargo.toml index 18478e7da..ee53d87f3 100644 --- a/bottlecap/Cargo.toml +++ b/bottlecap/Cargo.toml @@ -141,7 +141,6 @@ required-features = ["test-mode"] [[bin]] name = "mock-intake" -path = "src/bin/mock-intake/main.rs" required-features = ["mock-intake"] [profile.dev] diff --git a/bottlecap/src/bin/mock-intake/README.md b/bottlecap/src/bin/mock-intake/README.md index 5b26edbbf..2e8f57601 100644 --- a/bottlecap/src/bin/mock-intake/README.md +++ b/bottlecap/src/bin/mock-intake/README.md @@ -39,7 +39,7 @@ Each handled request produces one summary line on stderr (prefix `mock-intake:`) ## 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 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 crate decodes. Filenames are collision-resistant and never overwrite earlier 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 diff --git a/bottlecap/src/bin/mock-intake/main.rs b/bottlecap/src/bin/mock-intake/main.rs index a6a60f8ca..763101d98 100644 --- a/bottlecap/src/bin/mock-intake/main.rs +++ b/bottlecap/src/bin/mock-intake/main.rs @@ -5,8 +5,7 @@ //! //! 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 (added in -//! DataDog/datadog-lambda-extension#1216) at it and inspect +//! 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: @@ -65,12 +64,10 @@ fn parse_env_or(name: &str, default: T, expected: &str) -> 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() - ) - } + Err(std::env::VarError::NotUnicode(raw)) => bail!( + "mock-intake: {name} is not valid Unicode: {}", + raw.display() + ), } } From d5f6607c120c3897e701ba92d6de85d9363d9fb8 Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Sat, 7 Nov 2026 00:19:57 -0500 Subject: [PATCH 18/19] Deduplicate mock-intake env var docs --- bottlecap/src/bin/mock-intake/main.rs | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/bottlecap/src/bin/mock-intake/main.rs b/bottlecap/src/bin/mock-intake/main.rs index 763101d98..29b2922fe 100644 --- a/bottlecap/src/bin/mock-intake/main.rs +++ b/bottlecap/src/bin/mock-intake/main.rs @@ -8,13 +8,7 @@ //! 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 | -//! |---------------------------------|---------|--------------------------------------------------| -//! | `MOCK_INTAKE_PORT` | `8127` | Loopback listener port (`0` = OS-assigned) | -//! | `MOCK_INTAKE_FAIL_STATS_FIRST_N`| `0` | Return HTTP 500 for the first N stats attempts | -//! | `MOCK_INTAKE_DUMP_DIR` | unset | Write one JSON envelope per decoded request here | +//! Environment variables are documented in `README.md` next to this file. #![deny(clippy::all)] #![deny(clippy::pedantic)] From 99c5fb9c20e9b85f9f81396439cc1287be1d5d53 Mon Sep 17 00:00:00 2001 From: Lucas Pimentel Date: Sat, 7 Nov 2026 00:36:04 -0500 Subject: [PATCH 19/19] Run mock-intake binary tests in CI --- .github/workflows/rs_ci.yml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/rs_ci.yml b/.github/workflows/rs_ci.yml index 1a6bbef4e..5731b1add 100644 --- a/.github/workflows/rs_ci.yml +++ b/.github/workflows/rs_ci.yml @@ -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