From 6e5e1dbdfee72f05a3635e8c50f1dbe1ba35ca26 Mon Sep 17 00:00:00 2001 From: Jordan Gonzalez <30836115+duncanista@users.noreply.github.com> Date: Wed, 28 Oct 2026 15:00:55 -0400 Subject: [PATCH 1/2] [SVLS-9977] bench: criterion benches for the DogStatsD hot path and the agent config load --- .github/workflows/cargo.yml | 2 +- Cargo.lock | 174 +++++- LICENSE-3rdparty.csv | 17 + crates/datadog-agent-config/Cargo.toml | 11 + crates/datadog-agent-config/benches/load.rs | 262 +++++++++ crates/dogstatsd/Cargo.toml | 7 + crates/dogstatsd/benches/hotpath.rs | 601 ++++++++++++++++++++ 7 files changed, 1069 insertions(+), 5 deletions(-) create mode 100644 crates/datadog-agent-config/benches/load.rs create mode 100644 crates/dogstatsd/benches/hotpath.rs diff --git a/.github/workflows/cargo.yml b/.github/workflows/cargo.yml index db6553e3..e9c6fb03 100644 --- a/.github/workflows/cargo.yml +++ b/.github/workflows/cargo.yml @@ -65,7 +65,7 @@ jobs: # doesn't fail. export AWS_LC_FIPS_SYS_NO_ASM=1 fi - cargo clippy --workspace --all-features -- -D warnings + cargo clippy --workspace --all-targets --all-features -- -D warnings build: name: Build diff --git a/Cargo.lock b/Cargo.lock index f889c50f..4ec99bcb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -44,6 +44,18 @@ dependencies = [ "libc", ] +[[package]] +name = "anes" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" + +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + [[package]] name = "anyhow" version = "1.0.103" @@ -176,7 +188,7 @@ dependencies = [ "bitflags 2.11.0", "cexpr", "clang-sys", - "itertools", + "itertools 0.13.0", "log", "prettyplease", "proc-macro2", @@ -306,6 +318,12 @@ dependencies = [ "thiserror 1.0.69", ] +[[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" @@ -368,6 +386,33 @@ dependencies = [ "windows-link", ] +[[package]] +name = "ciborium" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e" +dependencies = [ + "ciborium-io", + "ciborium-ll", + "serde", +] + +[[package]] +name = "ciborium-io" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757" + +[[package]] +name = "ciborium-ll" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9" +dependencies = [ + "ciborium-io", + "half", +] + [[package]] name = "clang-sys" version = "1.8.1" @@ -379,6 +424,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" @@ -461,6 +531,40 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "criterion" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f" +dependencies = [ + "anes", + "cast", + "ciborium", + "clap", + "criterion-plot", + "is-terminal", + "itertools 0.10.5", + "num-traits", + "once_cell", + "oorandom", + "regex", + "serde", + "serde_derive", + "serde_json", + "tinytemplate", + "walkdir", +] + +[[package]] +name = "criterion-plot" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1" +dependencies = [ + "cast", + "itertools 0.10.5", +] + [[package]] name = "crossbeam-channel" version = "0.5.15" @@ -476,6 +580,12 @@ version = "0.8.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" +[[package]] +name = "crunchy" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" + [[package]] name = "crypto-common" version = "0.1.7" @@ -547,6 +657,7 @@ dependencies = [ name = "datadog-agent-config" version = "0.1.0" dependencies = [ + "criterion", "datadog-opentelemetry", "dogstatsd", "figment", @@ -556,6 +667,7 @@ dependencies = [ "serde", "serde-aux", "serde_json", + "tempfile", "tokio", "tracing", ] @@ -837,6 +949,7 @@ dependencies = [ name = "dogstatsd" version = "0.1.0" dependencies = [ + "criterion", "datadog-protos", "ddsketch-agent", "derive_more", @@ -1192,6 +1305,17 @@ dependencies = [ "tracing", ] +[[package]] +name = "half" +version = "2.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b" +dependencies = [ + "cfg-if", + "crunchy", + "zerocopy", +] + [[package]] name = "hashbrown" version = "0.12.3" @@ -1256,6 +1380,12 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "hermit-abi" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e17592d60ebacc7d5e169f4663c5f84f9161cc90328abcfe8456f41e4dfcb284" + [[package]] name = "hex" version = "0.4.3" @@ -1616,6 +1746,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.13.0" @@ -2332,6 +2482,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" @@ -2623,7 +2779,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ "heck", - "itertools", + "itertools 0.13.0", "log", "multimap", "once_cell", @@ -2643,7 +2799,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools", + "itertools 0.13.0", "proc-macro2", "quote", "syn 2.0.117", @@ -2656,7 +2812,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" dependencies = [ "anyhow", - "itertools", + "itertools 0.13.0", "proc-macro2", "quote", "syn 2.0.117", @@ -3756,6 +3912,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/LICENSE-3rdparty.csv b/LICENSE-3rdparty.csv index b1ce0464..d71b4f0f 100644 --- a/LICENSE-3rdparty.csv +++ b/LICENSE-3rdparty.csv @@ -4,6 +4,8 @@ ahash,https://github.com/tkaitchuck/ahash,MIT OR Apache-2.0,Tom Kaitchuck allocator-api2,https://github.com/zakarumych/allocator-api2,MIT OR Apache-2.0,Zakarum android_system_properties,https://github.com/nical/android_system_properties,MIT OR Apache-2.0,Nicolas Silva +anes,https://github.com/zrzka/anes-rs,MIT OR Apache-2.0,Robert Vojta +anstyle,https://github.com/rust-cli/anstyle,MIT OR Apache-2.0,The anstyle Authors anyhow,https://github.com/dtolnay/anyhow,MIT OR Apache-2.0,David Tolnay arc-swap,https://github.com/vorner/arc-swap,MIT OR Apache-2.0,Michal 'vorner' Vaner assert-json-diff,https://github.com/davidpdrsn/assert-json-diff,MIT,David Pedersen @@ -30,13 +32,20 @@ cadence,https://github.com/56quarters/cadence,Apache-2.0 OR MIT,Nick Pillitteri camino,https://github.com/camino-rs/camino,MIT OR Apache-2.0,"Without Boats , Ashley Williams , Steve Klabnik , Rain " cargo-platform,https://github.com/rust-lang/cargo,MIT OR Apache-2.0,The cargo-platform Authors cargo_metadata,https://github.com/oli-obk/cargo_metadata,MIT,Oliver Schneider +cast,https://github.com/japaric/cast.rs,MIT OR Apache-2.0,Jorge Aparicio cc,https://github.com/rust-lang/cc-rs,MIT OR Apache-2.0,Alex Crichton cesu8,https://github.com/emk/cesu8-rs,Apache-2.0 OR MIT,Eric Kidd cexpr,https://github.com/jethrogb/rust-cexpr,Apache-2.0 OR MIT,Jethro Beekman cfg-if,https://github.com/rust-lang/cfg-if,MIT OR Apache-2.0,Alex Crichton chacha20,https://github.com/RustCrypto/stream-ciphers,MIT OR Apache-2.0,RustCrypto Developers chrono,https://github.com/chronotope/chrono,MIT OR Apache-2.0,The chrono Authors +ciborium,https://github.com/enarx/ciborium,Apache-2.0,Nathaniel McCallum +ciborium-io,https://github.com/enarx/ciborium,Apache-2.0,Nathaniel McCallum +ciborium-ll,https://github.com/enarx/ciborium,Apache-2.0,Nathaniel McCallum clang-sys,https://github.com/KyleMayes/clang-sys,Apache-2.0,Kyle Mayes +clap,https://github.com/clap-rs/clap,MIT OR Apache-2.0,The clap Authors +clap_builder,https://github.com/clap-rs/clap,MIT OR Apache-2.0,The clap_builder Authors +clap_lex,https://github.com/clap-rs/clap,MIT OR Apache-2.0,The clap_lex Authors combine,https://github.com/Marwes/combine,MIT,Markus Westerlind const_format,https://github.com/rodrimati1992/const_format_crates,Zlib,rodrimati1992 const_format_proc_macros,https://github.com/rodrimati1992/const_format_crates,Zlib,rodrimati1992 @@ -44,8 +53,10 @@ core-foundation,https://github.com/servo/core-foundation-rs,MIT OR Apache-2.0,Th core-foundation-sys,https://github.com/servo/core-foundation-rs,MIT OR Apache-2.0,The Servo Project Developers cpufeatures,https://github.com/RustCrypto/utils,MIT OR Apache-2.0,RustCrypto Developers crc32fast,https://github.com/srijs/rust-crc32fast,MIT OR Apache-2.0,"Sam Rijs , Alex Crichton " +criterion-plot,https://github.com/bheisler/criterion.rs,MIT OR Apache-2.0,"Jorge Aparicio , Brook Heisler " crossbeam-channel,https://github.com/crossbeam-rs/crossbeam,MIT OR Apache-2.0,The crossbeam-channel Authors crossbeam-utils,https://github.com/crossbeam-rs/crossbeam,MIT OR Apache-2.0,The crossbeam-utils Authors +crunchy,https://github.com/eira-fransham/crunchy,MIT,Eira Fransham crypto-common,https://github.com/RustCrypto/traits,MIT OR Apache-2.0,RustCrypto Developers ctor,https://github.com/mmastrac/rust-ctor,Apache-2.0 OR MIT,Matt Mastracci darling,https://github.com/TedDriggs/darling,MIT,Ted Driggs @@ -93,10 +104,12 @@ generic-array,https://github.com/fizyk20/generic-array,MIT,"Bartłomiej Kamińsk getrandom,https://github.com/rust-random/getrandom,MIT OR Apache-2.0,The Rand Project Developers glob,https://github.com/rust-lang/glob,MIT OR Apache-2.0,The Rust Project Developers h2,https://github.com/hyperium/h2,MIT,"Carl Lerche , Sean McArthur " +half,https://github.com/VoidStarKat/half-rs,MIT OR Apache-2.0,Kathryn Long hashbrown,https://github.com/rust-lang/hashbrown,MIT OR Apache-2.0,Amanieu d'Antras headers,https://github.com/hyperium/headers,MIT,Sean McArthur headers-core,https://github.com/hyperium/headers,MIT,Sean McArthur heck,https://github.com/withoutboats/heck,MIT OR Apache-2.0,The heck Authors +hermit-abi,https://github.com/hermit-os/hermit-rs,MIT OR Apache-2.0,The hermit-abi Authors hex,https://github.com/KokaKiwi/rust-hex,MIT OR Apache-2.0,KokaKiwi home,https://github.com/rust-lang/cargo,MIT OR Apache-2.0,Brian Anderson http,https://github.com/hyperium/http,MIT OR Apache-2.0,"Alex Crichton , Carl Lerche , Sean McArthur " @@ -127,6 +140,7 @@ indexmap,https://github.com/indexmap-rs/indexmap,Apache-2.0 OR MIT,The indexmap inlinable_string,https://github.com/fitzgen/inlinable_string,Apache-2.0 OR MIT,Nick Fitzgerald ipnet,https://github.com/krisprice/ipnet,MIT OR Apache-2.0,Kris Price iri-string,https://github.com/lo48576/iri-string,MIT OR Apache-2.0,YOSHIOKA Takuma +is-terminal,https://github.com/sunfishcode/is-terminal,MIT,"softprops , Dan Gohman " itertools,https://github.com/rust-itertools/itertools,MIT OR Apache-2.0,bluss itoa,https://github.com/dtolnay/itoa,MIT OR Apache-2.0,David Tolnay jiff,https://github.com/BurntSushi/jiff,Unlicense OR MIT,Andrew Gallant @@ -182,6 +196,7 @@ nu-ansi-term,https://github.com/nushell/nu-ansi-term,MIT,"ogham@bsago.me, Ryan S num-conv,https://github.com/jhpratt/num-conv,MIT OR Apache-2.0,Jacob Pratt num-traits,https://github.com/rust-num/num-traits,MIT OR Apache-2.0,The Rust Project Developers once_cell,https://github.com/matklad/once_cell,MIT OR Apache-2.0,Aleksey Kladov +oorandom,https://hg.sr.ht/~icefox/oorandom,MIT,Simon Heath openssl-probe,https://github.com/alexcrichton/openssl-probe,MIT OR Apache-2.0,Alex Crichton opentelemetry,https://github.com/open-telemetry/opentelemetry-rust/tree/main/opentelemetry,Apache-2.0,The opentelemetry Authors opentelemetry-semantic-conventions,https://github.com/open-telemetry/opentelemetry-rust/tree/main/opentelemetry-semantic-conventions,Apache-2.0,The opentelemetry-semantic-conventions Authors @@ -302,6 +317,7 @@ time,https://github.com/time-rs/time,MIT OR Apache-2.0,"Jacob Pratt , Time contributors" time-macros,https://github.com/time-rs/time,MIT OR Apache-2.0,"Jacob Pratt , Time contributors" tinystr,https://github.com/unicode-org/icu4x,Unicode-3.0,The ICU4X Project Developers +tinytemplate,https://github.com/bheisler/TinyTemplate,Apache-2.0 OR MIT,Brook Heisler tinyvec,https://github.com/Lokathor/tinyvec,Zlib OR Apache-2.0 OR MIT,Lokathor tinyvec_macros,https://github.com/Soveu/tinyvec_macros,MIT OR Apache-2.0 OR Zlib,Soveu tokio,https://github.com/tokio-rs/tokio,MIT,Tokio Contributors @@ -336,6 +352,7 @@ utf8_iter,https://github.com/hsivonen/utf8_iter,Apache-2.0 OR MIT,Henri Sivonen uuid,https://github.com/uuid-rs/uuid,Apache-2.0 OR MIT,"Ashley Mannix, Dylan DPC, Hunar Roop Kahlon" valuable,https://github.com/tokio-rs/valuable,MIT,The valuable Authors wait-timeout,https://github.com/alexcrichton/wait-timeout,MIT OR Apache-2.0,Alex Crichton +walkdir,https://github.com/BurntSushi/walkdir,Unlicense OR MIT,Andrew Gallant want,https://github.com/seanmonstar/want,MIT,Sean McArthur wasi,https://github.com/bytecodealliance/wasi,Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT,The Cranelift Project Developers wasip2,https://github.com/bytecodealliance/wasi-rs,Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT,The wasip2 Authors diff --git a/crates/datadog-agent-config/Cargo.toml b/crates/datadog-agent-config/Cargo.toml index 43a68a4e..afb1576d 100644 --- a/crates/datadog-agent-config/Cargo.toml +++ b/crates/datadog-agent-config/Cargo.toml @@ -4,6 +4,9 @@ version = "0.1.0" edition.workspace = true license.workspace = true +[lib] +bench = false + [dependencies] figment = { version = "0.10", default-features = false, features = ["yaml", "env"] } libdd-trace-obfuscation.workspace = true @@ -18,7 +21,15 @@ dogstatsd = { path = "../dogstatsd", default-features = false } tokio = { version = "1.47", default-features = false, features = ["time"] } [dev-dependencies] +# default-features = false drops rayon, so the benches run on one thread; the +# `set_var` calls in the benches rely on that. +criterion = { version = "0.5", default-features = false, features = ["cargo_bench_support"] } figment = { version = "0.10", default-features = false, features = ["yaml", "env", "test"] } +tempfile = "3" + +[[bench]] +name = "load" +harness = false [features] # Disabled by default. Consumers that need a TLS provider on the transitive diff --git a/crates/datadog-agent-config/benches/load.rs b/crates/datadog-agent-config/benches/load.rs new file mode 100644 index 00000000..3d95ad71 --- /dev/null +++ b/crates/datadog-agent-config/benches/load.rs @@ -0,0 +1,262 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +//! Config load benchmarks. Run with `cargo bench -p datadog-agent-config`. +//! +//! The environment is process-global, so each bench sets its scenario once, +//! outside the timed loop. No tracing subscriber is installed, so the crate's +//! log calls cost nothing. + +use std::hint::black_box; + +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use datadog_agent_config::apm_replace_rule::deserialize_apm_replace_rules; +use datadog_agent_config::get_config; +use serde::de::IntoDeserializer; +use serde::de::value::{Error as ValueError, StrDeserializer}; + +// A Lambda sandbox environment (python3.13, arm64) without any DD_* variable. +// PATH, LD_LIBRARY_PATH and TMPDIR are left as the host has them. +const LAMBDA_BASE: &[(&str, &str)] = &[ + ("AWS_LAMBDA_FUNCTION_NAME", "my-function"), + ("AWS_LAMBDA_FUNCTION_VERSION", "$LATEST"), + ("AWS_LAMBDA_FUNCTION_MEMORY_SIZE", "1024"), + ("AWS_LAMBDA_LOG_GROUP_NAME", "/aws/lambda/my-function"), + ( + "AWS_LAMBDA_LOG_STREAM_NAME", + "2026/10/27/[$LATEST]0123456789abcdef0123456789abcdef", + ), + ("AWS_LAMBDA_LOG_FORMAT", "Text"), + ("AWS_LAMBDA_RUNTIME_API", "127.0.0.1:9001"), + ("AWS_LAMBDA_INITIALIZATION_TYPE", "on-demand"), + ("AWS_EXECUTION_ENV", "AWS_Lambda_python3.13"), + ("AWS_REGION", "us-east-1"), + ("AWS_DEFAULT_REGION", "us-east-1"), + ("AWS_ACCESS_KEY_ID", "ASIAEXAMPLEEXAMPLE12"), + ( + "AWS_SECRET_ACCESS_KEY", + "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + ), + ( + "AWS_SESSION_TOKEN", + "IQoJb3JpZ2luX2VjEJr//////////wEaCXVzLWVhc3QtMSJHMEUCIQDexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexampleexample", + ), + ("AWS_XRAY_DAEMON_ADDRESS", "169.254.79.129:2000"), + ("AWS_XRAY_CONTEXT_MISSING", "LOG_ERROR"), + ("_AWS_XRAY_DAEMON_ADDRESS", "169.254.79.129"), + ("_AWS_XRAY_DAEMON_PORT", "2000"), + ( + "_X_AMZN_TRACE_ID", + "Root=1-6720abcd-0123456789abcdef01234567;Parent=0123456789abcdef;Sampled=0;Lineage=1:abcd1234:0", + ), + ("LAMBDA_TASK_ROOT", "/var/task"), + ("LAMBDA_RUNTIME_DIR", "/var/runtime"), + ("LANG", "en_US.UTF-8"), + ("TZ", ":UTC"), + ("_HANDLER", "handler.handler"), + ("PYTHONPATH", "/var/runtime"), + ("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", "prod"), + ("DD_SERVICE", "checkout"), + ("DD_VERSION", "1.10.3"), + ("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 REPLACE_RULES_1: &str = r#"[{"name":"http.url","pattern":"(token=)[^&]*","repl":"${1}?"}]"#; +const REPLACE_RULES_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", + 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"}]"#, + ), + ("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_RULES_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"), +]; + +const TYPICAL_YAML: &str = r#"api_key: 0123456789abcdef0123456789abcdef +site: datadoghq.eu +log_level: info +env: staging +service: yaml-svc +version: 2.3.4 +tags: + - team:yaml + - tier:web +logs_enabled: true +logs_config: + processing_rules: + - type: exclude_at_match + name: exclude_health + pattern: "GET /health" + use_compression: true + compression_level: 6 +apm_config: + replace_tags: + - name: http.url + pattern: "(token=)[^&]*" + repl: "${1}?" + features: + - enable_stats_by_span_kind + obfuscation: + http: + remove_query_string: true +otlp_config: + receiver: + protocols: + http: + endpoint: localhost:4318 +serverless_flush_strategy: end +enhanced_metrics: true +capture_lambda_payload: true +api_key_secret_arn: arn:aws:secretsmanager:us-east-1:123456789012:secret:yaml +"#; + +fn set_all(pairs: &[(&str, &str)]) { + for (k, v) in pairs { + // SAFETY: criterion runs benches on this thread and no other thread is + // alive that reads the environment. + unsafe { std::env::set_var(k, v) } + } +} + +fn remove_dd_env() { + let keys: Vec<_> = std::env::vars_os() + .map(|(k, _)| k) + .filter(|k| { + let k = k.to_string_lossy(); + // `get_config` also reads the proxy variables. + k.starts_with("DD_") + || k.starts_with("DATADOG_") + || k.eq_ignore_ascii_case("HTTPS_PROXY") + || k.eq_ignore_ascii_case("NO_PROXY") + }) + .collect(); + for k in keys { + // SAFETY: see `set_all`. + unsafe { std::env::remove_var(k) } + } +} + +fn set_scenario(scenario: &str) { + remove_dd_env(); + set_all(LAMBDA_BASE); + match scenario { + "none" => {} + "typical" => set_all(TYPICAL), + "heavy" => { + set_all(TYPICAL); + set_all(HEAVY_EXTRA); + } + other => panic!("unknown scenario {other}"), + } +} + +fn bench_get_config(c: &mut Criterion) { + let empty_dir = tempfile::tempdir().expect("tempdir"); + let yaml_dir = tempfile::tempdir().expect("tempdir"); + std::fs::write(yaml_dir.path().join("datadog.yaml"), TYPICAL_YAML).expect("write yaml"); + + let mut group = c.benchmark_group("config"); + for scenario in ["none", "typical", "heavy"] { + set_scenario(scenario); + let config = get_config(empty_dir.path()); + assert_eq!(config.service.is_some(), scenario != "none"); + assert_eq!( + config.apm_replace_tags.map_or(0, |r| r.len()), + if scenario == "heavy" { 3 } else { 0 } + ); + group.bench_function(BenchmarkId::new("get_config", scenario), |b| { + b.iter(|| get_config(black_box(empty_dir.path()))) + }); + } + set_scenario("typical"); + let config = get_config(yaml_dir.path()); + assert_eq!(config.api_key, "0123456789abcdef0123456789abcdef"); + assert_eq!(config.service.as_deref(), Some("checkout")); + group.bench_function(BenchmarkId::new("get_config_yaml", "typical"), |b| { + b.iter(|| get_config(black_box(yaml_dir.path()))) + }); + remove_dd_env(); + group.finish(); +} + +fn bench_replace_rules(c: &mut Criterion) { + // The deserializer both config sources use, given the JSON string form. + let parse = |json: &str| { + let de: StrDeserializer<'_, ValueError> = json.into_deserializer(); + deserialize_apm_replace_rules(de) + }; + let mut group = c.benchmark_group("config"); + for (count, json) in [(1, REPLACE_RULES_1), (3, REPLACE_RULES_3)] { + let rules = parse(json).expect("deserialize").expect("rules"); + assert_eq!(rules.len(), count); + group.throughput(Throughput::Elements(count as u64)); + group.bench_function(BenchmarkId::new("replace_rules", count), |b| { + b.iter(|| parse(black_box(json))) + }); + } + group.finish(); +} + +criterion_group!(benches, bench_get_config, bench_replace_rules); +criterion_main!(benches); diff --git a/crates/dogstatsd/Cargo.toml b/crates/dogstatsd/Cargo.toml index c4d997d4..231365a0 100644 --- a/crates/dogstatsd/Cargo.toml +++ b/crates/dogstatsd/Cargo.toml @@ -27,11 +27,18 @@ regex = { version = "1.10.6", default-features = false } zstd = { version = "0.13.3", default-features = false } [dev-dependencies] +# default-features = false drops rayon, so the benches run on one thread; the +# `set_var` calls in the benches rely on that. +criterion = { version = "0.5", default-features = false, features = ["cargo_bench_support"] } http = "1" mockito = { version = "1.5.0", default-features = false } proptest = "1.4.0" tracing-test = { version = "0.2.5", default-features = false } +[[bench]] +name = "hotpath" +harness = false + [features] default = [] windows-pipes = ["tokio/io-util"] diff --git a/crates/dogstatsd/benches/hotpath.rs b/crates/dogstatsd/benches/hotpath.rs new file mode 100644 index 00000000..a29cfb75 --- /dev/null +++ b/crates/dogstatsd/benches/hotpath.rs @@ -0,0 +1,601 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +//! Hot-path benchmarks for DogStatsD: parse, packet, insert, flush, compress +//! and the `ustr` interner. +//! +//! Run with `cargo bench -p dogstatsd`, or filter with +//! `cargo bench -p dogstatsd -- dogstatsd/parse`. + +use std::hint::black_box; +use std::io::Write; +use std::process::Command; +use std::sync::Once; +use std::time::{Duration, Instant}; + +use criterion::{BatchSize, BenchmarkId, Criterion, SamplingMode, Throughput, criterion_group}; +use ddsketch_agent::DDSketch; +use dogstatsd::aggregator::Aggregator; +use dogstatsd::constants::MAX_CONTEXTS; +use dogstatsd::datadog::Series; +use dogstatsd::metric::{self, Metric, MetricValue, SortedTags, parse}; +use protobuf::Message; +use ustr::Ustr; +use zstd::zstd_safe::zstd_sys::ZSTD_EndDirective; +use zstd::zstd_safe::{CCtx, CParameter, InBuffer, OutBuffer}; + +const ENRICH_TAGS: &str = "function_arn:arn:aws:lambda:us-east-1:123456789012:function:my-checkout-fn,region:us-east-1,account_id:123456789012,aws_account:123456789012,functionname:my-checkout-fn,resource:my-checkout-fn,executedversion:$LATEST,memorysize:128,architecture:arm64,runtime:nodejs22.x,_dd.runtime_family:nodejs,env:prod,service:checkout,version:1.2.3,dd_extension_version:90,_dd.compute_stats:1"; +const TAGS_8: &str = "env:prod,service:checkout,version:1.2.3,region:us-east-1,team:payments,endpoint:/api/v1/pay,status:200,method:POST"; +const COUNT_2_TAGS: &str = "my.metric:1|c|#tag1:a,tag2:b"; +const DIST_2_TAGS: &str = "my.metric:1.5|d|#tag1:a,tag2:b"; +const DIST_8_TAGS: &str = "app.checkout.latency:12.5|d|#env:prod,service:checkout,version:1.2.3,region:us-east-1,team:payments,endpoint:/api/v1/pay,status:200,method:POST"; +const CONTEXT_COUNTS: [usize; 3] = [20, 1_000, 10_000]; +const INSERT_BATCH: usize = 1_000; +// The flusher's default level: `metrics_config_compression_level` in +// the `logs_config_compression_level` default in datadog-agent-config. +const ZSTD_LEVEL: i32 = 3; + +// The `ustr` interner is process-global, so its first use is measured in a +// child process: the bench binary re-executes itself with this variable set. +const CHILD_ENV: &str = "DOGSTATSD_BENCH_CHILD"; +const FIRST_PARSE_LINE: &str = "warm:1|c|#a:b"; + +fn enrich_tags() -> SortedTags { + SortedTags::parse(ENRICH_TAGS).expect("enrich tags parse") +} + +fn lcg(seed: &mut u64) -> u64 { + *seed = seed + .wrapping_mul(6_364_136_223_846_793_005) + .wrapping_add(1_442_695_040_888_963_407); + *seed >> 33 +} + +/// `n` values in `1..=distinct`, in pseudo-random order. +fn sketch_values(n: usize, distinct: u64) -> Vec { + let mut seed = 42; + (0..n) + .map(|_| (lcg(&mut seed) % distinct + 1) as f64) + .collect() +} + +fn unique_line(i: usize) -> String { + format!("my.metric:1|c|#env:prod,service:api,request_id:{i:08x}-0000-4000-8000-{i:012x}") +} + +fn build_aggregator(contexts: usize, distribution: bool) -> Aggregator { + let mut aggregator = Aggregator::new(enrich_tags(), MAX_CONTEXTS).expect("aggregator"); + for i in 0..contexts { + let line = if distribution { + format!( + "app.request.latency{}:{}|d|#endpoint:/api/v{i},status:200,method:GET", + i % 50, + (i % 900) + 1 + ) + } else { + format!( + "app.request.count{}:1|c|#endpoint:/api/v{i},status:200,method:GET", + i % 50 + ) + }; + aggregator + .insert(parse(&line).expect("parse")) + .expect("insert"); + } + aggregator +} + +fn series_bodies(series: &[Series]) -> Vec> { + series + .iter() + .map(|s| serde_json::to_vec(s).expect("serialize series")) + .collect() +} + +fn bench_parse(c: &mut Criterion) { + let mut group = c.benchmark_group("dogstatsd/parse"); + group.throughput(Throughput::Elements(1)); + for (id, line) in [ + ("count_2_tags", COUNT_2_TAGS), + ("distribution_2_tags", DIST_2_TAGS), + ("distribution_8_tags", DIST_8_TAGS), + ] { + parse(line).expect("parse"); + group.bench_function(id, |b| b.iter(|| parse(black_box(line)))); + } + group.bench_function("sorted_tags_8", |b| { + b.iter(|| SortedTags::parse(black_box(TAGS_8))) + }); + group.bench_function("sorted_tags_16_enrich", |b| { + b.iter(|| SortedTags::parse(black_box(ENRICH_TAGS))) + }); + let name = Ustr::from("app.checkout.latency"); + let tags = Some(SortedTags::parse(TAGS_8).expect("tags")); + group.bench_function("metric_id_8_tags", |b| { + b.iter(|| metric::id(black_box(name), black_box(&tags), black_box(1_700_000_000))) + }); + // The single-point sketch every distribution line builds. + group.bench_function("distribution_value", |b| { + b.iter(|| MetricValue::distribution(black_box(12.5))) + }); + group.finish(); +} + +/// Mirrors the per-datagram path up to the aggregator send, with no namespace: +/// `process_packet` and `insert_metrics` in src/dogstatsd.rs. Log calls are left out. +fn packet_today(buf: &[u8]) -> Vec { + let Ok(msgs) = std::str::from_utf8(buf) else { + return Vec::new(); + }; + let statsd_metric_strings: Vec<&str> = msgs.split('\n').collect(); + let metric_count_in_packet = statsd_metric_strings + .iter() + .filter(|m| !m.is_empty()) + .count(); + black_box(metric_count_in_packet); + statsd_metric_strings + .into_iter() + .filter(|m| { + !m.is_empty() + && !m.starts_with("_sc|") + && !m.starts_with("_e{") + && !m.starts_with("aws.lambda.enhanced.invocations") + }) + .map(|m| m.replace('\n', "")) + .filter_map(|m| parse(m.as_str()).ok()) + .collect() +} + +fn bench_packet(c: &mut Criterion) { + let lines: Vec = (0..20) + .map(|i| format!("my.metric{i}:1|c|#tag1:a,tag2:b")) + .collect(); + let packet = lines.join("\n"); + assert_eq!(packet_today(packet.as_bytes()).len(), 20); + + let mut group = c.benchmark_group("dogstatsd/packet"); + group.throughput(Throughput::Elements(20)); + group.bench_function("20_lines", |b| { + b.iter(|| packet_today(black_box(packet.as_bytes()))) + }); + group.finish(); +} + +fn bench_insert(c: &mut Criterion) { + let mut group = c.benchmark_group("dogstatsd/insert"); + + // Occupied path: every metric lands on one of 100 existing contexts. + let lines: Vec = (0..100) + // `|T` pins the bucket, so the batch and the base never straddle a 10 s boundary. + .map(|i| format!("my.metric{i}:1|c|#tag1:a,tag2:b|T1700000000")) + .collect(); + let mut base = Aggregator::new(enrich_tags(), MAX_CONTEXTS).expect("aggregator"); + for line in &lines { + base.insert(parse(line).expect("parse")).expect("insert"); + } + let batch: Vec = (0..INSERT_BATCH) + .map(|i| parse(&lines[i % lines.len()]).expect("parse")) + .collect(); + group.throughput(Throughput::Elements(INSERT_BATCH as u64)); + group.bench_function("count_100_contexts", |b| { + b.iter_batched( + || (base.clone(), batch.clone()), + |(mut aggregator, batch)| { + for m in batch { + let _ = aggregator.insert(m); + } + aggregator + }, + BatchSize::LargeInput, + ) + }); + + // Vacant path: 10k new contexts, including table growth and rehash. + let unique: Vec = (0..10_000) + .map(|i| parse(&unique_line(i)).expect("parse")) + .collect(); + group.throughput(Throughput::Elements(unique.len() as u64)); + group.bench_function("count_vacant_10000_contexts", |b| { + b.iter_batched( + || { + ( + Aggregator::new(enrich_tags(), MAX_CONTEXTS).expect("aggregator"), + unique.clone(), + ) + }, + |(mut aggregator, batch)| { + for m in batch { + let _ = aggregator.insert(m); + } + aggregator + }, + BatchSize::LargeInput, + ) + }); + + // Merge path: each metric carries a single-point sketch merged into one of + // 10 contexts. The base is warmed up so its sketches are at full size. + let lines: Vec = (0..10) + .map(|i| format!("my.dist{i}:1|d|#tag1:a,tag2:b|T1700000000")) + .collect(); + let dist_metrics = |values: &[f64]| -> Vec { + values + .iter() + .enumerate() + .map(|(i, v)| { + let mut m = parse(&lines[i % lines.len()]).expect("parse"); + m.value = MetricValue::distribution(*v); + m + }) + .collect() + }; + let mut base = Aggregator::new(enrich_tags(), MAX_CONTEXTS).expect("aggregator"); + for m in dist_metrics(&sketch_values(20_000, 2_000)) { + base.insert(m).expect("insert"); + } + let batch = dist_metrics(&sketch_values(INSERT_BATCH, 2_000)); + group.throughput(Throughput::Elements(INSERT_BATCH as u64)); + group.bench_function("distribution_10_contexts", |b| { + b.iter_batched( + || (base.clone(), batch.clone()), + |(mut aggregator, batch)| { + for m in batch { + let _ = aggregator.insert(m); + } + aggregator + }, + BatchSize::LargeInput, + ) + }); + + // What the merge path costs per value against a plain insert. Each batch + // starts from a fresh copy of the warm sketch: a full bin spills into a new + // bin, so a sketch fed for the whole run would keep growing. + group.throughput(Throughput::Elements(INSERT_BATCH as u64)); + for distinct in [1, 64, 2_000] { + let mut warm = DDSketch::default(); + for v in sketch_values(20_000, distinct) { + warm.insert(v); + } + let values = sketch_values(INSERT_BATCH, distinct); + let variant = format!("{distinct}_distinct"); + + group.bench_function( + BenchmarkId::new("sketch_merge_single_point", &variant), + |b| { + b.iter_batched_ref( + || warm.clone(), + |sketch| { + for v in &values { + let mut one = DDSketch::default(); + one.insert(*v); + sketch.merge(black_box(&one)); + } + }, + BatchSize::SmallInput, + ) + }, + ); + group.bench_function(BenchmarkId::new("sketch_insert", &variant), |b| { + b.iter_batched_ref( + || warm.clone(), + |sketch| { + for v in &values { + sketch.insert(black_box(*v)); + } + }, + BatchSize::SmallInput, + ) + }); + } + group.finish(); +} + +fn bench_flush(c: &mut Criterion) { + let mut group = c.benchmark_group("dogstatsd/flush"); + // Iterations take up to tens of milliseconds; linear sampling would need + // thousands of them. + group.sampling_mode(SamplingMode::Flat); + for contexts in CONTEXT_COUNTS { + group.throughput(Throughput::Elements(contexts as u64)); + + let counts = build_aggregator(contexts, false); + group.bench_function(BenchmarkId::new("consume_metrics", contexts), |b| { + b.iter_batched( + || counts.clone(), + |mut aggregator| aggregator.consume_metrics(), + BatchSize::LargeInput, + ) + }); + let series = counts.clone().consume_metrics(); + // The flusher deep-copies both batches to return them on error: + // `Flusher::flush` in src/flusher.rs. + // Clone and drop are both timed; production drops its copy too. + group.bench_function(BenchmarkId::new("series_clone", contexts), |b| { + b.iter(|| series.clone()) + }); + // Series serialization in src/datadog/shipping.rs. + group.bench_function(BenchmarkId::new("series_to_json", contexts), |b| { + b.iter(|| series_bodies(black_box(&series))) + }); + + let distributions = build_aggregator(contexts, true); + group.bench_function(BenchmarkId::new("consume_distributions", contexts), |b| { + b.iter_batched( + || distributions.clone(), + |mut aggregator| aggregator.consume_distributions(), + BatchSize::LargeInput, + ) + }); + let sketches = distributions.clone().consume_distributions(); + // Clone and drop are both timed; production drops its copy too. + group.bench_function(BenchmarkId::new("sketch_payload_clone", contexts), |b| { + b.iter(|| sketches.clone()) + }); + // Sketch serialization in src/datadog/shipping.rs. + group.bench_function(BenchmarkId::new("sketch_to_protobuf", contexts), |b| { + b.iter(|| { + sketches + .iter() + .map(|p| p.write_to_bytes().expect("serialize sketches")) + .collect::>() + }) + }); + } + group.finish(); +} + +/// Size of a level-3 `CCtx` after compressing `src` in one of the ways +/// `zstd` offers: streaming with or without a pledged size, or one shot. +fn cctx_size(src: &[u8], mode: &str) -> usize { + let mut cctx = CCtx::create(); + cctx.set_parameter(CParameter::CompressionLevel(ZSTD_LEVEL)) + .expect("level"); + let mut dst = vec![0u8; zstd::zstd_safe::compress_bound(src.len())]; + if mode == "one_shot" { + cctx.compress2(&mut dst[..], src).expect("compress2"); + return cctx.sizeof(); + } + if mode == "stream_pledged" { + cctx.set_pledged_src_size(Some(src.len() as u64)) + .expect("pledged size"); + } + let mut input = InBuffer::around(src); + let mut output = OutBuffer::around(&mut dst[..]); + while input.pos() < src.len() { + cctx.compress_stream2(&mut output, &mut input, ZSTD_EndDirective::ZSTD_e_continue) + .expect("compress_stream2"); + } + let mut empty = InBuffer::around(&[]); + while cctx + .compress_stream2(&mut output, &mut empty, ZSTD_EndDirective::ZSTD_e_end) + .expect("compress_stream2 end") + != 0 + {} + cctx.sizeof() +} + +fn print_zstd_memory(bodies_by_contexts: &[(usize, Vec>)]) { + for (contexts, bodies) in bodies_by_contexts { + let largest = bodies.iter().max_by_key(|b| b.len()).expect("bodies"); + let compressed: usize = bodies + .iter() + .map(|b| zstd::bulk::compress(b, ZSTD_LEVEL).expect("compress").len()) + .sum(); + let raw: usize = bodies.iter().map(Vec::len).sum(); + println!( + "zstd memory ({contexts} contexts, {} payloads, {raw} B raw, {compressed} B compressed, largest payload {} B): \ + CCtx stream={} B, stream_pledged={} B, one_shot={} B", + bodies.len(), + largest.len(), + cctx_size(largest, "stream"), + cctx_size(largest, "stream_pledged"), + cctx_size(largest, "one_shot"), + ); + } +} + +/// The three paths must give the same bytes back and sizes within 1 % of each other. +fn check_compress_outputs(bodies_by_contexts: &[(usize, Vec>)]) { + let mut reused = zstd::bulk::Compressor::new(ZSTD_LEVEL).expect("compressor"); + for (_, bodies) in bodies_by_contexts { + for body in bodies { + let mut encoder = + zstd::stream::write::Encoder::new(Vec::new(), ZSTD_LEVEL).expect("encoder"); + encoder.write_all(body).expect("write"); + let stream = encoder.finish().expect("finish"); + assert_eq!( + zstd::decode_all(stream.as_slice()).expect("decode"), + *body, + "stream output must decompress to the input" + ); + let bulk = zstd::bulk::compress(body, ZSTD_LEVEL).expect("compress"); + let reused_out = reused.compress(body).expect("compress"); + assert_eq!(bulk, reused_out, "bulk and reused compressor must match"); + let diff = stream.len().abs_diff(bulk.len()); + assert!( + diff * 100 <= stream.len(), + "stream {} B and bulk {} B differ by more than 1 %", + stream.len(), + bulk.len() + ); + } + } +} + +fn bench_compress(c: &mut Criterion) { + let bodies_by_contexts: Vec<(usize, Vec>)> = [1_000, 10_000] + .into_iter() + .map(|contexts| { + let series = build_aggregator(contexts, false).consume_metrics(); + (contexts, series_bodies(&series)) + }) + .collect(); + let print_memory = Once::new(); + + let mut group = c.benchmark_group("dogstatsd/compress"); + group.sampling_mode(SamplingMode::Flat); + for (contexts, bodies) in &bodies_by_contexts { + let raw: usize = bodies.iter().map(Vec::len).sum(); + group.throughput(Throughput::Bytes(raw as u64)); + + // What the shipper does today: the stream `Encoder` in src/datadog/shipping.rs. + group.bench_function(BenchmarkId::new("stream_encoder_level3", contexts), |b| { + print_memory.call_once(|| { + check_compress_outputs(&bodies_by_contexts); + print_zstd_memory(&bodies_by_contexts); + }); + b.iter(|| { + for body in bodies { + let mut encoder = + zstd::stream::write::Encoder::new(Vec::new(), ZSTD_LEVEL).expect("encoder"); + encoder.write_all(body).expect("write"); + black_box(encoder.finish().expect("finish")); + } + }) + }); + group.bench_function(BenchmarkId::new("bulk_compress_level3", contexts), |b| { + b.iter(|| { + for body in bodies { + black_box(zstd::bulk::compress(body, ZSTD_LEVEL).expect("compress")); + } + }) + }); + let mut compressor = zstd::bulk::Compressor::new(ZSTD_LEVEL).expect("compressor"); + group.bench_function( + BenchmarkId::new("bulk_compressor_reused_level3", contexts), + |b| { + b.iter(|| { + for body in bodies { + black_box(compressor.compress(body).expect("compress")); + } + }) + }, + ); + } + group.finish(); +} + +fn rss_kib() -> i64 { + Command::new("ps") + .args(["-o", "rss=", "-p", &std::process::id().to_string()]) + .output() + .ok() + .and_then(|out| String::from_utf8_lossy(&out.stdout).trim().parse().ok()) + .unwrap_or(-1) +} + +/// Entry point of the child process. Nothing touches `ustr` before this. +fn child(mode: &str) { + match mode { + "first_parse" => { + let start = Instant::now(); + black_box(parse(black_box(FIRST_PARSE_LINE)).expect("parse")); + println!("{}", start.elapsed().as_nanos()); + } + "report" => { + let rss_before = rss_kib(); + let start = Instant::now(); + black_box(parse(black_box(FIRST_PARSE_LINE)).expect("parse")); + let first_ns = start.elapsed().as_nanos(); + let rss_first = rss_kib(); + println!( + "ustr first parse: {first_ns} ns, RSS +{} KiB, arena allocated {} B, capacity {} B", + rss_first - rss_before, + ustr::total_allocated(), + ustr::total_capacity(), + ); + let mut done = 0; + for target in [1_000, 10_000, 100_000] { + let lines: Vec = (done..target).map(unique_line).collect(); + let start = Instant::now(); + for line in &lines { + black_box(parse(line).expect("parse")); + } + let ns_per_parse = start.elapsed().as_nanos() / lines.len() as u128; + drop(lines); + done = target; + println!( + "ustr after {target} unique tag values: {ns_per_parse} ns/parse, RSS +{} KiB since first parse, \ + arena allocated {} B, capacity {} B", + rss_kib() - rss_first, + ustr::total_allocated(), + ustr::total_capacity(), + ); + } + } + other => panic!("unknown {CHILD_ENV} mode {other}"), + } +} + +fn run_child(mode: &str) -> String { + let exe = std::env::current_exe().expect("current exe"); + let out = Command::new(exe) + .env(CHILD_ENV, mode) + .output() + .expect("spawn child"); + assert!(out.status.success(), "child {mode} failed: {out:?}"); + String::from_utf8(out.stdout).expect("utf8 child output") +} + +fn bench_ustr(c: &mut Criterion) { + let mut group = c.benchmark_group("dogstatsd/ustr"); + + let parts: Vec<&str> = TAGS_8.split([',', ':']).collect(); + for p in &parts { + Ustr::from(p); + } + group.throughput(Throughput::Elements(parts.len() as u64)); + group.bench_function("from_existing_16", |b| { + b.iter(|| { + for p in &parts { + black_box(Ustr::from(black_box(p))); + } + }) + }); + + // Each iteration is one child process; only the child-reported parse time + // counts. It includes every first-call cost of `parse`, the interner's being + // the largest. Criterion sizes the run from the wall time, spawn included. + let print_report = Once::new(); + group + .throughput(Throughput::Elements(1)) + .sampling_mode(SamplingMode::Flat) + .sample_size(10) + .warm_up_time(Duration::from_millis(100)) + .measurement_time(Duration::from_secs(1)); + group.bench_function("first_parse", |b| { + print_report.call_once(|| print!("{}", run_child("report"))); + b.iter_custom(|iters| { + (0..iters) + .map(|_| { + let ns = run_child("first_parse") + .trim() + .parse() + .expect("child nanoseconds"); + Duration::from_nanos(ns) + }) + .sum() + }) + }); + group.finish(); +} + +criterion_group!( + benches, + bench_parse, + bench_packet, + bench_insert, + bench_flush, + bench_compress, + bench_ustr +); + +fn main() { + if let Ok(mode) = std::env::var(CHILD_ENV) { + child(&mode); + return; + } + benches(); + Criterion::default().configure_from_args().final_summary(); +} From 1245ea36f78d5f41f831e200abac9afd12df2566 Mon Sep 17 00:00:00 2001 From: Jordan Gonzalez <30836115+duncanista@users.noreply.github.com> Date: Wed, 28 Oct 2026 17:13:41 -0400 Subject: [PATCH 2/2] test(trace-agent): collapse the named pipe readiness check for clippy --all-targets on windows --- crates/datadog-trace-agent/tests/integration_test.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/crates/datadog-trace-agent/tests/integration_test.rs b/crates/datadog-trace-agent/tests/integration_test.rs index 57de50ef..c29265c0 100644 --- a/crates/datadog-trace-agent/tests/integration_test.rs +++ b/crates/datadog-trace-agent/tests/integration_test.rs @@ -1149,11 +1149,11 @@ async fn test_mini_agent_named_pipe_with_real_flushers() { let mut server_ready = false; for _ in 0..20 { tokio::time::sleep(Duration::from_millis(50)).await; - if let Ok(response) = send_named_pipe_request(pipe_name, "/info", "GET", None).await { - if response.status().is_success() { - server_ready = true; - break; - } + if let Ok(response) = send_named_pipe_request(pipe_name, "/info", "GET", None).await + && response.status().is_success() + { + server_ready = true; + break; } } assert!(