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); +}