diff --git a/Cargo.lock b/Cargo.lock index 96098eee7f..1a31572fa7 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 ecdaf0c909..ce49298aad 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 0000000000..471a57cd7c --- /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 ca129ac644..a4b8eda25d 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 894ebd2fd2..53efe6b566 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 0000000000..f5d596b6c7 --- /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 0000000000..7eea6d0333 --- /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 0000000000..383a7dc43c --- /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 0000000000..080d69956a --- /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 1bff2d839b..5b3ea5a531 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 e45640b90f..9c675be1eb 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 eea855243e..273e9764fd 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 5905cd578d..828c99bd35 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 0334a5113b..0000000000 --- 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 9c27bb2573..0000000000 --- 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 fe2c690859..0000000000 --- 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 d7a1b67ef6..6220a0dc8b 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 0d0c16cbb3..d09e654e49 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 0000000000..c320d39532 --- /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 1576c772c1..973441dae8 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 fe72a630ba..c212d465f6 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 087473273f..4d11404c12 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 add2a52e6c..eb31127090 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 16ab74618c..b3b042d3f5 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 ae80beb0ae..81603d27c5 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 0a6886672a..29416458f9 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 ac80b8a559..18c98e6d66 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],