From f0e4939146f55a550cacc9d5772562799fe1e168 Mon Sep 17 00:00:00 2001 From: Jordan Gonzalez <30836115+duncanista@users.noreply.github.com> Date: Wed, 28 Oct 2026 14:49:03 -0400 Subject: [PATCH 1/4] [SVLS-9978] test: process-level fault-injection harness and the contract tests that pass today --- bottlecap/.config/nextest.toml | 20 + bottlecap/tests/HARNESS.md | 141 ++++++ bottlecap/tests/common/bottlecap_process.rs | 369 ++++++++++++++ bottlecap/tests/common/fake_extensions_api.rs | 295 +++++++++++ bottlecap/tests/common/fault_intake.rs | 466 ++++++++++++++++++ bottlecap/tests/common/harness.rs | 231 +++++++++ bottlecap/tests/common/port_guard.rs | 156 ++++++ bottlecap/tests/common/telemetry_events.rs | 58 +++ bottlecap/tests/config_robustness_test.rs | 85 ++++ bottlecap/tests/fault_injection_test.rs | 178 +++++++ bottlecap/tests/lifecycle_contract_test.rs | 70 +++ bottlecap/tests/port_conflict_test.rs | 115 +++++ 12 files changed, 2184 insertions(+) create mode 100644 bottlecap/.config/nextest.toml create mode 100644 bottlecap/tests/HARNESS.md create mode 100644 bottlecap/tests/common/bottlecap_process.rs create mode 100644 bottlecap/tests/common/fake_extensions_api.rs create mode 100644 bottlecap/tests/common/fault_intake.rs create mode 100644 bottlecap/tests/common/harness.rs create mode 100644 bottlecap/tests/common/port_guard.rs create mode 100644 bottlecap/tests/common/telemetry_events.rs create mode 100644 bottlecap/tests/config_robustness_test.rs create mode 100644 bottlecap/tests/fault_injection_test.rs create mode 100644 bottlecap/tests/lifecycle_contract_test.rs create mode 100644 bottlecap/tests/port_conflict_test.rs 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/tests/HARNESS.md b/bottlecap/tests/HARNESS.md new file mode 100644 index 000000000..fdd2bf950 --- /dev/null +++ b/bottlecap/tests/HARNESS.md @@ -0,0 +1,141 @@ +# 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 on SHUTDOWN, data is redriven after a fault clears, +nothing panics, memory stays bounded. 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 backlog item that fixes +the bug. 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 (benches, harness, simulator). Each test crate declares `mod fixtures;`. | +| `tests/lifecycle_contract_test.rs` | Shutdown and `/next` contracts. | +| `tests/fault_injection_test.rs` | Intake faults, secrets, oversized payloads, memory. | +| `tests/config_robustness_test.rs` | Wrong-typed config values, INIT-panic values. | +| `tests/port_conflict_test.rs` | One test per fixed port. | +| `.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 = "RV-2.x: "] +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 + assert!(s.invoke("req-1", secs(10), 1).await); // wait for /next #1, answer with INVOKE + s.intake.set_mode(FaultMode::Status(503)); // the fault + deliver_runtime_done(&s.api, "req-1", true).await.unwrap(); + assert!(s.api.wait_for_next_calls(2, secs(10)).await, "{}", bc.log_tail(40)); + s.shutdown_cleanly(&mut bc).await; // SHUTDOWN, exit 0, 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. +- Bound a flush by the invocation or shutdown deadline, not by a number from + an implementation. +- 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()`). +- 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)`, + `next_calls()`, `wait_for_next_calls(n, budget)`, `set_subscribe_status(StatusCode)`, + `request_lines()`, `send_telemetry(&[Value], budget)`, `send_telemetry_raw(Vec, budget)`, + `telemetry_port()` (from the subscribe body). 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` (shared with the Lambda simulator). + `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, budget)`, `wait_for_path`; `IntakeRequest::api_key()`, `decoded_body()`, + `body_text()`. `SynBlackhole::start()` (backlog 1, queue pre-filled), `is_saturated()`. +- `BottlecapProcess`: `scenario.bottlecap()` then `.intake(url)`, `.env`, `.envs`, `.remove_env`, + `.managed_instance()`, `.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_BUDGET`, `EXIT_BUDGET`, `TEST_API_KEY`. +- `Scenario`: `start(mode)`, `bottlecap()`, `invoke(id, deadline, poll)` (waits for the poll, + then answers it), `wait_next_or_exit`, `shutdown(&mut bc)`, `shutdown_cleanly(&mut bc)`. + Helpers: `deliver_runtime_done`, `msgpack_trace(trace_id, meta_bytes)`, `post_trace` (to 8126), + `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`. The Lambda simulator +uses the same names. + +## 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 # the not-yet-fixed ones +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. +- The T4(b) secret-recovery test 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..54c854618 --- /dev/null +++ b/bottlecap/tests/common/bottlecap_process.rs @@ -0,0 +1,369 @@ +// 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 { + env: BTreeMap, + task_root: tempfile::TempDir, +} + +impl BottlecapBuilder { + /// 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: &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 { 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. + pub fn managed_instance(self) -> Self { + 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 child = cmd + .stdin(Stdio::null()) + .stdout(Stdio::from(stdout)) + .stderr(Stdio::from(stderr)) + .spawn() + .expect("spawn bottlecap"); + BottlecapProcess { + pid: child.id(), + child, + started: Instant::now(), + stdout_path, + stderr_path, + _task_root: self.task_root, + } + } +} + +pub struct BottlecapProcess { + pub pid: u32, + child: Child, + pub started: 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() + } + + /// Exit status if the process already exited. + pub fn try_exit(&mut self) -> Option { + self.child.try_wait().ok().flatten() + } + + 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..8c3c23abb --- /dev/null +++ b/bottlecap/tests/common/fake_extensions_api.rs @@ -0,0 +1,295 @@ +// 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. 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 taken from the subscribe body). + +use std::collections::VecDeque; +use std::net::SocketAddr; +use std::sync::atomic::{AtomicU16, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use axum::{ + 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; + +/// Default Telemetry API listener port of the extension. +pub const DEFAULT_TELEMETRY_PORT: u16 = 8999; + +#[derive(Clone)] +struct Shared { + /// Events pushed by the test, served to `/next` in order. + next_queue: Arc>>, + next_notify: Arc, + next_calls: Arc, + /// `METHOD path` of every request, in arrival order. + requests: Arc>>, + subscribe_status: Arc, + /// Port in the `destination.URI` of the last subscribe body. + telemetry_port: 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_calls: Arc::new(AtomicUsize::new(0)), + requests: Arc::new(Mutex::new(Vec::new())), + subscribe_status: Arc::new(AtomicU16::new(StatusCode::OK.as_u16())), + telemetry_port: Arc::new(AtomicU16::new(DEFAULT_TELEMETRY_PORT)), + }; + 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); + } + + // --- 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(); + } + + pub fn push_invoke(&self, request_id: &str, deadline_in: Duration) { + self.push_next(invoke_event(request_id, deadline_in)); + } + + pub fn push_shutdown(&self, reason: &str, deadline_in: Duration) { + self.push_next(shutdown_event(reason, deadline_in)); + } + + /// Number of `/next` calls that arrived so far. + pub fn next_calls(&self) -> usize { + self.shared.next_calls.load(Ordering::SeqCst) + } + + /// Wait until at least `n` `/next` calls arrived. Returns whether it did + /// within `budget`. + pub async fn wait_for_next_calls(&self, n: usize, budget: Duration) -> bool { + let start = Instant::now(); + while self.next_calls() < n { + if start.elapsed() > budget { + return false; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + true + } + + /// `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. + pub fn telemetry_port(&self) -> u16 { + self.shared.telemetry_port.load(Ordering::SeqCst) + } + + pub fn telemetry_url(&self) -> String { + format!("http://127.0.0.1:{}/", self.telemetry_port()) + } + + /// 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. + pub async fn send_telemetry_raw(&self, body: Vec, budget: Duration) -> Result<(), String> { + let start = Instant::now(); + let mut last: String; + loop { + let result = self + .telemetry_client + .post(self.telemetry_url()) + .header("content-type", "application/json") + .body(body.clone()) + .send() + .await; + match result { + Ok(r) if r.status().is_success() => return Ok(()), + Ok(r) => last = format!("status {}", r.status()), + Err(e) => last = e.to_string(), + } + 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. + pub async fn send_telemetry(&self, events: &[Value], budget: Duration) -> Result<(), String> { + self.send_telemetry_raw( + Value::Array(events.to_vec()).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) -> impl IntoResponse { + record(&shared, "POST", "/2020-01-01/extension/register"); + let mut headers = HeaderMap::new(); + headers.insert( + "Lambda-Extension-Identifier", + "fake-extension-id".parse().expect("header"), + ); + ( + StatusCode::OK, + headers, + json!({"accountId": "123456789012", "functionName": "test", "functionVersion": "$LATEST", "handler": "index.handler"}).to_string(), + ) +} + +async fn subscribe(State(shared): State, body: Bytes) -> impl IntoResponse { + record(&shared, "PUT", "/2022-07-01/telemetry"); + if let Ok(v) = serde_json::from_slice::(&body) + && let Some(uri) = v["destination"]["URI"].as_str() + && let Some(port) = uri + .rsplit(':') + .next() + .and_then(|p| p.trim_end_matches('/').parse::().ok()) + { + shared.telemetry_port.store(port, Ordering::SeqCst); + } + let status = StatusCode::from_u16(shared.subscribe_status.load(Ordering::SeqCst)) + .unwrap_or(StatusCode::OK); + (status, "{}") +} + +async fn next(State(shared): State) -> impl IntoResponse { + record(&shared, "GET", "/2020-01-01/extension/event/next"); + shared.next_calls.fetch_add(1, Ordering::SeqCst); + loop { + let event = shared + .next_queue + .lock() + .expect("next queue lock") + .pop_front(); + if let Some(event) = event { + return (StatusCode::OK, event.to_string()); + } + shared.next_notify.notified().await; + } +} + +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..c4e79ab98 --- /dev/null +++ b/bottlecap/tests/common/fault_intake.rs @@ -0,0 +1,466 @@ +// 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 and +//! the mode that answered it. The fault mode can change at runtime, and one +//! path can get its own mode (for example `/credentials` for the secret-fetch +//! tests). +//! +//! Mode names, shared with the Lambda simulator: `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; + +use bytes::Bytes; +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 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, + /// 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. + pub fn decoded_body(&self) -> Vec { + match self.header("content-encoding") { + Some("zstd") => zstd::decode_all(self.body.as_slice()).unwrap_or_default(), + Some("gzip") => { + use std::io::Read; + let mut out = Vec::new(); + let _ = flate2::read::GzDecoder::new(self.body.as_slice()).read_to_end(&mut out); + out + } + _ => self.body.clone(), + } + } + + pub fn body_text(&self) -> String { + String::from_utf8_lossy(&self.decoded_body()).into_owned() + } +} + +#[derive(Clone)] +struct Shared { + mode: Arc>, + path_modes: Arc>>, + /// Bodies over this size get a 413, whatever the mode. + 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. + if *accept_shared.mode.lock().expect("mode lock") == FaultMode::Blackhole { + 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 match `pred`. Returns whether it happened within `budget`. + pub async fn wait_for( + &self, + n: usize, + pred: impl Fn(&IntakeRequest) -> bool, + budget: Duration, + ) -> bool { + let start = std::time::Instant::now(); + loop { + if self.requests().iter().filter(|r| pred(r)).count() >= n { + return true; + } + if start.elapsed() > budget { + return false; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + } + + /// Wait until `n` requests to a path containing `needle` arrived. + pub async fn wait_for_path(&self, needle: &str, n: usize, budget: Duration) -> bool { + self.wait_for(n, |r| r.path.contains(needle), budget).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>, Infallible> { + 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 body = req + .into_body() + .collect() + .await + .map(|c| c.to_bytes().to_vec()) + .unwrap_or_default(); + let mode = effective_mode(&shared, &path); + let reject_over = *shared.reject_over.lock().expect("reject lock"); + + let (status, response) = if reject_over.is_some_and(|max| 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(&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(delay) => { + tokio::time::sleep(*delay).await; + (202, status_response(202, "{}".into())) + } + FaultMode::Blackhole => { + // A request on a connection opened before the mode changed: + // record it, then hold it open for the life of the connection. + record(&shared, &method, &path, headers, body, &mode, 0); + std::future::pending::<()>().await; + unreachable!() + } + 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) + } + } + }; + record(&shared, &method, &path, headers, body, &mode, status); + Ok(response) +} + +fn record( + shared: &Shared, + method: &str, + path: &str, + headers: HashMap, + body: Vec, + mode: &FaultMode, + status: u16, +) { + shared + .requests + .lock() + .expect("requests lock") + .push(IntakeRequest { + method: method.to_string(), + path: path.to_string(), + headers, + body_len: body.len(), + body, + mode: mode.name(), + status, + }); +} + +/// 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..ca617843c --- /dev/null +++ b/bottlecap/tests/common/harness.rs @@ -0,0 +1,231 @@ +// 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::{DEFAULT_TELEMETRY_PORT, 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::{ + Scenario, deliver_runtime_done, incompressible_text, ms, msgpack_trace, post_trace, secs, + }; + pub use std::time::{Duration, Instant}; +} + +use std::process::ExitStatus; +use std::time::Duration; + +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) +} + +/// A fake Extensions API and a fault intake, started together. +/// +/// ```ignore +/// let s = Scenario::start(FaultMode::Ok).await; +/// let mut bc = s.bottlecap().spawn(); +/// assert!(s.invoke("req-1", secs(10), 1).await); +/// ``` +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` to be open, then answer it with an + /// INVOKE, as Lambda does. Returns `false` when the poll never opened. + pub async fn invoke(&self, request_id: &str, deadline_in: Duration, poll: usize) -> bool { + if !self + .api + .wait_for_next_calls(poll, bottlecap_process::STARTUP_BUDGET) + .await + { + return false; + } + self.api.push_invoke(request_id, deadline_in); + true + } + + /// Wait for `/next` call number `n`, giving up early when the process + /// exits (a crash at INIT should not cost the whole budget). + pub async fn wait_next_or_exit( + &self, + bc: &mut BottlecapProcess, + n: usize, + budget: Duration, + ) -> bool { + let start = std::time::Instant::now(); + while self.api.next_calls() < n { + if start.elapsed() > budget || !bc.is_running() { + return self.api.next_calls() >= n; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + true + } + + /// Answer the open `/next` with a SHUTDOWN (2 s deadline, Lambda's + /// default) and wait for the process to exit. Returns the exit status, + /// `None` when it had to be killed. + pub async fn shutdown(&self, bc: &mut BottlecapProcess) -> Option { + self.api.push_shutdown("spindown", secs(2)); + bc.wait_for_exit(bottlecap_process::EXIT_BUDGET).await + } + + /// [`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` then `platform.runtimeDone` for `request_id`, delivered +/// to the extension's Telemetry API listener. +pub async fn deliver_runtime_done( + api: &FakeExtensionsApi, + request_id: &str, + with_metrics: bool, +) -> Result<(), String> { + api.send_telemetry( + &[ + telemetry_events::platform_start(request_id), + 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..dfe926b13 --- /dev/null +++ b/bottlecap/tests/common/port_guard.rs @@ -0,0 +1,156 @@ +// 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`). +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..41a7e45a4 --- /dev/null +++ b/bottlecap/tests/config_robustness_test.rs @@ -0,0 +1,85 @@ +// 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::*; + +/// Run one invocation with `extra` in the env and report whether an intake +/// request carried `DD_API_KEY`, 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(); + assert!( + s.invoke("req-1", secs(10), 1).await, + "no /next poll. {}", + bc.log_tail(40) + ); + deliver_runtime_done(&s.api, "req-1", true).await.unwrap(); + assert!( + s.api.wait_for_next_calls(2, secs(10)).await, + "no second /next. {}", + bc.log_tail(40) + ); + let with_key = s + .intake + .wait_for(1, |r| r.api_key() == Some(TEST_API_KEY), secs(5)) + .await; + s.shutdown_cleanly(&mut bc).await; + if !with_key { + eprintln!( + "no intake request carried 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..49da5b5b2 --- /dev/null +++ b/bottlecap/tests/fault_injection_test.rs @@ -0,0 +1,178 @@ +// 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 (T1 to T4, T7, T8 of the +//! review's test plan). Each test asserts what Lambda or the customer would +//! observe: the Invoke phase ends before the deadline, data is redriven after +//! the fault clears, nothing panics, memory stays bounded. + +mod fixtures; +#[path = "common/harness.rs"] +mod harness; + +use harness::prelude::*; + +const TRACES_PATH: &str = "/api/v0.2/traces"; +/// Deliver a log and `runtimeDone` for `request_id`, then wait for `/next` +/// call number `next_call`. Returns `(released, time to release)`. +async fn end_invocation_and_wait( + s: &Scenario, + request_id: &str, + next_call: usize, + budget: Duration, +) -> (bool, Duration) { + s.api + .send_telemetry( + &[function_log(&format!("log for {request_id}"))], + STARTUP_BUDGET, + ) + .await + .unwrap(); + deliver_runtime_done(&s.api, request_id, true) + .await + .unwrap(); + let start = Instant::now(); + let released = s.api.wait_for_next_calls(next_call, budget).await; + (released, start.elapsed()) +} + +/// 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(); + assert!( + s.invoke("req-1", secs(10), 1).await, + "no /next poll. {}", + bc.log_tail(40) + ); + + s.intake.set_mode(FaultMode::Status(503)); + let (released, took) = end_invocation_and_wait(&s, "req-1", 2, secs(10)).await; + + s.intake.set_mode(FaultMode::Ok); + let exit = s.shutdown(&mut bc).await; + bc.assert_no_panic(); + assert!( + released, + "a 503 storm held /next for {took:?}. {}", + bc.log_tail(40) + ); + assert!( + s.intake.requests().iter().any(|r| r.status == 503), + "the storm never hit the intake. 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(); + assert!( + s.invoke("req-1", secs(10), 1).await, + "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", 2, secs(15)).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, + secs(5), + ) + .await; + + s.intake.set_reject_over(None); + assert!(s.invoke("req-2", secs(10), 2).await, "no second /next poll"); + post_trace(msgpack_trace(2, 0), STARTUP_BUDGET) + .await + .unwrap(); + let (released, _) = end_invocation_and_wait(&s, "req-2", 3, secs(15)).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.body_len < 1_000_000, + secs(10), + ) + .await; + + s.shutdown_cleanly(&mut bc).await; + assert!( + rejected, + "the 4 MB trace never reached the intake (accepted by the agent: {status}). Intake:\n{}", + s.intake.summary() + ); + assert!( + later_trace, + "the trace after the rejected one never arrived. Intake:\n{}\n{}", + s.intake.summary(), + bc.log_tail(40) + ); +} + +/// `close_mid_body`: the intake sends headers and part of the body, then +/// closes. The extension must treat it as a failed request and keep flowing. +/// 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(); + assert!( + s.invoke("req-1", secs(10), 1).await, + "no /next poll. {}", + bc.log_tail(40) + ); + + s.intake.set_mode(FaultMode::CloseMidBody); + let (released, took) = end_invocation_and_wait(&s, "req-1", 2, secs(10)).await; + + s.intake.set_mode(FaultMode::Ok); + let exit = s.shutdown(&mut bc).await; + bc.assert_no_panic(); + assert!( + released, + "a truncating intake held /next for {took:?}. {}", + bc.log_tail(40) + ); + assert!( + s.intake + .requests() + .iter() + .any(|r| r.mode == "close_mid_body"), + "the truncating mode never served a request. 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..09672c777 --- /dev/null +++ b/bottlecap/tests/lifecycle_contract_test.rs @@ -0,0 +1,70 @@ +// 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::*; + +/// 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(); + + assert!( + s.invoke("req-1", secs(10), 1).await, + "no /next poll. {}", + bc.log_tail(40) + ); + 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, secs(10)).await, + "no second /next after runtimeDone. {}", + bc.log_tail(40) + ); + + s.shutdown_cleanly(&mut bc).await; + // The end-of-invocation flush reached the intake with the configured key. + let with_key = s + .intake + .requests() + .iter() + .filter(|r| r.api_key() == Some(TEST_API_KEY)) + .count(); + assert!( + with_key > 0, + "no intake request carried the API key. Intake:\n{}", + s.intake.summary() + ); + // Enhanced metrics are distributions, so they land on the sketches path. + assert!( + s.intake.count("/api/beta/sketches") + s.intake.count("/api/v2/series") > 0, + "no metrics flushed. 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..db0d8d294 --- /dev/null +++ b/bottlecap/tests/port_conflict_test.rs @@ -0,0 +1,115 @@ +// 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::*; + +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 reached_next = s.wait_next_or_exit(&mut bc, 1, STARTUP_BUDGET).await; + let reached_at = bc.uptime(); + let telemetry_possible = port != TELEMETRY_PORT || s.api.telemetry_port() != TELEMETRY_PORT; + let mut flushed = false; + if reached_next && telemetry_possible { + s.api.push_invoke("req-1", secs(5)); + if deliver_runtime_done(&s.api, "req-1", true).await.is_ok() { + flushed = s.api.wait_for_next_calls(2, secs(10)).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 {reached_at:?}, telemetry possible={telemetry_possible}, invocation flushed={flushed}, exit {exit:?} at {:?}, intake requests={}", + bc.uptime(), + 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!( + 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) + ); + 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; +} From 51f541581a4827287dff7bb53a3a0209be542abe Mon Sep 17 00:00:00 2001 From: Jordan Gonzalez <30836115+duncanista@users.noreply.github.com> Date: Wed, 28 Oct 2026 17:27:31 -0400 Subject: [PATCH 2/4] test: address review feedback on the fault-injection harness - fault_injection_test: measure the /next budget from when the INVOKE was queued - harness: shutdown_cleanly asserts the exit came within the SHUTDOWN deadline --- bottlecap/tests/HARNESS.md | 2 +- bottlecap/tests/common/harness.rs | 21 ++++++++++++++++----- bottlecap/tests/fault_injection_test.rs | 24 ++++++++++++++++-------- 3 files changed, 33 insertions(+), 14 deletions(-) diff --git a/bottlecap/tests/HARNESS.md b/bottlecap/tests/HARNESS.md index fdd2bf950..edbe5f754 100644 --- a/bottlecap/tests/HARNESS.md +++ b/bottlecap/tests/HARNESS.md @@ -48,7 +48,7 @@ async fn my_fault() { s.intake.set_mode(FaultMode::Status(503)); // the fault deliver_runtime_done(&s.api, "req-1", true).await.unwrap(); assert!(s.api.wait_for_next_calls(2, secs(10)).await, "{}", bc.log_tail(40)); - s.shutdown_cleanly(&mut bc).await; // SHUTDOWN, exit 0, no panic + s.shutdown_cleanly(&mut bc).await; // SHUTDOWN, exit 0 within 2 s, no panic } ``` diff --git a/bottlecap/tests/common/harness.rs b/bottlecap/tests/common/harness.rs index ca617843c..2ad6e6af7 100644 --- a/bottlecap/tests/common/harness.rs +++ b/bottlecap/tests/common/harness.rs @@ -54,6 +54,9 @@ 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); + /// A fake Extensions API and a fault intake, started together. /// /// ```ignore @@ -111,17 +114,20 @@ impl Scenario { true } - /// Answer the open `/next` with a SHUTDOWN (2 s deadline, Lambda's - /// default) and wait for the process to exit. Returns the exit status, - /// `None` when it had to be killed. + /// Answer the open `/next` with a SHUTDOWN ([`SHUTDOWN_DEADLINE`]) and + /// wait for the process to exit. Returns the exit status, `None` when it + /// had to be killed. pub async fn shutdown(&self, bc: &mut BottlecapProcess) -> Option { - self.api.push_shutdown("spindown", secs(2)); + self.api.push_shutdown("spindown", SHUTDOWN_DEADLINE); bc.wait_for_exit(bottlecap_process::EXIT_BUDGET).await } - /// [`shutdown`](Self::shutdown), then assert no panic and a clean exit. + /// [`shutdown`](Self::shutdown), then assert no panic and a clean exit + /// within the SHUTDOWN deadline. pub async fn shutdown_cleanly(&self, bc: &mut BottlecapProcess) { + let start = std::time::Instant::now(); let exit = self.shutdown(bc).await; + let took = start.elapsed(); bc.assert_no_panic(); assert!( exit.is_some_and(|e| e.success()), @@ -129,6 +135,11 @@ impl Scenario { self.api.request_lines(), bc.log_tail(40) ); + assert!( + took <= SHUTDOWN_DEADLINE, + "exit took {took:?}, past the {SHUTDOWN_DEADLINE:?} SHUTDOWN deadline. {}", + bc.log_tail(40) + ); } } diff --git a/bottlecap/tests/fault_injection_test.rs b/bottlecap/tests/fault_injection_test.rs index 49da5b5b2..0de8f60f5 100644 --- a/bottlecap/tests/fault_injection_test.rs +++ b/bottlecap/tests/fault_injection_test.rs @@ -23,11 +23,13 @@ use harness::prelude::*; const TRACES_PATH: &str = "/api/v0.2/traces"; /// Deliver a log and `runtimeDone` for `request_id`, then wait for `/next` -/// call number `next_call`. Returns `(released, time to release)`. +/// call number `next_call` until `budget` after `invoked`, when the INVOKE +/// was queued. Returns `(released, time since the INVOKE)`. async fn end_invocation_and_wait( s: &Scenario, request_id: &str, next_call: usize, + invoked: Instant, budget: Duration, ) -> (bool, Duration) { s.api @@ -40,9 +42,11 @@ async fn end_invocation_and_wait( deliver_runtime_done(&s.api, request_id, true) .await .unwrap(); - let start = Instant::now(); - let released = s.api.wait_for_next_calls(next_call, budget).await; - (released, start.elapsed()) + let released = s + .api + .wait_for_next_calls(next_call, budget.saturating_sub(invoked.elapsed())) + .await; + (released, invoked.elapsed()) } /// T2 (no hang). A 503 storm must not hold the Invoke phase: retries are @@ -57,9 +61,10 @@ async fn status_503_storm_does_not_hold_next() { "no /next poll. {}", bc.log_tail(40) ); + let invoked = Instant::now(); s.intake.set_mode(FaultMode::Status(503)); - let (released, took) = end_invocation_and_wait(&s, "req-1", 2, secs(10)).await; + let (released, took) = end_invocation_and_wait(&s, "req-1", 2, invoked, secs(10)).await; s.intake.set_mode(FaultMode::Ok); let exit = s.shutdown(&mut bc).await; @@ -91,12 +96,13 @@ async fn oversized_trace_is_dropped_and_later_traces_flow() { "no /next poll. {}", bc.log_tail(40) ); + let invoked = Instant::now(); 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", 2, secs(15)).await; + let (released, _) = end_invocation_and_wait(&s, "req-1", 2, invoked, secs(15)).await; assert!( released, "the oversized trace held /next. {}", @@ -113,10 +119,11 @@ async fn oversized_trace_is_dropped_and_later_traces_flow() { s.intake.set_reject_over(None); assert!(s.invoke("req-2", secs(10), 2).await, "no second /next poll"); + let invoked = Instant::now(); post_trace(msgpack_trace(2, 0), STARTUP_BUDGET) .await .unwrap(); - let (released, _) = end_invocation_and_wait(&s, "req-2", 3, secs(15)).await; + let (released, _) = end_invocation_and_wait(&s, "req-2", 3, invoked, secs(15)).await; assert!(released, "no /next after the second invocation"); let later_trace = s .intake @@ -154,9 +161,10 @@ async fn intake_closing_mid_body_does_not_hold_next() { "no /next poll. {}", bc.log_tail(40) ); + let invoked = Instant::now(); s.intake.set_mode(FaultMode::CloseMidBody); - let (released, took) = end_invocation_and_wait(&s, "req-1", 2, secs(10)).await; + let (released, took) = end_invocation_and_wait(&s, "req-1", 2, invoked, secs(10)).await; s.intake.set_mode(FaultMode::Ok); let exit = s.shutdown(&mut bc).await; From 399b90b8e677741a4f3dc41669aa12d39aea8f65 Mon Sep 17 00:00:00 2001 From: Jordan Gonzalez <30836115+duncanista@users.noreply.github.com> Date: Wed, 28 Oct 2026 20:43:34 -0400 Subject: [PATCH 3/4] test: tighten the trace and lifecycle contract checks - oversized trace: wait for /next against the 10 s INVOKE deadline, and assert the rejected trace is never sent again after the rejection is lifted - lifecycle: read the intake before SHUTDOWN so the final flush cannot hide a missing end-of-invocation flush --- bottlecap/tests/fault_injection_test.rs | 16 ++++++++++++++-- bottlecap/tests/lifecycle_contract_test.rs | 16 ++++++++++------ 2 files changed, 24 insertions(+), 8 deletions(-) diff --git a/bottlecap/tests/fault_injection_test.rs b/bottlecap/tests/fault_injection_test.rs index 0de8f60f5..82b10d7e7 100644 --- a/bottlecap/tests/fault_injection_test.rs +++ b/bottlecap/tests/fault_injection_test.rs @@ -102,7 +102,7 @@ async fn oversized_trace_is_dropped_and_later_traces_flow() { .await .unwrap(); eprintln!("trace agent answered {status} to a 4 MB span"); - let (released, _) = end_invocation_and_wait(&s, "req-1", 2, invoked, secs(15)).await; + let (released, _) = end_invocation_and_wait(&s, "req-1", 2, invoked, secs(10)).await; assert!( released, "the oversized trace held /next. {}", @@ -123,7 +123,7 @@ async fn oversized_trace_is_dropped_and_later_traces_flow() { post_trace(msgpack_trace(2, 0), STARTUP_BUDGET) .await .unwrap(); - let (released, _) = end_invocation_and_wait(&s, "req-2", 3, invoked, secs(15)).await; + let (released, _) = end_invocation_and_wait(&s, "req-2", 3, invoked, secs(10)).await; assert!(released, "no /next after the second invocation"); let later_trace = s .intake @@ -140,6 +140,18 @@ async fn oversized_trace_is_dropped_and_later_traces_flow() { "the 4 MB trace never reached the intake (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. + 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, "the trace after the rejected one never arrived. Intake:\n{}\n{}", diff --git a/bottlecap/tests/lifecycle_contract_test.rs b/bottlecap/tests/lifecycle_contract_test.rs index 09672c777..c2e3f8e1f 100644 --- a/bottlecap/tests/lifecycle_contract_test.rs +++ b/bottlecap/tests/lifecycle_contract_test.rs @@ -48,23 +48,27 @@ async fn invoke_then_shutdown_exits_cleanly() { bc.log_tail(40) ); - s.shutdown_cleanly(&mut bc).await; - // The end-of-invocation flush reached the intake with the configured key. + // Read the intake before SHUTDOWN: its final flush would hide a missing + // end-of-invocation flush. let with_key = s .intake .requests() .iter() .filter(|r| r.api_key() == Some(TEST_API_KEY)) .count(); + // Enhanced metrics are distributions, so they land on the sketches path. + let metrics = s.intake.count("/api/beta/sketches") + s.intake.count("/api/v2/series"); + s.shutdown_cleanly(&mut bc).await; + + // The end-of-invocation flush reached the intake with the configured key. assert!( with_key > 0, - "no intake request carried the API key. Intake:\n{}", + "no intake request carried the API key before SHUTDOWN. Intake:\n{}", s.intake.summary() ); - // Enhanced metrics are distributions, so they land on the sketches path. assert!( - s.intake.count("/api/beta/sketches") + s.intake.count("/api/v2/series") > 0, - "no metrics flushed. Intake:\n{}", + metrics > 0, + "no metrics flushed before SHUTDOWN. Intake:\n{}", s.intake.summary() ); } From 3e32723992d91dbe240e50ddaa8cd86cdae705fe Mon Sep 17 00:00:00 2001 From: Jordan Gonzalez <30836115+duncanista@users.noreply.github.com> Date: Fri, 30 Oct 2026 16:28:44 -0400 Subject: [PATCH 4/4] test: enforce the invoke deadline in port-conflict tests, honor path modes in blackhole --- bottlecap/tests/common/fault_intake.rs | 13 +++++++++++-- bottlecap/tests/fault_injection_test.rs | 6 +++--- bottlecap/tests/port_conflict_test.rs | 16 ++++++++++++++-- 3 files changed, 28 insertions(+), 7 deletions(-) diff --git a/bottlecap/tests/common/fault_intake.rs b/bottlecap/tests/common/fault_intake.rs index c4e79ab98..a31f5291b 100644 --- a/bottlecap/tests/common/fault_intake.rs +++ b/bottlecap/tests/common/fault_intake.rs @@ -153,8 +153,17 @@ impl FaultIntake { continue; }; accept_shared.connections.fetch_add(1, Ordering::SeqCst); - // A connection accepted in blackhole mode is never read. - if *accept_shared.mode.lock().expect("mode lock") == FaultMode::Blackhole { + // A connection accepted in blackhole mode is never read, unless a path has + // its own mode: then `handle` reads the request 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; } diff --git a/bottlecap/tests/fault_injection_test.rs b/bottlecap/tests/fault_injection_test.rs index 82b10d7e7..855dd7dae 100644 --- a/bottlecap/tests/fault_injection_test.rs +++ b/bottlecap/tests/fault_injection_test.rs @@ -160,9 +160,9 @@ async fn oversized_trace_is_dropped_and_later_traces_flow() { ); } -/// `close_mid_body`: the intake sends headers and part of the body, then -/// closes. The extension must treat it as a failed request and keep flowing. -/// Passes today. +/// `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() { diff --git a/bottlecap/tests/port_conflict_test.rs b/bottlecap/tests/port_conflict_test.rs index db0d8d294..2facb8e73 100644 --- a/bottlecap/tests/port_conflict_test.rs +++ b/bottlecap/tests/port_conflict_test.rs @@ -20,6 +20,8 @@ mod harness; use harness::prelude::*; +const INVOKE_DEADLINE: Duration = secs(5); + enum Proto { Tcp, } @@ -49,9 +51,13 @@ async fn run_with_port_held( let telemetry_possible = port != TELEMETRY_PORT || s.api.telemetry_port() != TELEMETRY_PORT; let mut flushed = false; if reached_next && telemetry_possible { - s.api.push_invoke("req-1", secs(5)); + let invoked = Instant::now(); + s.api.push_invoke("req-1", INVOKE_DEADLINE); if deliver_runtime_done(&s.api, "req-1", true).await.is_ok() { - flushed = s.api.wait_for_next_calls(2, secs(10)).await; + flushed = s + .api + .wait_for_next_calls(2, INVOKE_DEADLINE.saturating_sub(invoked.elapsed())) + .await; } } let exit = s.shutdown(&mut bc).await; @@ -71,6 +77,12 @@ async fn run_with_port_held( s.api.request_lines(), bc.log_tail(40) ); + assert!( + flushed || !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{}",