diff --git a/crates/ourios-ingester/Cargo.toml b/crates/ourios-ingester/Cargo.toml index 3a3704d6f..37927167a 100644 --- a/crates/ourios-ingester/Cargo.toml +++ b/crates/ourios-ingester/Cargo.toml @@ -81,7 +81,7 @@ tonic = { version = "0.14", default-features = false, features = ["codegen"] } # Parse the snapshot-recorded WAL high-water segment id back into the # `uuid::Uuid` of `ourios_wal::WalOffset` (RFC 0001 §6.9 / RFC 0008 # §6.6). Same major as ourios-wal's, so the types unify. -uuid = { version = "1", default-features = false, features = ["std"] } +uuid = { version = "1", default-features = false, features = ["std", "v7"] } [dev-dependencies] tempfile = "3" diff --git a/crates/ourios-ingester/src/lib.rs b/crates/ourios-ingester/src/lib.rs index daf3a1f11..7d48b952b 100644 --- a/crates/ourios-ingester/src/lib.rs +++ b/crates/ourios-ingester/src/lib.rs @@ -35,6 +35,7 @@ pub mod compactor; pub mod metrics; pub mod receiver; +pub mod record_sink; pub mod recovery; pub mod snapshot_store; diff --git a/crates/ourios-ingester/src/record_sink.rs b/crates/ourios-ingester/src/record_sink.rs new file mode 100644 index 000000000..7e3e805fc --- /dev/null +++ b/crates/ourios-ingester/src/record_sink.rs @@ -0,0 +1,296 @@ +//! Production data write path (RFC 0014): a buffering [`RecordSink`] that +//! accumulates mined records per partition and flushes each to a Parquet +//! object on the RFC 0013 [`Store`] seam. +//! +//! Flush policy (RFC 0014 §3.2, hybrid): a partition flushes when its buffered +//! bytes reach [`FlushConfig::target_bytes`] (size, evaluated on `emit`), when +//! its oldest record reaches [`FlushConfig::max_buffer_age`] (age, evaluated by +//! [`ParquetRecordSink::flush_aged`] on the batch-window tick), or when the WAL +//! segment rotates ([`ParquetRecordSink::flush_all`], force-flushing *every* +//! partition). Buffered bytes are kept under [`FlushConfig::ceiling_bytes`] +//! by flushing the largest partition inline before `emit` would exceed it +//! (RFC 0014 §3.4) — a hard ceiling whenever the store accepts writes; a flush +//! failure retains the buffer and may transiently exceed it rather than +//! stalling ingest (see [`FlushConfig::ceiling_bytes`]). +//! +//! Records reach the sink only after the WAL is durable (`CLAUDE.md` §3.4), so +//! a buffer is a bounded accelerator, never the durability of record: a crash +//! re-mines the un-flushed tail from the WAL. A flush failure therefore retains +//! the buffer (counted, retried on the next trigger) rather than dropping data. +//! Buffers are keyed by [`PartitionKey`], which carries `tenant_id`, so they +//! are tenant-scoped by construction (`CLAUDE.md` §3.7). + +use std::collections::HashMap; +use std::path::Path; +use std::time::{Duration, Instant}; + +use ourios_core::record::{MinedRecord, RecordSink}; +use ourios_parquet::{ + DEFAULT_ZSTD_LEVEL, PartitionKey, Store, StoreError, WriterError, encode_records_to_parquet, +}; +use uuid::Uuid; + +/// Flush-policy knobs (RFC 0014 §3; RFC 0004 config at the call site). +#[derive(Debug, Clone)] +pub struct FlushConfig { + /// Size trigger: a partition flushes once its estimated buffered bytes + /// reach this. Production targets RFC 0005 §3.5's 256 MiB–2 GiB file band; + /// tests use small values. (Tuning is RFC 0014 §7.) + pub target_bytes: usize, + /// Age trigger: a partition flushes once its oldest buffered record's age + /// reaches this (inclusive), bounding low-volume staleness. + pub max_buffer_age: Duration, + /// Ceiling on total buffered bytes across all partitions; `emit` flushes + /// inline to stay at or under it whenever the store accepts writes. If a + /// flush fails, or a single record alone exceeds the ceiling (nothing left + /// to flush), the buffer is retained (the WAL is the durability of record) + /// and the ceiling may be transiently exceeded — rather than stalling + /// ingest. (A failed flush attempt is also counted as a flush error.) + pub ceiling_bytes: usize, +} + +/// A failed flush of one partition. Non-fatal — the buffer is retained and the +/// WAL remains the durability of record. Internal: the public `emit` / tick / +/// rotation surface is infallible (errors are swallowed + counted). +#[derive(Debug)] +enum FlushError { + /// Encoding the buffered records to Parquet failed. + Encode(WriterError), + /// Writing the encoded object to the store failed. + Store(StoreError), +} + +impl std::fmt::Display for FlushError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Encode(e) => write!(f, "encode buffered records: {e}"), + Self::Store(e) => write!(f, "put Parquet object: {e}"), + } + } +} + +impl std::error::Error for FlushError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Encode(e) => Some(e), + Self::Store(e) => Some(e), + } + } +} + +struct PartitionBuffer { + records: Vec, + est_bytes: usize, + oldest: Instant, +} + +/// The buffering Parquet record sink — the production replacement for +/// `NoOpRecordSink` (RFC 0014). See the module docs for the flush policy. +pub struct ParquetRecordSink { + store: Store, + config: FlushConfig, + buffers: HashMap, + total_bytes: usize, + flushes: u64, + records_flushed: u64, + flush_errors: u64, + derive_errors: u64, +} + +/// Cheap per-record footprint estimate driving the size trigger + ceiling — a +/// rough heuristic over the larger variable-length fields, not the exact +/// encoded (compressed) size and not every field. Good enough to bound memory +/// and roughly right-size files; precise estimation is RFC 0014 §7. +fn estimate_bytes(r: &MinedRecord) -> usize { + let opt = |o: &Option| o.as_ref().map_or(0, String::len); + // Fixed per-record overhead plus the variable-length payloads. + 96 + opt(&r.body) + + opt(&r.severity_text) + + opt(&r.scope_name) + + opt(&r.scope_version) + + r.params.iter().map(|p| p.value.len() + 8).sum::() + + r.separators.iter().map(String::len).sum::() + // Attributes are encoded as JSON; a flat per-entry estimate suffices. + + (r.attributes.len() + r.resource_attributes.len()) * 48 +} + +/// `/`-delimited object key for a partition's flushed file: the RFC 0005 §3.4 +/// Hive path (relative to the store root) plus a `UUIDv7` name. Mirrors +/// `ourios_parquet::Writer`'s key; object keys are `/`-delimited on every host. +fn object_key(partition: &PartitionKey) -> String { + let rel = partition.data_path(Path::new("")); + format!( + "{}/{}.parquet", + rel.to_string_lossy() + .replace(std::path::MAIN_SEPARATOR, "/"), + Uuid::now_v7() + ) +} + +impl ParquetRecordSink { + /// Create a sink flushing to `store` under `config`. + #[must_use] + pub fn new(store: Store, config: FlushConfig) -> Self { + Self { + store, + config, + buffers: HashMap::new(), + total_bytes: 0, + flushes: 0, + records_flushed: 0, + flush_errors: 0, + derive_errors: 0, + } + } + + /// Total estimated bytes currently buffered across all partitions. + #[must_use] + pub fn buffered_bytes(&self) -> usize { + self.total_bytes + } + + /// Number of partitions with a non-empty buffer. + #[must_use] + pub fn buffered_partitions(&self) -> usize { + self.buffers.len() + } + + /// Count of successful partition flushes. + #[must_use] + pub fn flushes(&self) -> u64 { + self.flushes + } + + /// Total records written out across all successful flushes. + #[must_use] + pub fn records_flushed(&self) -> u64 { + self.records_flushed + } + + /// Records currently buffered (not yet flushed) across all partitions. + #[must_use] + pub fn buffered_records(&self) -> usize { + self.buffers.values().map(|b| b.records.len()).sum() + } + + /// Force-flush every buffered partition — the WAL-segment-rotation trigger + /// (RFC0014.3), including sub-threshold low-volume partitions. + pub fn flush_all(&mut self) { + let keys: Vec = self.buffers.keys().cloned().collect(); + for key in keys { + self.flush_partition_swallow(&key); + } + } + + /// Flush partitions whose oldest record has reached `max_buffer_age` — the + /// age trigger (RFC0014.2), driven by the batch-window tick. + pub fn flush_aged(&mut self) { + let max = self.config.max_buffer_age; + let keys: Vec = self + .buffers + .iter() + .filter(|(_, b)| b.oldest.elapsed() >= max) + .map(|(k, _)| k.clone()) + .collect(); + for key in keys { + self.flush_partition_swallow(&key); + } + } + + /// Encode + put one partition's buffer. On success the buffer is removed + /// and the counters advance; the caller (via [`Self::flush_partition_swallow`]) + /// retains it on error. + fn flush_partition(&mut self, key: &PartitionKey) -> Result<(), FlushError> { + let bytes = match self.buffers.get(key) { + Some(buf) if !buf.records.is_empty() => { + encode_records_to_parquet(&buf.records, DEFAULT_ZSTD_LEVEL) + .map_err(FlushError::Encode)? + } + _ => return Ok(()), + }; + self.store + .put_blocking(&object_key(key), bytes) + .map_err(FlushError::Store)?; + if let Some(buf) = self.buffers.remove(key) { + self.total_bytes = self.total_bytes.saturating_sub(buf.est_bytes); + self.flushes += 1; + self.records_flushed += buf.records.len() as u64; + } + Ok(()) + } + + /// [`Self::flush_partition`] for the infallible `emit` / tick / rotation + /// paths: a failed flush retains the buffer (the WAL is the durability of + /// record) and is counted for observability. + fn flush_partition_swallow(&mut self, key: &PartitionKey) { + if self.flush_partition(key).is_err() { + self.flush_errors += 1; + } + } + + /// Flush the largest buffered partition to reclaim memory. Returns whether + /// a flush actually succeeded (so the ceiling loop stops if the store is + /// unavailable rather than spinning). + fn flush_largest(&mut self) -> bool { + let Some(key) = self + .buffers + .iter() + .filter(|(_, b)| !b.records.is_empty()) + .max_by_key(|(_, b)| b.est_bytes) + .map(|(k, _)| k.clone()) + else { + return false; + }; + if self.flush_partition(&key).is_ok() { + true + } else { + self.flush_errors += 1; + false + } + } +} + +impl RecordSink for ParquetRecordSink { + fn emit(&mut self, record: MinedRecord) { + let Ok(key) = PartitionKey::derive(&record) else { + // Un-partitionable (timestamp overflow, §3.4 fallback exhausted): + // can't route it. The WAL still holds it; count and drop here. + self.derive_errors += 1; + return; + }; + let est = estimate_bytes(&record); + + // Ceiling (RFC0014.4): flush the largest partition inline to make room + // before appending, so buffered bytes stay at or under the ceiling + // whenever the store accepts writes. If a flush fails (store + // unavailable) or nothing more can be flushed (a single oversized + // buffer), the loop stops rather than spinning — the record is still + // retained below (the WAL is the durability of record), and the + // ceiling may be transiently exceeded (counted via `flush_errors`) + // instead of deadlocking the ingest path. + while self.total_bytes.saturating_add(est) > self.config.ceiling_bytes + && self.flush_largest() + {} + + let buf = self + .buffers + .entry(key.clone()) + .or_insert_with(|| PartitionBuffer { + records: Vec::new(), + est_bytes: 0, + oldest: Instant::now(), + }); + buf.records.push(record); + // Saturating (matching `saturating_sub` on flush) so the byte counters + // stay monotonic and the triggers can't be corrupted by wraparound + // under prolonged retention (e.g. a store outage). + buf.est_bytes = buf.est_bytes.saturating_add(est); + let over_target = buf.est_bytes >= self.config.target_bytes; + self.total_bytes = self.total_bytes.saturating_add(est); + + // Size trigger (RFC0014.1): the emit that crosses the target flushes. + if over_target { + self.flush_partition_swallow(&key); + } + } +} diff --git a/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs b/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs index d238f0a7f..fd7620969 100644 --- a/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs +++ b/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs @@ -1,36 +1,155 @@ //! RFC 0014 — ingest write path (record sink + flush policy) acceptance //! scenarios (§5). //! -//! **Status: `red`.** These are the failing stubs that drive the `green` -//! implementation: each encodes one RFC 0014 §5 scenario and currently -//! `todo!()`s. They are `#[ignore]`d so the default `cargo test` (and CI) -//! stays green while the buffering `ParquetRecordSink` is built — `green` -//! replaces each body with a real assertion against the sink (per-partition -//! buffers; hybrid size + age + WAL-rotation flush; hard byte-ceiling with -//! blocking backpressure) and removes the `#[ignore]`. -//! -//! Placement may shift at `green`: RFC0014.5 (crash recovery) extends the RFC -//! 0008 WAL harness here in `ourios-ingester`; the buffer-trigger scenarios -//! (.1–.4, .6) may move next to the sink wherever it lands. +//! `.1`–`.4`/`.6` drive the [`ParquetRecordSink`] directly against a +//! `LocalFileSystem`-backed [`Store`] (synthetic `MinedRecord` streams); +//! reading the flushed objects back proves no loss + tenant isolation. `.5` +//! (crash recovery) stays `#[ignore]`d until the sink is wired into the ingest +//! pipeline and the RFC 0008 WAL crash harness is extended (`green`, part 2). //! //! See `docs/rfcs/0014-ingest-write-path.md` §5/§6. +use std::path::Path; +use std::time::Duration; + +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::{Reader, Store}; + +/// A clean-round-trip record for `tenant` at in-hour offset `i`. +fn rec_for(tenant: &str, i: u64) -> MinedRecord { + MinedRecord { + tenant_id: TenantId::new(tenant), + template_id: 1, + template_version: 1, + severity_number: 9, + severity_text: Some("INFO".to_string()), + scope_name: Some("lib.cart".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, + } +} + +/// Every flushed `*.parquet` object under `root`, one inner `Vec` per file +/// (so per-file tenant isolation can be asserted). +fn parquet_files(root: &Path) -> Vec> { + let mut files = Vec::new(); + let mut stack = vec![root.to_path_buf()]; + while let Some(dir) = stack.pop() { + let Ok(entries) = std::fs::read_dir(&dir) else { + continue; + }; + for entry in entries.flatten() { + let path = entry.path(); + if path.is_dir() { + stack.push(path); + } else if path.extension().is_some_and(|x| x == "parquet") { + let rows = Reader::open_file(&path) + .expect("open_file") + .read_all() + .expect("read_all"); + files.push(rows); + } + } + } + files +} + +fn all_rows(root: &Path) -> Vec { + parquet_files(root).into_iter().flatten().collect() +} + +fn sink(dir: &Path, config: FlushConfig) -> ParquetRecordSink { + ParquetRecordSink::new(Store::local(dir).expect("local store"), config) +} + +/// A "never trigger" sentinel for the size/ceiling knobs (pointer-width safe). +const HUGE: usize = usize::MAX; +const FOREVER: Duration = Duration::from_secs(86_400); + /// Scenario RFC0014.1 — Size trigger: the emit that crosses the size target /// flushes the partition to one right-sized Parquet object. /// See `docs/rfcs/0014-ingest-write-path.md` §5. #[test] -#[ignore = "RFC0014.1 — red until the ParquetRecordSink + flush policy land (green)"] fn rfc0014_1_size_trigger() { - todo!("RFC0014.1: a partition flushes on the emit that crosses the size target") + let dir = tempfile::TempDir::new().expect("temp dir"); + let mut s = sink( + dir.path(), + FlushConfig { + target_bytes: 500, + max_buffer_age: FOREVER, + ceiling_bytes: HUGE, + }, + ); + let records: Vec = (0..50).map(|i| rec_for("tenant-a", i)).collect(); + for r in &records { + s.emit(r.clone()); + } + // The size trigger alone (no age/rotation flush called) has fired. + assert!(s.flushes() >= 1, "size trigger flushed mid-stream"); + // Drain the buffered tail, then confirm no loss across the flushed objects. + s.flush_all(); + let mut got = all_rows(dir.path()); + assert_eq!( + got.len(), + records.len(), + "every record published, none lost" + ); + got.sort_by_key(|r| { + r.params[0] + .value + .parse::() + .expect("param round-trips as a parseable number") + }); + assert_eq!(got, records, "rows recovered byte-for-byte"); } /// Scenario RFC0014.2 — Age trigger: a sub-target low-volume partition flushes /// on the next batch-window tick once its oldest record reaches `max_buffer_age`. /// See `docs/rfcs/0014-ingest-write-path.md` §5. #[test] -#[ignore = "RFC0014.2 — red until the ParquetRecordSink + flush policy land (green)"] fn rfc0014_2_age_trigger() { - todo!("RFC0014.2: low-volume partition flushes on age") + let dir = tempfile::TempDir::new().expect("temp dir"); + // Size never triggers; age is inclusive at zero, so any buffered record is + // immediately "aged" — `flush_aged` (the tick) is the only thing that flushes. + let mut s = sink( + dir.path(), + FlushConfig { + target_bytes: HUGE, + max_buffer_age: Duration::ZERO, + ceiling_bytes: HUGE, + }, + ); + for i in 0..5 { + s.emit(rec_for("tenant-a", i)); + } + assert_eq!( + s.flushes(), + 0, + "size/ceiling did not flush a low-volume partition" + ); + s.flush_aged(); + assert_eq!(s.flushes(), 1, "the age sweep flushed the partition"); + assert_eq!(all_rows(dir.path()).len(), 5); } /// Scenario RFC0014.3 — Rotation force-flush: a WAL segment rotation flushes @@ -38,18 +157,69 @@ fn rfc0014_2_age_trigger() { /// the sealed segment. /// See `docs/rfcs/0014-ingest-write-path.md` §5. #[test] -#[ignore = "RFC0014.3 — red until the ParquetRecordSink + flush policy land (green)"] fn rfc0014_3_rotation_force_flush() { - todo!("RFC0014.3: rotation flushes every partition") + let dir = tempfile::TempDir::new().expect("temp dir"); + let mut s = sink( + dir.path(), + FlushConfig { + target_bytes: HUGE, + max_buffer_age: FOREVER, + ceiling_bytes: HUGE, + }, + ); + // Three tenants → three partitions, all below every other trigger. + for t in ["tenant-x", "tenant-y", "tenant-z"] { + for i in 0..3 { + s.emit(rec_for(t, i)); + } + } + assert_eq!(s.flushes(), 0, "nothing flushed before rotation"); + assert_eq!(s.buffered_partitions(), 3); + + s.flush_all(); // the WAL-segment-rotation trigger + + assert_eq!(s.flushes(), 3, "every partition flushed on rotation"); + assert_eq!( + s.buffered_partitions(), + 0, + "no buffered record predates the seal" + ); + assert_eq!(all_rows(dir.path()).len(), 9); } -/// Scenario RFC0014.4 — Bounded memory: the sink early-flushes under pressure -/// and, at the hard ceiling, `emit` blocks so buffered bytes never exceed it. +/// Scenario RFC0014.4 — Bounded memory: the sink flushes inline so buffered +/// bytes never exceed the hard ceiling; nothing is lost. /// See `docs/rfcs/0014-ingest-write-path.md` §5. #[test] -#[ignore = "RFC0014.4 — red until the ParquetRecordSink + flush policy land (green)"] fn rfc0014_4_bounded_memory() { - todo!("RFC0014.4: hard ceiling, never exceeded") + let dir = tempfile::TempDir::new().expect("temp dir"); + // Size/age never trigger; only the ceiling does. 100 records far exceed a + // 1 KiB ceiling, so the sink must flush inline to stay bounded. + let ceiling = 1024; + let mut s = sink( + dir.path(), + FlushConfig { + target_bytes: HUGE, + max_buffer_age: FOREVER, + ceiling_bytes: ceiling, + }, + ); + let n = 100; + for i in 0..n { + s.emit(rec_for("tenant-a", i)); + assert!( + s.buffered_bytes() <= ceiling, + "ceiling held after each emit: {} <= {ceiling}", + s.buffered_bytes(), + ); + } + assert!(s.flushes() >= 1, "the ceiling forced inline flushes"); + s.flush_all(); + assert_eq!( + all_rows(dir.path()).len() as u64, + n, + "no loss under backpressure" + ); } /// Scenario RFC0014.5 — No acknowledged-data loss: a crash with a non-empty @@ -57,7 +227,7 @@ fn rfc0014_4_bounded_memory() { /// record (`CLAUDE.md` §3.4). /// See `docs/rfcs/0014-ingest-write-path.md` §5. #[test] -#[ignore = "RFC0014.5 — red until the ParquetRecordSink + flush policy land (green)"] +#[ignore = "RFC0014.5 — green part 2: wire the sink into the ingest pipeline + extend the RFC 0008 crash harness"] fn rfc0014_5_no_acknowledged_data_loss() { todo!("RFC0014.5: crash mid-buffer loses no acknowledged data (WAL replay)") } @@ -66,7 +236,49 @@ fn rfc0014_5_no_acknowledged_data_loss() { /// only one tenant's rows; no buffer or flush crosses tenants (`CLAUDE.md` §3.7). /// See `docs/rfcs/0014-ingest-write-path.md` §5. #[test] -#[ignore = "RFC0014.6 — red until the ParquetRecordSink + flush policy land (green)"] fn rfc0014_6_tenant_isolation() { - todo!("RFC0014.6: no cross-tenant buffer or flush") + let dir = tempfile::TempDir::new().expect("temp dir"); + let mut s = sink( + dir.path(), + FlushConfig { + target_bytes: HUGE, + max_buffer_age: FOREVER, + ceiling_bytes: HUGE, + }, + ); + for i in 0..10 { + s.emit(rec_for("tenant-x", i)); + s.emit(rec_for("tenant-y", i)); + } + s.flush_all(); + + let files = parquet_files(dir.path()); + assert!(!files.is_empty(), "objects were published"); + for file in &files { + let tenants: std::collections::BTreeSet<&str> = + file.iter().map(|r| r.tenant_id.as_str()).collect(); + assert_eq!( + tenants.len(), + 1, + "each object holds exactly one tenant: {tenants:?}" + ); + } + let rows: Vec<&MinedRecord> = files.iter().flatten().collect(); + let tenants: std::collections::BTreeSet<&str> = + rows.iter().map(|r| r.tenant_id.as_str()).collect(); + assert_eq!( + tenants, + ["tenant-x", "tenant-y"].into_iter().collect(), + "exactly the two emitted tenants, no extras", + ); + assert_eq!(rows.len(), 20, "no extra rows beyond the 20 emitted"); + let x = rows + .iter() + .filter(|r| r.tenant_id.as_str() == "tenant-x") + .count(); + let y = rows + .iter() + .filter(|r| r.tenant_id.as_str() == "tenant-y") + .count(); + assert_eq!((x, y), (10, 10), "both tenants' rows present, unmixed"); }