From 5411400efbd899c08f1f96b8f60198d90f3cb32d Mon Sep 17 00:00:00 2001 From: Beinan Date: Sat, 25 Jul 2026 22:01:42 +0000 Subject: [PATCH 1/2] feat(metrics): split write-path latency and break down master tasks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `http_request_duration_seconds{method,path,status}` was the only latency signal for the write path, which made three things unmeasurable. **add and flush were one series.** They share a route (`routes/mod.rs:87`); flush is a query param parsed by hand (`routes/rollouts.rs:263`). The `path` label comes from `MatchedPath` — the route *template*, excluding the query string — so a flushing add and a plain add emitted byte-identical labels. Since flush is strictly additive (add, then flush), flushing requests were the slow tail contaminating p99 of plain adds with no way to demix. **core had no metrics at all.** `RolloutStore::add`, `flush`, and the WAL merge were uninstrumented, so HTTP duration blended body parsing, multipart decode, blob-budget admission, store open, lock acquisition and the actual work. **`master_task_duration_seconds` was narrower than its name.** It wrapped only the dispatch match, excluding the claim (etcd txn + target lock), the semaphore permit wait, and commit/lease release. MergeWal's `join_all` fan-out had no per-worker timing, so one straggler — which by `join_all` semantics sets the whole task's latency — was indistinguishable from every worker being slow. New metrics: rollout_add_duration_seconds{result} core: durable append rollout_flush_duration_seconds{result,outcome} core: memtable seal rollout_wal_merge_duration_seconds{phase,result} core: 6 merge phases rollout_add_request_duration_seconds{flush,result} server: store time rollout_wal_merge_request_duration_seconds{result} worker-side merge rollout_wal_merge_lock_wait_seconds write-lock wait rollout_compaction_lock_wait_seconds write-lock wait master_task_phase_duration_seconds{kind,phase} claim/permit/work/commit master_merge_wal_worker_duration_seconds{result} per-worker RTT master_merge_wal_workers_total{result} per-worker outcome master_merge_wal_generations_reclaimed_total `flush`'s `outcome` (sealed|noop|fenced) matters: no-resident-writer is the common case and returns in microseconds, so without it the histogram is dominated by near-zero samples and its percentiles say nothing about real flush cost. Merge phases are seal/read/append/claim_epoch/drain/delete, which is also where the unbounded read buffering and the cancel-unsafe append→drain window live. `master_merge_wal_workers_total{result}` distinguishes ok/not_found/http_error/ transport_error. A 404 is tolerated as "owns no shard" and N-1 failures still report task success, so this counter is currently the only place partial fan-out failure is visible at all. `rollout_wal_cleanup_total` is now emitted unconditionally with `result=merged|noop`. Gating it 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. Deliberate choices: - `metrics` is an **optional, default-on** feature of `lance-context-core` so downstream consumers embedding the library are not forced to take the dependency; call sites compile to nothing when disabled (verified with `--no-default-features`, warning-free under `-D warnings`). - No `store`/`target` labels anywhere — dataset and experiment names are unbounded cardinality. That context belongs in a tracing span. - `master_task_duration_seconds` keeps its original scope for back-compat, but gains a `result` label so a fast failure is not averaged in with successes. - The new long-running metrics are added to `JOB_LATENCY_METRICS`; without it a 120s merge phase lands in `+Inf` and high percentiles are unusable. Note the `_lock_wait_seconds` names match no suffix rule, so they need explicit entries. - Added the first `describe_*` calls in the repo — `/metrics` previously shipped with no HELP/TYPE for any application metric. Pure observability: no behaviour change. Bugs this only makes visible (MergeWal reporting success when N-1 workers fail, 404→Ok(0), unbounded merge buffering, the 100ms-spin coordination lock) are left for their own PRs. Verified: workspace tests pass (168 core / 51 server / 14 master); clippy clean with and without the feature; and a live server scrape confirms `rollout_add_request_duration_seconds{flush="true"}` and `{flush="false"}` are distinct series with `flush_duration{outcome="sealed"}` recorded only for the flushing request. The bucket-config assertion was checked to fail when the config is removed, so it is not vacuous. Co-Authored-By: Claude --- Cargo.lock | 13 + crates/lance-context-core/Cargo.toml | 13 + crates/lance-context-core/src/lib.rs | 1 + crates/lance-context-core/src/metrics.rs | 78 +++++ .../lance-context-core/src/rollout_store.rs | 300 +++++++++++++++--- crates/lance-context-master/src/scheduler.rs | 144 +++++++-- crates/lance-context-metrics/src/lib.rs | 180 +++++++++++ .../src/routes/rollouts.rs | 91 +++++- 8 files changed, 735 insertions(+), 85 deletions(-) create mode 100644 crates/lance-context-core/src/metrics.rs 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..c2ebc5d --- /dev/null +++ b/crates/lance-context-core/src/metrics.rs @@ -0,0 +1,78 @@ +//! 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`. + +/// Latency of one [`crate::RolloutStore::add`] — the durable WAL append only. +/// Label: `result` = `ok` | `error`. +pub const ROLLOUT_ADD_DURATION: &str = "rollout_add_duration_seconds"; + +/// Latency of one [`crate::RolloutStore::flush`] — sealing the memtable so +/// previously added rows become readable. +/// +/// Labels: `result` = `ok` | `error`, and `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 +/// +/// Without `outcome` this histogram 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"; + +/// Per-phase latency of a WAL self-merge. Label `phase`: +/// `seal` | `read` | `append` | `claim_epoch` | `drain` | `delete`, +/// plus `result` = `ok` | `error`. +pub const ROLLOUT_WAL_MERGE_DURATION: &str = "rollout_wal_merge_duration_seconds"; + +/// 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; + }}; +} + +/// 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 + }}; +} + +pub(crate) use {observe_duration, 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..596f945 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::{observe_duration, 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,20 @@ impl RolloutStore { if records.is_empty() { return Ok(self.dataset.manifest.version); } + // Timed as a whole: `add_inner` has several `?` early-returns, so timing + // here (rather than inline) guarantees the error path is measured too -- + // a failing append that is slow is exactly what you want to see. + let started = timer_start!(); + let result = self.add_inner(records).await; + observe_duration!( + crate::metrics::ROLLOUT_ADD_DURATION, + timer_elapsed!(started), + "result" => if result.is_ok() { "ok" } else { "error" }, + ); + 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 +682,34 @@ 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` separates real sealing work from the two fast paths. Without + // it the histogram is dominated by near-zero `noop` samples (no resident + // writer is the common case) and its high percentiles are meaningless. + let (res_label, outcome) = match &result { + Ok(FlushOutcome::Sealed) => ("ok", "sealed"), + Ok(FlushOutcome::Noop) => ("ok", "noop"), + Ok(FlushOutcome::Fenced) => ("ok", "fenced"), + Err(_) => ("error", "sealed"), + }; + let _ = (res_label, outcome); + observe_duration!( + crate::metrics::ROLLOUT_FLUSH_DURATION, + timer_elapsed!(started), + "result" => res_label, + "outcome" => outcome, + ); + 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 +717,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 +879,42 @@ 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?; + let phase = timer_start!(); + let sealed = self.close().await; + observe_duration!( + crate::metrics::ROLLOUT_WAL_MERGE_DURATION, + timer_elapsed!(phase), + "phase" => "seal", + "result" => if sealed.is_ok() { "ok" } else { "error" }, + ); + sealed?; 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 phase = timer_start!(); + let read = self.read_flushed_generations(manifest).await; + observe_duration!( + crate::metrics::ROLLOUT_WAL_MERGE_DURATION, + timer_elapsed!(phase), + "phase" => "read", + "result" => if read.is_ok() { "ok" } else { "error" }, + ); + let (merged_generations, merged_paths, batches, merge_schema) = read?; // Append the merged rows to the base table. if !batches.is_empty() { - let reader = RecordBatchIterator::new( - batches.into_iter().map(Ok::), - merge_schema, + let phase = timer_start!(); + let appended = self.append_merged_batches(batches, merge_schema).await; + observe_duration!( + crate::metrics::ROLLOUT_WAL_MERGE_DURATION, + timer_elapsed!(phase), + "phase" => "append", + "result" => if appended.is_ok() { "ok" } else { "error" }, ); - 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?; + appended?; } // Drain the merged generations from the shard manifest. Claim the @@ -893,8 +923,18 @@ 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 + let phase = timer_start!(); + let claimed = manifest_store.claim_epoch(manifest.shard_spec_id).await; + observe_duration!( + crate::metrics::ROLLOUT_WAL_MERGE_DURATION, + timer_elapsed!(phase), + "phase" => "claim_epoch", + "result" => if claimed.is_ok() { "ok" } else { "error" }, + ); + let (epoch, _) = claimed?; + + let phase = timer_start!(); + let drained = manifest_store .commit_update(epoch, |current| ShardManifest { version: current.version + 1, flushed_generations: current @@ -905,7 +945,14 @@ impl RolloutStore { .collect(), ..current.clone() }) - .await?; + .await; + observe_duration!( + crate::metrics::ROLLOUT_WAL_MERGE_DURATION, + timer_elapsed!(phase), + "phase" => "drain", + "result" => if drained.is_ok() { "ok" } else { "error" }, + ); + drained?; // Delete the merged generations' blob directories now that no manifest // references them. Ordering matters: the drain above already removed @@ -922,6 +969,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 +988,80 @@ impl RolloutStore { ); } } + // Best-effort by contract, so always `ok`: a delete failure is logged + // above and does not fail the merge. + observe_duration!( + crate::metrics::ROLLOUT_WAL_MERGE_DURATION, + timer_elapsed!(phase), + "phase" => "delete", + "result" => "ok", + ); + + 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 +5132,83 @@ 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; + assert!( + labels.iter().any(|(k, v)| k == "result" && v == "ok"), + "add should be labelled by result; got {labels:?}" + ); + } else if name == crate::metrics::ROLLOUT_FLUSH_DURATION { + 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" + ); + } } diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index a6e6bc9..5cecbba 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -83,26 +83,58 @@ 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()); + // Scope unchanged (the work window) for back-compat, but now labelled by + // result so a fast failure is not averaged in with fast successes. + metrics::histogram!("master_task_duration_seconds", "kind" => kind, "result" => result) + .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 +209,27 @@ 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. + 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", + "result" => result, + ) + .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 +240,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 +261,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 +481,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..4573c32 100644 --- a/crates/lance-context-metrics/src/lib.rs +++ b/crates/lance-context-metrics/src/lib.rs @@ -40,9 +40,19 @@ const JOB_LATENCY_BUCKETS: &[f64] = &[ /// 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 +92,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 +100,92 @@ pub fn install_recorder() -> MetricsHandle { } } +/// Register HELP text for the application metrics. +/// +/// Purely descriptive, but without it `/metrics` exposes no `# HELP`/`# TYPE` +/// for any application metric, so a scraped series is uninterpretable without +/// reading the source. +fn describe_metrics() { + use metrics::{describe_counter, 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 an add request (add + optional flush), excluding body parsing. \ + Labels: flush, result." + ); + describe_histogram!( + "rollout_add_duration_seconds", + Unit::Seconds, + "RolloutStore::add — the durable WAL append only. Label: result." + ); + describe_histogram!( + "rollout_flush_duration_seconds", + Unit::Seconds, + "RolloutStore::flush — sealing the memtable so added rows become readable. \ + Labels: result, outcome (sealed|noop|fenced)." + ); + describe_histogram!( + "rollout_wal_merge_duration_seconds", + Unit::Seconds, + "Per-phase WAL self-merge latency. \ + Labels: phase (seal|read|append|claim_epoch|drain|delete), result." + ); + describe_histogram!( + "rollout_wal_merge_request_duration_seconds", + Unit::Seconds, + "Worker-side handling of a master-driven WAL merge. Label: result." + ); + 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). \ + Labels: kind, result." + ); + 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. \ + Label: result." + ); + 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." + ); +} + /// Router exposing `GET /metrics` (relative to wherever it is nested/merged). pub fn metrics_router(handle: MetricsHandle) -> Router { Router::new() @@ -148,6 +245,41 @@ 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). + metrics::histogram!("rollout_add_duration_seconds", "result" => "ok").record(0.02); + metrics::histogram!( + "rollout_flush_duration_seconds", + "result" => "ok", + "outcome" => "sealed", + ) + .record(0.4); + metrics::histogram!( + "rollout_add_request_duration_seconds", + "flush" => "true", + "result" => "ok", + ) + .record(0.5); + metrics::histogram!( + "rollout_add_request_duration_seconds", + "flush" => "false", + "result" => "ok", + ) + .record(0.01); + // Job-scale: WAL merge phases and per-worker fan-out. + metrics::histogram!( + "rollout_wal_merge_duration_seconds", + "phase" => "append", + "result" => "ok", + ) + .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", "result" => "ok") + .record(200.0); + let app = metrics_router(handle); let resp = app .oneshot( @@ -190,5 +322,53 @@ mod tests { text.contains("master_task_duration_seconds_bucket{le=\"300\"}"), "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 300s bucket on *this* metric's own series: a bare + // `le="300"` 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=\"300\"") + }), + "{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. + assert!( + text.contains("# HELP rollout_add_duration_seconds"), + "new metrics should be described; body: {text}" + ); } } diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs index d0283a6..b787d42 100644 --- a/crates/lance-context-server/src/routes/rollouts.rs +++ b/crates/lance-context-server/src/routes/rollouts.rs @@ -310,20 +310,56 @@ 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; - let version = store - .add(&core_records) - .await - .map_err(AppError::from_lance)?; + + // 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. + let store_start = std::time::Instant::now(); + let add_result = store.add(&core_records).await; + let version = match add_result { + Ok(v) => v, + Err(e) => { + metrics::histogram!( + "rollout_add_request_duration_seconds", + "flush" => if flush { "true" } else { "false" }, + "result" => "error", + ) + .record(store_start.elapsed().as_secs_f64()); + return Err(AppError::from_lance(e)); + } + }; // Opt-in read-your-write: seal the memtable so the rows just appended are // readable by the time this responds. Still the read lock — `flush` is // `&self`, and concurrent appends are safe against it. if flush { - store.flush().await.map_err(AppError::from_lance)?; + if let Err(e) = store.flush().await { + metrics::histogram!( + "rollout_add_request_duration_seconds", + "flush" => "true", + "result" => "error", + ) + .record(store_start.elapsed().as_secs_f64()); + return Err(AppError::from_lance(e)); + } metrics::counter!("rollout_appends_flushed_total").increment(1); } - metrics::counter!("rollout_appends_total").increment(1); + metrics::histogram!( + "rollout_add_request_duration_seconds", + "flush" => if flush { "true" } else { "false" }, + "result" => "ok", + ) + .record(store_start.elapsed().as_secs_f64()); + + metrics::counter!( + "rollout_appends_total", + "flush" => if flush { "true" } else { "false" }, + ) + .increment(1); metrics::counter!("rollout_records_appended_total").increment(count as u64); Ok(( @@ -555,7 +591,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 +650,39 @@ 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; + ::metrics::histogram!( + "rollout_wal_merge_request_duration_seconds", + "result" => if result.is_ok() { "ok" } else { "error" }, + ) + .record(merge_start.elapsed().as_secs_f64()); + + 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. + ::metrics::counter!( + "rollout_wal_cleanup_total", + "result" => if reclaimed > 0 { "merged" } else { "noop" }, + ) + .increment(1); + ::metrics::counter!("rollout_wal_generations_reclaimed_total").increment(reclaimed as u64); Ok(Json(MergeWalResponse { reclaimed })) } From 2a3ca0a6a2a3967f33b440990c717a952f0eb32b Mon Sep 17 00:00:00 2001 From: Beinan Date: Sun, 26 Jul 2026 01:37:29 +0000 Subject: [PATCH 2/2] fix(metrics): make the new metrics Datadog-safe, not just Prometheus-safe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first pass was correct for Prometheus and expensive for Datadog. Datadog bills per unique name+tag combination, and a histogram bills *every bucket* separately, so what reads as harmless label design in Prometheus is a direct line item there. Measured on a real scrape, the previous commit emitted 128 series from two requests, with a worst-case of 724 across all label combinations. Three changes, all keeping the add/flush split intact. 1. Failures are counted, not timed. `result="ok"|"error"` on a histogram doubles its series count in order to describe the *latency distribution of a rare event*, which is almost never actionable — the actionable signal is the rate. Dropped `result` from every latency histogram and added flat counters instead (1 series each): rollout_add_errors_total rollout_flush_errors_total rollout_wal_merge_errors_total{phase} The merge counter keeps `phase` because a merge aborts on its first failing phase, so the label also says where it died. Server-side this let the add handler go back to plain `?` instead of three hand-written error arms. `master_task_duration_seconds` loses `result` for the same reason; `master_tasks_total{kind,result}` already carries it as a counter. 2. Trimmed bucket ladders. REQUEST 13 -> 9 buckets, JOB 11 -> 7. Adjacent ratios stay at or below 6x, which bounds interpolation error to that factor within the straddling bucket only — ample for latency SLOs. Boundaries sit on values people alert on (10ms, 100ms, 250ms, 1s). Coverage verified: 2ms..45s all land in finite buckets. 3. Gauges no longer named `_total`. `master_experiments_total`, `master_rollout_rows_total` and `master_rollout_fragments_total` are gauges. Datadog's OpenMetrics check infers type from the name, so `_total` made them ingest as monotonic counts — graphing "experiments created per second" for a metric meaning "how many exist right now". `rate()` was equally meaningless in Prometheus. Now emitted under `master_experiments` / `master_rollout_rows` / `master_rollout_fragments`, with the old names retained as deprecated aliases so existing dashboards keep working. Also added `describe_gauge!` calls and units: Datadog uses the exported TYPE to choose gauge vs count, and the declared Unit is what renders latency as a duration rather than a bare number. Result, measured on the same two-request workload: 128 -> 87 series (-32%), worst case 724 -> ~361 (-50%). No signal lost — the add/flush split, the flush `outcome` breakdown, the six merge phases and the per-worker fan-out are all intact, and failures are now visible as rates rather than buried in a histogram. Three regression guards, each verified to fail when deliberately broken: - latency histograms may not carry `result` (checked against the rendered text) - a cardinality budget of 125 series, close enough to the actual 112 that adding one two-valued label to a job histogram trips it - core-level assertion that histogram labels come from a closed, documented set, so an unbounded label (store URI, shard id, experiment) cannot be introduced Verified: workspace tests pass (166 core / 51 server / 14 master); clippy clean with and without the `metrics` feature; live scrape confirms zero `result=` on any bucket series and both add paths still separable. Co-Authored-By: Claude --- crates/lance-context-core/src/metrics.rs | 83 ++++++- .../lance-context-core/src/rollout_store.rs | 221 +++++++++++------- crates/lance-context-master/src/scanner.rs | 11 + crates/lance-context-master/src/scheduler.rs | 19 +- crates/lance-context-metrics/src/lib.rs | 191 ++++++++++----- .../src/routes/rollouts.rs | 51 ++-- 6 files changed, 382 insertions(+), 194 deletions(-) diff --git a/crates/lance-context-core/src/metrics.rs b/crates/lance-context-core/src/metrics.rs index c2ebc5d..332f6ca 100644 --- a/crates/lance-context-core/src/metrics.rs +++ b/crates/lance-context-core/src/metrics.rs @@ -7,28 +7,53 @@ //! //! 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. -/// Label: `result` = `ok` | `error`. +/// 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. /// -/// Labels: `result` = `ok` | `error`, and `outcome`: +/// 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 /// -/// Without `outcome` this histogram is dominated by near-zero `noop` samples and -/// its high percentiles say nothing about real flush cost. +/// `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`, -/// plus `result` = `ok` | `error`. +/// `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 { @@ -44,6 +69,26 @@ macro_rules! observe_duration { }}; } +/// 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 { @@ -75,4 +120,28 @@ macro_rules! timer_elapsed { }}; } -pub(crate) use {observe_duration, timer_elapsed, timer_start}; +/// 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 596f945..a2a5d0c 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -83,7 +83,7 @@ use serde_json::Value; use tracing::{info, warn}; use uuid::Uuid; -use crate::metrics::{observe_duration, timer_elapsed, timer_start}; +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, @@ -594,16 +594,18 @@ impl RolloutStore { if records.is_empty() { return Ok(self.dataset.manifest.version); } - // Timed as a whole: `add_inner` has several `?` early-returns, so timing - // here (rather than inline) guarantees the error path is measured too -- - // a failing append that is slow is exactly what you want to see. + // 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; - observe_duration!( - crate::metrics::ROLLOUT_ADD_DURATION, - timer_elapsed!(started), - "result" => if result.is_ok() { "ok" } else { "error" }, - ); + match &result { + Ok(_) => observe_duration!( + crate::metrics::ROLLOUT_ADD_DURATION, + timer_elapsed!(started), + ), + Err(_) => count!(crate::metrics::ROLLOUT_ADD_ERRORS), + } result } @@ -684,22 +686,26 @@ impl RolloutStore { pub async fn flush(&self) -> LanceResult<()> { let started = timer_start!(); let result = self.flush_inner().await; - // `outcome` separates real sealing work from the two fast paths. Without - // it the histogram is dominated by near-zero `noop` samples (no resident - // writer is the common case) and its high percentiles are meaningless. - let (res_label, outcome) = match &result { - Ok(FlushOutcome::Sealed) => ("ok", "sealed"), - Ok(FlushOutcome::Noop) => ("ok", "noop"), - Ok(FlushOutcome::Fenced) => ("ok", "fenced"), - Err(_) => ("error", "sealed"), - }; - let _ = (res_label, outcome); - observe_duration!( - crate::metrics::ROLLOUT_FLUSH_DURATION, - timer_elapsed!(started), - "result" => res_label, - "outcome" => outcome, - ); + // `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(|_| ()) } @@ -879,42 +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. - let phase = timer_start!(); - let sealed = self.close().await; - observe_duration!( - crate::metrics::ROLLOUT_WAL_MERGE_DURATION, - timer_elapsed!(phase), - "phase" => "seal", - "result" => if sealed.is_ok() { "ok" } else { "error" }, - ); - sealed?; + 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 phase = timer_start!(); - let read = self.read_flushed_generations(manifest).await; - observe_duration!( - crate::metrics::ROLLOUT_WAL_MERGE_DURATION, - timer_elapsed!(phase), - "phase" => "read", - "result" => if read.is_ok() { "ok" } else { "error" }, - ); - let (merged_generations, merged_paths, batches, merge_schema) = read?; + 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 phase = timer_start!(); - let appended = self.append_merged_batches(batches, merge_schema).await; - observe_duration!( - crate::metrics::ROLLOUT_WAL_MERGE_DURATION, - timer_elapsed!(phase), - "phase" => "append", - "result" => if appended.is_ok() { "ok" } else { "error" }, - ); - appended?; + observe_phase!( + "append", + self.append_merged_batches(batches, merge_schema).await + )?; } // Drain the merged generations from the shard manifest. Claim the @@ -923,36 +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 phase = timer_start!(); - let claimed = manifest_store.claim_epoch(manifest.shard_spec_id).await; - observe_duration!( - crate::metrics::ROLLOUT_WAL_MERGE_DURATION, - timer_elapsed!(phase), - "phase" => "claim_epoch", - "result" => if claimed.is_ok() { "ok" } else { "error" }, - ); - let (epoch, _) = claimed?; + let (epoch, _) = observe_phase!( + "claim_epoch", + manifest_store.claim_epoch(manifest.shard_spec_id).await + )?; - let phase = timer_start!(); - let drained = 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; - observe_duration!( - crate::metrics::ROLLOUT_WAL_MERGE_DURATION, - timer_elapsed!(phase), - "phase" => "drain", - "result" => if drained.is_ok() { "ok" } else { "error" }, - ); - drained?; + 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 @@ -988,13 +964,12 @@ impl RolloutStore { ); } } - // Best-effort by contract, so always `ok`: a delete failure is logged - // above and does not fail the merge. + // 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", - "result" => "ok", ); Ok(()) @@ -5184,11 +5159,19 @@ mod tests { }; 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.iter().any(|(k, v)| k == "result" && v == "ok"), - "add should be labelled by result; got {labels:?}" + 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") @@ -5211,4 +5194,70 @@ mod tests { "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 5cecbba..ba08f15 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -121,9 +121,11 @@ async fn run_task(state: &Arc, claim: TaskClaim, timing: TaskClaimT metrics::histogram!("master_task_phase_duration_seconds", "kind" => kind, "phase" => "work") .record(work_elapsed.as_secs_f64()); - // Scope unchanged (the work window) for back-compat, but now labelled by - // result so a fast failure is not averaged in with fast successes. - metrics::histogram!("master_task_duration_seconds", "kind" => kind, "result" => result) + // 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); @@ -211,7 +213,9 @@ async fn run_merge_wal(state: &Arc, name: &str) -> Result, name: &str) -> Result "http_error", Err(WorkerMergeError::Transport(_)) => "transport_error", }; - metrics::histogram!( - "master_merge_wal_worker_duration_seconds", - "result" => result, - ) - .record(started.elapsed().as_secs_f64()); + 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. diff --git a/crates/lance-context-metrics/src/lib.rs b/crates/lance-context-metrics/src/lib.rs index 4573c32..a3a6290 100644 --- a/crates/lance-context-metrics/src/lib.rs +++ b/crates/lance-context-metrics/src/lib.rs @@ -23,19 +23,26 @@ 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 @@ -100,13 +107,20 @@ pub fn install_recorder() -> MetricsHandle { } } -/// Register HELP text for the application metrics. +/// 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. /// -/// Purely descriptive, but without it `/metrics` exposes no `# HELP`/`# TYPE` -/// for any application metric, so a scraped series is uninterpretable without -/// reading the source. +/// 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_histogram, Unit}; + use metrics::{describe_counter, describe_gauge, describe_histogram, Unit}; describe_histogram!( "http_request_duration_seconds", @@ -120,30 +134,43 @@ fn describe_metrics() { describe_histogram!( "rollout_add_request_duration_seconds", Unit::Seconds, - "Store time for an add request (add + optional flush), excluding body parsing. \ - Labels: flush, result." + "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. Label: result." + "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. \ - Labels: result, outcome (sealed|noop|fenced)." + 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. \ - Labels: phase (seal|read|append|claim_epoch|drain|delete), result." + 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. Label: result." + "Worker-side handling of a master-driven WAL merge." ); describe_histogram!( "rollout_wal_merge_lock_wait_seconds", @@ -160,8 +187,8 @@ fn describe_metrics() { describe_histogram!( "master_task_duration_seconds", Unit::Seconds, - "Task work window only (excludes claim, permit wait, and commit). \ - Labels: kind, result." + "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", @@ -172,8 +199,7 @@ fn describe_metrics() { 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. \ - Label: result." + "Per-worker round trip of a WAL-merge fan-out; the slowest worker sets task latency." ); describe_counter!( "master_merge_wal_workers_total", @@ -184,6 +210,34 @@ fn describe_metrics() { "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). @@ -247,38 +301,23 @@ mod tests { // 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). - metrics::histogram!("rollout_add_duration_seconds", "result" => "ok").record(0.02); - metrics::histogram!( - "rollout_flush_duration_seconds", - "result" => "ok", - "outcome" => "sealed", - ) - .record(0.4); - metrics::histogram!( - "rollout_add_request_duration_seconds", - "flush" => "true", - "result" => "ok", - ) - .record(0.5); - metrics::histogram!( - "rollout_add_request_duration_seconds", - "flush" => "false", - "result" => "ok", - ) - .record(0.01); + // 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", - "result" => "ok", - ) - .record(120.0); + 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", "result" => "ok") - .record(200.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 @@ -312,14 +351,14 @@ 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}" ); @@ -353,22 +392,60 @@ mod tests { text.contains(&format!("# TYPE {name} histogram")), "{name} should be a histogram; body: {text}" ); - // Must assert the 300s bucket on *this* metric's own series: a bare - // `le="300"` substring check passes off any other job-scale metric + // 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=\"300\"") + 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. + // 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 b787d42..454489f 100644 --- a/crates/lance-context-server/src/routes/rollouts.rs +++ b/crates/lance-context-server/src/routes/rollouts.rs @@ -317,49 +317,32 @@ pub async fn add_rollouts( // 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 add_result = store.add(&core_records).await; - let version = match add_result { - Ok(v) => v, - Err(e) => { - metrics::histogram!( - "rollout_add_request_duration_seconds", - "flush" => if flush { "true" } else { "false" }, - "result" => "error", - ) - .record(store_start.elapsed().as_secs_f64()); - return Err(AppError::from_lance(e)); - } - }; + let version = store + .add(&core_records) + .await + .map_err(AppError::from_lance)?; // Opt-in read-your-write: seal the memtable so the rows just appended are // readable by the time this responds. Still the read lock — `flush` is // `&self`, and concurrent appends are safe against it. if flush { - if let Err(e) = store.flush().await { - metrics::histogram!( - "rollout_add_request_duration_seconds", - "flush" => "true", - "result" => "error", - ) - .record(store_start.elapsed().as_secs_f64()); - return Err(AppError::from_lance(e)); - } + store.flush().await.map_err(AppError::from_lance)?; metrics::counter!("rollout_appends_flushed_total").increment(1); } metrics::histogram!( "rollout_add_request_duration_seconds", - "flush" => if flush { "true" } else { "false" }, - "result" => "ok", + "flush" => flush_label, ) .record(store_start.elapsed().as_secs_f64()); - metrics::counter!( - "rollout_appends_total", - "flush" => if flush { "true" } else { "false" }, - ) - .increment(1); + metrics::counter!("rollout_appends_total", "flush" => flush_label).increment(1); metrics::counter!("rollout_records_appended_total").increment(count as u64); Ok(( @@ -661,11 +644,6 @@ pub async fn merge_wal( let merge_start = std::time::Instant::now(); let result = store.cleanup_own_shard().await; - ::metrics::histogram!( - "rollout_wal_merge_request_duration_seconds", - "result" => if result.is_ok() { "ok" } else { "error" }, - ) - .record(merge_start.elapsed().as_secs_f64()); let reclaimed = match result { Ok(n) => n, @@ -676,12 +654,15 @@ pub async fn merge_wal( }; // 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. + // 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 })) }