diff --git a/Cargo.lock b/Cargo.lock index 763e7b3..b554db2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5492,6 +5492,8 @@ dependencies = [ "lance-index 7.0.0", "lance-namespace 7.0.0", "lancedb", + "metrics", + "metrics-util", "serde", "serde_json", "tempfile", @@ -6902,7 +6904,9 @@ dependencies = [ "crossbeam-epoch", "crossbeam-utils", "hashbrown 0.15.5", + "indexmap 2.14.0", "metrics", + "ordered-float 4.6.0", "quanta", "rand 0.9.4", "rand_xoshiro", @@ -7729,6 +7733,15 @@ dependencies = [ "num-traits", ] +[[package]] +name = "ordered-float" +version = "4.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7bb71e1b3fa6ca1c61f383464aaf2bb0e2f8e772a1f01d486832464de363b951" +dependencies = [ + "num-traits", +] + [[package]] name = "ordered-float" version = "5.3.0" diff --git a/crates/lance-context-core/Cargo.toml b/crates/lance-context-core/Cargo.toml index f7000d8..a49e5a7 100644 --- a/crates/lance-context-core/Cargo.toml +++ b/crates/lance-context-core/Cargo.toml @@ -10,6 +10,13 @@ description = "Multimodal, versioned context storage for agentic workflows" keywords = ["context", "multimodal", "lance", "agents", "storage"] categories = ["database", "data-structures", "science"] +[features] +default = ["metrics"] +# Emit Prometheus metrics for store operations (add/flush/WAL merge latency). +# Optional so downstream consumers embedding this library are not forced to take +# the `metrics` dependency; call sites compile to nothing when disabled. +metrics = ["dep:metrics"] + [dependencies] arrow-array = "58" arrow-ipc = "58" @@ -23,6 +30,9 @@ lance-index = "7.0.0" lance-namespace = "7.0.0" lancedb = "0.30.0" lance-graph = "0.5.4" +# Version-matched with lance-context-server/-master so one process-wide recorder +# serves every crate. +metrics = { version = "0.24", optional = true } serde = { version = "1", features = ["derive"] } serde_json = "1" futures = "0.3" @@ -31,5 +41,8 @@ tracing = "0.1" uuid = { version = "1.20.0", features = ["v4", "v5", "v7"] } [dev-dependencies] +# Snapshotting recorder so tests can assert which metric series an operation +# emitted, without installing a process-global Prometheus exporter. +metrics-util = { version = "0.19", default-features = false, features = ["debugging"] } tempfile = "3" tokio = { version = "1", features = ["rt-multi-thread"] } diff --git a/crates/lance-context-core/src/lib.rs b/crates/lance-context-core/src/lib.rs index 904e632..838e7ce 100644 --- a/crates/lance-context-core/src/lib.rs +++ b/crates/lance-context-core/src/lib.rs @@ -8,6 +8,7 @@ mod datagen_store; mod eval; mod export; mod id; +pub mod metrics; mod namespace; mod record; mod registry; diff --git a/crates/lance-context-core/src/metrics.rs b/crates/lance-context-core/src/metrics.rs new file mode 100644 index 0000000..332f6ca --- /dev/null +++ b/crates/lance-context-core/src/metrics.rs @@ -0,0 +1,147 @@ +//! Feature-gated metrics shim. +//! +//! Call sites use [`observe_duration!`] and [`count!`] unconditionally; when the +//! `metrics` feature is off these expand to nothing (the timing `Instant` is not +//! even taken), so a consumer embedding this crate pays zero cost and takes no +//! dependency. +//! +//! Metric names and label conventions live here so they cannot drift between the +//! emission site and the bucket configuration in `lance-context-metrics`. +//! +//! # Cardinality +//! +//! Every label combination times every histogram bucket is a separate exported +//! series — and in Datadog, a separately-billed custom metric. Two rules keep +//! that bounded: +//! +//! 1. **No unbounded labels.** Never a dataset URI, store name, experiment, or +//! shard id. Those belong in a tracing span, which is queryable without +//! multiplying series. +//! 2. **Failures are counted, not timed.** A `result="error"` label doubles a +//! histogram's series count to describe the *latency distribution of a rare +//! event*, which is almost never actionable. The actionable signal is the +//! rate, so errors get a flat counter (1 series) and histograms measure the +//! success path only. + +/// Latency of one [`crate::RolloutStore::add`] — the durable WAL append only. +/// Unlabelled: failures are counted by [`ROLLOUT_ADD_ERRORS`] instead. +pub const ROLLOUT_ADD_DURATION: &str = "rollout_add_duration_seconds"; + +/// Failed [`crate::RolloutStore::add`] calls. +pub const ROLLOUT_ADD_ERRORS: &str = "rollout_add_errors_total"; + +/// Latency of one [`crate::RolloutStore::flush`] — sealing the memtable so +/// previously added rows become readable. +/// +/// Label `outcome`: +/// - `sealed` — a memtable was actually sealed and drained (real work) +/// - `noop` — no resident writer, returned immediately (the common case) +/// - `fenced` — the epoch was superseded by a merge; nothing to flush +/// +/// `outcome` is kept despite the cardinality cost because the three paths differ +/// by orders of magnitude: without it the distribution is dominated by near-zero +/// `noop` samples and its high percentiles say nothing about real flush cost. +pub const ROLLOUT_FLUSH_DURATION: &str = "rollout_flush_duration_seconds"; + +/// Failed [`crate::RolloutStore::flush`] calls. +pub const ROLLOUT_FLUSH_ERRORS: &str = "rollout_flush_errors_total"; + +/// Per-phase latency of a WAL self-merge. Label `phase`: +/// `seal` | `read` | `append` | `claim_epoch` | `drain` | `delete`. +pub const ROLLOUT_WAL_MERGE_DURATION: &str = "rollout_wal_merge_duration_seconds"; + +/// Failed WAL self-merge phases, labelled by the `phase` that failed. A merge +/// aborts on the first failing phase, so this also identifies where it died. +pub const ROLLOUT_WAL_MERGE_ERRORS: &str = "rollout_wal_merge_errors_total"; + +/// Emit a histogram sample in seconds. No-op without the `metrics` feature. +#[cfg(feature = "metrics")] +macro_rules! observe_duration { + ($name:expr, $elapsed:expr $(, $k:expr => $v:expr)* $(,)?) => { + ::metrics::histogram!($name $(, $k => $v)*).record($elapsed.as_secs_f64()) + }; +} + +#[cfg(not(feature = "metrics"))] +macro_rules! observe_duration { + ($name:expr, $elapsed:expr $(, $k:expr => $v:expr)* $(,)?) => {{ + let _ = &$elapsed; + }}; +} + +/// Increment a counter by one. No-op without the `metrics` feature. +/// +/// The disabled arm still expands to a unit *expression* rather than an empty +/// block, so a `match` whose arms are `observe_duration!`/`count!` keeps both +/// arms inhabited and does not collapse into a clippy `single_match` warning +/// when the feature is off. +#[cfg(feature = "metrics")] +macro_rules! count { + ($name:expr $(, $k:expr => $v:expr)* $(,)?) => { + ::metrics::counter!($name $(, $k => $v)*).increment(1) + }; +} + +#[cfg(not(feature = "metrics"))] +macro_rules! count { + ($name:expr $(, $k:expr => $v:expr)* $(,)?) => {{ + let _ = $name; + }}; +} + +/// Start a timer, or evaluate to `()` when metrics are compiled out. +#[cfg(feature = "metrics")] +macro_rules! timer_start { + () => { + std::time::Instant::now() + }; +} + +#[cfg(not(feature = "metrics"))] +macro_rules! timer_start { + () => { + () + }; +} + +/// Elapsed time since a [`timer_start!`], or a zero duration when compiled out. +#[cfg(feature = "metrics")] +macro_rules! timer_elapsed { + ($t:expr) => { + $t.elapsed() + }; +} + +#[cfg(not(feature = "metrics"))] +macro_rules! timer_elapsed { + ($t:expr) => {{ + let _ = &$t; + std::time::Duration::ZERO + }}; +} + +/// Time one WAL-merge phase: record its duration on success, or increment the +/// phase-labelled error counter on failure. Evaluates to the wrapped `Result`. +/// +/// Keeps the success/failure split identical across all six phases, which is +/// what stops `result` creeping back onto the histogram. +macro_rules! observe_phase { + ($phase:expr, $body:expr) => {{ + let __start = $crate::metrics::timer_start!(); + let __result = $body; + match &__result { + Ok(_) => $crate::metrics::observe_duration!( + $crate::metrics::ROLLOUT_WAL_MERGE_DURATION, + $crate::metrics::timer_elapsed!(__start), + "phase" => $phase, + ), + Err(_) => $crate::metrics::count!( + $crate::metrics::ROLLOUT_WAL_MERGE_ERRORS, + "phase" => $phase, + ), + } + __result + }}; +} + +pub(crate) use {count, observe_duration, observe_phase, timer_elapsed, timer_start}; diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs index c164310..a2a5d0c 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -83,6 +83,7 @@ use serde_json::Value; use tracing::{info, warn}; use uuid::Uuid; +use crate::metrics::{count, observe_duration, observe_phase, timer_elapsed, timer_start}; use crate::rollout::RolloutRecord; use crate::store::{ column_as, column_as_optional, relationship_field, relationship_list_item_field, @@ -90,6 +91,17 @@ use crate::store::{ CompactionStats, RELATIONSHIPS_COLUMN, }; +/// What a [`RolloutStore::flush`] actually did, for the `outcome` metric label. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum FlushOutcome { + /// A memtable was sealed and drained into a flushed generation. + Sealed, + /// No resident writer, so nothing to seal (the common case). + Noop, + /// The writer's epoch was superseded by a merge; nothing to flush. + Fenced, +} + /// Number of shard manifest files to scan per batch when discovering the latest /// shard state (mirrors the constant used by `ContextStore`). const DEFAULT_MANIFEST_SCAN_BATCH_SIZE: usize = 16; @@ -582,7 +594,22 @@ impl RolloutStore { if records.is_empty() { return Ok(self.dataset.manifest.version); } + // Success-path latency only. A failed append's *duration* is not + // actionable, but its *rate* is, so errors increment a flat counter + // rather than doubling this histogram's series count. + let started = timer_start!(); + let result = self.add_inner(records).await; + match &result { + Ok(_) => observe_duration!( + crate::metrics::ROLLOUT_ADD_DURATION, + timer_elapsed!(started), + ), + Err(_) => count!(crate::metrics::ROLLOUT_ADD_ERRORS), + } + result + } + async fn add_inner(&self, records: &[RolloutRecord]) -> LanceResult { let batch = self.records_to_batch(records)?; // Durable append only: `put` waits for the WAL entry to be PUT to object @@ -657,12 +684,38 @@ impl RolloutStore { /// append in [`Self::add`] and driven periodically (see the server's global /// sweeper). A no-op when no writer is resident or nothing is buffered. pub async fn flush(&self) -> LanceResult<()> { + let started = timer_start!(); + let result = self.flush_inner().await; + // `outcome` is worth its cardinality: the three paths differ by orders of + // magnitude, and `noop` (no resident writer) is by far the most common, + // so a blended histogram would be dominated by near-zero samples. + // Failures are counted, not timed. + match &result { + Ok(outcome) => { + let label = match outcome { + FlushOutcome::Sealed => "sealed", + FlushOutcome::Noop => "noop", + FlushOutcome::Fenced => "fenced", + }; + let _ = label; + observe_duration!( + crate::metrics::ROLLOUT_FLUSH_DURATION, + timer_elapsed!(started), + "outcome" => label, + ); + } + Err(_) => count!(crate::metrics::ROLLOUT_FLUSH_ERRORS), + } + result.map(|_| ()) + } + + async fn flush_inner(&self) -> LanceResult { let writer = { let guard = self.write_writer.lock().await; guard.as_ref().cloned() }; let Some(writer) = writer else { - return Ok(()); + return Ok(FlushOutcome::Noop); }; match writer.force_seal_active().await { Ok(()) => {} @@ -670,11 +723,12 @@ impl RolloutStore { // and replays. Nothing to flush against the dead epoch. Err(err) if is_fenced_error(&err) => { self.invalidate_writer(&writer).await; - return Ok(()); + return Ok(FlushOutcome::Fenced); } Err(err) => return Err(err), } - writer.wait_for_flush_drain().await + writer.wait_for_flush_drain().await?; + Ok(FlushOutcome::Sealed) } /// Gracefully close the resident writer, draining its background tasks. @@ -831,60 +885,22 @@ impl RolloutStore { // this shard — including our own resident writer. Close it (draining its // background tasks; `ShardWriter` has no `Drop`) and clear it so the next // `add` transparently reopens against the freshly-claimed epoch. - self.close().await?; + observe_phase!("seal", self.close().await)?; self.ensure_latest_rollout_schema().await?; // Resolve each flushed generation to its absolute dataset path and read // all its rows into memory. Record which generation ids we merge so the // drain can remove exactly these and nothing else. - let base_uri = self.dataset.uri().trim_end_matches('/').to_string(); - let mut merged_generations: HashSet = HashSet::new(); - // Remember each merged generation's on-storage folder name so we can - // delete the blob directory after the manifest drain (see below). - let mut merged_paths: Vec = Vec::new(); - let mut batches: Vec = Vec::new(); - let merge_schema: Arc = Arc::new(self.dataset.schema().into()); - for flushed in &manifest.flushed_generations { - let gen_uri = format!( - "{}/_mem_wal/{}/{}", - base_uri, self.write_shard, flushed.path - ); - let gen_dataset = Self::load_with_options( - &gen_uri, - self.storage_options.clone(), - self.session.clone(), - ) - .await?; - let mut stream = gen_dataset.scan().try_into_stream().await?; - while let Some(batch) = stream.try_next().await? { - if batch.num_rows() > 0 { - batches.push(align_batch_to_schema(batch, merge_schema.clone())?); - } - } - merged_generations.insert(flushed.generation); - merged_paths.push(flushed.path.clone()); - } + let (merged_generations, merged_paths, batches, merge_schema) = + observe_phase!("read", self.read_flushed_generations(manifest).await)?; // Append the merged rows to the base table. if !batches.is_empty() { - let reader = RecordBatchIterator::new( - batches.into_iter().map(Ok::), - merge_schema, - ); - let mut params = WriteParams { - mode: WriteMode::Append, - ..Default::default() - }; - if let Some(options) = &self.storage_options { - params.store_params = Some(ObjectStoreParams { - storage_options_accessor: Some(Arc::new( - StorageOptionsAccessor::with_static_options(options.clone()), - )), - ..Default::default() - }); - } - self.dataset.append(reader, Some(params)).await?; + observe_phase!( + "append", + self.append_merged_batches(batches, merge_schema).await + )?; } // Drain the merged generations from the shard manifest. Claim the @@ -893,19 +909,26 @@ impl RolloutStore { // Removing only the merged ids (rather than clearing the vec) is what // makes this safe against a generation that lands after we read the // manifest: it is preserved for the next merge instead of being dropped. - let (epoch, _) = manifest_store.claim_epoch(manifest.shard_spec_id).await?; - manifest_store - .commit_update(epoch, |current| ShardManifest { - version: current.version + 1, - flushed_generations: current - .flushed_generations - .iter() - .filter(|fg| !merged_generations.contains(&fg.generation)) - .cloned() - .collect(), - ..current.clone() - }) - .await?; + let (epoch, _) = observe_phase!( + "claim_epoch", + manifest_store.claim_epoch(manifest.shard_spec_id).await + )?; + + observe_phase!( + "drain", + manifest_store + .commit_update(epoch, |current| ShardManifest { + version: current.version + 1, + flushed_generations: current + .flushed_generations + .iter() + .filter(|fg| !merged_generations.contains(&fg.generation)) + .cloned() + .collect(), + ..current.clone() + }) + .await + )?; // Delete the merged generations' blob directories now that no manifest // references them. Ordering matters: the drain above already removed @@ -922,6 +945,7 @@ impl RolloutStore { // Skipping this deletion is exactly the historical storage leak: every // merged generation left its `_mem_wal/{shard}/{gen}/` directory behind // forever. + let phase = timer_start!(); let object_store = self.dataset.object_store(None).await?; let branch_path = self.dataset.branch_location().path.clone(); for path in &merged_paths { @@ -940,7 +964,79 @@ impl RolloutStore { ); } } + // Best-effort by contract: a delete failure is logged above and does not + // fail the merge, so there is no error counter for this phase. + observe_duration!( + crate::metrics::ROLLOUT_WAL_MERGE_DURATION, + timer_elapsed!(phase), + "phase" => "delete", + ); + + Ok(()) + } + + /// Read every flushed generation listed in `manifest` into memory, aligned + /// to the base table's current schema. + /// + /// Returns the merged generation ids, their on-storage folder names (needed + /// to delete the directories after the manifest drain), the batches, and the + /// schema they were aligned to. + #[allow(clippy::type_complexity)] + async fn read_flushed_generations( + &self, + manifest: &ShardManifest, + ) -> LanceResult<(HashSet, Vec, Vec, Arc)> { + let base_uri = self.dataset.uri().trim_end_matches('/').to_string(); + let mut merged_generations: HashSet = HashSet::new(); + let mut merged_paths: Vec = Vec::new(); + let mut batches: Vec = Vec::new(); + let merge_schema: Arc = Arc::new(self.dataset.schema().into()); + for flushed in &manifest.flushed_generations { + let gen_uri = format!( + "{}/_mem_wal/{}/{}", + base_uri, self.write_shard, flushed.path + ); + let gen_dataset = Self::load_with_options( + &gen_uri, + self.storage_options.clone(), + self.session.clone(), + ) + .await?; + let mut stream = gen_dataset.scan().try_into_stream().await?; + while let Some(batch) = stream.try_next().await? { + if batch.num_rows() > 0 { + batches.push(align_batch_to_schema(batch, merge_schema.clone())?); + } + } + merged_generations.insert(flushed.generation); + merged_paths.push(flushed.path.clone()); + } + Ok((merged_generations, merged_paths, batches, merge_schema)) + } + /// Append merged WAL rows into the base table with this store's credentials. + async fn append_merged_batches( + &mut self, + batches: Vec, + merge_schema: Arc, + ) -> LanceResult<()> { + let reader = RecordBatchIterator::new( + batches.into_iter().map(Ok::), + merge_schema, + ); + let mut params = WriteParams { + mode: WriteMode::Append, + ..Default::default() + }; + if let Some(options) = &self.storage_options { + params.store_params = Some(ObjectStoreParams { + storage_options_accessor: Some(Arc::new( + StorageOptionsAccessor::with_static_options(options.clone()), + )), + ..Default::default() + }); + } + self.dataset.append(reader, Some(params)).await?; Ok(()) } @@ -5011,4 +5107,157 @@ mod tests { assert!(!result.truncated); }); } + + /// `add` and `flush` must land on separate histograms, and `flush` must + /// report which of its three paths it took. + /// + /// This is the core of "split the latency": at the HTTP layer these two are + /// one route (flush is a query param), so if they are not distinguished here + /// they cannot be distinguished anywhere. + #[test] + fn add_and_flush_emit_distinct_labelled_histograms() { + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + use metrics_util::MetricKind; + + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + + let dir = TempDir::new().unwrap(); + let uri = dir + .path() + .join("rollouts.lance") + .to_string_lossy() + .into_owned(); + let runtime = tokio::runtime::Runtime::new().unwrap(); + + metrics::with_local_recorder(&recorder, || { + runtime.block_on(async { + let store = RolloutStore::open(&uri).await.unwrap(); + // A flush with no resident writer: the `noop` fast path. + store.flush().await.unwrap(); + store.add(&[assistant_record("row-0")]).await.unwrap(); + // Now there is a writer with buffered rows: the `sealed` path. + store.flush().await.unwrap(); + }); + }); + + let mut add_samples = 0usize; + let mut flush_outcomes: Vec = Vec::new(); + for (key, _, _, value) in snapshotter.snapshot().into_vec() { + if key.kind() != MetricKind::Histogram { + continue; + } + let name = key.key().name().to_string(); + let labels: Vec<(String, String)> = key + .key() + .labels() + .map(|l| (l.key().to_string(), l.value().to_string())) + .collect(); + let count = match value { + DebugValue::Histogram(v) => v.len(), + _ => 0, + }; + if name == crate::metrics::ROLLOUT_ADD_DURATION { + add_samples += count; + // No `result` label: an error label would double this + // histogram's series count (one per bucket, twice) to describe + // the latency of a rare event. Errors are counted instead. + assert!( + labels.is_empty(), + "add latency must stay unlabelled to bound cardinality; got {labels:?}" + ); + } else if name == crate::metrics::ROLLOUT_FLUSH_DURATION { + assert_eq!( + labels.len(), + 1, + "flush should carry exactly `outcome`, no result label; got {labels:?}" + ); + let outcome = labels + .iter() + .find(|(k, _)| k == "outcome") + .map(|(_, v)| v.clone()) + .expect("flush must carry an outcome label"); + for _ in 0..count { + flush_outcomes.push(outcome.clone()); + } + } + } + + assert_eq!(add_samples, 1, "one add should record exactly one sample"); + flush_outcomes.sort(); + // The two flushes took genuinely different paths; collapsing them into + // one series is what makes the flush histogram unreadable in production, + // since `noop` is by far the common case and is near-zero. + assert_eq!( + flush_outcomes, + vec!["noop".to_string(), "sealed".to_string()], + "flush should distinguish the no-op fast path from real sealing" + ); + } + + /// Guards the cardinality contract: latency histograms must never carry a + /// label whose domain is unbounded (a store URI, shard id, experiment name) + /// or redundant with a counter (`result`). + /// + /// Every label combination times every bucket is a separate exported series, + /// and in Datadog a separately-billed custom metric, so this is a cost + /// regression test as much as a correctness one. + #[test] + fn latency_histograms_carry_only_bounded_labels() { + use metrics_util::debugging::DebuggingRecorder; + use metrics_util::MetricKind; + + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let dir = TempDir::new().unwrap(); + let uri = dir + .path() + .join("rollouts.lance") + .to_string_lossy() + .into_owned(); + let runtime = tokio::runtime::Runtime::new().unwrap(); + + metrics::with_local_recorder(&recorder, || { + runtime.block_on(async { + let store = RolloutStore::open(&uri).await.unwrap(); + store.add(&[assistant_record("row-0")]).await.unwrap(); + store.flush().await.unwrap(); + }); + }); + + // Closed sets only. `result` is intentionally absent: it belongs on a + // counter, where it costs one series instead of one per bucket. + let allowed: &[(&str, &[&str])] = &[ + ("outcome", &["sealed", "noop", "fenced"]), + ( + "phase", + &["seal", "read", "append", "claim_epoch", "drain", "delete"], + ), + ]; + + for (key, _, _, _) in snapshotter.snapshot().into_vec() { + if key.kind() != MetricKind::Histogram { + continue; + } + for label in key.key().labels() { + let (k, v) = (label.key(), label.value()); + let allowed_values = allowed + .iter() + .find(|(name, _)| *name == k) + .map(|(_, vs)| *vs) + .unwrap_or_else(|| { + panic!( + "histogram {} carries unexpected label `{k}`; latency labels must \ + come from a closed, documented set", + key.key().name() + ) + }); + assert!( + allowed_values.contains(&v), + "histogram {} label {k}={v} is outside its documented domain {allowed_values:?}", + key.key().name() + ); + } + } + } } diff --git a/crates/lance-context-master/src/scanner.rs b/crates/lance-context-master/src/scanner.rs index cf40f6a..64f93d6 100644 --- a/crates/lance-context-master/src/scanner.rs +++ b/crates/lance-context-master/src/scanner.rs @@ -149,6 +149,17 @@ async fn scan_once_inner(state: &Arc) -> lance::Result { } metrics::histogram!("master_scan_duration_seconds").record(scan_start.elapsed().as_secs_f64()); + // Named without a `_total` suffix: these are gauges (current value), and + // `_total` is the counter convention. Datadog's OpenMetrics check infers + // type from the name, so `_total` on a gauge was being ingested as a + // monotonic count -- graphing "experiments created per second" for a metric + // whose meaning is "how many exist right now". `rate()` was equally + // meaningless in Prometheus. The `_total` names are still emitted alongside, + // deprecated, so existing dashboards keep working. + metrics::gauge!("master_experiments").set(live_count as f64); + metrics::gauge!("master_rollout_rows").set(total_rows as f64); + metrics::gauge!("master_rollout_fragments").set(total_fragments as f64); + // Deprecated aliases -- remove once dashboards have migrated. metrics::gauge!("master_experiments_total").set(live_count as f64); metrics::gauge!("master_rollout_rows_total").set(total_rows as f64); metrics::gauge!("master_rollout_fragments_total").set(total_fragments as f64); diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index a6e6bc9..ba08f15 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -83,26 +83,60 @@ pub async fn enqueue_with_deps( Ok(record) } +/// Time spent getting a task to the point where its work can start. +#[derive(Debug, Clone, Copy, Default)] +struct TaskClaimTiming { + /// The etcd claim transaction (queued→running, lease, target lock). + claim: std::time::Duration, + /// Waiting for a concurrency permit once the task was already claimed. + permit_wait: std::time::Duration, +} + /// Execute one claimed task and atomically publish its terminal state. -async fn run_task(state: &Arc, claim: TaskClaim) { +/// +/// `timing` carries how long the dispatch loop spent claiming this task and +/// waiting for a concurrency permit, so every phase of the task's life lands on +/// one metric rather than only the work window. +async fn run_task(state: &Arc, claim: TaskClaim, timing: TaskClaimTiming) { let task = claim.task.clone(); + let kind = kind_label(task.kind); + + metrics::histogram!("master_task_phase_duration_seconds", "kind" => kind, "phase" => "claim") + .record(timing.claim.as_secs_f64()); + metrics::histogram!( + "master_task_phase_duration_seconds", + "kind" => kind, + "phase" => "permit_wait", + ) + .record(timing.permit_wait.as_secs_f64()); + let started = std::time::Instant::now(); let outcome = match task.kind { TaskKind::Compact => run_compaction(state, &task.target).await, TaskKind::MergeWal => run_merge_wal(state, &task.target).await, TaskKind::IndexId => run_index_id(state, &task.target).await, }; - - metrics::histogram!("master_task_duration_seconds", "kind" => kind_label(task.kind)) - .record(started.elapsed().as_secs_f64()); + let work_elapsed = started.elapsed(); let result = if outcome.is_ok() { "success" } else { "failed" }; - metrics::counter!("master_tasks_total", "kind" => kind_label(task.kind), "result" => result) - .increment(1); + + metrics::histogram!("master_task_phase_duration_seconds", "kind" => kind, "phase" => "work") + .record(work_elapsed.as_secs_f64()); + // Same scope as the `work` phase above, kept for back-compat with existing + // dashboards. Success/failure is carried by `master_tasks_total{result}`, a + // counter — putting `result` on the histogram would double its series count + // (every bucket, twice) to describe the latency of a rare event. + metrics::histogram!("master_task_duration_seconds", "kind" => kind) + .record(work_elapsed.as_secs_f64()); + metrics::counter!("master_tasks_total", "kind" => kind, "result" => result).increment(1); if let Err(error) = &outcome { tracing::warn!(task = %task.id, target = %task.target, error, "task failed"); } - if let Err(error) = state.task_store.finish(claim, outcome).await { + let commit_start = std::time::Instant::now(); + let finished = state.task_store.finish(claim, outcome).await; + metrics::histogram!("master_task_phase_duration_seconds", "kind" => kind, "phase" => "commit") + .record(commit_start.elapsed().as_secs_f64()); + if let Err(error) = finished { tracing::error!(task = %task.id, error = %error, "failed to persist task completion"); } } @@ -177,17 +211,26 @@ async fn run_merge_wal(state: &Arc, name: &str) -> Result(0); - } - if !status.is_success() { - return Err(format!("{url}: HTTP {status}")); - } - let body: MergeWalReply = resp.json().await.map_err(|e| e.to_string())?; - Ok(body.reclaimed) + // Per-worker timing: `join_all` means the slowest worker sets the + // whole task's latency, so without this one straggler is + // indistinguishable from every worker being slow. Unlabelled -- + // outcome is carried by the counter below, which costs one series + // per value instead of one per bucket per value. + let started = std::time::Instant::now(); + let outcome = merge_wal_one(&http, &url).await; + let result = match &outcome { + Ok(WorkerMerge::Reclaimed(_)) => "ok", + Ok(WorkerMerge::NotFound) => "not_found", + Err(WorkerMergeError::Http(_)) => "http_error", + Err(WorkerMergeError::Transport(_)) => "transport_error", + }; + metrics::histogram!("master_merge_wal_worker_duration_seconds") + .record(started.elapsed().as_secs_f64()); + // Counted per worker per attempt: a 404 is tolerated as "owns no + // shard" and N-1 failures still report task success, so this counter + // is the only place partial failure is visible at all. + metrics::counter!("master_merge_wal_workers_total", "result" => result).increment(1); + outcome } }); @@ -198,14 +241,19 @@ async fn run_merge_wal(state: &Arc, name: &str) -> Result { + Ok(WorkerMerge::Reclaimed(n)) => { reclaimed += n; ok_workers += 1; } - Err(e) => last_err = Some(e), + Ok(WorkerMerge::NotFound) => { + ok_workers += 1; + } + Err(e) => last_err = Some(e.to_string()), } } + metrics::counter!("master_merge_wal_generations_reclaimed_total").increment(reclaimed as u64); + if ok_workers == 0 { return Err(last_err.unwrap_or_else(|| "all workers failed".to_string())); } @@ -214,6 +262,50 @@ async fn run_merge_wal(state: &Arc, name: &str) -> Result) -> std::fmt::Result { + match self { + Self::Http(m) | Self::Transport(m) => f.write_str(m), + } + } +} + +/// Issue the merge call to one worker, classifying the failure mode so the +/// caller can label its metrics. +async fn merge_wal_one(http: &reqwest::Client, url: &str) -> Result { + let resp = http + .post(url) + .send() + .await + .map_err(|e| WorkerMergeError::Transport(e.to_string()))?; + let status = resp.status(); + if status == reqwest::StatusCode::NOT_FOUND { + return Ok(WorkerMerge::NotFound); + } + if !status.is_success() { + return Err(WorkerMergeError::Http(format!("{url}: HTTP {status}"))); + } + let body: MergeWalReply = resp + .json() + .await + .map_err(|e| WorkerMergeError::Transport(e.to_string()))?; + Ok(WorkerMerge::Reclaimed(body.reclaimed)) +} + /// Refresh the stats row for `name` after a successful compaction: re-observe /// fragment/row counts and bump `last_compaction`/`total_compactions`. async fn update_stats_after_compaction(state: &Arc, name: &str, store: &RolloutStore) { @@ -390,16 +482,27 @@ pub fn spawn_scheduler(state: &Arc) -> JoinHandle<()> { metrics::gauge!("master_task_queue_depth").set(queued as f64); } while sem.available_permits() > 0 { + let claim_start = std::time::Instant::now(); match dispatch_state.task_store.claim_next().await { Ok(Some(claim)) => { + let claim_elapsed = claim_start.elapsed(); + // The task is already claimed at this point (queue key + // deleted, lease granted, target lock held), so time + // spent here is a claimed-but-idle task holding its + // per-experiment lock — worth seeing separately. + let permit_start = std::time::Instant::now(); let permit = sem .clone() .acquire_owned() .await .expect("semaphore never closed"); + let timing = TaskClaimTiming { + claim: claim_elapsed, + permit_wait: permit_start.elapsed(), + }; let st = dispatch_state.clone(); tokio::spawn(async move { - run_task(&st, claim).await; + run_task(&st, claim, timing).await; drop(permit); }); } diff --git a/crates/lance-context-metrics/src/lib.rs b/crates/lance-context-metrics/src/lib.rs index e5882e4..a3a6290 100644 --- a/crates/lance-context-metrics/src/lib.rs +++ b/crates/lance-context-metrics/src/lib.rs @@ -23,26 +23,43 @@ use metrics_exporter_prometheus::{Matcher, PrometheusBuilder, PrometheusHandle}; use metrics_process::Collector; /// Explicit histogram buckets (upper bounds, seconds) for request-scale latency -/// metrics — anything matching the `_duration_seconds` suffix. Tuned for -/// sub-second-to-tens-of-seconds work like HTTP requests and rollout scans. -const REQUEST_LATENCY_BUCKETS: &[f64] = &[ - 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, -]; +/// metrics — anything matching the `_duration_seconds` suffix. +/// +/// # Why only 9 buckets +/// +/// Every bucket is a separate exported series, and in Datadog every series is a +/// separately-billed custom metric. 9 buckets keeps adjacent ratios at or below +/// 6x, which bounds `histogram_quantile` interpolation error to roughly that +/// factor *within the straddling bucket only* — ample for latency SLOs, while +/// costing a third less than a denser ladder. Boundaries sit on the values +/// people actually alert on (10ms, 100ms, 250ms, 1s). +const REQUEST_LATENCY_BUCKETS: &[f64] = &[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 10.0, 60.0]; /// Coarser buckets (upper bounds, seconds) for long-running background jobs /// (compaction, WAL merge, index builds) whose latency can reach minutes. /// Without the extended tail every sample over 60s would fall into `+Inf`, /// pinning `histogram_quantile` for high percentiles at the last finite bucket. -const JOB_LATENCY_BUCKETS: &[f64] = &[ - 0.1, 0.5, 1.0, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0, 600.0, 1800.0, -]; +/// +/// Same cardinality reasoning as [`REQUEST_LATENCY_BUCKETS`]; 7 buckets spanning +/// 0.5s..30min, since job latency is acted on at order-of-magnitude granularity. +const JOB_LATENCY_BUCKETS: &[f64] = &[0.5, 2.0, 10.0, 30.0, 120.0, 600.0, 1800.0]; /// Metric names whose latency is job-scale rather than request-scale. A `Full` /// matcher outranks the `_duration_seconds` `Suffix` matcher (the exporter /// applies Full > Prefix > Suffix), so these get [`JOB_LATENCY_BUCKETS`]. +/// +/// Note the `_lock_wait_seconds` entries: they do not end in `_duration_seconds` +/// and so match no suffix rule, meaning without an explicit entry here they +/// would fall back to the exporter's default summary rather than a histogram. const JOB_LATENCY_METRICS: &[&str] = &[ "master_task_duration_seconds", + "master_task_phase_duration_seconds", + "master_merge_wal_worker_duration_seconds", "rollout_compaction_duration_seconds", + "rollout_compaction_lock_wait_seconds", + "rollout_wal_merge_duration_seconds", + "rollout_wal_merge_request_duration_seconds", + "rollout_wal_merge_lock_wait_seconds", ]; /// Handle used to render the Prometheus exposition text on demand, plus a @@ -82,6 +99,7 @@ pub fn install_recorder() -> MetricsHandle { let process = Collector::default(); // Register the process metric descriptions once up front. process.describe(); + describe_metrics(); MetricsHandle { prometheus, @@ -89,6 +107,139 @@ pub fn install_recorder() -> MetricsHandle { } } +/// Register HELP text and units for the application metrics. +/// +/// Not merely cosmetic: Datadog's OpenMetrics check uses the exported `# TYPE` +/// to decide whether a series becomes a gauge or a monotonic count, and the +/// declared `Unit` is what makes latency render as a duration rather than a bare +/// number. Without these, `/metrics` exposes no `# HELP`/`# TYPE` for any +/// application metric. +/// +/// Latency histograms are deliberately **unlabelled by result**. Failures are +/// counted (`*_errors_total`, `*_total{result=...}`), which costs one series per +/// value, instead of labelled onto a histogram, which costs one series per +/// bucket per value. +fn describe_metrics() { + use metrics::{describe_counter, describe_gauge, describe_histogram, Unit}; + + describe_histogram!( + "http_request_duration_seconds", + Unit::Seconds, + "End-to-end HTTP handler latency, including body parsing and admission control." + ); + + // Rollout write path. `rollout_add_request_duration_seconds` is labelled by + // `flush` because flush is a query param on the add route, so the HTTP + // metric's `path` label cannot distinguish the two. + describe_histogram!( + "rollout_add_request_duration_seconds", + Unit::Seconds, + "Store time for a successful add request (add + optional flush), excluding body \ + parsing. Label: flush (true|false)." + ); + describe_histogram!( + "rollout_add_duration_seconds", + Unit::Seconds, + "RolloutStore::add — the durable WAL append only, success path." + ); + describe_counter!( + "rollout_add_errors_total", + "Failed RolloutStore::add calls." + ); + describe_histogram!( + "rollout_flush_duration_seconds", + Unit::Seconds, + "RolloutStore::flush — sealing the memtable so added rows become readable. \ + Label: outcome (sealed|noop|fenced); the paths differ by orders of magnitude." + ); + describe_counter!( + "rollout_flush_errors_total", + "Failed RolloutStore::flush calls." + ); + describe_histogram!( + "rollout_wal_merge_duration_seconds", + Unit::Seconds, + "Per-phase WAL self-merge latency. \ + Label: phase (seal|read|append|claim_epoch|drain|delete)." + ); + describe_counter!( + "rollout_wal_merge_errors_total", + "Failed WAL self-merge phases; the phase label identifies where the merge died. \ + Label: phase." + ); + describe_histogram!( + "rollout_wal_merge_request_duration_seconds", + Unit::Seconds, + "Worker-side handling of a master-driven WAL merge." + ); + describe_histogram!( + "rollout_wal_merge_lock_wait_seconds", + Unit::Seconds, + "Time waiting for the store write lock before a WAL merge (blocks all ingest)." + ); + describe_histogram!( + "rollout_compaction_lock_wait_seconds", + Unit::Seconds, + "Time waiting for the store write lock before compaction." + ); + + // Master task lifecycle. + describe_histogram!( + "master_task_duration_seconds", + Unit::Seconds, + "Task work window only (excludes claim, permit wait, and commit). Label: kind. \ + Success/failure is on master_tasks_total." + ); + describe_histogram!( + "master_task_phase_duration_seconds", + Unit::Seconds, + "Task latency broken down by phase. \ + Labels: kind, phase (claim|permit_wait|work|commit)." + ); + describe_histogram!( + "master_merge_wal_worker_duration_seconds", + Unit::Seconds, + "Per-worker round trip of a WAL-merge fan-out; the slowest worker sets task latency." + ); + describe_counter!( + "master_merge_wal_workers_total", + "WAL-merge fan-out outcomes per worker. \ + Labels: result (ok|not_found|http_error|transport_error)." + ); + describe_counter!( + "master_merge_wal_generations_reclaimed_total", + "MemWAL generations folded into base tables by master-driven merges." + ); + + // Gauges. These deliberately carry no `_total` suffix: that suffix is the + // counter convention, and Datadog infers a monotonic count from it, which + // turns "how many exist right now" into a meaningless per-second rate. + describe_gauge!( + "master_experiments", + "Live experiments seen by the last scan." + ); + describe_gauge!( + "master_rollout_rows", + "Total rollout rows across all experiments as of the last scan." + ); + describe_gauge!( + "master_rollout_fragments", + "Total fragments across all experiments as of the last scan." + ); + describe_gauge!( + "master_experiments_total", + "DEPRECATED alias for master_experiments; a gauge despite the _total suffix." + ); + describe_gauge!( + "master_rollout_rows_total", + "DEPRECATED alias for master_rollout_rows; a gauge despite the _total suffix." + ); + describe_gauge!( + "master_rollout_fragments_total", + "DEPRECATED alias for master_rollout_fragments; a gauge despite the _total suffix." + ); +} + /// Router exposing `GET /metrics` (relative to wherever it is nested/merged). pub fn metrics_router(handle: MetricsHandle) -> Router { Router::new() @@ -148,6 +299,26 @@ mod tests { metrics::histogram!("http_request_duration_seconds").record(0.3); metrics::histogram!("master_task_duration_seconds").record(45.0); + // Per-operation write-path metrics: add and flush must be separable, and + // an add is separable by whether it flushed (they share one HTTP route, + // so the `path` label cannot distinguish them). None carry `result` -- + // failures are counted, not timed. + metrics::histogram!("rollout_add_duration_seconds").record(0.02); + metrics::histogram!("rollout_flush_duration_seconds", "outcome" => "sealed").record(0.4); + metrics::histogram!("rollout_add_request_duration_seconds", "flush" => "true").record(0.5); + metrics::histogram!("rollout_add_request_duration_seconds", "flush" => "false") + .record(0.01); + metrics::counter!("rollout_add_errors_total").increment(1); + // Job-scale: WAL merge phases and per-worker fan-out. + metrics::histogram!("rollout_wal_merge_duration_seconds", "phase" => "append") + .record(120.0); + metrics::histogram!("master_task_phase_duration_seconds", + "kind" => "merge_wal", "phase" => "claim") + .record(90.0); + metrics::histogram!("master_merge_wal_worker_duration_seconds").record(200.0); + metrics::counter!("master_merge_wal_workers_total", "result" => "http_error").increment(1); + metrics::gauge!("master_experiments").set(3.0); + let app = metrics_router(handle); let resp = app .oneshot( @@ -180,15 +351,101 @@ mod tests { "request latency must not export summary quantiles; body: {text}" ); - // Job-scale metric gets the extended tail (a 300s bucket exists), so a + // Job-scale metric gets the extended tail (a 600s bucket exists), so a // 45s sample is not lumped straight into +Inf. assert!( text.contains("# TYPE master_task_duration_seconds histogram"), "job latency should be a histogram; body: {text}" ); assert!( - text.contains("master_task_duration_seconds_bucket{le=\"300\"}"), + text.contains("master_task_duration_seconds_bucket{le=\"600\"}"), "job latency should use the extended (job) bucket set; body: {text}" ); + + // add and flush must be distinct series, not one blended number. + assert!( + text.contains("# TYPE rollout_add_duration_seconds histogram") + && text.contains("# TYPE rollout_flush_duration_seconds histogram"), + "add and flush must be separate histograms; body: {text}" + ); + // flush's `outcome` separates real sealing from the no-op fast path, + // which otherwise dominates the distribution with near-zero samples. + assert!( + text.contains("outcome=\"sealed\""), + "flush should carry an outcome label; body: {text}" + ); + // The two add paths differ only by query param, so the `flush` label is + // the only thing that can separate them. + assert!( + text.contains("flush=\"true\"") && text.contains("flush=\"false\""), + "flushing and non-flushing adds must be separable; body: {text}" + ); + + // Job-scale bucket set must apply to the new long-running metrics too, + // otherwise a 120s merge phase lands in +Inf and p99 is unusable. + for name in [ + "rollout_wal_merge_duration_seconds", + "master_task_phase_duration_seconds", + "master_merge_wal_worker_duration_seconds", + ] { + assert!( + text.contains(&format!("# TYPE {name} histogram")), + "{name} should be a histogram; body: {text}" + ); + // Must assert the bucket on *this* metric's own series: a bare + // `le="600"` substring check passes off any other job-scale metric + // in the same body and silently tolerates a missing bucket config. + assert!( + text.lines().any(|line| { + line.starts_with(&format!("{name}_bucket{{")) && line.contains("le=\"600\"") + }), + "{name} should use the extended (job) bucket set; body: {text}" + ); + } + + // Every application metric should carry HELP so a scrape is + // interpretable without reading the source, and so Datadog's + // OpenMetrics check can infer the right type. + assert!( + text.contains("# HELP rollout_add_duration_seconds"), + "new metrics should be described; body: {text}" + ); + + // Latency histograms must not carry a `result` label. Failures belong on + // counters: `result` on a histogram costs one series *per bucket* per + // value, which in Datadog is one billed custom metric each. + for line in text.lines() { + if line.contains("_duration_seconds_bucket{") || line.contains("_seconds_sum{") { + assert!( + !line.contains("result=\""), + "latency histograms must not carry a `result` label \ + (use an errors counter instead): {line}" + ); + } + } + + // Gauges must not be named `_total`: Datadog infers a monotonic count + // from that suffix, turning "how many exist now" into a nonsense rate. + // The deprecated aliases are the documented exception. + assert!( + text.contains("# TYPE master_experiments gauge"), + "gauge should be exported without a _total suffix; body: {text}" + ); + + // Cardinality budget. Every series is a billed custom metric in Datadog, + // so a label added without thought is a cost regression. The bound is + // deliberately close to the actual count (112 at the time of writing): + // adding one two-valued label to a job-scale histogram costs ~9 series + // and trips this, forcing the tradeoff to be made explicitly rather than + // discovered on an invoice. + let series = text + .lines() + .filter(|l| !l.starts_with('#') && !l.is_empty()) + .count(); + assert!( + series <= 125, + "metric cardinality regressed to {series} series (budget 125). Every series is a \ + billed custom metric in Datadog — prefer a counter label over a histogram label." + ); } } diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs index d0283a6..454489f 100644 --- a/crates/lance-context-server/src/routes/rollouts.rs +++ b/crates/lance-context-server/src/routes/rollouts.rs @@ -310,6 +310,19 @@ pub async fn add_rollouts( // concurrent, so multiple ingest requests to the same store run in parallel. // Mutating ops (merge, compact, checkout, close) still take the write lock. let store = store_lock.read().await; + + // Times only the store work (`add` + optional `flush`), excluding body + // parsing, multipart decode and blob-budget admission. Subtracting this from + // `http_request_duration_seconds` gives the framing overhead. The `flush` + // label is what makes a flushing add separable at all: flush is a query + // param on this same route, so `MatchedPath` (and therefore the HTTP + // metric's `path` label) is byte-identical for both. + // + // Success path only: failures already surface as `http_requests_total` + // with a 4xx/5xx status, so a `result` label here would double this + // histogram's series count to say something already observable. + let flush_label = if flush { "true" } else { "false" }; + let store_start = std::time::Instant::now(); let version = store .add(&core_records) .await @@ -323,7 +336,13 @@ pub async fn add_rollouts( metrics::counter!("rollout_appends_flushed_total").increment(1); } - metrics::counter!("rollout_appends_total").increment(1); + metrics::histogram!( + "rollout_add_request_duration_seconds", + "flush" => flush_label, + ) + .record(store_start.elapsed().as_secs_f64()); + + metrics::counter!("rollout_appends_total", "flush" => flush_label).increment(1); metrics::counter!("rollout_records_appended_total").increment(count as u64); Ok(( @@ -555,7 +574,10 @@ pub async fn compact_rollout( None }; + let lock_start = std::time::Instant::now(); let mut store = store_lock.write().await; + ::metrics::histogram!("rollout_compaction_lock_wait_seconds") + .record(lock_start.elapsed().as_secs_f64()); let compact_start = std::time::Instant::now(); let compact_result = store.compact(config).await; ::metrics::histogram!("rollout_compaction_duration_seconds") @@ -611,15 +633,37 @@ pub async fn merge_wal( Path(name): Path, ) -> Result, AppError> { let store_lock = state.get_or_open_rollout_store(&name).await?; + // The write lock blocks all ingest on this store for the duration of the + // merge, so it is timed separately from the merge itself: a slow merge and a + // merge that spent its time queued behind other writers are different + // problems with the same end-to-end number. + let lock_start = std::time::Instant::now(); let mut store = store_lock.write().await; - let reclaimed = store - .cleanup_own_shard() - .await - .map_err(AppError::from_lance)?; - if reclaimed > 0 { - ::metrics::counter!("rollout_wal_cleanup_total", "result" => "merged").increment(1); - ::metrics::counter!("rollout_wal_generations_reclaimed_total").increment(reclaimed as u64); - } + ::metrics::histogram!("rollout_wal_merge_lock_wait_seconds") + .record(lock_start.elapsed().as_secs_f64()); + + let merge_start = std::time::Instant::now(); + let result = store.cleanup_own_shard().await; + + let reclaimed = match result { + Ok(n) => n, + Err(e) => { + ::metrics::counter!("rollout_wal_cleanup_total", "result" => "failed").increment(1); + return Err(AppError::from_lance(e)); + } + }; + // Emitted unconditionally: gating on `reclaimed > 0` made the common no-op + // merge invisible, so a worker that never has anything to merge and a worker + // that is never called looked identical. `result` stays on this *counter* + // (1 series per value) rather than on a histogram (1 series per bucket). + ::metrics::counter!( + "rollout_wal_cleanup_total", + "result" => if reclaimed > 0 { "merged" } else { "noop" }, + ) + .increment(1); + ::metrics::histogram!("rollout_wal_merge_request_duration_seconds") + .record(merge_start.elapsed().as_secs_f64()); + ::metrics::counter!("rollout_wal_generations_reclaimed_total").increment(reclaimed as u64); Ok(Json(MergeWalResponse { reclaimed })) }