diff --git a/AGENTS.md b/AGENTS.md index 102c32899..5eb14fd12 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -8,7 +8,18 @@ The Datadog Lambda Extension ("Bottlecap") is an AWS Lambda Extension written in ## Common Commands -All Rust code lives under `bottlecap/`. Run these from that directory unless noted. +All Rust code lives under `bottlecap/`. Run these from that directory unless noted. CI runs the same commands (`.github/workflows/rs_ci.yml`). + +- Build: `cargo build` +- Test: `cargo nextest run --workspace` +- Lint: `cargo clippy --workspace --all-targets --features default`. CI also runs it with `--features default,test-mode` and with `--no-default-features --features fips`. +- Format: `cargo fmt --all -- --check` + +## Testing + +- Unit tests live next to the code. Integration tests are in `bottlecap/tests/`. +- Process-level tests run the real binary against a fake Lambda API and an intake that fails on demand. Read `bottlecap/tests/HARNESS.md` before you add or change one. Each reliability fix should come with one of these tests. +- Shared test inputs (telemetry batches, logs, traces, metrics, trigger events) are in `bottlecap/tests/fixtures/`. See `bottlecap/tests/fixtures/README.md`. ## Where New Code Goes diff --git a/bottlecap/.config/nextest.toml b/bottlecap/.config/nextest.toml new file mode 100644 index 000000000..e493b13d6 --- /dev/null +++ b/bottlecap/.config/nextest.toml @@ -0,0 +1,20 @@ +# Nextest configuration for the bottlecap workspace (CI runs `cargo nextest run +# --workspace` from this directory, see .github/workflows/rs_ci.yml). +# +# Nextest runs every test in its own process, so `#[serial]` cannot protect a +# fixed port across test binaries. The process-level contract tests spawn the +# real `bottlecap` binary, which binds 8124, 8125/udp, 8126, 8999 and 9000. +# Those binaries go in the `fixed-ports` group: one test at a time, across all +# of them. See tests/HARNESS.md. + +[test-groups] +fixed-ports = { max-threads = 1 } + +# A regex so the config stays valid when one of these binaries is absent +# (nextest rejects an exact `binary(name)` that matches nothing). +[[profile.default.overrides]] +filter = 'binary(/(^|_)(lifecycle_contract|fault_injection|config_robustness|port_conflict)_test$/)' +test-group = "fixed-ports" +# A debug build of the extension, a blocking flush and a shutdown deadline add +# up; the longest test (a storm with a memory window) stays under this. +slow-timeout = { period = "60s", terminate-after = 3 } diff --git a/bottlecap/Cargo.lock b/bottlecap/Cargo.lock index 3d7e4c4c0..b9b7c1b46 100644 --- a/bottlecap/Cargo.lock +++ b/bottlecap/Cargo.lock @@ -511,6 +511,7 @@ dependencies = [ "opentelemetry-semantic-conventions 0.30.0", "ordered_hash_map", "prost 0.14.3", + "protobuf", "rand 0.8.6", "regex", "reqwest", diff --git a/bottlecap/Cargo.toml b/bottlecap/Cargo.toml index ec3238011..f437d5ad4 100644 --- a/bottlecap/Cargo.toml +++ b/bottlecap/Cargo.toml @@ -122,6 +122,8 @@ tokio-rustls = { version = "0.26", default-features = false, features = ["loggin # 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 } [build-dependencies] # No external dependencies needed for the build script diff --git a/bottlecap/tests/HARNESS.md b/bottlecap/tests/HARNESS.md new file mode 100644 index 000000000..41994b5fd --- /dev/null +++ b/bottlecap/tests/HARNESS.md @@ -0,0 +1,187 @@ +# Fault-injection test harness + +Process-level contract tests run the real `bottlecap` binary against a fake +Lambda control plane and a Datadog intake that misbehaves on demand. Each test +asserts what Lambda or the customer would see: `/next` is called before the +deadline, the process exits within the SHUTDOWN deadline, data reaches the +intake, and nothing panics. No test reads internal state. + +Every reliability fix should come with one of these tests. A test for a bug that +is not fixed yet is committed `#[ignore = ": "]`, +so CI stays green and the test turns into a regression test when the fix lands +(remove the attribute in the same PR). The ID names the Jira ticket that tracks +the fix (for example `SVLS-10061`). A reason that starts with `needs owner decision` marks +a test whose expected behavior is a design choice, not a plain bug. + +## Layout + +| File | Role | +|---|---| +| `tests/common/harness.rs` | Umbrella module and `prelude`. | +| `tests/common/fake_extensions_api.rs` | `FakeExtensionsApi`: register, telemetry subscribe (status knob), test-driven `/next`, request log, Telemetry API sender. | +| `tests/common/fault_intake.rs` | `FaultIntake` with `FaultMode`, per-path modes, body-size rejection, request log; `SynBlackhole`. | +| `tests/common/bottlecap_process.rs` | `BottlecapProcess`: spawn the binary with the base env, capture stdout/stderr, RSS/CPU, exit with a deadline, kill on drop. | +| `tests/common/port_guard.rs` | Hold or probe the fixed ports (8124, 8125/udp, 8126, 8999, 9000). | +| `tests/common/telemetry_events.rs` | Telemetry API records as `Value`, thin wrappers over `tests/fixtures/telemetry.rs`. | +| `tests/fixtures/` | The shared corpus for benches and these tests (see its `README.md`). Each test crate declares `mod fixtures;`. | +| `tests/lifecycle_contract_test.rs` | Shutdown and `/next` contracts. | +| `tests/fault_injection_test.rs` | Intake faults: a 503 storm, an oversized trace, a response cut off mid-body. | +| `tests/config_robustness_test.rs` | Wrong-typed config values. | +| `tests/port_conflict_test.rs` | Another process holds port 8124 or 8126. | +| `.config/nextest.toml` | `fixed-ports` test group (`max-threads = 1`) for these binaries. | + +## Write a new fault test in 10 lines + +```rust +mod fixtures; +#[path = "common/harness.rs"] +mod harness; +use harness::prelude::*; + +#[tokio::test(flavor = "multi_thread")] +#[serial_test::serial(bottlecap_ports)] +#[ignore = "SVLS-xxxxx: "] +async fn my_fault() { + let s = Scenario::start(FaultMode::Ok).await; // fake API + intake, fixed ports checked + let mut bc = s.bottlecap().spawn(); // real binary, base env points at the fakes + let startup = bc.startup_deadline(); // /next #1 must arrive by it + let deadline = s.invoke("req-1", secs(10), 1, startup).await.unwrap(); // answer it with an INVOKE + s.intake.set_mode(FaultMode::Status(503)); // the fault + deliver_start(&s.api, "req-1", true).await.unwrap(); // initStart (cold start), then platform.start + deliver_runtime_done(&s.api, "req-1", true).await.unwrap(); + // /next #2 must arrive by the INVOKE's deadline, counted from before it was queued + assert!(s.api.wait_for_next_calls(2, deadline).await, "{}", bc.log_tail(40)); + s.shutdown_cleanly(&mut bc).await; // SHUTDOWN, exit 0 within 2 s, no panic +} +``` + +## Rules + +- Never sleep a fixed time. Wait on a recorded event with a deadline: + `api.wait_for_next_calls`, `intake.wait_for(...)`, `bc.wait_for_exit`. +- Assert the customer-visible contract (a `/next` call, an exit, an intake + request), never a log line. Print `bc.log_tail(40)` and `s.intake.summary()` + in the message. One exception: a log line may prove that the fault was really + injected (the port-conflict tests check that the held port conflicted), never + that the contract holds. +- Bound a flush by the invocation or shutdown deadline, not by a number from + an implementation. +- Count every deadline from the instant its event was queued: use the `Instant` that `invoke`, `push_invoke` or `push_shutdown` returns, and `bc.startup_deadline()` for the first `/next`. +- Compare a deadline with the instant the fakes recorded (`next_call_at(n)`, `IntakeRequest::at`, `bc.seen_running()`), never with `Instant::now()` after a wait: a slow test thread delays the wait, not the event. +- Match decoded content in every data assertion (the log text in `body_text()`, a name in `metric_names()`, an ID in `trace_ids()`), not only a path, status, size or API key; a check that only counts requests says why in a comment. +- Require `api_key() == Some(TEST_API_KEY)` in every positive data assertion, a faulted request too: the fake intake accepts a request without the key, the real one does not. +- Always check `bc.assert_no_panic()` (or use `shutdown_cleanly`). Dev builds + unwind and keep running, release builds abort; both print `panicked at`. +- Check memory growth only on Linux (`BottlecapProcess::can_measure_memory()`). +- Send telemetry in Lambda's order: `deliver_start` (with `platform.initStart` on the cold + start), then the function logs, then `deliver_runtime_done`. Lambda sends `platform.report` + only after the extension's next `/next`. Until `platform.initStart` arrives the extension + holds every log and the cold start's `aws.lambda.enhanced.invocations`. +- When the extension could break a Lambda phase rule, make the fake refuse it (a subscribe + after the first `/next` gets `403`) and list the rule in the API summary. +- Every spawning test carries `#[serial_test::serial(bottlecap_ports)]`: the + binary binds fixed ports, and plain `cargo test` runs tests in parallel. New + test binaries must also match the regex in `.config/nextest.toml`. +- Run with a per-worktree `CARGO_TARGET_DIR`: the binary under test is the + uplifted `target/debug/bottlecap`, which a shared target dir overwrites. + `log_tail` prints the binary path and age. + +## API summary + +- `FakeExtensionsApi`: `push_invoke(id, deadline_in)`, `push_shutdown(reason, deadline_in)` (each + returns the event's deadline as an `Instant`, taken before the event is queued), `next_calls()`, `next_call_at(n)`, `wait_for_next_calls(n, deadline)` (true only when `/next` call `n` arrived by + the `Instant`), `set_subscribe_status(StatusCode)`, `request_lines()`, + `send_telemetry(&[Value], budget)`, `send_telemetry_raw(Vec, budget)`, `telemetry_port()` + (from the last accepted subscribe body, `None` before one; `send_telemetry` fails without it, + as Lambda delivers telemetry only to subscribers). Both post to the port and path of the + subscribed URI. Like Lambda, `send_telemetry` withholds + events whose type (`platform`, `function`, `extension`) the subscribe body did not ask for; + `send_telemetry_raw` posts its body as is. Like Lambda, `/next` serves only the event types from + the register body and drops the rest, a `/next` while another is pending gets `403`, `/next` + and subscribe get `403` without the `Lambda-Extension-Identifier` issued at register, subscribe + gets `403` after the first `/next` (Lambda takes subscriptions only during init), register + gets `403` without `Lambda-Extension-Name: datadog-agent` or with an invalid body, a subscribe + body outside the Telemetry API reference rules gets `400` (also an `https://` URI, port 9001, + or a missing `Content-Type: application/json`), and + `accountId` needs `Lambda-Extension-Accept-Feature`. `set_managed_instance()` applies the Lambda + Managed Instances rules: register gets `403` for `INVOKE`, and subscribe gets `400` for any + `schemaVersion` but `2025-01-29`. The SHUTDOWN reason is lower case (`spindown`), as the Runtime + Interface Emulator sends it. + Any other path gets `202`. +- `FaultIntake` (HTTP/1.1): modes `Ok`, `Status(code)`, `Slow(d)`, `Blackhole`, + `RetryAfter{status, secs}`, `CloseMidBody`; string forms `ok`, `status:`, `slow:`, + `blackhole`, `retry_after::`, `close_mid_body`. + `Blackhole` never reads a body: with a path mode set, it reads only the request head to pick + the mode, and records the request with `body_len` 0. An upload that breaks off is not + recorded and gets no answer. + `set_mode`, `set_path_mode(path, mode)`, `clear_path_mode`, `set_reject_over(Some(bytes))`, + `set_ok_body(path, body)`; `requests()`, `requests_to`, `count`, `connections()`, `summary()`, + `wait_for(n, pred, deadline)` (true only when `n` matching requests arrived by the `Instant`; + `Slow` records a request before its delay), + `wait_for_path`; `IntakeRequest::at` (arrival), `api_key()`, `decoded_body()`, `body_text()`, + `trace_ids()` (every span's trace ID in an `AgentPayload`) and `metric_names()` (a sketches or + series payload), each `Err` when the body does not decode. `SynBlackhole::start()` (backlog 1, + queue pre-filled), `is_saturated()`. +- `BottlecapProcess`: `scenario.bottlecap()` then `.intake(url)`, `.env`, `.envs`, `.remove_env`, + `.managed_instance()` (also switches the fake API to its rules), `.flush_strategy(s)`, + `.secret_from(credentials_url)`, `.spawn()`; `wait_for_exit(budget)` (kills on timeout), + `is_running()`, `kill()`, `stdout()`, `stderr()`, `log_tail(n)`, `log_contains`, `panic_lines()`, + `assert_no_panic()`, `rss_kb()`, `hwm_kb()`, `cpu_ms()`, `uptime()`, `startup_deadline()` + (`STARTUP_BUDGET` from before the spawn), `seen_running()` (start of the last poll that found + the process running); `STARTUP_BUDGET`, + `EXIT_BUDGET`, `TEST_API_KEY`. +- `Scenario`: `start(mode)`, `bottlecap()`, `invoke(id, deadline_in, poll, poll_by)` (waits for + the poll by `poll_by`, then answers it; returns the INVOKE's deadline, `None` when the poll + missed `poll_by`), `wait_next_or_exit(bc, n, deadline)`, `shutdown(&mut bc)` (asserts the process is still + running when SHUTDOWN is queued and no poll saw it running after the 2 s SHUTDOWN deadline), + `shutdown_cleanly(&mut bc)` (also asserts exit 0 and no panic). + Helpers: `deliver_start(api, id, cold_start)` (`platform.initStart` on a cold start, then + `platform.start`), `deliver_runtime_done(api, id, with_metrics)` (`platform.runtimeDone` only), + `msgpack_trace(trace_id, meta_bytes)`, `post_trace` (to 8126), + `INVOCATIONS_METRIC`, + `incompressible_text`, `secs`, `ms`. Records: `platform_init_start`, `platform_start`, + `platform_runtime_done(id, status, with_metrics)`, `platform_report`, `function_log`, + `platform_logs_dropped`, `unknown_event`; larger bodies come from `fixtures::telemetry`. +- `PortGuard::tcp(port)` / `udp(port)` / `*_wait` bind the address the extension uses. + +## Base environment + +`AWS_LAMBDA_RUNTIME_API=`, `AWS_LAMBDA_FUNCTION_NAME=test`, +`AWS_LAMBDA_FUNCTION_MEMORY_SIZE=128`, `AWS_LAMBDA_FUNCTION_VERSION=$LATEST`, +`AWS_REGION`/`AWS_DEFAULT_REGION=us-east-1`, `LAMBDA_TASK_ROOT=`, +`DD_API_KEY=`, `DD_SITE=datadoghq.com`, `DD_LOG_LEVEL=debug`, +`DD_SERVERLESS_FLUSH_STRATEGY=end`, +`DD_DD_URL`/`DD_URL`/`DD_APM_DD_URL`/`DD_LOGS_CONFIG_LOGS_DD_URL=http://`, +`DD_HTTP_PROTOCOL=http1` (the fakes speak HTTP/1.1; the default client uses h2 +prior knowledge), `DD_PROXY_HTTPS=http://127.0.0.1:1` (trace stats follow +`DD_SITE` over https; an https-only proxy on a closed port fails them at once +while the `http://` intake URLs bypass it). Managed Instance mode adds +`AWS_LAMBDA_INITIALIZATION_TYPE=lambda-managed-instances`. + +## Running + +```sh +cd bottlecap +cargo nextest run --test lifecycle_contract_test --test fault_injection_test \ + --test config_robustness_test --test port_conflict_test # CI set, ignored skipped +cargo nextest run --test fault_injection_test --run-ignored only # not-yet-fixed tests (none yet) +cargo test --test fault_injection_test # also serialized +``` + +`.config/nextest.toml` puts these binaries in the `fixed-ports` group +(`max-threads = 1`); nextest runs each test in its own process, so the +`serial` attribute alone would not protect the ports across binaries. A test +fails fast with `fixed ports still busy before spawn` when a leaked process +holds one. + +## Hand-offs + +- The fault modes belong in the shared `datadog-mock-intake` crate once + bottlecap consumes it: a `FaultMode` enum with the string forms above, + `set_mode`/`set_path_mode`, body-size rejection, and the recorded status and + mode per request. `blackhole` needs the accept loop, so the crate must expose + it. Until then `fault_intake.rs` stays here next to the shared crate's payload + decoding. +- A secret-recovery test (the secret fetch fails, then recovers) needs the extension to honor + `AWS_ENDPOINT_URL_` / `AWS_ENDPOINT_URL` (the AWS SDK convention) for + its Secrets Manager call; that hook is a separate small change. diff --git a/bottlecap/tests/common/bottlecap_process.rs b/bottlecap/tests/common/bottlecap_process.rs new file mode 100644 index 000000000..92c1938b5 --- /dev/null +++ b/bottlecap/tests/common/bottlecap_process.rs @@ -0,0 +1,398 @@ +// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +// `#[path]`-included by several test binaries; not every binary uses every item. +#![allow(dead_code)] +//! The real `bottlecap` binary as a child process, pointed at the fakes. +//! +//! `BottlecapProcess::builder(&api).intake(&url).spawn()` starts the binary +//! cargo built for this test run with a base environment where every Datadog +//! URL points at the fake. stdout and stderr go to files in the task root; +//! RSS and CPU come from `/proc` on Linux and `ps` elsewhere. Dropping the +//! value kills the process. +//! +//! The binary is `CARGO_BIN_EXE_bottlecap`, the uplifted `target/debug/bottlecap`. +//! With a target dir shared between worktrees that file is whichever worktree +//! built last, so run these tests with a per-worktree `CARGO_TARGET_DIR`; +//! `log_tail` prints the binary path and mtime so a stale one is visible. + +use std::collections::BTreeMap; +use std::path::PathBuf; +use std::process::{Child, Command, ExitStatus, Stdio}; +use std::time::{Duration, Instant, SystemTime}; + +use crate::harness::fake_extensions_api::FakeExtensionsApi; +use crate::harness::fault_intake::CLOSED_PORT_URL; + +/// Budget for a debug build to register, subscribe and reach `/next`. +pub const STARTUP_BUDGET: Duration = Duration::from_secs(15); +/// Budget for the process to exit after a SHUTDOWN, generous for debug builds. +pub const EXIT_BUDGET: Duration = Duration::from_secs(8); + +/// The API key the base env gives the extension. +pub const TEST_API_KEY: &str = "0000000000000000000000000000000a"; + +const BINARY: &str = env!("CARGO_BIN_EXE_bottlecap"); + +pub struct BottlecapBuilder<'api> { + api: &'api FakeExtensionsApi, + env: BTreeMap, + task_root: tempfile::TempDir, +} + +impl<'api> BottlecapBuilder<'api> { + /// Base env for a process that talks to `api`. Every intake URL points at + /// `api` too, until [`intake`](Self::intake) replaces them. + pub fn new(api: &'api FakeExtensionsApi) -> Self { + let task_root = tempfile::tempdir().expect("tempdir"); + let mut env = BTreeMap::new(); + let set = |env: &mut BTreeMap, k: &str, v: &str| { + env.insert(k.to_string(), v.to_string()); + }; + set(&mut env, "AWS_LAMBDA_RUNTIME_API", &api.runtime_api()); + set(&mut env, "AWS_LAMBDA_FUNCTION_NAME", "test"); + set(&mut env, "AWS_LAMBDA_FUNCTION_MEMORY_SIZE", "128"); + set(&mut env, "AWS_LAMBDA_FUNCTION_VERSION", "$LATEST"); + set(&mut env, "AWS_REGION", "us-east-1"); + set(&mut env, "AWS_DEFAULT_REGION", "us-east-1"); + set( + &mut env, + "LAMBDA_TASK_ROOT", + &task_root.path().display().to_string(), + ); + set(&mut env, "DD_API_KEY", TEST_API_KEY); + set(&mut env, "DD_SITE", "datadoghq.com"); + set(&mut env, "DD_LOG_LEVEL", "debug"); + set(&mut env, "DD_SERVERLESS_FLUSH_STRATEGY", "end"); + // The fakes speak HTTP/1.1; the default client uses h2 prior knowledge. + set(&mut env, "DD_HTTP_PROTOCOL", "http1"); + // Trace stats follow DD_SITE over https. Route https through a closed + // port so they fail at once instead of reaching the network; the http + // intake URLs bypass an https-only proxy. + set(&mut env, "DD_PROXY_HTTPS", CLOSED_PORT_URL); + let mut builder = Self { + api, + env, + task_root, + }; + builder.set_intake(&api.base_url()); + builder + } + + fn set_intake(&mut self, url: &str) { + for key in [ + "DD_DD_URL", + "DD_URL", + "DD_APM_DD_URL", + "DD_LOGS_CONFIG_LOGS_DD_URL", + ] { + self.env.insert(key.to_string(), url.to_string()); + } + } + + /// Point every Datadog intake URL at `url` (`http://host:port`). + pub fn intake(mut self, url: &str) -> Self { + self.set_intake(url); + self + } + + pub fn env(mut self, key: &str, value: &str) -> Self { + self.env.insert(key.to_string(), value.to_string()); + self + } + + pub fn envs<'a>(mut self, pairs: impl IntoIterator) -> Self { + for (k, v) in pairs { + self.env.insert(k.to_string(), v.to_string()); + } + self + } + + pub fn remove_env(mut self, key: &str) -> Self { + self.env.remove(key); + self + } + + /// Lambda Managed Instances mode, for the process and the fake API. + pub fn managed_instance(self) -> Self { + self.api.set_managed_instance(); + self.env("AWS_LAMBDA_INITIALIZATION_TYPE", "lambda-managed-instances") + } + + pub fn flush_strategy(self, strategy: &str) -> Self { + self.env("DD_SERVERLESS_FLUSH_STRATEGY", strategy) + } + + /// Fetch the API key from Secrets Manager with container credentials + /// served by `credentials_url`. The env carries no static AWS keys, so + /// the extension must call that URL first. + pub fn secret_from(self, credentials_url: &str) -> Self { + self.remove_env("DD_API_KEY") + .env( + "DD_API_KEY_SECRET_ARN", + "arn:aws:secretsmanager:us-east-1:123456789012:secret:dd-api-key", + ) + .env("AWS_CONTAINER_CREDENTIALS_FULL_URI", credentials_url) + .env("AWS_CONTAINER_AUTHORIZATION_TOKEN", "test-token") + } + + pub fn spawn(self) -> BottlecapProcess { + let stdout_path = self.task_root.path().join("bottlecap.stdout.log"); + let stderr_path = self.task_root.path().join("bottlecap.stderr.log"); + let stdout = std::fs::File::create(&stdout_path).expect("create stdout log"); + let stderr = std::fs::File::create(&stderr_path).expect("create stderr log"); + let mut cmd = Command::new(BINARY); + cmd.env_clear(); + // Keep PATH so `ps` and friends resolve for the child's own tooling. + if let Ok(path) = std::env::var("PATH") { + cmd.env("PATH", path); + } + for (k, v) in &self.env { + cmd.env(k, v); + } + let started = Instant::now(); + let child = cmd + .stdin(Stdio::null()) + .stdout(Stdio::from(stdout)) + .stderr(Stdio::from(stderr)) + .spawn() + .expect("spawn bottlecap"); + BottlecapProcess { + pid: child.id(), + child, + started, + seen_running: started, + stdout_path, + stderr_path, + _task_root: self.task_root, + } + } +} + +pub struct BottlecapProcess { + pub pid: u32, + child: Child, + /// Taken before the spawn, so a startup deadline counts from it. + pub started: Instant, + /// See [`seen_running`](Self::seen_running). + seen_running: Instant, + stdout_path: PathBuf, + stderr_path: PathBuf, + _task_root: tempfile::TempDir, +} + +impl BottlecapProcess { + pub fn builder(api: &FakeExtensionsApi) -> BottlecapBuilder<'_> { + BottlecapBuilder::new(api) + } + + pub fn uptime(&self) -> Duration { + self.started.elapsed() + } + + /// When the first `/next` must have arrived: [`STARTUP_BUDGET`] after the spawn. + pub fn startup_deadline(&self) -> Instant { + self.started + STARTUP_BUDGET + } + + /// Exit status if the process already exited. + pub fn try_exit(&mut self) -> Option { + let polled = Instant::now(); + let status = self.child.try_wait().ok().flatten(); + if status.is_none() { + self.seen_running = polled; + } + status + } + + /// Start of the last poll that found the process running (the spawn + /// before one). A process seen running after a deadline missed it; the + /// instant a wait returns does not prove that, as a slow test thread + /// delays it. + pub fn seen_running(&self) -> Instant { + self.seen_running + } + + pub fn is_running(&mut self) -> bool { + self.try_exit().is_none() + } + + /// Poll until the process exits or `budget` passes. On timeout the + /// process is killed and `None` is returned. + pub async fn wait_for_exit(&mut self, budget: Duration) -> Option { + let start = Instant::now(); + loop { + if let Some(status) = self.try_exit() { + return Some(status); + } + if start.elapsed() > budget { + self.kill(); + return None; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + } + + pub fn kill(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } + + pub fn stdout(&self) -> String { + std::fs::read_to_string(&self.stdout_path).unwrap_or_default() + } + + pub fn stderr(&self) -> String { + std::fs::read_to_string(&self.stderr_path).unwrap_or_default() + } + + /// The binary under test and its build time, for assertion messages. + pub fn binary_info() -> String { + let mtime = std::fs::metadata(BINARY) + .and_then(|m| m.modified()) + .ok() + .and_then(|t| t.duration_since(SystemTime::UNIX_EPOCH).ok()) + .map_or_else( + || "unknown".to_string(), + |d| { + format!( + "{}s ago", + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .map(|now| now.as_secs().saturating_sub(d.as_secs())) + .unwrap_or_default() + ) + }, + ); + format!("binary {BINARY} (built {mtime})") + } + + /// Last `n` lines of stdout and stderr, for assertion messages. Drops the + /// per-tick `/proc` noise a macOS host produces. + pub fn log_tail(&self, n: usize) -> String { + let out = self.stdout(); + let err = self.stderr(); + let tail = |text: &str| { + let lines: Vec<&str> = text + .lines() + .filter(|l| !l.contains("Could not list /proc files")) + .collect(); + let start = lines.len().saturating_sub(n); + lines[start..].join("\n") + }; + format!( + "{}\n--- stdout tail ---\n{}\n--- stderr tail ---\n{}", + Self::binary_info(), + tail(&out), + tail(&err) + ) + } + + pub fn log_contains(&self, needle: &str) -> bool { + self.stdout().contains(needle) || self.stderr().contains(needle) + } + + /// Lines of stderr that report a Rust panic. Dev builds unwind and keep + /// running; release builds abort. Both print `panicked at`. + pub fn panic_lines(&self) -> Vec { + self.stderr() + .lines() + .filter(|l| l.contains("panicked at")) + .map(str::to_string) + .collect() + } + + pub fn assert_no_panic(&self) { + let panics = self.panic_lines(); + assert!( + panics.is_empty(), + "bottlecap panicked:\n{}\n{}", + panics.join("\n"), + self.log_tail(40) + ); + } + + // --- resource usage ------------------------------------------------------- + + /// Resident set size in KiB, if the process is alive. + pub fn rss_kb(&self) -> Option { + proc_status_kb(self.pid, "VmRSS:").or_else(|| ps_kb(self.pid, "rss=")) + } + + /// Peak resident set size in KiB (Linux only). + pub fn hwm_kb(&self) -> Option { + proc_status_kb(self.pid, "VmHWM:") + } + + /// CPU time (user + system) in milliseconds, if the process is alive. + pub fn cpu_ms(&self) -> Option { + proc_stat_cpu_ms(self.pid).or_else(|| ps_cpu_ms(self.pid)) + } + + /// Whether RSS growth checks are meaningful on this platform. + pub fn can_measure_memory() -> bool { + cfg!(target_os = "linux") + } +} + +impl Drop for BottlecapProcess { + fn drop(&mut self) { + self.kill(); + } +} + +fn proc_status_kb(pid: u32, key: &str) -> Option { + let status = std::fs::read_to_string(format!("/proc/{pid}/status")).ok()?; + status + .lines() + .find(|l| l.starts_with(key))? + .split_whitespace() + .nth(1)? + .parse() + .ok() +} + +fn proc_stat_cpu_ms(pid: u32) -> Option { + let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?; + let after_comm = stat.rsplit(')').next()?; + let fields: Vec<&str> = after_comm.split_whitespace().collect(); + // Fields after `)`: state(0) ... utime(11) stime(12). + let utime: u64 = fields.get(11)?.parse().ok()?; + let stime: u64 = fields.get(12)?.parse().ok()?; + // CLK_TCK is 100 on every Linux we run on (Lambda included); a test that + // needs the exact value should read it through libc::sysconf. + let ticks_per_sec = 100; + Some((utime + stime) * 1000 / ticks_per_sec) +} + +fn ps_field(pid: u32, field: &str) -> Option { + let out = Command::new("ps") + .args(["-o", field, "-p", &pid.to_string()]) + .output() + .ok()?; + if !out.status.success() { + return None; + } + let text = String::from_utf8_lossy(&out.stdout).trim().to_string(); + if text.is_empty() { None } else { Some(text) } +} + +fn ps_kb(pid: u32, field: &str) -> Option { + ps_field(pid, field)?.parse().ok() +} + +/// `ps -o time=` prints `MM:SS.ss`, `HH:MM:SS` or `DD-HH:MM:SS`. +fn ps_cpu_ms(pid: u32) -> Option { + let text = ps_field(pid, "time=")?; + let (days, clock) = match text.split_once('-') { + Some((d, rest)) => (d.parse::().ok()?, rest), + None => (0.0, text.as_str()), + }; + let mut total_ms: f64 = days * 24.0 * 3600.0 * 1000.0; + let mut clock_ms = 0.0; + for part in clock.split(':') { + clock_ms = clock_ms * 60.0 + part.parse::().ok()? * 1000.0; + } + total_ms += clock_ms; + Some(total_ms as u64) +} diff --git a/bottlecap/tests/common/fake_extensions_api.rs b/bottlecap/tests/common/fake_extensions_api.rs new file mode 100644 index 000000000..1e39d8eb9 --- /dev/null +++ b/bottlecap/tests/common/fake_extensions_api.rs @@ -0,0 +1,541 @@ +// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +// `#[path]`-included by several test binaries; not every binary uses every item. +#![allow(dead_code)] +//! In-process fake Lambda Extensions API for process-level contract tests. +//! +//! Serves `register`, the Telemetry API `subscribe` (with a status knob), and a +//! `/next` that the test drives: push an INVOKE or a SHUTDOWN and the pending +//! long-poll returns it. Like Lambda, `/next` serves only the event types the +//! extension registered for, a `/next` while another is pending gets a `403`, +//! `/next` and `subscribe` get a `403` without the identifier issued at +//! register, register gets a `403` without the extension's name or with an +//! invalid body, a subscribe after the first `/next` gets a `403`, an invalid +//! subscribe body gets a `400`, and register returns `accountId` only when the +//! extension asks for it. In Lambda Managed Instances +//! mode ([`set_managed_instance`](FakeExtensionsApi::set_managed_instance)), +//! register gets a `403` for `INVOKE` and subscribe a `400` for any schema +//! version but `2025-01-29`. +//! It also answers any other request with `202`, so the same port can stand in +//! for the Datadog intake when a test does not need a `FaultIntake`. +//! +//! It also plays Lambda's Telemetry API sender: `send_telemetry` posts a batch +//! to the extension's listener (the port and path of the subscribe URI). Like +//! Lambda, it delivers nothing until a subscribe call was accepted, and then +//! only the event types that call asked for. + +use std::collections::VecDeque; +use std::net::SocketAddr; +use std::ops::RangeInclusive; +use std::sync::atomic::{AtomicBool, AtomicU16, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use axum::{ + Json, Router, + body::Bytes, + extract::State, + http::{HeaderMap, StatusCode}, + response::IntoResponse, + routing::{any, get, post, put}, +}; +use serde_json::{Value, json}; +use tokio::net::TcpListener; +use tokio::sync::Notify; +use tokio::task::JoinHandle; + +const EXTENSION_NAME_HEADER: &str = "Lambda-Extension-Name"; +const ACCEPT_FEATURE_HEADER: &str = "Lambda-Extension-Accept-Feature"; +/// The extension's file name under `/opt/extensions`. +const EXTENSION_NAME: &str = "datadog-agent"; +const EXTENSION_ID_HEADER: &str = "Lambda-Extension-Identifier"; +/// The identifier the fake issues at register, a UUID like Lambda's. +const EXTENSION_ID: &str = "3f8e4b6a-1c2d-4e5f-8a9b-0c1d2e3f4a5b"; + +#[derive(Clone)] +struct Shared { + /// Events pushed by the test, served to `/next` in order. + next_queue: Arc>>, + next_notify: Arc, + /// Arrival time of every accepted `/next` call, in order. + next_arrivals: Arc>>, + /// An accepted `/next` call waits for an event. + next_pending: Arc, + /// Lambda Managed Instances rules for register and subscribe. + managed_instance: Arc, + /// Event types from the last accepted register body. + registered_events: Arc>>, + /// `METHOD path` of every request, in arrival order. + requests: Arc>>, + subscribe_status: Arc, + /// `destination.URI` of the last accepted subscribe body with the host set + /// to `127.0.0.1`, `None` until one. + telemetry_url: Arc>>, + /// `types` of the last accepted subscribe body. + telemetry_types: Arc>>, +} + +pub struct FakeExtensionsApi { + pub addr: SocketAddr, + shared: Shared, + handle: JoinHandle<()>, + telemetry_client: reqwest::Client, +} + +/// A `/next` INVOKE body with a deadline `deadline_in` from now. +pub fn invoke_event(request_id: &str, deadline_in: Duration) -> Value { + json!({ + "eventType": "INVOKE", + "requestId": request_id, + "invokedFunctionArn": "arn:aws:lambda:us-east-1:123456789012:function:test", + "deadlineMs": epoch_ms(SystemTime::now() + deadline_in), + "tracing": {"type": "X-Amzn-Trace-Id", "value": "Root=1-0;Parent=0;Sampled=0"} + }) +} + +/// A `/next` SHUTDOWN body with a deadline `deadline_in` from now. +pub fn shutdown_event(reason: &str, deadline_in: Duration) -> Value { + json!({ + "eventType": "SHUTDOWN", + "shutdownReason": reason, + "deadlineMs": epoch_ms(SystemTime::now() + deadline_in), + }) +} + +fn epoch_ms(t: SystemTime) -> u64 { + u64::try_from(t.duration_since(UNIX_EPOCH).unwrap_or_default().as_millis()).unwrap_or(u64::MAX) +} + +impl FakeExtensionsApi { + /// Start the fake on `127.0.0.1:0` with an empty `/next` queue. + pub async fn start() -> Self { + let shared = Shared { + next_queue: Arc::new(Mutex::new(VecDeque::new())), + next_notify: Arc::new(Notify::new()), + next_arrivals: Arc::new(Mutex::new(Vec::new())), + next_pending: Arc::new(AtomicBool::new(false)), + managed_instance: Arc::new(AtomicBool::new(false)), + registered_events: Arc::new(Mutex::new(Vec::new())), + requests: Arc::new(Mutex::new(Vec::new())), + subscribe_status: Arc::new(AtomicU16::new(StatusCode::OK.as_u16())), + telemetry_url: Arc::new(Mutex::new(None)), + telemetry_types: Arc::new(Mutex::new(Vec::new())), + }; + let router = Router::new() + .route("/2020-01-01/extension/register", post(register)) + .route("/2020-01-01/extension/event/next", get(next)) + .route("/2022-07-01/telemetry", put(subscribe)) + .fallback(any(accept_anything)) + .with_state(shared.clone()); + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind fake extensions api"); + let addr = listener.local_addr().expect("local addr"); + let handle = tokio::spawn(async move { + // The server ends when the test drops the fake; nothing to report. + let _ = axum::serve(listener, router).await; + }); + Self { + addr, + shared, + handle, + // Test-only client towards a local fake; the FIPS adapter is for production. + #[allow(clippy::disallowed_methods)] + telemetry_client: reqwest::Client::builder() + .timeout(Duration::from_secs(5)) + .build() + .expect("telemetry client"), + } + } + + /// `host:port`, the value for `AWS_LAMBDA_RUNTIME_API`. + pub fn runtime_api(&self) -> String { + self.addr.to_string() + } + + /// `http://host:port`. + pub fn base_url(&self) -> String { + format!("http://{}", self.addr) + } + + /// Status for the Telemetry API subscribe call. Lambda returns 4xx for a + /// bad subscription. Takes effect for calls made after it. + pub fn set_subscribe_status(&self, status: StatusCode) { + self.shared + .subscribe_status + .store(status.as_u16(), Ordering::SeqCst); + } + + /// Apply the Lambda Managed Instances rules to calls made after it. + pub fn set_managed_instance(&self) { + self.shared.managed_instance.store(true, Ordering::SeqCst); + } + + // --- driving /next ------------------------------------------------------ + + /// Queue an event for the next `/next` long-poll. + pub fn push_next(&self, event: Value) { + self.shared + .next_queue + .lock() + .expect("next queue lock") + .push_back(event); + self.shared.next_notify.notify_one(); + } + + /// Queue an INVOKE whose `deadlineMs` is `deadline_in` from now. Returns + /// that deadline, taken before the event is queued. + pub fn push_invoke(&self, request_id: &str, deadline_in: Duration) -> Instant { + let deadline = Instant::now() + deadline_in; + self.push_next(invoke_event(request_id, deadline_in)); + deadline + } + + /// Queue a SHUTDOWN, and return its deadline as `push_invoke` does. + pub fn push_shutdown(&self, reason: &str, deadline_in: Duration) -> Instant { + let deadline = Instant::now() + deadline_in; + self.push_next(shutdown_event(reason, deadline_in)); + deadline + } + + /// Number of `/next` calls that arrived so far. + pub fn next_calls(&self) -> usize { + self.shared + .next_arrivals + .lock() + .expect("next arrivals lock") + .len() + } + + /// When `/next` call number `n` (1-based) arrived, `None` before it does. + pub fn next_call_at(&self, n: usize) -> Option { + self.shared + .next_arrivals + .lock() + .expect("next arrivals lock") + .get(n.checked_sub(1)?) + .copied() + } + + /// Wait for `/next` call number `n`. Returns whether it arrived by + /// `deadline`, so a call that came late counts as missed even when the + /// wait starts after it. + pub async fn wait_for_next_calls(&self, n: usize, deadline: Instant) -> bool { + loop { + let now = Instant::now(); + if let Some(at) = self.next_call_at(n) { + return at <= deadline; + } + if now > deadline { + return false; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + + /// `METHOD path` for each request, in order. + pub fn request_lines(&self) -> Vec { + self.shared.requests.lock().expect("requests lock").clone() + } + + // --- Telemetry API sender ---------------------------------------------- + + /// Port the extension asked Lambda to deliver telemetry to, `None` until + /// a subscribe call was accepted. + pub fn telemetry_port(&self) -> Option { + self.telemetry_url_parsed()?.port() + } + + /// Where Lambda would post telemetry: the subscribed port and path. + pub fn telemetry_url(&self) -> Option { + self.telemetry_url_parsed().map(String::from) + } + + fn telemetry_url_parsed(&self) -> Option { + self.shared + .telemetry_url + .lock() + .expect("telemetry url lock") + .clone() + } + + /// Post a raw body (a JSON array of events) to the extension's listener. + /// Retries until it is accepted or `budget` passes, since the listener + /// comes up a moment after the process starts. Fails without an accepted + /// subscription, as Lambda would deliver nothing. + pub async fn send_telemetry_raw(&self, body: Vec, budget: Duration) -> Result<(), String> { + let start = Instant::now(); + let mut last: String; + loop { + let result = match self.telemetry_url() { + Some(url) => self + .telemetry_client + .post(url) + .header("content-type", "application/json") + .body(body.clone()) + .send() + .await + .map_err(|e| e.to_string()), + None => Err("no accepted Telemetry API subscription".to_string()), + }; + match result { + Ok(r) if r.status().is_success() => return Ok(()), + Ok(r) => last = format!("status {}", r.status()), + Err(e) => last = e, + } + if start.elapsed() > budget { + return Err(format!("telemetry not accepted within {budget:?}: {last}")); + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + + /// Post one batch of events to the extension's listener. Like Lambda, it + /// withholds events whose type (`platform`, `function`, `extension`) the + /// subscription did not ask for, and posts nothing when none is left. + pub async fn send_telemetry(&self, events: &[Value], budget: Duration) -> Result<(), String> { + if self.telemetry_port().is_none() { + return Err("no accepted Telemetry API subscription".to_string()); + } + let types = self + .shared + .telemetry_types + .lock() + .expect("telemetry types lock") + .clone(); + let (subscribed, withheld): (Vec, Vec) = + events.iter().cloned().partition(|e| { + let kind = e["type"].as_str().unwrap_or_default(); + types + .iter() + .any(|t| kind.split('.').next() == Some(t.as_str())) + }); + for e in &withheld { + eprintln!( + "fake extensions api: withheld an unsubscribed {} event", + e["type"] + ); + } + if subscribed.is_empty() { + return Ok(()); + } + self.send_telemetry_raw(Value::Array(subscribed).to_string().into_bytes(), budget) + .await + } +} + +impl Drop for FakeExtensionsApi { + fn drop(&mut self) { + self.handle.abort(); + } +} + +fn record(shared: &Shared, method: &str, path: &str) { + shared + .requests + .lock() + .expect("requests lock") + .push(format!("{method} {path}")); +} + +async fn register( + State(shared): State, + headers: HeaderMap, + body: Bytes, +) -> Result { + record(&shared, "POST", "/2020-01-01/extension/register"); + // Lambda answers `403` without a name. Any name but the file under + // `/opt/extensions` registers an internal extension, which cannot take SHUTDOWN. + if headers + .get(EXTENSION_NAME_HEADER) + .is_none_or(|v| v != EXTENSION_NAME) + { + return Err(StatusCode::FORBIDDEN); + } + let managed_instance = shared.managed_instance.load(Ordering::SeqCst); + let events = register_events(&body, managed_instance).ok_or(StatusCode::FORBIDDEN)?; + *shared + .registered_events + .lock() + .expect("registered events lock") = events; + let mut response = + json!({"functionName": "test", "functionVersion": "$LATEST", "handler": "index.handler"}); + // Lambda adds `accountId` only when the extension asks for that feature. + if headers + .get(ACCEPT_FEATURE_HEADER) + .and_then(|v| v.to_str().ok()) + .is_some_and(|v| v.split(',').any(|f| f.trim() == "accountId")) + { + response["accountId"] = json!("123456789012"); + } + let mut headers = HeaderMap::new(); + headers.insert(EXTENSION_ID_HEADER, EXTENSION_ID.parse().expect("header")); + Ok((StatusCode::OK, headers, Json(response))) +} + +/// Event types of a register body, `None` when Lambda answers `403`. Lambda +/// accepts a body without `events`, and only `INVOKE` and `SHUTDOWN` in it +/// (only `SHUTDOWN` on Lambda Managed Instances). +fn register_events(body: &[u8], managed_instance: bool) -> Option> { + let body = serde_json::from_slice::(body).ok()?; + match body.as_object()?.get("events") { + None | Some(Value::Null) => Some(Vec::new()), + Some(events) => events + .as_array()? + .iter() + .map(|e| { + e.as_str() + .filter(|e| *e == "SHUTDOWN" || (*e == "INVOKE" && !managed_instance)) + .map(str::to_string) + }) + .collect(), + } +} + +/// Lambda answers `403` to a `/next` or `subscribe` call without the +/// identifier issued at register. +fn check_extension_id(headers: &HeaderMap) -> Result<(), StatusCode> { + if headers + .get(EXTENSION_ID_HEADER) + .is_some_and(|v| v == EXTENSION_ID) + { + Ok(()) + } else { + Err(StatusCode::FORBIDDEN) + } +} + +async fn subscribe( + State(shared): State, + headers: HeaderMap, + body: Bytes, +) -> Result { + record(&shared, "PUT", "/2022-07-01/telemetry"); + check_extension_id(&headers)?; + // Lambda takes subscriptions only during init, which the first `/next` ends. + if !shared + .next_arrivals + .lock() + .expect("next arrivals lock") + .is_empty() + { + return Err(StatusCode::FORBIDDEN); + } + // The Subscribe API takes a JSON body with `Content-Type: application/json`. + if !headers + .get("content-type") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.split(';').next()) + .is_some_and(|v| v.trim().eq_ignore_ascii_case("application/json")) + { + return Err(StatusCode::BAD_REQUEST); + } + let managed_instance = shared.managed_instance.load(Ordering::SeqCst); + let (url, types) = subscription(&body, managed_instance).ok_or(StatusCode::BAD_REQUEST)?; + let status = StatusCode::from_u16(shared.subscribe_status.load(Ordering::SeqCst)) + .unwrap_or(StatusCode::OK); + if status.is_success() { + *shared.telemetry_types.lock().expect("telemetry types lock") = types; + *shared.telemetry_url.lock().expect("telemetry url lock") = Some(url); + } + Ok((status, "{}")) +} + +/// Delivery URL and `types` of a subscribe body, `None` when Lambda answers +/// `400`: the Telemetry API reference rules, and an `http://sandbox` URI with a +/// port (Lambda does not deliver over HTTPS). Lambda Managed Instances take +/// only schema `2025-01-29`. The fake sender speaks only HTTP, so it also +/// rejects a `TCP` destination. +fn subscription(body: &[u8], managed_instance: bool) -> Option<(reqwest::Url, Vec)> { + let v = serde_json::from_slice::(body).ok()?; + let types = v["types"] + .as_array() + .filter(|types| !types.is_empty())? + .iter() + .map(|t| { + t.as_str() + .filter(|t| matches!(*t, "platform" | "function" | "extension")) + .map(str::to_string) + }) + .collect::>>()?; + let buffering_ok = v.get("buffering").is_none_or(|b| { + let in_range = |key: &str, range: RangeInclusive| { + b.get(key) + .is_none_or(|n| n.as_u64().is_some_and(|n| range.contains(&n))) + }; + b.is_object() + && in_range("maxItems", 1_000..=10_000) + && in_range("maxBytes", 262_144..=1_048_576) + && in_range("timeoutMs", 25..=30_000) + }); + let schema_ok = match v["schemaVersion"].as_str() { + Some("2025-01-29") => true, + Some("2022-07-01" | "2022-12-13") => !managed_instance, + _ => false, + }; + let mut uri = reqwest::Url::parse(v["destination"]["URI"].as_str()?).ok()?; + let valid = schema_ok + && v["destination"]["protocol"] == "HTTP" + && uri.scheme() == "http" + && matches!(uri.host_str(), Some("sandbox" | "sandbox.localdomain")) + // Lambda reserves port 9001 for the Runtime API. + && uri.port().is_some_and(|p| p != 0 && p != 9001) + && buffering_ok; + if !valid { + return None; + } + // Lambda posts to the URI's port and path inside the sandbox. + uri.set_host(Some("127.0.0.1")).ok()?; + Some((uri, types)) +} + +async fn next( + State(shared): State, + headers: HeaderMap, +) -> Result { + record(&shared, "GET", "/2020-01-01/extension/event/next"); + check_extension_id(&headers)?; + // Lambda answers `403` to a `/next` while the extension's last one waits. + // A poll the client drops stays pending, as in Lambda. + if shared.next_pending.swap(true, Ordering::SeqCst) { + return Err(StatusCode::FORBIDDEN); + } + shared + .next_arrivals + .lock() + .expect("next arrivals lock") + .push(Instant::now()); + loop { + let event = shared + .next_queue + .lock() + .expect("next queue lock") + .pop_front(); + let Some(event) = event else { + shared.next_notify.notified().await; + continue; + }; + let event_type = event["eventType"].as_str().unwrap_or_default(); + let registered = shared + .registered_events + .lock() + .expect("registered events lock") + .iter() + .any(|e| e == event_type); + if registered { + shared.next_pending.store(false, Ordering::SeqCst); + return Ok((StatusCode::OK, Json(event))); + } + // Lambda never sends an event type the extension did not register for. + eprintln!("fake extensions api: dropped an unregistered {event_type} event"); + } +} + +async fn accept_anything( + State(shared): State, + request: axum::extract::Request, +) -> impl IntoResponse { + record(&shared, request.method().as_str(), request.uri().path()); + (StatusCode::ACCEPTED, "{}") +} diff --git a/bottlecap/tests/common/fault_intake.rs b/bottlecap/tests/common/fault_intake.rs new file mode 100644 index 000000000..367b64abe --- /dev/null +++ b/bottlecap/tests/common/fault_intake.rs @@ -0,0 +1,527 @@ +// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +// `#[path]`-included by several test binaries; not every binary uses every item. +#![allow(dead_code)] +//! A Datadog intake that misbehaves on demand. +//! +//! Plain HTTP/1.1 on a random local port (set `DD_HTTP_PROTOCOL=http1` on the +//! extension). Every request is recorded with its path, size, headers, arrival +//! time and the mode that answered it; an upload that breaks off is not recorded and +//! gets no answer. The fault mode can change at runtime, and one +//! path can get its own mode (for example `/credentials`, for a secret fetch). +//! +//! Mode names: `ok`, `status:`, `slow:`, `blackhole`, +//! `retry_after::`, `close_mid_body`, and `syn_blackhole` for +//! [`SynBlackhole`]. These modes belong in the shared +//! `datadog-mock-intake` crate once bottlecap consumes it; see `tests/HARNESS.md`. + +use std::collections::HashMap; +use std::convert::Infallible; +use std::net::SocketAddr; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use bytes::Bytes; +use datadog_protos::metrics::{MetricPayload, SketchPayload}; +use futures::stream; +use http_body_util::{BodyExt, Full, StreamBody, combinators::BoxBody}; +use hyper::body::{Frame, Incoming}; +use hyper::server::conn::http1; +use hyper::service::service_fn; +use hyper::{Request, Response, StatusCode}; +use hyper_util::rt::TokioIo; +use libdd_trace_protobuf::pb; +use prost::Message as _; +use protobuf::Message as _; +use tokio::net::{TcpListener, TcpSocket, TcpStream}; +use tokio::task::JoinHandle; + +/// How the intake answers. +#[derive(Clone, Debug, PartialEq)] +pub enum FaultMode { + /// `202 Accepted` with `{}`. + Ok, + /// Always this status, with a plain-text (non-JSON) body. + Status(u16), + /// Wait, then `202`. + Slow(Duration), + /// Accept the connection, never read, never answer. + Blackhole, + /// This status with a `Retry-After: ` header. + RetryAfter { status: u16, secs: u64 }, + /// Status line and headers, a few body bytes, then close the connection. + CloseMidBody, +} + +impl FaultMode { + /// The shared string form. + pub fn name(&self) -> String { + match self { + Self::Ok => "ok".into(), + Self::Status(c) => format!("status:{c}"), + Self::Slow(d) => format!("slow:{}", d.as_millis()), + Self::Blackhole => "blackhole".into(), + Self::RetryAfter { status, secs } => format!("retry_after:{status}:{secs}"), + Self::CloseMidBody => "close_mid_body".into(), + } + } +} + +/// One request the intake received. +#[derive(Clone, Debug)] +pub struct IntakeRequest { + pub method: String, + pub path: String, + /// Lowercase header names; the last value wins on repeats. + pub headers: HashMap, + pub body_len: usize, + pub body: Vec, + /// When the intake had the whole body (the head, for `Blackhole`). + pub at: Instant, + /// Mode in force when it was answered. + pub mode: String, + pub status: u16, +} + +impl IntakeRequest { + pub fn header(&self, name: &str) -> Option<&str> { + self.headers + .get(&name.to_ascii_lowercase()) + .map(String::as_str) + } + + pub fn api_key(&self) -> Option<&str> { + self.header("dd-api-key") + } + + /// Body, zstd- or gzip-decoded when `Content-Encoding` says so. A body that + /// does not decode is an error, never partial output. + pub fn decoded_body(&self) -> Result, String> { + let decoded = match self.header("content-encoding") { + Some("zstd") => zstd::decode_all(self.body.as_slice()), + Some("gzip") => { + use std::io::Read; + let mut out = Vec::new(); + flate2::read::GzDecoder::new(self.body.as_slice()) + .read_to_end(&mut out) + .map(|_| out) + } + _ => Ok(self.body.clone()), + }; + decoded.map_err(|e| format!("{} {}: undecodable body: {e}", self.method, self.path)) + } + + pub fn body_text(&self) -> Result { + self.decoded_body() + .map(|b| String::from_utf8_lossy(&b).into_owned()) + } + + /// Trace ID of every span in a `/api/v0.2/traces` body (`AgentPayload`). + pub fn trace_ids(&self) -> Result, String> { + let payload = pb::AgentPayload::decode(self.decoded_body()?.as_slice()).map_err(|e| { + format!( + "{} {}: undecodable AgentPayload: {e}", + self.method, self.path + ) + })?; + Ok(payload + .tracer_payloads + .iter() + .flat_map(|t| &t.chunks) + .flat_map(|c| &c.spans) + .map(|span| span.trace_id) + .collect()) + } + + /// Metric name of every sketch or series in a `/api/beta/sketches` or + /// `/api/v2/series` body. + pub fn metric_names(&self) -> Result, String> { + let body = self.decoded_body()?; + let names = match self.path.as_str() { + "/api/beta/sketches" => SketchPayload::parse_from_bytes(&body) + .map(|p| p.sketches.iter().map(|m| m.metric.to_string()).collect()), + "/api/v2/series" => MetricPayload::parse_from_bytes(&body) + .map(|p| p.series.iter().map(|m| m.metric.to_string()).collect()), + _ => return Err(format!("{} {}: not a metrics path", self.method, self.path)), + }; + names.map_err(|e| format!("{} {}: undecodable metrics: {e}", self.method, self.path)) + } +} + +#[derive(Clone)] +struct Shared { + mode: Arc>, + path_modes: Arc>>, + /// Bodies over this size get a 413, in any mode but `Blackhole`. + reject_over: Arc>>, + /// Bodies sent back for a path in `Ok` mode, instead of `{}`. + ok_bodies: Arc>>, + requests: Arc>>, + connections: Arc, + held: Arc>>, +} + +pub struct FaultIntake { + pub addr: SocketAddr, + shared: Shared, + handle: JoinHandle<()>, +} + +impl FaultIntake { + /// Bind `127.0.0.1:0` and serve in `mode`. + pub async fn start(mode: FaultMode) -> Self { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind fault intake"); + let addr = listener.local_addr().expect("local addr"); + let shared = Shared { + mode: Arc::new(Mutex::new(mode)), + path_modes: Arc::new(Mutex::new(HashMap::new())), + reject_over: Arc::new(Mutex::new(None)), + ok_bodies: Arc::new(Mutex::new(HashMap::new())), + requests: Arc::new(Mutex::new(Vec::new())), + connections: Arc::new(AtomicUsize::new(0)), + held: Arc::new(Mutex::new(Vec::new())), + }; + let accept_shared = shared.clone(); + let handle = tokio::spawn(async move { + loop { + let Ok((stream, _)) = listener.accept().await else { + // EMFILE and friends: do not spin. + tokio::time::sleep(Duration::from_millis(10)).await; + continue; + }; + accept_shared.connections.fetch_add(1, Ordering::SeqCst); + // A connection accepted in blackhole mode is never read, unless a path has + // its own mode: then `handle` reads the request head to pick the mode. + let blackhole = + *accept_shared.mode.lock().expect("mode lock") == FaultMode::Blackhole; + if blackhole + && accept_shared + .path_modes + .lock() + .expect("path modes lock") + .is_empty() + { + accept_shared.held.lock().expect("held lock").push(stream); + continue; + } + let conn_shared = accept_shared.clone(); + tokio::spawn(async move { + let service = service_fn(move |req| handle(conn_shared.clone(), req)); + let _ = http1::Builder::new() + .keep_alive(true) + .serve_connection(TokioIo::new(stream), service) + .await; + }); + } + }); + Self { + addr, + shared, + handle, + } + } + + /// `http://host:port`. + pub fn url(&self) -> String { + format!("http://{}", self.addr) + } + + // --- knobs ------------------------------------------------------------ + + pub fn set_mode(&self, mode: FaultMode) { + *self.shared.mode.lock().expect("mode lock") = mode; + } + + /// Mode for one exact path, overriding the global mode. + pub fn set_path_mode(&self, path: &str, mode: FaultMode) { + self.shared + .path_modes + .lock() + .expect("path modes lock") + .insert(path.to_string(), mode); + } + + pub fn clear_path_mode(&self, path: &str) { + self.shared + .path_modes + .lock() + .expect("path modes lock") + .remove(path); + } + + /// Answer `413` to any body larger than `bytes`. + pub fn set_reject_over(&self, bytes: Option) { + *self.shared.reject_over.lock().expect("reject lock") = bytes; + } + + /// Body returned for `path` on a `202`, instead of `{}`. + pub fn set_ok_body(&self, path: &str, body: &str) { + self.shared + .ok_bodies + .lock() + .expect("ok bodies lock") + .insert(path.to_string(), body.to_string()); + } + + // --- request log -------------------------------------------------------- + + pub fn requests(&self) -> Vec { + self.shared.requests.lock().expect("requests lock").clone() + } + + /// Requests whose path contains `needle`. + pub fn requests_to(&self, needle: &str) -> Vec { + self.requests() + .into_iter() + .filter(|r| r.path.contains(needle)) + .collect() + } + + pub fn count(&self, needle: &str) -> usize { + self.requests_to(needle).len() + } + + /// Accepted TCP connections, including blackholed ones. + pub fn connections(&self) -> usize { + self.shared.connections.load(Ordering::SeqCst) + } + + /// `METHOD path -> status [mode] size` per request, for assertion messages. + pub fn summary(&self) -> String { + self.requests() + .iter() + .map(|r| { + format!( + "{} {} -> {} [{}] {}B", + r.method, r.path, r.status, r.mode, r.body_len + ) + }) + .collect::>() + .join("\n") + } + + /// Wait until `n` requests that match `pred` arrived by `deadline`. A + /// request that came late counts as missed even when the wait starts after it. + pub async fn wait_for( + &self, + n: usize, + pred: impl Fn(&IntakeRequest) -> bool, + deadline: Instant, + ) -> bool { + loop { + let now = Instant::now(); + let matched = self + .requests() + .iter() + .filter(|r| r.at <= deadline && pred(r)) + .count(); + if matched >= n { + return true; + } + if now > deadline { + return false; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + } + + /// Wait until `n` requests to a path containing `needle` arrived by `deadline`. + pub async fn wait_for_path(&self, needle: &str, n: usize, deadline: Instant) -> bool { + self.wait_for(n, |r| r.path.contains(needle), deadline) + .await + } +} + +impl Drop for FaultIntake { + fn drop(&mut self) { + self.handle.abort(); + } +} + +fn effective_mode(shared: &Shared, path: &str) -> FaultMode { + if let Some(m) = shared.path_modes.lock().expect("path modes lock").get(path) { + return m.clone(); + } + shared.mode.lock().expect("mode lock").clone() +} + +fn full(body: impl Into) -> BoxBody { + Full::new(body.into()).boxed() +} + +fn status_response(status: u16, body: String) -> Response> { + Response::builder() + .status(StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR)) + .header("content-type", "text/plain") + .body(full(body)) + .expect("response") +} + +async fn handle( + shared: Shared, + req: Request, +) -> Result>, hyper::Error> { + let method = req.method().as_str().to_string(); + let path = req.uri().path().to_string(); + let headers: HashMap = req + .headers() + .iter() + .map(|(k, v)| { + ( + k.as_str().to_ascii_lowercase(), + v.to_str().unwrap_or("").to_string(), + ) + }) + .collect(); + let mode = effective_mode(&shared, &path); + let mut request = IntakeRequest { + method, + path, + headers, + body_len: 0, + body: Vec::new(), + at: Instant::now(), + mode: mode.name(), + status: 0, + }; + if mode == FaultMode::Blackhole { + // Reached when a path has its own mode, or the mode changed on an open + // connection: record the head, never read the body, hold the connection. + record(&shared, request); + match std::future::pending::().await {} + } + request.body = match req.into_body().collect().await { + Ok(c) => c.to_bytes().to_vec(), + Err(e) => { + // The upload broke off, so the intake never received it: record + // nothing and close the connection without an answer. + eprintln!( + "fault intake: incomplete {} {}: {e}", + request.method, request.path + ); + return Err(e); + } + }; + request.body_len = request.body.len(); + request.at = Instant::now(); + let reject_over = *shared.reject_over.lock().expect("reject lock"); + + let mut delay = None; + let (status, response) = if reject_over.is_some_and(|max| request.body_len > max) { + ( + 413, + status_response(413, "fault:413 payload too large".into()), + ) + } else { + match &mode { + FaultMode::Ok => { + let body = shared + .ok_bodies + .lock() + .expect("ok bodies lock") + .get(&request.path) + .cloned() + .unwrap_or_else(|| "{}".to_string()); + ( + 202, + Response::builder() + .status(StatusCode::ACCEPTED) + .header("content-type", "application/json") + .body(full(body)) + .expect("response"), + ) + } + FaultMode::Status(code) => (*code, status_response(*code, format!("fault:{code}"))), + FaultMode::Slow(d) => { + delay = Some(*d); + (202, status_response(202, "{}".into())) + } + FaultMode::Blackhole => unreachable!("held before the body is read"), + FaultMode::RetryAfter { status, secs } => ( + *status, + Response::builder() + .status(StatusCode::from_u16(*status).unwrap_or(StatusCode::TOO_MANY_REQUESTS)) + .header("retry-after", secs.to_string()) + .body(full(format!("fault:{status} retry later"))) + .expect("response"), + ), + FaultMode::CloseMidBody => { + // A `Content-Length` larger than the one chunk we send: hyper + // ends the connection early and the client sees a truncated body. + let frames = vec![Ok::<_, Infallible>(Frame::data(Bytes::from_static( + b"{\"partial\":", + )))]; + let body = StreamBody::new(stream::iter(frames)).boxed(); + let response = Response::builder() + .status(StatusCode::OK) + .header("content-length", "4096") + .body(body) + .expect("response"); + (200, response) + } + } + }; + request.status = status; + // Record before a `Slow` delay, so `wait_for` sees the request from its arrival. + record(&shared, request); + if let Some(delay) = delay { + tokio::time::sleep(delay).await; + } + Ok(response) +} + +fn record(shared: &Shared, request: IntakeRequest) { + shared.requests.lock().expect("requests lock").push(request); +} + +/// A listener whose accept queue is full, so new connections never complete +/// their handshake: the kernel drops the SYN or leaves it half-open and the +/// client's connect blocks until its timeout. Needs no privileges. Linux +/// honors the backlog exactly; a kernel that does not is detected by +/// [`is_saturated`](Self::is_saturated). Build it inside a tokio runtime. +pub struct SynBlackhole { + pub addr: SocketAddr, + _listener: TcpListener, + _fillers: Vec, +} + +impl SynBlackhole { + pub fn start() -> Self { + let socket = TcpSocket::new_v4().expect("socket"); + socket + .bind("127.0.0.1:0".parse().expect("addr")) + .expect("bind"); + let listener = socket.listen(1).expect("listen"); + let addr = listener.local_addr().expect("local addr"); + // Fill the accept queue. Once full, a connect hangs, so stop on timeout. + let mut fillers = Vec::new(); + for _ in 0..8 { + match std::net::TcpStream::connect_timeout(&addr, Duration::from_millis(200)) { + Ok(s) => fillers.push(s), + Err(_) => break, + } + } + Self { + addr, + _listener: listener, + _fillers: fillers, + } + } + + pub fn url(&self) -> String { + format!("http://{}", self.addr) + } + + /// Whether the queue is really full: a fresh connect must time out. + pub fn is_saturated(&self) -> bool { + std::net::TcpStream::connect_timeout(&self.addr, Duration::from_millis(300)).is_err() + } +} + +/// A `host:port` nothing listens on. Connects fail at once with ECONNREFUSED, +/// which keeps trace stats (they follow `DD_SITE`) off the network. +pub const CLOSED_PORT_URL: &str = "http://127.0.0.1:1"; diff --git a/bottlecap/tests/common/harness.rs b/bottlecap/tests/common/harness.rs new file mode 100644 index 000000000..15de5b203 --- /dev/null +++ b/bottlecap/tests/common/harness.rs @@ -0,0 +1,282 @@ +// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +// `#[path]`-included by several test binaries; not every binary uses every item. +#![allow(dead_code)] +//! Fault-injection harness for process-level contract tests. +//! +//! Include it from a test binary as +//! `mod fixtures; #[path = "common/harness.rs"] mod harness; use harness::prelude::*;`. +//! The module must be named `harness`: the submodules refer to each other +//! through `crate::harness::...`, and `telemetry_events` reads the shared +//! corpus through `crate::fixtures`. See `tests/HARNESS.md`. + +#[path = "bottlecap_process.rs"] +pub mod bottlecap_process; +#[path = "fake_extensions_api.rs"] +pub mod fake_extensions_api; +#[path = "fault_intake.rs"] +pub mod fault_intake; +#[path = "port_guard.rs"] +pub mod port_guard; +#[path = "telemetry_events.rs"] +pub mod telemetry_events; + +#[allow(unused_imports)] +pub mod prelude { + pub use super::bottlecap_process::{ + BottlecapBuilder, BottlecapProcess, EXIT_BUDGET, STARTUP_BUDGET, TEST_API_KEY, + }; + pub use super::fake_extensions_api::FakeExtensionsApi; + pub use super::fault_intake::{FaultIntake, FaultMode, IntakeRequest, SynBlackhole}; + pub use super::port_guard::{ + DOGSTATSD_UDP_PORT, LIFECYCLE_PORT, PROXY_PORT, PortGuard, TELEMETRY_PORT, TRACE_AGENT_PORT, + }; + pub use super::telemetry_events::*; + pub use super::{ + INVOCATIONS_METRIC, Scenario, deliver_runtime_done, deliver_start, incompressible_text, ms, + msgpack_trace, post_trace, secs, + }; + pub use std::time::{Duration, Instant}; +} + +use std::process::ExitStatus; +use std::time::{Duration, Instant}; + +use bottlecap_process::BottlecapProcess; +use fake_extensions_api::FakeExtensionsApi; +use fault_intake::{FaultIntake, FaultMode}; + +pub const fn secs(n: u64) -> Duration { + Duration::from_secs(n) +} + +pub const fn ms(n: u64) -> Duration { + Duration::from_millis(n) +} + +/// The deadline of the SHUTDOWN event, Lambda's default. +const SHUTDOWN_DEADLINE: Duration = secs(2); + +/// The enhanced metric every invocation emits. +pub const INVOCATIONS_METRIC: &str = "aws.lambda.enhanced.invocations"; + +/// A fake Extensions API and a fault intake, started together. +/// +/// ```ignore +/// let s = Scenario::start(FaultMode::Ok).await; +/// let mut bc = s.bottlecap().spawn(); +/// let deadline = s.invoke("req-1", secs(10), 1, bc.startup_deadline()).await.unwrap(); +/// ``` +pub struct Scenario { + pub api: FakeExtensionsApi, + pub intake: FaultIntake, +} + +impl Scenario { + pub async fn start(mode: FaultMode) -> Self { + port_guard::require_fixed_ports_free().await; + let api = FakeExtensionsApi::start().await; + let intake = FaultIntake::start(mode).await; + Self { api, intake } + } + + /// A process builder pointed at this scenario's fakes. + pub fn bottlecap(&self) -> bottlecap_process::BottlecapBuilder<'_> { + BottlecapProcess::builder(&self.api).intake(&self.intake.url()) + } + + /// Wait for `/next` call number `poll`, which must arrive by `poll_by`, + /// then answer it with an INVOKE, as Lambda does. Returns the INVOKE's + /// deadline (see [`FakeExtensionsApi::push_invoke`]), `None` when the + /// poll missed `poll_by`. + pub async fn invoke( + &self, + request_id: &str, + deadline_in: Duration, + poll: usize, + poll_by: Instant, + ) -> Option { + if !self.api.wait_for_next_calls(poll, poll_by).await { + return None; + } + Some(self.api.push_invoke(request_id, deadline_in)) + } + + /// Wait for `/next` call number `n` and return whether it arrived by + /// `deadline`, giving up early when the process exits (a crash at INIT + /// should not cost the whole wait). + pub async fn wait_next_or_exit( + &self, + bc: &mut BottlecapProcess, + n: usize, + deadline: Instant, + ) -> bool { + loop { + if let Some(at) = self.api.next_call_at(n) { + return at <= deadline; + } + if Instant::now() > deadline || !bc.is_running() { + return self.api.next_call_at(n).is_some_and(|at| at <= deadline); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + + /// Answer the open `/next` with a SHUTDOWN ([`SHUTDOWN_DEADLINE`]), wait + /// for the process to exit, and assert no poll saw it running after that + /// deadline (Lambda kills it after). Also asserts the process is still + /// running when the SHUTDOWN is queued: Lambda treats an earlier exit as a + /// crash. Returns the exit status, `None` when it had to be killed. + pub async fn shutdown(&self, bc: &mut BottlecapProcess) -> Option { + let early = bc.try_exit(); + assert!( + early.is_none(), + "exited {early:?} before SHUTDOWN. API: {:?}\n{}", + self.api.request_lines(), + bc.log_tail(40) + ); + // Lambda sends the reason in lower case (Runtime Interface Emulator + // source); the Extensions API docs list it in upper case. + let deadline = self.api.push_shutdown("spindown", SHUTDOWN_DEADLINE); + let exit = bc.wait_for_exit(bottlecap_process::EXIT_BUDGET).await; + let late = bc.seen_running().saturating_duration_since(deadline); + assert!( + late.is_zero(), + "still running {late:?} after the {SHUTDOWN_DEADLINE:?} SHUTDOWN deadline. {}", + bc.log_tail(40) + ); + exit + } + + /// [`shutdown`](Self::shutdown), then assert no panic and a clean exit. + pub async fn shutdown_cleanly(&self, bc: &mut BottlecapProcess) { + let exit = self.shutdown(bc).await; + bc.assert_no_panic(); + assert!( + exit.is_some_and(|e| e.success()), + "exit {exit:?} after SHUTDOWN. API: {:?}\n{}", + self.api.request_lines(), + bc.log_tail(40) + ); + } +} + +/// `platform.start` for `request_id`, after `platform.initStart` on a cold +/// start, delivered to the extension's Telemetry API listener. Until +/// `platform.initStart` arrives the extension holds every log and the cold +/// start's invocation metric. Send the invocation's function logs after it. +pub async fn deliver_start( + api: &FakeExtensionsApi, + request_id: &str, + cold_start: bool, +) -> Result<(), String> { + let start = telemetry_events::platform_start(request_id); + let events = if cold_start { + vec![telemetry_events::platform_init_start(), start] + } else { + vec![start] + }; + api.send_telemetry(&events, bottlecap_process::STARTUP_BUDGET) + .await +} + +/// `platform.runtimeDone` for `request_id`, delivered after the invocation's +/// function logs. Lambda sends `platform.report` only after the extension's +/// next `/next`. +pub async fn deliver_runtime_done( + api: &FakeExtensionsApi, + request_id: &str, + with_metrics: bool, +) -> Result<(), String> { + api.send_telemetry( + &[telemetry_events::platform_runtime_done( + request_id, + "success", + with_metrics, + )], + bottlecap_process::STARTUP_BUDGET, + ) + .await +} + +/// `n` bytes of alphanumerics from a fixed-seed LCG. Compresses to roughly +/// three quarters of its size, so a padded payload stays large on the wire. +pub fn incompressible_text(n: usize, seed: u64) -> String { + const ALPHABET: &[u8] = b"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"; + let mut state = seed.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1); + (0..n) + .map(|_| { + state = state + .wrapping_mul(6_364_136_223_846_793_005) + .wrapping_add(1_442_695_040_888_963_407); + ALPHABET[((state >> 33) % ALPHABET.len() as u64) as usize] as char + }) + .collect() +} + +/// A one-span trace in the v0.4 msgpack form tracers post to the trace agent. +/// `meta_bytes` pads a tag with incompressible text so a test can make the +/// payload large on the wire. +pub fn msgpack_trace(trace_id: u64, meta_bytes: usize) -> Vec { + let now_ns = i64::try_from( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(), + ) + .unwrap_or(i64::MAX); + let padding = incompressible_text(meta_bytes, trace_id); + let span = serde_json::json!({ + "trace_id": trace_id, + "span_id": trace_id, + "parent_id": 0u64, + "name": "aws.lambda", + "resource": "test-handler", + "service": "test-service", + "type": "serverless", + "start": now_ns, + "duration": 1_000_000i64, + "error": 0, + "meta": {"_dd.origin": "lambda", "harness.padding": padding}, + "metrics": {"_sampling_priority_v1": 1.0} + }); + rmp_serde::to_vec_named(&vec![vec![span]]).expect("msgpack trace") +} + +/// Post a v0.4 trace payload to the trace agent on 8126, retrying until it +/// is up or `budget` passes. Returns the HTTP status. +pub async fn post_trace(body: Vec, budget: Duration) -> Result { + // Test-only client towards the local trace agent; the FIPS adapter is for production. + #[allow(clippy::disallowed_methods)] + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build() + .expect("client"); + let start = std::time::Instant::now(); + let mut last: String; + loop { + let result = client + .post(format!( + "http://127.0.0.1:{}/v0.4/traces", + port_guard::TRACE_AGENT_PORT + )) + .header("content-type", "application/msgpack") + .header("x-datadog-trace-count", "1") + .header("datadog-meta-lang", "harness") + .header("datadog-meta-tracer-version", "0.0.0") + .body(body.clone()) + .send() + .await; + match result { + Ok(r) => return Ok(r.status().as_u16()), + Err(e) => last = e.to_string(), + } + if start.elapsed() > budget { + return Err(format!( + "trace agent not reachable within {budget:?}: {last}" + )); + } + tokio::time::sleep(Duration::from_millis(50)).await; + } +} diff --git a/bottlecap/tests/common/port_guard.rs b/bottlecap/tests/common/port_guard.rs new file mode 100644 index 000000000..a685da0f4 --- /dev/null +++ b/bottlecap/tests/common/port_guard.rs @@ -0,0 +1,158 @@ +// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +// `#[path]`-included by several test binaries; not every binary uses every item. +#![allow(dead_code)] +//! Hold or probe the fixed ports the extension binds. +//! +//! The binary binds 8124 (lifecycle listener, `127.0.0.1`), 8125/udp +//! (DogStatsD, `0.0.0.0`), 8126 (trace agent, `127.0.0.1`), 8999 (Telemetry +//! API listener, `0.0.0.0`) and 9000 (runtime API proxy, `0.0.0.0`, only +//! with LWA or the Datadog wrapper). A guard must bind the same address as +//! the extension: on macOS a `127.0.0.1` bind coexists with a `0.0.0.0` one +//! when `SO_REUSEADDR` is set (tokio sets it), so a wildcard guard would not +//! conflict with the trace agent there. Tests that spawn the binary are +//! serialized (`#[serial_test::serial(bottlecap_ports)]` and the nextest +//! `fixed-ports` group); this module checks the ports are free before a +//! spawn and holds one of them for the conflict tests. + +use std::net::{IpAddr, Ipv4Addr, TcpListener, UdpSocket}; +use std::time::{Duration, Instant}; + +pub const LIFECYCLE_PORT: u16 = 8124; +pub const DOGSTATSD_UDP_PORT: u16 = 8125; +pub const TRACE_AGENT_PORT: u16 = 8126; +pub const TELEMETRY_PORT: u16 = 8999; +pub const PROXY_PORT: u16 = 9000; + +const LOOPBACK: IpAddr = IpAddr::V4(Ipv4Addr::LOCALHOST); +const ANY: IpAddr = IpAddr::V4(Ipv4Addr::UNSPECIFIED); + +/// TCP ports the extension binds in every mode. +pub const FIXED_TCP_PORTS: [u16; 3] = [LIFECYCLE_PORT, TRACE_AGENT_PORT, TELEMETRY_PORT]; + +/// The address the extension binds `port` on. +pub fn bind_ip(port: u16) -> IpAddr { + match port { + LIFECYCLE_PORT | TRACE_AGENT_PORT => LOOPBACK, + _ => ANY, + } +} + +/// Keeps a port bound until dropped. +pub struct PortGuard { + pub port: u16, + tcp: Option, + udp: Option, +} + +impl PortGuard { + /// Bind `port` over TCP on the address the extension uses. `None` when + /// something else holds it. + pub fn tcp(port: u16) -> Option { + let tcp = TcpListener::bind((bind_ip(port), port)).ok()?; + Some(Self { + port, + tcp: Some(tcp), + udp: None, + }) + } + + /// Bind `port` over UDP on the address the extension uses. + pub fn udp(port: u16) -> Option { + let udp = UdpSocket::bind((bind_ip(port), port)).ok()?; + Some(Self { + port, + tcp: None, + udp: Some(udp), + }) + } + + /// Bind over TCP, waiting up to `budget` for the port to free up. + pub async fn tcp_wait(port: u16, budget: Duration) -> Option { + let start = Instant::now(); + loop { + if let Some(g) = Self::tcp(port) { + return Some(g); + } + if start.elapsed() > budget { + return None; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + + pub async fn udp_wait(port: u16, budget: Duration) -> Option { + let start = Instant::now(); + loop { + if let Some(g) = Self::udp(port) { + return Some(g); + } + if start.elapsed() > budget { + return None; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + } +} + +/// Whether a TCP bind on `port` succeeds on both the loopback and the +/// wildcard address, since a listener on either would conflict on Linux. +/// The two probes run one after the other: the first listener is dropped +/// before the second bind, or on Linux the second would fail against our own +/// probe (two listeners cannot overlap, even with `SO_REUSEADDR`). Like +/// Tokio's, `std`'s `TcpListener::bind` sets `SO_REUSEADDR` on Unix, so a +/// connection left in TIME_WAIT does not count as busy. +pub fn tcp_port_free(port: u16) -> bool { + let loopback = TcpListener::bind((LOOPBACK, port)).is_ok(); + if !loopback { + return false; + } + TcpListener::bind((ANY, port)).is_ok() +} + +/// Same as [`tcp_port_free`] for UDP. +pub fn udp_port_free(port: u16) -> bool { + let loopback = UdpSocket::bind((LOOPBACK, port)).is_ok(); + if !loopback { + return false; + } + UdpSocket::bind((ANY, port)).is_ok() +} + +/// Ports among the fixed set that are busy right now. +pub fn busy_fixed_ports() -> Vec { + let mut busy: Vec = FIXED_TCP_PORTS + .iter() + .filter(|p| !tcp_port_free(**p)) + .map(|p| format!("tcp/{p}")) + .collect(); + if !udp_port_free(DOGSTATSD_UDP_PORT) { + busy.push(format!("udp/{DOGSTATSD_UDP_PORT}")); + } + busy +} + +/// Wait until every fixed port is free (a previous child may still be +/// dying). Returns the ports still busy after `budget`. +pub async fn wait_fixed_ports_free(budget: Duration) -> Vec { + let start = Instant::now(); + loop { + let busy = busy_fixed_ports(); + if busy.is_empty() || start.elapsed() > budget { + return busy; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } +} + +/// Fail fast with a clear message when the fixed ports are not free. Call it +/// before each spawn so a leaked process from another test is named, not +/// mistaken for a bug in the extension. +pub async fn require_fixed_ports_free() { + let busy = wait_fixed_ports_free(Duration::from_secs(5)).await; + assert!( + busy.is_empty(), + "fixed ports still busy before spawn: {busy:?}. Spawning tests must carry #[serial_test::serial(bottlecap_ports)] and run in the `fixed-ports` nextest group." + ); +} diff --git a/bottlecap/tests/common/telemetry_events.rs b/bottlecap/tests/common/telemetry_events.rs new file mode 100644 index 000000000..9dc0abb40 --- /dev/null +++ b/bottlecap/tests/common/telemetry_events.rs @@ -0,0 +1,58 @@ +// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +// `#[path]`-included by several test binaries; not every binary uses every item. +#![allow(dead_code)] +//! Telemetry API records as `serde_json::Value`, for `FakeExtensionsApi::send_telemetry`. +//! +//! Thin wrappers over the shared corpus in `tests/fixtures/telemetry.rs`, the +//! single source for record shapes. Each test crate declares `mod fixtures;` +//! at its root so these wrappers can reach it. + +use serde_json::Value; + +use crate::fixtures::telemetry as corpus; + +fn value(event: String) -> Value { + serde_json::from_str(&event).expect("fixture event is valid JSON") +} + +/// `platform.initStart` for an on-demand runtime. +pub fn platform_init_start() -> Value { + value(corpus::init_start()) +} + +/// `platform.start` for `request_id`. +pub fn platform_start(request_id: &str) -> Value { + value(corpus::platform_start(request_id)) +} + +/// `platform.runtimeDone` for `request_id`. Lambda omits `metrics` on some +/// failed invocations (a timeout), so `with_metrics = false` reproduces that. +pub fn platform_runtime_done(request_id: &str, status: &str, with_metrics: bool) -> Value { + value(corpus::runtime_done_with(request_id, status, with_metrics)) +} + +/// `platform.report` for `request_id`. +pub fn platform_report(request_id: &str) -> Value { + value(corpus::report(request_id)) +} + +/// A plain-text `function` log line. +pub fn function_log(message: &str) -> Value { + value(corpus::event( + 60, + "function", + &Value::String(message.to_string()).to_string(), + )) +} + +/// A `platform.logsDropped` record, as Lambda sends when its buffer overflows. +pub fn platform_logs_dropped(dropped_records: u64) -> Value { + value(corpus::logs_dropped(dropped_records)) +} + +/// An event of a type this extension does not know. Must never break a batch. +pub fn unknown_event() -> Value { + value(corpus::unknown_event()) +} diff --git a/bottlecap/tests/config_robustness_test.rs b/bottlecap/tests/config_robustness_test.rs new file mode 100644 index 000000000..544ec8ce4 --- /dev/null +++ b/bottlecap/tests/config_robustness_test.rs @@ -0,0 +1,100 @@ +// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 +// Integration tests are exempt from the production lint policy in Cargo.toml. +#![allow( + unreachable_pub, + clippy::pedantic, + clippy::panic, + clippy::unwrap_used, + clippy::expect_used +)] + +//! T5. Configuration fields get a value of the wrong type. The expected +//! contract, one bad value costs only that field and `DD_API_KEY` survives, +//! is a design decision for the config owner (today the whole environment +//! source is dropped). The INIT-panic cases are not a policy question. +//! + +mod fixtures; +#[path = "common/harness.rs"] +mod harness; + +use harness::prelude::*; + +const INVOKE_DEADLINE: Duration = secs(10); +const SKETCHES_PATH: &str = "/api/beta/sketches"; + +/// Run one invocation with `extra` in the env and report whether its +/// invocation metric reached the intake with `DD_API_KEY` by the invoke +/// deadline, then shut down cleanly. +async fn runs_and_flushes_with_api_key(extra: &[(&str, &str)]) -> bool { + let s = Scenario::start(FaultMode::Ok).await; + let mut bc = s.bottlecap().envs(extra.iter().copied()).spawn(); + let deadline = s + .invoke("req-1", INVOKE_DEADLINE, 1, bc.startup_deadline()) + .await + .unwrap_or_else(|| panic!("no /next poll. {}", bc.log_tail(40))); + deliver_start(&s.api, "req-1", true).await.unwrap(); + deliver_runtime_done(&s.api, "req-1", true).await.unwrap(); + assert!( + s.api.wait_for_next_calls(2, deadline).await, + "no second /next within the {INVOKE_DEADLINE:?} invoke deadline. {}", + bc.log_tail(40) + ); + let with_key = s + .intake + .wait_for( + 1, + |r| { + r.path.contains(SKETCHES_PATH) + && r.api_key() == Some(TEST_API_KEY) + && r.metric_names() + .unwrap() + .iter() + .any(|n| n == INVOCATIONS_METRIC) + }, + deadline, + ) + .await; + s.shutdown_cleanly(&mut bc).await; + if !with_key { + eprintln!( + "no {INVOCATIONS_METRIC} reached the intake with DD_API_KEY. Intake:\n{}\n{}", + s.intake.summary(), + bc.log_tail(40) + ); + } + with_key +} + +/// One wrong field at a time, for the fields the review names. Each must +/// cost only itself. One test per case so none runs into the slow-test +/// timeout. Fields with a lenient deserializer, and every extension field, +/// already cost only themselves: those tests run in CI and keep it so. +macro_rules! one_wrong_field_costs_only_itself { + ($name:ident, $key:literal, $value:literal) => { + #[tokio::test(flavor = "multi_thread")] + #[serial_test::serial(bottlecap_ports)] + async fn $name() { + assert!( + runs_and_flushes_with_api_key(&[($key, $value)]).await, + "{}={} cost more than itself", + $key, + $value + ); + } + }; +} + +one_wrong_field_costs_only_itself!( + wrong_type_flush_strategy, + "DD_SERVERLESS_FLUSH_STRATEGY", + "1000" +); +one_wrong_field_costs_only_itself!(wrong_type_flush_timeout, "DD_FLUSH_TIMEOUT", "soon"); +one_wrong_field_costs_only_itself!( + wrong_type_processing_rules, + "DD_LOGS_CONFIG_PROCESSING_RULES", + "1" +); +one_wrong_field_costs_only_itself!(wrong_type_log_level, "DD_LOG_LEVEL", "[a]"); diff --git a/bottlecap/tests/fault_injection_test.rs b/bottlecap/tests/fault_injection_test.rs new file mode 100644 index 000000000..20a61974a --- /dev/null +++ b/bottlecap/tests/fault_injection_test.rs @@ -0,0 +1,215 @@ +// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 +// Integration tests are exempt from the production lint policy in Cargo.toml. +#![allow( + unreachable_pub, + clippy::pedantic, + clippy::panic, + clippy::unwrap_used, + clippy::expect_used +)] + +//! Fault-injection contract tests: the real `bottlecap` binary against a fake +//! Extensions API and an intake that misbehaves. Each test asserts what Lambda +//! or the customer would observe: the Invoke phase ends before the deadline, +//! the faulted request reaches the intake, and nothing panics. + +mod fixtures; +#[path = "common/harness.rs"] +mod harness; + +use harness::prelude::*; + +const TRACES_PATH: &str = "/api/v0.2/traces"; +const LOGS_PATH: &str = "/api/v2/logs"; + +/// The function log [`end_invocation_and_wait`] sends for `request_id`. +fn log_text(request_id: &str) -> String { + format!("log for {request_id}") +} + +/// Deliver `platform.start`, a log and `runtimeDone` for `request_id`, then +/// return whether `/next` call number `next_call` arrived by `deadline`, the +/// INVOKE's. +async fn end_invocation_and_wait( + s: &Scenario, + request_id: &str, + cold_start: bool, + next_call: usize, + deadline: Instant, +) -> bool { + deliver_start(&s.api, request_id, cold_start).await.unwrap(); + s.api + .send_telemetry(&[function_log(&log_text(request_id))], STARTUP_BUDGET) + .await + .unwrap(); + deliver_runtime_done(&s.api, request_id, true) + .await + .unwrap(); + s.api.wait_for_next_calls(next_call, deadline).await +} + +/// T2 (no hang). A 503 storm must not hold the Invoke phase: retries are +/// bounded and `/next` is called before the deadline. +#[tokio::test(flavor = "multi_thread")] +#[serial_test::serial(bottlecap_ports)] +async fn status_503_storm_does_not_hold_next() { + let s = Scenario::start(FaultMode::Ok).await; + let mut bc = s.bottlecap().spawn(); + let deadline = s + .invoke("req-1", secs(10), 1, bc.startup_deadline()) + .await + .unwrap_or_else(|| panic!("no /next poll. {}", bc.log_tail(40))); + + s.intake.set_mode(FaultMode::Status(503)); + let released = end_invocation_and_wait(&s, "req-1", true, 2, deadline).await; + // Read before the fault clears: the end-of-invocation flush sends the log. + let logs_faulted = s.intake.requests_to(LOGS_PATH).iter().any(|r| { + r.status == 503 + && r.api_key() == Some(TEST_API_KEY) + && r.body_text().unwrap().contains(&log_text("req-1")) + }); + + s.intake.set_mode(FaultMode::Ok); + let exit = s.shutdown(&mut bc).await; + bc.assert_no_panic(); + assert!( + released, + "a 503 storm held /next past the invoke deadline. {}", + bc.log_tail(40) + ); + assert!( + logs_faulted, + "the storm never hit the logs intake with the function log and the API key. Intake:\n{}", + s.intake.summary() + ); + assert!(exit.is_some_and(|e| e.success()), "exit {exit:?}"); +} + +/// T7 (trace). A trace payload the intake rejects with 413 must be dropped, +/// not redriven forever; the next normal trace still flows and nothing +/// panics. Passes today: the current code drops every non-`Err` result. +#[tokio::test(flavor = "multi_thread")] +#[serial_test::serial(bottlecap_ports)] +async fn oversized_trace_is_dropped_and_later_traces_flow() { + let s = Scenario::start(FaultMode::Ok).await; + s.intake.set_reject_over(Some(1_000_000)); + let mut bc = s.bottlecap().spawn(); + let deadline = s + .invoke("req-1", secs(10), 1, bc.startup_deadline()) + .await + .unwrap_or_else(|| panic!("no /next poll. {}", bc.log_tail(40))); + + let status = post_trace(msgpack_trace(1, 4_000_000), STARTUP_BUDGET) + .await + .unwrap(); + eprintln!("trace agent answered {status} to a 4 MB span"); + let released = end_invocation_and_wait(&s, "req-1", true, 2, deadline).await; + assert!( + released, + "the oversized trace held /next. {}", + bc.log_tail(40) + ); + let rejected = s + .intake + .wait_for( + 1, + |r| { + r.path.contains(TRACES_PATH) + && r.status == 413 + && r.api_key() == Some(TEST_API_KEY) + && r.trace_ids().unwrap().contains(&1) + }, + deadline, + ) + .await; + + s.intake.set_reject_over(None); + let deadline = s + .invoke("req-2", secs(10), 2, deadline) + .await + .expect("no second /next poll"); + post_trace(msgpack_trace(2, 0), STARTUP_BUDGET) + .await + .unwrap(); + let released = end_invocation_and_wait(&s, "req-2", false, 3, deadline).await; + assert!(released, "no /next after the second invocation"); + let later_trace = s + .intake + .wait_for( + 1, + |r| { + r.path.contains(TRACES_PATH) + && r.status == 202 + && r.api_key() == Some(TEST_API_KEY) + && r.trace_ids().unwrap().contains(&2) + }, + deadline, + ) + .await; + + s.shutdown_cleanly(&mut bc).await; + assert!( + rejected, + "the 4 MB trace (ID 1) never reached the intake with the API key by its invoke deadline (accepted by the agent: {status}). Intake:\n{}", + s.intake.summary() + ); + // The flusher retries the 413 within its flush today; the contract is + // that the 4 MB trace is never sent again once the rejection is lifted. + // Matched by size on purpose, not decoded: a negative check must not + // pass because a body failed to decode. + let redriven = s + .intake + .requests() + .iter() + .any(|r| r.path.contains(TRACES_PATH) && r.body_len > 1_000_000 && r.status != 413); + assert!( + !redriven, + "the rejected trace was redriven after the fault cleared. Intake:\n{}", + s.intake.summary() + ); + assert!( + later_trace, + "trace ID 2, after the rejected one, never arrived with the API key by its invoke deadline. Intake:\n{}\n{}", + s.intake.summary(), + bc.log_tail(40) + ); +} + +/// `close_mid_body`: the intake sends a 2xx with headers and part of the body, +/// then closes. The request reached the intake, so this checks only that the +/// extension does not hold `/next` or crash. Passes today. +#[tokio::test(flavor = "multi_thread")] +#[serial_test::serial(bottlecap_ports)] +async fn intake_closing_mid_body_does_not_hold_next() { + let s = Scenario::start(FaultMode::Ok).await; + let mut bc = s.bottlecap().spawn(); + let deadline = s + .invoke("req-1", secs(10), 1, bc.startup_deadline()) + .await + .unwrap_or_else(|| panic!("no /next poll. {}", bc.log_tail(40))); + + s.intake.set_mode(FaultMode::CloseMidBody); + let released = end_invocation_and_wait(&s, "req-1", true, 2, deadline).await; + // Read before the fault clears: the end-of-invocation flush sends the log. + let logs_faulted = s.intake.requests_to(LOGS_PATH).iter().any(|r| { + r.mode == "close_mid_body" + && r.api_key() == Some(TEST_API_KEY) + && r.body_text().unwrap().contains(&log_text("req-1")) + }); + + s.intake.set_mode(FaultMode::Ok); + let exit = s.shutdown(&mut bc).await; + bc.assert_no_panic(); + assert!( + released, + "a truncating intake held /next past the invoke deadline. {}", + bc.log_tail(40) + ); + assert!( + logs_faulted, + "the truncating mode never served the function log with the API key. Intake:\n{}", + s.intake.summary() + ); + assert!(exit.is_some_and(|e| e.success()), "exit {exit:?}"); +} diff --git a/bottlecap/tests/lifecycle_contract_test.rs b/bottlecap/tests/lifecycle_contract_test.rs new file mode 100644 index 000000000..1882b3f1c --- /dev/null +++ b/bottlecap/tests/lifecycle_contract_test.rs @@ -0,0 +1,91 @@ +// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 +// Integration tests are exempt from the production lint policy in Cargo.toml. +#![allow( + unreachable_pub, + clippy::pedantic, + clippy::panic, + clippy::unwrap_used, + clippy::expect_used +)] + +//! Lifecycle contract tests: the real `bottlecap` binary against a fake +//! Extensions API. Each test asserts what Lambda would observe (calls to +//! `/next`, process exit, time to exit), never internal state. +//! +//! Every test here spawns the binary, which binds fixed ports, so each one +//! carries `#[serial_test::serial(bottlecap_ports)]` (for `cargo test`) and +//! the binary is in the `fixed-ports` nextest group. + +mod fixtures; +#[path = "common/harness.rs"] +mod harness; + +use harness::prelude::*; + +const INVOKE_DEADLINE: Duration = secs(10); + +/// Happy path, also a self-test of the harness: an invocation ends, the +/// extension flushes to the intake and calls `/next` again, then exits on +/// SHUTDOWN with no panic. +#[tokio::test(flavor = "multi_thread")] +#[serial_test::serial(bottlecap_ports)] +async fn invoke_then_shutdown_exits_cleanly() { + let s = Scenario::start(FaultMode::Ok).await; + let mut bc = s.bottlecap().spawn(); + + let deadline = s + .invoke("req-1", INVOKE_DEADLINE, 1, bc.startup_deadline()) + .await + .unwrap_or_else(|| panic!("no /next poll. {}", bc.log_tail(40))); + deliver_start(&s.api, "req-1", true).await.unwrap(); + s.api + .send_telemetry(&[function_log("hello from the function")], STARTUP_BUDGET) + .await + .unwrap(); + deliver_runtime_done(&s.api, "req-1", true).await.unwrap(); + assert!( + s.api.wait_for_next_calls(2, deadline).await, + "no second /next within the {INVOKE_DEADLINE:?} invoke deadline. {}", + bc.log_tail(40) + ); + + // Read the intake before SHUTDOWN: its final flush would hide a missing + // end-of-invocation flush. + // Enhanced metrics are distributions, so they land on the sketches path. + let metrics = [ + s.intake.requests_to("/api/beta/sketches"), + s.intake.requests_to("/api/v2/series"), + ] + .concat(); + let metrics_keyed = metrics.iter().all(|r| r.api_key() == Some(TEST_API_KEY)); + let invocations_flushed = metrics.iter().any(|r| { + r.metric_names() + .unwrap() + .iter() + .any(|n| n == INVOCATIONS_METRIC) + }); + let log_flushed = s.intake.requests_to("/api/v2/logs").iter().any(|r| { + r.api_key() == Some(TEST_API_KEY) + && r.body_text().unwrap().contains("hello from the function") + }); + s.shutdown_cleanly(&mut bc).await; + + // The end-of-invocation flush reached the intake, each payload with the + // configured key. + assert!( + invocations_flushed, + "no {INVOCATIONS_METRIC} flushed before SHUTDOWN. Intake:\n{}", + s.intake.summary() + ); + assert!( + metrics_keyed, + "a metrics payload without the API key. Intake:\n{}", + s.intake.summary() + ); + assert!( + log_flushed, + "the function log was not flushed with the API key before SHUTDOWN. Intake:\n{}", + s.intake.summary() + ); +} diff --git a/bottlecap/tests/port_conflict_test.rs b/bottlecap/tests/port_conflict_test.rs new file mode 100644 index 000000000..76a7948a9 --- /dev/null +++ b/bottlecap/tests/port_conflict_test.rs @@ -0,0 +1,129 @@ +// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 +// Integration tests are exempt from the production lint policy in Cargo.toml. +#![allow( + unreachable_pub, + clippy::pedantic, + clippy::panic, + clippy::unwrap_used, + clippy::expect_used +)] + +//! T6. Another extension or the runtime holds one of the fixed ports the +//! extension binds. The contract: no panic, the extension still registers and +//! answers `/next`, and it exits on SHUTDOWN. Only the listener on that port +//! is lost. + +mod fixtures; +#[path = "common/harness.rs"] +mod harness; + +use harness::prelude::*; + +const INVOKE_DEADLINE: Duration = secs(5); + +enum Proto { + Tcp, +} + +/// Run one invocation and a shutdown with `port` held. Telemetry is delivered +/// only when the extension can receive it: when the held port is the +/// Telemetry API listener's own and the extension still subscribed with it, +/// no listener is up, so the scenario ends with a SHUTDOWN on the first +/// `/next` instead. `conflict_marker` is the log line the extension writes +/// when it handles the conflict without a panic. +async fn run_with_port_held( + proto: Proto, + port: u16, + extra_env: &[(&str, &str)], + conflict_marker: Option<&str>, +) -> (Scenario, BottlecapProcess) { + // The scenario checks the fixed ports are free, so hold the port after it. + let s = Scenario::start(FaultMode::Ok).await; + let guard = match proto { + Proto::Tcp => PortGuard::tcp_wait(port, secs(5)).await, + } + .unwrap_or_else(|| panic!("could not hold port {port}")); + let mut bc = s.bottlecap().envs(extra_env.iter().copied()).spawn(); + + let startup_deadline = bc.startup_deadline(); + let reached_next = s.wait_next_or_exit(&mut bc, 1, startup_deadline).await; + let telemetry_possible = + port != TELEMETRY_PORT || s.api.telemetry_port() != Some(TELEMETRY_PORT); + let mut ended = false; + if reached_next && telemetry_possible { + let deadline = s.api.push_invoke("req-1", INVOKE_DEADLINE); + if deliver_start(&s.api, "req-1", true).await.is_ok() + && deliver_runtime_done(&s.api, "req-1", true).await.is_ok() + { + ended = s.api.wait_for_next_calls(2, deadline).await; + } + } + let exit = s.shutdown(&mut bc).await; + // Release the port only after the process is gone, so the conflict holds + // for the process's whole life. + drop(guard); + eprintln!( + "port {port} held: /next at {:?}, telemetry possible={telemetry_possible}, invocation ended={ended}, exit {exit:?}, last seen running at {:?}, intake requests={}", + s.api.next_call_at(1).map(|at| at - bc.started), + bc.seen_running() - bc.started, + s.intake.requests().len() + ); + + bc.assert_no_panic(); + assert!( + reached_next, + "with port {port} held, the extension never reached /next. API: {:?}\n{}", + s.api.request_lines(), + bc.log_tail(40) + ); + assert!( + ended || !telemetry_possible, + "with port {port} held, the invocation did not end within its {INVOKE_DEADLINE:?} deadline. API: {:?}\n{}", + s.api.request_lines(), + bc.log_tail(40) + ); + assert!( + exit.is_some_and(|e| e.success()), + "with port {port} held, exit was {exit:?}. API: {:?}\nIntake:\n{}\n{}", + s.api.request_lines(), + s.intake.summary(), + bc.log_tail(40) + ); + // A precondition, not the contract: the log line proves the held port + // really conflicted (see the log-line exception in HARNESS.md). + if let Some(marker) = conflict_marker { + assert!( + bc.log_contains(marker), + "port {port} was held but the extension did not report the conflict ({marker:?}); the guard did not conflict. {}", + bc.log_tail(40) + ); + } + (s, bc) +} + +/// 8124, the lifecycle listener: the bind error is returned and logged today. +#[tokio::test(flavor = "multi_thread")] +#[serial_test::serial(bottlecap_ports)] +async fn lifecycle_listener_port_8124_held() { + run_with_port_held( + Proto::Tcp, + LIFECYCLE_PORT, + &[], + Some("Error starting lifecycle listener"), + ) + .await; +} + +/// 8126, the trace agent: the bind error is returned and logged today. +#[tokio::test(flavor = "multi_thread")] +#[serial_test::serial(bottlecap_ports)] +async fn trace_agent_port_8126_held() { + run_with_port_held( + Proto::Tcp, + TRACE_AGENT_PORT, + &[], + Some("Error starting trace agent"), + ) + .await; +}