feat(ingester): instrument the compaction sweep with RFC 0009 §3.6 metrics

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

crates/ourios-ingester/src/compactor.rs

@@ -4,17 +4,20 @@
 //! 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, 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 +67,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 +119,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 +210,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 +225,9 @@ impl Compactor {
     where
         F: FnMut(Result<SweepReport, IngestError>),
     {
+        // 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 +238,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).

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;

crates/ourios-ingester/src/metrics.rs

@@ -0,0 +1,176 @@
+//! 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<u64>,
+    partitions: Counter<u64>,
+    files: Counter<u64>,
+    rows: Counter<u64>,
+    orphan_files: Counter<u64>,
+    duration: Histogram<f64>,
+}
+
+impl CompactionMetrics {
+    /// 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)
+            .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 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();
+
+        // 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, &[]);
+        }
+
+        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<SweepReport, IngestError>, 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<String> {
+        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:?}",
+            );
+        }
+    }
+}