diff --git a/.gitignore b/.gitignore index 06fcc1e10..1da84db82 100644 --- a/.gitignore +++ b/.gitignore @@ -30,3 +30,6 @@ integration-tests/cdk.context.json /CLAUDE_PERSONAL.md # Generated benchmark fixture corpus (see bottlecap/tests/fixtures/README.md) bottlecap/tests/fixtures/generated +# Criterion baseline exports are machine specific; keep them as CI artifacts, not in git. +bottlecap/benches/baselines/*.json +bottlecap/target-linux-arm64 diff --git a/bottlecap/Cargo.lock b/bottlecap/Cargo.lock index b9b7c1b46..38890c990 100644 --- a/bottlecap/Cargo.lock +++ b/bottlecap/Cargo.lock @@ -35,6 +35,18 @@ version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" +[[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" @@ -472,6 +484,7 @@ dependencies = [ "bytes", "chrono", "cookie", + "criterion", "datadog-agent-config", "datadog-agent-trace-sampler", "datadog-fips", @@ -588,6 +601,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" @@ -648,6 +667,33 @@ dependencies = [ "serde", ] +[[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" @@ -659,6 +705,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" @@ -772,6 +843,42 @@ 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", + "futures", + "is-terminal", + "itertools 0.10.5", + "num-traits", + "once_cell", + "oorandom", + "regex", + "serde", + "serde_derive", + "serde_json", + "tinytemplate", + "tokio", + "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" @@ -1381,6 +1488,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.13.2" @@ -1850,6 +1968,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" @@ -2668,6 +2806,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" @@ -4246,6 +4390,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 f437d5ad4..e729fc497 100644 --- a/bottlecap/Cargo.toml +++ b/bottlecap/Cargo.toml @@ -124,16 +124,49 @@ rmp-serde = { version = "1.3.1", default-features = false } flate2 = { version = "1.1", default-features = false, features = ["rust_backend"] } # fault-intake test harness: decode the `datadog-protos` metrics payloads protobuf = { version = "3.7", default-features = false } +# Hot-path benchmark suite under `benches/`. See `benches/README.md`. +criterion = { version = "0.5", default-features = false, features = ["cargo_bench_support", "async_tokio"] } [build-dependencies] # No external dependencies needed for the build script +[lib] +# `cargo bench` runs only the criterion targets under `benches/`, not the +# libtest harness of the library and the binaries. +bench = false + [[bin]] name = "bottlecap" +bench = false [[bin]] name = "bottlecap-test-mode" required-features = ["test-mode"] +bench = false + +[[bench]] +name = "logs" +harness = false + +[[bench]] +name = "traces" +harness = false + +[[bench]] +name = "invocation" +harness = false + +[[bench]] +name = "enhanced_metrics" +harness = false + +[[bench]] +name = "config" +harness = false + +[[bench]] +name = "init" +harness = false [profile.dev] debug = true # same as debuginfo=2 and no stripping diff --git a/bottlecap/benches/README.md b/bottlecap/benches/README.md new file mode 100644 index 000000000..a4201bf9d --- /dev/null +++ b/bottlecap/benches/README.md @@ -0,0 +1,216 @@ +# Bottlecap benchmark suite + +Criterion benches for the hot paths of the extension. The extension runs at about 0.07 vCPU +at 128 MB, so per-line, per-span and per-invocation costs show up in billed duration. Every +performance fix should come with a bench from this suite, run before and after. + +The companion benches for the shared crates live in serverless-components: +`crates/dogstatsd/benches/hotpath.rs` (groups `dogstatsd/{parse,packet,insert,flush,compress,enhanced,ustr}`) +and `crates/datadog-agent-config/benches/load.rs` (groups `config/{get_config,get_config_yaml,replace_rules}`). +Run them with `cargo bench -p dogstatsd` and `cargo bench -p datadog-agent-config`; the +workspace release profile is the same opt-level "z" + LTO. Those benches use the system +allocator, not jemalloc. + +## Benches + +| Target (`--bench`) | Group ids | What it measures | +|---|---|---| +| `logs` | `logs/listener_parse`, `logs/processor_process`, `logs/processor_parts`, `logs/oom_check`, `logs/stitch`, `logs/aggregator`, `logs/compress` | The Telemetry API body parse, `LambdaProcessor::process` per line (text, JSON object, JSON string), the operations inside it and their alternatives, the OOM check, split-record stitching, the aggregator batch build, the zstd flush encode | +| `traces` | `traces/decode_v04`, `traces/dedup`, `traces/process_traces`, `traces/stats_copy`, `traces/body_size_encoded_len`, `traces/input_clone`, `traces/tag_injection`, `traces/obfuscate_span`, `traces/normalize_and_top_level`, `traces/regex_compile_per_root_span`, `traces/stats`, `traces/flush` | A 2000-span v0.4 POST: decode, span dedup, `process_traces`, the stats copy, tag injection, obfuscation, normalization, regex filters, stats generation, the flush encode and compress | +| `invocation` | `invocation/trigger`, `invocation/payload`, `invocation/lwa` | Trigger identification on the 38 payloads plus a 220 KB SQS and a 100-record DynamoDB event, the payload parse and clones, the LWA re-parse | +| `enhanced_metrics` | `enhanced_metrics/set_max`, `enhanced_metrics/usage_tick`, `enhanced_metrics/per_invocation_reads`, `enhanced_metrics/dynamic_tags`, `enhanced_metrics/metric_new` | `set_max_enhanced_metrics` (2 regex compiles + `/proc//limits`), the usage-monitor tick, the per-invocation `/proc` reads, the dynamic tag rebuild, `Metric::new`. Real `/proc` variants run on Linux only; fixture-tree variants run everywhere | +| `config` | `config/get_config`, `config/replace_rules`, `config/processing_rules` | `get_config` with the none, typical and heavy env sets; the replace-rule and processing-rule compiles | +| `init` | `init/ustr`, `init/http_client`, `init/tokio_runtime` | The `Ustr` first use (in a child process, because the interner is global), the HTTP client builds, the runtime build | + +Inputs come from the generators in `tests/fixtures/` (see `tests/fixtures/README.md`). Benches +whose id starts with `alt_` measure an alternative from the review, not production code; they +are the target a fix should reach. The `alt_*` log-line benches are lower bounds: they keep the +status mapping and the OOM scan but skip the processing rules, the durable-function handling and +the aggregator insert. + +`criterion` is a dev-dependency with `default-features = false`: no `plotters` (so no HTML +report under `target/criterion`, use critcmp or the CLI output) and no `rayon`, which keeps the +benches on one thread; the `std::env::set_var` calls in `config` and `common::setup_env` rely on that. + +## Run + +From `bottlecap/`: + +```sh +cargo bench # everything, production profile (slow link, see below) +cargo bench --bench logs # one target +cargo bench --bench logs -- 'processor' # ids matching a regex +cargo bench -- --quick # fewer samples, for a smoke run +cargo bench --no-run # compile only +``` + +`cargo bench` builds with the `bench` profile, which inherits `[profile.release]` from +`Cargo.toml`: `opt-level = "z"`, `lto = true`, `codegen-units = 1`. That is the profile the +extension ships with, so the numbers are comparable with production (cargo forces +`panic = "unwind"` for bench binaries; nothing else differs). The price is the link: a fat-LTO +link of one bench binary takes a few minutes on a laptop. + +To iterate on a bench, skip LTO. The numbers are then not the production profile; the review +measured `opt-level 3` 1.4 to 1.8 times faster than `z` on the logs paths (item 5.15): + +```sh +CARGO_PROFILE_BENCH_LTO=false CARGO_PROFILE_BENCH_CODEGEN_UNITS=16 cargo bench --bench logs -- --quick +# or +BENCH_FAST=1 ../scripts/bench-compare.sh save scratch logs -- --quick +``` + +## Baselines and comparison + +`scripts/bench-compare.sh` wraps criterion's `--save-baseline` and +[critcmp](https://github.com/BurntSushi/critcmp) (`cargo install critcmp`): + +```sh +scripts/bench-compare.sh save main # run the suite, save baseline "main" +# ... apply a change ... +scripts/bench-compare.sh save fix logs traces # only two targets +scripts/bench-compare.sh compare main fix -t 5 # table, differences over 5 % only +scripts/bench-compare.sh export main /tmp/main.json +scripts/bench-compare.sh list +``` + +Without critcmp, criterion compares on its own: `cargo bench -- --baseline main` prints the +change against the saved baseline after each bench. + +`save` writes `target/criterion/baseline-.txt` with the machine, the load average, the +profile (with any `CARGO_PROFILE_BENCH_*` override) and the commit, so a number is never read +without its context. `scripts/bench-docker-arm64.sh` writes the same note, with the image and +the container limits. + +### First baseline + +The first baseline, `mac-main`, is the table at the end of this file; its notes are in +`benches/baselines/mac-main.txt`. The critcmp export (`scripts/bench-compare.sh export`) is +machine specific and 350 KB, so it is not in git (`benches/baselines/*.json` is ignored); CI +keeps exports as artifacts, and a local one compares with `critcmp old.json ` (critcmp +accepts a file path in place of a baseline name). + +- Commit: `8a2cc021` on `main`, plus this suite. +- Machine: Apple M4 Max, 16 cores, 64 GB, macOS 26.6. Recorded 2026-10-28 with a 1-minute + load average of about 5 (other builds were running), so treat the numbers as +-10 %. The + Linux-only `/proc` benches were skipped; see the Docker recipe. +- Profile: `bench` = release (`opt-level = "z"`, `lto = true`, `codegen-units = 1`). +- Allocator: jemalloc, as in production (`benches/common/mod.rs`). + +## Noise control + +- Run on a quiet machine: close browsers and IDE indexers, and check `uptime`; the load + average should be below 2 before `save`. The script records it. +- Keep the CPU fixed: on a laptop, plug in the power and let it cool between runs. Criterion's + warm-up handles frequency ramp-up, not thermal throttling. On Linux, pin the run + (`taskset -c 2,3 cargo bench ...`) and, if you can, set the governor to `performance`. +- Use the production profile for any number that goes into a PR or into the table below. + `BENCH_FAST=1` is for checking that a bench runs. +- Compare on the same machine. Across machines, compare ratios (fix vs main), not absolute + times. +- Treat changes under 5 % as noise unless criterion reports `p < 0.05` on a quiet machine and + the change repeats. `critcmp -t 5` hides the small ones. +- Per-line and per-span benches report `Throughput::Elements`. Criterion and critcmp still + print the time of the whole batch; the throughput is a rate in elements per second. Divide + the batch time by the element count for the per-element time, which is what the backlog + quotes. +- The `init/ustr` benches spawn a child process per iteration. Their variance is higher; look + at the median and run them twice. + +## Linux arm64 in Docker + +The extension ships on Linux arm64 (Graviton). `scripts/bench-docker-arm64.sh` runs the suite +in a `rust:1.93-bookworm` arm64 container (native on Apple Silicon), limited to 2 CPUs like a +Lambda sandbox, with the target directory in `bottlecap/target-linux-arm64/` so the results +never mix with host baselines: + +```sh +scripts/bench-docker-arm64.sh linux-main # whole suite +scripts/bench-docker-arm64.sh linux-main enhanced_metrics # the /proc benches, Linux only +CARGO_TARGET_DIR=bottlecap/target-linux-arm64 scripts/bench-compare.sh compare linux-main linux-fix +``` + +A Lambda at 128 MB gets about 0.07 vCPU, so absolute numbers in the container are still +faster than production. The ratios between benches, and between main and a fix, transfer. +A lambda-sim environment (not in this repository) can reuse this image and the fixture +corpus to measure the whole extension under a CPU quota. + +## Backlog to bench + +Review = `consolidated-findings-v3` (Confluence page 7384761389). Numbers are the first +baseline on this Mac (see above), production profile, marked **measured**; a dash means the +bench exists but the number was not recorded yet. Per-element numbers are what to watch. + +Rows marked **measured** (pre-fix) were recorded before these fixes to the benches. Re-measure +them before you quote them: + +- The shared setup set `AWS_EXECUTION_ENV`, which Lambda does not give to extensions, and + version `1.42.0`. The tags provider then added a `runtime` tag: 17 tags on every span and log + line, now 16. This affects `traces/{process_traces,stats_copy,body_size_encoded_len,tag_injection,stats,flush}` + and `logs/{processor_process,processor_parts,aggregator,compress}`. +- The `config` benches timed the drop of the config and of the compiled rules, and + `get_config` also set `AWS_EXECUTION_ENV`, `LAMBDA_TASK_ROOT` and 9 more variables that an + extension never sees, plus the Python-only `PYTHONPATH`. +- `init/http_client` timed the client drops, and the first-client RSS did not count the + clients that stay alive. +- `traces/process_traces/regex_filters_2` dropped every chunk at the require filter, so the + work after the filter did not run. + +| Review item | Cost it names | Bench id | Baseline on this Mac | +|---|---|---|---| +| 1, 3.8 | Config parse 18 µs (none), 126 µs (20 vars), 370 µs (45 vars) on Linux arm64 | `config/get_config/{none,typical,heavy,heavy_no_replace_tags}`; sc `config/get_config/*` | **measured** (pre-fix) none 23.8 µs, typical 62.1 µs, heavy 978 µs, heavy without `DD_APM_REPLACE_TAGS` 179 µs (the host has 149 env vars; a sandbox has about 32). sc crate alone (`NoExtension`): none 24.1 µs, typical 62.2 µs, heavy 823 µs, typical from `datadog.yaml` 247 µs | +| 3.1, 4.1 | `Ustr` first use writes 9 MiB, 1.4 to 2 ms at cold start; never frees | `init/ustr/first_use`, `init/ustr/first_sorted_tags_parse`; sc `dogstatsd/ustr/*` | **measured** `Ustr::from("")` 747 µs, first `SortedTags::parse` 732 µs; child RSS +9392 KiB (9.2 MiB) on the first use; warm `Ustr::from` x16 163 ns. sc: first parse 655 µs, RSS +8.4 MiB; growth with unique tag values printed once: +1.2 MiB at 1k, +2.4 MiB at 10k, +19.5 MiB at 100k | +| 3.2 | libddwaf 1.8 to 2 MB of binary | none: binary size, use `cargo bloat` (#1443) | n/a | +| 3.3 | Additional-endpoint secrets block init | none: network; harness test T4 | n/a | +| 3.4 | 0.6 ms per replace rule compiled before `/next`; log rules too | `config/replace_rules/compile/{1,3}`, `config/processing_rules/compile/3` | **measured** (pre-fix) 1 rule 281 µs, 3 rules 766 µs (inside `get_config`, via `deserialize_apm_replace_rules` and libdd `parse_rules_from_string`); 3 log processing rules 444 µs | +| 3.5 | FIPS CA bundle loaded 2 or 3 times (1.76 ms each on M4) | `init/http_client/*` (default build; run with `--no-default-features --features fips` for the FIPS number) | **measured** (pre-fix; default build, webpki roots) first TLS client in a fresh process 37.8 µs; register + shared + trace client in `main.rs` order 54.1 µs, RSS +528 KiB; warm builds 1.9 to 3.6 µs | +| 3.6 | Register first; the register client build | `init/http_client/{register_client,first_register_client}` | **measured** (pre-fix) warm 2.3 µs, first in process 37.8 µs | +| 3.7, 3.10, 3.11 | Non-PIE, dependency weight, wrapper script | none: build flags and binary size; lambda-sim measures RSS and page faults | n/a | +| 4.2 to 4.5 | Uncollected tasks and unbounded queues grow memory | none: time series, harness test T8 (memory under a 503 storm) | n/a | +| 4.6 | `Vec::with_capacity(100)` per context (5.9 MB per 1000); new zstd encoder per request (3.66 MB) | sc `dogstatsd/flush/consume_metrics`, `dogstatsd/compress/{stream_encoder,bulk_compress,bulk_compressor_reused}_level3` | **measured** (sc, system allocator) `consume_metrics` 2.12 µs per context (2.12 ms per 1000); zstd level 3 on 668 KB: stream 135 µs, bulk 129 µs, reused 128 µs; on 6.7 MB: 1.46 / 1.32 / 1.25 ms; CCtx memory printed once: stream 3.66 MB, bulk 1.30 MB | +| 5.0 | Tick burst after a freeze: 36 µs per scan, up to 6000 scans | `enhanced_metrics/usage_tick/*` (cost per tick; the burst count is a lambda-sim test) | **measured** fake `/proc` tree (4 pids, 32 fds, 8 tasks) 194 µs on macOS, where a file open costs about 16 µs; `statfs("/tmp")` 0.85 µs. The real `/proc` tick is Linux-only: not recorded yet, run `scripts/bench-docker-arm64.sh` | +| 5.1 | Trace POST replies after processing: 26 ms per 2000 spans, 16 ms after decode | `traces/decode_v04/libdd_from_request_body` + `traces/process_traces/default` (the end-to-end HTTP seed is not ported, see "Not done") | **measured** (pre-fix) decode 10.7 ms (5.4 µs per span) + `process_traces` 11.2 ms (5.6 µs per span) = 21.9 ms of work before the reply, for a 2000-span POST | +| 5.2 | Stats deep copy 3.8 ms per 2000 spans | `traces/stats_copy/{clone,drop}` | **measured** (pre-fix) clone 2.4 ms + drop 1.58 ms (1.2 + 0.8 µs per span) | +| 5.3 | Tag injection 2.2 µs per span; regex filter compiled per span | `traces/tag_injection`, `traces/process_traces/regex_filters_2`, `traces/regex_compile_per_root_span`, alternatives `traces/tag_injection/{reserve_get_mut,root_span_only}` | **measured** (pre-fix) tag injection 4.1 ms (2.06 µs per span, 17 tags); `process_traces` with 2 regex filters 40.2 ms vs 11.2 ms without (3.6x, with every chunk dropped); 2 `Regex::new` per root span 28.8 ms per 100 roots (144 µs each); `reserve_get_mut` 3.5 ms, root-span-only 0.99 ms | +| 5.4 | libdd `from_slice` 34 % faster; encode without copy 21 % faster | `traces/decode_v04/{libdd_from_request_body,libdd_from_slice,libdd_from_slice_to_owned,rmp_serde_from_slice}`, `traces/flush/{payload_to_vec,prost_encode_agent_payload,encode/current,encode/borrowed_manual}` | **measured** (pre-fix) decode: libdd `from_request_body` 10.7 ms, libdd borrowed `from_slice` 2.2 ms, borrowed + convert to `pb::Span` 4.1 ms, `rmp_serde::from_slice` 6.5 ms; flush encode: `to_vec` + `AgentPayload` encode 10.7 ms vs borrowed manual encode 6.3 ms (byte-identical), of which `to_vec` 2.3 ms and prost encode 7.1 ms | +| 5.5 | JSON log line parsed twice, copied, serialized 2 or 3 times: 32 to 41 µs vs 7 to 9 µs | `logs/listener_parse/*`, `logs/processor_process/per_line/*`, `logs/processor_parts/*`, target `alt_raw_value_end_to_end` | **measured** (pre-fix) per line: listener parse (flatten) json_object 12.6 µs, json_string 3.8 µs, text 0.63 µs; `process()` json_object 23.6 µs, json_string 17.0 µs, text 2.8 µs (so 36 µs today for a 2 KB JSON object line); parts: event clone 2.3 µs, `to_string(obj)` 2.4 µs, re-parse 8.9 µs, `IntakeLog` serialize 2.5 µs; lower bound `alt_raw_value_end_to_end` json_object 6.8 µs (parse + extract + status + OOM scan + one serialize; no rules, durable handling or aggregator insert), `raw_value_record` parse 2.2 µs | +| 5.6 | OOM check 7 scans per line: 1 to 4 µs vs under 0.4 µs | `logs/oom_check/is_oom_error_7_contains/{200,2000}`, target `alt_single_regex` | **measured** 1.02 µs (200 B), 3.5 µs (2000 B), 0.40 µs on a hit; one regex 0.031 µs and 0.19 µs | +| 5.7 | Fixed fields serialized per line: 0.6 to 0.9 µs | `logs/processor_parts/serialize_intake_log`, target `alt_prebuilt_suffix_serialize` | **measured** (pre-fix) text 0.77 µs, JSON 2.5 µs per line; prebuilt suffix 0.18 µs and 1.8 µs | +| 5.8 | `/proc` poll every 10 ms: 26 to 36 µs | `enhanced_metrics/usage_tick/*`, `enhanced_metrics/per_invocation_reads/*` | **measured** on macOS fixtures only: `/proc/stat` 16.4 µs, `/proc/net/dev` 16.2 µs, `/proc/uptime` 16.1 µs (file-open bound). Linux numbers pending (Docker recipe) | +| 5.9 | `set_max_enhanced_metrics` 485 to 506 µs vs 54 µs | `enhanced_metrics/set_max/*` | **measured** fixture tree 367 µs, of which the 2 `Regex::new` are 236 µs; hoisted regex 70.5 µs; `strip_prefix` 64 µs; `Lambda::set_max_enhanced_metrics` on macOS (empty pid list) 242 µs | +| 5.10 | Payload parsed then copied 3 or 4 times: 519 µs (220 KB SQS) + 557 µs LWA | `invocation/payload/{parse,clone,parse_plus_3_clones}/*`, `invocation/trigger/from_value/*`, `invocation/lwa/parse/*`, target `payload/typed_partial_parse` | **measured** SQS 220 KB: parse 480 µs (noisy; 383 µs in a second run), `Value` clone 10 µs (10 large strings), parse + 3 clones 436 µs, LWA re-parse 1.11 ms, typed partial parse 202 µs; DynamoDB 100 records (126 KB): parse 1.10 ms, clone 308 µs, parse + 3 clones 2.9 ms, LWA 2.6 ms; `IdentifiedTrigger::from_value` 1.3 to 9.2 µs on the 37 checked-in events (msk with headers 21 µs), 12.6 µs on SQS 220 KB, 505 µs on DynamoDB 100 records; it runs up to 3 times per invocation | +| 5.11 | Dual-shipped traces not compressed | `traces/flush/zstd/{1,3}` (the cost of compressing) | **measured** (pre-fix) 1.84 MB protobuf: zstd level 1 1.65 ms (to 146 KB), level 3 1.80 ms (to 143 KB); `SendData::send` with a fake HTTP client 12.7 ms in total | +| 5.12 | Metrics flush copies series and sketches 2 or more times: 4.7 MB, 1.5 ms per 2000 contexts | sc `dogstatsd/flush/{series_clone,sketch_payload_clone,series_to_json,sketch_to_protobuf}` | **measured** (sc) per context: series clone 415 ns, sketch payload clone 222 to 274 ns, series to JSON 815 ns, sketch to protobuf 527 to 586 ns; `consume_distributions` 1.34 µs | +| 5.13 | Connection pooling off: 2 to 6 ms per flush | none: network; lambda-sim | n/a | +| 5.14 | zstd inline 10 to 17 ms per 5 MiB; stitching 4 MB 100 to 181 ms | `logs/compress/*`, `logs/stitch/*`, `logs/aggregator/*` | **measured** (pre-fix) zstd stream level 3: 2.26 MB noisy batch 6.8 ms (316 MB/s; synthetic text compresses 140x and runs at 4.3 GB/s, so use the noisy row), level 1 2.1 ms; stitch one 1 MiB record 16.6 ms, 4 MiB 88.5 ms, 4 MiB across 3 POSTs 128 ms; the failed full parse before it 0.77 ms; aggregator 10k lines 481 µs (48 ns/line) | +| 5.15 | opt-level 3 vs z, 1.4 to 1.8x on logs | the whole suite with `CARGO_PROFILE_BENCH_OPT_LEVEL=3`, compared with `scripts/bench-compare.sh compare` | — | +| 5.16 | Dedup round trip per span, window of 100 too small | `traces/dedup/{service_round_trip/{no_timeout,timeout_5s},inline_mutex}` and the printed caught count | **measured** channel round trip 692 µs per 2000 spans (346 ns per span), 1.62 ms with the 5 s timeout (810 ns); inline `Mutex` 171 µs (86 ns); a resend of the same 2000 spans catches 0 duplicates with capacity 100 | +| 5.17 | Stats: a copy and a channel message per span, 5.3 µs | `traces/stats/{generator_send,send_and_drain}`, `traces/process_traces/compute_stats` | **measured** (pre-fix) `StatsGenerator::send` 2.2 ms (1.1 µs per span), send + concentrator drain 6.3 ms (3.2 µs per span); `process_traces` with compute_stats 10.6 ms | +| 5.18 | DogStatsD: merge 30 to 685 ns vs insert 7 to 94 ns; hash 264 vs 10 ns; dedup before sort | sc `dogstatsd/insert/{sketch_merge_single_point,sketch_insert}/{1,64,2000}_distinct`, `dogstatsd/insert/{count_100_contexts,count_vacant_10000_contexts,distribution_10_contexts}`, `dogstatsd/parse/*`, `dogstatsd/packet/20_lines` | **measured** (sc) merge 43.5 / 173 / 684 ns vs insert 6.9 / 27.5 / 93.9 ns per value (6 to 7x); count insert 70.8 ns per metric (occupied), 341 ns (vacant, with rehash); distribution insert 643 ns; parse count 328 ns, distribution 8 tags 1.12 µs, `SortedTags::parse` 8 tags 678 ns, 16 enrich tags 1.69 µs, `metric::id` 250 ns; packet of 20 lines 8.3 µs (416 ns per line) | +| 5.19, 5.22 | Size cap no-op; Managed Instance actor throughput | none: correctness and load test | n/a | +| 5.20 | Proxy tee body copies 3 times, 2 to 4 ms per 6 MB | not covered yet (proxy users only), see "Not done" | — | +| 5.21 | Enhanced metric tags rebuilt 15 to 26 times per invocation: 530 ns vs 22 ns | `enhanced_metrics/dynamic_tags/{rebuild_3_tags,cached_clone}`, `enhanced_metrics/metric_new/distribution`; sc `dogstatsd/enhanced/*` | **measured** rebuild 499 ns vs cached clone 18.7 ns; `Metric::new` 176 ns | +| 5.23 | No hot-path benchmarks | this suite | — | +| 9.4 patch 0003 | `zstd::bulk`, no deep copies per metrics flush | sc `dogstatsd/compress/*`, `dogstatsd/flush/*` (before numbers) | **measured** (sc) see 4.6 and 5.12; the stream encoder costs 5 to 15 % more time and 2.8x the memory of bulk | +| 9.4 patch 0006 | Replace rules compiled on first use | `config/replace_rules/compile/*`, `config/get_config/heavy` (before numbers) | **measured** (pre-fix) heavy 978 µs vs 179 µs without replace tags: the 3-rule compile is about 0.8 ms of a heavy `get_config` | +| 9.4 `ustr` branch | RSS +9.3 MiB to +144 KiB after the first parse | `init/ustr/*`; sc `dogstatsd/ustr/*` (before numbers) | **measured** 747 µs and +9.2 MiB on the first use | +| 9.4 config loader | 30 to 190 µs and 150 KB saved | `config/get_config/*` (before numbers) | **measured** (pre-fix) none 23.8 µs, typical 62.1 µs, heavy 978 µs | + +### Not done + +- The end-to-end HTTP round trip of a trace POST (seed `bench2_http_end_to_end`) needs a + `TraceAgent` with `InvocationProcessorHandle::noop()`, which only exists under the + `test-mode` feature. Port it as `traces/http_round_trip` with `required-features = ["test-mode"]`. +- The proxy tee body (5.20) has no bench; `proxy/tee_body.rs` needs a hyper body fixture. +- Memory benches (RSS, allocations) are one-off prints, not criterion series. A counting + allocator like the one in the dogstatsd seed would make them tracked numbers. + +## Items exposed for the benches + +Production code changed only to make these reachable; each is `#[doc(hidden)] pub` and has +no behavior change: + +| Item | Why | +|---|---| +| `extension::telemetry::stitch` (module), `FragmentBuffer`, `Stitch`, `FragmentBuffer::stitch` | `logs/stitch` and the fixture test feed split bodies to the stitcher | +| `logs::lambda::processor::is_oom_error` | `logs/oom_check` measures the 7-substring scan on its own | +| `proc::{get_network_data, get_cpu_data, get_uptime, get_fd_max_data, get_fd_use_data, get_threads_max_data, get_threads_use_data}_from_path` | `enhanced_metrics` runs the `/proc` readers on fixture trees, so the bench works on macOS | diff --git a/bottlecap/benches/baselines/mac-main.txt b/bottlecap/benches/baselines/mac-main.txt new file mode 100644 index 000000000..c6705985c --- /dev/null +++ b/bottlecap/benches/baselines/mac-main.txt @@ -0,0 +1,3 @@ +machine: Darwin arm64, Apple M4 Max; load average: 4.99 5.16 9.46; profile: bench = release (opt-level z, lto, codegen-units 1) +saved: 2026-10-28T06:23:10Z +git: 8a2cc021 diff --git a/bottlecap/benches/common/mod.rs b/bottlecap/benches/common/mod.rs new file mode 100644 index 000000000..65b653dca --- /dev/null +++ b/bottlecap/benches/common/mod.rs @@ -0,0 +1,84 @@ +//! Shared setup for the criterion benches: the production allocator, a Lambda-like config +//! and tags provider, and a 2-worker runtime (Lambda exposes 2 vCPUs). +#![allow(dead_code)] + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use bottlecap::LAMBDA_RUNTIME_SLUG; +use bottlecap::config::Config; +use bottlecap::tags::provider::Provider; +use criterion::Criterion; + +use crate::fixtures::invocation::{ + ENV, FUNCTION_ARN, FUNCTION_NAME, FUNCTION_VERSION, INIT_TYPE, MEMORY_SIZE_MB, REGION, SERVICE, + VERSION, +}; + +/// Production uses jemalloc; measure with it. +#[global_allocator] +static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; + +/// Sets the AWS variables the tags provider and the config read, for the fixture function. No +/// `AWS_EXECUTION_ENV`: Lambda gives it to the runtime process only, not to extensions. +pub fn setup_env() { + for (k, v) in [ + ("AWS_LAMBDA_FUNCTION_NAME", FUNCTION_NAME.to_string()), + ("AWS_LAMBDA_FUNCTION_VERSION", FUNCTION_VERSION.to_string()), + ( + "AWS_LAMBDA_FUNCTION_MEMORY_SIZE", + MEMORY_SIZE_MB.to_string(), + ), + ("AWS_LAMBDA_INITIALIZATION_TYPE", INIT_TYPE.to_string()), + ("AWS_REGION", REGION.to_string()), + ] { + // Benches are single-threaded at setup time. + unsafe { std::env::set_var(k, v) }; + } +} + +/// A typical customer config: the fixture service, env and version, and 4 custom tags. +pub fn config() -> Arc { + let mut config = Config { + service: Some(SERVICE.to_string()), + env: Some(ENV.to_string()), + version: Some(VERSION.to_string()), + tags: HashMap::from([ + ("env".to_string(), ENV.to_string()), + ("team".to_string(), "checkout".to_string()), + ("cost_center".to_string(), "cc-1234".to_string()), + ("owner".to_string(), "payments-platform".to_string()), + ]), + apm_dd_url: "https://trace.agent.datadoghq.com".to_string(), + ..Config::default() + }; + config.ext.serverless_logs_enabled = true; + Arc::new(config) +} + +pub fn provider(config: Arc) -> Arc { + Arc::new(Provider::new( + config, + LAMBDA_RUNTIME_SLUG.to_string(), + &HashMap::from([("function_arn".to_string(), FUNCTION_ARN.to_string())]), + )) +} + +/// A multi-thread runtime with 2 workers, like the extension's. +pub fn runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .expect("build tokio runtime") +} + +/// Criterion with the suite's defaults. `cargo bench -- --quick` still shortens runs. +pub fn criterion() -> Criterion { + Criterion::default() + .warm_up_time(Duration::from_millis(500)) + .measurement_time(Duration::from_secs(3)) + .sample_size(50) + .configure_from_args() +} diff --git a/bottlecap/benches/config.rs b/bottlecap/benches/config.rs new file mode 100644 index 000000000..e44cb7756 --- /dev/null +++ b/bottlecap/benches/config.rs @@ -0,0 +1,223 @@ +//! Config load: `get_config` with the none, typical and heavy env sets, and the APM +//! replace-rule compile. +//! +//! Review items: 1, 3.4, 3.8. +//! +//! No tracing subscriber is installed, so the config warnings cost only their level check. The +//! extension keeps the config and the compiled rules, so their drop is outside the measurement. + +mod common; +#[path = "../tests/fixtures/mod.rs"] +mod fixtures; + +use std::hint::black_box; + +use bottlecap::config::apm_replace_rule::deserialize_apm_replace_rules; +use bottlecap::config::processing_rule::ProcessingRule; +use bottlecap::logs::lambda::IntakeLog; +use bottlecap::logs::lambda::processor::LambdaProcessor; +use bottlecap::logs::processor::Processor; +use criterion::{Criterion, criterion_group, criterion_main}; +use fixtures::invocation::{ENV, REGION, SERVICE, VERSION}; +use serde::de::value::{Error as DeError, StrDeserializer}; + +// The rest of the environment Lambda gives an extension (`common::setup_env` sets the function +// identity), without any DD_* variable. PATH is left out so the bench keeps the host PATH. The +// variables Lambda gives only to the runtime process (`AWS_EXECUTION_ENV`, `LAMBDA_TASK_ROOT`, +// `_HANDLER`, the log group and X-Ray ones; see the Extensions API docs) and the per-invocation +// `_X_AMZN_TRACE_ID` are left out. +const LAMBDA_BASE: &[(&str, &str)] = &[ + ("AWS_LAMBDA_LOG_FORMAT", "Text"), + ("AWS_LAMBDA_RUNTIME_API", "127.0.0.1:9001"), + ("AWS_DEFAULT_REGION", REGION), + ("AWS_ACCESS_KEY_ID", "ASIAEXAMPLEEXAMPLE12"), + ( + "AWS_SECRET_ACCESS_KEY", + "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + ), + ( + "AWS_SESSION_TOKEN", + "IQoJb3JpZ2luX2VjEJr//////////wEaCXVzLWVhc3QtMSJHMEUCIQDexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexample", + ), + ( + "LD_LIBRARY_PATH", + "/var/lang/lib:/lib64:/usr/lib64:/var/runtime:/var/runtime/lib:/var/task:/var/task/lib:/opt/lib", + ), + ("LANG", "en_US.UTF-8"), + ("TZ", ":UTC"), + ("AWS_LAMBDA_EXEC_WRAPPER", "/opt/datadog_wrapper"), + ("SHLVL", "0"), +]; + +// A typical Datadog Lambda setup, as the Serverless Framework / CDK plugins generate it. +const TYPICAL: &[(&str, &str)] = &[ + ( + "DD_API_KEY_SECRET_ARN", + "arn:aws:secretsmanager:us-east-1:123456789012:secret:dd-api-key-AbCdEf", + ), + ("DD_SITE", "datadoghq.com"), + ("DD_ENV", ENV), + ("DD_SERVICE", SERVICE), + ("DD_VERSION", VERSION), + ("DD_TAGS", "team:payments,tier:backend"), + ("DD_LOG_LEVEL", "debug"), + ("DD_TRACE_ENABLED", "true"), + ("DD_SERVERLESS_LOGS_ENABLED", "true"), + ("DD_LAMBDA_HANDLER", "app.handler"), + ("DD_FLUSH_TO_LOG", "false"), + ("DD_MERGE_XRAY_TRACES", "false"), + ("DD_CAPTURE_LAMBDA_PAYLOAD", "false"), + ("DD_ENHANCED_METRICS", "true"), + ("DD_TRACE_OTEL_ENABLED", "false"), + ("DD_LOGS_INJECTION", "true"), + ("DD_SERVERLESS_FLUSH_STRATEGY", "end"), + ("DD_EXTENSION_VERSION", "next"), + ("DD_PROFILING_ENABLED", "false"), + ("DD_SERVERLESS_APPSEC_ENABLED", "false"), +]; + +const ENDPOINTS_JSON: &str = r#"{"https://app.datadoghq.eu":["0123456789abcdef0123456789abcde1","0123456789abcdef0123456789abcde2"],"https://app.us5.datadoghq.com":["0123456789abcdef0123456789abcde3"]}"#; + +const PROCESSING_RULES_3: &str = r#"[{"type":"exclude_at_match","name":"exclude_health","pattern":"GET /health"},{"type":"mask_sequences","name":"mask_cc","pattern":"\\d{4}-\\d{4}-\\d{4}-\\d{4}","replace_placeholder":"[REDACTED]"},{"type":"include_at_match","name":"inc","pattern":"ERROR|WARN"}]"#; + +const REPLACE_TAGS_1: &str = r#"[{"name":"http.url","pattern":"(token=)[^&]*","repl":"${1}?"}]"#; +const REPLACE_TAGS_3: &str = r#"[{"name":"http.url","pattern":"(token=)[^&]*","repl":"${1}?"},{"name":"resource.name","pattern":"/users/\\d+","repl":"/users/?"},{"name":"*","pattern":"\\d{16}","repl":"?"}]"#; + +// A heavy setup: processing rules, dual shipping, APM replace rules, filters, OTLP, DSM. +const HEAVY_EXTRA: &[(&str, &str)] = &[ + ("DD_LOGS_CONFIG_PROCESSING_RULES", PROCESSING_RULES_3), + ("DD_ADDITIONAL_ENDPOINTS", ENDPOINTS_JSON), + ("DD_APM_ADDITIONAL_ENDPOINTS", ENDPOINTS_JSON), + ( + "DD_LOGS_CONFIG_ADDITIONAL_ENDPOINTS", + r#"[{"api_key":"0123456789abcdef0123456789abcde4","Host":"agent-http-intake.logs.datadoghq.eu","Port":443,"is_reliable":true}]"#, + ), + ("DD_APM_REPLACE_TAGS", REPLACE_TAGS_3), + ( + "DD_SERVICE_MAPPING", + "lambda_sqs:my-queue,lambda_sns:my-topic", + ), + ("DD_APM_FILTER_TAGS_REQUIRE", "env:prod"), + ("DD_APM_FILTER_TAGS_REJECT", "debug:true"), + ("DD_TRACE_PROPAGATION_STYLE", "datadog,tracecontext,b3multi"), + ("DD_PROXY_HTTPS", "http://proxy.internal:3128"), + ("DD_PROXY_NO_PROXY", "169.254.169.254,localhost"), + ( + "DD_OTLP_CONFIG_RECEIVER_PROTOCOLS_HTTP_ENDPOINT", + "localhost:4318", + ), + ("DD_OTLP_CONFIG_TRACES_SPAN_NAME_REMAPPINGS", "a:b,c:d"), + ("DD_LAMBDA_CUSTOMER_METRICS_EXCLUDE_TAGS", "requestid,host"), + ("DD_SPAN_DEDUP_TIMEOUT", "5"), + ("DD_APPSEC_WAF_TIMEOUT", "5000"), + ("DD_DATA_STREAMS_ENABLED", "true"), + ("DD_COMPRESSION_LEVEL", "6"), + ("DD_FLUSH_TIMEOUT", "10"), + ("DD_LOGS_CONFIG_USE_COMPRESSION", "true"), + ("DD_APM_FEATURES", "enable_stats_by_span_kind"), + ("DD_LAMBDA_EXTENSION_COMPUTE_STATS", "true"), + ("DD_TRACE_STATS_ADDITIONAL_TAGS", "http.status_code"), + ("DD_TRACE_EXPERIMENTAL_FEATURES_ENABLED", "true"), + ("DD_API_KEY_SECRET_RELOAD_INTERVAL", "3600"), + ("DD_LAMBDA_DURABLE_FUNCTION_LOG_BUFFER_SIZE", "10"), +]; + +fn set_all(pairs: &[(&str, &str)]) { + for (k, v) in pairs { + // Benches are single-threaded at setup time. + unsafe { std::env::set_var(k, v) }; + } +} + +/// Removes every `DD_*` and `DATADOG_*` variable, including any from the host shell. +fn clear_dd_env() { + let keys: Vec<_> = std::env::vars_os() + .map(|(k, _)| k) + .filter(|k| { + k.to_str() + .is_some_and(|k| k.starts_with("DD_") || k.starts_with("DATADOG_")) + }) + .collect(); + for k in keys { + unsafe { std::env::remove_var(k) }; + } +} + +fn set_scenario(name: &str) { + clear_dd_env(); + common::setup_env(); + set_all(LAMBDA_BASE); + match name { + "none" => {} + "typical" => set_all(TYPICAL), + "heavy" => { + set_all(TYPICAL); + set_all(HEAVY_EXTRA); + } + // Isolates the replace-rule compile inside `heavy`. + "heavy_no_replace_tags" => { + set_all(TYPICAL); + set_all(HEAVY_EXTRA); + unsafe { std::env::remove_var("DD_APM_REPLACE_TAGS") }; + } + other => panic!("unknown scenario {other}"), + } +} + +fn get_config(c: &mut Criterion) { + let mut g = c.benchmark_group("config/get_config"); + // `main` reads `$LAMBDA_TASK_ROOT/datadog.yaml` (`/var/task` when unset, as for an + // extension); here the file does not exist. + let dir = tempfile::tempdir().expect("tempdir"); + for scenario in ["none", "typical", "heavy", "heavy_no_replace_tags"] { + set_scenario(scenario); + println!( + "config/get_config/{scenario}: {} env vars in the process", + std::env::vars_os().count() + ); + g.bench_function(scenario, |b| { + b.iter_with_large_drop(|| bottlecap::config::get_config(black_box(dir.path()))); + }); + } + clear_dd_env(); + g.finish(); +} + +fn replace_rules(c: &mut Criterion) { + // `DD_APM_REPLACE_TAGS` is compiled once, inside `get_config`: the env source's field uses + // `deserialize_apm_replace_rules` (datadog-agent-config `deserializers/apm_replace_rule.rs`), + // which validates the JSON, then calls `libdd_trace_obfuscation::replacer:: + // parse_rules_from_string` (one `Regex::new` per rule). `build_trace_agent` only clones the + // compiled rules into `ObfuscationConfig`. + let mut g = c.benchmark_group("config/replace_rules/compile"); + for (n, json) in [("1", REPLACE_TAGS_1), ("3", REPLACE_TAGS_3)] { + g.bench_function(n, |b| { + b.iter_with_large_drop(|| { + deserialize_apm_replace_rules(StrDeserializer::::new(black_box(json))) + .expect("replace rules") + }); + }); + } + g.finish(); +} + +fn processing_rules(c: &mut Criterion) { + // `LambdaProcessor::new` (`src/logs/lambda/processor.rs`) compiles the rules with the + // `Processor::compile_rules` default method (`src/logs/processor.rs`). + let rules: Option> = + Some(serde_json::from_str(PROCESSING_RULES_3).expect("processing rules")); + let mut g = c.benchmark_group("config/processing_rules/compile"); + g.bench_function("3", |b| { + b.iter_with_large_drop(|| { + >::compile_rules(black_box(&rules)) + }); + }); + g.finish(); +} + +criterion_group! { + name = benches; + config = common::criterion(); + targets = get_config, replace_rules, processing_rules +} +criterion_main!(benches); diff --git a/bottlecap/benches/enhanced_metrics.rs b/bottlecap/benches/enhanced_metrics.rs new file mode 100644 index 000000000..f824aaa32 --- /dev/null +++ b/bottlecap/benches/enhanced_metrics.rs @@ -0,0 +1,300 @@ +//! Enhanced metrics: `set_max_enhanced_metrics` (2 regex compiles + `/proc//limits`), +//! the usage-monitor tick (`/proc` readers), the per-invocation `/proc` reads and the +//! dynamic tag rebuild. +//! +//! Review items: 5.0, 5.8, 5.9, 5.21. +//! +//! The `*_real_proc` and `set_max/fd_and_threads_max` benches read the live `/proc` and run only +//! on Linux. The `*_fixture` and `fake_proc_tree` variants read `tests/proc` or a temp tree, so +//! they run everywhere. + +mod common; +#[path = "../tests/fixtures/mod.rs"] +mod fixtures; + +use std::collections::HashMap; +use std::fs::{self, File}; +use std::hint::black_box; +use std::io::{self, BufRead}; +use std::path::Path; + +use bottlecap::metrics::enhanced::lambda::Lambda; +use bottlecap::metrics::enhanced::statfs; +use bottlecap::proc::{self, constants}; +use criterion::{Criterion, criterion_group, criterion_main}; +use dogstatsd::aggregator::AggregatorService; +use dogstatsd::metric::{EMPTY_TAGS, Metric, MetricValue, SortedTags}; +use fixtures::invocation::{BASE_NS, INIT_TYPE, RUNTIME}; +use regex::Regex; + +const PROC_FIXTURE: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/proc"); +const LIMITS_FIXTURE: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/proc/process/valid"); +const LIMITS_PIDS: [i64; 2] = [9, 31]; + +// `get_fd_max_data_from_path` and `get_threads_max_data_from_path` in `src/proc/mod.rs`. +const FD_MAX_RE: &str = r"^Max open files\s+(\d+)"; +const THREADS_MAX_RE: &str = r"^Max processes\s+(\d+)"; + +fn skip_unless_linux(id: &str) -> bool { + if cfg!(target_os = "linux") { + return false; + } + println!("enhanced_metrics/{id}: skipped, reads the live /proc (Linux only)"); + true +} + +/// `get_fd_max_data_from_path` with the regex compiled once by the caller (PR #1278). +fn limits_min_hoisted(root: &str, pids: &[i64], re: &Regex, default: f64) -> f64 { + let mut min = default; + for &pid in pids { + let Ok(file) = File::open(format!("{root}/{pid}/limits")) else { + continue; + }; + for line in io::BufReader::new(file).lines().map_while(Result::ok) { + if let Some(caps) = re.captures(&line) { + if let Ok(v) = caps[1].parse::() { + min = min.min(v); + } + break; + } + } + } + min +} + +/// The no-regex alternative: one read per file and `strip_prefix` (review 5.9). +fn limits_min_strip_prefix(root: &str, pids: &[i64], prefix: &str, default: f64) -> f64 { + let mut min = default; + for &pid in pids { + let Ok(s) = fs::read_to_string(format!("{root}/{pid}/limits")) else { + continue; + }; + if let Some(v) = s + .lines() + .find_map(|l| l.strip_prefix(prefix)) + .and_then(|rest| rest.split_whitespace().next()) + .and_then(|v| v.parse::().ok()) + { + min = min.min(v); + } + } + min +} + +fn set_max(c: &mut Criterion) { + let mut g = c.benchmark_group("enhanced_metrics/set_max"); + + // `Lambda::set_max_enhanced_metrics`, `src/metrics/enhanced/lambda.rs`. + if !skip_unless_linux("set_max/fd_and_threads_max") { + g.bench_function("fd_and_threads_max", |b| { + b.iter(|| { + let pids = proc::get_pid_list(); + ( + proc::get_fd_max_data(&pids), + proc::get_threads_max_data(&pids), + ) + }); + }); + } + + g.bench_function("fd_and_threads_max_fixture", |b| { + b.iter(|| { + ( + proc::get_fd_max_data_from_path(LIMITS_FIXTURE, &LIMITS_PIDS), + proc::get_threads_max_data_from_path(LIMITS_FIXTURE, &LIMITS_PIDS), + ) + }); + }); + + g.bench_function("regex_compile_x2", |b| { + b.iter(|| { + ( + Regex::new(black_box(FD_MAX_RE)).expect("fd regex"), + Regex::new(black_box(THREADS_MAX_RE)).expect("threads regex"), + ) + }); + }); + + let fd_re = Regex::new(FD_MAX_RE).expect("fd regex"); + let threads_re = Regex::new(THREADS_MAX_RE).expect("threads regex"); + g.bench_function("hoisted_regex_fixture", |b| { + b.iter(|| { + ( + limits_min_hoisted( + LIMITS_FIXTURE, + &LIMITS_PIDS, + &fd_re, + constants::LAMBDA_FILE_DESCRIPTORS_DEFAULT_LIMIT, + ), + limits_min_hoisted( + LIMITS_FIXTURE, + &LIMITS_PIDS, + &threads_re, + constants::LAMBDA_EXECUTION_PROCESSES_DEFAULT_LIMIT, + ), + ) + }); + }); + + g.bench_function("no_regex_strip_prefix_fixture", |b| { + b.iter(|| { + ( + limits_min_strip_prefix( + LIMITS_FIXTURE, + &LIMITS_PIDS, + "Max open files", + constants::LAMBDA_FILE_DESCRIPTORS_DEFAULT_LIMIT, + ), + limits_min_strip_prefix( + LIMITS_FIXTURE, + &LIMITS_PIDS, + "Max processes", + constants::LAMBDA_EXECUTION_PROCESSES_DEFAULT_LIMIT, + ), + ) + }); + }); + + // The method itself: pid list, the 2 limits reads, tag rebuild, 2 `Metric::new` and 2 + // aggregator sends. Outside Linux the pid list is empty, so only the regex compiles, + // the tags and the sends remain. + let rt = common::runtime(); + let _guard = rt.enter(); + let (service, handle) = AggregatorService::new(EMPTY_TAGS, 1024).expect("aggregator"); + rt.spawn(service.run()); + let mut lambda = Lambda::new(handle, common::config()); + lambda.set_init_tags(false, true); + lambda.set_runtime_tag(RUNTIME); + g.bench_function("lambda_method", |b| { + b.iter(|| lambda.set_max_enhanced_metrics()) + }); + + g.finish(); +} + +/// A `/proc`-like tree: 4 pids, each with 32 `fd/` entries, 8 `task/` dirs and a `limits`. +fn fake_proc_tree() -> tempfile::TempDir { + let dir = tempfile::tempdir().expect("tempdir"); + let limits = Path::new(LIMITS_FIXTURE).join("9/limits"); + for pid in 101..105 { + let pid_dir = dir.path().join(pid.to_string()); + fs::create_dir_all(pid_dir.join("fd")).expect("fd dir"); + for fd in 0..32 { + File::create(pid_dir.join("fd").join(fd.to_string())).expect("fd file"); + } + for tid in 0..8 { + fs::create_dir_all(pid_dir.join("task").join((pid * 100 + tid).to_string())) + .expect("task dir"); + } + fs::copy(&limits, pid_dir.join("limits")).expect("limits"); + } + dir +} + +fn usage_tick(c: &mut Criterion) { + let mut g = c.benchmark_group("enhanced_metrics/usage_tick"); + + // `start_usage_metrics_task`, `src/metrics/enhanced/lambda.rs`. + if !skip_unless_linux("usage_tick/real_proc") { + g.bench_function("real_proc", |b| { + b.iter(|| { + let pids = proc::get_pid_list(); + ( + statfs::get_tmp_used().ok(), + proc::get_fd_use_data(&pids), + proc::get_threads_use_data(&pids).ok(), + ) + }); + }); + } + + let tree = fake_proc_tree(); + let root = tree.path().to_str().expect("utf-8 temp path"); + g.bench_function("fake_proc_tree", |b| { + b.iter(|| { + let pids = proc::get_pid_list_from_path(root); + ( + proc::get_fd_use_data_from_path(root, &pids), + proc::get_threads_use_data_from_path(root, &pids).ok(), + ) + }); + }); + + g.bench_function("statfs_tmp", |b| b.iter(|| statfs::get_tmp_used().ok())); + + g.finish(); +} + +fn per_invocation_reads(c: &mut Criterion) { + let mut g = c.benchmark_group("enhanced_metrics/per_invocation_reads"); + + let stat = format!("{PROC_FIXTURE}/stat/valid_stat"); + let net_dev = format!("{PROC_FIXTURE}/net/valid_dev"); + let uptime = format!("{PROC_FIXTURE}/uptime/valid_uptime"); + g.bench_function("cpu_stat", |b| { + b.iter(|| proc::get_cpu_data_from_path(&stat).ok()); + }); + g.bench_function("net_dev", |b| { + b.iter(|| proc::get_network_data_from_path(&net_dev).ok()); + }); + g.bench_function("uptime", |b| { + b.iter(|| proc::get_uptime_from_path(&uptime).ok()); + }); + + if !skip_unless_linux("per_invocation_reads/*_real_proc") { + g.bench_function("cpu_stat_real_proc", |b| { + b.iter(|| proc::get_cpu_data().ok()) + }); + // Off Lambda the interface is missing and this returns an error after the full read. + g.bench_function("net_dev_real_proc", |b| { + b.iter(|| proc::get_network_data().ok()); + }); + g.bench_function("uptime_real_proc", |b| b.iter(|| proc::get_uptime().ok())); + } + + g.finish(); +} + +fn dynamic_tags(c: &mut Criterion) { + let mut g = c.benchmark_group("enhanced_metrics/dynamic_tags"); + + let tags: HashMap = HashMap::from([ + ("cold_start".to_string(), "false".to_string()), + ("runtime".to_string(), RUNTIME.to_string()), + ("init_type".to_string(), INIT_TYPE.to_string()), + ]); + // `Lambda::get_dynamic_value_tags`, `src/metrics/enhanced/lambda.rs`. + g.bench_function("rebuild_3_tags", |b| { + b.iter(|| { + let vec_tags: Vec = tags.iter().map(|(k, v)| format!("{k}:{v}")).collect(); + SortedTags::parse(&vec_tags.join(",")).ok() + }); + }); + + let cached = SortedTags::parse(&format!( + "cold_start:false,runtime:{RUNTIME},init_type:{INIT_TYPE}" + )) + .ok(); + g.bench_function("cached_clone", |b| b.iter(|| cached.clone())); + g.finish(); + + let mut g = c.benchmark_group("enhanced_metrics/metric_new"); + g.bench_function("distribution", |b| { + b.iter(|| { + Metric::new( + "aws.lambda.enhanced.duration".into(), + MetricValue::distribution(0.123), + cached.clone(), + Some(BASE_NS / 1_000_000_000), + ) + }); + }); + g.finish(); +} + +criterion_group! { + name = benches; + config = common::criterion(); + targets = set_max, usage_tick, per_invocation_reads, dynamic_tags +} +criterion_main!(benches); diff --git a/bottlecap/benches/init.rs b/bottlecap/benches/init.rs new file mode 100644 index 000000000..6f8d905c1 --- /dev/null +++ b/bottlecap/benches/init.rs @@ -0,0 +1,289 @@ +//! Init pieces that run in-process: the `Ustr` first use (measured in a child process, since +//! the interner is global) and the HTTP client builds. +//! +//! Review items: 3.1, 3.5, 3.6, H15. +//! +//! With `BOTTLECAP_BENCH_CHILD=` the binary runs one step in a fresh process, prints the +//! elapsed nanoseconds on stdout and exits. The child builds the runtime first, as `main` does. +//! The `first_*` benches sum those child timings with `iter_custom`; the process spawn is not +//! measured. The child never drops what a step builds (`mem::forget`, then exit), so the timing +//! leaves out the teardown and the RSS sample counts everything the step keeps alive. + +mod common; +#[path = "../tests/fixtures/mod.rs"] +mod fixtures; + +use std::hint::black_box; +use std::mem; +use std::process::Command; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use bottlecap::config::Config; +use criterion::{BatchSize, Criterion, SamplingMode, criterion_group}; +use datadog_fips::reqwest_adapter::create_reqwest_client_builder; +use dogstatsd::metric::SortedTags; +use fixtures::dogstatsd::enrich_tags; +use ustr::Ustr; + +const CHILD_ENV: &str = "BOTTLECAP_BENCH_CHILD"; +/// When set, the child prints its RSS in KiB before and after the step instead of the time. +const CHILD_RSS_ENV: &str = "BOTTLECAP_BENCH_CHILD_RSS"; + +const USTR_FIRST_USE: &str = "ustr_first_use"; +const SORTED_TAGS_FIRST_PARSE: &str = "sorted_tags_first_parse"; +const FIRST_REGISTER_CLIENT: &str = "first_register_client"; +const FIRST_CLIENT_SEQUENCE: &str = "first_client_sequence"; + +/// the register client in `src/bin/bottlecap/main.rs`. +fn register_client() -> reqwest::Client { + create_reqwest_client_builder() + .expect("client builder") + .no_proxy() + .build() + .expect("register client") +} + +/// `build_trace_agent` in `src/startup.rs`. +fn trace_client(config: &Config) -> impl Sized { + bottlecap::traces::http_client::create_client( + config.proxy_https.as_ref(), + config.tls_cert_file.as_ref(), + config.skip_ssl_validation, + ) + .expect("trace client") +} + +/// RSS in KiB: `VmRSS` from `/proc/self/status` on Linux (no `ps` needed), `ps` elsewhere. +fn rss_kib() -> u64 { + if cfg!(target_os = "linux") { + let status = std::fs::read_to_string("/proc/self/status").expect("read /proc/self/status"); + return status + .lines() + .find_map(|l| l.strip_prefix("VmRSS:")) + .and_then(|v| v.split_whitespace().next()) + .and_then(|v| v.parse().ok()) + .expect("VmRSS in /proc/self/status"); + } + let out = Command::new("ps") + .args(["-o", "rss=", "-p", &std::process::id().to_string()]) + .output() + .expect("run ps"); + String::from_utf8_lossy(&out.stdout) + .trim() + .parse() + .expect("ps prints the RSS") +} + +fn run_step(step: &str, config: &Arc) { + match step { + // `init_ustr` in `src/bin/bottlecap/main.rs`. + USTR_FIRST_USE => { + black_box(Ustr::from("")); + } + // The first interner use when `init_ustr` has not finished: `start_dogstatsd` parses + // the enrichment tags. + SORTED_TAGS_FIRST_PARSE => { + mem::forget(black_box( + SortedTags::parse(&enrich_tags()).expect("enrichment tags"), + )); + } + FIRST_REGISTER_CLIENT => { + mem::forget(black_box(register_client())); + } + // The three client builds of a cold start, in order: register, shared `http::get_client`, + // and the trace client from `build_trace_agent`. All three stay alive, as in `main`. + FIRST_CLIENT_SEQUENCE => { + mem::forget(black_box(register_client())); + mem::forget(black_box(bottlecap::http::get_client(config))); + mem::forget(black_box(trace_client(config))); + } + other => panic!("unknown child step {other}"), + } +} + +fn child(step: &str) -> ! { + let rt = common::runtime(); + let _guard = rt.enter(); + let config = common::config(); + bottlecap::fips::prepare_client_provider().expect("crypto provider"); + if std::env::var_os(CHILD_RSS_ENV).is_some() { + let before = rss_kib(); + run_step(step, &config); + println!("{before} {}", rss_kib()); + } else { + let start = Instant::now(); + run_step(step, &config); + println!("{}", start.elapsed().as_nanos()); + } + std::process::exit(0); +} + +fn spawn_child(step: &str, rss: bool) -> String { + let mut cmd = Command::new(std::env::current_exe().expect("current exe")); + cmd.env(CHILD_ENV, step); + if rss { + cmd.env(CHILD_RSS_ENV, "1"); + } + let out = cmd.output().expect("spawn child"); + assert!( + out.status.success(), + "child {step} failed: {}", + String::from_utf8_lossy(&out.stderr) + ); + String::from_utf8(out.stdout).expect("utf-8 child output") +} + +fn print_child_rss(group: &str, step: &str) { + let out = spawn_child(step, true); + let mut kib = out + .split_whitespace() + .map(|v| v.parse::().expect("KiB")); + let (before, after) = (kib.next().expect("before"), kib.next().expect("after")); + println!( + "{group}: child RSS around {step}: {before} KiB before, {after} KiB after ({:+} KiB)", + after - before + ); +} + +/// Benches `step` in fresh child processes, in its own group `group` so the short sample +/// settings do not leak into the in-process benches. One spawn per iteration, so the counts +/// stay small; criterion's "Unable to complete 10 samples in 20.0ms" warning is expected. +fn bench_child(c: &mut Criterion, group: &str, id: &str, step: &str) { + let mut g = c.benchmark_group(group); + g.sample_size(10) + .warm_up_time(Duration::from_millis(1)) + .measurement_time(Duration::from_millis(20)) + .sampling_mode(SamplingMode::Flat); + g.bench_function(id, |b| { + b.iter_custom(|iters| { + (0..iters) + .map(|_| { + let ns = spawn_child(step, false) + .trim() + .parse() + .expect("nanoseconds"); + Duration::from_nanos(ns) + }) + .sum() + }); + }); + g.finish(); +} + +fn ustr(c: &mut Criterion) { + let mut g = c.benchmark_group("init/ustr"); + + let tags = enrich_tags(); + let existing: Vec<&str> = tags.split(',').collect(); + for s in &existing { + Ustr::from(s); + } + g.bench_function(format!("from_existing_x{}", existing.len()), |b| { + b.iter(|| { + for s in &existing { + black_box(Ustr::from(black_box(*s))); + } + }); + }); + + g.finish(); + + print_child_rss("init/ustr", USTR_FIRST_USE); + print_child_rss("init/ustr", SORTED_TAGS_FIRST_PARSE); + bench_child(c, "init/ustr", "first_use", USTR_FIRST_USE); + bench_child( + c, + "init/ustr", + "first_sorted_tags_parse", + SORTED_TAGS_FIRST_PARSE, + ); +} + +fn http_client(c: &mut Criterion) { + let rt = common::runtime(); + let _guard = rt.enter(); + let config = common::config(); + let config_http1 = Arc::new(Config { + http_protocol: Some("http1".to_string()), + ..(*config).clone() + }); + bottlecap::fips::prepare_client_provider().expect("crypto provider"); + + // Warm builds: the second and later client in a process. `main` keeps its clients, so the + // drop is outside the measurement. + let mut g = c.benchmark_group("init/http_client"); + // `http::get_client` as `main` calls it, HTTP/2 by default (no proxy). + g.bench_function("shared_get_client", |b| { + b.iter_batched( + || (), + |()| bottlecap::http::get_client(&config), + BatchSize::PerIteration, + ); + }); + g.bench_function("shared_get_client_http1", |b| { + b.iter_batched( + || (), + |()| bottlecap::http::get_client(&config_http1), + BatchSize::PerIteration, + ); + }); + g.bench_function("register_client", |b| { + b.iter_batched(|| (), |()| register_client(), BatchSize::PerIteration); + }); + g.bench_function("trace_client", |b| { + b.iter_batched(|| (), |()| trace_client(&config), BatchSize::PerIteration); + }); + + g.finish(); + + // Cold builds: the first client in a fresh process pays the TLS and lazy-static setup. + print_child_rss("init/http_client", FIRST_CLIENT_SEQUENCE); + bench_child( + c, + "init/http_client", + "first_register_client", + FIRST_REGISTER_CLIENT, + ); + bench_child( + c, + "init/http_client", + "first_client_sequence", + FIRST_CLIENT_SEQUENCE, + ); +} + +fn tokio_runtime(c: &mut Criterion) { + // `#[tokio::main]` on `main`; the drop (worker shutdown) is outside the measurement. + let mut g = c.benchmark_group("init/tokio_runtime/build"); + for workers in [2, 4] { + g.bench_function(workers.to_string(), |b| { + b.iter_batched( + || (), + |()| { + tokio::runtime::Builder::new_multi_thread() + .worker_threads(workers) + .enable_all() + .build() + .expect("tokio runtime") + }, + BatchSize::PerIteration, + ); + }); + } + g.finish(); +} + +criterion_group! { + name = benches; + config = common::criterion(); + targets = ustr, http_client, tokio_runtime +} + +fn main() { + if let Ok(step) = std::env::var(CHILD_ENV) { + child(&step); + } + benches(); + common::criterion().final_summary(); +} diff --git a/bottlecap/benches/invocation.rs b/bottlecap/benches/invocation.rs new file mode 100644 index 000000000..793a2e601 --- /dev/null +++ b/bottlecap/benches/invocation.rs @@ -0,0 +1,174 @@ +//! Per-invocation payload work: trigger identification, payload parse and clones, and the +//! LWA parse. +//! +//! Review item: 5.10. + +mod common; +#[path = "../tests/fixtures/mod.rs"] +mod fixtures; + +use std::borrow::Cow; +use std::collections::HashMap; +use std::hint::black_box; + +use bottlecap::lifecycle::invocation::triggers::IdentifiedTrigger; +use criterion::{Criterion, Throughput, criterion_group, criterion_main}; +use serde::Deserialize; +use serde_json::{Value, json}; + +/// Payloads in `tests/payloads/` that are not trigger events: `eventbridge_span` is the expected +/// inferred span of a unit test. +const UNKNOWN_PAYLOADS: &[&str] = &["eventbridge_span"]; + +fn len_u64(bytes: &[u8]) -> u64 { + bytes.len().try_into().expect("payload length fits in u64") +} + +fn triggers(c: &mut Criterion) { + let mut payloads = fixtures::events::all_payloads(); + payloads.push(("sqs_220kb".to_string(), fixtures::events::sqs_220kb())); + payloads.push(( + "dynamodb_100_records".to_string(), + fixtures::events::dynamodb_100_records(), + )); + + let mut g = c.benchmark_group("invocation"); + for (name, value) in &payloads { + let expect_unknown = UNKNOWN_PAYLOADS.contains(&name.as_str()); + assert_eq!( + IdentifiedTrigger::from_value(value).is_unknown(), + expect_unknown, + "{name}: unexpected trigger identification" + ); + let bytes = serde_json::to_vec(value).expect("serialize payload"); + g.throughput(Throughput::Bytes(len_u64(&bytes))); + // Review 5.10: production calls this up to 3 times per invocation, each with a full + // `Value` clone (`IdentifiedTrigger::from_value`): two call sites in `span_inferrer.rs` + // and `InvocationProcessor::record_dsm_consume_from_payload`. + g.bench_function(format!("trigger/from_value/{name}"), |b| { + b.iter_with_large_drop(|| IdentifiedTrigger::from_value(black_box(value))); + }); + } + g.finish(); +} + +/// The fields the SQS trigger reads, borrowed from the input where no unescape is needed. +#[derive(Deserialize)] +struct SqsPartial<'a> { + #[serde(rename = "Records", borrow)] + records: Vec>, +} + +#[derive(Deserialize)] +struct SqsRecordPartial<'a> { + #[serde(rename = "eventSource", borrow)] + event_source: Option>, + #[serde(rename = "eventSourceARN", borrow)] + event_source_arn: Option>, + #[serde(rename = "messageAttributes", borrow)] + message_attributes: Option, MsgAttrPartial<'a>>>, +} + +#[derive(Deserialize)] +struct MsgAttrPartial<'a> { + #[serde(rename = "stringValue", borrow)] + string_value: Option>, +} + +fn sqs_datadog_contexts(bytes: &[u8]) -> usize { + let parsed: SqsPartial = serde_json::from_slice(bytes).expect("typed sqs parse"); + parsed + .records + .iter() + .filter(|r| r.event_source.as_deref() == Some("aws:sqs") && r.event_source_arn.is_some()) + .filter_map(|r| r.message_attributes.as_ref()) + .filter_map(|m| m.get("_datadog")) + .filter_map(|a| a.string_value.as_deref()) + .count() +} + +fn large_payloads() -> [(&'static str, Vec); 2] { + [ + ( + "sqs_220kb", + serde_json::to_vec(&fixtures::events::sqs_220kb()).expect("serialize sqs"), + ), + ( + "dynamodb_100", + serde_json::to_vec(&fixtures::events::dynamodb_100_records()) + .expect("serialize dynamodb"), + ), + ] +} + +fn payload(c: &mut Criterion) { + let payloads = large_payloads(); + let mut g = c.benchmark_group("invocation"); + for (name, bytes) in &payloads { + let value: Value = serde_json::from_slice(bytes).expect("parse payload"); + g.throughput(Throughput::Bytes(len_u64(bytes))); + // The start-invocation handler in src/lifecycle/listener.rs + g.bench_function(format!("payload/parse/{name}"), |b| { + b.iter_with_large_drop(|| serde_json::from_slice::(bytes).expect("parse")); + }); + // `IdentifiedTrigger::from_value` + g.bench_function(format!("payload/clone/{name}"), |b| { + b.iter_with_large_drop(|| value.clone()); + }); + // Today's tracer path: one parse, then one clone per `from_value` call. Drops are timed. + g.bench_function(format!("payload/parse_plus_3_clones/{name}"), |b| { + b.iter(|| { + let v = serde_json::from_slice::(bytes).expect("parse"); + let clones = [v.clone(), v.clone(), v.clone()]; + black_box((&v, &clones)); + }); + }); + } + + // The comparison target for the SQS case: a borrowed typed parse of the fields used. + let (name, bytes) = &payloads[0]; + assert_eq!( + sqs_datadog_contexts(bytes), + 10, + "one _datadog context per record" + ); + g.throughput(Throughput::Bytes(len_u64(bytes))); + g.bench_function(format!("payload/typed_partial_parse/{name}"), |b| { + b.iter(|| sqs_datadog_contexts(bytes)); + }); + g.finish(); +} + +/// Mirrors `process_invocation_next` in src/lwa/mod.rs: parse, serialize again, then +/// parse a clone of the new body. +fn lwa_parse(body_bytes: &[u8]) -> (Value, Vec, Value) { + let inner_payload = serde_json::from_slice::(body_bytes).unwrap_or_else(|_| json!({})); + let body = serde_json::to_vec(&inner_payload).unwrap_or_default(); + let payload_value = + serde_json::from_slice::(&body.clone()).unwrap_or_else(|_| json!({})); + (inner_payload, body, payload_value) +} + +fn lwa(c: &mut Criterion) { + let [sqs, dynamodb] = large_payloads(); + let api_gateway = ( + "api_gateway_http_event", + fixtures::events::payload("api_gateway_http_event").into_bytes(), + ); + let mut g = c.benchmark_group("invocation"); + for (name, bytes) in [sqs, dynamodb, api_gateway] { + assert!(lwa_parse(&bytes).2.is_object(), "{name}: parse failed"); + g.throughput(Throughput::Bytes(len_u64(&bytes))); + g.bench_function(format!("lwa/parse/{name}"), |b| { + b.iter_with_large_drop(|| lwa_parse(&bytes)); + }); + } + g.finish(); +} + +criterion_group! { + name = benches; + config = common::criterion(); + targets = triggers, payload, lwa +} +criterion_main!(benches); diff --git a/bottlecap/benches/logs.rs b/bottlecap/benches/logs.rs new file mode 100644 index 000000000..c0dde152f --- /dev/null +++ b/bottlecap/benches/logs.rs @@ -0,0 +1,783 @@ +//! Logs pipeline hot path: listener parse, per-line processing, the OOM check, stitching, +//! the aggregator batch build and the flush compression. +//! +//! Review items: 5.5 (JSON line parsed twice, copied, serialized 2 or 3 times), 5.6 (OOM check), +//! 5.7 (fixed fields serialized per line), 5.14 (zstd inline, quadratic stitch), 2.2. +//! +//! Every per-line bench runs over `LINES` events and reports with `Throughput::Elements`. +//! Criterion's time is for the whole batch; divide it by the element count for the per-line cost. + +mod common; +#[path = "../tests/fixtures/mod.rs"] +mod fixtures; + +use std::io::Write as _; +use std::time::{Duration, Instant}; + +use bottlecap::event_bus::Event; +use bottlecap::extension::telemetry::events::{TelemetryEvent, TelemetryRecord}; +use bottlecap::extension::telemetry::stitch::{FragmentBuffer, Stitch}; +use bottlecap::logs::aggregator::Aggregator; +use bottlecap::logs::aggregator_service::{AggregatorHandle, AggregatorService}; +use bottlecap::logs::lambda::processor::{LambdaProcessor, is_oom_error}; +use bottlecap::logs::lambda::{IntakeLog, Lambda, Message}; +use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use fixtures::invocation::{BASE_NS, SERVICE, log_millis, millis_to_ns}; +use fixtures::logs; +use fixtures::telemetry::{self, LogKind, REQUEST_ID}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use serde_json::value::RawValue; + +/// Events per POST body, the Telemetry API `maxItems`. +const LINES: usize = 1000; + +/// The `timestamp` of the generated intake logs: the first function log line, in milliseconds. +fn log_timestamp_ms() -> i64 { + (BASE_NS + millis_to_ns(log_millis(0))) / 1_000_000 +} + +fn parse_events(body: &[u8]) -> Vec { + serde_json::from_slice(body).expect("fixture body parses") +} + +fn message_of(event: &TelemetryEvent) -> String { + match &event.record { + TelemetryRecord::Function(Value::Object(o)) => { + serde_json::to_string(o).expect("serialize object") + } + TelemetryRecord::Function(Value::String(s)) => s.clone(), + other => panic!("unexpected record {other:?}"), + } +} + +/// A processor in the state it has after `platform.initStart` and `platform.start`, with the +/// aggregator service running on `rt`. +struct Pipeline { + processor: LambdaProcessor, + handle: AggregatorHandle, + tags: String, + arn: String, + /// `process()` awaits `event_bus.send` for platform and OOM events; drain it between + /// iterations (see `drain_event_bus`) so a full channel never stalls the bench. + event_bus_rx: tokio::sync::mpsc::Receiver, +} + +impl Pipeline { + fn drain_event_bus(&mut self) { + while self.event_bus_rx.try_recv().is_ok() {} + } +} + +fn pipeline(rt: &tokio::runtime::Runtime) -> Pipeline { + let config = common::config(); + let provider = common::provider(config.clone()); + let (tx, rx) = tokio::sync::mpsc::channel::(1024); + let mut processor = LambdaProcessor::new(provider.clone(), config, tx, false); + let (service, handle) = AggregatorService::default(); + rt.spawn(service.run()); + rt.block_on(async { + for body in [ + telemetry::init_start(), + telemetry::platform_start(REQUEST_ID), + ] { + let event: TelemetryEvent = serde_json::from_str(&body).expect("platform event"); + processor.process(event, &handle).await; + } + let _ = handle.get_batches().await; + }); + let mut pipeline = Pipeline { + processor, + handle, + tags: provider.get_tags_string(), + arn: provider.get_canonical_id().unwrap_or_default(), + event_bus_rx: rx, + }; + pipeline.drain_event_bus(); + pipeline +} + +/// `map_log_level_to_status` in `src/logs/lambda/processor.rs`. +fn map_log_level_to_status(level: &str) -> Option<&'static str> { + match level.to_uppercase().as_str() { + "FATAL" | "CRITICAL" => Some("critical"), + "ERROR" | "ERR" => Some("error"), + "WARN" | "WARNING" => Some("warn"), + "INFO" | "INFORMATION" => Some("info"), + "DEBUG" => Some("debug"), + "TRACE" => Some("trace"), + _ => None, + } +} + +// --------------------------------------------------------------------------------------------- +// Listener parse (`Listener::handle` in src/extension/telemetry/listener.rs) and the alternatives from the review. +// --------------------------------------------------------------------------------------------- + +/// Non-flatten event with a `Value` record (same tree, no `#[serde(flatten)]`). +#[derive(Deserialize)] +struct PlainEvent { + #[allow(dead_code)] + time: chrono::DateTime, + #[serde(rename = "type")] + #[allow(dead_code)] + kind: String, + #[allow(dead_code)] + record: Value, +} + +/// Event whose record stays as raw bytes (the 5.5 prototype). +#[derive(Deserialize)] +struct RawEvent<'a> { + #[allow(dead_code)] + time: chrono::DateTime, + #[serde(rename = "type", borrow)] + #[allow(dead_code)] + kind: &'a str, + #[serde(borrow)] + record: &'a RawValue, +} + +fn listener_parse(c: &mut Criterion) { + let mut g = c.benchmark_group("logs/listener_parse"); + g.throughput(Throughput::Elements(LINES as u64)); + for kind in LogKind::ALL { + let body = telemetry::log_batch(kind, LINES); + g.bench_with_input( + BenchmarkId::new("flatten_telemetry_event", kind.name()), + &body, + |b, body| { + b.iter(|| parse_events(body)); + }, + ); + g.bench_with_input( + BenchmarkId::new("plain_struct_value_record", kind.name()), + &body, + |b, body| { + b.iter(|| serde_json::from_slice::>(body).expect("parse")); + }, + ); + g.bench_with_input( + BenchmarkId::new("raw_value_record", kind.name()), + &body, + |b, body| { + b.iter(|| serde_json::from_slice::>(body).expect("parse")); + }, + ); + } + g.finish(); +} + +// --------------------------------------------------------------------------------------------- +// Processor: LambdaProcessor::process per line, and the operations inside it. +// --------------------------------------------------------------------------------------------- + +fn processor_process(c: &mut Criterion) { + common::setup_env(); + let rt = common::runtime(); + let mut p = pipeline(&rt); + + let mut g = c.benchmark_group("logs/processor_process"); + g.throughput(Throughput::Elements(LINES as u64)); + for kind in LogKind::ALL { + let events = parse_events(&telemetry::log_batch(kind, LINES)); + g.bench_function(BenchmarkId::new("per_line", kind.name()), |b| { + b.iter_custom(|iters| { + let mut total = Duration::ZERO; + for _ in 0..iters { + let batch = events.clone(); + rt.block_on(async { + let start = Instant::now(); + for event in batch { + p.processor.process(event, &p.handle).await; + } + total += start.elapsed(); + // Keep the aggregator from growing across iterations. + let _ = p.handle.get_batches().await; + }); + p.drain_event_bus(); + } + total + }); + }); + } + g.finish(); +} + +/// Borrowed intake log: the lower bound for one serialization with no per-line String clones. +#[derive(Serialize)] +struct LambdaRef<'a> { + arn: &'a str, + request_id: Option<&'a str>, +} +#[derive(Serialize)] +struct MessageRef<'a> { + message: &'a str, + lambda: LambdaRef<'a>, + timestamp: i64, + status: &'a str, +} +#[derive(Serialize)] +struct IntakeLogRef<'a> { + message: MessageRef<'a>, + hostname: &'a str, + service: &'a str, + ddtags: &'a str, + ddsource: &'a str, +} + +/// The fields `get_intake_log` reads from a JSON line, borrowed. +#[derive(Deserialize)] +struct Fields<'a> { + #[serde(rename = "requestId", borrow, default)] + request_id: Option>, + #[serde(borrow, default)] + level: Option>, +} + +fn intake_log(message: &str, arn: &str, tags: &str) -> IntakeLog { + IntakeLog { + message: Message { + message: message.to_string(), + lambda: Lambda { + arn: arn.to_string(), + request_id: Some(REQUEST_ID.to_string()), + ..Lambda::default() + }, + timestamp: log_timestamp_ms(), + status: "info".to_string(), + }, + hostname: arn.to_string(), + service: SERVICE.to_string(), + tags: tags.to_string(), + source: "lambda".to_string(), + } +} + +fn processor_parts(c: &mut Criterion) { + common::setup_env(); + let rt = common::runtime(); + let p = pipeline(&rt); + let (arn, tags) = (p.arn.clone(), p.tags.clone()); + + let mut g = c.benchmark_group("logs/processor_parts"); + g.throughput(Throughput::Elements(LINES as u64)); + for kind in LogKind::ALL { + let body = telemetry::log_batch(kind, LINES); + let events = parse_events(&body); + let messages: Vec = events.iter().map(message_of).collect(); + let intake: Vec = messages + .iter() + .map(|m| intake_log(m, &arn, &tags)) + .collect(); + + // `get_message`: `let copy = event.clone()`. + g.bench_function(BenchmarkId::new("event_clone", kind.name()), |b| { + b.iter(|| { + events.iter().fold(0usize, |n, e| { + std::hint::black_box(e.clone()); + n + 1 + }) + }); + }); + // `get_message`: `serde_json::to_string(&obj)` for object records. + if kind == LogKind::JsonObject { + g.bench_function( + BenchmarkId::new("to_string_object_record", kind.name()), + |b| { + b.iter(|| { + events + .iter() + .map(|e| std::hint::black_box(message_of(e)).len()) + .sum::() + }); + }, + ); + } + // `get_intake_log`: `serde_json::from_str::(message)`. + g.bench_function( + BenchmarkId::new("reparse_message_as_value", kind.name()), + |b| { + b.iter(|| { + messages + .iter() + .filter(|m| serde_json::from_str::(m).is_ok()) + .count() + }); + }, + ); + // get_intake_log: `message.clone()` plus the tags, arn, service and source clones. + g.bench_function(BenchmarkId::new("intake_field_clones", kind.name()), |b| { + b.iter(|| { + messages.iter().fold(0usize, |acc, m| { + let fields = std::hint::black_box(( + m.clone(), + arn.clone(), + arn.clone(), + tags.clone(), + SERVICE.to_string(), + "lambda".to_string(), + )); + acc + fields.0.len() + }) + }); + }); + // aggregator insert: `serde_json::to_string(&IntakeLog)` per line. + g.bench_function(BenchmarkId::new("serialize_intake_log", kind.name()), |b| { + b.iter(|| { + intake + .iter() + .map(|l| serde_json::to_string(l).expect("serialize").len()) + .sum::() + }); + }); + // Lower bound: extract from the parsed tree, borrow every field, map the status, run + // the OOM scan, serialize once. It skips the processing rules, the durable-function + // handling and the aggregator insert, which production keeps. + g.bench_function( + BenchmarkId::new("alt_borrowed_single_serialize", kind.name()), + |b| { + b.iter_batched( + || events.clone(), + |events| { + let mut out = 0usize; + let mut oom = 0usize; + for e in events { + let (msg, rid, status) = match e.record { + TelemetryRecord::Function(Value::Object(o)) => { + let rid = o + .get("requestId") + .and_then(Value::as_str) + .map(str::to_owned); + let st = + o.get("level").and_then(Value::as_str).map(str::to_owned); + (serde_json::to_string(&o).expect("serialize"), rid, st) + } + TelemetryRecord::Function(Value::String(s)) => { + if s.starts_with('{') { + let f: Option = serde_json::from_str(&s).ok(); + let (rid, st) = f + .map(|f| { + ( + f.request_id.map(|c| c.into_owned()), + f.level.map(|c| c.into_owned()), + ) + }) + .unwrap_or((None, None)); + (s, rid, st) + } else { + (s, None, None) + } + } + _ => unreachable!(), + }; + oom += usize::from(is_oom_error(&msg)); + let log = IntakeLogRef { + message: MessageRef { + message: &msg, + lambda: LambdaRef { + arn: &arn, + request_id: rid.as_deref().or(Some(REQUEST_ID)), + }, + timestamp: e.time.timestamp_millis(), + status: status + .as_deref() + .and_then(map_log_level_to_status) + .unwrap_or("info"), + }, + hostname: &arn, + service: SERVICE, + ddtags: &tags, + ddsource: "lambda", + }; + out += serde_json::to_string(&log).expect("serialize").len(); + } + (out, oom) + }, + BatchSize::LargeInput, + ); + }, + ); + // 5.5 prototype, same lower bound: RawValue parse + borrowed field extraction + status + // map + OOM scan + one serialization. + g.bench_function( + BenchmarkId::new("alt_raw_value_end_to_end", kind.name()), + |b| { + b.iter(|| { + let mut out = 0usize; + let mut oom = 0usize; + let parsed: Vec = serde_json::from_slice(&body).expect("parse"); + for e in parsed { + let raw = e.record.get(); + let (msg, rid, status): ( + std::borrow::Cow, + Option, + Option, + ) = if raw.starts_with('{') { + let f: Fields = serde_json::from_str(raw).expect("fields"); + ( + std::borrow::Cow::Borrowed(raw), + f.request_id.map(|c| c.into_owned()), + f.level.map(|c| c.into_owned()), + ) + } else { + let s: String = serde_json::from_str(raw).expect("string record"); + if s.starts_with('{') { + let f: Option = serde_json::from_str(&s).ok(); + let (rid, st) = f + .map(|f| { + ( + f.request_id.map(|c| c.into_owned()), + f.level.map(|c| c.into_owned()), + ) + }) + .unwrap_or((None, None)); + (std::borrow::Cow::Owned(s), rid, st) + } else { + (std::borrow::Cow::Owned(s), None, None) + } + }; + oom += usize::from(is_oom_error(&msg)); + let log = IntakeLogRef { + message: MessageRef { + message: &msg, + lambda: LambdaRef { + arn: &arn, + request_id: rid.as_deref().or(Some(REQUEST_ID)), + }, + timestamp: e.time.timestamp_millis(), + status: status + .as_deref() + .and_then(map_log_level_to_status) + .unwrap_or("info"), + }, + hostname: &arn, + service: SERVICE, + ddtags: &tags, + ddsource: "lambda", + }; + out += serde_json::to_string(&log).expect("serialize").len(); + } + (out, oom) + }); + }, + ); + // 5.7: serialize only the per-line fields, append the fixed suffix once computed. + let suffix = { + let mut s = String::from(",\"hostname\":"); + s.push_str(&serde_json::to_string(&arn).expect("arn")); + s.push_str(",\"service\":"); + s.push_str(&serde_json::to_string(SERVICE).expect("service")); + s.push_str(",\"ddtags\":"); + s.push_str(&serde_json::to_string(&tags).expect("tags")); + s.push_str(",\"ddsource\":\"lambda\"}"); + s + }; + let arn_json = serde_json::to_string(&arn).expect("arn"); + let message_tail = format!( + "\"}},\"timestamp\":{},\"status\":\"info\"}}", + log_timestamp_ms() + ); + let prebuilt = |m: &str| -> Vec { + let mut out = Vec::with_capacity(m.len() + suffix.len() + 160); + out.extend_from_slice(b"{\"message\":{\"message\":"); + serde_json::to_writer(&mut out, m).expect("message"); + out.extend_from_slice(b",\"lambda\":{\"arn\":"); + out.extend_from_slice(arn_json.as_bytes()); + out.extend_from_slice(b",\"request_id\":\""); + out.extend_from_slice(REQUEST_ID.as_bytes()); + out.extend_from_slice(message_tail.as_bytes()); + out.extend_from_slice(suffix.as_bytes()); + out + }; + assert_eq!( + String::from_utf8(prebuilt(&messages[0])).expect("utf-8"), + serde_json::to_string(&intake[0]).expect("serialize"), + "the prebuilt suffix must give the same bytes as serde" + ); + g.bench_function( + BenchmarkId::new("alt_prebuilt_suffix_serialize", kind.name()), + |b| { + b.iter(|| messages.iter().map(|m| prebuilt(m).len()).sum::()); + }, + ); + } + + // `process()`: one `Vec` and one unbounded channel send per line. + g.bench_function("per_line_vec_and_unbounded_send", |b| { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::>(); + let lines: Vec = (0..LINES).map(|i| i.to_string()).collect(); + b.iter_batched( + || lines.clone(), + |lines| { + for s in lines { + tx.send(vec![s]).expect("send"); + } + while rx.try_recv().is_ok() {} + }, + BatchSize::LargeInput, + ); + }); + g.finish(); +} + +// --------------------------------------------------------------------------------------------- +// OOM check: 7 substring scans per line (`is_oom_error`) vs one regex alternation (5.6). +// --------------------------------------------------------------------------------------------- + +/// `OOM_ERRORS` in `src/logs/lambda/processor.rs`, the needles `is_oom_error` searches for. +const OOM_NEEDLES: [&str; 7] = [ + "fatal error: runtime: out of memory", + "java.lang.OutOfMemoryError", + "JavaScript heap out of memory", + "Runtime exited with error: signal: killed", + "MemoryError", + "failed to allocate memory (NoMemoryError)", + "OutOfMemoryException", +]; + +fn oom_check(c: &mut Criterion) { + let mut g = c.benchmark_group("logs/oom_check"); + g.throughput(Throughput::Elements(LINES as u64)); + let regex = regex::Regex::new( + &OOM_NEEDLES + .iter() + .map(|n| regex::escape(n)) + .collect::>() + .join("|"), + ) + .expect("oom regex"); + // The alternative must agree with `is_oom_error` on every hit and on 1000 clean lines. + for line in logs::OOM_LINES { + assert!(is_oom_error(line) && regex.is_match(line), "{line}"); + } + for i in 0..LINES { + let line = logs::text_line(i); + assert!(!is_oom_error(&line) && !regex.is_match(&line), "{line}"); + } + for len in [200usize, 2000] { + let lines: Vec = (0..LINES).map(|i| logs::text_line_of_len(i, len)).collect(); + g.bench_with_input( + BenchmarkId::new("is_oom_error_7_contains", len), + &lines, + |b, lines| { + b.iter(|| lines.iter().filter(|l| is_oom_error(l)).count()); + }, + ); + g.bench_with_input( + BenchmarkId::new("alt_single_regex", len), + &lines, + |b, lines| { + b.iter(|| lines.iter().filter(|l| regex.is_match(l)).count()); + }, + ); + } + let hits: Vec = (0..LINES) + .map(|i| logs::OOM_LINES[i % logs::OOM_LINES.len()].to_string()) + .collect(); + g.bench_function("is_oom_error_7_contains/hit", |b| { + b.iter(|| hits.iter().filter(|l| is_oom_error(l)).count()); + }); + g.finish(); +} + +// --------------------------------------------------------------------------------------------- +// Stitching split records (src/extension/telemetry/stitch.rs). Review 5.14, 2.2. +// --------------------------------------------------------------------------------------------- + +fn stitch(c: &mut Criterion) { + let mut g = c.benchmark_group("logs/stitch"); + g.sample_size(20); + let piece = 256 << 10; + for (label, total) in [("1mib", 1usize << 20), ("4mib", 4usize << 20)] { + let single = telemetry::split_record_single_post(total, piece); + g.throughput(Throughput::Bytes(single.len() as u64)); + // The listener tries a full parse first (`Listener::handle`); it fails on a split body. + g.bench_with_input( + BenchmarkId::new("failed_full_parse", label), + &single, + |b, body| { + b.iter(|| serde_json::from_slice::>(body).is_err()); + }, + ); + g.bench_with_input( + BenchmarkId::new("single_post_seams", label), + &single, + |b, body| { + b.iter(|| match FragmentBuffer::default().stitch(body) { + Stitch::Complete(events) => events.len(), + other => panic!("expected complete, got {other:?}"), + }); + }, + ); + let posts = telemetry::split_record_posts(total, piece, 6); + g.bench_with_input( + BenchmarkId::new("across_posts", label), + &posts, + |b, posts| { + b.iter(|| { + let fragments = FragmentBuffer::default(); + let mut events = 0; + for post in posts { + if let Stitch::Complete(e) = fragments.stitch(post) { + events += e.len(); + } + } + events + }); + }, + ); + } + g.finish(); +} + +// --------------------------------------------------------------------------------------------- +// Aggregator batch build (src/logs/aggregator.rs) and flush compression (`LogsFlusher::encode`). +// --------------------------------------------------------------------------------------------- + +fn intake_lines(kind: LogKind, n: usize, arn: &str, tags: &str) -> Vec { + (0..n) + .map(|i| { + let message = match kind { + LogKind::Text => logs::text_line(i), + _ => logs::json_line(i), + }; + serde_json::to_string(&intake_log(&message, arn, tags)).expect("serialize") + }) + .collect() +} + +fn drain(agg: &mut Aggregator) -> Vec> { + let mut batches = Vec::new(); + loop { + let batch = agg.get_batch(); + if batch.is_empty() { + break; + } + batches.push(batch); + } + batches +} + +fn aggregator(c: &mut Criterion) { + common::setup_env(); + let config = common::config(); + let provider = common::provider(config); + let (arn, tags) = ( + provider.get_canonical_id().unwrap_or_default(), + provider.get_tags_string(), + ); + const N: usize = 10_000; + + let mut g = c.benchmark_group("logs/aggregator"); + g.throughput(Throughput::Elements(N as u64)); + for kind in [LogKind::Text, LogKind::JsonObject] { + let lines = intake_lines(kind, N, &arn, &tags); + g.bench_function( + BenchmarkId::new("add_batch_and_get_batches", kind.name()), + |b| { + b.iter_batched( + || lines.clone(), + |lines| { + let mut agg = Aggregator::default(); + agg.add_batch(lines); + drain(&mut agg).len() + }, + BatchSize::LargeInput, + ); + }, + ); + } + g.finish(); +} + +fn stream_encode(data: &[u8], level: i32) -> usize { + // `LogsFlusher::encode` in src/logs/flusher.rs + let mut encoder = zstd::stream::write::Encoder::new(Vec::new(), level).expect("encoder"); + encoder.write_all(data).expect("write"); + encoder.finish().expect("finish").len() +} + +fn compress(c: &mut Criterion) { + common::setup_env(); + let config = common::config(); + let provider = common::provider(config.clone()); + let (arn, tags) = ( + provider.get_canonical_id().unwrap_or_default(), + provider.get_tags_string(), + ); + let level = config.logs_config_compression_level; + + let mut g = c.benchmark_group("logs/compress"); + g.sample_size(20); + let mut inputs: Vec<(String, Vec)> = Vec::new(); + for kind in [LogKind::Text, LogKind::JsonObject] { + let mut agg = Aggregator::default(); + agg.add_batch(intake_lines(kind, 5000, &arn, &tags)); + let batch = drain(&mut agg).remove(0); + inputs.push((format!("{}_batch", kind.name()), batch)); + } + // Real logs compress far less than synthetic ones: a batch of pseudo-random hex words. + let noisy: Vec = (0..1000) + .map(|i| { + let mut x: u64 = 0x9E37_79B9_7F4A_7C15u64.wrapping_mul(i as u64 + 1); + let mut msg = String::with_capacity(1600); + while msg.len() < 1500 { + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + msg.push_str(&format!("{x:x} ")); + } + serde_json::to_string(&intake_log(&msg, &arn, &tags)).expect("serialize") + }) + .collect(); + let mut agg = Aggregator::default(); + agg.add_batch(noisy); + inputs.push(("noisy_batch".to_string(), drain(&mut agg).remove(0))); + + for (label, data) in &inputs { + g.throughput(Throughput::Bytes(data.len() as u64)); + println!( + "logs/compress {label}: {} B raw, zstd level {level} (config default) -> {} B", + data.len(), + stream_encode(data, level) + ); + g.bench_with_input( + BenchmarkId::new("zstd_stream_config_level", label), + data, + |b, data| { + b.iter(|| stream_encode(data, level)); + }, + ); + g.bench_with_input( + BenchmarkId::new("zstd_stream_l3", label), + data, + |b, data| { + b.iter(|| stream_encode(data, 3)); + }, + ); + g.bench_with_input( + BenchmarkId::new("zstd_stream_l1", label), + data, + |b, data| { + b.iter(|| stream_encode(data, 1)); + }, + ); + g.bench_with_input( + BenchmarkId::new("alt_zstd_bulk_l3_reused_ctx", label), + data, + |b, data| { + let mut compressor = zstd::bulk::Compressor::new(3).expect("compressor"); + b.iter(|| compressor.compress(data).expect("compress").len()); + }, + ); + } + g.finish(); +} + +criterion_group! { + name = benches; + config = common::criterion(); + targets = listener_parse, processor_process, processor_parts, oom_check, stitch, aggregator, compress +} +criterion_main!(benches); diff --git a/bottlecap/benches/traces.rs b/bottlecap/benches/traces.rs new file mode 100644 index 000000000..e6a2c09e7 --- /dev/null +++ b/bottlecap/benches/traces.rs @@ -0,0 +1,601 @@ +//! Traces hot path on a 2000-span v0.4 POST: decode, dedup, `process_traces`, tag injection, +//! the stats copy, and the flush encode and compress. +//! +//! Review items: 5.1, 5.2, 5.3, 5.4, 5.11, 5.16, 5.17. + +mod common; +#[path = "../tests/fixtures/mod.rs"] +mod fixtures; + +use std::collections::HashMap; +use std::hint::black_box; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use bottlecap::config::Config; +use bottlecap::traces::span_dedup::{DedupKey, Deduper}; +use bottlecap::traces::span_dedup_service::DedupService; +use bottlecap::traces::stats_concentrator_service::StatsConcentratorService; +use bottlecap::traces::stats_generator::StatsGenerator; +use bottlecap::traces::trace_aggregator::SendDataBuilderInfo; +use bottlecap::traces::trace_processor::{self, ServerlessTraceProcessor, TraceProcessor}; +use criterion::measurement::WallTime; +use criterion::{ + BatchSize, BenchmarkGroup, Criterion, Throughput, criterion_group, criterion_main, +}; +use fixtures::invocation::SERVICE; +use libdd_capabilities::MaybeSend; +use libdd_capabilities::http::{HttpClientCapability, HttpError}; +use libdd_capabilities::sleep::SleepCapability; +use libdd_common::http_common; +use libdd_trace_obfuscation::obfuscate::obfuscate_pb_span; +use libdd_trace_obfuscation::obfuscation_config::{HttpConfig, ObfuscationConfig}; +use libdd_trace_protobuf::pb; +use libdd_trace_utils::send_with_retry::{RetryBackoffType, RetryStrategy}; +use libdd_trace_utils::trace_utils; +use libdd_trace_utils::tracer_payload::TracerPayloadCollection; +use prost::Message; + +const SPANS: u64 = 2000; +const N_TRACES: usize = 100; + +fn group(c: &mut Criterion) -> BenchmarkGroup<'_, WallTime> { + let mut g = c.benchmark_group("traces"); + g.throughput(Throughput::Elements(SPANS)); + g +} + +fn make_config(edit: impl FnOnce(&mut Config)) -> Arc { + let mut config = (*common::config()).clone(); + edit(&mut config); + Arc::new(config) +} + +/// Built as `build_trace_agent` in `src/startup.rs` does, error sampler off. +fn processor(config: &Config) -> ServerlessTraceProcessor { + ServerlessTraceProcessor { + obfuscation_config: Arc::new(ObfuscationConfig { + tag_replace_rules: bottlecap::startup::build_tag_replace_rules(config), + http: HttpConfig { + remove_path_digits: config.apm_config_obfuscation_http_remove_paths_with_digits, + remove_query_string: config.apm_config_obfuscation_http_remove_query_string, + }, + ..Default::default() + }), + error_sampler: trace_processor::new_error_sampler(false), + error_rare_sample_tracer_drop: false, + } +} + +/// One untimed `process_traces` call, for setup. +fn process( + config: &Arc, + traces: Vec>, + body_len: usize, +) -> (Option, TracerPayloadCollection) { + processor(config).process_traces( + config.clone(), + common::provider(config.clone()), + fixtures::traces::header_tags(), + traces, + body_len, + None, + ) +} + +fn decode(c: &mut Criterion) { + let rt = common::runtime(); + let body = bytes::Bytes::from(fixtures::traces::v04_2000_spans()); + let mut g = group(c); + + // `TraceAgent::handle_traces` in src/traces/trace_agent.rs + g.bench_function("decode_v04/libdd_from_request_body", |b| { + b.to_async(&rt).iter_with_large_drop(|| async { + trace_utils::get_traces_from_request_body(http_common::Body::from_bytes(body.clone())) + .await + .expect("decode v0.4") + }); + }); + g.bench_function("decode_v04/libdd_from_slice", |b| { + b.iter_with_large_drop(|| { + libdd_trace_utils::msgpack_decoder::v04::from_slice(&body).expect("decode v0.4") + }); + }); + g.bench_function("decode_v04/libdd_from_slice_to_owned", |b| { + b.iter_with_large_drop(|| { + let (traces, _) = + libdd_trace_utils::msgpack_decoder::v04::from_slice(&body).expect("decode v0.4"); + traces + .into_iter() + .map(|trace| { + trace + .into_iter() + .map(|s| pb::Span { + service: s.service.to_string(), + name: s.name.to_string(), + resource: s.resource.to_string(), + #[allow(clippy::cast_possible_truncation)] + trace_id: s.trace_id as u64, + span_id: s.span_id, + parent_id: s.parent_id, + start: s.start, + duration: s.duration, + error: s.error, + meta: s + .meta + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(), + metrics: s.metrics.iter().map(|(k, v)| (k.to_string(), *v)).collect(), + r#type: s.r#type.to_string(), + ..Default::default() + }) + .collect::>() + }) + .collect::>() + }); + }); + g.bench_function("decode_v04/rmp_serde_from_slice", |b| { + b.iter_with_large_drop(|| { + rmp_serde::from_slice::>>(&body).expect("decode v0.4") + }); + }); + g.finish(); +} + +fn dedup(c: &mut Criterion) { + let rt = common::runtime(); + let traces = fixtures::traces::trace_2000_spans(); + let keys: Arc> = Arc::new( + traces + .iter() + .flatten() + .map(|s| (s.trace_id, s.span_id)) + .collect(), + ); + + // Review 5.16: the window holds 100 keys, so a resend of the same payload slips through. + let mut deduper = Deduper::new(100); + for &(trace_id, span_id) in keys.iter() { + deduper.check_and_add(DedupKey::new(trace_id, span_id)); + } + let caught = keys + .iter() + .filter(|&&(trace_id, span_id)| !deduper.check_and_add(DedupKey::new(trace_id, span_id))) + .count(); + println!( + "dedup: resend of the same {SPANS}-span payload, {caught} duplicates caught (capacity 100)" + ); + + let mut g = group(c); + // One round trip per span, as `TraceAgent::handle_traces` does. Keys change every round so + // every check is a miss. + for (name, timeout) in [ + ("no_timeout", None), + ("timeout_5s", Some(Duration::from_secs(5))), + ] { + let (service, handle) = DedupService::new(); + rt.spawn(service.run()); + let mut round = 0u64; + g.bench_function(format!("dedup/service_round_trip/{name}"), |b| { + b.iter_custom(|iters| { + let first = round + 1; + round += iters; + let (handle, keys) = (handle.clone(), keys.clone()); + rt.block_on(rt.spawn(async move { + let start = Instant::now(); + for r in first..first + iters { + for &(trace_id, span_id) in keys.iter() { + let key = DedupKey::new(trace_id + r * 1_000_000_000, span_id); + black_box(handle.check_and_add(key, timeout).await.ok()); + } + } + start.elapsed() + })) + .expect("dedup task") + }); + }); + } + + let inline = std::sync::Mutex::new(Deduper::new(100)); + let mut round = 0u64; + g.bench_function("dedup/inline_mutex", |b| { + b.iter(|| { + round += 1; + let mut deduper = inline.lock().expect("dedup lock"); + for &(trace_id, span_id) in keys.iter() { + let key = DedupKey::new(trace_id + round * 1_000_000_000, span_id); + black_box(deduper.check_and_add(key)); + } + }); + }); + g.finish(); +} + +fn process_traces(c: &mut Criterion) { + common::setup_env(); + let traces = fixtures::traces::trace_2000_spans(); + let body_len = fixtures::traces::v04_msgpack(&traces).len(); + let variants = [ + ("default", make_config(|_| {})), + // Review 5.3: each regex compiles once per root span (`span_matches_tag_regex` in trace_processor.rs). + // The require filter matches the fixture service: every chunk is kept, both filters run. + ( + "regex_filters_2", + make_config(|c| { + c.apm_filter_tags_regex_require = Some(vec![format!("service:^{SERVICE}$")]); + c.apm_filter_tags_regex_reject = Some(vec!["http.url:.*healthcheck.*".to_string()]); + }), + ), + ( + "compute_stats", + make_config(|c| c.ext.lambda_extension_compute_stats = true), + ), + ]; + + let mut g = group(c); + for (name, config) in variants { + let provider = common::provider(config.clone()); + let processor = processor(&config); + let (info, stats) = process(&config, traces.clone(), body_len); + assert!(info.is_some(), "{name}: every chunk was dropped"); + let TracerPayloadCollection::V07(payloads) = stats else { + unreachable!("process_traces returns V07") + }; + let kept: usize = payloads + .iter() + .flat_map(|tp| &tp.chunks) + .map(|chunk| chunk.spans.len()) + .sum(); + assert_eq!(kept as u64, SPANS, "{name}: spans were filtered out"); + g.bench_function(format!("process_traces/{name}"), |b| { + b.iter_batched( + || traces.clone(), + |input| { + processor.process_traces( + config.clone(), + provider.clone(), + fixtures::traces::header_tags(), + input, + body_len, + None, + ) + }, + BatchSize::LargeInput, + ); + }); + } + g.finish(); +} + +fn parts(c: &mut Criterion) { + common::setup_env(); + let config = make_config(|_| {}); + let provider = common::provider(config.clone()); + let traces = fixtures::traces::trace_2000_spans(); + let body_len = fixtures::traces::v04_msgpack(&traces).len(); + let (_, stats) = process(&config, traces.clone(), body_len); + let TracerPayloadCollection::V07(stats_vec) = stats else { + unreachable!("process_traces returns V07") + }; + let tags = provider.get_tags_map().clone(); + println!("traces: {} tags injected into every span", tags.len()); + + let mut g = group(c); + // Review 5.2: the copy for stats generation in `process_traces`, dropped unused + // when compute_stats is off. + g.bench_function("stats_copy/clone", |b| { + b.iter_with_large_drop(|| stats_vec.clone()) + }); + g.bench_function("stats_copy/drop", |b| { + b.iter_batched(|| stats_vec.clone(), drop, BatchSize::LargeInput); + }); + // `process_traces` body size: prost `encoded_len` over every payload. + g.bench_function("body_size_encoded_len", |b| { + b.iter(|| stats_vec.iter().map(Message::encoded_len).sum::()); + }); + // `trace.to_vec()` in libdd `collect_pb_trace_chunks` . + g.bench_function("input_clone", |b| b.iter_with_large_drop(|| traces.clone())); + + // `ChunkProcessor::process` in src/traces/trace_processor.rs + g.bench_function("tag_injection", |b| { + b.iter_batched( + || traces.clone(), + |mut input| { + for span in input.iter_mut().flatten() { + tags.iter().for_each(|(k, v)| { + span.meta.insert(k.clone(), v.clone()); + }); + span.meta + .insert("_dd.compute_stats".to_string(), "1".to_string()); + span.meta.insert("origin".to_string(), "lambda".to_string()); + span.meta + .insert("_dd.origin".to_string(), "lambda".to_string()); + } + input + }, + BatchSize::LargeInput, + ); + }); + g.bench_function("tag_injection/reserve_get_mut", |b| { + b.iter_batched( + || traces.clone(), + |mut input| { + for span in input.iter_mut().flatten() { + span.meta.reserve(tags.len() + 3); + for (k, v) in &tags { + match span.meta.get_mut(k) { + Some(existing) => existing.clone_from(v), + None => { + span.meta.insert(k.clone(), v.clone()); + } + } + } + span.meta + .insert("_dd.compute_stats".to_string(), "1".to_string()); + span.meta.insert("origin".to_string(), "lambda".to_string()); + span.meta + .insert("_dd.origin".to_string(), "lambda".to_string()); + } + input + }, + BatchSize::LargeInput, + ); + }); + // Upper bound only: tags on the root span alone is a behavior change. + g.bench_function("tag_injection/root_span_only", |b| { + b.iter_batched( + || traces.clone(), + |mut input| { + for chunk in &mut input { + if let Some(root) = chunk.first_mut() { + tags.iter().for_each(|(k, v)| { + root.meta.insert(k.clone(), v.clone()); + }); + } + for span in chunk.iter_mut() { + span.meta + .insert("_dd.compute_stats".to_string(), "1".to_string()); + span.meta.insert("origin".to_string(), "lambda".to_string()); + span.meta + .insert("_dd.origin".to_string(), "lambda".to_string()); + } + } + input + }, + BatchSize::LargeInput, + ); + }); + + // The `obfuscate_pb_span` call in `ChunkProcessor::process`. + let obfuscation = processor(&config).obfuscation_config; + g.bench_function("obfuscate_span", |b| { + b.iter_batched( + || traces.clone(), + |mut input| { + for span in input.iter_mut().flatten() { + obfuscate_pb_span(span, &obfuscation); + } + input + }, + BatchSize::LargeInput, + ); + }); + // libdd `collect_pb_trace_chunks` . + g.bench_function("normalize_and_top_level", |b| { + b.iter_batched( + || traces.clone(), + |mut input| { + for trace in &mut input { + black_box(libdd_trace_normalization::normalizer::normalize_trace(trace).ok()); + trace_utils::compute_top_level_span(trace); + } + input + }, + BatchSize::LargeInput, + ); + }); + // 2 compiles per root span, not a per-span cost. + g.throughput(Throughput::Elements(2 * N_TRACES as u64)); + g.bench_function("regex_compile_per_root_span", |b| { + b.iter(compile_filters_per_root_span) + }); + g.finish(); +} + +/// Review 5.3: `span_matches_tag_regex` (trace_processor.rs) compiles its pattern on +/// every call, once per root span and filter. +#[allow(clippy::regex_creation_in_loops)] +fn compile_filters_per_root_span() { + for _ in 0..N_TRACES { + black_box(regex::Regex::new("^my-.*$").expect("regex")); + black_box(regex::Regex::new(".*healthcheck.*").expect("regex")); + } +} + +fn stats(c: &mut Criterion) { + common::setup_env(); + let rt = common::runtime(); + let config = make_config(|c| c.ext.lambda_extension_compute_stats = true); + // The concentrator keeps a time window and drops spans outside it, so these start now. + let traces = fixtures::traces::make_traces_at(fixtures::traces::now_ns(), N_TRACES, 20, true); + let body_len = fixtures::traces::v04_msgpack(&traces).len(); + let (_, stats) = process(&config, traces, body_len); + let stats = Arc::new(stats); + + let (service, handle) = StatsConcentratorService::new(config); + rt.spawn(service.run()); + let generator = Arc::new(StatsGenerator::new(handle.clone())); + + let mut g = group(c); + // Review 5.17: one boxed span clone and one channel send per span. + for (name, include_drain) in [("generator_send", false), ("send_and_drain", true)] { + g.bench_function(format!("stats/{name}"), |b| { + b.iter_custom(|iters| { + let (generator, handle, stats) = (generator.clone(), handle.clone(), stats.clone()); + rt.block_on(rt.spawn(async move { + let mut total = Duration::ZERO; + for _ in 0..iters { + let start = Instant::now(); + generator.send(&stats).expect("stats send"); + let sent = start.elapsed(); + // The flush reply comes after every queued add is aggregated. + handle.flush(false).await.expect("stats flush"); + total += if include_drain { start.elapsed() } else { sent }; + } + total + })) + .expect("stats task") + }); + }); + } + g.finish(); +} + +#[derive(Clone, Debug)] +struct FakeClient; + +impl HttpClientCapability for FakeClient { + fn new_client() -> Self { + FakeClient + } + + fn new_periodic() -> Self { + FakeClient + } + + fn request( + &self, + req: http::Request, + ) -> impl Future, HttpError>> + MaybeSend { + black_box(req.body().len()); + async { + Ok(http::Response::builder() + .status(200) + .body(bytes::Bytes::new()) + .expect("build response")) + } + } +} + +impl SleepCapability for FakeClient { + fn new() -> Self { + FakeClient + } + + fn sleep(&self, duration: Duration) -> impl Future + MaybeSend { + tokio::time::sleep(duration) + } +} + +/// `AgentPayload` encoded straight from the borrowed payloads; byte-identical to libdd's. +fn encode_agent_payload_borrowed(tps: &[pb::TracerPayload]) -> Vec { + use prost::encoding::{double, message}; + let len = tps + .iter() + .map(|tp| message::encoded_len(5, tp)) + .sum::() + + double::encoded_len(8, &60.0) + + double::encoded_len(9, &60.0); + let mut buf = Vec::with_capacity(len); + for tp in tps { + message::encode(5, tp, &mut buf); + } + double::encode(8, &60.0, &mut buf); + double::encode(9, &60.0, &mut buf); + buf +} + +/// libdd `construct_agent_payload`. +fn agent_payload(tracer_payloads: Vec) -> pb::AgentPayload { + pb::AgentPayload { + host_name: String::new(), + env: String::new(), + agent_version: String::new(), + error_tps: 60.0, + target_tps: 60.0, + tags: HashMap::new(), + tracer_payloads, + rare_sampler_enabled: false, + idx_tracer_payloads: Vec::new(), + } +} + +/// libdd `serialize_proto_payload`. +fn encode(agent: &pb::AgentPayload) -> Vec { + let mut buf = Vec::with_capacity(agent.encoded_len()); + agent.encode(&mut buf).expect("encode AgentPayload"); + buf +} + +fn flush(c: &mut Criterion) { + common::setup_env(); + let rt = common::runtime(); + let config = make_config(|_| {}); + let traces = fixtures::traces::trace_2000_spans(); + let body_len = fixtures::traces::v04_msgpack(&traces).len(); + let (info, _) = process(&config, traces, body_len); + // As `TraceFlusher::flush` in src/traces/trace_flusher.rs, with one attempt and no backoff. + let send_data = info + .expect("trace payload") + .builder + .with_api_key("0123456789abcdef0123456789abcdef") + .with_retry_strategy(RetryStrategy::new(1, 0, RetryBackoffType::Constant, None)) + .build(); + let TracerPayloadCollection::V07(tps) = send_data.get_payloads() else { + unreachable!("process_traces returns V07") + }; + let tps = tps.clone(); + let agent = agent_payload(tps.clone()); + let encoded = encode(&agent); + assert_eq!( + encode_agent_payload_borrowed(&tps), + encoded, + "borrowed encode must be byte-identical" + ); + let sent = rt.block_on(send_data.send(&FakeClient)); + assert_eq!(sent.requests_count, 1, "one request per send"); + println!( + "flush: protobuf {} bytes, zstd 1 {} bytes, zstd 3 {} bytes, on wire {} bytes (level {})", + encoded.len(), + zstd::bulk::compress(&encoded, 1).expect("zstd").len(), + zstd::bulk::compress(&encoded, 3).expect("zstd").len(), + sent.bytes_sent, + config.apm_config_compression_level, + ); + + let mut g = group(c); + // Review 5.4, 5.11: payload clone, protobuf encode and zstd in libdd `send_with_protobuf`. + g.bench_function("flush/send_data_fake_http", |b| { + b.to_async(&rt) + .iter(|| async { send_data.send(&FakeClient).await }); + }); + // libdd `SendData::send_with_protobuf` + g.bench_function("flush/payload_to_vec", |b| { + b.iter_with_large_drop(|| tps.to_vec()) + }); + // From here on the unit of work is the encoded protobuf, not the spans. + g.throughput(Throughput::Bytes(encoded.len() as u64)); + g.bench_function("flush/prost_encode_agent_payload", |b| { + b.iter_with_large_drop(|| encode(&agent)); + }); + g.bench_function("flush/encode/current", |b| { + b.iter_with_large_drop(|| encode(&agent_payload(tps.to_vec()))); + }); + g.bench_function("flush/encode/borrowed_manual", |b| { + b.iter_with_large_drop(|| encode_agent_payload_borrowed(&tps)); + }); + for level in [1, 3] { + g.bench_function(format!("flush/zstd/{level}"), |b| { + b.iter_with_large_drop(|| zstd::bulk::compress(&encoded, level).expect("zstd")); + }); + } + g.finish(); +} + +criterion_group! { + name = benches; + config = common::criterion(); + targets = decode, dedup, process_traces, parts, stats, flush +} +criterion_main!(benches); diff --git a/bottlecap/src/logs/lambda/processor.rs b/bottlecap/src/logs/lambda/processor.rs index e795f2ad8..45c7cb0e9 100644 --- a/bottlecap/src/logs/lambda/processor.rs +++ b/bottlecap/src/logs/lambda/processor.rs @@ -79,7 +79,9 @@ const OOM_ERRORS: [&str; 7] = [ "OutOfMemoryException", // .NET ]; -fn is_oom_error(error_msg: &str) -> bool { +#[doc(hidden)] +#[must_use] +pub fn is_oom_error(error_msg: &str) -> bool { OOM_ERRORS .iter() .any(|&oom_str| error_msg.contains(oom_str)) diff --git a/bottlecap/src/proc/mod.rs b/bottlecap/src/proc/mod.rs index 30dc43e5b..f630380e8 100644 --- a/bottlecap/src/proc/mod.rs +++ b/bottlecap/src/proc/mod.rs @@ -52,7 +52,8 @@ pub fn get_network_data() -> Result { get_network_data_from_path(PROC_NET_DEV_PATH) } -fn get_network_data_from_path(path: &str) -> Result { +#[doc(hidden)] +pub fn get_network_data_from_path(path: &str) -> Result { let file = File::open(path)?; let reader = io::BufReader::new(file); @@ -107,7 +108,8 @@ pub fn get_cpu_data() -> Result { get_cpu_data_from_path(PROC_STAT_PATH) } -fn get_cpu_data_from_path(path: &str) -> Result { +#[doc(hidden)] +pub fn get_cpu_data_from_path(path: &str) -> Result { let file = File::open(path)?; let reader = io::BufReader::new(file); @@ -185,7 +187,8 @@ pub fn get_uptime() -> Result { get_uptime_from_path(PROC_UPTIME_PATH) } -fn get_uptime_from_path(path: &str) -> Result { +#[doc(hidden)] +pub fn get_uptime_from_path(path: &str) -> Result { let file = File::open(path)?; let reader = io::BufReader::new(file); @@ -220,7 +223,9 @@ pub fn get_fd_max_data(pids: &[i64]) -> f64 { get_fd_max_data_from_path(PROC_PATH, pids) } -fn get_fd_max_data_from_path(path: &str, pids: &[i64]) -> f64 { +#[doc(hidden)] +#[must_use] +pub fn get_fd_max_data_from_path(path: &str, pids: &[i64]) -> f64 { let mut fd_max = constants::LAMBDA_FILE_DESCRIPTORS_DEFAULT_LIMIT; // regex to capture the soft limit value (first numeric value after the title) let re = Regex::new(r"^Max open files\s+(\d+)").expect("Failed to create regex"); @@ -252,7 +257,9 @@ pub fn get_fd_use_data(pids: &[i64]) -> f64 { get_fd_use_data_from_path(PROC_PATH, pids) } -fn get_fd_use_data_from_path(path: &str, pids: &[i64]) -> f64 { +#[doc(hidden)] +#[must_use] +pub fn get_fd_use_data_from_path(path: &str, pids: &[i64]) -> f64 { let mut fd_use = 0; for &pid in pids { @@ -271,7 +278,9 @@ pub fn get_threads_max_data(pids: &[i64]) -> f64 { get_threads_max_data_from_path(PROC_PATH, pids) } -fn get_threads_max_data_from_path(path: &str, pids: &[i64]) -> f64 { +#[doc(hidden)] +#[must_use] +pub fn get_threads_max_data_from_path(path: &str, pids: &[i64]) -> f64 { let mut threads_max = constants::LAMBDA_EXECUTION_PROCESSES_DEFAULT_LIMIT; // regex to capture the soft limit value (first numeric value after the title) let re = Regex::new(r"^Max processes\s+(\d+)").expect("Failed to create regex"); @@ -302,7 +311,8 @@ pub fn get_threads_use_data(pids: &[i64]) -> Result { get_threads_use_data_from_path(PROC_PATH, pids) } -fn get_threads_use_data_from_path(path: &str, pids: &[i64]) -> Result { +#[doc(hidden)] +pub fn get_threads_use_data_from_path(path: &str, pids: &[i64]) -> Result { let mut threads_use = 0; for &pid in pids { diff --git a/bottlecap/src/startup.rs b/bottlecap/src/startup.rs index cf12b0939..05798858b 100644 --- a/bottlecap/src/startup.rs +++ b/bottlecap/src/startup.rs @@ -53,7 +53,9 @@ pub struct TraceAgentPipeline { pub trace_aggregator_handle: TraceAggregatorHandle, } -fn build_tag_replace_rules(config: &Config) -> Vec { +#[doc(hidden)] +#[must_use] +pub fn build_tag_replace_rules(config: &Config) -> Vec { config .apm_replace_tags .as_ref() diff --git a/scripts/bench-compare.sh b/scripts/bench-compare.sh new file mode 100755 index 000000000..82d7bdeee --- /dev/null +++ b/scripts/bench-compare.sh @@ -0,0 +1,152 @@ +#!/usr/bin/env bash +# Runs the bottlecap criterion suite with the production profile, saves the results as a named +# baseline, and compares it with another baseline. +# +# Usage: +# scripts/bench-compare.sh save [bench...] [-- ] +# Run the suite (or only the named benches: logs, traces, invocation, enhanced_metrics, +# config, init) and save the baseline . +# scripts/bench-compare.sh compare [critcmp args] +# Print a table of vs . Extra args go to critcmp (for example `-t 5` hides +# differences of 5% or less, `-f 'logs/.*'` filters by regex). +# scripts/bench-compare.sh export +# Export a baseline to one JSON file, so it survives `cargo clean` and can be compared +# from another machine with `critcmp old.json new.json`. +# scripts/bench-compare.sh list +# List the baselines critcmp knows about. +# +# Environment: +# CARGO_TARGET_DIR where cargo builds and where criterion keeps baselines +# (default: bottlecap/target). critcmp reads the same directory. +# BENCH_FAST=1 build the bench profile without LTO and with 16 codegen units. Fast to +# link, but NOT the production profile; use it only to check that a bench +# runs, never for a baseline. +# CRITCMP path to critcmp (default: critcmp on PATH). Install with `cargo install critcmp`. +# +# The bench profile inherits [profile.release] from bottlecap/Cargo.toml: opt-level "z", lto, +# codegen-units 1, panic abort (cargo forces unwind for bench binaries). That is the shipped +# profile, so the numbers are comparable with the extension. See bottlecap/benches/README.md +# for noise control. +set -euo pipefail + +here="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +crate="$here/bottlecap" +export CARGO_TARGET_DIR="${CARGO_TARGET_DIR:-$crate/target}" +# cargo and critcmp run from the crate, so a relative path must not resolve twice. +[[ "$CARGO_TARGET_DIR" == /* ]] || CARGO_TARGET_DIR="$PWD/$CARGO_TARGET_DIR" + +critcmp_bin="${CRITCMP:-}" +if [[ -z "$critcmp_bin" ]] && command -v critcmp >/dev/null 2>&1; then + critcmp_bin="critcmp" +fi + +usage() { + sed -n '2,29p' "${BASH_SOURCE[0]}" | sed 's/^# \{0,1\}//' + exit 1 +} + +need_critcmp() { + if [[ -z "$critcmp_bin" ]]; then + echo "critcmp not found. Install it with: cargo install critcmp" >&2 + echo "Or compare with criterion only: cargo bench -- --baseline " >&2 + exit 1 + fi +} + +# HEAD, plus a hash of the uncommitted diff when the tree has changes, so two baselines saved +# on the same commit before and after an edit are not reported as the same code. +git_note() { + local rev + rev="$(git -C "$here" rev-parse --short HEAD 2>/dev/null)" || { echo unknown; return; } + if [[ -n "$(git -C "$here" status --porcelain --untracked-files=no)" ]]; then + rev="$rev-dirty (diff $(git -C "$here" diff HEAD | git hash-object --stdin | cut -c1-12))" + fi + echo "$rev" +} + +# The bench profile, with every CARGO_PROFILE_BENCH_* override (BENCH_FAST sets two), so a +# baseline saved with, say, CARGO_PROFILE_BENCH_OPT_LEVEL=3 is not recorded as production. +profile_note() { + local overrides + overrides="$(env | grep '^CARGO_PROFILE_BENCH_' | sort | tr '\n' ' ' || true)" + if [[ -n "$overrides" ]]; then + echo "bench with ${overrides}(NOT production)" + else + echo 'bench = release (opt-level z, lto, codegen-units 1)' + fi +} + +machine_note() { + local os cpu load + os="$(uname -sm)" + if [[ "$(uname -s)" == "Darwin" ]]; then + cpu="$(sysctl -n machdep.cpu.brand_string 2>/dev/null || echo unknown)" + else + cpu="$(grep -m1 'model name' /proc/cpuinfo 2>/dev/null | cut -d: -f2- | sed 's/^ //' || echo unknown)" + fi + load="$(uptime | sed 's/.*load average[s]*: //')" + echo "machine: $os, $cpu; load average: $load; profile: $(profile_note)" +} + +cmd="${1:-}" +[[ -n "$cmd" ]] || usage +shift + +case "$cmd" in + save) + name="${1:-}" + [[ -n "$name" ]] || usage + shift + benches=() + extra=() + while [[ $# -gt 0 ]]; do + if [[ "$1" == "--" ]]; then + shift + extra=("$@") + break + fi + benches+=("--bench" "$1") + shift + done + if [[ "${BENCH_FAST:-0}" == 1 ]]; then + export CARGO_PROFILE_BENCH_LTO=false + export CARGO_PROFILE_BENCH_CODEGEN_UNITS=16 + fi + note="$(machine_note)" + echo "$note" + # `${arr[@]+"${arr[@]}"}` expands to nothing when the array is empty; a bare "${arr[@]}" + # is an unbound variable under `set -u` on bash 3.2 (/bin/bash on macOS). + (cd "$crate" && cargo bench ${benches[@]+"${benches[@]}"} -- --save-baseline "$name" ${extra[@]+"${extra[@]}"}) + mkdir -p "$CARGO_TARGET_DIR/criterion" + printf '%s\nsaved: %s\ngit: %s\n' "$note" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" \ + "$(git_note)" \ + > "$CARGO_TARGET_DIR/criterion/baseline-$name.txt" + echo "saved baseline '$name' under $CARGO_TARGET_DIR/criterion (notes in baseline-$name.txt)" + ;; + compare) + old="${1:-}" + new="${2:-}" + [[ -n "$old" && -n "$new" ]] || usage + shift 2 + need_critcmp + for n in "$old" "$new"; do + [[ -f "$CARGO_TARGET_DIR/criterion/baseline-$n.txt" ]] && { echo "== $n"; cat "$CARGO_TARGET_DIR/criterion/baseline-$n.txt"; } + done + (cd "$crate" && "$critcmp_bin" "$old" "$new" "$@") + ;; + export) + name="${1:-}" + out="${2:-}" + [[ -n "$name" && -n "$out" ]] || usage + need_critcmp + (cd "$crate" && "$critcmp_bin" --export "$name") > "$out" + echo "exported '$name' to $out" + ;; + list) + need_critcmp + (cd "$crate" && "$critcmp_bin" --baselines) + ;; + *) + usage + ;; +esac diff --git a/scripts/bench-docker-arm64.sh b/scripts/bench-docker-arm64.sh new file mode 100755 index 000000000..0388c1311 --- /dev/null +++ b/scripts/bench-docker-arm64.sh @@ -0,0 +1,101 @@ +#!/usr/bin/env bash +# Runs the bottlecap criterion suite on Linux arm64 inside Docker, the platform the extension +# ships on. Lambda arm64 is Graviton; an Apple Silicon host runs the arm64 image natively, so +# this is the closest local stand-in and the only way to run the Linux-only `/proc` benches. +# +# Usage: +# scripts/bench-docker-arm64.sh [bench...] [-- ] +# +# Examples: +# scripts/bench-docker-arm64.sh linux-main +# scripts/bench-docker-arm64.sh linux-main enhanced_metrics -- --quick +# +# Results land in bottlecap/target-linux-arm64/criterion///, so they never mix +# with the host baselines, with the same baseline-.txt note as `bench-compare.sh save` +# (machine, image and limits, load, profile, commit). Compare them on the host with +# CARGO_TARGET_DIR=bottlecap/target-linux-arm64 scripts/bench-compare.sh compare +# +# Environment: +# RUST_IMAGE docker image (default rust:1.93-bookworm; keep it on the stable toolchain CI +# uses). bottlecap needs protoc for the build script: the image installs it. +# DOCKER_CPUS CPU limit for the container (default 2, like a Lambda sandbox's 2 vCPUs. Lambda +# at 128 MB gets about 0.07 vCPU of that, so absolute numbers here are still +# faster than production; the ratios between benches are what transfer). +# DOCKER_MEM memory limit (default 2g; the bench binaries hold several MB of fixtures). +# BENCH_FAST=1 no LTO, 16 codegen units; fast link, NOT the production profile. +set -euo pipefail + +name="${1:-}" +[[ -n "$name" ]] || { sed -n '2,25p' "${BASH_SOURCE[0]}" | sed 's/^# \{0,1\}//'; exit 1; } +shift + +here="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +image="${RUST_IMAGE:-rust:1.93-bookworm}" +cpus="${DOCKER_CPUS:-2}" +mem="${DOCKER_MEM:-2g}" + +benches=() +extra=() +while [[ $# -gt 0 ]]; do + if [[ "$1" == "--" ]]; then + shift + extra=("$@") + break + fi + benches+=("--bench" "$1") + shift +done + +# `${arr[@]+"${arr[@]}"}` expands to nothing when the array is empty; a bare "${arr[@]}" is an +# unbound variable under `set -u` on bash 3.2 (/bin/bash on macOS). The cargo arguments reach +# the container shell as positional parameters, not as script text, so filters with spaces or +# shell metacharacters pass through unchanged. +cargo_args=(${benches[@]+"${benches[@]}"} -- --save-baseline "$name" ${extra[@]+"${extra[@]}"}) + +# HEAD, plus a hash of the uncommitted diff when the tree has changes, as `bench-compare.sh` +# records it. +git_rev="$(git -C "$here" rev-parse --short HEAD 2>/dev/null)" || git_rev=unknown +if [[ "$git_rev" != unknown && -n "$(git -C "$here" status --porcelain --untracked-files=no)" ]]; then + git_rev="$git_rev-dirty (diff $(git -C "$here" diff HEAD | git hash-object --stdin | cut -c1-12))" +fi + +# Optional docker flags. +docker_flags=() +if [[ "${BENCH_FAST:-0}" == 1 ]]; then + docker_flags+=(-e CARGO_PROFILE_BENCH_LTO=false -e CARGO_PROFILE_BENCH_CODEGEN_UNITS=16) +fi +if [[ -t 0 ]]; then + docker_flags+=(-it) +fi + +# Named volumes keep the registry and git checkouts between runs. +docker run --rm \ + --platform linux/arm64 \ + --cpus "$cpus" --memory "$mem" \ + -v "$here:/src" \ + -v bottlecap-bench-cargo-registry:/usr/local/cargo/registry \ + -v bottlecap-bench-cargo-git:/usr/local/cargo/git \ + -w /src/bottlecap \ + -e CARGO_TARGET_DIR=/src/bottlecap/target-linux-arm64 \ + -e CARGO_NET_GIT_FETCH_WITH_CLI=true \ + -e BENCH_NAME="$name" \ + -e BENCH_GIT="$git_rev" \ + -e BENCH_CONTAINER="$image --cpus $cpus --memory $mem" \ + ${docker_flags[@]+"${docker_flags[@]}"} \ + "$image" \ + bash -c " + set -euo pipefail + if ! command -v protoc >/dev/null; then + apt-get update -qq && apt-get install -y -qq --no-install-recommends protobuf-compiler >/dev/null + fi + cpu=\$(grep -m1 'model name' /proc/cpuinfo | cut -d: -f2- | sed 's/^ //' || true) + overrides=\$(env | grep '^CARGO_PROFILE_BENCH_' | sort | tr '\n' ' ' || true) + profile=\${overrides:+bench with \${overrides}(NOT production)} + note=\"machine: \$(uname -sm), \$(nproc) cpus visible\${cpu:+, \$cpu}; container: \$BENCH_CONTAINER; load average: \$(cut -d' ' -f1-3 /proc/loadavg); profile: \${profile:-bench = release (opt-level z, lto, codegen-units 1)}\" + echo \"\$note\" + cargo bench \"\$@\" + mkdir -p \"\$CARGO_TARGET_DIR/criterion\" + printf '%s\nsaved: %s\ngit: %s\n' \"\$note\" \"\$(date -u +%Y-%m-%dT%H:%M:%SZ)\" \"\$BENCH_GIT\" \\ + > \"\$CARGO_TARGET_DIR/criterion/baseline-\$BENCH_NAME.txt\" + " bench "${cargo_args[@]}" +echo "saved baseline '$name' under bottlecap/target-linux-arm64/criterion (notes in baseline-$name.txt)"