From e9165e4232527f2e6b7dea60ad4e834cd8902673 Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Wed, 3 Jun 2026 20:57:21 +0200 Subject: [PATCH 1/4] =?UTF-8?q?feat(ingester):=20instrument=20the=20compac?= =?UTF-8?q?tion=20sweep=20with=20RFC=200009=20=C2=A73.6=20metrics?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3 of the telemetry code work. Records the compaction metric set through OpenTelemetry, using the generated `ourios-semconv` name constants and the global meter (RFC 0001 §6.8 API/SDK split — the ingester depends only on the lightweight `opentelemetry` API; the SDK + OTLP exporter live in `ourios-telemetry`, configured by the binary). - New `CompactionMetrics` (crates/ourios-ingester/src/metrics.rs): builds the `ourios.compaction` instruments and seeds each counter with a zero measurement at construction so the full set is visible to the exporter before the first sweep (§6.8 collect-on-read). The duration histogram is deliberately not seeded — a synthetic `record(0)` would distort it. - `Compactor::run` builds the instruments once (seeded) before the loop and records each sweep: `sweeps` + `duration` carry the `ourios.compaction.result` attribute (committed / noop / error), and `partitions` / `files` / `rows` / `orphan.files` carry the volume. Sweep wall-clock is timed inside the blocking task. - `SweepReport` gains `files_compacted` (sum of each consolidated partition's `files_before`) for the H4 `ourios.compaction.files` signal. Deferred to a follow-up (need data this slice doesn't have): `ourios.compaction.io` (per-compaction byte counts aren't on `CompactionOutcome`), `ourios.compaction.backlog` (gauge semantics), and `ourios.storage.parquet.file.size` (an all-files H4 detector that belongs to the writer / a partition scan, not the compactor). Test: stands up an in-memory MeterProvider via `ourios-telemetry`'s `init_in_memory`, records a sweep, force-flushes, and asserts the exported stream carries every recorded instrument. Verification: cargo fmt --all --check, clippy --all-targets --all-features -D warnings, cargo test --workspace --all-features — all green. Co-Authored-By: Claude Opus 4.8 --- Cargo.lock | 4 + crates/ourios-ingester/Cargo.toml | 15 +++ crates/ourios-ingester/src/compactor.rs | 39 ++++-- crates/ourios-ingester/src/lib.rs | 2 + crates/ourios-ingester/src/metrics.rs | 158 ++++++++++++++++++++++++ 5 files changed, 209 insertions(+), 9 deletions(-) create mode 100644 crates/ourios-ingester/src/metrics.rs diff --git a/Cargo.lock b/Cargo.lock index 5b9bc06db..e76035e6d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2464,8 +2464,12 @@ dependencies = [ name = "ourios-ingester" version = "0.0.0" dependencies = [ + "opentelemetry", + "opentelemetry_sdk", "ourios-core", "ourios-parquet", + "ourios-semconv", + "ourios-telemetry", "tempfile", "tokio", ] diff --git a/crates/ourios-ingester/Cargo.toml b/crates/ourios-ingester/Cargo.toml index ca5d29333..f12104f57 100644 --- a/crates/ourios-ingester/Cargo.toml +++ b/crates/ourios-ingester/Cargo.toml @@ -23,9 +23,24 @@ ourios-parquet = { path = "../ourios-parquet" } # interval timer, `rt` for `spawn_blocking` (compaction is blocking # filesystem/Parquet work), `macros` for `#[tokio::main]`/tests. tokio = { version = "1", default-features = false, features = ["rt", "time", "macros"] } +# Telemetry: the lightweight OTel metrics API only (RFC 0001 §6.8 — the +# heavy SDK/OTLP live in ourios-telemetry; library crates resolve +# instruments through the global meter). +opentelemetry = { version = "0.32", default-features = false, features = ["metrics"] } +# Generated metric/attribute name constants (RFC 0009 §3.6); a leaf, +# zero-dep crate. +ourios-semconv = { path = "../ourios-semconv" } [dev-dependencies] tempfile = "3" +# Stand up an in-memory MeterProvider to assert the compaction metric +# stream; `testing` exposes `init_in_memory`. +ourios-telemetry = { path = "../ourios-telemetry", features = ["testing"] } +# The SDK data types (`ResourceMetrics` etc.) the metrics test reads. +opentelemetry_sdk = { version = "0.32", default-features = false, features = ["metrics", "testing"] } +# The metrics test collects via the in-memory reader's periodic-export +# path, which needs a multi-thread runtime. +tokio = { version = "1", default-features = false, features = ["rt", "rt-multi-thread", "time", "macros"] } [lints] workspace = true diff --git a/crates/ourios-ingester/src/compactor.rs b/crates/ourios-ingester/src/compactor.rs index 2fd9330f2..284f3e19b 100644 --- a/crates/ourios-ingester/src/compactor.rs +++ b/crates/ourios-ingester/src/compactor.rs @@ -9,12 +9,14 @@ //! wiring is a later slice). use std::path::{Path, PathBuf}; -use std::time::{Duration, SystemTime}; +use std::time::{Duration, Instant, SystemTime}; use ourios_parquet::{ CompactionError, CompactionPolicy, compact_partition, percent_decode_tenant, plan_candidates, }; +use crate::metrics::CompactionMetrics; + /// Failure during a compaction sweep. #[derive(Debug)] #[non_exhaustive] @@ -64,6 +66,10 @@ pub struct SweepReport { /// Partitions actually consolidated (a candidate that wasn't a /// no-op). pub partitions_compacted: usize, + /// Total input files merged away across those partitions (the + /// `files_before` of each consolidated partition) — the H4 + /// small-file signal (RFC 0009 §3.6 `ourios.compaction.files`). + pub files_compacted: u64, /// Total rows rewritten across those partitions. pub rows_compacted: u64, /// Superseded inputs that couldn't be removed post-commit (orphans @@ -112,6 +118,7 @@ pub fn run_sweep( Ok(outcome) => { if outcome.committed.is_some() { report.partitions_compacted += 1; + report.files_compacted += to_u64(outcome.files_before); report.rows_compacted += outcome.rows; } report.gc_failures += outcome.gc_failures; @@ -202,10 +209,11 @@ impl Compactor { /// Run sweeps forever, one per `interval` tick. Each sweep runs on /// the blocking pool (compaction is blocking I/O) as of the current /// wall clock; its [`SweepReport`]/[`IngestError`] result is handed - /// to `on_sweep` for logging/metrics — so one failing sweep is - /// observed, not fatal, and the loop keeps ticking. (Telemetry - /// wiring is a later slice; the observer is the seam.) Does not - /// return. + /// to `on_sweep` for logging — so one failing sweep is observed, + /// not fatal, and the loop keeps ticking. RFC 0009 §3.6 metrics are + /// recorded for every sweep via the `ourios.compaction` meter + /// (instruments built and seeded once here, before the loop). Does + /// not return. /// /// # Panics /// @@ -216,6 +224,9 @@ impl Compactor { where F: FnMut(Result), { + // Built (and zero-seeded) once, before the loop, so the metric + // set is visible to the exporter even before the first sweep. + let metrics = CompactionMetrics::new(); let mut ticker = tokio::time::interval(self.interval); // A maintenance sweep that overruns `interval` must not make // the next ticks fire back-to-back (the default `Burst`) — @@ -226,15 +237,25 @@ impl Compactor { ticker.tick().await; let bucket = self.bucket_root.clone(); let policy = self.policy; - let result = - tokio::task::spawn_blocking(move || run_sweep(&bucket, now_unix_nanos(), &policy)) - .await - .expect("compaction sweep task should not panic"); + let (result, elapsed) = tokio::task::spawn_blocking(move || { + let start = Instant::now(); + let result = run_sweep(&bucket, now_unix_nanos(), &policy); + (result, start.elapsed()) + }) + .await + .expect("compaction sweep task should not panic"); + metrics.record_sweep(&result, elapsed); on_sweep(result); } } } +/// Saturating `usize` → `u64` (lossless on 64-bit; saturates rather +/// than truncating on a theoretically wider target). +pub(crate) fn to_u64(value: usize) -> u64 { + u64::try_from(value).unwrap_or(u64::MAX) +} + /// `SystemTime::now()` as Unix nanoseconds (`0` if the clock is before /// the epoch; saturated at `u64::MAX` past year 2554 — neither is /// reachable in practice). diff --git a/crates/ourios-ingester/src/lib.rs b/crates/ourios-ingester/src/lib.rs index 88e72c8a4..b7a414c22 100644 --- a/crates/ourios-ingester/src/lib.rs +++ b/crates/ourios-ingester/src/lib.rs @@ -23,6 +23,8 @@ #![deny(unsafe_code)] pub mod compactor; +pub mod metrics; pub mod receiver; pub use compactor::{Compactor, IngestError, SweepReport, run_sweep}; +pub use metrics::CompactionMetrics; diff --git a/crates/ourios-ingester/src/metrics.rs b/crates/ourios-ingester/src/metrics.rs new file mode 100644 index 000000000..1ae56f5ff --- /dev/null +++ b/crates/ourios-ingester/src/metrics.rs @@ -0,0 +1,158 @@ +//! OpenTelemetry instruments for the compaction sweep (RFC 0009 §3.6). +//! +//! Names come from the generated [`ourios_semconv`] constants; the +//! instruments resolve through the **global** meter, per the RFC 0001 +//! §6.8 API/SDK split (the SDK + OTLP exporter live in +//! `ourios-telemetry`, configured once by the binary). When no provider +//! is installed the global meter is a no-op, so constructing and +//! recording is always safe. +//! +//! This slice records the metric set that the existing +//! [`CompactionOutcome`](ourios_parquet::CompactionOutcome) already +//! exposes — sweeps, partitions, files, rows, orphan files, and sweep +//! duration. `ourios.compaction.io` (needs per-compaction byte counts), +//! `ourios.compaction.backlog`, and `ourios.storage.parquet.file.size` +//! are deferred to a follow-up that plumbs the additional data. + +use std::time::Duration; + +use opentelemetry::metrics::{Counter, Histogram}; +use opentelemetry::{KeyValue, global}; +use ourios_semconv as semconv; + +use crate::compactor::{IngestError, SweepReport, to_u64}; + +/// The compaction metric instruments (RFC 0009 §3.6). Build one per +/// process and call [`CompactionMetrics::record_sweep`] once per sweep. +#[derive(Debug)] +pub struct CompactionMetrics { + sweeps: Counter, + partitions: Counter, + files: Counter, + rows: Counter, + orphan_files: Counter, + duration: Histogram, +} + +impl CompactionMetrics { + /// Build the instruments on the `ourios.compaction` meter, seeding + /// each counter with a zero measurement so the full set is visible + /// to the exporter before the first sweep records anything (RFC 0001 + /// §6.8 collect-on-read). The histogram is intentionally **not** + /// seeded: a synthetic `record(0)` would distort its distribution; + /// it surfaces on the first real sweep. + #[must_use] + pub fn new() -> Self { + let meter = global::meter("ourios.compaction"); + let sweeps = meter.u64_counter(semconv::OURIOS_COMPACTION_SWEEPS).build(); + let partitions = meter + .u64_counter(semconv::OURIOS_COMPACTION_PARTITIONS) + .build(); + let files = meter.u64_counter(semconv::OURIOS_COMPACTION_FILES).build(); + let rows = meter.u64_counter(semconv::OURIOS_COMPACTION_ROWS).build(); + let orphan_files = meter + .u64_counter(semconv::OURIOS_COMPACTION_ORPHAN_FILES) + .build(); + let duration = meter + .f64_histogram(semconv::OURIOS_COMPACTION_DURATION) + .with_unit("s") + .build(); + + for counter in [&sweeps, &partitions, &files, &rows, &orphan_files] { + counter.add(0, &[]); + } + + Self { + sweeps, + partitions, + files, + rows, + orphan_files, + duration, + } + } + + /// Record one sweep's outcome and wall-clock `elapsed`. The + /// `ourios.compaction.result` attribute classifies the sweep: + /// `error` if any tenant/partition failed (or the sweep itself + /// failed to scan the store), else `committed` if anything was + /// consolidated, else `noop`. + pub fn record_sweep(&self, result: &Result, elapsed: Duration) { + let outcome = match result { + Ok(report) if !report.errors.is_empty() => "error", + Ok(report) if report.partitions_compacted > 0 => "committed", + Ok(_) => "noop", + Err(_) => "error", + }; + let attrs = [KeyValue::new(semconv::OURIOS_COMPACTION_RESULT, outcome)]; + + self.sweeps.add(1, &attrs); + self.duration.record(elapsed.as_secs_f64(), &attrs); + + if let Ok(report) = result { + self.partitions + .add(to_u64(report.partitions_compacted), &[]); + self.files.add(report.files_compacted, &[]); + self.rows.add(report.rows_compacted, &[]); + self.orphan_files.add(to_u64(report.gc_failures), &[]); + } + } +} + +impl Default for CompactionMetrics { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use opentelemetry_sdk::metrics::data::{ResourceMetrics, ScopeMetrics}; + + use super::*; + + // Collected metric names across the in-memory export. + fn collected_names(rms: &[ResourceMetrics]) -> Vec { + rms.iter() + .flat_map(ResourceMetrics::scope_metrics) + .flat_map(ScopeMetrics::metrics) + .map(|metric| metric.name().to_string()) + .collect() + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + async fn record_sweep_exports_the_compaction_metric_set() { + // Arrange — an in-memory provider, then the instruments (built + // after, so they resolve against it). + let (guard, exporter) = ourios_telemetry::init_in_memory("ourios-test"); + let metrics = CompactionMetrics::new(); + let report = SweepReport { + tenants_scanned: 2, + partitions_compacted: 3, + files_compacted: 7, + rows_compacted: 100, + gc_failures: 1, + errors: Vec::new(), + }; + + // Act + metrics.record_sweep(&Ok(report), Duration::from_millis(50)); + guard.force_flush().expect("force_flush succeeds"); + + // Assert — every recorded instrument is in the exported stream. + let names = collected_names(&exporter.get_finished_metrics().expect("metrics exported")); + for expected in [ + semconv::OURIOS_COMPACTION_SWEEPS, + semconv::OURIOS_COMPACTION_PARTITIONS, + semconv::OURIOS_COMPACTION_FILES, + semconv::OURIOS_COMPACTION_ROWS, + semconv::OURIOS_COMPACTION_ORPHAN_FILES, + semconv::OURIOS_COMPACTION_DURATION, + ] { + assert!( + names.iter().any(|name| name == expected), + "exported stream missing {expected}, got {names:?}", + ); + } + } +} From 5daa3fee5dcb5c3ea0ca6968ff071d110071700a Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Wed, 3 Jun 2026 21:08:32 +0200 Subject: [PATCH 2/4] =?UTF-8?q?fixup!=20feat(ingester):=20instrument=20the?= =?UTF-8?q?=20compaction=20sweep=20with=20RFC=200009=20=C2=A73.6=20metrics?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/ourios-ingester/src/compactor.rs | 7 +++-- crates/ourios-ingester/src/metrics.rs | 38 ++++++++++++++++++------- 2 files changed, 32 insertions(+), 13 deletions(-) diff --git a/crates/ourios-ingester/src/compactor.rs b/crates/ourios-ingester/src/compactor.rs index 284f3e19b..bb3c7e71d 100644 --- a/crates/ourios-ingester/src/compactor.rs +++ b/crates/ourios-ingester/src/compactor.rs @@ -4,9 +4,10 @@ //! select its sealed candidate partitions and consolidate them. It is //! synchronous (blocking filesystem + Parquet work) and deterministic, //! so it's the unit the tests exercise. [`Compactor::run`] is the thin -//! daemon: it calls `run_sweep` on a fixed cadence via `spawn_blocking` -//! and hands each result to a caller-supplied observer (metrics/log -//! wiring is a later slice). +//! daemon: it calls `run_sweep` on a fixed cadence via `spawn_blocking`, +//! records the RFC 0009 §3.6 metrics for each sweep +//! ([`crate::metrics::CompactionMetrics`]), and hands each result to a +//! caller-supplied observer for logging. use std::path::{Path, PathBuf}; use std::time::{Duration, Instant, SystemTime}; diff --git a/crates/ourios-ingester/src/metrics.rs b/crates/ourios-ingester/src/metrics.rs index 1ae56f5ff..7f9181687 100644 --- a/crates/ourios-ingester/src/metrics.rs +++ b/crates/ourios-ingester/src/metrics.rs @@ -35,30 +35,48 @@ pub struct CompactionMetrics { } impl CompactionMetrics { - /// Build the instruments on the `ourios.compaction` meter, seeding - /// each counter with a zero measurement so the full set is visible - /// to the exporter before the first sweep records anything (RFC 0001 - /// §6.8 collect-on-read). The histogram is intentionally **not** - /// seeded: a synthetic `record(0)` would distort its distribution; - /// it surfaces on the first real sweep. + /// Build the instruments on the `ourios.compaction` meter, with the + /// UCUM units the registry (`semconv/registry/metrics.yaml`) + /// declares for each. + /// + /// The attribute-free counters are seeded with a zero measurement so + /// they are visible to the exporter before the first sweep records + /// anything (RFC 0001 §6.8 collect-on-read). `sweeps` and `duration` + /// are **not** seeded: their `ourios.compaction.result` attribute is + /// *required* (a `add(0, &[])` would emit a point missing it), and a + /// histogram `record(0)` would distort the distribution — both + /// surface on the first sweep with a real `result`. #[must_use] pub fn new() -> Self { let meter = global::meter("ourios.compaction"); - let sweeps = meter.u64_counter(semconv::OURIOS_COMPACTION_SWEEPS).build(); + let sweeps = meter + .u64_counter(semconv::OURIOS_COMPACTION_SWEEPS) + .with_unit("{sweep}") + .build(); let partitions = meter .u64_counter(semconv::OURIOS_COMPACTION_PARTITIONS) + .with_unit("{partition}") + .build(); + let files = meter + .u64_counter(semconv::OURIOS_COMPACTION_FILES) + .with_unit("{file}") + .build(); + let rows = meter + .u64_counter(semconv::OURIOS_COMPACTION_ROWS) + .with_unit("{row}") .build(); - let files = meter.u64_counter(semconv::OURIOS_COMPACTION_FILES).build(); - let rows = meter.u64_counter(semconv::OURIOS_COMPACTION_ROWS).build(); let orphan_files = meter .u64_counter(semconv::OURIOS_COMPACTION_ORPHAN_FILES) + .with_unit("{file}") .build(); let duration = meter .f64_histogram(semconv::OURIOS_COMPACTION_DURATION) .with_unit("s") .build(); - for counter in [&sweeps, &partitions, &files, &rows, &orphan_files] { + // Only the attribute-free counters; `sweeps` carries a required + // `result` attribute, so it is not zero-seeded here. + for counter in [&partitions, &files, &rows, &orphan_files] { counter.add(0, &[]); } From 900484dc0d652e92fd2b3b711d3d5c1ce202c1c2 Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Wed, 3 Jun 2026 21:16:48 +0200 Subject: [PATCH 3/4] =?UTF-8?q?fixup!=20fixup!=20feat(ingester):=20instrum?= =?UTF-8?q?ent=20the=20compaction=20sweep=20with=20RFC=200009=20=C2=A73.6?= =?UTF-8?q?=20metrics?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/ourios-ingester/src/metrics.rs | 35 +++++++++++++++++++++++++-- 1 file changed, 33 insertions(+), 2 deletions(-) diff --git a/crates/ourios-ingester/src/metrics.rs b/crates/ourios-ingester/src/metrics.rs index 7f9181687..e2612465d 100644 --- a/crates/ourios-ingester/src/metrics.rs +++ b/crates/ourios-ingester/src/metrics.rs @@ -125,7 +125,9 @@ impl Default for CompactionMetrics { #[cfg(test)] mod tests { - use opentelemetry_sdk::metrics::data::{ResourceMetrics, ScopeMetrics}; + use opentelemetry_sdk::metrics::data::{ + AggregatedMetrics, MetricData, ResourceMetrics, ScopeMetrics, + }; use super::*; @@ -158,7 +160,8 @@ mod tests { guard.force_flush().expect("force_flush succeeds"); // Assert — every recorded instrument is in the exported stream. - let names = collected_names(&exporter.get_finished_metrics().expect("metrics exported")); + let rms = exporter.get_finished_metrics().expect("metrics exported"); + let names = collected_names(&rms); for expected in [ semconv::OURIOS_COMPACTION_SWEEPS, semconv::OURIOS_COMPACTION_PARTITIONS, @@ -172,5 +175,33 @@ mod tests { "exported stream missing {expected}, got {names:?}", ); } + + // …and every `sweeps` datapoint carries the *required* + // `ourios.compaction.result` attribute (the round-1 seed bug + // emitted an attribute-less point), with the committed sweep + // classified as such. + let sweeps = rms + .iter() + .flat_map(ResourceMetrics::scope_metrics) + .flat_map(ScopeMetrics::metrics) + .find(|metric| metric.name() == semconv::OURIOS_COMPACTION_SWEEPS) + .expect("sweeps metric present"); + let AggregatedMetrics::U64(MetricData::Sum(sum)) = sweeps.data() else { + panic!("sweeps should be a u64 sum, got {:?}", sweeps.data()); + }; + assert!(sum.data_points().count() > 0, "a sweep was recorded"); + assert!( + sum.data_points().all(|dp| dp + .attributes() + .any(|kv| kv.key.as_str() == semconv::OURIOS_COMPACTION_RESULT)), + "every sweeps datapoint must carry the required result attribute", + ); + assert!( + sum.data_points().any(|dp| dp.attributes().any(|kv| { + kv.key.as_str() == semconv::OURIOS_COMPACTION_RESULT + && kv.value.as_str() == "committed" + })), + "the committed sweep is classified result=committed", + ); } } From 655cab17b0cc26ccedffd5a4e676e9f36a0f6049 Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Wed, 3 Jun 2026 21:30:21 +0200 Subject: [PATCH 4/4] =?UTF-8?q?fixup!=20fixup!=20fixup!=20feat(ingester):?= =?UTF-8?q?=20instrument=20the=20compaction=20sweep=20with=20RFC=200009=20?= =?UTF-8?q?=C2=A73.6=20metrics?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/ourios-ingester/src/compactor.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/crates/ourios-ingester/src/compactor.rs b/crates/ourios-ingester/src/compactor.rs index bb3c7e71d..9132801d4 100644 --- a/crates/ourios-ingester/src/compactor.rs +++ b/crates/ourios-ingester/src/compactor.rs @@ -340,6 +340,10 @@ mod tests { assert_eq!(report.tenants_scanned, 1); assert_eq!(report.partitions_compacted, 1); assert_eq!(report.rows_compacted, 2); + assert_eq!( + report.files_compacted, 2, + "both input files are merged away (the H4 signal)" + ); } #[test]