chore(antithesis): Bound contexts in runs

This commit introduces a new protocol to the intake node to allow for
context bounding in runs. The purposes are twofold. First, while it's
interesting to explore timelines in which there are too many contexts
for either lane to handle these are less relevant to me _now_, so I've
capped total contexts at 1M. Second, in order for timeseries equality
to _work_ I need to be certain that I have at least _some_ timeseries
with more than one point emitted into them. Previously drivers were
emitting essentially random contexts and it was very unlikely for more
than one point to be emitted per context. Oops.

Drivers now request N contexts from the intake service for each sub-kind
of dogstatsd they emit, meaning all contexts are sourced from a single
spot in a topology and not from pure randomness.

Author: blt

Cargo.lock

@@ -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 }

test/antithesis/harness/Cargo.toml

@@ -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

test/antithesis/harness/proptest-regressions/contexts.txt

@@ -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;
 
@@ -67,6 +67,15 @@ fn main() -> anyhow::Result<()> {
     fs::write(&driver_path, config.driver_config(&mut rng).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).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() }));

test/antithesis/harness/src/bin/first_sample_config/main.rs

@@ -186,6 +186,9 @@ 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;
+
 /// 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.
@@ -197,6 +200,9 @@ 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, each
+    /// boundary-biased log-uniform in `1..=MAX_WORKING_SET`.
+    pub context_count: usize,
 }
 
 impl DriverConfig {
@@ -211,6 +217,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),
         }
     }
 
@@ -237,6 +244,61 @@ impl DriverConfig {
     }
 }
 
+/// Upper bound on the distinct contexts a shared pool holds per kind.
+const MAX_CONTEXTS_PER_KIND: 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 independently, so the distinct-context count per kind varies at random.
+#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
+pub struct ContextSourceConfig {
+    /// Distinct metric contexts the pool holds.
+    pub metric_contexts: usize,
+    /// Distinct event contexts the pool holds.
+    pub event_contexts: usize,
+    /// Distinct service-check contexts the pool holds.
+    pub service_check_contexts: usize,
+}
+
+impl ContextSourceConfig {
+    /// Sample the per-kind caps, each boundary-biased log-uniform in `1..=MAX_CONTEXTS_PER_KIND`.
+    #[must_use]
+    pub fn sample<R: Rng + ?Sized>(rng: &mut R) -> Self {
+        Self {
+            metric_contexts: sample_cap(rng),
+            event_contexts: sample_cap(rng),
+            service_check_contexts: sample_cap(rng),
+        }
+    }
+
+    /// Render `self` as a `context_source.yaml` string.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if serialization fails.
+    pub fn to_yaml(&self) -> anyhow::Result<String> {
+        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<Self> {
+        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. The Probe max is well within `usize` on every supported target, so the
+/// saturating conversion is unreachable in practice.
+fn sample_cap<R: Rng + ?Sized>(rng: &mut R) -> usize {
+    usize::try_from(Probe::new(1, MAX_CONTEXTS_PER_KIND).sample(rng)).unwrap_or(usize::MAX)
+}
+
 #[cfg(test)]
 mod tests {
     use std::convert::Infallible;