|
| 1 | +//! Per-timeline lading config sampling. |
| 2 | +//! |
| 3 | +//! The sampler builds lading's real [`tcp::Config`] so the value menu is exactly |
| 4 | +//! the config schema, then serializes it under the `generator: [{ tcp: … }]` |
| 5 | +//! shape lading parses. The transport is fixed to TCP against the sink oracle, |
| 6 | +//! which counts bytes and so catches any payload variant; only the free axes |
| 7 | +//! (payload variant, rate, connections, block sizes, seed) vary. Transport |
| 8 | +//! variation waits on a multi-protocol sink. |
| 9 | +
|
| 10 | +use lading::generator::tcp; |
| 11 | +use lading_payload as payload; |
| 12 | +use rand::seq::IndexedRandom as _; |
| 13 | +use rand::{Rng, RngExt as _}; |
| 14 | + |
| 15 | +/// Fixed address of the sink oracle every sampled config targets. |
| 16 | +const SINK_ADDR: &str = "sink:9000"; |
| 17 | + |
| 18 | +/// Payload variants the TCP sink catches with no decoding. All are fieldless, so |
| 19 | +/// they need no extra configuration to sample. |
| 20 | +fn variant_menu() -> [payload::Config; 6] { |
| 21 | + [ |
| 22 | + payload::Config::Ascii, |
| 23 | + payload::Config::Syslog5424, |
| 24 | + payload::Config::Json, |
| 25 | + payload::Config::Fluent, |
| 26 | + payload::Config::ApacheCommon, |
| 27 | + payload::Config::DatadogLog, |
| 28 | + ] |
| 29 | +} |
| 30 | + |
| 31 | +/// Sample a lading TCP generator config for one timeline. |
| 32 | +/// |
| 33 | +/// Draws the payload variant, throughput, parallel connections, and seed from |
| 34 | +/// `rng`; the transport and target stay fixed to the sink. |
| 35 | +#[must_use] |
| 36 | +pub fn sample<R: Rng>(rng: &mut R) -> tcp::Config { |
| 37 | + // Structured choices come from the caller's rng -- AntithesisRng in |
| 38 | + // production -- so Antithesis branches each pick and sweeps the menu. |
| 39 | + let variant = variant_menu() |
| 40 | + .choose(rng) |
| 41 | + .cloned() |
| 42 | + .unwrap_or(payload::Config::Ascii); |
| 43 | + |
| 44 | + let bps_mib = [1_u64, 5, 10, 50, 100].choose(rng).copied().unwrap_or(10); |
| 45 | + let bytes_per_second_bytes = bps_mib * 1024 * 1024; |
| 46 | + let bytes_per_second = Some(byte_unit::Byte::from_u64(bytes_per_second_bytes)); |
| 47 | + |
| 48 | + let parallel_connections = rng.random_range(1..=8_u16); |
| 49 | + |
| 50 | + // Cap the block size at the smallest per-connection throttle capacity. lading |
| 51 | + // divides `bytes_per_second` evenly across `parallel_connections`; a block |
| 52 | + // larger than a worker's divided capacity is rejected by the throttle, and |
| 53 | + // the TCP worker then busy-spins discarding it with no backoff. Keeping |
| 54 | + // `maximum_block_size <= bytes_per_second / parallel_connections` guarantees |
| 55 | + // every block fits, so no timeline degrades into a discard spin. |
| 56 | + let per_connection_capacity = bytes_per_second_bytes / u64::from(parallel_connections); |
| 57 | + let maximum_block_size = |
| 58 | + byte_unit::Byte::from_u64(per_connection_capacity.clamp(1, 1024 * 1024)); |
| 59 | + |
| 60 | + // lading seeds its own payload PRNG from `seed`. The Antithesis docs warn |
| 61 | + // against seeding your own RNG from SDK randomness, so draw the seed from |
| 62 | + // system entropy rather than the (Antithesis) `rng`. This makes payload byte |
| 63 | + // content opaque to Antithesis and effectively fixed across timelines, which |
| 64 | + // is fine: the sink asserts on bytes received, not on content. |
| 65 | + let mut seed = [0u8; 32]; |
| 66 | + rand::rng().fill_bytes(&mut seed); |
| 67 | + |
| 68 | + tcp::Config { |
| 69 | + seed, |
| 70 | + addr: SINK_ADDR.to_string(), |
| 71 | + variant, |
| 72 | + bytes_per_second, |
| 73 | + maximum_block_size, |
| 74 | + maximum_prebuild_cache_size_bytes: byte_unit::Byte::from_u64(8 * 1024 * 1024), |
| 75 | + parallel_connections, |
| 76 | + throttle: None, |
| 77 | + } |
| 78 | +} |
| 79 | + |
| 80 | +/// Serialize a sampled `tcp::Config` into the top-level `generator: [{ tcp: … }]` |
| 81 | +/// YAML that lading consumes. |
| 82 | +/// |
| 83 | +/// # Errors |
| 84 | +/// |
| 85 | +/// Returns an error if serialization fails. |
| 86 | +pub fn to_yaml(cfg: &tcp::Config) -> Result<String, serde_yaml::Error> { |
| 87 | + let mut tcp_item = serde_yaml::Mapping::new(); |
| 88 | + tcp_item.insert(serde_yaml::Value::from("tcp"), serde_yaml::to_value(cfg)?); |
| 89 | + let generators = serde_yaml::Value::Sequence(vec![serde_yaml::Value::Mapping(tcp_item)]); |
| 90 | + let mut top = serde_yaml::Mapping::new(); |
| 91 | + top.insert(serde_yaml::Value::from("generator"), generators); |
| 92 | + serde_yaml::to_string(&serde_yaml::Value::Mapping(top)) |
| 93 | +} |
| 94 | + |
| 95 | +/// Short, stable label for a payload variant, for tagging the Antithesis sample. |
| 96 | +#[must_use] |
| 97 | +pub fn variant_label(variant: &payload::Config) -> &'static str { |
| 98 | + match variant { |
| 99 | + payload::Config::Ascii => "ascii", |
| 100 | + payload::Config::Syslog5424 => "syslog5424", |
| 101 | + payload::Config::Json => "json", |
| 102 | + payload::Config::Fluent => "fluent", |
| 103 | + payload::Config::ApacheCommon => "apache_common", |
| 104 | + payload::Config::DatadogLog => "datadog_log", |
| 105 | + _ => "other", |
| 106 | + } |
| 107 | +} |
| 108 | + |
| 109 | +#[cfg(test)] |
| 110 | +mod tests { |
| 111 | + use super::{sample, to_yaml}; |
| 112 | + use rand::SeedableRng as _; |
| 113 | + use rand::rngs::StdRng; |
| 114 | + |
| 115 | + #[test] |
| 116 | + fn sampled_config_deserializes_as_valid_lading_config() { |
| 117 | + // The load-bearing invariant: whatever we sample must be a config lading |
| 118 | + // actually accepts. Sweep many seeds so the whole menu is exercised. |
| 119 | + for s in 0..256_u64 { |
| 120 | + let mut rng = StdRng::seed_from_u64(s); |
| 121 | + let cfg = sample(&mut rng); |
| 122 | + let yaml = to_yaml(&cfg).expect("serialize sampled config"); |
| 123 | + let parsed: Result<lading::config::Config, _> = serde_yaml::from_str(&yaml); |
| 124 | + assert!( |
| 125 | + parsed.is_ok(), |
| 126 | + "seed {s} produced a config lading rejects: {err:?}\n{yaml}", |
| 127 | + err = parsed.err() |
| 128 | + ); |
| 129 | + } |
| 130 | + } |
| 131 | + |
| 132 | + #[test] |
| 133 | + fn sampled_config_holds_invariants() { |
| 134 | + for s in 0..256_u64 { |
| 135 | + let mut rng = StdRng::seed_from_u64(s); |
| 136 | + let cfg = sample(&mut rng); |
| 137 | + assert_eq!(cfg.addr, "sink:9000"); |
| 138 | + assert!((1..=8).contains(&cfg.parallel_connections)); |
| 139 | + assert!(cfg.bytes_per_second.is_some()); |
| 140 | + } |
| 141 | + } |
| 142 | + |
| 143 | + #[test] |
| 144 | + fn block_size_never_exceeds_per_connection_capacity() { |
| 145 | + // Regression: lading divides bytes_per_second across parallel_connections, |
| 146 | + // and a block larger than a worker's divided capacity busy-spins on |
| 147 | + // discard. maximum_block_size must fit the smallest per-connection share. |
| 148 | + for s in 0..256_u64 { |
| 149 | + let mut rng = StdRng::seed_from_u64(s); |
| 150 | + let cfg = sample(&mut rng); |
| 151 | + let bps = cfg.bytes_per_second.expect("bytes_per_second set").as_u64(); |
| 152 | + let per_connection = bps / u64::from(cfg.parallel_connections); |
| 153 | + assert!( |
| 154 | + cfg.maximum_block_size.as_u64() <= per_connection, |
| 155 | + "seed {s}: block {block} exceeds per-connection capacity {per_connection}", |
| 156 | + block = cfg.maximum_block_size.as_u64() |
| 157 | + ); |
| 158 | + } |
| 159 | + } |
| 160 | +} |
0 commit comments