krabka_parse_benches/
contract.rs

1//! The regression contract: the workload matrix and the budgets that a
2//! performance comparison is judged by.
3//!
4//! The contract is declared before any trial runs. The harness records the
5//! SHA-256 of the contract's bytes when it starts, and
6//! [`evaluate`](crate::budget::evaluate) refuses a record whose digest does
7//! not match the contract it is handed. A budget edited after the numbers are
8//! in therefore cannot judge them.
9
10use {
11    serde::{Deserialize, Serialize},
12    sha2::{Digest, Sha256},
13    std::{
14        collections::{BTreeMap, BTreeSet},
15        io,
16    },
17    thiserror::Error,
18};
19
20/// The only contract and record format version this crate reads and writes.
21pub const FORMAT_VERSION: u32 = 1;
22
23/// The fewest interleaved rounds a contract may declare. The criterion lane's
24/// [`compare_summaries`](crate::compare_summaries) has the same floor: a
25/// median absolute deviation over fewer than three samples says nothing.
26pub const MINIMUM_ROUNDS: usize = 3;
27
28/// The smallest record the workload can carry: it writes an 8-byte send
29/// timestamp and an 8-byte sequence number into every value.
30pub const MINIMUM_RECORD_BYTES: u32 = 16;
31
32/// Errors from loading a contract, assembling a record, or evaluating one.
33#[derive(Debug, Error)]
34#[non_exhaustive]
35pub enum BudgetError {
36    /// The contract or record declares a format version this crate cannot read.
37    #[error("{what} format version {found} is not {FORMAT_VERSION}")]
38    FormatVersion {
39        /// Which document carried the version.
40        what: &'static str,
41        /// The version it declared.
42        found: u32,
43    },
44
45    /// The contract is internally inconsistent.
46    #[error("invalid contract: {0}")]
47    InvalidContract(String),
48
49    /// The record was measured under a different contract.
50    #[error("record was measured under contract {recorded} but evaluation was given {supplied}")]
51    ContractMismatch {
52        /// The contract name and digest the record carries.
53        recorded: String,
54        /// The contract name and digest supplied for evaluation.
55        supplied: String,
56    },
57
58    /// The record is structurally invalid.
59    #[error("invalid record: {0}")]
60    InvalidRecord(String),
61
62    /// A harness artifact could not be parsed.
63    #[error("invalid artifact {path}: {reason}")]
64    InvalidArtifact {
65        /// The artifact path, relative to the artifact directory.
66        path: String,
67        /// What was wrong with it.
68        reason: String,
69    },
70
71    /// Underlying I/O error.
72    #[error("I/O error: {0}")]
73    Io(#[from] io::Error),
74
75    /// Underlying JSON error.
76    #[error("JSON error: {0}")]
77    Json(#[from] serde_json::Error),
78}
79
80/// What a comparison measures the candidate against.
81#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
82#[serde(rename_all = "snake_case")]
83pub enum Mode {
84    /// The reference is an earlier krabka revision.
85    BeforeAfter,
86    /// The reference is the pinned Apache Kafka image.
87    PinnedKafka,
88}
89
90impl Mode {
91    /// The name the harness and the record use for this mode.
92    #[must_use]
93    pub fn name(self) -> &'static str {
94        match self {
95            Self::BeforeAfter => "before_after",
96            Self::PinnedKafka => "pinned_kafka",
97        }
98    }
99}
100
101/// Which way a metric gets better.
102#[derive(Debug, Clone, Copy, PartialEq, Eq)]
103pub enum Direction {
104    /// A larger value is better, as for throughput.
105    HigherIsBetter,
106    /// A smaller value is better, as for latency, CPU and memory.
107    LowerIsBetter,
108}
109
110/// A quantity every completed trial reports.
111#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
112#[serde(rename_all = "snake_case")]
113pub enum Metric {
114    /// Acknowledged and consumed records per second, end to end.
115    RecordsPerSecond,
116    /// Median end-to-end latency in milliseconds.
117    LatencyMsP50,
118    /// 95th percentile end-to-end latency in milliseconds.
119    LatencyMsP95,
120    /// 99th percentile end-to-end latency in milliseconds.
121    LatencyMsP99,
122    /// Broker CPU, summed over every broker, per acknowledged record.
123    CpuMicrosPerRecord,
124    /// Peak resident set size, summed over every broker, in MiB.
125    PeakRssMib,
126}
127
128impl Metric {
129    /// Every metric, in report order.
130    pub const ALL: [Self; 6] = [
131        Self::RecordsPerSecond,
132        Self::LatencyMsP50,
133        Self::LatencyMsP95,
134        Self::LatencyMsP99,
135        Self::CpuMicrosPerRecord,
136        Self::PeakRssMib,
137    ];
138
139    /// Which way this metric gets better.
140    #[must_use]
141    pub fn direction(self) -> Direction {
142        match self {
143            Self::RecordsPerSecond => Direction::HigherIsBetter,
144            Self::LatencyMsP50
145            | Self::LatencyMsP95
146            | Self::LatencyMsP99
147            | Self::CpuMicrosPerRecord
148            | Self::PeakRssMib => Direction::LowerIsBetter,
149        }
150    }
151
152    /// The column heading the rendered report uses.
153    #[must_use]
154    pub fn label(self) -> &'static str {
155        match self {
156            Self::RecordsPerSecond => "records/s",
157            Self::LatencyMsP50 => "p50 ms",
158            Self::LatencyMsP95 => "p95 ms",
159            Self::LatencyMsP99 => "p99 ms",
160            Self::CpuMicrosPerRecord => "CPU µs/record",
161            Self::PeakRssMib => "peak RSS MiB",
162        }
163    }
164}
165
166/// How a cell's record values are filled.
167#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
168#[serde(rename_all = "snake_case")]
169pub enum Payload {
170    /// Bytes from a seeded random pool, which compression cannot shrink.
171    Random,
172    /// Zero bytes, which compress almost completely.
173    Zeros,
174}
175
176impl Payload {
177    /// The argument `BrokerPerformanceWorkload` takes for this payload.
178    #[must_use]
179    pub fn name(self) -> &'static str {
180        match self {
181            Self::Random => "random",
182            Self::Zeros => "zeros",
183        }
184    }
185}
186
187/// One workload in the matrix.
188#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
189#[serde(deny_unknown_fields)]
190pub struct Cell {
191    /// Stable identifier, also the trial directory name: `[a-z0-9-]+`.
192    pub id: String,
193    /// Bytes per record value.
194    pub record_bytes: u32,
195    /// How the values are filled.
196    pub payload: Payload,
197    /// The producer's `compression.type`.
198    pub compression: String,
199    /// Partitions of the cell's fresh topic.
200    pub partitions: u32,
201    /// Records every trial must produce and consume.
202    pub records: u64,
203    /// Requested records per second, or `None` to saturate.
204    pub rate: Option<u64>,
205    /// Seconds the subscribed consumer idles before the first record is sent.
206    pub idle_seconds: u32,
207    /// Seconds the workload may run before it fails.
208    pub timeout_seconds: u32,
209    /// The metrics a budget judges for this cell. Every metric is recorded.
210    pub budgeted: Vec<Metric>,
211}
212
213/// The materiality and variance budget for one metric.
214#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
215#[serde(deny_unknown_fields)]
216pub struct Budget {
217    /// A median change smaller than this percentage is never material.
218    pub materiality_percent: f64,
219    /// A median change must also exceed this many times the observed spread.
220    pub spread_multiplier: f64,
221    /// A side whose spread exceeds this percentage makes the cell inconclusive.
222    pub max_spread_percent: f64,
223}
224
225/// What a mode gates on beyond correctness and variance.
226#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
227#[serde(deny_unknown_fields)]
228pub struct ModePolicy {
229    /// Whether a material regression rejects the comparison.
230    pub gate_performance: bool,
231}
232
233/// The cluster shape every trial runs against.
234#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
235#[serde(deny_unknown_fields)]
236pub struct Cluster {
237    /// Combined broker/controller nodes.
238    pub brokers: u32,
239    /// Replication factor of every cell's topic.
240    pub replication_factor: u32,
241    /// `min.insync.replicas` of every cell's topic.
242    pub min_insync_replicas: u32,
243}
244
245/// The whole regression contract.
246#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
247#[serde(deny_unknown_fields)]
248pub struct Contract {
249    /// Must be [`FORMAT_VERSION`].
250    pub format_version: u32,
251    /// The contract's name, which every record and report carries.
252    pub name: String,
253    /// Interleaved rounds; each round runs every cell once per side.
254    pub rounds: usize,
255    /// The cluster every trial runs against.
256    pub cluster: Cluster,
257    /// The pinned Kafka image a [`Mode::PinnedKafka`] comparison runs, and
258    /// whose client jars drive both sides in every mode.
259    pub kafka_image: String,
260    /// The workload matrix.
261    pub cells: Vec<Cell>,
262    /// One budget per budgeted metric.
263    pub budgets: BTreeMap<Metric, Budget>,
264    /// What each mode gates on.
265    pub modes: BTreeMap<Mode, ModePolicy>,
266}
267
268fn invalid(reason: impl Into<String>) -> BudgetError { BudgetError::InvalidContract(reason.into()) }
269
270fn is_cell_id(id: &str) -> bool {
271    !id.is_empty()
272        && id
273            .bytes()
274            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-')
275}
276
277impl Contract {
278    /// Parse and validate a contract from its JSON bytes.
279    ///
280    /// # Errors
281    ///
282    /// Returns [`BudgetError`] when the bytes are not a contract or the
283    /// contract fails [`Contract::validate`].
284    pub fn from_json(bytes: &[u8]) -> Result<Self, BudgetError> {
285        let contract: Self = serde_json::from_slice(bytes)?;
286        contract.validate()?;
287        Ok(contract)
288    }
289
290    /// Check that the contract can judge a comparison.
291    ///
292    /// # Errors
293    ///
294    /// Returns [`BudgetError::InvalidContract`] naming the first problem.
295    pub fn validate(&self) -> Result<(), BudgetError> {
296        if self.format_version != FORMAT_VERSION {
297            return Err(BudgetError::FormatVersion {
298                what: "contract",
299                found: self.format_version,
300            });
301        }
302        if self.rounds < MINIMUM_ROUNDS {
303            return Err(invalid(format!(
304                "rounds must be at least {MINIMUM_ROUNDS}, got {}",
305                self.rounds
306            )));
307        }
308        let cluster = &self.cluster;
309        if cluster.replication_factor == 0
310            || cluster.replication_factor > cluster.brokers
311            || cluster.min_insync_replicas == 0
312            || cluster.min_insync_replicas > cluster.replication_factor
313        {
314            return Err(invalid(
315                "cluster needs 0 < min_insync_replicas <= replication_factor <= brokers",
316            ));
317        }
318        if self.cells.is_empty() {
319            return Err(invalid("the workload matrix has no cells"));
320        }
321        let mut ids = BTreeSet::new();
322        for cell in &self.cells {
323            if !is_cell_id(&cell.id) {
324                return Err(invalid(format!("cell id {:?} is not [a-z0-9-]+", cell.id)));
325            }
326            if !ids.insert(cell.id.as_str()) {
327                return Err(invalid(format!("cell id {} appears twice", cell.id)));
328            }
329            if cell.record_bytes < MINIMUM_RECORD_BYTES {
330                return Err(invalid(format!(
331                    "cell {} has records smaller than {MINIMUM_RECORD_BYTES} bytes",
332                    cell.id
333                )));
334            }
335            if cell.records == 0 || cell.partitions == 0 || cell.timeout_seconds == 0 {
336                return Err(invalid(format!(
337                    "cell {} needs positive records, partitions and timeout",
338                    cell.id
339                )));
340            }
341            if cell.rate == Some(0) {
342                return Err(invalid(format!(
343                    "cell {} has rate 0; omit it to saturate",
344                    cell.id
345                )));
346            }
347            if cell.budgeted.is_empty() {
348                return Err(invalid(format!("cell {} budgets no metric", cell.id)));
349            }
350            for metric in &cell.budgeted {
351                if !self.budgets.contains_key(metric) {
352                    return Err(invalid(format!(
353                        "cell {} budgets {metric:?}, which has no budget",
354                        cell.id
355                    )));
356                }
357            }
358        }
359        for (metric, budget) in &self.budgets {
360            let values = [
361                budget.materiality_percent,
362                budget.spread_multiplier,
363                budget.max_spread_percent,
364            ];
365            if values
366                .iter()
367                .any(|value| !value.is_finite() || *value < 0.0)
368            {
369                return Err(invalid(format!(
370                    "budget for {metric:?} must be finite and non-negative"
371                )));
372            }
373        }
374        for mode in [Mode::BeforeAfter, Mode::PinnedKafka] {
375            if !self.modes.contains_key(&mode) {
376                return Err(invalid(format!("no policy for mode {}", mode.name())));
377            }
378        }
379        Ok(())
380    }
381
382    /// The cell with this id.
383    #[must_use]
384    pub fn cell(&self, id: &str) -> Option<&Cell> { self.cells.iter().find(|cell| cell.id == id) }
385}
386
387/// The lowercase hex SHA-256 of a contract's bytes, as `sha256sum` prints it.
388#[must_use]
389pub fn contract_digest(bytes: &[u8]) -> String { hex::encode(Sha256::digest(bytes)) }
krabka_parse_benches/contract.rs

krabka_parse_benches/
contract.rs
