krabka_parse_benches/
record.rs

1//! The performance record: every raw trial of one comparison, with the
2//! provenance needed to reproduce it.
3//!
4//! `packaging/performance/qualify.sh budget` writes one directory per trial.
5//! [`assemble`] reads that layout into a [`PerformanceRecord`], which is what
6//! is published and what [`evaluate`](crate::budget::evaluate) judges.
7//!
8//! ```text
9//! <artifact-dir>/
10//!   contract.json              the contract, copied before the first trial
11//!   contract.sha256            its digest, as `sha256sum` printed it
12//!   provenance.env             revisions, images, digests, settings
13//!   host.env                   host resources
14//!   trials/round-<r>/<side>/<cell>/
15//!     order                    the trial's position in the run, from 1
16//!     workload.json            BrokerPerformanceWorkload's result, if it finished
17//!     workload.log             its stderr
18//!     resources-before.txt     per-broker /proc snapshot before the trial
19//!     resources-after.txt      the same snapshot after it
20//! ```
21
22use {
23    crate::{
24        contract::{BudgetError, Contract, FORMAT_VERSION, Mode, contract_digest},
25        num::to_f64,
26    },
27    serde::{Deserialize, Serialize},
28    std::{
29        collections::BTreeMap,
30        fs,
31        io::ErrorKind,
32        path::{Path, PathBuf},
33        str::FromStr,
34    },
35};
36
37/// Which side of the comparison a trial measured.
38#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
39#[serde(rename_all = "snake_case")]
40pub enum Side {
41    /// The revision or broker the candidate is compared against.
42    Reference,
43    /// The revision under test.
44    Candidate,
45}
46
47impl Side {
48    /// Both sides, reference first.
49    pub const BOTH: [Self; 2] = [Self::Reference, Self::Candidate];
50
51    /// The directory name and record spelling of this side.
52    #[must_use]
53    pub fn name(self) -> &'static str {
54        match self {
55            Self::Reference => "reference",
56            Self::Candidate => "candidate",
57        }
58    }
59}
60
61/// One side's exact build.
62#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
63#[serde(deny_unknown_fields)]
64pub struct Subject {
65    /// What the report calls this side, such as `krabka` or `kafka`.
66    pub label: String,
67    /// The git revision, or the Kafka release for a pinned Kafka reference.
68    pub revision: String,
69    /// The image reference the cluster ran.
70    pub image: String,
71    /// The image's content digest.
72    pub image_digest: String,
73}
74
75/// The host every trial ran on.
76#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
77#[serde(deny_unknown_fields)]
78pub struct Host {
79    /// `model name` from `/proc/cpuinfo`.
80    pub cpu_model: String,
81    /// Logical CPUs online.
82    pub logical_cpus: u32,
83    /// `MemTotal`, in bytes.
84    pub memory_bytes: u64,
85    /// Bytes available on the artifact filesystem when the run started.
86    pub disk_available_bytes: u64,
87    /// `uname -r`.
88    pub kernel: String,
89    /// `getconf CLK_TCK`, the unit of `/proc/<pid>/stat` CPU times.
90    pub clock_ticks_per_second: u32,
91}
92
93/// What one completed workload measured.
94#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
95#[serde(deny_unknown_fields)]
96pub struct Measurement {
97    /// Records the producer saw acknowledged.
98    pub sent: u64,
99    /// Distinct records the consumer read back.
100    pub consumed: u64,
101    /// Records the consumer read more than once.
102    pub duplicates: u64,
103    /// Producer send failures.
104    pub errors: u64,
105    /// Wall time from the first send to the last record consumed.
106    pub seconds: f64,
107    /// Consumed records per second over that wall time.
108    pub records_per_second: f64,
109    /// Consumed payload MiB per second over that wall time.
110    pub mib_per_second: f64,
111    /// Median end-to-end latency.
112    pub latency_ms_p50: f64,
113    /// 95th percentile end-to-end latency.
114    pub latency_ms_p95: f64,
115    /// 99th percentile end-to-end latency.
116    pub latency_ms_p99: f64,
117    /// Broker CPU seconds over the trial, summed over every broker.
118    pub cpu_seconds: f64,
119    /// Peak resident set size, summed over every broker, in bytes.
120    pub peak_rss_bytes: u64,
121}
122
123/// How a trial ended.
124#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
125#[serde(tag = "status", rename_all = "snake_case")]
126pub enum TrialOutcome {
127    /// The workload finished and reported a result.
128    Completed(Measurement),
129    /// The workload or the harness failed; nothing is measured.
130    Failed {
131        /// Why, as specifically as the artifacts say.
132        reason: String,
133    },
134}
135
136/// One raw trial.
137#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
138#[serde(deny_unknown_fields)]
139pub struct Trial {
140    /// The contract cell this trial ran.
141    pub cell: String,
142    /// The interleaved round, from 1.
143    pub round: usize,
144    /// Which side ran.
145    pub side: Side,
146    /// The trial's position in the whole run, from 1.
147    pub order: usize,
148    /// How it ended.
149    pub outcome: TrialOutcome,
150}
151
152/// Every trial of one comparison and where it came from.
153#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
154#[serde(deny_unknown_fields)]
155pub struct PerformanceRecord {
156    /// Must be [`FORMAT_VERSION`].
157    pub format_version: u32,
158    /// The contract's name.
159    pub contract_name: String,
160    /// The SHA-256 of the contract's bytes, captured before the first trial.
161    pub contract_sha256: String,
162    /// What the candidate was compared against.
163    pub mode: Mode,
164    /// The reference side.
165    pub reference: Subject,
166    /// The candidate side.
167    pub candidate: Subject,
168    /// The client image whose jars drove both sides.
169    pub client_image: String,
170    /// That image's digest.
171    pub client_image_digest: String,
172    /// The host.
173    pub host: Host,
174    /// When the first trial started, RFC 3339 UTC.
175    pub started_at: String,
176    /// Every other harness setting, verbatim from `provenance.env`.
177    pub settings: BTreeMap<String, String>,
178    /// Every trial that ran, in no particular order.
179    pub trials: Vec<Trial>,
180}
181
182impl PerformanceRecord {
183    /// Read a record from its JSON bytes.
184    ///
185    /// # Errors
186    ///
187    /// Returns [`BudgetError`] when the bytes are not a record or declare
188    /// another format version.
189    pub fn from_json(bytes: &[u8]) -> Result<Self, BudgetError> {
190        let record: Self = serde_json::from_slice(bytes)?;
191        if record.format_version != FORMAT_VERSION {
192            return Err(BudgetError::FormatVersion {
193                what: "record",
194                found: record.format_version,
195            });
196        }
197        Ok(record)
198    }
199}
200
201fn artifact_error(path: &str, reason: impl Into<String>) -> BudgetError {
202    BudgetError::InvalidArtifact {
203        path: path.to_owned(),
204        reason: reason.into(),
205    }
206}
207
208/// Parse `key=value` lines. Blank lines and `#` comments are skipped; a
209/// repeated key is an error, so a snapshot cannot silently lose a value.
210///
211/// # Errors
212///
213/// Returns [`BudgetError::InvalidArtifact`] for a line without `=` or a
214/// repeated key.
215pub fn parse_env(path: &str, text: &str) -> Result<BTreeMap<String, String>, BudgetError> {
216    let mut values = BTreeMap::new();
217    for line in text.lines() {
218        let line = line.trim();
219        if line.is_empty() || line.starts_with('#') {
220            continue;
221        }
222        let Some((key, value)) = line.split_once('=') else {
223            return Err(artifact_error(path, format!("line {line:?} has no '='")));
224        };
225        if values
226            .insert(key.trim().to_owned(), value.trim().to_owned())
227            .is_some()
228        {
229            return Err(artifact_error(path, format!("key {key} appears twice")));
230        }
231    }
232    Ok(values)
233}
234
235fn take(
236    path: &str,
237    values: &mut BTreeMap<String, String>,
238    key: &str,
239) -> Result<String, BudgetError> {
240    values
241        .remove(key)
242        .filter(|value| !value.is_empty())
243        .ok_or_else(|| artifact_error(path, format!("missing {key}")))
244}
245
246fn take_number<T: FromStr>(
247    path: &str,
248    values: &mut BTreeMap<String, String>,
249    key: &str,
250) -> Result<T, BudgetError> {
251    let raw = take(path, values, key)?;
252    raw.parse()
253        .map_err(|_| artifact_error(path, format!("{key}={raw} is not a number")))
254}
255
256/// Parse `host.env`.
257///
258/// # Errors
259///
260/// Returns [`BudgetError::InvalidArtifact`] for a missing or malformed key.
261pub fn parse_host(text: &str) -> Result<Host, BudgetError> {
262    let path = "host.env";
263    let mut values = parse_env(path, text)?;
264    let host = Host {
265        cpu_model: take(path, &mut values, "cpu_model")?,
266        logical_cpus: take_number(path, &mut values, "logical_cpus")?,
267        memory_bytes: take_number(path, &mut values, "memory_bytes")?,
268        disk_available_bytes: take_number(path, &mut values, "disk_available_bytes")?,
269        kernel: take(path, &mut values, "kernel")?,
270        clock_ticks_per_second: take_number(path, &mut values, "clock_ticks_per_second")?,
271    };
272    if host.clock_ticks_per_second == 0 {
273        return Err(artifact_error(
274            path,
275            "clock_ticks_per_second must be positive",
276        ));
277    }
278    Ok(host)
279}
280
281/// One broker process in a resource snapshot.
282#[derive(Debug, Clone, Copy, PartialEq, Eq)]
283pub struct ProcessSample {
284    /// `utime + stime` from `/proc/<pid>/stat`, in clock ticks.
285    pub cpu_ticks: u64,
286    /// `starttime` from `/proc/<pid>/stat`, in clock ticks since boot. A
287    /// container restart keeps the pod name but changes this.
288    pub process_start_ticks: u64,
289    /// `VmHWM` from `/proc/<pid>/status`: peak RSS since the process started.
290    pub rss_hwm_kib: u64,
291}
292
293/// Parse a resource snapshot: blocks that each open with `pod=<name>`.
294///
295/// Keys this function does not need, such as `fd_count` or the network
296/// counters, are kept in the raw artifact and skipped here.
297///
298/// # Errors
299///
300/// Returns [`BudgetError::InvalidArtifact`] for a key outside a pod block, a
301/// pod named twice, or a pod without `cpu_ticks`, `process_start_ticks` or
302/// `rss_hwm_kib`.
303pub fn parse_snapshot(
304    path: &str,
305    text: &str,
306) -> Result<BTreeMap<String, ProcessSample>, BudgetError> {
307    let mut pods: BTreeMap<String, BTreeMap<String, String>> = BTreeMap::new();
308    let mut current: Option<String> = None;
309    for line in text.lines() {
310        let line = line.trim();
311        if line.is_empty() {
312            continue;
313        }
314        let Some((key, value)) = line.split_once('=') else {
315            return Err(artifact_error(path, format!("line {line:?} has no '='")));
316        };
317        if key == "pod" {
318            if pods.insert(value.to_owned(), BTreeMap::new()).is_some() {
319                return Err(artifact_error(path, format!("pod {value} appears twice")));
320            }
321            current = Some(value.to_owned());
322            continue;
323        }
324        let Some(pod) = &current else {
325            return Err(artifact_error(
326                path,
327                format!("{key} precedes the first pod="),
328            ));
329        };
330        if let Some(block) = pods.get_mut(pod) {
331            block.insert(key.to_owned(), value.to_owned());
332        }
333    }
334    if pods.is_empty() {
335        return Err(artifact_error(path, "no pod in the snapshot"));
336    }
337    pods.into_iter()
338        .map(|(pod, mut block)| {
339            let sample = ProcessSample {
340                cpu_ticks: take_number(path, &mut block, "cpu_ticks")?,
341                process_start_ticks: take_number(path, &mut block, "process_start_ticks")?,
342                rss_hwm_kib: take_number(path, &mut block, "rss_hwm_kib")?,
343            };
344            Ok((pod, sample))
345        })
346        .collect()
347}
348
349/// Broker CPU seconds and summed peak RSS bytes between two snapshots.
350///
351/// # Errors
352///
353/// Returns the failure reason when the two snapshots name different pods, or a
354/// pod's process start time or CPU counter differs in a way one process cannot
355/// produce. Each means a broker process was replaced during the trial, so its
356/// numbers describe no single process. A restarted container keeps its pod
357/// name, so the start time is what catches it.
358pub fn resource_delta(
359    before: &BTreeMap<String, ProcessSample>,
360    after: &BTreeMap<String, ProcessSample>,
361    clock_ticks_per_second: u32,
362) -> Result<(f64, u64), String> {
363    if before.keys().ne(after.keys()) {
364        return Err("the broker pod set changed during the trial".to_owned());
365    }
366    let mut ticks = 0_u64;
367    let mut rss_bytes = 0_u64;
368    for (pod, start) in before {
369        let end = after[pod];
370        if end.process_start_ticks != start.process_start_ticks {
371            return Err(format!(
372                "{pod}'s broker process started again during the trial"
373            ));
374        }
375        let Some(used) = end.cpu_ticks.checked_sub(start.cpu_ticks) else {
376            return Err(format!(
377                "{pod}'s CPU counter went backwards: its broker process restarted"
378            ));
379        };
380        ticks = ticks.saturating_add(used);
381        rss_bytes = rss_bytes.saturating_add(end.rss_hwm_kib.saturating_mul(1024));
382    }
383    Ok((to_f64(ticks) / f64::from(clock_ticks_per_second), rss_bytes))
384}
385
386/// The JSON object `BrokerPerformanceWorkload` prints on success.
387#[derive(Debug, Deserialize)]
388struct WorkloadResult {
389    sent: u64,
390    consumed: u64,
391    duplicates: u64,
392    errors: u64,
393    seconds: f64,
394    records_per_second: f64,
395    mib_per_second: f64,
396    latency_ms_p50: f64,
397    latency_ms_p95: f64,
398    latency_ms_p99: f64,
399}
400
401/// The last non-empty line of a workload log, which is where the workload's
402/// own `IllegalStateException` names the counts that disagreed.
403fn last_log_line(log: &str) -> Option<&str> {
404    log.lines().map(str::trim).rfind(|line| !line.is_empty())
405}
406
407fn read_optional(path: &Path) -> Result<Option<String>, BudgetError> {
408    match fs::read_to_string(path) {
409        Ok(text) => Ok(Some(text)),
410        Err(error) if error.kind() == ErrorKind::NotFound => Ok(None),
411        Err(error) => Err(error.into()),
412    }
413}
414
415/// Read one trial directory into its outcome.
416fn trial_outcome(
417    directory: &Path,
418    relative: &str,
419    clock_ticks_per_second: u32,
420) -> Result<TrialOutcome, BudgetError> {
421    let failed = |reason: String| Ok(TrialOutcome::Failed { reason });
422    let Some(workload) = read_optional(&directory.join("workload.json"))? else {
423        let log = read_optional(&directory.join("workload.log"))?.unwrap_or_default();
424        return failed(last_log_line(&log).map_or_else(
425            || "the workload reported no result".to_owned(),
426            |line| format!("the workload reported no result: {line}"),
427        ));
428    };
429    let Ok(result) = serde_json::from_str::<WorkloadResult>(&workload) else {
430        return failed("workload.json is not a workload result".to_owned());
431    };
432    let snapshot = |name: &str| -> Result<Option<BTreeMap<String, ProcessSample>>, BudgetError> {
433        read_optional(&directory.join(name))?
434            .map(|text| parse_snapshot(&format!("{relative}/{name}"), &text))
435            .transpose()
436    };
437    let (Some(before), Some(after)) = (
438        snapshot("resources-before.txt")?,
439        snapshot("resources-after.txt")?,
440    ) else {
441        return failed("a resource snapshot is missing".to_owned());
442    };
443    let (cpu_seconds, peak_rss_bytes) =
444        match resource_delta(&before, &after, clock_ticks_per_second) {
445            Ok(delta) => delta,
446            Err(reason) => return failed(reason),
447        };
448    Ok(TrialOutcome::Completed(Measurement {
449        sent: result.sent,
450        consumed: result.consumed,
451        duplicates: result.duplicates,
452        errors: result.errors,
453        seconds: result.seconds,
454        records_per_second: result.records_per_second,
455        mib_per_second: result.mib_per_second,
456        latency_ms_p50: result.latency_ms_p50,
457        latency_ms_p95: result.latency_ms_p95,
458        latency_ms_p99: result.latency_ms_p99,
459        cpu_seconds,
460        peak_rss_bytes,
461    }))
462}
463
464fn parse_mode(path: &str, raw: &str) -> Result<Mode, BudgetError> {
465    [Mode::BeforeAfter, Mode::PinnedKafka]
466        .into_iter()
467        .find(|mode| mode.name() == raw)
468        .ok_or_else(|| artifact_error(path, format!("mode={raw} is not a mode")))
469}
470
471/// Assemble the record of a `qualify.sh budget` artifact directory.
472///
473/// Every (round, side, cell) the contract expects and whose trial directory
474/// exists becomes one [`Trial`]. A directory that is absent means the harness
475/// stopped before reaching it: that trial is left out, and evaluation reports
476/// it missing.
477///
478/// # Errors
479///
480/// Returns [`BudgetError`] when the copied contract is invalid or does not
481/// match its recorded digest, or a provenance, host or snapshot file is
482/// malformed.
483pub fn assemble(artifact_dir: &Path) -> Result<PerformanceRecord, BudgetError> {
484    let contract_bytes = fs::read(artifact_dir.join("contract.json"))?;
485    let contract = Contract::from_json(&contract_bytes)?;
486    let mut provenance = parse_env(
487        "provenance.env",
488        &fs::read_to_string(artifact_dir.join("provenance.env"))?,
489    )?;
490    let path = "provenance.env";
491    let recorded_digest = take(path, &mut provenance, "contract_sha256")?;
492    if recorded_digest != contract_digest(&contract_bytes) {
493        return Err(artifact_error(
494            "contract.json",
495            "its bytes do not match the contract_sha256 recorded before the run",
496        ));
497    }
498    let mode = parse_mode(path, &take(path, &mut provenance, "mode")?)?;
499    let mut subject = |side: Side| -> Result<Subject, BudgetError> {
500        let prefix = side.name();
501        Ok(Subject {
502            label: take(path, &mut provenance, &format!("{prefix}_label"))?,
503            revision: take(path, &mut provenance, &format!("{prefix}_revision"))?,
504            image: take(path, &mut provenance, &format!("{prefix}_image"))?,
505            image_digest: take(path, &mut provenance, &format!("{prefix}_image_digest"))?,
506        })
507    };
508    let reference = subject(Side::Reference)?;
509    let candidate = subject(Side::Candidate)?;
510    let client_image = take(path, &mut provenance, "client_image")?;
511    let client_image_digest = take(path, &mut provenance, "client_image_digest")?;
512    let started_at = take(path, &mut provenance, "started_at")?;
513    let host = parse_host(&fs::read_to_string(artifact_dir.join("host.env"))?)?;
514
515    let mut trials = Vec::new();
516    for round in 1..=contract.rounds {
517        for side in Side::BOTH {
518            for cell in &contract.cells {
519                let relative = format!("trials/round-{round}/{}/{}", side.name(), cell.id);
520                let directory: PathBuf = artifact_dir.join(&relative);
521                if !directory.is_dir() {
522                    continue;
523                }
524                let order_path = format!("{relative}/order");
525                let order_text = fs::read_to_string(directory.join("order"))?;
526                let order = order_text
527                    .trim()
528                    .parse()
529                    .map_err(|_| artifact_error(&order_path, "not a trial position"))?;
530                trials.push(Trial {
531                    cell: cell.id.clone(),
532                    round,
533                    side,
534                    order,
535                    outcome: trial_outcome(&directory, &relative, host.clock_ticks_per_second)?,
536                });
537            }
538        }
539    }
540    trials.sort_by_key(|trial| trial.order);
541    Ok(PerformanceRecord {
542        format_version: FORMAT_VERSION,
543        contract_name: contract.name,
544        contract_sha256: recorded_digest,
545        mode,
546        reference,
547        candidate,
548        client_image,
549        client_image_digest,
550        host,
551        started_at,
552        settings: provenance,
553        trials,
554    })
555}
krabka_parse_benches/record.rs

krabka_parse_benches/
record.rs
