diff --git a/.github/workflows/rs_ci.yml b/.github/workflows/rs_ci.yml index 58ef0796b..133f82ebc 100644 --- a/.github/workflows/rs_ci.yml +++ b/.github/workflows/rs_ci.yml @@ -84,10 +84,11 @@ jobs: run: cargo clippy --workspace --all-targets --features default - working-directory: bottlecap run: cargo clippy --workspace --all-targets --no-default-features --features fips - # No other job compiles the test-mode feature: it gates test-only - # constructors that are absent from default and fips builds. + # No other job compiles the test-mode or mock-intake features: they gate + # test-only constructors and the standalone mock-intake binary, which are + # absent from default and fips builds. - working-directory: bottlecap - run: cargo clippy --workspace --all-targets --features default,test-mode + run: cargo clippy --workspace --all-targets --features default,test-mode,mock-intake build-all: name: Build All diff --git a/.gitlab/templates/pipeline.yaml.tpl b/.gitlab/templates/pipeline.yaml.tpl index cc829caef..f23aa9420 100644 --- a/.gitlab/templates/pipeline.yaml.tpl +++ b/.gitlab/templates/pipeline.yaml.tpl @@ -61,11 +61,12 @@ 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 - # No other job compiles the test-mode feature: it gates test-only - # constructors that are absent from default and fips builds. - - cargo clippy --workspace --features default,test-mode + - cargo clippy --workspace --all-targets --features default + - cargo clippy --workspace --all-targets --no-default-features --features fips + # No other job compiles the test-mode or mock-intake features: they gate + # test-only constructors and the standalone mock-intake binary, which are + # absent from default and fips builds. + - 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 3475b27e9..1474158dd 100644 --- a/bottlecap/Cargo.lock +++ b/bottlecap/Cargo.lock @@ -499,6 +499,7 @@ dependencies = [ "datadog-agent-config", "datadog-agent-trace-sampler", "datadog-fips", + "datadog-mock-intake", "datadog-opentelemetry", "datadog-protos", "ddsketch-agent", @@ -920,6 +921,26 @@ dependencies = [ "tracing", ] +[[package]] +name = "datadog-mock-intake" +version = "0.1.0" +source = "git+https://github.com/DataDog/serverless-components?rev=b3e6a032f00a29a34ce61b50138677991d43c1e5#b3e6a032f00a29a34ce61b50138677991d43c1e5" +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.5.2" @@ -1152,7 +1173,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -1785,7 +1806,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.6.3", + "socket2 0.5.10", "tokio", "tower-service", "tracing", @@ -3169,7 +3190,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ "heck", - "itertools 0.14.0", + "itertools 0.11.0", "log", "multimap", "once_cell", @@ -3189,7 +3210,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.11.0", "proc-macro2", "quote", "syn 2.0.117", @@ -3202,7 +3223,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.11.0", "proc-macro2", "quote", "syn 2.0.117", @@ -3297,7 +3318,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls", - "socket2 0.6.3", + "socket2 0.5.10", "thiserror 2.0.18", "tokio", "tracing", @@ -3335,7 +3356,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.6.3", + "socket2 0.5.10", "tracing", "windows-sys 0.60.2", ] @@ -3665,7 +3686,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3733,7 +3754,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4171,7 +4192,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -4302,7 +4323,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -5064,7 +5085,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] diff --git a/bottlecap/Cargo.toml b/bottlecap/Cargo.toml index 3f56cd2bb..83df90643 100644 --- a/bottlecap/Cargo.toml +++ b/bottlecap/Cargo.toml @@ -90,6 +90,10 @@ ddsketch-agent = { version = "0.1.0", default-features = false, git = "https://g libdd-capabilities = "3.0.0" libdd-common = { version = "5.2.0", default-features = false } libdd-trace-protobuf = "4.0.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 = "b3e6a032f00a29a34ce61b50138677991d43c1e5", optional = true } libdd-trace-utils = { version = "10.1.0", default-features = false, features = ["mini_agent"] } libdd-trace-normalization = "3.0.1" libdd-trace-obfuscation = { version = "6.0.0", default-features = false } @@ -113,9 +117,9 @@ 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"] } +# 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 = "b3e6a032f00a29a34ce61b50138677991d43c1e5" } [build-dependencies] # No external dependencies needed for the build script @@ -123,6 +127,11 @@ flate2 = { version = "1.1", default-features = false, features = ["rust_backend" [[bin]] name = "bottlecap" +[[bin]] +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 strip = false @@ -187,6 +196,11 @@ fips = [ # Not enabled in `default` or `fips`, so the items it gates do not appear # in production builds. test-mode = [] +# Builds the standalone mock-intake binary (`src/bin/mock-intake`). The +# APM / DSM integration tests do not require this feature: they use +# `datadog-mock-intake` as a dev-dependency and run in every test build. +# Not enabled in `default` or `fips`. +mock-intake = ["dep:datadog-mock-intake", "tokio/signal"] [lints.rust] unexpected_cfgs = { level = "warn", check-cfg = ['cfg(coverage,coverage_nightly)'] } diff --git a/bottlecap/LICENSE-3rdparty.csv b/bottlecap/LICENSE-3rdparty.csv index 07ec6ff60..50f10a435 100644 --- a/bottlecap/LICENSE-3rdparty.csv +++ b/bottlecap/LICENSE-3rdparty.csv @@ -51,6 +51,7 @@ darling_macro,https://github.com/TedDriggs/darling,MIT,Ted Driggs datadog-protos,https://github.com/DataDog/saluki,Apache-2.0,The datadog-protos Authors ddsketch-agent,https://github.com/DataDog/saluki,Apache-2.0,The ddsketch-agent Authors diff --git a/bottlecap/README.md b/bottlecap/README.md index 60a9db35e..bf0769183 100644 --- a/bottlecap/README.md +++ b/bottlecap/README.md @@ -3,6 +3,19 @@ ## Development Use the `/scripts/build_bottlecap_layer.sh` and either publish it as a layer and test in Lambda or copy the binary into a container image and test there. Ask AJ or see the internal wiki for more. +## Local APM debugging with mock-intake + +The `mock-intake` feature builds a standalone mock Datadog APM intake: a small HTTP server that accepts the same APM endpoints the extension flushes to, decodes the payloads, and reports them locally. It is gated behind a feature and is never part of `default` or `fips` production builds. + +```bash +cd bottlecap +cargo build --bin mock-intake --features mock-intake +``` + +The binary wraps the shared `datadog-mock-intake` crate from the `serverless-components` repository. The same crate backs the `apm_integration_test` and `dsm_integration_test` integration tests, which run in every test build via a dev-dependency. + +For configuration, endpoints, failure injection, request summaries, JSON dumps, and a smoke procedure, see [mock-intake/README.md](src/bin/mock-intake/README.md). + ## Flush Strategies Bottlecap supports several flush strategies that control when and how observability data (metrics, logs, traces) is sent to Datadog. The strategy is configured via the `DD_SERVERLESS_FLUSH_STRATEGY` environment variable. 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. diff --git a/bottlecap/src/bin/mock-intake/main.rs b/bottlecap/src/bin/mock-intake/main.rs new file mode 100644 index 000000000..11c843df4 --- /dev/null +++ b/bottlecap/src/bin/mock-intake/main.rs @@ -0,0 +1,110 @@ +// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +//! Standalone mock Datadog APM intake for local debugging. +//! +//! Runs the shared `datadog-mock-intake` crate with request summaries enabled +//! and accepts the same APM endpoints the Lambda extension flushes to. Point a +//! tracer or the `bottlecap-test-mode` trace processor (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: +//! +//! | 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 | + +#![deny(clippy::all)] +#![deny(clippy::pedantic)] +#![deny(clippy::unwrap_used)] +#![deny(unused_extern_crates)] +#![deny(unused_allocation)] +#![deny(unused_assignments)] +#![deny(unused_comparisons)] +#![deny(unreachable_pub)] +#![deny(missing_copy_implementations)] +#![deny(missing_debug_implementations)] + +use std::path::PathBuf; + +use anyhow::{Context, bail}; +use datadog_mock_intake::{MockIntake, MockIntakeOptions}; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let options = parse_options()?; + let intake = MockIntake::start_with_options(options) + .await + .context("mock-intake: failed to start")?; + + println!("mock-intake: listening on {}", intake.base_url()); + println!("mock-intake: stats endpoint POST {}", intake.stats_url()); + println!("mock-intake: traces endpoint POST {}", intake.traces_url()); + println!( + "mock-intake: DSM endpoint POST {}", + intake.pipeline_stats_url() + ); + println!("mock-intake: waiting for SIGINT or SIGTERM to shut down"); + + wait_for_shutdown() + .await + .context("mock-intake: failed to wait for shutdown signal")?; + drop(intake); + println!("mock-intake: shut down"); + Ok(()) +} + +/// Parse an environment variable via `FromStr`, falling back to `default` when +/// unset. `expected` describes the accepted format for the error message. +fn parse_env_or(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: {name} is not valid Unicode: {}", + raw.display() + ) + } + }; + Ok(value) +} + +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 = std::env::var_os("MOCK_INTAKE_DUMP_DIR").map(PathBuf::from); + + 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("mock-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 2eed5e12f..745a49117 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: @@ -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"; @@ -104,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 = FakeIntake::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"); @@ -126,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 { @@ -137,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(), @@ -182,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]; @@ -197,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]; @@ -216,17 +213,17 @@ async fn stats_payload_roundtrip_through_fake_intake() { } #[tokio::test] -async fn trace_payload_roundtrip_through_fake_intake() { - let fake_intake = FakeIntake::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, @@ -279,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]; @@ -293,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); @@ -301,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 @@ -328,20 +325,20 @@ 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. +/// 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, traces: Vec>, ) -> PipelineOutcome { - let fake_intake = FakeIntake::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() @@ -423,14 +420,14 @@ async fn run_processor_pipeline_with_traces( 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(), } } @@ -446,7 +443,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, @@ -489,7 +486,7 @@ fn captured_compute_stats(traces: &[pb::AgentPayload]) -> Option { async fn e2e_error_sampler_rescues_only_errored_p0_traces() { let make_span = |trace_id: u64, error: i32| { 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, @@ -678,7 +675,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, @@ -863,7 +860,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, @@ -879,10 +876,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: &FakeIntake, +async fn flush_spans_to_mock_intake( + mock_intake: &MockIntake, config: Arc, spans: &[pb::Span], ) -> pb::StatsPayload { @@ -908,7 +905,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; @@ -917,7 +914,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") } @@ -927,11 +924,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 = FakeIntake::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 @@ -955,14 +952,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 = FakeIntake::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 @@ -1003,8 +1000,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 = FakeIntake::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"), @@ -1021,7 +1018,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 @@ -1062,8 +1059,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 = FakeIntake::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"), @@ -1076,7 +1073,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!( @@ -1101,8 +1098,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 = FakeIntake::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"), @@ -1117,7 +1114,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!( @@ -1151,8 +1148,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 = FakeIntake::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"), @@ -1170,7 +1167,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/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/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 diff --git a/bottlecap/tests/dsm_integration_test.rs b/bottlecap/tests/dsm_integration_test.rs index a086a8ebe..597ababba 100644 --- a/bottlecap/tests/dsm_integration_test.rs +++ b/bottlecap/tests/dsm_integration_test.rs @@ -1,14 +1,14 @@ // 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: //! //! `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"; @@ -55,8 +52,8 @@ fn tags_provider(config: &Arc) -> Arc { } #[tokio::test] -async fn dsm_pipeline_stats_roundtrip_through_fake_intake() { - let fake_intake = FakeIntake::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") @@ -66,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(), @@ -77,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), ); @@ -106,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, @@ -114,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");