From 31214a04fc88c80a527e8b9a185b657b29ae2a57 Mon Sep 17 00:00:00 2001 From: Tianning Li Date: Wed, 21 Oct 2026 15:53:04 -0400 Subject: [PATCH] test(logs): add contract tests, a fake logs intake and a benchmark for the logs pipeline Why: we are about to replace the logs aggregator and flusher with the shared datadog-logs-agent crate, and Lambda users must not notice. These changes record exactly what bottlecap does today, so any change in what it sends shows up as a failing test instead of a surprise in production, and give a performance baseline to compare against. What: - refactor: the code that creates the logs aggregator, agent and flusher moves from main.rs into build_logs_pipeline() in the library, so tests run exactly the production wiring. main.rs still starts the agent in the background; runtime behavior is unchanged. A unit test checks that a log sent into the pipeline reaches the aggregator. - fake logs intake: a test server that accepts log requests over HTTP or HTTPS (self-signed test certificate), records each request (method, URL, headers, body), decompresses zstd bodies, and answers with status codes chosen by the test or hangs on purpose. It has its own tests. - contract tests (tests/logs_contract_test.rs) through the real pipeline: - every log field by field: START/END/REPORT lines, plain and JSON function logs, logs that arrive before START, Managed Instance mode, processing rules, OOM lines and durable execution logs; - the URL path, API key, headers, method and compression of each request for the Datadog intake (compressed and not), Observability Pipelines Worker, two additional HTTPS endpoints, and a custom intake URL with a path; - retry, redrive and drop rules: 403 dropped with no retry; 400, 401, 413, 429 and 500 tried three times, handed back and resent once more; a redrive resends only the failed request; timeouts and refused connections are handed back; additional-endpoint failures are resent only to that endpoint; a missing API key drops the primary's logs with an error while additional endpoints still get them; - batching (2,500 small logs and three 2 MiB logs arrive exactly once within the per-request limits) and oversized logs (a 2 MiB log is sent with one warning); - how many warning and error lines the pipeline writes. Two tests record known bugs on purpose (OPW applied to additional endpoints; one log over 5 MiB blocking the queue); the next changes fix them and update those tests. - benchmark (benches/logs_pipeline.rs): turning Lambda telemetry events into batched logs, and flushing 1,000 varied logs to a local fake intake with zero or two additional endpoints, using the extension's allocator. Adds criterion as a dev-dependency only; the extension binary is unchanged. Co-Authored-By: Claude Opus 5.5 --- bottlecap/Cargo.lock | 152 +++ bottlecap/Cargo.toml | 7 + bottlecap/benches/logs_pipeline.rs | 249 +++++ bottlecap/src/bin/bottlecap/main.rs | 35 +- bottlecap/src/logs/mod.rs | 1 + bottlecap/src/logs/pipeline.rs | 112 ++ bottlecap/tests/common/fake_logs_intake.rs | 379 +++++++ bottlecap/tests/logs_contract_test.rs | 1169 ++++++++++++++++++++ 8 files changed, 2083 insertions(+), 21 deletions(-) create mode 100644 bottlecap/benches/logs_pipeline.rs create mode 100644 bottlecap/src/logs/pipeline.rs create mode 100644 bottlecap/tests/common/fake_logs_intake.rs create mode 100644 bottlecap/tests/logs_contract_test.rs diff --git a/bottlecap/Cargo.lock b/bottlecap/Cargo.lock index 3475b27e9..1915e850b 100644 --- a/bottlecap/Cargo.lock +++ b/bottlecap/Cargo.lock @@ -44,6 +44,18 @@ dependencies = [ "libc", ] +[[package]] +name = "anes" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" + +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + [[package]] name = "anyhow" version = "1.0.103" @@ -496,6 +508,7 @@ dependencies = [ "bytes", "chrono", "cookie", + "criterion", "datadog-agent-config", "datadog-agent-trace-sampler", "datadog-fips", @@ -624,6 +637,12 @@ dependencies = [ "crossbeam-channel", ] +[[package]] +name = "cast" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" + [[package]] name = "cc" version = "1.2.57" @@ -686,6 +705,33 @@ dependencies = [ "windows-link", ] +[[package]] +name = "ciborium" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e" +dependencies = [ + "ciborium-io", + "ciborium-ll", + "serde", +] + +[[package]] +name = "ciborium-io" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757" + +[[package]] +name = "ciborium-ll" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9" +dependencies = [ + "ciborium-io", + "half", +] + [[package]] name = "clang-sys" version = "1.8.1" @@ -697,6 +743,31 @@ dependencies = [ "libloading", ] +[[package]] +name = "clap" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aa8876b300ab35ba921adea3dfd70157a46249b33f95c9084ae5709785478946" +dependencies = [ + "clap_builder", +] + +[[package]] +name = "clap_builder" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0797fb7aeb1406c84efac526901f7ec3ead2124f946b494e72879d4b54704d" +dependencies = [ + "anstyle", + "clap_lex", +] + +[[package]] +name = "clap_lex" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c133bc6a41be0d194c306b5506d15e6feeea7b1d6604bd3f8310dfb2ca96486" + [[package]] name = "cmake" version = "0.1.58" @@ -810,6 +881,40 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "criterion" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f" +dependencies = [ + "anes", + "cast", + "ciborium", + "clap", + "criterion-plot", + "is-terminal", + "itertools 0.10.5", + "num-traits", + "once_cell", + "oorandom", + "regex", + "serde", + "serde_derive", + "serde_json", + "tinytemplate", + "walkdir", +] + +[[package]] +name = "criterion-plot" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1" +dependencies = [ + "cast", + "itertools 0.10.5", +] + [[package]] name = "crossbeam-channel" version = "0.5.15" @@ -1479,6 +1584,17 @@ dependencies = [ "tracing", ] +[[package]] +name = "half" +version = "2.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b" +dependencies = [ + "cfg-if", + "crunchy", + "zerocopy", +] + [[package]] name = "hashbrown" version = "0.12.3" @@ -1974,6 +2090,26 @@ dependencies = [ "serde", ] +[[package]] +name = "is-terminal" +version = "0.4.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" +dependencies = [ + "hermit-abi", + "libc", + "windows-sys 0.61.2", +] + +[[package]] +name = "itertools" +version = "0.10.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0fd2260e829bddf4cb6ea802289de2f86d6a7a690192fbe91b3f46e0f2c8473" +dependencies = [ + "either", +] + [[package]] name = "itertools" version = "0.11.0" @@ -2803,6 +2939,12 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "oorandom" +version = "11.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e" + [[package]] name = "openssl-probe" version = "0.1.6" @@ -4441,6 +4583,16 @@ dependencies = [ "zerovec", ] +[[package]] +name = "tinytemplate" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "tinyvec" version = "1.11.0" diff --git a/bottlecap/Cargo.toml b/bottlecap/Cargo.toml index 3f56cd2bb..c32a9dcc9 100644 --- a/bottlecap/Cargo.toml +++ b/bottlecap/Cargo.toml @@ -117,6 +117,13 @@ tokio-rustls = { version = "0.26", default-features = false, features = ["loggin rmp-serde = { version = "1.3.1", default-features = false } flate2 = { version = "1.1", default-features = false, features = ["rust_backend"] } +# logs pipeline benchmark (benches/logs_pipeline.rs) +criterion = { version = "0.5", default-features = false, features = ["cargo_bench_support"] } + +[[bench]] +name = "logs_pipeline" +harness = false + [build-dependencies] # No external dependencies needed for the build script diff --git a/bottlecap/benches/logs_pipeline.rs b/bottlecap/benches/logs_pipeline.rs new file mode 100644 index 000000000..1e5223a61 --- /dev/null +++ b/bottlecap/benches/logs_pipeline.rs @@ -0,0 +1,249 @@ +// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +//! Logs pipeline benchmark, used as the performance baseline for the migration to +//! `datadog-logs-agent`. Compare runs with `--save-baseline` / `--baseline`. + +// Its `self_tests` module is unused when clippy checks this bench in test mode. +#[allow(unused_imports)] +#[path = "../tests/common/fake_logs_intake.rs"] +mod fake_logs_intake; + +use std::collections::HashMap; +use std::sync::Arc; + +use bottlecap::LAMBDA_RUNTIME_SLUG; +use bottlecap::config::Config; +use bottlecap::config::logs_additional_endpoints::LogsAdditionalEndpoint; +use bottlecap::event_bus::EventBus; +use bottlecap::extension::telemetry::events::TelemetryEvent; +use bottlecap::logs::pipeline::{LogsPipeline, build_logs_pipeline}; +use bottlecap::tags::provider::Provider; +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use dogstatsd::api_key::ApiKeyFactory; +use fake_logs_intake::FakeLogsIntake; +use serde_json::json; + +// Same allocator as the extension binary. +#[cfg(not(target_env = "msvc"))] +#[global_allocator] +static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; + +const ARN: &str = "arn:aws:lambda:us-east-1:123456789012:function:bench"; +const INIT_START: &str = r#"{"time":"2022-10-21T14:05:03.000Z","type":"platform.initStart","record":{"initializationType":"on-demand","phase":"init"}}"#; +const WORDS: [&str; 16] = [ + "request", "handler", "cache", "miss", "user", "order", "payment", "retry", "timeout", "queue", + "db", "query", "ok", "error", "shard", "session", +]; + +fn runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime") +} + +/// Must be called inside the runtime (the pipeline spawns its aggregator task). +fn build_pipeline(config: Config, is_managed_instance_mode: bool) -> (LogsPipeline, EventBus) { + let config = Arc::new(config); + let tags_provider = Arc::new(Provider::new( + Arc::clone(&config), + LAMBDA_RUNTIME_SLUG.to_string(), + &HashMap::from([("function_arn".to_string(), ARN.to_string())]), + )); + let (event_bus, event_bus_tx) = EventBus::run(); + // Pooled so loopback connections are reused: a fresh connection per request exhausts + // ephemeral ports within a criterion run, and connection setup isn't what the + // migration changes. + let client = datadog_fips::reqwest_adapter::create_reqwest_client_builder() + .expect("client builder") + .danger_accept_invalid_certs(true) + .build() + .expect("client"); + let pipeline = build_logs_pipeline( + &config, + Arc::new(ApiKeyFactory::new("bench-key")), + &tags_provider, + event_bus_tx, + is_managed_instance_mode, + &client, + ); + (pipeline, event_bus) +} + +fn config(primary: &FakeLogsIntake, extras: &[FakeLogsIntake]) -> Config { + Config { + http_protocol: Some("http1".to_string()), + skip_ssl_validation: true, + logs_config_logs_dd_url: primary.base_url(), + logs_config_use_compression: true, + logs_config_compression_level: 3, + logs_config_additional_endpoints: extras + .iter() + .map(|e| LogsAdditionalEndpoint { + api_key: "extra-key".to_string(), + host: "127.0.0.1".to_string(), + port: u32::from(e.port()), + is_reliable: true, + }) + .collect(), + ..Config::default() + } +} + +/// Varied but deterministic, so a batch compresses about 9:1 with zstd level 3. +fn log_message(i: usize) -> String { + let h = (i as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15); + let span = (h ^ (h >> 31)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + let user = (span ^ (span >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + let w = |shift: u32| WORDS[usize::try_from((h >> shift) & 0xF).expect("word index")]; + format!( + "log line {i:08} trace={h:016x} span={span:016x} user={user:016x} {} {} {} {} latency_ms={} bytes={}", + w(0), + w(4), + w(8), + w(12), + h % 5_000, + user % 100_000 + ) +} + +async fn process_all(pipeline: &mut LogsPipeline, events: &[TelemetryEvent]) -> Vec> { + for event in events { + pipeline + .telemetry_tx + .send(event.clone()) + .await + .expect("send"); + pipeline.agent.sync_consume().await; + } + // HARNESS: aggregator-specific. + pipeline + .aggregator_handle + .get_batches() + .await + .expect("batches") +} + +async fn flush_once(pipeline: &LogsPipeline, batch: &[String]) { + // HARNESS: aggregator-specific. + pipeline + .aggregator_handle + .insert_batch(batch.to_vec()) + .expect("insert"); + // HARNESS: flusher-specific. + assert!(pipeline.flusher.flush(None).await.is_empty()); +} + +/// Telemetry event → processor → aggregator → batches. +fn bench_process_and_batch(c: &mut Criterion) { + let rt = runtime(); + let mut group = c.benchmark_group("logs_process_and_batch"); + for (count, size) in [(1_000_usize, 200_usize), (10_000, 200), (1_000, 4_096)] { + let events: Vec = (0..count) + .map(|i| { + serde_json::from_value(json!({ + "time": "2022-10-21T14:05:04.000Z", + "type": "function", + "record": format!("{i:08} {}", "x".repeat(size)) + })) + .expect("event") + }) + .collect(); + let (mut pipeline, _event_bus, _intake) = rt.block_on(async { + let intake = FakeLogsIntake::start_http().await; + // Managed Instance mode ships logs without a request_id immediately. + let (mut pipeline, event_bus) = build_pipeline(config(&intake, &[]), true); + let init: TelemetryEvent = serde_json::from_str(INIT_START).expect("init"); + pipeline.telemetry_tx.send(init).await.expect("send"); + pipeline.agent.sync_consume().await; + // HARNESS: aggregator-specific. + let _ = pipeline.aggregator_handle.get_batches().await; + (pipeline, event_bus, intake) + }); + + let batches = rt.block_on(process_all(&mut pipeline, &events)); + let total: usize = batches + .iter() + .map(|b| { + serde_json::from_slice::>(b) + .expect("batch") + .len() + }) + .sum(); + assert_eq!(total, count, "every event should reach a batch"); + + group.throughput(Throughput::Elements(count as u64)); + group.bench_with_input( + BenchmarkId::from_parameter(format!("{count}x{size}B")), + &events, + |b, events| { + b.iter(|| rt.block_on(process_all(&mut pipeline, events))); + }, + ); + } + group.finish(); +} + +/// Aggregator → batch → zstd → HTTP, to the primary and 0 or 2 additional endpoints. +fn bench_flush(c: &mut Criterion) { + let rt = runtime(); + let mut group = c.benchmark_group("logs_flush"); + let batch: Vec = (0..1_000) + .map(|i| { + json!({ + "message": { + "message": log_message(i), + "lambda": {"arn": ARN, "request_id": "459921b5-681c-4a96-beb0-81e0aa586026"}, + "timestamp": 1_666_361_104_000_i64, + "status": "info" + }, + "hostname": ARN, + "service": "bench", + "ddtags": "env:bench", + "ddsource": "lambda" + }) + .to_string() + }) + .collect(); + + for extras_count in [0_usize, 2] { + let (pipeline, _event_bus, primary, extras) = rt.block_on(async { + let primary = FakeLogsIntake::start_http().await; + let mut extras = Vec::new(); + for _ in 0..extras_count { + extras.push(FakeLogsIntake::start_https().await); + } + let (pipeline, event_bus) = build_pipeline(config(&primary, &extras), false); + (pipeline, event_bus, primary, extras) + }); + + rt.block_on(flush_once(&pipeline, &batch)); + for intake in std::iter::once(&primary).chain(&extras) { + let requests = intake.take_requests(); + assert_eq!(requests.len(), 1, "one request per endpoint"); + let logs = requests[0].json(); + assert_eq!(logs.as_array().expect("array").len(), batch.len()); + } + + group.throughput(Throughput::Elements(batch.len() as u64)); + group.bench_function( + BenchmarkId::from_parameter(format!("1000_logs_{extras_count}_extras")), + |b| { + b.iter(|| { + rt.block_on(async { + flush_once(&pipeline, &batch).await; + primary.take_requests(); + for extra in &extras { + extra.take_requests(); + } + }); + }); + }, + ); + } + group.finish(); +} + +criterion_group!(benches, bench_process_and_batch, bench_flush); +criterion_main!(benches); diff --git a/bottlecap/src/bin/bottlecap/main.rs b/bottlecap/src/bin/bottlecap/main.rs index 3078b25b6..4e63fe8e9 100644 --- a/bottlecap/src/bin/bottlecap/main.rs +++ b/bottlecap/src/bin/bottlecap/main.rs @@ -49,12 +49,10 @@ use bottlecap::{ cold_start::{ColdStartDurations, elapsed_ms}, }, logs::{ - agent::LogsAgent, - aggregator_service::{ - AggregatorHandle as LogsAggregatorHandle, AggregatorService as LogsAggregatorService, - }, + aggregator_service::AggregatorHandle as LogsAggregatorHandle, flusher::LogsFlusher, lambda::DurableContextUpdate, + pipeline::{LogsPipeline, build_logs_pipeline}, }, otlp::{agent::Agent as OtlpAgent, should_enable_otlp_agent}, proxy::{interceptor, should_start_proxy}, @@ -1160,18 +1158,19 @@ fn start_logs_agent( LogsAggregatorHandle, Sender, ) { - let (aggregator_service, aggregator_handle) = LogsAggregatorService::default(); - // Start service in background - tokio::spawn(async move { - aggregator_service.run().await; - }); - - let (mut agent, tx, durable_context_tx) = LogsAgent::new( - Arc::clone(tags_provider), - Arc::clone(config), + let LogsPipeline { + mut agent, + telemetry_tx, + durable_context_tx, + flusher, + aggregator_handle, + } = build_logs_pipeline( + config, + api_key_factory, + tags_provider, event_bus, - aggregator_handle.clone(), is_managed_instance_mode, + client, ); let cancel_token = agent.cancel_token(); // Start logs agent in background @@ -1182,14 +1181,8 @@ fn start_logs_agent( drop(agent); }); - let flusher = LogsFlusher::new( - api_key_factory, - aggregator_handle.clone(), - config.clone(), - client.clone(), - ); ( - tx, + telemetry_tx, flusher, cancel_token, aggregator_handle, diff --git a/bottlecap/src/logs/mod.rs b/bottlecap/src/logs/mod.rs index 3c40f2983..cebda00c5 100644 --- a/bottlecap/src/logs/mod.rs +++ b/bottlecap/src/logs/mod.rs @@ -4,4 +4,5 @@ pub mod aggregator_service; pub mod constants; pub mod flusher; pub mod lambda; +pub mod pipeline; pub mod processor; diff --git a/bottlecap/src/logs/pipeline.rs b/bottlecap/src/logs/pipeline.rs new file mode 100644 index 000000000..76029e781 --- /dev/null +++ b/bottlecap/src/logs/pipeline.rs @@ -0,0 +1,112 @@ +//! Wiring for the logs pipeline, shared by `main.rs` and the logs contract tests so both +//! exercise exactly the same aggregator, agent and flusher setup. + +use std::sync::Arc; + +use dogstatsd::api_key::ApiKeyFactory; +use reqwest::Client; +use tokio::sync::mpsc::Sender; + +use crate::config::Config; +use crate::event_bus::Event; +use crate::extension::telemetry::events::TelemetryEvent; +use crate::logs::agent::LogsAgent; +use crate::logs::aggregator_service::{AggregatorHandle, AggregatorService}; +use crate::logs::flusher::LogsFlusher; +use crate::logs::lambda::DurableContextUpdate; +use crate::tags::provider::Provider as TagProvider; + +/// The pieces of a logs pipeline. The aggregator task is already running; the agent is +/// returned unspawned so the caller decides how to drive it: `main.rs` spawns `spin()`, +/// tests call `sync_consume()`. +pub struct LogsPipeline { + pub agent: LogsAgent, + pub telemetry_tx: Sender, + pub durable_context_tx: Sender, + pub flusher: LogsFlusher, + pub aggregator_handle: AggregatorHandle, +} + +/// Creates the logs aggregator (and starts its background task), the logs agent and the +/// flusher. Must be called from within a tokio runtime. +#[must_use] +pub fn build_logs_pipeline( + config: &Arc, + api_key_factory: Arc, + tags_provider: &Arc, + event_bus: Sender, + is_managed_instance_mode: bool, + client: &Client, +) -> LogsPipeline { + let (aggregator_service, aggregator_handle) = AggregatorService::default(); + tokio::spawn(async move { + aggregator_service.run().await; + }); + + let (agent, telemetry_tx, durable_context_tx) = LogsAgent::new( + Arc::clone(tags_provider), + Arc::clone(config), + event_bus, + aggregator_handle.clone(), + is_managed_instance_mode, + ); + + let flusher = LogsFlusher::new( + api_key_factory, + aggregator_handle.clone(), + Arc::clone(config), + client.clone(), + ); + + LogsPipeline { + agent, + telemetry_tx, + durable_context_tx, + flusher, + aggregator_handle, + } +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use super::*; + use crate::LAMBDA_RUNTIME_SLUG; + use crate::event_bus::EventBus; + use std::collections::HashMap; + + #[tokio::test] + async fn build_logs_pipeline_routes_telemetry_to_the_aggregator() { + let config = Arc::new(Config::default()); + let tags_provider = Arc::new(TagProvider::new( + Arc::clone(&config), + LAMBDA_RUNTIME_SLUG.to_string(), + &HashMap::from([("function_arn".to_string(), "test-arn".to_string())]), + )); + let (_event_bus, event_bus_tx) = EventBus::run(); + let client = crate::http::get_client(&config); + + // Managed Instance mode ships logs without a request_id immediately, + // so a single function log is enough to reach the aggregator. + let mut pipeline = build_logs_pipeline( + &config, + Arc::new(ApiKeyFactory::new("test-key")), + &tags_provider, + event_bus_tx, + true, + &client, + ); + let event: TelemetryEvent = serde_json::from_str( + r#"{"time":"2024-04-24T12:34:56.789Z","type":"function","record":"hello"}"#, + ) + .unwrap(); + pipeline.telemetry_tx.send(event).await.unwrap(); + pipeline.agent.sync_consume().await; + + let batches = pipeline.aggregator_handle.get_batches().await.unwrap(); + assert_eq!(batches.len(), 1); + let logs: serde_json::Value = serde_json::from_slice(&batches[0]).unwrap(); + assert_eq!(logs[0]["message"]["message"], "hello"); + assert_eq!(logs[0]["hostname"], "test-arn"); + } +} diff --git a/bottlecap/tests/common/fake_logs_intake.rs b/bottlecap/tests/common/fake_logs_intake.rs new file mode 100644 index 000000000..416b17dd0 --- /dev/null +++ b/bottlecap/tests/common/fake_logs_intake.rs @@ -0,0 +1,379 @@ +// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +//! Fake logs intake for the logs contract tests and the logs benchmark. +//! +//! Accepts requests on any path over plain HTTP/1.1 or HTTPS (HTTP/1.1 only — clients must +//! not use HTTP/2 prior knowledge; set bottlecap's `http_protocol: "http1"`). HTTPS uses a +//! self-signed test certificate, so clients must skip certificate validation (bottlecap: +//! `skip_ssl_validation: true`). Every request is recorded (method, path, headers, body); +//! responses use scripted status codes, falling back to `202 Accepted`. Requests can also be +//! scripted to hang (never answered) so clients time out. + +#![allow(dead_code)] // each test binary uses a different subset + +use std::collections::{HashMap, VecDeque}; +use std::convert::Infallible; +use std::io::BufReader; +use std::sync::{Arc, Mutex}; + +use bytes::Bytes; +use http_body_util::{BodyExt, Full}; +use hyper::body::Incoming; +use hyper::server::conn::http1; +use hyper::service::service_fn; +use hyper::{Request, Response}; +use hyper_util::rt::TokioIo; +use tokio::net::TcpListener; +use tokio::task::JoinHandle; + +// Same self-signed leaf certificate as the TLS tests in `src/http.rs`. +const TEST_SERVER_CERT_PEM: &str = r" +-----BEGIN CERTIFICATE----- +MIIBvTCCAWSgAwIBAgIUOkYng/RZxQcYlMY5I+RAXzR7dSkwCgYIKoZIzj0EAwIw +HDEaMBgGA1UEAwwRQm90dGxlY2FwIFRlc3QgQ0EwIBcNMjYwNjI1MTgzNjMwWhgP +MjEyNjA2MDExODM2MzBaMBQxEjAQBgNVBAMMCWxvY2FsaG9zdDBZMBMGByqGSM49 +AgEGCCqGSM49AwEHA0IABOuKun0roY5MAEOgA3NNebQa9l56HVsNFwGJ4a5chM2T +s+vToAoyflPMZfxuS6PUv2WHrahmUH5WZKr0XaT/RIijgYkwgYYwCQYDVR0TBAIw +ADAOBgNVHQ8BAf8EBAMCB4AwEwYDVR0lBAwwCgYIKwYBBQUHAwEwFAYDVR0RBA0w +C4IJbG9jYWxob3N0MB0GA1UdDgQWBBTcqKduDXHEOeUXAL6fqwAYl50VVTAfBgNV +HSMEGDAWgBSbWR21hL9RSNCmnDP+HLSZFLjINjAKBggqhkjOPQQDAgNHADBEAiAT +nwOMjOwnQoeLvZgHhNqrzlVGNqnT4FPudJTgTKrAAAIgVMw8wqk70JXLm4pEdN8/ +oMUo5j7nV6S8Hd02b5hW5pM= +-----END CERTIFICATE----- +"; +const TEST_SERVER_KEY_PEM: &str = r" +-----BEGIN PRIVATE KEY----- +MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgkflx4lEcI8Fafj8E +EgBejgnIKCAkaQXo+p0h3a6IsQOhRANCAATrirp9K6GOTABDoANzTXm0GvZeeh1b +DRcBieGuXITNk7Pr06AKMn5TzGX8bkuj1L9lh62oZlB+VmSq9F2k/0SI +-----END PRIVATE KEY----- +"; + +/// One request received by the fake intake. +#[derive(Clone, Debug)] +pub struct RecordedRequest { + pub method: String, + pub path: String, + /// Header names are lowercase; if a header repeats, the last value wins. + pub headers: HashMap, + pub body: Vec, +} + +impl RecordedRequest { + #[must_use] + pub fn header(&self, name: &str) -> Option<&str> { + self.headers + .get(&name.to_ascii_lowercase()) + .map(String::as_str) + } + + /// Body parsed as JSON, decompressing first when `Content-Encoding: zstd` is set. + #[must_use] + pub fn json(&self) -> serde_json::Value { + let raw = if self.header("content-encoding") == Some("zstd") { + zstd::decode_all(self.body.as_slice()).expect("fake_logs_intake: invalid zstd body") + } else { + self.body.clone() + }; + serde_json::from_slice(&raw).expect("fake_logs_intake: body is not JSON") + } +} + +#[derive(Default)] +struct State { + inner: Mutex, +} + +#[derive(Default)] +struct Inner { + requests: Vec, + replies: VecDeque, +} + +/// How the intake answers one request. +enum Reply { + Status(u16), + /// Record the request, then never answer. + Hang, +} + +pub struct FakeLogsIntake { + base_url: String, + port: u16, + state: Arc, + task: JoinHandle<()>, +} + +impl FakeLogsIntake { + /// Plain HTTP on `127.0.0.1`, OS-assigned port. + pub async fn start_http() -> Self { + Self::start(None).await + } + + /// HTTPS on `127.0.0.1`, OS-assigned port, self-signed certificate. + pub async fn start_https() -> Self { + Self::start(Some(tls_acceptor())).await + } + + async fn start(acceptor: Option) -> Self { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("fake_logs_intake: bind"); + let port = listener + .local_addr() + .expect("fake_logs_intake: local_addr") + .port(); + let scheme = if acceptor.is_some() { "https" } else { "http" }; + let state = Arc::new(State::default()); + let server_state = Arc::clone(&state); + + let task = tokio::spawn(async move { + loop { + let Ok((stream, _)) = listener.accept().await else { + return; + }; + let state = Arc::clone(&server_state); + let acceptor = acceptor.clone(); + tokio::spawn(async move { + let service = service_fn(move |req| handle(Arc::clone(&state), req)); + match acceptor { + Some(acceptor) => { + let Ok(tls) = acceptor.accept(stream).await else { + return; + }; + let _ = http1::Builder::new() + .serve_connection(TokioIo::new(tls), service) + .await; + } + None => { + let _ = http1::Builder::new() + .serve_connection(TokioIo::new(stream), service) + .await; + } + } + }); + } + }); + + Self { + base_url: format!("{scheme}://127.0.0.1:{port}"), + port, + state, + task, + } + } + + #[must_use] + pub fn base_url(&self) -> String { + self.base_url.clone() + } + + #[must_use] + pub fn port(&self) -> u16 { + self.port + } + + /// Replies to the next requests with `statuses`, in order; later requests get 202. + pub fn script_statuses(&self, statuses: &[u16]) { + self.state + .inner + .lock() + .expect("fake_logs_intake: lock") + .replies + .extend(statuses.iter().copied().map(Reply::Status)); + } + + /// Records the next `count` requests but never answers them, so clients time out. + pub fn script_hangs(&self, count: usize) { + self.state + .inner + .lock() + .expect("fake_logs_intake: lock") + .replies + .extend(std::iter::repeat_with(|| Reply::Hang).take(count)); + } + + #[must_use] + pub fn requests(&self) -> Vec { + self.state + .inner + .lock() + .expect("fake_logs_intake: lock") + .requests + .clone() + } + + /// Returns the recorded requests and clears the log. + pub fn take_requests(&self) -> Vec { + std::mem::take( + &mut self + .state + .inner + .lock() + .expect("fake_logs_intake: lock") + .requests, + ) + } +} + +impl Drop for FakeLogsIntake { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn handle( + state: Arc, + req: Request, +) -> Result>, Infallible> { + let method = req.method().to_string(); + let path = req.uri().path().to_string(); + let headers = req + .headers() + .iter() + .map(|(k, v)| { + ( + k.as_str().to_string(), + v.to_str().unwrap_or_default().to_string(), + ) + }) + .collect(); + let body = req + .into_body() + .collect() + .await + .map(|b| b.to_bytes().to_vec()) + .unwrap_or_default(); + let reply = { + let mut inner = state.inner.lock().expect("fake_logs_intake: lock"); + let reply = inner.replies.pop_front().unwrap_or(Reply::Status(202)); + inner.requests.push(RecordedRequest { + method, + path, + headers, + body, + }); + reply + }; + let status = match reply { + Reply::Status(status) => status, + Reply::Hang => std::future::pending().await, + }; + Ok(Response::builder() + .status(status) + .body(Full::new(Bytes::new())) + .expect("fake_logs_intake: response")) +} + +fn tls_acceptor() -> tokio_rustls::TlsAcceptor { + // Mirrors `ensure_crypto_provider` in `src/http.rs`; idempotent. + #[cfg(feature = "fips")] + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + #[cfg(not(feature = "fips"))] + let _ = rustls::crypto::ring::default_provider().install_default(); + + let certs = rustls_pemfile::certs(&mut BufReader::new(TEST_SERVER_CERT_PEM.as_bytes())) + .collect::, _>>() + .expect("fake_logs_intake: parse cert"); + let key = rustls_pemfile::private_key(&mut BufReader::new(TEST_SERVER_KEY_PEM.as_bytes())) + .expect("fake_logs_intake: read key") + .expect("fake_logs_intake: key present"); + let config = rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(certs, key) + .expect("fake_logs_intake: tls config"); + tokio_rustls::TlsAcceptor::from(Arc::new(config)) +} + +#[cfg(test)] +#[allow(clippy::disallowed_methods)] // plain reqwest client against a local server +mod self_tests { + use super::*; + + #[tokio::test] + async fn records_requests_and_replays_scripted_statuses() { + let intake = FakeLogsIntake::start_http().await; + intake.script_statuses(&[500]); + let client = reqwest::Client::builder().build().expect("client"); + + let first = client + .post(format!("{}/api/v2/logs", intake.base_url())) + .header("DD-API-KEY", "k") + .body("[1]") + .send() + .await + .expect("send"); + let second = client + .post(format!("{}/other", intake.base_url())) + .body("[2]") + .send() + .await + .expect("send"); + + assert_eq!(first.status().as_u16(), 500); + assert_eq!(second.status().as_u16(), 202); + let requests = intake.take_requests(); + assert_eq!(requests.len(), 2); + assert_eq!(requests[0].path, "/api/v2/logs"); + assert_eq!(requests[0].header("DD-API-KEY"), Some("k")); + assert_eq!(requests[0].json(), serde_json::json!([1])); + assert_eq!(requests[1].path, "/other"); + assert!(intake.requests().is_empty(), "take_requests clears the log"); + } + + #[tokio::test] + async fn decodes_zstd_bodies() { + let intake = FakeLogsIntake::start_http().await; + let client = reqwest::Client::builder().build().expect("client"); + let body = zstd::encode_all(&b"[{\"a\":1}]"[..], 3).expect("zstd"); + client + .post(format!("{}/api/v2/logs", intake.base_url())) + .header("Content-Encoding", "zstd") + .body(body) + .send() + .await + .expect("send"); + assert_eq!(intake.requests()[0].json(), serde_json::json!([{"a": 1}])); + } + + #[tokio::test] + async fn serves_https_with_the_test_certificate() { + let intake = FakeLogsIntake::start_https().await; + assert!(intake.base_url().starts_with("https://127.0.0.1:")); + let client = reqwest::Client::builder() + .danger_accept_invalid_certs(true) + .build() + .expect("client"); + let resp = client + .post(format!("{}/api/v2/logs", intake.base_url())) + .body("[]") + .send() + .await + .expect("send"); + assert_eq!(resp.status().as_u16(), 202); + assert_eq!(intake.requests().len(), 1); + } + + #[tokio::test] + async fn hung_request_times_out_and_is_recorded() { + let intake = FakeLogsIntake::start_http().await; + intake.script_hangs(1); + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_millis(200)) + .build() + .expect("client"); + let result = client + .post(format!("{}/api/v2/logs", intake.base_url())) + .body("[]") + .send() + .await; + assert!( + result.expect_err("a hung request must fail").is_timeout(), + "the client times out" + ); + let requests = intake.requests(); + assert_eq!(requests.len(), 1, "the hung request is still recorded"); + assert_eq!(requests[0].method, "POST"); + } +} diff --git a/bottlecap/tests/logs_contract_test.rs b/bottlecap/tests/logs_contract_test.rs new file mode 100644 index 000000000..dd9761bfa --- /dev/null +++ b/bottlecap/tests/logs_contract_test.rs @@ -0,0 +1,1169 @@ +// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +//! Contract tests for the logs pipeline. +//! +//! They pin what bottlecap sends to the logs intake today: the JSON of every log, request +//! URLs and headers, retry/redrive/drop behavior, and how many `warn!`/`error!` lines the +//! pipeline emits. Migrating to `datadog-logs-agent` must keep every expectation in this +//! file unchanged. Only HARNESS code (see the doc on `impl Harness`) may change with the +//! implementation. + +#[path = "common/fake_logs_intake.rs"] +mod fake_logs_intake; + +use std::collections::{BTreeSet, HashMap}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use bottlecap::LAMBDA_RUNTIME_SLUG; +use bottlecap::config::Config; +use bottlecap::config::logs_additional_endpoints::LogsAdditionalEndpoint; +use bottlecap::config::processing_rule::{Kind, ProcessingRule}; +use bottlecap::event_bus::EventBus; +use bottlecap::extension::telemetry::events::TelemetryEvent; +use bottlecap::logs::pipeline::{LogsPipeline, build_logs_pipeline}; +use bottlecap::tags::provider::Provider; +use dogstatsd::api_key::ApiKeyFactory; +use fake_logs_intake::{FakeLogsIntake, RecordedRequest}; +use serde_json::{Value, json}; +use tracing::Level; +use tracing_subscriber::layer::SubscriberExt as _; + +const ARN: &str = "arn:aws:lambda:us-east-1:123456789012:function:contract-test"; +const REQUEST_ID: &str = "459921b5-681c-4a96-beb0-81e0aa586026"; +const API_KEY: &str = "primary-api-key"; +const SERVICE: &str = "contract-service"; // configured as "Contract-Service"; bottlecap lowercases it + +const INIT_START: &str = r#"{"time":"2022-10-21T14:05:03.000Z","type":"platform.initStart","record":{"initializationType":"on-demand","phase":"init"}}"#; +const START: &str = r#"{"time":"2022-10-21T14:05:03.165Z","type":"platform.start","record":{"requestId":"459921b5-681c-4a96-beb0-81e0aa586026","version":"$LATEST"}}"#; +const RUNTIME_DONE: &str = r#"{"time":"2022-10-21T14:05:05.764Z","type":"platform.runtimeDone","record":{"requestId":"459921b5-681c-4a96-beb0-81e0aa586026","status":"success","metrics":{"durationMs":2599.0,"producedBytes":8}}}"#; +const REPORT: &str = r#"{"time":"2022-10-21T14:05:05.766Z","type":"platform.report","record":{"requestId":"459921b5-681c-4a96-beb0-81e0aa586026","metrics":{"durationMs":2599.4,"billedDurationMs":2600,"memorySizeMB":128,"maxMemoryUsedMB":94,"initDurationMs":549.04},"status":"success"}}"#; +const MANAGED_INSTANCE_REPORT: &str = r#"{"time":"2022-10-21T14:05:05.766Z","type":"platform.report","record":{"requestId":"459921b5-681c-4a96-beb0-81e0aa586026","status":"success","metrics":{"durationMs":1.148}}}"#; + +const TS_INIT_START: i64 = 1_666_361_103_000; +const TS_START: i64 = 1_666_361_103_165; +const TS_FUNCTION: i64 = 1_666_361_104_000; +const TS_RUNTIME_DONE: i64 = 1_666_361_105_764; +const TS_REPORT: i64 = 1_666_361_105_766; + +const START_LINE: &str = "START RequestId: 459921b5-681c-4a96-beb0-81e0aa586026 Version: $LATEST"; +const INIT_START_LINE: &str = "INIT_START Runtime Version: ? Runtime Version ARN: ?"; + +fn function_log(record: impl Into) -> String { + json!({"time": "2022-10-21T14:05:04.000Z", "type": "function", "record": record.into()}) + .to_string() +} + +fn extension_log(record: &str) -> String { + json!({"time": "2022-10-21T14:05:04.000Z", "type": "extension", "record": record}).to_string() +} + +/// Config shared by all tests: logs go to `intake` uncompressed over HTTP/1. +fn base_config(intake: &FakeLogsIntake) -> Config { + Config { + http_protocol: Some("http1".to_string()), + skip_ssl_validation: true, + logs_config_logs_dd_url: intake.base_url(), + logs_config_use_compression: false, + logs_config_compression_level: 3, + service: Some("Contract-Service".to_string()), + flush_timeout: 5, + ..Config::default() + } +} + +/// The logs from the one and only request `intake` received. +#[track_caller] +fn single_request_logs(intake: &FakeLogsIntake) -> Value { + let requests = intake.requests(); + assert_eq!(requests.len(), 1, "expected exactly one request"); + requests[0].json() +} + +/// One line per log entry (message length and the first 80 characters), for readable +/// failure messages when payloads are megabytes long. +fn summarize(logs: &Value) -> String { + logs.as_array() + .map(|entries| { + entries + .iter() + .map(|entry| { + let message = entry["message"]["message"].as_str().unwrap_or_default(); + let preview: String = message.chars().take(80).collect(); + format!("{} bytes: {preview}", message.len()) + }) + .collect::>() + .join("\n") + }) + .unwrap_or_else(|| "".to_string()) +} + +/// Header names on a request, minus transport headers whose values depend on the +/// connection (`host`, `content-length`). +fn header_names(request: &RecordedRequest) -> BTreeSet { + request + .headers + .keys() + .filter(|name| !matches!(name.as_str(), "host" | "content-length")) + .cloned() + .collect() +} + +/// `entry` with its message cut to 80 characters, for readable failure output. +fn abbreviate(entry: &Value) -> Value { + let mut entry = entry.clone(); + if let Some(message) = entry["message"]["message"].as_str() { + let preview: String = message.chars().take(80).collect(); + let abbreviated = format!("{preview}... ({} bytes)", message.len()); + entry["message"]["message"] = Value::String(abbreviated); + } + entry +} + +/// Asserts that `actual` holds every log in `expected` exactly once, in any order (logs are +/// matched by message, so messages must be unique). +#[track_caller] +fn assert_each_log_delivered_once(mut actual: Vec, mut expected: Vec) { + let message = |entry: &Value| entry["message"]["message"].as_str().map(str::to_string); + actual.sort_by_cached_key(message); + expected.sort_by_cached_key(message); + for (a, e) in actual.iter().zip(&expected) { + assert!( + a == e, + "logs differ\n--- actual ---\n{:#}\n--- expected ---\n{:#}", + abbreviate(a), + abbreviate(e) + ); + } + assert_eq!(actual.len(), expected.len(), "number of logs delivered"); +} + +fn additional_endpoint(intake: &FakeLogsIntake, api_key: &str) -> LogsAdditionalEndpoint { + LogsAdditionalEndpoint { + api_key: api_key.to_string(), + host: "127.0.0.1".to_string(), + port: u32::from(intake.port()), + is_reliable: true, + } +} + +struct Harness { + pipeline: LogsPipeline, + tags: String, + _event_bus: EventBus, // keeps the event bus receiver alive so sends succeed +} + +/// Everything in this `impl` except `send_invocation_start`, `log`, `log_with` and +/// `invocation_start_logs` is HARNESS: it wires the test to the current pipeline +/// implementation and may change when the implementation changes. Expected values elsewhere +/// in this file must not change. `tags` must come from the same tag provider the pipeline +/// uses. +impl Harness { + fn new(config: Config) -> Self { + Self::build(config, false, Arc::new(ApiKeyFactory::new(API_KEY))) + } + + fn managed_instance(config: Config) -> Self { + Self::build(config, true, Arc::new(ApiKeyFactory::new(API_KEY))) + } + + /// An on-demand pipeline whose API key resolver never returns a key. + fn with_unresolvable_api_key(config: Config) -> Self { + let resolver: dogstatsd::api_key::ApiKeyResolverFn = Arc::new( + || -> std::pin::Pin> + Send>> { + Box::pin(async { None }) + }, + ); + let factory = Arc::new(ApiKeyFactory::new_from_resolver(resolver, None)); + Self::build(config, false, factory) + } + + fn build( + config: Config, + is_managed_instance_mode: bool, + api_key_factory: Arc, + ) -> Self { + let config = Arc::new(config); + let tags_provider = Arc::new(Provider::new( + Arc::clone(&config), + LAMBDA_RUNTIME_SLUG.to_string(), + &HashMap::from([("function_arn".to_string(), ARN.to_string())]), + )); + let tags = tags_provider.get_tags_string(); + let (event_bus, event_bus_tx) = EventBus::run(); + let client = bottlecap::http::get_client(&config); + let pipeline = build_logs_pipeline( + &config, + api_key_factory, + &tags_provider, + event_bus_tx, + is_managed_instance_mode, + &client, + ); + Self { + pipeline, + tags, + _event_bus: event_bus, + } + } + + async fn send(&mut self, events: &[&str]) { + for raw in events { + let event: TelemetryEvent = + serde_json::from_str(raw).expect("test event must be valid telemetry"); + self.pipeline + .telemetry_tx + .send(event) + .await + .expect("send telemetry event"); + self.pipeline.agent.sync_consume().await; + } + } + + async fn send_invocation_start(&mut self) { + self.send(&[INIT_START, START]).await; + } + + /// One normal flush, returning requests that should be redriven. + async fn flush(&self) -> Vec { + self.pipeline.flusher.flush(None).await + } + + /// Mirrors `FlushingService::await_handles` — one redrive pass per failed + /// request, results discarded. (Runs them one after another; the real service runs them + /// concurrently, which makes no difference with one failed request. Blocking flushes + /// discard failures entirely and never redrive.) + async fn redrive(&self, failed: Vec) { + for request in failed { + let _ = self.pipeline.flusher.flush(Some(request)).await; + } + } + + /// Drains the aggregator and returns how many batches were still waiting. This removes + /// them, so only call it when the test no longer needs those logs. + async fn drain_pending_batches(&self) -> usize { + self.pipeline + .aggregator_handle + .get_batches() + .await + .expect("aggregator reachable") + .len() + } + + /// Expected JSON for one log, using this function's ARN, service and tags. + fn log(&self, message: &str, request_id: Option<&str>, timestamp: i64, status: &str) -> Value { + self.log_with( + message, + json!({"arn": ARN, "request_id": request_id}), + timestamp, + status, + &self.tags, + ) + } + + fn log_with( + &self, + message: &str, + lambda: Value, + timestamp: i64, + status: &str, + tags: &str, + ) -> Value { + json!({ + "message": { + "message": message, + "lambda": lambda, + "timestamp": timestamp, + "status": status, + }, + "hostname": ARN, + "service": SERVICE, + "ddtags": tags, + "ddsource": "lambda", + }) + } + + /// The two logs every `send_invocation_start()` produces, in shipping order. INIT_START + /// arrives before any request_id exists, so it is held as an orphan and shipped right + /// after START with START's request_id. + fn invocation_start_logs(&self) -> Vec { + vec![ + self.log(START_LINE, Some(REQUEST_ID), TS_START, "info"), + self.log(INIT_START_LINE, Some(REQUEST_ID), TS_INIT_START, "info"), + ] + } +} + +/// Counts `warn!`/`error!` events from the logs pipeline (bottlecap today, the shared crate +/// after the migration). Tests must run on the current-thread runtime so tasks spawned by +/// the pipeline run on the test thread and are counted. +#[derive(Clone, Default)] +struct LevelCounts(Arc>>); + +const PIPELINE_TARGETS: [&str; 2] = ["bottlecap", "datadog_logs_agent"]; + +impl tracing_subscriber::Layer for LevelCounts { + fn on_event( + &self, + event: &tracing::Event<'_>, + _ctx: tracing_subscriber::layer::Context<'_, S>, + ) { + let meta = event.metadata(); + if PIPELINE_TARGETS + .iter() + .any(|t| meta.target().starts_with(t)) + { + *self + .0 + .lock() + .expect("level counts lock") + .entry(*meta.level()) + .or_default() += 1; + } + } +} + +impl LevelCounts { + fn install() -> (Self, tracing::subscriber::DefaultGuard) { + let counts = Self::default(); + let guard = + tracing::subscriber::set_default(tracing_subscriber::registry().with(counts.clone())); + (counts, guard) + } + + fn get(&self, level: Level) -> usize { + self.0 + .lock() + .expect("level counts lock") + .get(&level) + .copied() + .unwrap_or(0) + } + + #[track_caller] + fn assert(&self, warns: usize, errors: usize) { + assert_eq!( + (self.get(Level::WARN), self.get(Level::ERROR)), + (warns, errors), + "(warn!, error!) lines emitted by the logs pipeline" + ); + } +} + +// --------------------------------------------------------------------------------------- +// Log content +// --------------------------------------------------------------------------------------- + +#[tokio::test(flavor = "current_thread")] +async fn on_demand_invocation_lines() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::new(base_config(&intake)); + + h.send(&[ + INIT_START, + START, + &function_log("hello"), + RUNTIME_DONE, + REPORT, + ]) + .await; + assert!(h.flush().await.is_empty()); + + let mut expected = h.invocation_start_logs(); + expected.extend([ + h.log("hello", Some(REQUEST_ID), TS_FUNCTION, "info"), + h.log( + &format!("END RequestId: {REQUEST_ID}"), + Some(REQUEST_ID), + TS_RUNTIME_DONE, + "info", + ), + h.log( + &format!( + "REPORT RequestId: {REQUEST_ID} Duration: 2599.40 ms Runtime Duration: 2599.00 ms \ + Post Runtime Duration: 0.40 ms Billed Duration: 2600 ms Memory Size: 128 MB \ + Max Memory Used: 94 MB Init Duration: 549.04 ms" + ), + Some(REQUEST_ID), + TS_REPORT, + "info", + ), + ]); + assert_eq!(single_request_logs(&intake), Value::Array(expected)); + counts.assert(0, 0); +} + +#[tokio::test(flavor = "current_thread")] +async fn json_function_log_uses_level_and_extracts_ddtags() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::new(base_config(&intake)); + + h.send_invocation_start().await; + // ddtags is last so removing it cannot reorder the remaining keys. + h.send(&[&function_log(json!({ + "level": "WARN", + "message": "disk almost full", + "ddtags": "team:logs" + }))]) + .await; + assert!(h.flush().await.is_empty()); + + let mut expected = h.invocation_start_logs(); + expected.push(h.log_with( + r#"{"level":"WARN","message":"disk almost full"}"#, + json!({"arn": ARN, "request_id": REQUEST_ID}), + TS_FUNCTION, + "warn", + &format!("{},team:logs", h.tags), + )); + assert_eq!(single_request_logs(&intake), Value::Array(expected)); + counts.assert(0, 0); +} + +#[tokio::test(flavor = "current_thread")] +async fn non_object_json_and_extension_logs_are_shipped_as_text() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::new(base_config(&intake)); + + h.send_invocation_start().await; + h.send(&[ + &function_log("[1,2,3]"), + &extension_log("extension says hi"), + ]) + .await; + assert!(h.flush().await.is_empty()); + + let mut expected = h.invocation_start_logs(); + expected.extend([ + h.log("[1,2,3]", Some(REQUEST_ID), TS_FUNCTION, "info"), + h.log("extension says hi", Some(REQUEST_ID), TS_FUNCTION, "info"), + ]); + assert_eq!(single_request_logs(&intake), Value::Array(expected)); + counts.assert(0, 0); +} + +#[tokio::test(flavor = "current_thread")] +async fn function_log_before_start_is_attached_to_the_next_request() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::new(base_config(&intake)); + + h.send(&[INIT_START, &function_log("before start"), START]) + .await; + assert!(h.flush().await.is_empty()); + + let mut expected = h.invocation_start_logs(); + expected.push(h.log("before start", Some(REQUEST_ID), TS_FUNCTION, "info")); + assert_eq!(single_request_logs(&intake), Value::Array(expected)); + counts.assert(0, 0); +} + +#[tokio::test(flavor = "current_thread")] +async fn managed_instance_ships_logs_without_request_id_immediately() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::managed_instance(base_config(&intake)); + + h.send(&[ + INIT_START, + &function_log("sandbox log"), + MANAGED_INSTANCE_REPORT, + ]) + .await; + assert!(h.flush().await.is_empty()); + + let expected = json!([ + h.log(INIT_START_LINE, None, TS_INIT_START, "info"), + h.log("sandbox log", None, TS_FUNCTION, "info"), + h.log( + &format!("REPORT RequestId: {REQUEST_ID} Runtime Duration: 1.15 ms"), + Some(REQUEST_ID), + TS_REPORT, + "info" + ), + ]); + assert_eq!(single_request_logs(&intake), expected); + counts.assert(0, 0); +} + +#[tokio::test(flavor = "current_thread")] +async fn processing_rules_exclude_and_mask() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::new(Config { + logs_config_processing_rules: Some(vec![ + ProcessingRule { + kind: Kind::ExcludeAtMatch, + name: "drop-healthchecks".to_string(), + pattern: "healthcheck".to_string(), + replace_placeholder: None, + }, + ProcessingRule { + kind: Kind::MaskSequences, + name: "mask-cards".to_string(), + pattern: r"\d{4}-\d{4}".to_string(), + replace_placeholder: Some("[masked]".to_string()), + }, + ]), + ..base_config(&intake) + }); + + h.send_invocation_start().await; + h.send(&[ + &function_log("GET /healthcheck"), + &function_log("card 1234-5678 used"), + ]) + .await; + assert!(h.flush().await.is_empty()); + + let mut expected = h.invocation_start_logs(); + expected.push(h.log("card [masked] used", Some(REQUEST_ID), TS_FUNCTION, "info")); + assert_eq!(single_request_logs(&intake), Value::Array(expected)); + counts.assert(0, 0); +} + +#[tokio::test(flavor = "current_thread")] +async fn oom_line_is_shipped_like_any_other_log() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::new(base_config(&intake)); + + h.send_invocation_start().await; + h.send(&[&function_log("FATAL ERROR: JavaScript heap out of memory")]) + .await; + assert!(h.flush().await.is_empty()); + + let mut expected = h.invocation_start_logs(); + expected.push(h.log( + "FATAL ERROR: JavaScript heap out of memory", + Some(REQUEST_ID), + TS_FUNCTION, + "info", + )); + assert_eq!(single_request_logs(&intake), Value::Array(expected)); + counts.assert(0, 0); +} + +#[tokio::test(flavor = "current_thread")] +async fn durable_sdk_log_carries_execution_context() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::new(base_config(&intake)); + + h.send_invocation_start().await; + h.send(&[&function_log(json!({ + "executionArn": "arn:aws:lambda:us-east-1:123456789012:function:f/durable-execution/my-exec/abc-123", + "message": "step done" + }))]) + .await; + assert!(h.flush().await.is_empty()); + + let mut expected = h.invocation_start_logs(); + expected.push(h.log_with( + r#"{"executionArn":"arn:aws:lambda:us-east-1:123456789012:function:f/durable-execution/my-exec/abc-123","message":"step done"}"#, + json!({ + "arn": ARN, + "request_id": REQUEST_ID, + "durable_function.execution_id": "abc-123", + "durable_function.execution_name": "my-exec" + }), + TS_FUNCTION, + "info", + &h.tags, + )); + assert_eq!(single_request_logs(&intake), Value::Array(expected)); + counts.assert(0, 0); +} + +// --------------------------------------------------------------------------------------- +// Endpoints and request shape +// --------------------------------------------------------------------------------------- + +#[tokio::test(flavor = "current_thread")] +async fn datadog_intake_request_shape_uncompressed() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::new(base_config(&intake)); + + h.send_invocation_start().await; + assert!(h.flush().await.is_empty()); + + let requests = intake.requests(); + assert_eq!(requests.len(), 1); + let r = &requests[0]; + assert_eq!(r.method, "POST"); + assert_eq!(r.path, "/api/v2/logs"); + assert_eq!( + header_names(r), + BTreeSet::from([ + "accept".to_string(), + "content-type".to_string(), + "dd-api-key".to_string(), + "dd-protocol".to_string() + ]) + ); + assert_eq!(r.header("DD-API-KEY"), Some(API_KEY)); + assert_eq!(r.header("DD-PROTOCOL"), Some("agent-json")); + assert_eq!(r.header("Content-Type"), Some("application/json")); + assert_eq!(r.header("Content-Encoding"), None); + assert_eq!(r.json(), Value::Array(h.invocation_start_logs())); + counts.assert(0, 0); +} + +#[tokio::test(flavor = "current_thread")] +async fn datadog_intake_request_shape_compressed() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::new(Config { + logs_config_use_compression: true, + ..base_config(&intake) + }); + + h.send_invocation_start().await; + assert!(h.flush().await.is_empty()); + + let requests = intake.requests(); + assert_eq!(requests.len(), 1); + let r = &requests[0]; + assert_eq!(r.method, "POST"); + assert_eq!(r.path, "/api/v2/logs"); + assert_eq!( + header_names(r), + BTreeSet::from([ + "accept".to_string(), + "content-encoding".to_string(), + "content-type".to_string(), + "dd-api-key".to_string(), + "dd-protocol".to_string() + ]) + ); + assert_eq!(r.header("DD-API-KEY"), Some(API_KEY)); + assert_eq!(r.header("DD-PROTOCOL"), Some("agent-json")); + assert_eq!(r.header("Content-Type"), Some("application/json")); + assert_eq!(r.header("Content-Encoding"), Some("zstd")); + assert_eq!(r.json(), Value::Array(h.invocation_start_logs())); + counts.assert(0, 0); +} + +#[tokio::test(flavor = "current_thread")] +async fn observability_pipelines_worker_request_shape() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::new(Config { + observability_pipelines_worker_logs_enabled: true, + observability_pipelines_worker_logs_url: format!("{}/opw/path", intake.base_url()), + logs_config_use_compression: true, // OPW must still be uncompressed + ..base_config(&intake) + }); + + h.send_invocation_start().await; + assert!(h.flush().await.is_empty()); + + let requests = intake.requests(); + assert_eq!(requests.len(), 1); + let r = &requests[0]; + assert_eq!(r.method, "POST"); + assert_eq!(r.path, "/opw/path"); + assert_eq!( + header_names(r), + BTreeSet::from([ + "accept".to_string(), + "content-type".to_string(), + "dd-api-key".to_string() + ]) + ); + assert_eq!(r.header("DD-API-KEY"), Some(API_KEY)); + assert_eq!(r.header("DD-PROTOCOL"), None); + assert_eq!(r.header("Content-Type"), Some("application/json")); + assert_eq!(r.header("Content-Encoding"), None); + assert_eq!(r.json(), Value::Array(h.invocation_start_logs())); + counts.assert(0, 0); +} + +#[tokio::test(flavor = "current_thread")] +async fn additional_endpoints_receive_the_same_logs() { + let (counts, _guard) = LevelCounts::install(); + let primary = FakeLogsIntake::start_http().await; + let extra_1 = FakeLogsIntake::start_https().await; + let extra_2 = FakeLogsIntake::start_https().await; + let mut h = Harness::new(Config { + logs_config_use_compression: true, + logs_config_additional_endpoints: vec![ + additional_endpoint(&extra_1, "extra-key-1"), + additional_endpoint(&extra_2, "extra-key-2"), + ], + ..base_config(&primary) + }); + + h.send_invocation_start().await; + assert!(h.flush().await.is_empty()); + + let expected = Value::Array(h.invocation_start_logs()); + assert_eq!(single_request_logs(&primary), expected); + for (extra, key) in [(&extra_1, "extra-key-1"), (&extra_2, "extra-key-2")] { + let requests = extra.requests(); + assert_eq!(requests.len(), 1); + let r = &requests[0]; + assert_eq!(r.method, "POST"); + assert_eq!(r.path, "/api/v2/logs"); + assert_eq!( + header_names(r), + BTreeSet::from([ + "accept".to_string(), + "content-encoding".to_string(), + "content-type".to_string(), + "dd-api-key".to_string(), + "dd-protocol".to_string() + ]) + ); + assert_eq!(r.header("DD-API-KEY"), Some(key)); + assert_eq!(r.header("DD-PROTOCOL"), Some("agent-json")); + assert_eq!(r.header("Content-Type"), Some("application/json")); + assert_eq!(r.header("Content-Encoding"), Some("zstd")); + assert_eq!(r.json(), expected); + } + counts.assert(0, 0); +} + +/// Pins a known bug on purpose (a later change fixes it): with OPW enabled, additional +/// endpoints are posted to `https://host:port` with no path, without DD-PROTOCOL and +/// uncompressed. +#[tokio::test(flavor = "current_thread")] +async fn observability_pipelines_worker_with_additional_endpoint() { + let (counts, _guard) = LevelCounts::install(); + let primary = FakeLogsIntake::start_http().await; + let extra = FakeLogsIntake::start_https().await; + let mut h = Harness::new(Config { + observability_pipelines_worker_logs_enabled: true, + observability_pipelines_worker_logs_url: format!("{}/opw", primary.base_url()), + logs_config_use_compression: true, + logs_config_additional_endpoints: vec![additional_endpoint(&extra, "extra-key-1")], + ..base_config(&primary) + }); + + h.send_invocation_start().await; + assert!(h.flush().await.is_empty()); + + let primary_requests = primary.requests(); + assert_eq!(primary_requests.len(), 1); + assert_eq!(primary_requests[0].path, "/opw"); + assert_eq!( + primary_requests[0].json(), + Value::Array(h.invocation_start_logs()) + ); + let requests = extra.requests(); + assert_eq!(requests.len(), 1); + let r = &requests[0]; + assert_eq!(r.path, "/"); + assert_eq!(r.header("DD-API-KEY"), Some("extra-key-1")); + assert_eq!(r.header("DD-PROTOCOL"), None); + assert_eq!(r.header("Content-Type"), Some("application/json")); + assert_eq!(r.header("Content-Encoding"), None); + assert_eq!(r.json(), Value::Array(h.invocation_start_logs())); + counts.assert(0, 0); +} + +/// The configured logs URL keeps its path; `/api/v2/logs` is appended to it. +#[tokio::test(flavor = "current_thread")] +async fn logs_dd_url_with_a_path_prefix_keeps_it() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::new(Config { + logs_config_logs_dd_url: format!("{}/prefix", intake.base_url()), + ..base_config(&intake) + }); + + h.send_invocation_start().await; + assert!(h.flush().await.is_empty()); + + let requests = intake.requests(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].path, "/prefix/api/v2/logs"); + assert_eq!(requests[0].json(), Value::Array(h.invocation_start_logs())); + counts.assert(0, 0); +} + +// --------------------------------------------------------------------------------------- +// Retries, redrive and drops +// --------------------------------------------------------------------------------------- + +#[tokio::test(flavor = "current_thread")] +async fn status_403_is_dropped_without_retry() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + intake.script_statuses(&[403]); + let mut h = Harness::new(base_config(&intake)); + + h.send_invocation_start().await; + let failed = h.flush().await; + + assert!(failed.is_empty(), "403 is never redriven"); + assert_eq!(intake.requests().len(), 1); + counts.assert(0, 1); +} + +async fn assert_retried_three_times_then_redriven_once(status: u16) { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + intake.script_statuses(&[status; 6]); + let mut h = Harness::new(base_config(&intake)); + + h.send_invocation_start().await; + let failed = h.flush().await; + assert_eq!(failed.len(), 1, "one request handed back for redrive"); + assert_eq!( + intake.requests().len(), + 3, + "three attempts in the first flush" + ); + + h.redrive(failed).await; + assert_eq!( + intake.requests().len(), + 6, + "three more attempts in the single redrive" + ); + + assert!(h.flush().await.is_empty()); + assert_eq!( + intake.requests().len(), + 6, + "nothing left to send after the redrive" + ); + for r in intake.requests() { + assert_eq!(r.json(), Value::Array(h.invocation_start_logs())); + } + counts.assert(0, 2); +} + +#[tokio::test(flavor = "current_thread")] +async fn status_500_is_retried_then_redriven_once() { + assert_retried_three_times_then_redriven_once(500).await; +} + +#[tokio::test(flavor = "current_thread")] +async fn status_400_is_retried_then_redriven_once() { + assert_retried_three_times_then_redriven_once(400).await; +} + +#[tokio::test(flavor = "current_thread")] +async fn status_401_is_retried_then_redriven_once() { + assert_retried_three_times_then_redriven_once(401).await; +} + +#[tokio::test(flavor = "current_thread")] +async fn status_413_is_retried_then_redriven_once() { + assert_retried_three_times_then_redriven_once(413).await; +} + +#[tokio::test(flavor = "current_thread")] +async fn status_429_is_retried_then_redriven_once() { + assert_retried_three_times_then_redriven_once(429).await; +} + +/// A redrive resends only the failed request. Logs that arrive in between wait for the +/// next regular flush, so a failure there still gets its own redrive. +#[tokio::test(flavor = "current_thread")] +async fn redrive_does_not_send_new_logs() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + intake.script_statuses(&[500; 3]); + let mut h = Harness::new(base_config(&intake)); + + h.send_invocation_start().await; + let failed = h.flush().await; + assert_eq!(failed.len(), 1); + h.send(&[&function_log("after failure")]).await; + + h.redrive(failed).await; + let requests = intake.requests(); + assert_eq!(requests.len(), 4, "only the failed request is re-sent"); + assert_eq!(requests[3].json(), Value::Array(h.invocation_start_logs())); + + assert!(h.flush().await.is_empty()); + let requests = intake.requests(); + assert_eq!( + requests.len(), + 5, + "the new log ships on the next regular flush" + ); + assert_eq!( + requests[4].json(), + json!([h.log("after failure", Some(REQUEST_ID), TS_FUNCTION, "info")]) + ); + counts.assert(0, 1); +} + +#[tokio::test(flavor = "current_thread")] +async fn connection_error_is_handed_back() { + let (counts, _guard) = LevelCounts::install(); + let closed_port = { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind"); + listener.local_addr().expect("addr").port() + }; // listener dropped: connections are refused + let intake = FakeLogsIntake::start_http().await; // only for base_config; never called + let mut h = Harness::new(Config { + logs_config_logs_dd_url: format!("http://127.0.0.1:{closed_port}"), + ..base_config(&intake) + }); + + h.send_invocation_start().await; + let failed = h.flush().await; + + assert_eq!(failed.len(), 1); + assert!(intake.requests().is_empty()); + counts.assert(0, 1); +} + +/// A request that gets no answer within `flush_timeout` is retried, then handed back. +#[tokio::test(flavor = "current_thread")] +async fn request_timeout_is_retried_then_handed_back() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + intake.script_hangs(3); + let mut h = Harness::new(Config { + flush_timeout: 1, + ..base_config(&intake) + }); + + h.send_invocation_start().await; + let failed = tokio::time::timeout(Duration::from_secs(30), h.flush()) + .await + .expect("flush must give up on its own within flush_timeout"); + + assert_eq!(failed.len(), 1); + assert_eq!(intake.requests().len(), 3, "three attempts, each timed out"); + + h.redrive(failed).await; + let requests = intake.requests(); + assert_eq!(requests.len(), 4, "the redrive is answered with 202"); + assert_eq!(requests[3].json(), Value::Array(h.invocation_start_logs())); + counts.assert(0, 1); +} + +#[tokio::test(flavor = "current_thread")] +async fn additional_endpoint_failure_is_handed_back_for_redrive() { + let (counts, _guard) = LevelCounts::install(); + let primary = FakeLogsIntake::start_http().await; + let extra = FakeLogsIntake::start_https().await; + extra.script_statuses(&[500, 500, 500]); + let mut h = Harness::new(Config { + logs_config_additional_endpoints: vec![additional_endpoint(&extra, "extra-key-1")], + ..base_config(&primary) + }); + + h.send_invocation_start().await; + let failed = h.flush().await; + + assert_eq!( + failed.len(), + 1, + "the additional endpoint's request is handed back" + ); + assert_eq!(primary.requests().len(), 1); + assert_eq!(extra.requests().len(), 3); + + h.redrive(failed).await; + assert_eq!(primary.requests().len(), 1, "primary is not re-sent"); + assert_eq!( + extra.requests().len(), + 4, + "redrive goes to the additional endpoint" + ); + assert_eq!( + extra.requests()[3].header("DD-API-KEY"), + Some("extra-key-1") + ); + counts.assert(0, 1); +} + +#[tokio::test(flavor = "current_thread")] +async fn api_key_resolution_failure_drops_the_batch() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::with_unresolvable_api_key(base_config(&intake)); + + h.send_invocation_start().await; + let failed = h.flush().await; + + assert!(failed.is_empty()); + assert!(intake.requests().is_empty()); + assert_eq!( + h.drain_pending_batches().await, + 0, + "drained logs are dropped, not kept for later" + ); + counts.assert(0, 1); + + // The error is logged on every flush, even with nothing to send. + assert!(h.flush().await.is_empty()); + counts.assert(0, 2); +} + +/// When the primary's API key cannot be resolved, only the primary is skipped: additional +/// endpoints still receive the logs with their own keys. +#[tokio::test(flavor = "current_thread")] +async fn api_key_resolution_failure_still_sends_to_additional_endpoints() { + let (counts, _guard) = LevelCounts::install(); + let primary = FakeLogsIntake::start_http().await; + let extra = FakeLogsIntake::start_https().await; + let mut h = Harness::with_unresolvable_api_key(Config { + logs_config_additional_endpoints: vec![additional_endpoint(&extra, "extra-key-1")], + ..base_config(&primary) + }); + + h.send_invocation_start().await; + assert!(h.flush().await.is_empty()); + + assert!(primary.requests().is_empty()); + let requests = extra.requests(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].header("DD-API-KEY"), Some("extra-key-1")); + assert_eq!(requests[0].json(), Value::Array(h.invocation_start_logs())); + counts.assert(0, 1); +} + +// --------------------------------------------------------------------------------------- +// Batching +// --------------------------------------------------------------------------------------- + +const MIB: usize = 1024 * 1024; + +/// Split points are not pinned: only that every log arrives once, within the per-request +/// entry limit, and in order within each request. +#[tokio::test(flavor = "current_thread")] +async fn many_logs_are_split_into_batches_and_all_delivered() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::new(base_config(&intake)); + let messages: Vec = (0..2_500).map(|i| format!("log {i:04}")).collect(); + + h.send_invocation_start().await; + for message in &messages { + h.send(&[&function_log(message.as_str())]).await; + } + assert!(h.flush().await.is_empty()); + + let mut delivered = Vec::new(); + for (i, request) in intake.requests().iter().enumerate() { + let Value::Array(entries) = request.json() else { + panic!("request {i} is not a JSON array"); + }; + assert!( + entries.len() <= 1000, + "request {i} has {} entries, more than 1000", + entries.len() + ); + let numbers: Vec = entries + .iter() + .filter_map(|e| e["message"]["message"].as_str()?.strip_prefix("log ")) + .map(|n| n.parse().expect("numbered log")) + .collect(); + assert!( + numbers.is_sorted_by(|a, b| a < b), + "request {i}: numbered logs out of order" + ); + delivered.extend(entries); + } + + let mut expected = h.invocation_start_logs(); + expected.extend( + messages + .iter() + .map(|m| h.log(m, Some(REQUEST_ID), TS_FUNCTION, "info")), + ); + assert_each_log_delivered_once(delivered, expected); + counts.assert(0, 0); +} + +#[tokio::test(flavor = "current_thread")] +async fn logs_over_5_mib_in_total_are_split_across_requests() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::new(base_config(&intake)); + let bigs: Vec = ["a", "b", "c"].iter().map(|c| c.repeat(2 * MIB)).collect(); + + h.send_invocation_start().await; + for big in &bigs { + h.send(&[&function_log(big.as_str())]).await; + } + assert!(h.flush().await.is_empty()); + + let mut delivered = Vec::new(); + for (i, request) in intake.requests().iter().enumerate() { + assert!( + request.body.len() <= 5 * MIB, + "request {i} body is {} bytes, more than 5 MiB", + request.body.len() + ); + let Value::Array(entries) = request.json() else { + panic!("request {i} is not a JSON array"); + }; + delivered.extend(entries); + } + + let mut expected = h.invocation_start_logs(); + expected.extend( + bigs.iter() + .map(|big| h.log(big, Some(REQUEST_ID), TS_FUNCTION, "info")), + ); + assert_each_log_delivered_once(delivered, expected); + counts.assert(3, 0); +} + +// --------------------------------------------------------------------------------------- +// Oversized logs +// --------------------------------------------------------------------------------------- + +#[tokio::test(flavor = "current_thread")] +async fn log_between_1_and_5_mib_is_sent_with_one_warning() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::new(base_config(&intake)); + let big = "a".repeat(2 * MIB); + + h.send_invocation_start().await; + h.send(&[&function_log(big.as_str())]).await; + assert!(h.flush().await.is_empty()); + + let mut expected = h.invocation_start_logs(); + expected.push(h.log(&big, Some(REQUEST_ID), TS_FUNCTION, "info")); + let actual = single_request_logs(&intake); + let expected = Value::Array(expected); + assert!( + actual == expected, + "logs differ\n--- actual ---\n{}\n--- expected ---\n{}", + summarize(&actual), + summarize(&expected) + ); + counts.assert(1, 0); +} + +/// Pins a known bug on purpose (a later change fixes it): a log too large for any payload +/// stays at the head of the queue forever, so every log behind it is never sent. +#[tokio::test(flavor = "current_thread")] +async fn log_over_5_mib_blocks_later_logs() { + let (counts, _guard) = LevelCounts::install(); + let intake = FakeLogsIntake::start_http().await; + let mut h = Harness::new(base_config(&intake)); + + h.send_invocation_start().await; + h.send(&[&function_log("a".repeat(6 * MIB)), &function_log("after")]) + .await; + assert!(h.flush().await.is_empty()); + assert!(h.flush().await.is_empty()); + + let actual = single_request_logs(&intake); + let expected = Value::Array(h.invocation_start_logs()); + assert!( + actual == expected, + "only the logs queued before the huge one are sent: logs differ\n--- actual ---\n{}\n--- expected ---\n{}", + summarize(&actual), + summarize(&expected) + ); + counts.assert(0, 0); +}