From 461651fcd45db768b85716477b67680a6ad27350 Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Thu, 18 Jun 2026 17:05:17 +0200 Subject: [PATCH 1/4] test(bench): ingest write-path criterion benches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds `ourios-bench/benches/ingest_write_path.rs` covering CLAUDE.md §6.2's named ingest hot paths, alongside the existing b1/b2/recovery/compaction benches: - `wal_append` — OTLP → WAL: append one batch frame + fsync (the WAL-before-ack unit), mirroring `ourios.wal.append.duration`. - `sink_write/{1000,10000}` — WAL → Parquet: feed N mined records to a `ParquetRecordSink` and flush to one Parquet object, mirroring `ourios.sink.flush.*`. Throughput in records. Both rebuild their fixture in the untimed `iter_batched` setup so only the durable append / emit + flush is measured. Supportive **wall-clock** evidence (like recovery/b1/b2) — to be run indicatively on the CI runner first; the authoritative baseline-hardware numbers are a separate, opt-in run (the `bench-on-ci-runner-first` discipline). Adds `ourios-ingester` as a dev-dep for the sink half. Co-Authored-By: Claude Opus 4.8 --- Cargo.lock | 1 + crates/ourios-bench/Cargo.toml | 8 + .../ourios-bench/benches/ingest_write_path.rs | 139 ++++++++++++++++++ 3 files changed, 148 insertions(+) create mode 100644 crates/ourios-bench/benches/ingest_write_path.rs diff --git a/Cargo.lock b/Cargo.lock index 3f46caac1..4e017815a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3001,6 +3001,7 @@ dependencies = [ "criterion", "opentelemetry-proto", "ourios-core", + "ourios-ingester", "ourios-miner", "ourios-parquet", "ourios-querier", diff --git a/crates/ourios-bench/Cargo.toml b/crates/ourios-bench/Cargo.toml index 29c7550cc..8893f543c 100644 --- a/crates/ourios-bench/Cargo.toml +++ b/crates/ourios-bench/Cargo.toml @@ -72,11 +72,19 @@ tokio = { version = "1", default-features = false, features = ["rt"] } # (the O(N) shape) mirroring b1/b2's role; the structural side is # pinned by `ourios-wal`'s `rfc0008_3_*` integration tests. ourios-wal = { path = "../ourios-wal" } +# Ingest write-path bench (CLAUDE.md §6.2 hot paths): drives the RFC 0014 +# `ParquetRecordSink` for the WAL → Parquet half; the WAL half uses +# `ourios-wal` directly. +ourios-ingester = { path = "../ourios-ingester" } [[bench]] name = "b1" harness = false +[[bench]] +name = "ingest_write_path" +harness = false + [[bench]] name = "b2" harness = false diff --git a/crates/ourios-bench/benches/ingest_write_path.rs b/crates/ourios-bench/benches/ingest_write_path.rs new file mode 100644 index 000000000..f185d1832 --- /dev/null +++ b/crates/ourios-bench/benches/ingest_write_path.rs @@ -0,0 +1,139 @@ +//! Ingest write-path wall-clock benches (`CLAUDE.md` §6.2 hot paths: +//! OTLP → WAL, WAL → Parquet). Supportive, **indicative** evidence — run on +//! the CI runner first (see the `bench-on-ci-runner-first` discipline); the +//! authoritative baseline-hardware numbers are a separate, opt-in run. +//! +//! - `wal_append` — the OTLP → WAL half: append one batch frame and fsync it +//! (the WAL-before-ack unit; one batch → one frame → one sync). Mirrors the +//! `ourios.wal.append.duration` metric. Throughput is reported in bytes. +//! - `sink_write/{N}` — the WAL → Parquet half: feed `N` mined records to a +//! `ParquetRecordSink` and flush them to one Parquet object on a +//! `LocalFileSystem` store (encode + put). Mirrors `ourios.sink.flush.*`. +//! Throughput is reported in records. +//! +//! Both rebuild their fixture in the (untimed) `iter_batched` setup so only +//! the durable append / the emit + flush is measured. + +use std::hint::black_box; +use std::path::Path; +use std::time::Duration; + +use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; + +use ourios_core::audit::ParamType; +use ourios_core::record::{BodyKind, MinedRecord, Param, RecordSink}; +use ourios_core::tenant::TenantId; +use ourios_ingester::record_sink::{FlushConfig, ParquetRecordSink}; +use ourios_parquet::Store; +use ourios_wal::{FrameKind, Wal, WalConfig}; + +/// A representative ack unit: a single batch frame (~one small OTLP export). +const FRAME_LEN: usize = 4 * 1024; +/// Record counts for the sink-flush throughput sweep. +const SINK_RECORDS: [usize; 2] = [1_000, 10_000]; + +fn wal_config(root: &Path) -> WalConfig { + WalConfig { + root: root.to_path_buf(), + batch_window_ms: 100, + segment_size_bytes: 128 * 1024 * 1024, + segment_age_secs: 600, + housekeeping_secs: 60, + macos_full_fsync: false, + } +} + +/// A clean mined record whose varying param keeps the column non-degenerate +/// (so encode/compression is representative, not a trivial all-equal run). +fn rec(i: u64) -> MinedRecord { + MinedRecord { + tenant_id: TenantId::new("bench"), + template_id: 1, + template_version: 1, + severity_number: 9, + severity_text: Some("INFO".to_string()), + scope_name: Some("lib.bench".to_string()), + scope_version: Some("1.0.0".to_string()), + time_unix_nano: 1_775_127_480_000_000_000 + i * 1_000, + observed_time_unix_nano: Some(1_775_127_480_000_000_000 + i * 1_000 + 1), + attributes: Vec::new(), + dropped_attributes_count: 0, + resource_attributes: Vec::new(), + trace_id: None, + span_id: None, + flags: 0x01, + event_name: None, + body_kind: BodyKind::String, + params: vec![Param { + type_tag: ParamType::Num, + value: format!("{i}"), + }], + separators: vec![String::new(), " ".to_string()], + body: None, + confidence: 1.0, + lossy_flag: false, + } +} + +/// OTLP → WAL: append one frame + fsync (the durable-ack unit). +fn wal_append(c: &mut Criterion) { + let mut group = c.benchmark_group("wal_append"); + group.throughput(Throughput::Bytes(FRAME_LEN as u64)); + let payload = vec![0xA5u8; FRAME_LEN]; + group.bench_function("batch", |b| { + b.iter_batched( + || { + // Keep the `TempDir` alive alongside the `Wal` — dropping it + // would delete the segment directory mid-measurement. + let dir = tempfile::TempDir::new().expect("temp"); + let wal = Wal::open(wal_config(dir.path())).expect("open"); + (dir, wal) + }, + |(_dir, mut wal)| { + wal.append(FrameKind::OtlpBatch, &payload).expect("append"); + black_box(wal.sync().expect("sync")); + }, + BatchSize::SmallInput, + ); + }); + group.finish(); +} + +/// WAL → Parquet: feed `N` records to the sink and flush to one object. +fn sink_write(c: &mut Criterion) { + let mut group = c.benchmark_group("sink_write"); + for n in SINK_RECORDS { + let records: Vec = (0..n as u64).map(rec).collect(); + group.throughput(Throughput::Elements(n as u64)); + group.bench_with_input(BenchmarkId::from_parameter(n), &records, |b, records| { + b.iter_batched( + || { + let dir = tempfile::TempDir::new().expect("temp"); + // Size never triggers mid-stream: emit the whole batch, + // then time one explicit flush of the full partition. + let sink = ParquetRecordSink::new( + Store::local(dir.path()).expect("store"), + FlushConfig { + target_bytes: usize::MAX, + max_buffer_age: Duration::from_secs(86_400), + ceiling_bytes: usize::MAX, + }, + ); + (dir, sink) + }, + |(_dir, mut sink)| { + for r in records { + sink.emit(r.clone()); + } + sink.flush_all(); + black_box(sink.flushes()); + }, + BatchSize::SmallInput, + ); + }); + } + group.finish(); +} + +criterion_group!(benches, wal_append, sink_write); +criterion_main!(benches); From 3a57de72958b0ef31d71e9fdf91576edda0a4372 Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Thu, 18 Jun 2026 17:46:57 +0200 Subject: [PATCH 2/4] test(bench): isolate the timed path (review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review on #248: - Clone the record batch in the untimed `iter_batched` setup; the timed `sink_write` routine now emits owned records, so per-record clone cost no longer pollutes the WAL→Parquet signal (sink throughput rises to ~1.09 M rec/s at 10k, previously understated). - `wal_append` warms up with one append + sync in setup so the timed sync measures steady state — the first sync on a fresh WAL also fsyncs the segment directory (entry durability), which steady-state syncs don't. Co-Authored-By: Claude Opus 4.8 --- .../ourios-bench/benches/ingest_write_path.rs | 22 ++++++++++++++----- 1 file changed, 16 insertions(+), 6 deletions(-) diff --git a/crates/ourios-bench/benches/ingest_write_path.rs b/crates/ourios-bench/benches/ingest_write_path.rs index f185d1832..da9da7c01 100644 --- a/crates/ourios-bench/benches/ingest_write_path.rs +++ b/crates/ourios-bench/benches/ingest_write_path.rs @@ -84,9 +84,16 @@ fn wal_append(c: &mut Criterion) { b.iter_batched( || { // Keep the `TempDir` alive alongside the `Wal` — dropping it - // would delete the segment directory mid-measurement. + // would delete the segment directory mid-measurement. Warm up + // with one append + sync: the *first* sync on a fresh WAL also + // fsyncs the segment directory (entry durability), which + // steady-state syncs don't, so doing it here (untimed) isolates + // the steady-state per-batch append + fsync cost. let dir = tempfile::TempDir::new().expect("temp"); - let wal = Wal::open(wal_config(dir.path())).expect("open"); + let mut wal = Wal::open(wal_config(dir.path())).expect("open"); + wal.append(FrameKind::OtlpBatch, &payload) + .expect("warm append"); + wal.sync().expect("warm sync"); (dir, wal) }, |(_dir, mut wal)| { @@ -119,11 +126,14 @@ fn sink_write(c: &mut Criterion) { ceiling_bytes: usize::MAX, }, ); - (dir, sink) + // Clone the batch in the untimed setup — production `emit` + // takes owned records, so the timed routine emits by value + // (no clone cost polluting the WAL→Parquet signal). + (dir, sink, records.clone()) }, - |(_dir, mut sink)| { - for r in records { - sink.emit(r.clone()); + |(_dir, mut sink, batch)| { + for r in batch { + sink.emit(r); } sink.flush_all(); black_box(sink.flushes()); From 0fbac0126ff7ed25edadba7b73a12a109f6e7eb0 Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Thu, 18 Jun 2026 18:04:08 +0200 Subject: [PATCH 3/4] test(bench): time only the work via iter_custom (review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 2 review on #248: `iter_batched` ran fixture teardown (dropping `TempDir`/`Wal`/sink → file close, Parquet unlink, dir delete) inside the timed routine. Switch both benches to `iter_custom` with an explicit `Instant` around just the append+fsync / emit+flush, so build and teardown are excluded and only one fixture is alive at a time (deferring drops via iter_batched outputs would instead accumulate open WALs/temp dirs across a sample batch). Teardown was a large fraction — `sink_write/1000` ~halves and sink throughput rises to ~1.56 M rec/s at 10k. Module doc updated. Co-Authored-By: Claude Opus 4.8 --- .../ourios-bench/benches/ingest_write_path.rs | 73 +++++++++++-------- 1 file changed, 41 insertions(+), 32 deletions(-) diff --git a/crates/ourios-bench/benches/ingest_write_path.rs b/crates/ourios-bench/benches/ingest_write_path.rs index da9da7c01..2c9cddac3 100644 --- a/crates/ourios-bench/benches/ingest_write_path.rs +++ b/crates/ourios-bench/benches/ingest_write_path.rs @@ -11,14 +11,17 @@ //! `LocalFileSystem` store (encode + put). Mirrors `ourios.sink.flush.*`. //! Throughput is reported in records. //! -//! Both rebuild their fixture in the (untimed) `iter_batched` setup so only -//! the durable append / the emit + flush is measured. +//! Both use `iter_custom` and time only the durable append / the emit + flush +//! with an explicit `Instant` — fixture build **and** teardown (`TempDir` +//! delete, file close, Parquet unlink) fall outside the measured span, and +//! only one fixture is alive at a time (no accumulation of open WALs / temp +//! dirs across a sample batch). use std::hint::black_box; use std::path::Path; -use std::time::Duration; +use std::time::{Duration, Instant}; -use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; use ourios_core::audit::ParamType; use ourios_core::record::{BodyKind, MinedRecord, Param, RecordSink}; @@ -81,27 +84,28 @@ fn wal_append(c: &mut Criterion) { group.throughput(Throughput::Bytes(FRAME_LEN as u64)); let payload = vec![0xA5u8; FRAME_LEN]; group.bench_function("batch", |b| { - b.iter_batched( - || { - // Keep the `TempDir` alive alongside the `Wal` — dropping it - // would delete the segment directory mid-measurement. Warm up - // with one append + sync: the *first* sync on a fresh WAL also - // fsyncs the segment directory (entry durability), which - // steady-state syncs don't, so doing it here (untimed) isolates - // the steady-state per-batch append + fsync cost. + b.iter_custom(|iters| { + let mut total = Duration::ZERO; + for _ in 0..iters { + // Untimed setup. Warm up with one append + sync: the *first* + // sync on a fresh WAL also fsyncs the segment directory (entry + // durability), which steady-state syncs don't — so it stays out + // of the measured span, isolating the steady-state cost. let dir = tempfile::TempDir::new().expect("temp"); let mut wal = Wal::open(wal_config(dir.path())).expect("open"); wal.append(FrameKind::OtlpBatch, &payload) .expect("warm append"); wal.sync().expect("warm sync"); - (dir, wal) - }, - |(_dir, mut wal)| { + + // Timed: one steady-state append + fsync (the durable-ack unit). + let start = Instant::now(); wal.append(FrameKind::OtlpBatch, &payload).expect("append"); black_box(wal.sync().expect("sync")); - }, - BatchSize::SmallInput, - ); + total += start.elapsed(); + // `wal` + `dir` drop here, after the timer — teardown untimed. + } + total + }); }); group.finish(); } @@ -113,12 +117,16 @@ fn sink_write(c: &mut Criterion) { let records: Vec = (0..n as u64).map(rec).collect(); group.throughput(Throughput::Elements(n as u64)); group.bench_with_input(BenchmarkId::from_parameter(n), &records, |b, records| { - b.iter_batched( - || { + b.iter_custom(|iters| { + let mut total = Duration::ZERO; + for _ in 0..iters { + // Untimed setup. Size never triggers mid-stream (target = + // MAX), so the whole batch buffers and one explicit + // `flush_all` writes it. The batch is cloned here because + // production `emit` takes owned records — keeping the clone + // out of the measured span. let dir = tempfile::TempDir::new().expect("temp"); - // Size never triggers mid-stream: emit the whole batch, - // then time one explicit flush of the full partition. - let sink = ParquetRecordSink::new( + let mut sink = ParquetRecordSink::new( Store::local(dir.path()).expect("store"), FlushConfig { target_bytes: usize::MAX, @@ -126,20 +134,21 @@ fn sink_write(c: &mut Criterion) { ceiling_bytes: usize::MAX, }, ); - // Clone the batch in the untimed setup — production `emit` - // takes owned records, so the timed routine emits by value - // (no clone cost polluting the WAL→Parquet signal). - (dir, sink, records.clone()) - }, - |(_dir, mut sink, batch)| { + let batch = records.clone(); + + // Timed: emit the owned batch + flush to one Parquet object. + let start = Instant::now(); for r in batch { sink.emit(r); } sink.flush_all(); black_box(sink.flushes()); - }, - BatchSize::SmallInput, - ); + total += start.elapsed(); + // `sink` + `dir` drop here, after the timer — teardown + // (Parquet unlink, dir delete) untimed. + } + total + }); }); } group.finish(); From 6d95d1cd1e956f11500d4fc133d9b725ba12a23f Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Thu, 18 Jun 2026 18:16:29 +0200 Subject: [PATCH 4/4] test(bench): assert the sink drained, so a failed flush fails loudly (review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 3 review on #248: `ParquetRecordSink::flush_all` swallows encode/ store errors and retains the buffer, so a silent flush failure would still report a (bogus) `sink_write` timing for the error path. Add an untimed `assert_eq!(sink.buffered_records(), 0)` right after the timer — a failed flush now panics the bench instead of benchmarking the retry path. (`wal_append` is already covered: its append/sync `.expect()`s panic on error.) Co-Authored-By: Claude Opus 4.8 --- crates/ourios-bench/benches/ingest_write_path.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/crates/ourios-bench/benches/ingest_write_path.rs b/crates/ourios-bench/benches/ingest_write_path.rs index 2c9cddac3..6b0008ec7 100644 --- a/crates/ourios-bench/benches/ingest_write_path.rs +++ b/crates/ourios-bench/benches/ingest_write_path.rs @@ -142,8 +142,14 @@ fn sink_write(c: &mut Criterion) { sink.emit(r); } sink.flush_all(); - black_box(sink.flushes()); total += start.elapsed(); + // Untimed: `flush_all` swallows encode/store errors and + // retains the buffer, so assert the partition actually + // drained — otherwise a silent failure would report + // throughput for an error path. (`flush_all`'s store write + // is a real side effect, so it isn't elided without a + // `black_box`.) + assert_eq!(sink.buffered_records(), 0, "sink flush did not drain"); // `sink` + `dir` drop here, after the timer — teardown // (Parquet unlink, dir delete) untimed. }