From 966965a01627c6ce37121b70dd0e46ef23ff84f7 Mon Sep 17 00:00:00 2001 From: "Brian L. Troutwine" Date: Fri, 24 Jul 2026 23:06:53 +0000 Subject: [PATCH] chore(antithesis): Bound contexts in runs This commit introduces a new protocol to the intake node to allow for context bounding in runs. The purposes are twofold. First, while it's interesting to explore timelines in which there are too many contexts for either lane to handle these are less relevant to me _now_, so I've capped total contexts at 1M. Second, in order for timeseries equality to _work_ I need to be certain that I have at least _some_ timeseries with more than one point emitted into them. Previously drivers were emitting essentially random contexts and it was very unlikely for more than one point to be emitted per context. Oops. Drivers now request N contexts from the intake service for each sub-kind of dogstatsd they emit, meaning all contexts are sourced from a single spot in a topology and not from pure randomness. --- Cargo.lock | 5 + test/antithesis/harness/Cargo.toml | 1 + .../harness/proptest-regressions/contexts.txt | 7 + .../src/bin/first_sample_config/main.rs | 16 +- test/antithesis/harness/src/config.rs | 126 ++++++- test/antithesis/harness/src/contexts.rs | 329 ++++++++++++++++++ test/antithesis/harness/src/contexts/event.rs | 140 ++++++++ .../antithesis/harness/src/contexts/metric.rs | 273 +++++++++++++++ .../harness/src/contexts/service_check.rs | 132 +++++++ test/antithesis/harness/src/driver.rs | 91 ++++- test/antithesis/harness/src/lib.rs | 1 + .../harness/src/payload/dogstatsd.rs | 240 ++++++++----- .../harness/src/payload/dogstatsd/common.rs | 155 +++++++-- .../harness/src/payload/dogstatsd/events.rs | 91 ----- .../harness/src/payload/dogstatsd/metrics.rs | 157 --------- .../src/payload/dogstatsd/service_checks.rs | 81 ----- test/antithesis/intake/Cargo.toml | 4 +- test/antithesis/intake/src/bin/intake.rs | 18 +- test/antithesis/intake/src/context_pool.rs | 261 ++++++++++++++ test/antithesis/intake/src/http/antithesis.rs | 40 ++- test/antithesis/intake/src/http/state.rs | 24 +- test/antithesis/intake/src/lib.rs | 1 + .../intake/tests/series_decode_interop.rs | 9 +- .../differential/docker-compose.yaml | 7 +- ...llel_driver_send_dogstatsd_differential.rs | 6 + .../scenarios/general/docker-compose.yaml | 5 +- .../src/bin/parallel_driver_send_dogstatsd.rs | 9 + 27 files changed, 1743 insertions(+), 486 deletions(-) create mode 100644 test/antithesis/harness/proptest-regressions/contexts.txt create mode 100644 test/antithesis/harness/src/contexts.rs create mode 100644 test/antithesis/harness/src/contexts/event.rs create mode 100644 test/antithesis/harness/src/contexts/metric.rs create mode 100644 test/antithesis/harness/src/contexts/service_check.rs delete mode 100644 test/antithesis/harness/src/payload/dogstatsd/events.rs delete mode 100644 test/antithesis/harness/src/payload/dogstatsd/metrics.rs delete mode 100644 test/antithesis/harness/src/payload/dogstatsd/service_checks.rs create mode 100644 test/antithesis/intake/src/context_pool.rs diff --git a/Cargo.lock b/Cargo.lock index 96098eee7f1..1a31572fa72 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -200,11 +200,13 @@ dependencies = [ "axum", "clap", "datadog-protos", + "harness", "headers", "http-body-util", "mime", "proptest", "protobuf", + "rand 0.10.1", "serde", "serde_json", "serde_yaml", @@ -472,6 +474,7 @@ checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ "axum-core", "bytes", + "form_urlencoded", "futures-util", "http", "http-body", @@ -487,6 +490,7 @@ dependencies = [ "serde_core", "serde_json", "serde_path_to_error", + "serde_urlencoded", "sync_wrapper", "tokio", "tower", @@ -1936,6 +1940,7 @@ dependencies = [ "num-traits", "proptest", "rand 0.10.1", + "reqwest", "ryu", "serde", "serde_json", diff --git a/test/antithesis/harness/Cargo.toml b/test/antithesis/harness/Cargo.toml index ecdaf0c909b..ce49298aad0 100644 --- a/test/antithesis/harness/Cargo.toml +++ b/test/antithesis/harness/Cargo.toml @@ -20,6 +20,7 @@ itoa = { workspace = true } libc = { workspace = true } num-traits = { workspace = true } rand = { workspace = true } +reqwest = { workspace = true, features = ["blocking"] } ryu = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } diff --git a/test/antithesis/harness/proptest-regressions/contexts.txt b/test/antithesis/harness/proptest-regressions/contexts.txt new file mode 100644 index 00000000000..471a57cd7ca --- /dev/null +++ b/test/antithesis/harness/proptest-regressions/contexts.txt @@ -0,0 +1,7 @@ +# Seeds for failure cases proptest has generated in the past. It is +# automatically read and these particular cases re-run before any +# novel cases are generated. +# +# It is recommended to check this file in to source control so that +# everyone who runs the test benefits from these saved cases. +cc 671f360ca24ca5add047c2f93acf185116ebfa9c663637a430a15fa6abc56151 # shrinks to seed = 8436120490560261333, kind = Metric diff --git a/test/antithesis/harness/src/bin/first_sample_config/main.rs b/test/antithesis/harness/src/bin/first_sample_config/main.rs index ca129ac6449..a4b8eda25d9 100644 --- a/test/antithesis/harness/src/bin/first_sample_config/main.rs +++ b/test/antithesis/harness/src/bin/first_sample_config/main.rs @@ -14,7 +14,7 @@ use antithesis_sdk::prelude::*; use antithesis_sdk::random::AntithesisRng; use anyhow::Context; use clap::Parser; -use harness::config::DatadogConfig; +use harness::config::{ContextSourceConfig, DatadogConfig}; use rand::rand_core::UnwrapErr; use serde_json::json; @@ -63,10 +63,22 @@ fn main() -> anyhow::Result<()> { // Load-generator view of the same sample, so a generator caps its datagrams // to the SUT's receive buffer without reading the Agent config. + let driver = config.driver_config(&mut rng); let driver_path = cli.config_dir.join("driver.yaml"); - fs::write(&driver_path, config.driver_config(&mut rng).to_yaml()?.as_bytes()) + fs::write(&driver_path, driver.to_yaml()?.as_bytes()) .with_context(|| format!("write driver config {}", driver_path.display()))?; + // Per-kind context-pool caps for this timeline, read lazily by the intake's shared pool. Sampled + // here so the pool's cardinality varies per timeline like every other sampled knob. + let context_source_path = cli.config_dir.join("context_source.yaml"); + fs::write( + &context_source_path, + ContextSourceConfig::sample(&mut rng, driver.payload_byte_limit) + .to_yaml()? + .as_bytes(), + ) + .with_context(|| format!("write context source config {}", context_source_path.display()))?; + // Per-timeline anchor: counting these in triage tells us how many distinct // configs the run sampled. let details = serde_json::to_value(&config).unwrap_or_else(|e| json!({ "serialize_error": e.to_string() })); diff --git a/test/antithesis/harness/src/config.rs b/test/antithesis/harness/src/config.rs index 894ebd2fd26..53efe6b5667 100644 --- a/test/antithesis/harness/src/config.rs +++ b/test/antithesis/harness/src/config.rs @@ -245,6 +245,13 @@ cloud_provider_metadata: [] /// Upper bound on datagrams one driver invocation ships in a timeline. const MAX_DATAGRAMS: usize = 10_000; +/// Upper bound on the working set one driver invocation fetches from the shared context pool. +const MAX_WORKING_SET: u64 = 1_024; + +/// The intake's ceiling on one `/contexts` request. A `context_count` past it is rejected there, so a +/// config carrying one is rejected here instead. +const MAX_CONTEXTS_PER_REQUEST: usize = 65_536; + /// Config a load generator reads to shape its output to this timeline's SUT. /// `first_sample_config` samples it beside `datadog.yaml` from one draw, so the /// generator and the SUT are driven together. @@ -256,6 +263,14 @@ pub struct DriverConfig { pub payload_byte_limit: usize, /// Datagrams a driver invocation ships this timeline. pub datagram_count: usize, + /// Distinct contexts a driver invocation fetches from the shared pool as its working set. + /// + /// Sampled boundary-biased log-uniform in `1..=1_024`. Valid values run `1..=65_536`, the intake's + /// per-request ceiling. Outside that the intake rejects every `/contexts` request, the driver waits + /// out its fetch budget and ships nothing, so [`Self::read`] rejects such a config rather than + /// letting a timeline generate no load. A larger working set spreads load over more identities and + /// so puts fewer points in each, which is the trade against recurrence. + pub context_count: usize, } impl DriverConfig { @@ -270,6 +285,7 @@ impl DriverConfig { Self { payload_byte_limit, datagram_count: rng.random_range(0..=MAX_DATAGRAMS), + context_count: usize::try_from(Probe::new(1, MAX_WORKING_SET).sample(rng)).unwrap_or(usize::MAX), } } @@ -292,10 +308,96 @@ impl DriverConfig { pub fn read(config_dir: &Path) -> anyhow::Result { let path = config_dir.join("driver.yaml"); let yaml = fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?; - serde_yaml::from_str(&yaml).with_context(|| format!("parse driver config from {}", path.display())) + let config: Self = + serde_yaml::from_str(&yaml).with_context(|| format!("parse driver config from {}", path.display()))?; + anyhow::ensure!( + (1..=MAX_CONTEXTS_PER_REQUEST).contains(&config.context_count), + "context_count {} outside 1..={} in {}, the intake would reject every fetch and the driver would ship nothing", + config.context_count, + MAX_CONTEXTS_PER_REQUEST, + path.display() + ); + Ok(config) + } +} + +/// Upper bound on the distinct contexts a shared pool holds across every kind. The pool retains each +/// minted context, so the ceiling belongs to the total rather than to any one kind. +const MAX_CONTEXTS_TOTAL: u64 = 1_000_000; + +/// The per-kind caps a timeline's shared context pool fills to before it recurs existing contexts. +/// `first_sample_config` samples this beside `datadog.yaml` so cardinality varies per timeline. Each +/// cap is drawn against the budget the earlier draws left, so a kind's cardinality still varies at +/// random while the three together stay under [`MAX_CONTEXTS_TOTAL`]. +#[derive(Clone, Copy, Debug, Serialize, Deserialize)] +pub struct ContextSourceConfig { + /// Bytes a rendered line of any pooled context must fit, this timeline's real datagram budget + /// rather than the protocol ceiling. Mint builds identities against it, so every served context + /// has a rendering the driver can pack. Defaults to the sampled `payload_byte_limit`, which is the + /// smaller of the SUT's receive buffer and [`PAYLOAD_BYTE_LIMIT`]. + pub payload_byte_limit: usize, + /// Distinct metric contexts the pool holds before it recurs the ones it has. + /// + /// Sampled in `1..=MAX_CONTEXTS_TOTAL` minus what the other kinds took. A larger cap explores more + /// identities and puts fewer points in each, and costs memory: the pool retains every context it + /// mints for the life of the run. Zero is not sampled and serves nothing for the kind. + pub metric_contexts: usize, + /// Distinct event contexts the pool holds. Same range and trade as [`Self::metric_contexts`]. + pub event_contexts: usize, + /// Distinct service-check contexts the pool holds. Same range and trade as + /// [`Self::metric_contexts`]. + pub service_check_contexts: usize, +} + +impl ContextSourceConfig { + /// Sample the per-kind caps, each boundary-biased log-uniform against the budget still free. + /// + /// The draws run in order and each spends from one shared ceiling, so the total is bounded by + /// construction rather than by scaling three independent draws afterwards. Every kind keeps at + /// least one context, since a kind capped at zero panics the pool's draw. + #[must_use] + pub fn sample(rng: &mut R, payload_byte_limit: usize) -> Self { + // Two contexts held back so the later kinds can each keep their one. + let metric_contexts = sample_cap(rng, MAX_CONTEXTS_TOTAL - 2); + let free = MAX_CONTEXTS_TOTAL - metric_contexts as u64; + let event_contexts = sample_cap(rng, free - 1); + let free = free - event_contexts as u64; + Self { + payload_byte_limit, + metric_contexts, + event_contexts, + service_check_contexts: sample_cap(rng, free), + } + } + + /// Render `self` as a `context_source.yaml` string. + /// + /// # Errors + /// + /// Returns an error if serialization fails. + pub fn to_yaml(&self) -> anyhow::Result { + serde_yaml::to_string(self).context("serialize context_source.yaml") + } + + /// Read the context-source config from the `context_source.yaml` that `first_sample_config` wrote + /// to `config_dir`. + /// + /// # Errors + /// + /// Returns an error if the config is unreadable or is not valid YAML. + pub fn read(config_dir: &Path) -> anyhow::Result { + let path = config_dir.join("context_source.yaml"); + let yaml = fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?; + serde_yaml::from_str(&yaml).with_context(|| format!("parse context source config from {}", path.display())) } } +/// A single per-kind cap in `1..=ceiling`. The ceiling is well within `usize` on every supported +/// target, so the saturating conversion is unreachable in practice. +fn sample_cap(rng: &mut R, ceiling: u64) -> usize { + usize::try_from(Probe::new(1, ceiling.max(1)).sample(rng)).unwrap_or(usize::MAX) +} + #[cfg(test)] mod tests { use std::collections::BTreeSet; @@ -401,6 +503,28 @@ mod tests { assert_eq!(seen, [true, true]); } + // The pool holds every minted context, so the ceiling is on the total across kinds rather than on + // each kind alone. Three independent draws at the ceiling would retain three million. + #[test] + fn context_caps_sum_within_the_total_ceiling() { + for seed in 0..64 { + let caps = ContextSourceConfig::sample(&mut SeqRng(seed), 8_192); + let total = (caps.metric_contexts + caps.event_contexts + caps.service_check_contexts) as u64; + assert!(total <= MAX_CONTEXTS_TOTAL, "seed {seed} sampled {total}"); + // A kind of zero panics the pool's `random_range(0..0)`, so every kind keeps at least one. + assert!(caps.metric_contexts >= 1 && caps.event_contexts >= 1 && caps.service_check_contexts >= 1); + } + } + + // Randomness still drives each kind rather than the total being split evenly. + #[test] + fn context_caps_vary_per_kind() { + let spread: BTreeSet = (0..64) + .map(|seed| ContextSourceConfig::sample(&mut SeqRng(seed), 8_192).metric_contexts) + .collect(); + assert!(spread.len() > 8, "metric cap barely varies: {spread:?}"); + } + #[test] fn compressor_samples_every_kind() { let mut seen = BTreeSet::new(); diff --git a/test/antithesis/harness/src/contexts.rs b/test/antithesis/harness/src/contexts.rs new file mode 100644 index 00000000000..f5d596b6c71 --- /dev/null +++ b/test/antithesis/harness/src/contexts.rs @@ -0,0 +1,329 @@ +//! The context protocol: reusable `DogStatsD` identities a driver renders load against. +//! +//! A [`Context`] is a per-type stable identity — a metric's `(kind, name, tags)`, an event's title + +//! tags + option fields, a service check's name + tags + host. It is minted over the free generator's +//! full content alphabet (`payload/dogstatsd/common.rs`) and accepted only when a probe render is +//! `!is_malformed` (see [`crate::dogstatsd`]) — conforming to the Agent parser and nothing stricter. +//! The driver varies the per-occurrence payload (value/text/status, extensions, timestamp) each +//! render, so a pooled identity recurs while its load varies. +//! +//! A shared intake pool mints identities up to a per-kind cap then recurs them, and serves them to +//! drivers over the length-prefixed binary codec here ([`encode_response`] / [`decode_response`]), +//! which carries non-UTF-8 names and tags that JSON could not. + +use rand::{Rng, RngExt}; + +use crate::dogstatsd::is_malformed; + +pub mod event; +pub mod metric; +pub mod service_check; + +/// How many times to re-mint an identity whose probe render the Agent would drop before falling back +/// to a guaranteed-sound identity. Mint is mostly-valid, so the fallback is rare. +const REMINT_TRIES: usize = 16; + +/// How many times to re-render a context whose per-occurrence payload the Agent would drop before +/// falling back to a guaranteed-well-formed line. +const RENDER_TRIES: usize = 8; + +/// Digits allowed for a rendered length field, generous so a floor is never an underestimate. +pub(crate) const LEN_DIGITS: usize = 5; + +/// Digits allowed for a rendered timestamp, likewise generous. +pub(crate) const TS_DIGITS: usize = 20; + +/// The three context kinds. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Kind { + /// A metric context. + Metric, + /// An event context. + Event, + /// A service-check context. + ServiceCheck, +} + +impl Kind { + /// Sample a kind by the message-type weight — 98% metric, 1% event, 1% service check. + #[must_use] + pub fn sample(rng: &mut (impl Rng + ?Sized)) -> Kind { + match rng.random_range(0..100u32) { + 0 => Kind::Event, + 1 => Kind::ServiceCheck, + _ => Kind::Metric, + } + } +} + +/// A reusable `DogStatsD` identity of one of the three kinds. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub enum Context { + /// A metric identity. + Metric(metric::MetricContext), + /// An event identity. + Event(event::EventContext), + /// A service-check identity. + ServiceCheck(service_check::ServiceCheckContext), +} + +impl Context { + /// Mint a context of `kind` whose renders fit `budget` and that the Agent forwards, or `None` when + /// no such identity is available. + /// + /// The identity is built against the budget, so it fits by construction and nothing is minted then + /// measured. The remaining re-mint loop is about content alone: the alphabet carries protocol + /// delimiters and some combinations land on the drop side, which a probe render is what detects. + /// An exhausted loop yields `None` rather than an identity of another kind, so the caller never + /// stores a metric in the event or service-check working set. + #[must_use] + pub fn mint_within(kind: Kind, rng: &mut (impl Rng + ?Sized), budget: usize) -> Option { + for _ in 0..REMINT_TRIES { + let context = match kind { + Kind::Metric => Context::Metric(metric::MetricContext::mint_within(rng, budget)?), + Kind::Event => Context::Event(event::EventContext::mint_within(rng, budget)?), + Kind::ServiceCheck => { + Context::ServiceCheck(service_check::ServiceCheckContext::mint_within(rng, budget)?) + } + }; + let mut probe = Vec::new(); + context.render(rng, &mut probe); + if is_malformed(&probe).is_ok() { + return Some(context); + } + } + None + } + + /// Render one datagram line (no trailing `\n`) for this identity with a fresh per-occurrence + /// payload. Returns the packed multi-value run length, or zero. + pub fn render(&self, rng: &mut (impl Rng + ?Sized), out: &mut Vec) -> usize { + match self { + Context::Metric(c) => c.render(rng, out), + Context::Event(c) => c.render(rng, out), + Context::ServiceCheck(c) => c.render(rng, out), + } + } + + /// Bytes every render of this identity must spend, whatever the per-occurrence payload. + #[must_use] + pub fn floor(&self) -> usize { + match self { + Context::Metric(c) => c.floor(), + Context::Event(c) => c.floor(), + Context::ServiceCheck(c) => c.floor(), + } + } + + /// Render one line within `budget`, or `None` when the budget cannot hold this identity. + fn render_within(&self, rng: &mut (impl Rng + ?Sized), out: &mut Vec, budget: usize) -> Option { + match self { + Context::Metric(c) => c.render_within(rng, out, budget), + Context::Event(c) => c.render_within(rng, out, budget), + Context::ServiceCheck(c) => c.render_within(rng, out, budget), + } + } + + /// Render one datagram line the Agent forwards within `budget`, or `None` when this context has no + /// forwardable rendering that fits. The budget is a construction input to each attempt rather than + /// a filter on the result, and an exhausted attempt count yields nothing rather than a line + /// carrying an identity the pool never issued. + pub fn render_wellformed_within( + &self, rng: &mut (impl Rng + ?Sized), out: &mut Vec, budget: usize, + ) -> Option { + for _ in 0..RENDER_TRIES { + let start = out.len(); + let packed = self.render_within(rng, out, budget)?; + if is_malformed(&out[start..]).is_ok() { + return Some(packed); + } + out.truncate(start); + } + None + } + + /// Append this context's tagged, length-prefixed encoding. + pub fn encode(&self, out: &mut Vec) { + match self { + Context::Metric(c) => { + put_u8(out, 0); + c.encode(out); + } + Context::Event(c) => { + put_u8(out, 1); + c.encode(out); + } + Context::ServiceCheck(c) => { + put_u8(out, 2); + c.encode(out); + } + } + } + + /// Decode one context, advancing `*pos`. Returns `None` on truncation or an unknown tag. + fn decode(buf: &[u8], pos: &mut usize) -> Option { + Some(match get_u8(buf, pos)? { + 0 => Context::Metric(metric::MetricContext::decode(buf, pos)?), + 1 => Context::Event(event::EventContext::decode(buf, pos)?), + 2 => Context::ServiceCheck(service_check::ServiceCheckContext::decode(buf, pos)?), + _ => return None, + }) + } +} + +/// Encode a `GET /contexts` response body: a `u32` count then each context. +#[must_use] +pub fn encode_response(contexts: &[Context]) -> Vec { + let mut out = Vec::new(); + // A response holds the N contexts a driver asked for, far below u32::MAX. + let count = u32::try_from(contexts.len()).unwrap_or(u32::MAX); + out.extend_from_slice(&count.to_le_bytes()); + for context in contexts { + context.encode(&mut out); + } + out +} + +/// Decode a `GET /contexts` response body. Returns `None` on any truncation or malformed field, so a +/// partial or corrupt body is an error, not a panic. Never pre-sizes from the wire count. +#[must_use] +pub fn decode_response(buf: &[u8]) -> Option> { + let mut pos = 0; + let count = get_u32(buf, &mut pos)?; + let mut contexts = Vec::new(); + for _ in 0..count { + contexts.push(Context::decode(buf, &mut pos)?); + } + Some(contexts) +} + +/// A fresh per-occurrence Unix timestamp for a `d:` field. Any positive integer forwards. +pub(crate) fn fresh_timestamp(rng: &mut (impl Rng + ?Sized)) -> u64 { + rng.random_range(1..=2_000_000_000u64) +} + +// --- shared length-prefixed binary codec --- + +/// Append one byte. +pub(crate) fn put_u8(out: &mut Vec, b: u8) { + out.push(b); +} + +/// Append `len` as a little-endian `u16`, saturating an over-long field. Minted fields are bounded +/// well under `u16::MAX`, so saturation never fires in practice. +fn put_u16(out: &mut Vec, len: usize) { + let len = u16::try_from(len).unwrap_or(u16::MAX); + out.extend_from_slice(&len.to_le_bytes()); +} + +/// Append a `u16` length prefix then the bytes. +pub(crate) fn put_bytes(out: &mut Vec, bytes: &[u8]) { + put_u16(out, bytes.len()); + out.extend_from_slice(bytes); +} + +/// Append a `u16` count then each byte run. +pub(crate) fn put_tags(out: &mut Vec, tags: &[Vec]) { + put_u16(out, tags.len()); + for tag in tags { + put_bytes(out, tag); + } +} + +/// Read one byte, advancing `*pos`. +pub(crate) fn get_u8(buf: &[u8], pos: &mut usize) -> Option { + let byte = *buf.get(*pos)?; + *pos += 1; + Some(byte) +} + +/// Read a little-endian `u16` as a `usize`, advancing `*pos`. +fn get_u16(buf: &[u8], pos: &mut usize) -> Option { + let end = pos.checked_add(2)?; + let slice = buf.get(*pos..end)?; + *pos = end; + Some(u16::from_le_bytes([slice[0], slice[1]]) as usize) +} + +/// Read a little-endian `u32` as a `usize`, advancing `*pos`. +fn get_u32(buf: &[u8], pos: &mut usize) -> Option { + let end = pos.checked_add(4)?; + let slice = buf.get(*pos..end)?; + *pos = end; + Some(u32::from_le_bytes([slice[0], slice[1], slice[2], slice[3]]) as usize) +} + +/// Read a `u16`-prefixed byte run, advancing `*pos`. +pub(crate) fn get_bytes<'a>(buf: &'a [u8], pos: &mut usize) -> Option<&'a [u8]> { + let len = get_u16(buf, pos)?; + let end = pos.checked_add(len)?; + let slice = buf.get(*pos..end)?; + *pos = end; + Some(slice) +} + +/// Read a `u16`-counted list of byte runs, advancing `*pos`. Does not pre-size from the wire count. +pub(crate) fn get_tags(buf: &[u8], pos: &mut usize) -> Option>> { + let count = get_u16(buf, pos)?; + let mut tags = Vec::new(); + for _ in 0..count { + tags.push(get_bytes(buf, pos)?.to_vec()); + } + Some(tags) +} + +#[cfg(test)] +mod tests { + use proptest::prelude::*; + use rand::rngs::SmallRng; + use rand::SeedableRng; + + use super::{decode_response, encode_response, Context, Kind}; + use crate::dogstatsd::is_malformed; + + fn any_kind() -> impl Strategy { + prop_oneof![Just(Kind::Metric), Just(Kind::Event), Just(Kind::ServiceCheck)] + } + + proptest! { + /// A minted context conforms to is_malformed. Content carries delimiters, so a raw render may + /// land on the drop side. The repair loop is the sorter, and any line it does yield forwards. + #[test] + fn property_test_render_wellformed_always_forwards(seed: u64, kind in any_kind()) { + let mut rng = SmallRng::seed_from_u64(seed); + let Some(context) = Context::mint_within(kind, &mut rng, 8_192) else { return Ok(()) }; + for _ in 0..8 { + let mut line = Vec::new(); + if context.render_wellformed_within(&mut rng, &mut line, 8_192).is_some() { + prop_assert_eq!(is_malformed(&line), Ok(()), "a rendered line was droppable"); + } + } + } + + /// A response of minted contexts round-trips through the binary codec, non-UTF-8 and all. + #[test] + fn property_test_response_round_trips(seed: u64) { + let mut rng = SmallRng::seed_from_u64(seed); + let contexts: Vec = (0..8) + .filter_map(|_| Context::mint_within(Kind::sample(&mut rng), &mut rng, 8_192)) + .collect(); + let wire = encode_response(&contexts); + let decoded = decode_response(&wire); + prop_assert_eq!(decoded.as_deref(), Some(contexts.as_slice())); + } + + /// Decode never panics and rejects every truncated prefix of a valid body. + #[test] + fn property_test_decode_rejects_truncation(seed: u64) { + let mut rng = SmallRng::seed_from_u64(seed); + let contexts: Vec = (0..4) + .filter_map(|_| Context::mint_within(Kind::sample(&mut rng), &mut rng, 8_192)) + .collect(); + let wire = encode_response(&contexts); + for cut in 0..wire.len() { + let _ = decode_response(&wire[..cut]); + } + prop_assert_eq!(decode_response(&wire).map(|c| c.len()), Some(contexts.len())); + } + } +} diff --git a/test/antithesis/harness/src/contexts/event.rs b/test/antithesis/harness/src/contexts/event.rs new file mode 100644 index 00000000000..7eea6d0333a --- /dev/null +++ b/test/antithesis/harness/src/contexts/event.rs @@ -0,0 +1,140 @@ +//! Event contexts: the stable `_e{}` identity a driver renders text and timestamps against. +//! +//! The Agent does not aggregate events, but each event has a natural stable identity — title, tags, +//! and its option fields (aggregation key, host, source, alert type, priority) — distinct from the +//! per-occurrence text and timestamp. The pool recurs the identity; the driver varies text and +//! timestamp each render. + +use rand::{Rng, RngExt}; + +use super::{fresh_timestamp, get_bytes, get_tags, put_bytes, put_tags}; +use super::{LEN_DIGITS, TS_DIGITS}; +use crate::payload::dogstatsd::common; + +/// Identity option prefixes: aggregation key, hostname, source type, alert type, priority. Bad values +/// are logged and defaulted by the Agent, never dropped, so any content forwards. +const OPT_PREFIXES: &[&[u8]] = &[b"k:", b"h:", b"s:", b"t:", b"p:"]; + +/// Identity option counts: mostly none, a small body. +const OPT_COUNTS: &[usize] = &[0, 0, 1, 1, 2, 3]; + +/// An event identity: title, tags, and fixed option chunks. Text and timestamp vary per render. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct EventContext { + /// Title content, non-empty. + pub title: Vec, + /// `key:value` tags. + pub tags: Vec>, + /// Fixed option chunks, each carrying its own prefix. + pub options: Vec>, +} + +impl EventContext { + /// Mint an event identity that renders within `budget`, or `None` when the budget cannot hold the + /// smallest one. Title, options and tags are built against the room the render's header, timestamp + /// and separators leave. + pub(crate) fn mint_within(rng: &mut (impl Rng + ?Sized), budget: usize) -> Option { + const RESERVED: usize = "_e{".len() + LEN_DIGITS + 1 + LEN_DIGITS + "}:".len() + 1 + "|d:".len() + TS_DIGITS; + let title = common::identifier_within(rng, budget.checked_sub(RESERVED)?); + if title.is_empty() { + return None; + } + let mut room = budget - RESERVED - title.len(); + let count = OPT_COUNTS[rng.random_range(0..OPT_COUNTS.len())]; + let mut options = Vec::new(); + for _ in 0..count { + let prefix = OPT_PREFIXES[rng.random_range(0..OPT_PREFIXES.len())]; + let Some(body_room) = room.checked_sub(1 + prefix.len()) else { + break; + }; + let mut chunk = prefix.to_vec(); + chunk.extend_from_slice(&common::optional_text_within(rng, body_room)); + room -= 1 + chunk.len(); + options.push(chunk); + } + Some(Self { + title, + tags: common::tags_within(rng, room), + options, + }) + } + + /// Render `_e{title_len,text_len}:title|text[|opt...]|d:ts[|#tags]` for a fresh text and + /// timestamp. The header lengths are the true byte lengths, so the Agent never rejects on a + /// length mismatch. Returns zero (events carry no packed run). + /// Bytes every render of this identity must spend: the header, the title, the fixed options, the + /// timestamp and the tag set. Only the text is variable. Length and timestamp digits are allowed + /// generously so the floor is never an underestimate. + pub(crate) fn floor(&self) -> usize { + const HEADER: usize = "_e{".len() + LEN_DIGITS + 1 + LEN_DIGITS + "}:".len(); + HEADER + + self.title.len() + + 1 + + self.options.iter().map(|opt| 1 + opt.len()).sum::() + + "|d:".len() + + TS_DIGITS + + common::tags_len(&self.tags) + } + + /// Render within `budget`, or `None` when the budget cannot hold the identity. Only the text is + /// sampled, and it is sampled against the room left rather than trimmed afterwards. + pub(crate) fn render_within( + &self, rng: &mut (impl Rng + ?Sized), out: &mut Vec, budget: usize, + ) -> Option { + let text_room = budget.checked_sub(self.floor())?; + let text = common::optional_text_within(rng, text_room); + let mut itoa = itoa::Buffer::new(); + out.extend_from_slice(b"_e{"); + out.extend_from_slice(itoa.format(self.title.len()).as_bytes()); + out.push(b','); + out.extend_from_slice(itoa.format(text.len()).as_bytes()); + out.extend_from_slice(b"}:"); + out.extend_from_slice(&self.title); + out.push(b'|'); + out.extend_from_slice(&text); + for opt in &self.options { + out.push(b'|'); + out.extend_from_slice(opt); + } + out.extend_from_slice(b"|d:"); + out.extend_from_slice(itoa.format(fresh_timestamp(rng)).as_bytes()); + common::serialize_tags(&self.tags, out); + Some(0) + } + + pub(crate) fn render(&self, rng: &mut (impl Rng + ?Sized), out: &mut Vec) -> usize { + let text = common::optional_text(rng); + let mut itoa = itoa::Buffer::new(); + out.extend_from_slice(b"_e{"); + out.extend_from_slice(itoa.format(self.title.len()).as_bytes()); + out.push(b','); + out.extend_from_slice(itoa.format(text.len()).as_bytes()); + out.extend_from_slice(b"}:"); + out.extend_from_slice(&self.title); + out.push(b'|'); + out.extend_from_slice(&text); + for opt in &self.options { + out.push(b'|'); + out.extend_from_slice(opt); + } + out.extend_from_slice(b"|d:"); + out.extend_from_slice(itoa.format(fresh_timestamp(rng)).as_bytes()); + common::serialize_tags(&self.tags, out); + 0 + } + + /// Append this context's length-prefixed encoding. + pub(crate) fn encode(&self, out: &mut Vec) { + put_bytes(out, &self.title); + put_tags(out, &self.tags); + put_tags(out, &self.options); + } + + /// Decode one event context, advancing `*pos`. + pub(crate) fn decode(buf: &[u8], pos: &mut usize) -> Option { + let title = get_bytes(buf, pos)?.to_vec(); + let tags = get_tags(buf, pos)?; + let options = get_tags(buf, pos)?; + Some(Self { title, tags, options }) + } +} diff --git a/test/antithesis/harness/src/contexts/metric.rs b/test/antithesis/harness/src/contexts/metric.rs new file mode 100644 index 00000000000..383a7dc43c7 --- /dev/null +++ b/test/antithesis/harness/src/contexts/metric.rs @@ -0,0 +1,273 @@ +//! Metric contexts: the `(kind, name, tags)` identity a driver renders values against. + +use rand::{Rng, RngExt}; + +use super::{get_bytes, get_tags, get_u8, put_bytes, put_tags, put_u8}; +use crate::payload::dogstatsd::common; + +/// The six metric types. +const METRIC_TYPES: [MetricType; 6] = [ + MetricType::Count, + MetricType::Gauge, + MetricType::Timing, + MetricType::Histogram, + MetricType::Set, + MetricType::Distribution, +]; + +/// Extension-chunk counts per render: mostly none, with a boundary tail. +const EXT_COUNTS: &[usize] = &[0, 0, 0, 0, 1, 1, 2, 3, 127, 255]; + +/// A metric type. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub enum MetricType { + /// Count. + Count, + /// Gauge. + Gauge, + /// Timing. + Timing, + /// Histogram. + Histogram, + /// Set. + Set, + /// Distribution. + Distribution, +} + +impl MetricType { + /// The on-wire type symbol. + fn token(self) -> &'static [u8] { + match self { + MetricType::Count => b"c", + MetricType::Gauge => b"g", + MetricType::Timing => b"ms", + MetricType::Histogram => b"h", + MetricType::Set => b"s", + MetricType::Distribution => b"d", + } + } + + /// A stable codec byte. + fn to_byte(self) -> u8 { + match self { + MetricType::Count => 0, + MetricType::Gauge => 1, + MetricType::Timing => 2, + MetricType::Histogram => 3, + MetricType::Set => 4, + MetricType::Distribution => 5, + } + } + + /// Decode a codec byte. + fn from_byte(b: u8) -> Option { + Some(match b { + 0 => MetricType::Count, + 1 => MetricType::Gauge, + 2 => MetricType::Timing, + 3 => MetricType::Histogram, + 4 => MetricType::Set, + 5 => MetricType::Distribution, + _ => return None, + }) + } + + /// Whether this is the set type, whose value the Agent never parses. + fn is_set(self) -> bool { + matches!(self, MetricType::Set) + } +} + +/// A metric identity: type, name, and tags. The value and extensions vary per render. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct MetricContext { + /// The metric type. + pub kind: MetricType, + /// Name content. + pub name: Vec, + /// `key:value` tags. + pub tags: Vec>, +} + +impl MetricContext { + /// Mint a metric identity that renders within `budget`, or `None` when the budget cannot hold the + /// smallest one. The name and tags are built against the room the render's own skeleton leaves, so + /// the identity fits by construction and no probe is needed to find that out. + pub(crate) fn mint_within(rng: &mut (impl Rng + ?Sized), budget: usize) -> Option { + let kind = METRIC_TYPES[rng.random_range(0..METRIC_TYPES.len())]; + // `:value|type` is what every render of this identity must carry beyond the name and tags. + let reserved = 1 + common::min_value_token() + 1 + kind.token().len(); + let name = common::identifier_within(rng, budget.checked_sub(reserved)?); + if name.is_empty() { + return None; + } + let tags = common::tags_within(rng, budget - reserved - name.len()); + Some(Self { kind, name, tags }) + } + + /// Bytes every render of this identity must spend: `name:value|type` and the tag set. A render + /// spends anything past this on extra packed values and extension chunks. + pub(crate) fn floor(&self) -> usize { + self.fixed() + common::min_value_token() + } + + /// The identity's cost without the value placeholder. + fn fixed(&self) -> usize { + self.name.len() + 1 + 1 + self.kind.token().len() + common::tags_len(&self.tags) + } + + /// Render `name:value|type[|#tags][|ext...]` within `budget`, or `None` when the budget cannot + /// hold the identity. Values and extensions are sampled against the room left, so no byte is built + /// and then thrown away. + pub(crate) fn render_within( + &self, rng: &mut (impl Rng + ?Sized), out: &mut Vec, budget: usize, + ) -> Option { + let fixed = self.fixed(); + let value_room = budget.checked_sub(fixed)?; + if value_room < common::min_value_token() { + return None; + } + out.extend_from_slice(&self.name); + out.push(b':'); + let mut used = 0; + let packed = if self.kind.is_set() { + let value = common::opaque_value_within(rng, value_room); + used += value.len(); + out.extend_from_slice(&value); + 0 + } else { + let first = common::float_token_within(rng, value_room); + used += first.len(); + out.extend_from_slice(&first); + let mut count = 1; + for _ in 1..value_count(rng) { + let room = value_room - used; + let Some(token_room) = room.checked_sub(1) else { + break; + }; + let value = common::float_token_within(rng, token_room); + if value.is_empty() { + break; + } + out.push(b':'); + out.extend_from_slice(&value); + used += 1 + value.len(); + count += 1; + } + if count > 1 { + count + } else { + 0 + } + }; + out.push(b'|'); + out.extend_from_slice(self.kind.token()); + common::serialize_tags(&self.tags, out); + let mut room = budget - (fixed + used); + let ext_count = EXT_COUNTS[rng.random_range(0..EXT_COUNTS.len())]; + for _ in 0..ext_count { + let Some(chunk_room) = room.checked_sub(1) else { + break; + }; + let Some(chunk) = ext_chunk_within(rng, chunk_room) else { + break; + }; + out.push(b'|'); + out.extend_from_slice(&chunk); + room -= 1 + chunk.len(); + } + Some(packed) + } + + /// Render `name:value|type[|#tags][|ext...]` for a fresh value and extensions. Returns the packed + /// multi-value run length, or zero for a single value or a set. + pub(crate) fn render(&self, rng: &mut (impl Rng + ?Sized), out: &mut Vec) -> usize { + out.extend_from_slice(&self.name); + out.push(b':'); + let packed = if self.kind.is_set() { + out.extend_from_slice(&common::opaque_value(rng)); + 0 + } else { + let count = value_count(rng); + for i in 0..count { + if i > 0 { + out.push(b':'); + } + out.extend_from_slice(&common::float_token(rng)); + } + if count > 1 { + count + } else { + 0 + } + }; + out.push(b'|'); + out.extend_from_slice(self.kind.token()); + common::serialize_tags(&self.tags, out); + let ext_count = EXT_COUNTS[rng.random_range(0..EXT_COUNTS.len())]; + for _ in 0..ext_count { + out.push(b'|'); + ext_chunk(rng, out); + } + packed + } + + /// Append this context's length-prefixed encoding. + pub(crate) fn encode(&self, out: &mut Vec) { + put_u8(out, self.kind.to_byte()); + put_bytes(out, &self.name); + put_tags(out, &self.tags); + } + + /// Decode one metric context, advancing `*pos`. + pub(crate) fn decode(buf: &[u8], pos: &mut usize) -> Option { + let kind = MetricType::from_byte(get_u8(buf, pos)?)?; + let name = get_bytes(buf, pos)?.to_vec(); + let tags = get_tags(buf, pos)?; + Some(Self { kind, name, tags }) + } +} + +/// The `:`-packed value run length. Overwhelmingly one, with a short tail. +fn value_count(rng: &mut (impl Rng + ?Sized)) -> usize { + match rng.random_range(0..800u16) { + 0..792 => 1, + 792..796 => 2, + 796..798 => 3, + 798 => 4, + _ => 5, + } +} + +/// One extension chunk within `budget`, or `None` when the budget cannot hold the shortest one. +fn ext_chunk_within(rng: &mut (impl Rng + ?Sized), budget: usize) -> Option> { + let prefix: &[u8] = match rng.random_range(0..4u8) { + 0 => b"@", + 1 => b"c:", + 2 => b"e:", + _ => b"card:", + }; + let body_room = budget.checked_sub(prefix.len())?; + let body = if prefix == b"@" { + common::rate_token_within(rng, body_room)? + } else { + common::optional_text_within(rng, body_room) + }; + let mut chunk = prefix.to_vec(); + chunk.extend_from_slice(&body); + Some(chunk) +} + +/// Append one extension chunk (prefix + body). `@` is a parseable rate; the origin chunks carry free +/// content the Agent never drops on. +fn ext_chunk(rng: &mut (impl Rng + ?Sized), out: &mut Vec) { + let (prefix, body): (&[u8], Vec) = match rng.random_range(0..4u8) { + 0 => (b"@", common::rate_token(rng)), + 1 => (b"c:", common::optional_text(rng)), + 2 => (b"e:", common::optional_text(rng)), + _ => (b"card:", common::optional_text(rng)), + }; + out.extend_from_slice(prefix); + out.extend_from_slice(&body); +} diff --git a/test/antithesis/harness/src/contexts/service_check.rs b/test/antithesis/harness/src/contexts/service_check.rs new file mode 100644 index 00000000000..080d69956a8 --- /dev/null +++ b/test/antithesis/harness/src/contexts/service_check.rs @@ -0,0 +1,132 @@ +//! Service-check contexts: the stable `_sc` identity a driver renders status, message, and timestamp +//! against. +//! +//! The Agent does not aggregate service checks, but each has a natural stable identity — name, tags, +//! and host — distinct from the per-occurrence status, message, and timestamp. The pool recurs the +//! identity; the driver varies status, message, and timestamp each render. + +use rand::{Rng, RngExt}; + +use super::TS_DIGITS; +use super::{fresh_timestamp, get_bytes, get_tags, put_bytes, put_tags}; +use crate::payload::dogstatsd::common; + +/// The status symbols: OK, warning, critical, unknown. +const STATUS: &[&[u8]] = &[b"0", b"1", b"2", b"3"]; + +/// Identity option counts: usually none, sometimes a host. +const OPT_COUNTS: &[usize] = &[0, 0, 0, 1]; + +/// A service-check identity: name, tags, and fixed option chunks (a host). Status, message, and +/// timestamp vary per render. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct ServiceCheckContext { + /// Name content, non-empty. + pub name: Vec, + /// `key:value` tags. + pub tags: Vec>, + /// Fixed option chunks (a `h:` host), each carrying its own prefix. + pub options: Vec>, +} + +impl ServiceCheckContext { + /// Mint a service-check identity that renders within `budget`, or `None` when the budget cannot + /// hold the smallest one. Name, options and tags are built against the room the render's skeleton, + /// timestamp and trailing message leave. + pub(crate) fn mint_within(rng: &mut (impl Rng + ?Sized), budget: usize) -> Option { + const RESERVED: usize = "_sc|".len() + 1 + 1 + "|d:".len() + TS_DIGITS + "|m:".len(); + let name = common::identifier_within(rng, budget.checked_sub(RESERVED)?); + if name.is_empty() { + return None; + } + let mut room = budget - RESERVED - name.len(); + let count = OPT_COUNTS[rng.random_range(0..OPT_COUNTS.len())]; + let mut options = Vec::new(); + for _ in 0..count { + let Some(body_room) = room.checked_sub(1 + "h:".len()) else { + break; + }; + let mut chunk = b"h:".to_vec(); + chunk.extend_from_slice(&common::optional_text_within(rng, body_room)); + room -= 1 + chunk.len(); + options.push(chunk); + } + Some(Self { + name, + tags: common::tags_within(rng, room), + options, + }) + } + + /// Render `_sc|name|status[|opt...]|d:ts[|#tags]|m:message` for a fresh status, message, and + /// timestamp. Returns zero (service checks carry no packed run). + /// Bytes every render of this identity must spend: `_sc|name|status`, the fixed options, the + /// timestamp, the tag set and the empty `|m:` message. Only the message body is variable. + pub(crate) fn floor(&self) -> usize { + "_sc|".len() + + self.name.len() + + 1 + + 1 + + self.options.iter().map(|opt| 1 + opt.len()).sum::() + + "|d:".len() + + TS_DIGITS + + common::tags_len(&self.tags) + + "|m:".len() + } + + /// Render within `budget`, or `None` when the budget cannot hold the identity. Only the message is + /// sampled, against the room left. + pub(crate) fn render_within( + &self, rng: &mut (impl Rng + ?Sized), out: &mut Vec, budget: usize, + ) -> Option { + let message_room = budget.checked_sub(self.floor())?; + out.extend_from_slice(b"_sc|"); + out.extend_from_slice(&self.name); + out.push(b'|'); + out.extend_from_slice(STATUS[rng.random_range(0..STATUS.len())]); + for opt in &self.options { + out.push(b'|'); + out.extend_from_slice(opt); + } + let mut itoa = itoa::Buffer::new(); + out.extend_from_slice(b"|d:"); + out.extend_from_slice(itoa.format(fresh_timestamp(rng)).as_bytes()); + common::serialize_tags(&self.tags, out); + out.extend_from_slice(b"|m:"); + out.extend_from_slice(&common::optional_text_within(rng, message_room)); + Some(0) + } + + pub(crate) fn render(&self, rng: &mut (impl Rng + ?Sized), out: &mut Vec) -> usize { + out.extend_from_slice(b"_sc|"); + out.extend_from_slice(&self.name); + out.push(b'|'); + out.extend_from_slice(STATUS[rng.random_range(0..STATUS.len())]); + for opt in &self.options { + out.push(b'|'); + out.extend_from_slice(opt); + } + let mut itoa = itoa::Buffer::new(); + out.extend_from_slice(b"|d:"); + out.extend_from_slice(itoa.format(fresh_timestamp(rng)).as_bytes()); + common::serialize_tags(&self.tags, out); + out.extend_from_slice(b"|m:"); + out.extend_from_slice(&common::optional_text(rng)); + 0 + } + + /// Append this context's length-prefixed encoding. + pub(crate) fn encode(&self, out: &mut Vec) { + put_bytes(out, &self.name); + put_tags(out, &self.tags); + put_tags(out, &self.options); + } + + /// Decode one service-check context, advancing `*pos`. + pub(crate) fn decode(buf: &[u8], pos: &mut usize) -> Option { + let name = get_bytes(buf, pos)?.to_vec(); + let tags = get_tags(buf, pos)?; + let options = get_tags(buf, pos)?; + Some(Self { name, tags, options }) + } +} diff --git a/test/antithesis/harness/src/driver.rs b/test/antithesis/harness/src/driver.rs index 1bff2d839b7..5b3ea5a5314 100644 --- a/test/antithesis/harness/src/driver.rs +++ b/test/antithesis/harness/src/driver.rs @@ -1,9 +1,10 @@ //! Shared `DogStatsD` load-driver engine. //! -//! A producer thread samples lines into a bounded channel; a consumer thread -//! fans each line out to every socket and tallies per-socket sends. Drivers -//! differ only in how many sockets they target and which anchors they fire, so -//! both the single-socket and differential drivers run on this one engine. +//! The engine fetches a working set of contexts from the shared intake pool, then a producer thread +//! renders per-occurrence payloads against them into a bounded channel while a consumer thread fans +//! each datagram out to every socket and tallies per-socket sends. Drivers differ only in how many +//! sockets they target and which anchors they fire, so both the single-socket and differential +//! drivers run on this one engine. //! //! NOTE: this driver intentionally blocks on backpressure from the SUT. Retry //! and backoff timers are meant to endure transient errors. @@ -19,12 +20,20 @@ use antithesis_sdk::prelude::*; use rand::Rng; use serde_json::json; +use crate::contexts::{decode_response, Context}; use crate::dogstatsd::is_malformed; use crate::payload::dogstatsd; const SEND_RETRY_BUDGET: Duration = Duration::from_secs(5); const SEND_RETRY_BACKOFF: Duration = Duration::from_millis(1); +/// How long to keep retrying the context fetch before giving up and running no load this invocation. +const CONTEXT_FETCH_BUDGET: Duration = Duration::from_secs(30); +/// Backoff between context-fetch attempts. +const CONTEXT_FETCH_BACKOFF: Duration = Duration::from_millis(250); +/// Per-request timeout on a single context fetch. +const CONTEXT_FETCH_TIMEOUT: Duration = Duration::from_secs(10); + /// A generated payload queued for the sockets: the packed bytes and what they hold. struct Datagram { /// The `\n`-packed payload bytes to ship over a socket. @@ -49,30 +58,51 @@ pub struct Stats { pub timed_out: bool, } -/// Drive `count` sampled `DogStatsD` datagrams to every socket, packing each to -/// at most `limit_bytes` and blocking through transient backpressure so every -/// datagram reaches every socket. Both `count` and `limit_bytes` come from a load -/// generator's [`crate::config::DriverConfig`], so a datagram never truncates on -/// receive. +impl Stats { + /// The zero result for `sockets` sockets: nothing received, nothing sent. Reported when the + /// context pool is unreachable so a driver invocation degrades to a no-op rather than an error. + fn empty(sockets: usize) -> Self { + Self { + received: 0, + sent: vec![0; sockets], + max_packed: vec![0; sockets], + timed_out: false, + } + } +} + +/// Fetch a working set of `context_count` contexts from the intake pool at `intake_addr`, then drive +/// `count` datagrams to every socket, each a fresh render of a sampled context packed to at most +/// `limit_bytes`, blocking through transient backpressure so every datagram reaches every socket. +/// `context_count`, `count`, and `limit_bytes` come from a load generator's +/// [`crate::config::DriverConfig`], so a datagram never truncates on receive. /// -/// A peer that leaves mid-batch, or backpressure that outlasts the retry budget, -/// ends the run early with a partial [`Stats`] rather than an error. +/// An unreachable pool ends the run with an empty [`Stats`] rather than an error, so a driver started +/// before the intake serves degrades to a no-op. A peer that leaves mid-batch, or backpressure that +/// outlasts the retry budget, ends the run early with a partial [`Stats`]. /// /// # Errors /// /// Errors if a worker thread panics. Sustained backpressure is reported via /// [`Stats::timed_out`], not as an error. pub fn run( - mut rng: R, limit_bytes: usize, count: usize, sockets: Vec, + mut rng: R, intake_addr: &str, context_count: usize, limit_bytes: usize, count: usize, sockets: Vec, ) -> anyhow::Result { + let contexts = match fetch_contexts(intake_addr, context_count) { + // Ordered by floor once here, not per datagram: the working set is fixed for the invocation. + Some(contexts) if !contexts.is_empty() => dogstatsd::WorkingSet::new(contexts), + // Pool unreachable or empty. No load this invocation, not a failure. + _ => return Ok(Stats::empty(sockets.len())), + }; + let (tx, rx) = sync_channel::(2024); let producer = thread::spawn(move || { for _ in 0..count { let mut bytes = Vec::new(); - let payload = dogstatsd::write_payload(&mut rng, &mut bytes, limit_bytes); - // Green by construction: write_payload emits only lines the Agent forwards. The anchor - // catches any generator drift that would ship a droppable datagram. + let payload = dogstatsd::write_payload(&mut rng, &contexts, &mut bytes, limit_bytes); + // Green by construction: write_payload packs only rendered lines the Agent forwards. The + // anchor catches any drift that would ship a droppable datagram. assert_always!( is_malformed(&bytes).is_ok(), "driver payload is well-formed", @@ -125,6 +155,37 @@ pub fn run( .map_err(|_| anyhow::anyhow!("consumer thread panicked"))? } +/// Fetch a working set of `n` contexts from the pool at `intake_addr` over blocking HTTP, retrying +/// through [`CONTEXT_FETCH_BUDGET`]. Returns `None` if the pool never answers or the body does not +/// decode, so the caller degrades to no load rather than failing. +fn fetch_contexts(intake_addr: &str, n: usize) -> Option> { + let url = format!("http://{intake_addr}/contexts?n={n}"); + let client = reqwest::blocking::Client::builder() + .timeout(CONTEXT_FETCH_TIMEOUT) + .build() + .ok()?; + let deadline = Instant::now() + CONTEXT_FETCH_BUDGET; + loop { + if let Some(contexts) = try_fetch_contexts(&client, &url) { + return Some(contexts); + } + if Instant::now() >= deadline { + return None; + } + sleep(CONTEXT_FETCH_BACKOFF); + } +} + +/// One context-fetch attempt: `None` on a transport error, a non-success status, or a body that does +/// not decode, so a partial or corrupt response is retried rather than trusted. +fn try_fetch_contexts(client: &reqwest::blocking::Client, url: &str) -> Option> { + let response = client.get(url).send().ok()?; + if !response.status().is_success() { + return None; + } + decode_response(&response.bytes().ok()?) +} + /// Outcome of delivering one line to a socket. enum Delivery { /// The line reached the socket. diff --git a/test/antithesis/harness/src/lib.rs b/test/antithesis/harness/src/lib.rs index e45640b90fd..9c675be1ebb 100644 --- a/test/antithesis/harness/src/lib.rs +++ b/test/antithesis/harness/src/lib.rs @@ -3,6 +3,7 @@ use std::time::Duration; pub mod config; +pub mod contexts; pub mod dogstatsd; #[cfg(unix)] pub mod driver; diff --git a/test/antithesis/harness/src/payload/dogstatsd.rs b/test/antithesis/harness/src/payload/dogstatsd.rs index eea855243e1..273e9764fd4 100644 --- a/test/antithesis/harness/src/payload/dogstatsd.rs +++ b/test/antithesis/harness/src/payload/dogstatsd.rs @@ -1,35 +1,23 @@ -//! `DogStatsD` payload generation. +//! `DogStatsD` payload generation from a pooled context working set. //! -//! Generation builds three structured message types, metric, event, and service check, then -//! serializes each to datagram bytes. Every message is checked with [`crate::dogstatsd::is_malformed`] -//! on its serialized bytes and repaired until the Datadog Agent would forward it. There is no -//! clean/feral/mixed configuration and no per-line vibe. The legal space is exactly the set of -//! payloads `is_malformed` accepts. Content fields range over the full forwarded alphabet including -//! delimiter bytes, and the predicate sorts the delimiter-bearing cases the Agent keeps from the ones -//! it drops. +//! A driver no longer samples a fresh identity per line. It fetches a working set of [`Context`]s +//! from the shared intake pool, [`crate::contexts`], and renders a fresh per-occurrence payload +//! against a sampled context each line, so identities recur while their load varies. Every rendered +//! line is repaired by [`Context::render_wellformed_within`] to one the Datadog Agent forwards, so a packed +//! datagram holds only whole lines the Agent keeps. There is no clean/feral/mixed configuration. The +//! legal space is exactly the set of payloads [`crate::dogstatsd::is_malformed`] accepts. //! //! ```text //! metric: :(:)*|[|@][|#][|c:..][|e:..][|card:..] -//! event: _e{,}:|<TEXT>[|h:..][|k:..][|p:..][|s:..][|t:..][|#<TAGS>] -//! service check: _sc|<NAME>|<STATUS>[|h:..][|m:..][|#<TAGS>] +//! event: _e{<TITLE_LEN>,<TEXT_LEN>}:<TITLE>|<TEXT>[|opt...]|d:<TS>[|#<TAGS>] +//! service check: _sc|<NAME>|<STATUS>[|opt...]|d:<TS>[|#<TAGS>]|m:<MESSAGE> //! ``` use rand::{Rng, RngExt}; -use crate::dogstatsd::is_malformed; +use crate::contexts::Context; -mod common; -mod events; -mod metrics; -mod service_checks; - -/// How many times to re-sample a message whose serialized bytes the Agent would drop before falling -/// back to a guaranteed-well-formed line. Construction is mostly-valid, so the fallback is rare. -const REPAIR_TRIES: usize = 16; - -/// A guaranteed-well-formed metric, used only if repair is exhausted, so the pack loop always -/// terminates. -const FALLBACK_LINE: &[u8] = b"harness.fallback:1|c"; +pub(crate) mod common; /// Ceiling on a generated datagram, the Datadog Agent's default `dogstatsd_buffer_size`. A run caps /// each datagram to the smaller of this and the SUT's sampled receive buffer, so a packed datagram @@ -45,79 +33,80 @@ pub struct Payload { pub max_packed: usize, } -/// One of the three `DogStatsD` message types, structured. -enum Message { - Metric(metrics::Metric), - Event(events::Event), - ServiceCheck(service_checks::ServiceCheck), +/// How many affordable contexts a line tries before the datagram ends. A context whose occurrence +/// payloads keep landing on the drop side is one to move past, not a reason to stop packing. +const RENDER_PICKS: usize = 4; + +/// A driver's working set, ordered by how many bytes each identity forces on every render. +/// +/// The order is computed once per fetch. Packing a line then finds the affordable contexts by +/// partition point rather than filtering the whole set, which matters because one invocation packs +/// thousands of datagrams from a set of up to a thousand identities. +#[derive(Clone, Debug)] +pub struct WorkingSet { + /// Contexts by ascending floor. + contexts: Vec<Context>, + /// `contexts[i].floor()`, same order, so the affordable contexts are a prefix. + floors: Vec<usize>, } -impl Message { - /// Sample a message type. The mix is heavily metric-weighted: 98% metric, 1% event, 1% service - /// check. Metrics drive the aggregate context and sketch paths, so the bulk of load goes there - /// while events and service checks still fire often enough to keep their anchors non-vacuous. - fn generate(rng: &mut (impl Rng + ?Sized), budget: usize) -> Option<Self> { - match rng.random_range(0..100u32) { - 0 => events::Event::generate(rng, budget).map(Message::Event), - 1 => service_checks::ServiceCheck::generate(rng, budget).map(Message::ServiceCheck), - _ => metrics::Metric::generate(rng, budget).map(Message::Metric), - } +impl WorkingSet { + /// Order `contexts` by floor and record each one. + #[must_use] + pub fn new(mut contexts: Vec<Context>) -> Self { + contexts.sort_by_key(Context::floor); + let floors = contexts.iter().map(Context::floor).collect(); + Self { contexts, floors } } - /// Serialize to datagram bytes, without the trailing `\n`. - fn serialize(&self, out: &mut Vec<u8>) { - match self { - Message::Metric(m) => m.serialize(out), - Message::Event(e) => e.serialize(out), - Message::ServiceCheck(s) => s.serialize(out), - } + /// Whether the set holds nothing. + #[must_use] + pub fn is_empty(&self) -> bool { + self.contexts.is_empty() } - /// The packed multi-value run length this message contributes, or zero. - fn packed(&self) -> usize { - match self { - Message::Metric(m) => m.packed(), - Message::Event(_) | Message::ServiceCheck(_) => 0, - } + /// The smallest floor in the set, or `None` when the set is empty. + #[must_use] + pub fn smallest_floor(&self) -> Option<usize> { + self.floors.first().copied() } -} -/// Sample one message and serialize it to bytes the Agent forwards, within `budget` bytes including -/// the trailing `\n`. Re-samples up to [`REPAIR_TRIES`] times a message the Agent would drop or one -/// that overruns the budget, then falls back to [`FALLBACK_LINE`]. Returns `None` only when even the -/// fallback does not fit. -/// -/// The budget is a construction input, not a filter applied afterwards: a sample that does not fit is -/// another sample to take, never a reason to end the payload. Ending it would leave the datagram -/// short, or empty when the very first sample overran, and a short datagram is workload the SUT never -/// sees while the driver still spends one of its configured sends. -fn wellformed_line(rng: &mut (impl Rng + ?Sized), budget: usize) -> Option<(Vec<u8>, usize)> { - for _ in 0..REPAIR_TRIES { - // The message is built against the budget, so it fits by construction. The retry is only for - // the Agent's own drop rules, which are about content rather than size. - let Some(message) = Message::generate(rng, budget.saturating_sub(1)) else { - break; - }; - let mut bytes = Vec::new(); - message.serialize(&mut bytes); - if bytes.len() < budget && is_malformed(&bytes).is_ok() { - return Some((bytes, message.packed())); - } + /// Pick uniformly among the contexts whose identity fits `budget`, or `None` when none does. + fn pick(&self, rng: &mut (impl Rng + ?Sized), budget: usize) -> Option<&Context> { + let affordable = self.floors.partition_point(|&floor| floor <= budget); + (affordable > 0).then(|| &self.contexts[rng.random_range(0..affordable)]) } - (FALLBACK_LINE.len() < budget).then(|| (FALLBACK_LINE.to_vec(), 0)) } -/// Pack whole `\n`-terminated lines into `buf` until the budget cannot hold another line. Each line is -/// sampled against the budget still free, so the datagram fills rather than ending on the first -/// oversized sample, never exceeds `limit_bytes`, and holds only whole lines every one of which the -/// Agent forwards. Clears `buf` first. -pub fn write_payload(rng: &mut (impl Rng + ?Sized), buf: &mut Vec<u8>, limit_bytes: usize) -> Payload { +/// Pack whole `\n`-terminated lines into `buf`, each a fresh render of a context that fits the room +/// left under `limit_bytes`. A context is picked from those the remaining budget can hold and rendered +/// against that budget, so the datagram fills instead of ending on the first oversized render and +/// nothing is built and thrown away. A context whose occurrence payloads all land on the drop side is +/// passed over for another, up to [`RENDER_PICKS`], rather than ending the datagram. Clears `buf` +/// first. An empty working set, or a budget too small for any identity, yields an empty payload. +pub fn write_payload( + rng: &mut (impl Rng + ?Sized), contexts: &WorkingSet, buf: &mut Vec<u8>, limit_bytes: usize, +) -> Payload { buf.clear(); let mut payload = Payload::default(); + let mut line = Vec::new(); loop { - // The budget left, `\n` included. Every accepted line consumes at least one byte, so the - // remaining budget strictly shrinks and the loop ends once nothing fits. - let Some((line, packed)) = wellformed_line(rng, limit_bytes - buf.len()) else { + // `\n` is part of what a line costs. + let Some(budget) = limit_bytes.checked_sub(buf.len() + 1) else { + break; + }; + let mut rendered = None; + for _ in 0..RENDER_PICKS { + let Some(context) = contexts.pick(rng, budget) else { + break; + }; + line.clear(); + if let Some(packed) = context.render_wellformed_within(rng, &mut line, budget) { + rendered = Some(packed); + break; + } + } + let Some(packed) = rendered else { break; }; buf.extend_from_slice(&line); @@ -134,9 +123,19 @@ mod test { use rand::rngs::SmallRng; use rand::SeedableRng; - use super::{write_payload, FALLBACK_LINE, PAYLOAD_BYTE_LIMIT}; + use super::{write_payload, WorkingSet, PAYLOAD_BYTE_LIMIT}; + use crate::contexts::{Context, Kind}; use crate::dogstatsd::is_malformed; + /// A working set of minted contexts across all three kinds. + fn pool(rng: &mut SmallRng, n: usize) -> WorkingSet { + WorkingSet::new( + (0..n) + .filter_map(|_| Context::mint_within(Kind::sample(rng), rng, PAYLOAD_BYTE_LIMIT)) + .collect(), + ) + } + /// Lines carry no interior newline and each is `\n`-terminated, so the line count equals the /// newline count. #[allow(clippy::naive_bytecount)] @@ -145,24 +144,78 @@ mod test { } proptest! { - /// Every datagram the generator emits is one the Agent forwards. A packed datagram must be - /// entirely well-formed. + /// A rendered metric carries exactly the tag set its identity holds. The Agent assigns the tag + /// set from every `#`-prefixed optional field it sees, last one winning, so a second such field + /// would swap the identity's tags for whatever an occurrence body happened to contain. That + /// would make one pooled identity render as several, putting cardinality past its cap and one + /// point in each of many series. Occurrence bodies do carry `|` and `#`, but segments are + /// always joined by a separator, so the two never land adjacent. This pins that. + #[test] + fn property_test_metric_carries_one_tag_field(seed: u64) { + let mut rng = SmallRng::seed_from_u64(seed); + let contexts = pool(&mut rng, 8); + let mut buf = Vec::new(); + for _ in 0..4 { + write_payload(&mut rng, &contexts, &mut buf, PAYLOAD_BYTE_LIMIT); + for line in buf.split(|&b| b == b'\n') { + if line.is_empty() || line.starts_with(b"_e{") || line.starts_with(b"_sc") { + continue; + } + // Field 0 is `name:value` and field 1 the type. The Agent reads a tag set only from + // the optional fields after those, so a `#` opening either of the first two is not + // one. + let tag_fields = line + .split(|&b| b == b'|') + .skip(2) + .filter(|field| field.first() == Some(&b'#')) + .count(); + prop_assert!( + tag_fields <= 1, + "{tag_fields} tag fields in {:?}", + String::from_utf8_lossy(line) + ); + } + } + } + + /// Every datagram the driver packs from pooled contexts is one the Agent forwards. A packed + /// datagram must be entirely well-formed. #[test] fn property_test_every_payload_is_well_formed(seed: u64) { let mut rng = SmallRng::seed_from_u64(seed); + let contexts = pool(&mut rng, 8); let mut buf = Vec::new(); for _ in 0..8 { - write_payload(&mut rng, &mut buf, PAYLOAD_BYTE_LIMIT); + write_payload(&mut rng, &contexts, &mut buf, PAYLOAD_BYTE_LIMIT); prop_assert_eq!(is_malformed(&buf), Ok(()), "emitted a droppable datagram: {:?}", String::from_utf8_lossy(&buf)); } } + /// A budget that can hold some context's identity yields load. An empty datagram spends one of + /// the driver's configured sends without reaching the SUT, so the packer must fill the budget it + /// was given rather than give up on the first context that does not fit. The fallback line is + /// the floor below which a render that keeps landing on the drop side has nowhere to go. + #[test] + fn property_test_payload_is_never_empty_when_a_context_fits(seed: u64, limit_bytes in 1..8_192usize) { + let mut rng = SmallRng::seed_from_u64(seed); + let contexts = pool(&mut rng, 8); + let mut buf = Vec::new(); + let payload = write_payload(&mut rng, &contexts, &mut buf, limit_bytes); + + let smallest = contexts.smallest_floor().expect("pool is non-empty"); + if limit_bytes > smallest { + prop_assert!(!buf.is_empty(), "empty datagram at limit {limit_bytes}, smallest floor {smallest}"); + prop_assert!(payload.lines > 0); + } + } + #[test] fn property_test_payload_stays_within_its_limit(seed: u64, limit_bytes: u16) { let mut rng = SmallRng::seed_from_u64(seed); + let contexts = pool(&mut rng, 8); let limit_bytes = usize::from(limit_bytes); let mut buf = Vec::new(); - let payload = write_payload(&mut rng, &mut buf, limit_bytes); + let payload = write_payload(&mut rng, &contexts, &mut buf, limit_bytes); prop_assert!(buf.len() <= limit_bytes); prop_assert_eq!(newline_count(&buf), payload.lines); @@ -171,17 +224,14 @@ mod test { } } - /// A budget that admits any line at all yields load. An empty datagram burns one of the - /// driver's configured sends without reaching the SUT, so the generator must fill the budget - /// it was given rather than give up on the first oversized sample. + /// An empty working set yields no load, not a panic. #[test] - fn property_test_payload_is_never_empty_when_a_line_fits(seed: u64, limit_bytes in (FALLBACK_LINE.len() + 1)..4096usize) { + fn property_test_empty_pool_yields_empty_payload(seed: u64) { let mut rng = SmallRng::seed_from_u64(seed); let mut buf = Vec::new(); - let payload = write_payload(&mut rng, &mut buf, limit_bytes); - - prop_assert!(!buf.is_empty(), "empty datagram at limit {}", limit_bytes); - prop_assert!(payload.lines > 0); + let payload = write_payload(&mut rng, &WorkingSet::new(Vec::new()), &mut buf, PAYLOAD_BYTE_LIMIT); + prop_assert_eq!(payload.lines, 0); + prop_assert!(buf.is_empty()); } } } diff --git a/test/antithesis/harness/src/payload/dogstatsd/common.rs b/test/antithesis/harness/src/payload/dogstatsd/common.rs index 5905cd578d4..828c99bd35a 100644 --- a/test/antithesis/harness/src/payload/dogstatsd/common.rs +++ b/test/antithesis/harness/src/payload/dogstatsd/common.rs @@ -102,26 +102,31 @@ const RATE_TOKEN: &[&[u8]] = &[ b"-inf", b"nan", ]; -/// Segment counts for a required field: at least one, with a large-boundary tail so huge fields stay -/// in the explored surface. -const COUNTS_REQUIRED: &[usize] = &[1, 1, 2, 2, 3, 3, 4, 5, 6, 127, 255]; +/// Segment counts for a required field: at least one, a small body. Field length crosses the intake +/// byte caps through the long-field path below, not through a huge segment count, so a pooled context +/// never compounds into a behemoth. +const COUNTS_REQUIRED: &[usize] = &[1, 1, 2, 2, 3, 3, 4, 5, 6, 7, 8]; /// Segment counts for an optional field: the required body plus zero. -const COUNTS_OPTIONAL: &[usize] = &[0, 1, 1, 2, 2, 3, 3, 4, 5, 6, 127, 255]; +const COUNTS_OPTIONAL: &[usize] = &[0, 1, 1, 2, 2, 3, 3, 4, 5, 6, 7, 8]; + +/// Tag counts. Reaches past `MaxTags` (100) so the tag-count cap is exercised, but bounded so a +/// context stays a few KiB even at the high end. +const TAG_COUNTS: &[usize] = &[0, 1, 1, 2, 2, 3, 3, 4, 5, 6, 16, 64, 101, 127]; + +/// Byte-length targets for a long field, crossing `MaxTagLength` (200) and the metric-name cap (350) +/// directly rather than via a huge segment count. +const LONG_TARGETS: &[usize] = &[128, 199, 200, 201, 256, 300, 349, 350, 351, 400]; /// Pick one item from a static, non-empty pool by index. fn pick<'a>(rng: &mut (impl Rng + ?Sized), pool: &[&'a [u8]]) -> &'a [u8] { pool[rng.random_range(0..pool.len())] } -/// The shortest item `pools` can yield, which is the floor cost of one more segment. -pub(crate) fn min_item(pools: &[&[&[u8]]]) -> usize { - pools - .iter() - .flat_map(|pool| pool.iter()) - .map(|item| item.len()) - .min() - .unwrap_or(0) +/// Pick one item from the union of static, non-empty pools by index. +fn pick_union<'a>(rng: &mut (impl Rng + ?Sized), pools: &[&[&'a [u8]]]) -> &'a [u8] { + let pool = pools[rng.random_range(0..pools.len())]; + pick(rng, pool) } /// Pick an item no longer than `budget`, or `None` when the pools hold nothing that small. Counts the @@ -149,6 +154,30 @@ fn pick_within<'a>(rng: &mut (impl Rng + ?Sized), pools: &[&[&'a [u8]]], budget: /// Sampling against the room left is what keeps the generator from building a field it would have to /// throw away. fn field_within(rng: &mut (impl Rng + ?Sized), pools: &[&[&[u8]]], counts: &[usize], budget: usize) -> Vec<u8> { + // One in eight: fill to a byte target that crosses the intake length caps, taking the largest + // target the budget affords. Without this an identity built against a budget never reaches the + // 350-byte name or 200-byte tag boundary, since the segment counts alone do not get there. + if rng.random_range(0..8u8) == 0 { + let affordable = LONG_TARGETS.iter().filter(|&&target| target <= budget).count(); + if affordable > 0 { + let target = LONG_TARGETS[rng.random_range(0..affordable)]; + let mut out = Vec::new(); + while out.len() < target { + // Separated like the segment run below. Concatenating raw pool items would let a `|` + // and a `#` land adjacent, which opens a second tags field and swaps the identity's + // tag set for occurrence content. + let separator = usize::from(!out.is_empty()); + let Some(item) = pick_within(rng, pools, (target - out.len()).saturating_sub(separator)) else { + break; + }; + if !out.is_empty() { + out.push(NAME_SEPARATORS[rng.random_range(0..NAME_SEPARATORS.len())]); + } + out.extend_from_slice(item); + } + return out; + } + } let count = counts[rng.random_range(0..counts.len())]; let mut out = Vec::new(); for i in 0..count { @@ -165,22 +194,55 @@ fn field_within(rng: &mut (impl Rng + ?Sized), pools: &[&[&[u8]]], counts: &[usi out } +/// Build a field of separator-joined segments drawn from `pools`, with the segment count sampled from +/// `counts`. A `counts` slice containing `0` can yield an empty field. Used to mint an identity, which +/// carries no budget: the budget applies when a context is rendered, not when it is minted. +fn field(rng: &mut (impl Rng + ?Sized), pools: &[&[&[u8]]], counts: &[usize]) -> Vec<u8> { + // One in eight: a long field filled to a byte target that crosses the intake length caps. This + // reaches the same length surface as a huge segment count without the behemoth a pooled context + // cannot afford. + if rng.random_range(0..8u8) == 0 { + let target = LONG_TARGETS[rng.random_range(0..LONG_TARGETS.len())]; + let mut out = Vec::new(); + while out.len() < target { + if !out.is_empty() { + out.push(NAME_SEPARATORS[rng.random_range(0..NAME_SEPARATORS.len())]); + } + out.extend_from_slice(pick_union(rng, pools)); + } + return out; + } + let count = counts[rng.random_range(0..counts.len())]; + let mut out = Vec::new(); + for i in 0..count { + if i > 0 { + out.push(NAME_SEPARATORS[rng.random_range(0..NAME_SEPARATORS.len())]); + } + out.extend_from_slice(pick_union(rng, pools)); + } + out +} + +/// The shortest item `pools` can yield, the floor cost of one more segment. +fn min_item(pools: &[&[&[u8]]]) -> usize { + pools + .iter() + .flat_map(|pool| pool.iter()) + .map(|item| item.len()) + .min() + .unwrap_or(0) +} + /// A required identifier within `budget`. Empty when the budget cannot hold one segment, which the -/// caller must treat as "no line fits" rather than emitting an invalid name. +/// caller treats as "no identity fits" rather than minting an invalid name. pub(crate) fn identifier_within(rng: &mut (impl Rng + ?Sized), budget: usize) -> Vec<u8> { field_within(rng, WORD_POOLS, COUNTS_REQUIRED, budget) } -/// An optional free-text field within `budget`. -pub(crate) fn optional_text_within(rng: &mut (impl Rng + ?Sized), budget: usize) -> Vec<u8> { - field_within(rng, WORD_POOLS, COUNTS_OPTIONAL, budget) -} - /// A tag set serialized within `budget` bytes, `|#` and separating commas included. Each tag is built -/// against the room left, and the run stops when the next one cannot fit. +/// against the room left and the run stops when the next one cannot fit. pub(crate) fn tags_within(rng: &mut (impl Rng + ?Sized), budget: usize) -> Vec<Vec<u8>> { - let count = COUNTS_OPTIONAL[rng.random_range(0..COUNTS_OPTIONAL.len())]; - // `|#` before the first tag, then a comma before each later one. + let count = TAG_COUNTS[rng.random_range(0..TAG_COUNTS.len())]; let Some(mut room) = budget.checked_sub(2) else { return Vec::new(); }; @@ -190,7 +252,6 @@ pub(crate) fn tags_within(rng: &mut (impl Rng + ?Sized), budget: usize) -> Vec<V let Some(tag_room) = room.checked_sub(separator) else { break; }; - // A tag is `key:value` with a required key, so it needs a segment plus the colon. if tag_room < min_item(TAG_KEY_POOLS) + 1 { break; } @@ -208,6 +269,16 @@ pub(crate) fn tags_within(rng: &mut (impl Rng + ?Sized), budget: usize) -> Vec<V out } +/// An optional free-text field for an identity. +pub(crate) fn optional_text(rng: &mut (impl Rng + ?Sized)) -> Vec<u8> { + field(rng, WORD_POOLS, COUNTS_OPTIONAL) +} + +/// An optional free-text field within `budget`. +pub(crate) fn optional_text_within(rng: &mut (impl Rng + ?Sized), budget: usize) -> Vec<u8> { + field_within(rng, WORD_POOLS, COUNTS_OPTIONAL, budget) +} + /// Serialize a tag set as `|#key:value,key:value`. An empty set appends nothing. pub(crate) fn serialize_tags(tags: &[Vec<u8>], out: &mut Vec<u8>) { for (i, tag) in tags.iter().enumerate() { @@ -238,6 +309,16 @@ pub(crate) fn float_token(rng: &mut (impl Rng + ?Sized)) -> Vec<u8> { } } +/// A sample-rate token. +pub(crate) fn rate_token(rng: &mut (impl Rng + ?Sized)) -> Vec<u8> { + pick(rng, RATE_TOKEN).to_vec() +} + +/// A set-type value: an opaque required field over the full alphabet. +pub(crate) fn opaque_value(rng: &mut (impl Rng + ?Sized)) -> Vec<u8> { + field(rng, WORD_POOLS, COUNTS_REQUIRED) +} + /// The floor cost of a value token, the shortest `SPECIAL_VALUE` entry. pub(crate) fn min_value_token() -> usize { SPECIAL_VALUE.iter().map(|item| item.len()).min().unwrap_or(1) @@ -307,3 +388,33 @@ fn pick_padding_run(rng: &mut (impl Rng + ?Sized)) -> u8 { const RUNS: &[u8] = &[0, 1, 2, 8, 16, 32, 64, 127]; RUNS[rng.random_range(0..RUNS.len())] } + +#[cfg(test)] +mod tests { + use rand::rngs::SmallRng; + use rand::SeedableRng; + + use super::{identifier_within, tags_within}; + + // A field built against a budget must still reach the intake's length boundaries, 350 bytes for a + // metric name and 200 for a tag. The segment counts stop well short of both, so the long-target arm + // is what gets there, and giving the builder a budget must not cost that coverage. + #[test] + fn budgeted_fields_reach_the_length_boundaries() { + let mut max_name = 0; + let mut max_tag = 0; + for seed in 0..512u64 { + let mut rng = SmallRng::seed_from_u64(seed); + max_name = max_name.max(identifier_within(&mut rng, 8_000).len()); + max_tag = max_tag.max(tags_within(&mut rng, 8_000).iter().map(Vec::len).max().unwrap_or(0)); + } + assert!( + max_name > 350, + "longest field {max_name} never crossed the 350-byte name cap" + ); + assert!( + max_tag > 200, + "longest tag {max_tag} never crossed the 200-byte tag cap" + ); + } +} diff --git a/test/antithesis/harness/src/payload/dogstatsd/events.rs b/test/antithesis/harness/src/payload/dogstatsd/events.rs deleted file mode 100644 index 0334a5113b6..00000000000 --- a/test/antithesis/harness/src/payload/dogstatsd/events.rs +++ /dev/null @@ -1,91 +0,0 @@ -//! Structured `DogStatsD` event generation. - -use rand::{Rng, RngExt}; - -use super::common; - -/// Optional-field prefixes: hostname, aggregation key, priority, source type, alert type. Bad -/// priority and alert values are logged and defaulted by the Agent, never dropped, so any content -/// forwards. -const OPT_PREFIXES: &[&[u8]] = &[b"h:", b"k:", b"p:", b"s:", b"t:"]; - -/// Optional-field counts: mostly none, with a boundary tail. -const OPT_COUNTS: &[usize] = &[0, 0, 0, 0, 1, 1, 2, 3, 127, 255]; - -/// A structured event: `_e{title_len,text_len}:title|text[|opt...]`. -#[derive(Clone, Debug)] -pub(crate) struct Event { - /// Title content, required non-empty. - title: Vec<u8>, - /// Text content, may be empty. - text: Vec<u8>, - /// `key:value` tags. - tags: Vec<Vec<u8>>, - /// Optional chunks, each carrying its own prefix. - options: Vec<Vec<u8>>, -} - -impl Event { - /// Sample an event with content over the full forwarded alphabet. The body is length-delimited, - /// so title and text carry any bytes including delimiters. - pub(crate) fn generate(rng: &mut (impl Rng + ?Sized), budget: usize) -> Option<Self> { - // `_e{N,M}:title|text` is the skeleton. The header digits grow with the field lengths, so a - // conservative four bytes per length keeps the estimate above what serialization will write for - // any field this generator can build. - const HEADER: usize = "_e{".len() + 4 + 1 + 4 + "}:".len() + 1; - let title_room = budget.checked_sub(HEADER)?; - let title = common::identifier_within(rng, title_room); - if title.is_empty() { - return None; - } - let spent = HEADER + title.len(); - let text = common::optional_text_within(rng, budget.saturating_sub(spent)); - let spent = spent + text.len(); - let tags = common::tags_within(rng, budget.saturating_sub(spent)); - let spent = spent + common::tags_len(&tags); - Some(Self { - title, - text, - tags, - options: options(rng, budget.saturating_sub(spent)), - }) - } - - /// Serialize the event to datagram bytes, without the trailing `\n`. The header lengths are the - /// true byte lengths of the title and text, so the Agent never rejects on a length mismatch. - pub(crate) fn serialize(&self, out: &mut Vec<u8>) { - let mut itoa = itoa::Buffer::new(); - out.extend_from_slice(b"_e{"); - out.extend_from_slice(itoa.format(self.title.len()).as_bytes()); - out.push(b','); - out.extend_from_slice(itoa.format(self.text.len()).as_bytes()); - out.extend_from_slice(b"}:"); - out.extend_from_slice(&self.title); - out.push(b'|'); - out.extend_from_slice(&self.text); - common::serialize_tags(&self.tags, out); - for opt in &self.options { - out.push(b'|'); - out.extend_from_slice(opt); - } - } -} - -/// Sample a run of optional chunks. -fn options(rng: &mut (impl Rng + ?Sized), budget: usize) -> Vec<Vec<u8>> { - let count = OPT_COUNTS[rng.random_range(0..OPT_COUNTS.len())]; - let mut out = Vec::new(); - let mut room = budget; - for _ in 0..count { - let prefix = OPT_PREFIXES[rng.random_range(0..OPT_PREFIXES.len())]; - // Each chunk carries a leading `|` on the wire. - let Some(body_room) = room.checked_sub(1 + prefix.len()) else { - break; - }; - let mut chunk = prefix.to_vec(); - chunk.extend_from_slice(&common::optional_text_within(rng, body_room)); - room -= 1 + chunk.len(); - out.push(chunk); - } - out -} diff --git a/test/antithesis/harness/src/payload/dogstatsd/metrics.rs b/test/antithesis/harness/src/payload/dogstatsd/metrics.rs deleted file mode 100644 index 9c27bb2573f..00000000000 --- a/test/antithesis/harness/src/payload/dogstatsd/metrics.rs +++ /dev/null @@ -1,157 +0,0 @@ -//! Structured `DogStatsD` metric generation. - -use rand::{Rng, RngExt}; - -use super::common; - -/// The metric types: count, gauge, timing, histogram, set, distribution. -const METRIC_TYPES: &[&[u8]] = &[b"c", b"g", b"ms", b"h", b"s", b"d"]; - -/// Extension-field counts: mostly none, with a boundary tail. -const EXT_COUNTS: &[usize] = &[0, 0, 0, 0, 1, 1, 2, 3, 127, 255]; - -/// A structured metric: `name:value[:value...]|type[|ext...]`. -#[derive(Clone, Debug)] -pub(crate) struct Metric { - /// Metric name content. - name: Vec<u8>, - /// One value for a set, else a `:`-packed run of Go-float tokens. - values: Vec<Vec<u8>>, - /// The type symbol. - kind: &'static [u8], - /// `key:value` tags. - tags: Vec<Vec<u8>>, - /// Extension chunks, each carrying its own prefix: `@`, `c:`, `e:`, or `card:`. - extensions: Vec<Vec<u8>>, -} - -impl Metric { - /// Sample a metric with content over the full forwarded alphabet. - pub(crate) fn generate(rng: &mut (impl Rng + ?Sized), budget: usize) -> Option<Self> { - let kind = METRIC_TYPES[rng.random_range(0..METRIC_TYPES.len())]; - // `name:value|kind` is the whole of a valid metric, so the name is built against what is left - // once the rest of that skeleton is reserved. - let reserved = 1 + common::min_value_token() + 1 + kind.len(); - let name_room = budget.checked_sub(reserved)?; - let name = common::identifier_within(rng, name_room); - if name.is_empty() { - return None; - } - let mut spent = name.len() + reserved; - let values = if kind == b"s" { - vec![common::opaque_value_within( - rng, - budget - spent + common::min_value_token(), - )] - } else { - // The first value is already reserved. Each extra one costs a `:` and its own bytes. - let mut values = vec![common::float_token_within( - rng, - budget - spent + common::min_value_token(), - )]; - spent = spent - common::min_value_token() + values[0].len(); - for _ in 1..value_count(rng) { - let room = budget.saturating_sub(spent + 1); - let value = common::float_token_within(rng, room); - if value.is_empty() { - break; - } - spent += 1 + value.len(); - values.push(value); - } - values - }; - if values.iter().any(Vec::is_empty) { - return None; - } - let spent: usize = - name.len() + 1 + values.iter().map(Vec::len).sum::<usize>() + values.len() - 1 + 1 + kind.len(); - let tags = common::tags_within(rng, budget.saturating_sub(spent)); - let spent = spent + common::tags_len(&tags); - Some(Self { - name, - values, - kind, - tags, - extensions: extensions(rng, budget.saturating_sub(spent)), - }) - } - - /// Serialize the metric to datagram bytes, without the trailing `\n`. - pub(crate) fn serialize(&self, out: &mut Vec<u8>) { - out.extend_from_slice(&self.name); - out.push(b':'); - for (i, value) in self.values.iter().enumerate() { - if i > 0 { - out.push(b':'); - } - out.extend_from_slice(value); - } - out.push(b'|'); - out.extend_from_slice(self.kind); - common::serialize_tags(&self.tags, out); - for ext in &self.extensions { - out.push(b'|'); - out.extend_from_slice(ext); - } - } - - /// The packed multi-value run length, or zero for a single value or a set. - pub(crate) fn packed(&self) -> usize { - if self.kind != b"s" && self.values.len() > 1 { - self.values.len() - } else { - 0 - } - } -} - -/// The `:`-packed value run length. Overwhelmingly one, with a short tail. -fn value_count(rng: &mut (impl Rng + ?Sized)) -> usize { - match rng.random_range(0..800u16) { - 0..792 => 1, - 792..796 => 2, - 796..798 => 3, - 798 => 4, - _ => 5, - } -} - -/// Sample a run of extension chunks. -fn extensions(rng: &mut (impl Rng + ?Sized), budget: usize) -> Vec<Vec<u8>> { - let count = EXT_COUNTS[rng.random_range(0..EXT_COUNTS.len())]; - let mut out = Vec::new(); - let mut room = budget; - for _ in 0..count { - // Each chunk carries a leading `|`. - let Some(chunk_room) = room.checked_sub(1) else { - break; - }; - let Some(chunk) = ext_chunk(rng, chunk_room) else { - break; - }; - room -= 1 + chunk.len(); - out.push(chunk); - } - out -} - -/// One extension chunk with its prefix. The `@` rate is a parseable token. The origin chunks carry -/// free content the Agent never drops on. -fn ext_chunk(rng: &mut (impl Rng + ?Sized), budget: usize) -> Option<Vec<u8>> { - let prefix: &[u8] = match rng.random_range(0..4u8) { - 0 => b"@", - 1 => b"c:", - 2 => b"e:", - _ => b"card:", - }; - let body_room = budget.checked_sub(prefix.len())?; - let body = if prefix == b"@" { - common::rate_token_within(rng, body_room)? - } else { - common::optional_text_within(rng, body_room) - }; - let mut chunk = prefix.to_vec(); - chunk.extend_from_slice(&body); - Some(chunk) -} diff --git a/test/antithesis/harness/src/payload/dogstatsd/service_checks.rs b/test/antithesis/harness/src/payload/dogstatsd/service_checks.rs deleted file mode 100644 index fe2c6908591..00000000000 --- a/test/antithesis/harness/src/payload/dogstatsd/service_checks.rs +++ /dev/null @@ -1,81 +0,0 @@ -//! Structured `DogStatsD` service-check generation. - -use rand::{Rng, RngExt}; - -use super::common; - -/// Status symbols: OK, warning, critical, unknown. -const STATUS: &[&[u8]] = &[b"0", b"1", b"2", b"3"]; - -/// Optional-field prefixes: hostname, message. -const OPT_PREFIXES: &[&[u8]] = &[b"h:", b"m:"]; - -/// Optional-field counts: mostly none, with a boundary tail. -const OPT_COUNTS: &[usize] = &[0, 0, 0, 0, 1, 1, 2, 3, 127, 255]; - -/// A structured service check: `_sc|name|status[|opt...]`. -#[derive(Clone, Debug)] -pub(crate) struct ServiceCheck { - /// Name content, required non-empty. - name: Vec<u8>, - /// The status symbol, always one of `0` `1` `2` `3`. - status: &'static [u8], - /// `key:value` tags. - tags: Vec<Vec<u8>>, - /// Optional chunks, each carrying its own prefix. - options: Vec<Vec<u8>>, -} - -impl ServiceCheck { - /// Sample a service check with content over the full forwarded alphabet. - pub(crate) fn generate(rng: &mut (impl Rng + ?Sized), budget: usize) -> Option<Self> { - // `_sc|name|status` is the skeleton, with a one-byte status. - const SKELETON: usize = "_sc|".len() + 1 + 1; - let name_room = budget.checked_sub(SKELETON)?; - let name = common::identifier_within(rng, name_room); - if name.is_empty() { - return None; - } - let spent = SKELETON + name.len(); - let tags = common::tags_within(rng, budget.saturating_sub(spent)); - let spent = spent + common::tags_len(&tags); - Some(Self { - name, - status: STATUS[rng.random_range(0..STATUS.len())], - tags, - options: options(rng, budget.saturating_sub(spent)), - }) - } - - /// Serialize the service check to datagram bytes, without the trailing `\n`. - pub(crate) fn serialize(&self, out: &mut Vec<u8>) { - out.extend_from_slice(b"_sc|"); - out.extend_from_slice(&self.name); - out.push(b'|'); - out.extend_from_slice(self.status); - common::serialize_tags(&self.tags, out); - for opt in &self.options { - out.push(b'|'); - out.extend_from_slice(opt); - } - } -} - -/// Sample a run of optional chunks. -fn options(rng: &mut (impl Rng + ?Sized), budget: usize) -> Vec<Vec<u8>> { - let count = OPT_COUNTS[rng.random_range(0..OPT_COUNTS.len())]; - let mut out = Vec::new(); - let mut room = budget; - for _ in 0..count { - let prefix = OPT_PREFIXES[rng.random_range(0..OPT_PREFIXES.len())]; - // Each chunk carries a leading `|` on the wire. - let Some(body_room) = room.checked_sub(1 + prefix.len()) else { - break; - }; - let mut chunk = prefix.to_vec(); - chunk.extend_from_slice(&common::optional_text_within(rng, body_room)); - room -= 1 + chunk.len(); - out.push(chunk); - } - out -} diff --git a/test/antithesis/intake/Cargo.toml b/test/antithesis/intake/Cargo.toml index d7a1b67ef66..6220a0dc8bd 100644 --- a/test/antithesis/intake/Cargo.toml +++ b/test/antithesis/intake/Cargo.toml @@ -16,13 +16,15 @@ workspace = true [dependencies] antithesis_sdk = { workspace = true, features = ["full"] } anyhow = { workspace = true, features = ["std"] } -axum = { workspace = true, features = ["http1", "json", "tokio", "tracing"] } +axum = { workspace = true, features = ["http1", "json", "query", "tokio", "tracing"] } clap = { workspace = true, features = ["derive", "env", "error-context", "help", "std", "usage"] } datadog-protos = { workspace = true } +harness = { path = "../harness" } headers = { workspace = true } http-body-util = { workspace = true } mime = { workspace = true } protobuf = { workspace = true } +rand = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } serde_yaml = { workspace = true } diff --git a/test/antithesis/intake/src/bin/intake.rs b/test/antithesis/intake/src/bin/intake.rs index 0d0c16cbb37..d09e654e493 100644 --- a/test/antithesis/intake/src/bin/intake.rs +++ b/test/antithesis/intake/src/bin/intake.rs @@ -6,9 +6,11 @@ mod unix_intake { use std::future::IntoFuture; use std::path::PathBuf; + use std::sync::Arc; use antithesis_intake::{ capture::State, + context_pool::Pool, http::{build_router, state::AppState}, }; use antithesis_sdk::prelude::*; @@ -31,9 +33,11 @@ mod unix_intake { /// Optional ADP-target HTTP bind address. #[arg(long = "adp-listen-addr", env = "ADP_LISTEN_ADDR")] adp_listen_addr: Option<String>, - /// Directory holding the timeline's sampled `datadog.yaml`. - #[arg(long = "agent-config-dir", env = "AGENT_CONFIG_DIR", default_value = "/agent-config")] - agent_config_dir: PathBuf, + /// Directory holding this timeline's sampled `datadog.yaml` and `context_source.yaml`. The + /// former backs Pyld60, the latter resolves the pool's per-kind caps on the first `/contexts` + /// request. + #[arg(long = "config-dir", env = "CONFIG_DIR", default_value = "/agent-config")] + config_dir: PathBuf, } #[tokio::main] @@ -64,6 +68,8 @@ mod unix_intake { spawn_signal_handlers(shutdown_tx).context("Failed to configure signal handlers.")?; let capture = State::new(); + // One pool backs every lane, so the drivers draw recurring identities across lanes. + let pool = Arc::new(Pool::new(config.config_dir.clone())); if let (Some(agent_addr), Some(adp_addr)) = (config.agent_listen_addr.as_ref(), config.adp_listen_addr.as_ref()) { @@ -78,8 +84,8 @@ mod unix_intake { agent_addr, adp_addr ); - let agent_router = build_router(AppState::agent(&capture, &config.agent_config_dir)); - let adp_router = build_router(AppState::adp(&capture, &config.agent_config_dir)); + let agent_router = build_router(AppState::agent(&capture, pool.clone(), &config.config_dir)); + let adp_router = build_router(AppState::adp(&capture, pool.clone(), &config.config_dir)); // Both servers drain in flight requests on the shared shutdown signal. let agent_server = axum::serve(agent_listener, agent_router) @@ -99,7 +105,7 @@ mod unix_intake { axum::serve( listener, - build_router(AppState::adp(&capture, &config.agent_config_dir)), + build_router(AppState::adp(&capture, pool, &config.config_dir)), ) .with_graceful_shutdown(wait_for_shutdown(shutdown_rx)) .await diff --git a/test/antithesis/intake/src/context_pool.rs b/test/antithesis/intake/src/context_pool.rs new file mode 100644 index 00000000000..c320d395322 --- /dev/null +++ b/test/antithesis/intake/src/context_pool.rs @@ -0,0 +1,261 @@ +//! The context pool: per-kind bounded, lazily-filled sets the differential drivers draw from so +//! contexts recur across flushes. +//! +//! Every driver invocation is a fresh process; without coordination each would mint its own contexts +//! and the space would grow without bound and never recur. The pool holds one shared set per kind +//! behind a hard per-kind cap: it mints a new context of a sampled kind while that kind is under its +//! cap, then draws an existing one of that kind at random. Once a kind's cumulative requests exceed +//! its cap the set is exhausted and its contexts recur across flushes, which is what gives the +//! differential oracle multi-point curves to align. +//! +//! The caps are read from `context_source.yaml` on the first [`Pool::serve`] call. The intake starts +//! before `first_sample_config` samples that file, but the first `/contexts` request only ever comes +//! from a driver, which runs after `first_sample_config` — so the config is present by then. + +use std::collections::hash_map::DefaultHasher; +use std::collections::HashSet; +use std::hash::{Hash, Hasher}; +use std::path::PathBuf; +use std::sync::{Mutex, PoisonError}; + +use antithesis_sdk::prelude::*; +use anyhow::Context as _; +use harness::config::ContextSourceConfig; +use harness::contexts::{Context, Kind}; +use rand::{Rng, RngExt}; +use serde_json::json; + +/// A per-kind bounded, lazily-filled pool of contexts. +#[derive(Debug)] +pub struct Pool { + /// Directory holding `context_source.yaml`, read once on the first serve. + config_dir: PathBuf, + /// The resolved caps and the minted contexts, behind one lock. + state: Mutex<PoolState>, +} + +/// The pool's mutable state: the resolved per-kind caps and the minted contexts. +#[derive(Debug, Default)] +struct PoolState { + /// The per-kind caps, resolved from the config on the first serve. + caps: Option<ContextSourceConfig>, + /// Minted metric contexts, grown to the metric cap then drawn from. + metric: Vec<Context>, + /// Minted event contexts. + event: Vec<Context>, + /// Minted service-check contexts. + service_check: Vec<Context>, + /// Hashes of every context held, so a duplicate mint does not spend a cap slot. Hashes rather + /// than the contexts themselves, since the pool already holds up to a million of them and a + /// second copy would double that. A hash collision rejects a distinct identity, which costs one + /// remint and nothing else. + seen: HashSet<u64>, +} + +/// How many times a duplicate mint is retried before the slot is left for a later request. A small +/// alphabet under a tight budget collides often, and spinning here would stall the serve. +const MINT_TRIES: usize = 8; + +/// The hash a pooled identity is deduplicated by. +fn digest(context: &Context) -> u64 { + let mut hasher = DefaultHasher::new(); + context.hash(&mut hasher); + hasher.finish() +} + +impl Pool { + /// A pool that resolves its caps from `context_source.yaml` in `config_dir` on the first serve. + #[must_use] + pub fn new(config_dir: PathBuf) -> Self { + Self { + config_dir, + state: Mutex::new(PoolState::default()), + } + } + + /// Serve `n` contexts: for each slot sample a kind, mint a fresh one while that kind is under its + /// cap, else draw an existing one of that kind. The whole operation holds the lock, so concurrent + /// requests never race on `rng` or overshoot a cap. + /// + /// # Errors + /// + /// Returns an error if `context_source.yaml` cannot be read on the first call. + pub fn serve<R: Rng + ?Sized>(&self, n: usize, rng: &mut R) -> anyhow::Result<Vec<Context>> { + // A poisoned lock still holds a valid pool; recover it rather than panic. + let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner); + + let caps = if let Some(caps) = state.caps { + caps + } else { + let resolved = + ContextSourceConfig::read(&self.config_dir).context("read context source config for the pool caps")?; + state.caps = Some(resolved); + resolved + }; + + let mut out = Vec::with_capacity(n); + let mut served_existing = false; + for _ in 0..n { + let kind = Kind::sample(rng); + let PoolState { + metric, + event, + service_check, + seen, + .. + } = &mut *state; + let (pool, cap) = match kind { + Kind::Metric => (metric, caps.metric_contexts), + Kind::Event => (event, caps.event_contexts), + Kind::ServiceCheck => (service_check, caps.service_check_contexts), + }; + // The identity is minted against this timeline's real datagram budget, so every served + // context has a rendering the driver can pack. A budget too small for this kind, or an + // alphabet that keeps landing on the drop side, yields nothing rather than a context of + // another kind, so a kind's working set never fills with identities it did not ask for. + // A duplicate identity would spend a cap slot without adding a distinct context, so the + // kind would stop minting early and the run would explore fewer identities than configured. + // Remint on a duplicate instead of pushing it. + let mut minted = None; + if pool.len() < cap { + for _ in 0..MINT_TRIES { + let Some(context) = Context::mint_within(kind, rng, caps.payload_byte_limit.saturating_sub(1)) + else { + break; + }; + if seen.insert(digest(&context)) { + minted = Some(context); + break; + } + } + } + if let Some(context) = minted { + pool.push(context.clone()); + out.push(context); + } else if !pool.is_empty() { + let context = pool[rng.random_range(0..pool.len())].clone(); + served_existing = true; + out.push(context); + } + } + let metric = state.metric.len(); + let event = state.event.len(); + let service_check = state.service_check.len(); + drop(state); + + assert_always!( + metric <= caps.metric_contexts + && event <= caps.event_contexts + && service_check <= caps.service_check_contexts, + "context pool never exceeds its per-kind caps", + &json!({ + "metric": metric, + "event": event, + "service_check": service_check, + "caps": { "metric": caps.metric_contexts, "event": caps.event_contexts, "service_check": caps.service_check_contexts } + }) + ); + assert_sometimes!(served_existing, "context source served an existing context", &json!({})); + Ok(out) + } +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeSet; + use std::collections::HashSet; + use std::path::PathBuf; + use std::sync::atomic::{AtomicUsize, Ordering}; + + use harness::config::ContextSourceConfig; + use harness::contexts::{decode_response, encode_response, Context}; + use rand::rngs::SmallRng; + use rand::SeedableRng; + + use super::Pool; + + /// Write a `context_source.yaml` with the given per-kind caps into a fresh temp dir. + fn temp_config(metric: usize, event: usize, service_check: usize) -> PathBuf { + static SEQ: AtomicUsize = AtomicUsize::new(0); + let dir = std::env::temp_dir().join(format!( + "ctxpool-{}-{}", + std::process::id(), + SEQ.fetch_add(1, Ordering::Relaxed) + )); + std::fs::create_dir_all(&dir).expect("create temp config dir"); + let config = ContextSourceConfig { + payload_byte_limit: 8_192, + metric_contexts: metric, + event_contexts: event, + service_check_contexts: service_check, + }; + std::fs::write( + dir.join("context_source.yaml"), + config.to_yaml().expect("render config"), + ) + .expect("write config"); + dir + } + + #[test] + fn fills_to_caps_then_repeats() { + let mut rng = SmallRng::seed_from_u64(0); + // Small metric cap so metric contexts recur within the run. + let pool = Pool::new(temp_config(4, 1_000, 1_000)); + + let mut metric_wire = BTreeSet::new(); + let mut metric_total = 0; + for _ in 0..50 { + for context in pool.serve(5, &mut rng).expect("serve") { + if let Context::Metric(_) = context { + let mut wire = Vec::new(); + context.encode(&mut wire); + metric_wire.insert(wire); + metric_total += 1; + } + } + } + assert!( + metric_wire.len() <= 4, + "distinct metric {} exceeds cap 4", + metric_wire.len() + ); + assert!(metric_total > metric_wire.len(), "expected metric repeats"); + } + + // A cap counts distinct identities. A duplicate mint that consumed a slot would stop the kind + // minting early and leave the run exploring fewer identities than configured. + #[test] + fn pooled_contexts_are_distinct() { + let mut rng = SmallRng::seed_from_u64(11); + let pool = Pool::new(temp_config(64, 64, 64)); + let mut all = Vec::new(); + for _ in 0..16 { + all.extend(pool.serve(32, &mut rng).expect("serve")); + } + let state = pool.state.lock().expect("lock"); + for held in [&state.metric, &state.event, &state.service_check] { + let distinct: HashSet<&Context> = held.iter().collect(); + assert_eq!(distinct.len(), held.len(), "a kind holds a duplicate identity"); + } + } + + #[test] + fn serves_exactly_n_and_round_trips_the_wire() { + let mut rng = SmallRng::seed_from_u64(7); + let pool = Pool::new(temp_config(1_000, 1_000, 1_000)); + let contexts = pool.serve(9, &mut rng).expect("serve"); + assert_eq!(contexts.len(), 9); + + let wire = encode_response(&contexts); + let decoded = decode_response(&wire); + assert_eq!(decoded.as_deref(), Some(contexts.as_slice())); + } + + #[test] + fn missing_config_is_an_error() { + let mut rng = SmallRng::seed_from_u64(1); + let pool = Pool::new(std::env::temp_dir().join("ctxpool-does-not-exist")); + assert!(pool.serve(1, &mut rng).is_err()); + } +} diff --git a/test/antithesis/intake/src/http/antithesis.rs b/test/antithesis/intake/src/http/antithesis.rs index 1576c772c19..973441dae8f 100644 --- a/test/antithesis/intake/src/http/antithesis.rs +++ b/test/antithesis/intake/src/http/antithesis.rs @@ -4,19 +4,33 @@ //! //! - `GET /antithesis/metrics/{target}`: returns one lane's captured contexts and the intake's //! current time for `agent` or `adp`. +//! - `GET /contexts?n=N`: serves `N` contexts from the shared pool over the binary codec, so the +//! drivers render recurring identities. +use antithesis_sdk::prelude::*; +use antithesis_sdk::random::AntithesisRng; use axum::{ - extract::{Path, State}, + extract::{Path, Query, State}, http::StatusCode, routing::get, Json, Router, }; +use harness::contexts::encode_response; +use rand::rand_core::UnwrapErr; +use serde::Deserialize; +use serde_json::json; use super::state::AppState; use crate::capture; +/// The largest working set a single `/contexts` request may ask for. A request over this is rejected +/// rather than served, so a malformed `n` cannot make the pool mint an unbounded response. +const MAX_CONTEXTS_PER_REQUEST: usize = 65_536; + pub(super) fn routes() -> Router<AppState> { - Router::new().route("/antithesis/metrics/{target}", get(metrics)) + Router::new() + .route("/antithesis/metrics/{target}", get(metrics)) + .route("/contexts", get(contexts)) } async fn metrics( @@ -33,3 +47,25 @@ async fn metrics( contexts: state.recorder.contexts(target), })) } + +/// The `n` query parameter of `GET /contexts`. +#[derive(Debug, Deserialize)] +struct ContextQuery { + /// How many contexts to serve. + n: usize, +} + +/// Serve `n` contexts from the shared pool, encoded with the binary codec so non-UTF-8 names and tags +/// round-trip. Rejects an out-of-range `n` with 400 and a pool read error with 500. +async fn contexts(State(state): State<AppState>, Query(query): Query<ContextQuery>) -> Result<Vec<u8>, StatusCode> { + if query.n == 0 || query.n > MAX_CONTEXTS_PER_REQUEST { + return Err(StatusCode::BAD_REQUEST); + } + let mut rng = UnwrapErr(AntithesisRng); + let contexts = state + .pool + .serve(query.n, &mut rng) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + assert_reachable!("context source served a request", &json!({ "n": query.n })); + Ok(encode_response(&contexts)) +} diff --git a/test/antithesis/intake/src/http/state.rs b/test/antithesis/intake/src/http/state.rs index fe72a630ba1..c212d465f6e 100644 --- a/test/antithesis/intake/src/http/state.rs +++ b/test/antithesis/intake/src/http/state.rs @@ -4,9 +4,11 @@ use std::path::Path; use std::sync::{Arc, OnceLock}; use crate::capture; +use crate::context_pool::Pool; use crate::sut_config::SutConfig; -/// Per-router state: the shared recorder handle plus the lane this router writes to. +/// Per-router state: the shared recorder handle, the lane this router writes to, and the shared +/// context pool the drivers draw from. #[derive(Clone, Debug)] pub struct AppState { pub(crate) recorder: capture::State, @@ -14,8 +16,11 @@ pub struct AppState { /// First non-empty host resolved on this lane, set once. Pyld17 requires every series across all /// inbound traffic on the lane to resolve to this same host. pub(crate) established_host: Arc<OnceLock<String>>, + /// The shared context pool served by `GET /contexts`. One pool backs every lane, so the drivers + /// draw recurring identities across lanes. + pub(crate) pool: Arc<Pool>, /// Directory holding the timeline's sampled `datadog.yaml`. - agent_config_dir: Arc<Path>, + config_dir: Arc<Path>, /// The sampled config, read on the first request that finds the file written. sut_config: Arc<OnceLock<SutConfig>>, } @@ -23,22 +28,23 @@ pub struct AppState { impl AppState { /// Creates router state for Datadog Agent intake. #[must_use] - pub fn agent(recorder: &capture::State, agent_config_dir: &Path) -> Self { - Self::new(recorder, capture::Target::Agent, agent_config_dir) + pub fn agent(recorder: &capture::State, pool: Arc<Pool>, config_dir: &Path) -> Self { + Self::new(recorder, capture::Target::Agent, pool, config_dir) } /// Creates router state for ADP intake. #[must_use] - pub fn adp(recorder: &capture::State, agent_config_dir: &Path) -> Self { - Self::new(recorder, capture::Target::Adp, agent_config_dir) + pub fn adp(recorder: &capture::State, pool: Arc<Pool>, config_dir: &Path) -> Self { + Self::new(recorder, capture::Target::Adp, pool, config_dir) } - fn new(recorder: &capture::State, target: capture::Target, agent_config_dir: &Path) -> Self { + fn new(recorder: &capture::State, target: capture::Target, pool: Arc<Pool>, config_dir: &Path) -> Self { Self { recorder: recorder.clone(), target, established_host: Arc::default(), - agent_config_dir: Arc::from(agent_config_dir), + pool, + config_dir: Arc::from(config_dir), sut_config: Arc::default(), } } @@ -49,7 +55,7 @@ impl AppState { if let Some(config) = self.sut_config.get() { return Some(config); } - let config = SutConfig::load(&self.agent_config_dir)?; + let config = SutConfig::load(&self.config_dir)?; Some(self.sut_config.get_or_init(|| config)) } } diff --git a/test/antithesis/intake/src/lib.rs b/test/antithesis/intake/src/lib.rs index 087473273f6..4d11404c12d 100644 --- a/test/antithesis/intake/src/lib.rs +++ b/test/antithesis/intake/src/lib.rs @@ -34,6 +34,7 @@ #![deny(warnings)] pub mod capture; +pub mod context_pool; pub mod http; mod lenient_decode; diff --git a/test/antithesis/intake/tests/series_decode_interop.rs b/test/antithesis/intake/tests/series_decode_interop.rs index add2a52e6cc..eb311270907 100644 --- a/test/antithesis/intake/tests/series_decode_interop.rs +++ b/test/antithesis/intake/tests/series_decode_interop.rs @@ -7,9 +7,11 @@ //! each producer. use std::io::Write as _; -use std::path::Path; +use std::path::{Path, PathBuf}; +use std::sync::Arc; use antithesis_intake::capture; +use antithesis_intake::context_pool::Pool; use antithesis_intake::http::{build_router, state::AppState}; use axum::body::Body; use axum::http::{Request, StatusCode}; @@ -60,7 +62,10 @@ async fn adp_async_compression(raw: &[u8]) -> Vec<u8> { const AGENT_UA: &str = "datadog-agent/7.55.0"; async fn post_series(encoding: Option<&str>, user_agent: Option<&str>, body: Vec<u8>) -> StatusCode { - let state = AppState::agent(&capture::State::new(), Path::new("/nonexistent-agent-config")); + // This test drives only the series decode path, never `/contexts`, so neither the pool's config + // nor the sampled `datadog.yaml` is read. + let pool = Arc::new(Pool::new(PathBuf::from("/nonexistent"))); + let state = AppState::agent(&capture::State::new(), pool, Path::new("/nonexistent")); let app = build_router(state); let mut builder = Request::builder() .method("POST") diff --git a/test/antithesis/scenarios/differential/docker-compose.yaml b/test/antithesis/scenarios/differential/docker-compose.yaml index 16ab74618c4..b3b042d3f5d 100644 --- a/test/antithesis/scenarios/differential/docker-compose.yaml +++ b/test/antithesis/scenarios/differential/docker-compose.yaml @@ -16,8 +16,11 @@ services: DD_HOSTNAME: "antithesis-differential" AGENT_LISTEN_ADDR: "0.0.0.0:2049" ADP_LISTEN_ADDR: "0.0.0.0:2050" + CONFIG_DIR: "/agent-config" volumes: - # The sampled datadog.yaml, so the intake can hold each target to what its config asked for. + # first_sample_config (workload) writes this timeline's datadog.yaml and context_source.yaml + # here. The intake reads the first to hold each target to its config and the second for the + # context pool's caps. - agent-config:/agent-config:ro healthcheck: test: ["CMD-SHELL", "bash -c 'exec 3<>/dev/tcp/localhost/2049'"] @@ -92,6 +95,8 @@ services: environment: NO_COLOR: "1" INTAKE_CONTROL_ADDR: "intake:2049" + # Both lanes share one pool, so either lane's address serves the same contexts. + INTAKE_ADDR: "intake:2049" AGENT_DOGSTATSD_SOCKET: "/var/run/datadog-agent/dsd.socket" ADP_DOGSTATSD_SOCKET: "/var/run/adp/dsd.socket" DOGSTATSD_SOCKET: "/var/run/dogstatsd/dsd.socket" diff --git a/test/antithesis/scenarios/differential/src/bin/parallel_driver_send_dogstatsd_differential.rs b/test/antithesis/scenarios/differential/src/bin/parallel_driver_send_dogstatsd_differential.rs index ae80beb0ae3..81603d27c55 100644 --- a/test/antithesis/scenarios/differential/src/bin/parallel_driver_send_dogstatsd_differential.rs +++ b/test/antithesis/scenarios/differential/src/bin/parallel_driver_send_dogstatsd_differential.rs @@ -34,6 +34,10 @@ mod unix_driver { /// to the shared sampled receive buffer. #[arg(long = "config-dir", env = "CONFIG_DIR", default_value = "/agent-config")] config_dir: PathBuf, + /// `host:port` of the intake that serves the shared context pool over `GET /contexts`. Both + /// lanes share one pool, so either lane's address serves the same contexts. + #[arg(long = "intake-addr", env = "INTAKE_ADDR", default_value = "intake:2049")] + intake_addr: String, } pub(super) fn run() -> anyhow::Result<()> { @@ -64,6 +68,8 @@ mod unix_driver { // Agent first, ADP second: `stats.sent` and `stats.max_packed` are indexed in this order. let stats = driver::run( AntithesisRng, + &config.intake_addr, + driver_config.context_count, driver_config.payload_byte_limit, driver_config.datagram_count, vec![agent_socket, adp_socket], diff --git a/test/antithesis/scenarios/general/docker-compose.yaml b/test/antithesis/scenarios/general/docker-compose.yaml index 0a6886672a4..29416458f90 100644 --- a/test/antithesis/scenarios/general/docker-compose.yaml +++ b/test/antithesis/scenarios/general/docker-compose.yaml @@ -13,8 +13,11 @@ services: image: intake:${ANTITHESIS_IMAGE_TAG:-latest} environment: NO_COLOR: "1" + CONFIG_DIR: "/agent-config" volumes: - # The sampled datadog.yaml, so the intake can hold the SUT to what its config asked for. + # first_sample_config (workload) writes this timeline's datadog.yaml and context_source.yaml + # here. The intake reads the first to hold the SUT to its config and the second for the context + # pool's caps. - agent-config:/agent-config:ro healthcheck: # antithesis-intake serves HTTP on :2049. /dev/tcp avoids needing curl in the image. diff --git a/test/antithesis/scenarios/general/src/bin/parallel_driver_send_dogstatsd.rs b/test/antithesis/scenarios/general/src/bin/parallel_driver_send_dogstatsd.rs index ac80b8a5594..18c98e6d661 100644 --- a/test/antithesis/scenarios/general/src/bin/parallel_driver_send_dogstatsd.rs +++ b/test/antithesis/scenarios/general/src/bin/parallel_driver_send_dogstatsd.rs @@ -26,11 +26,18 @@ mod unix_driver { /// to the SUT's sampled receive buffer. #[arg(long = "config-dir", env = "CONFIG_DIR", default_value = "/agent-config")] config_dir: PathBuf, + /// `host:port` of the intake that serves the shared context pool over `GET /contexts`. + #[arg(long = "intake-addr", env = "INTAKE_ADDR", default_value = "intake:2049")] + intake_addr: String, } pub(super) fn run() -> anyhow::Result<()> { antithesis_init(); + // reqwest is built with `rustls-no-provider` in this crate, so install the process-wide + // crypto provider before the driver builds its context-fetch client, which otherwise panics. + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let config = Config::try_parse()?; // Socket unavailable (ADP booting, or a fault). No-op exit, not a failure. @@ -41,6 +48,8 @@ mod unix_driver { let driver_config = DriverConfig::read(&config.config_dir)?; let stats = driver::run( AntithesisRng, + &config.intake_addr, + driver_config.context_count, driver_config.payload_byte_limit, driver_config.datagram_count, vec![socket],