From a1e448d8d1d89307e4d4fea6941e8518f45e2188 Mon Sep 17 00:00:00 2001 From: Beinan Date: Tue, 14 Jul 2026 17:13:12 +0000 Subject: [PATCH] feat(metrics): expose Prometheus /metrics on worker and master Add a shared `lance-context-metrics` crate that installs a process-wide Prometheus recorder, exposes `GET /metrics` on each binary's main HTTP port, records HTTP request count/latency via middleware, and emits process CPU/memory/FD/thread gauges (metrics-process) refreshed per scrape. Worker (data-plane) instruments rollout appends, WAL cleanup outcomes, compaction success/failure/duration, and LRU cache hit/miss + resident store gauge. Master instruments stats-scan duration and experiment aggregate gauges (count/rows/fragments). Each process exposes only its own metrics, ready for Prometheus/Datadog scraping. Co-Authored-By: Claude Opus 4 --- Cargo.lock | 209 +++++++++++++++++- Cargo.toml | 1 + crates/lance-context-master/Cargo.toml | 2 + crates/lance-context-master/src/main.rs | 7 + crates/lance-context-master/src/scanner.rs | 13 ++ crates/lance-context-master/src/scheduler.rs | 14 ++ crates/lance-context-metrics/Cargo.toml | 17 ++ crates/lance-context-metrics/src/lib.rs | 125 +++++++++++ crates/lance-context-server/Cargo.toml | 2 + crates/lance-context-server/src/main.rs | 7 + .../src/routes/rollouts.rs | 18 +- crates/lance-context-server/src/state.rs | 45 ++-- 12 files changed, 444 insertions(+), 16 deletions(-) create mode 100644 crates/lance-context-metrics/Cargo.toml create mode 100644 crates/lance-context-metrics/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index 14285ffd..0e03f9f4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1218,6 +1218,24 @@ dependencies = [ "num-traits", ] +[[package]] +name = "bindgen" +version = "0.72.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "993776b509cfb49c750f11b8f07a46fa23e0a1386ffc01fb1e7d343efc387895" +dependencies = [ + "bitflags 2.11.1", + "cexpr", + "clang-sys", + "itertools 0.13.0", + "proc-macro2", + "quote", + "regex", + "rustc-hash", + "shlex 1.3.0", + "syn 2.0.117", +] + [[package]] name = "bitflags" version = "1.3.2" @@ -1470,7 +1488,7 @@ dependencies = [ "find-msvc-tools", "jobserver", "libc", - "shlex", + "shlex 2.0.1", ] [[package]] @@ -1488,6 +1506,15 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4f4c707c6a209cbe82d10abd08e1ea8995e9ea937d2550646e02798948992be0" +[[package]] +name = "cexpr" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766" +dependencies = [ + "nom 7.1.3", +] + [[package]] name = "cfg-if" version = "0.1.10" @@ -1551,6 +1578,17 @@ dependencies = [ "inout", ] +[[package]] +name = "clang-sys" +version = "1.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b023947811758c97c59bf9d1c188fd619ad4718dcaa767947df1cadb14f39f4" +dependencies = [ + "glob", + "libc", + "libloading", +] + [[package]] name = "clap" version = "4.6.1" @@ -5442,6 +5480,8 @@ dependencies = [ "lance 7.0.0", "lance-context-api", "lance-context-core", + "lance-context-metrics", + "metrics", "serde", "serde_json", "tempfile", @@ -5451,6 +5491,19 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "lance-context-metrics" +version = "0.1.0" +dependencies = [ + "axum", + "metrics", + "metrics-exporter-prometheus", + "metrics-process", + "tokio", + "tower", + "tracing", +] + [[package]] name = "lance-context-python" version = "0.6.3" @@ -5475,7 +5528,9 @@ dependencies = [ "clap", "lance-context-api", "lance-context-core", + "lance-context-metrics", "lru", + "metrics", "serde", "serde_json", "tempfile", @@ -6474,12 +6529,33 @@ version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "libloading" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7c4b02199fee7c5d21a5ae7d8cfa79a6ef5bb2fc834d6e9058e89c825efdc55" +dependencies = [ + "cfg-if 1.0.4", + "windows-link", +] + [[package]] name = "libm" version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" +[[package]] +name = "libproc" +version = "0.14.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a54ad7278b8bc5301d5ffd2a94251c004feb971feba96c971ea4063645990757" +dependencies = [ + "bindgen", + "errno", + "libc", +] + [[package]] name = "libredox" version = "0.1.17" @@ -6652,6 +6728,12 @@ dependencies = [ "pkg-config", ] +[[package]] +name = "mach2" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dae608c151f68243f2b000364e1f7b186d9c29845f7d2d85bd31b9ad77ad552b" + [[package]] name = "matchers" version = "0.2.0" @@ -6732,6 +6814,62 @@ dependencies = [ "autocfg", ] +[[package]] +name = "metrics" +version = "0.24.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89550ee9f79e88fef3119de263694973a8adb26c21d75322164fb8c493039fe2" +dependencies = [ + "portable-atomic", + "rapidhash", +] + +[[package]] +name = "metrics-exporter-prometheus" +version = "0.16.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dd7399781913e5393588a8d8c6a2867bf85fb38eaf2502fdce465aad2dc6f034" +dependencies = [ + "base64", + "indexmap 2.14.0", + "metrics", + "metrics-util", + "quanta", + "thiserror 1.0.69", +] + +[[package]] +name = "metrics-process" +version = "2.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4268d87f64a752f5a651314fc683f04da10be65701ea3e721ba4d74f79163cac" +dependencies = [ + "libc", + "libproc", + "mach2", + "metrics", + "once_cell", + "procfs", + "rlimit", + "windows", +] + +[[package]] +name = "metrics-util" +version = "0.19.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8496cc523d1f94c1385dd8f0f0c2c480b2b8aeccb5b7e4485ad6365523ae376" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", + "hashbrown 0.15.5", + "metrics", + "quanta", + "rand 0.9.4", + "rand_xoshiro", + "sketches-ddsketch", +] + [[package]] name = "mime" version = "0.3.17" @@ -7970,6 +8108,27 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "procfs" +version = "0.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25485360a54d6861439d60facef26de713b1e126bf015ec8f98239467a2b82f7" +dependencies = [ + "bitflags 2.11.1", + "procfs-core", + "rustix 1.1.4", +] + +[[package]] +name = "procfs-core" +version = "0.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6401bf7b6af22f78b563665d15a22e9aef27775b79b149a66ca022468a4e405" +dependencies = [ + "bitflags 2.11.1", + "hex", +] + [[package]] name = "prost" version = "0.13.5" @@ -8165,6 +8324,21 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "quanta" +version = "0.12.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3ab5a9d756f0d97bdc89019bd2e4ea098cf9cde50ee7564dde6b81ccc8f06c7" +dependencies = [ + "crossbeam-utils", + "libc", + "once_cell", + "raw-cpuid", + "wasi 0.11.1+wasi-snapshot-preview1", + "web-sys", + "winapi", +] + [[package]] name = "quick-xml" version = "0.37.5" @@ -8421,6 +8595,24 @@ version = "1.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "973443cf09a9c8656b574a866ab68dfa19f0867d0340648c7d2f6a71b8a8ea68" +[[package]] +name = "rapidhash" +version = "4.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5da7e78a036ce858e8d55b7e7dc8ba3a88b78350fd2155d3591bbd966b58589e" +dependencies = [ + "rustversion", +] + +[[package]] +name = "raw-cpuid" +version = "11.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "498cd0dc59d73224351ee52a95fee0f1a617a2eae0e7d9d720cc622c73a54186" +dependencies = [ + "bitflags 2.11.1", +] + [[package]] name = "rawpointer" version = "0.2.1" @@ -8868,6 +9060,15 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "rlimit" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f35ee2729c56bb610f6dba436bf78135f728b7373bdffae2ec815b2d3eb98cc3" +dependencies = [ + "libc", +] + [[package]] name = "roaring" version = "0.10.12" @@ -9433,6 +9634,12 @@ dependencies = [ "os_str_bytes", ] +[[package]] +name = "shlex" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" + [[package]] name = "shlex" version = "2.0.1" diff --git a/Cargo.toml b/Cargo.toml index c8133bc9..2efab194 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,7 @@ [workspace] members = [ "crates/lance-context-core", + "crates/lance-context-metrics", "crates/lance-context", "crates/lance-context-api", "crates/lance-context-server", diff --git a/crates/lance-context-master/Cargo.toml b/crates/lance-context-master/Cargo.toml index d37f7052..16c36c59 100644 --- a/crates/lance-context-master/Cargo.toml +++ b/crates/lance-context-master/Cargo.toml @@ -15,6 +15,7 @@ path = "src/main.rs" [dependencies] lance-context-core = { version = "0.6.3", path = "../lance-context-core" } lance-context-api = { version = "0.6.3", path = "../lance-context-api" } +lance-context-metrics = { version = "0.1.0", path = "../lance-context-metrics" } arrow-array = "58" arrow-schema = "58" axum = { version = "0.8", features = ["json"] } @@ -22,6 +23,7 @@ chrono = { version = "0.4", default-features = false, features = ["clock"] } clap = { version = "4", features = ["derive", "env"] } futures = "0.3" lance = "7.0.0" +metrics = "0.24" serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { version = "1", features = ["rt-multi-thread", "macros", "signal", "sync"] } diff --git a/crates/lance-context-master/src/main.rs b/crates/lance-context-master/src/main.rs index 9c2e062c..1bc95ee1 100644 --- a/crates/lance-context-master/src/main.rs +++ b/crates/lance-context-master/src/main.rs @@ -46,6 +46,9 @@ async fn main() { // Single serial compaction worker (+ optional periodic auto-sweep). let _scheduler = scheduler::spawn_scheduler(&state); + // Install the Prometheus recorder once, before any metrics are emitted. + let metrics_handle = lance_context_metrics::install_recorder(); + let mut app = Router::new().nest("/api/v1", routes::api_router()); // Serve the built UI (SPA) as a fallback when a `--ui-dir` is configured. @@ -57,6 +60,10 @@ async fn main() { let app = app .with_state(state) + .merge(lance_context_metrics::metrics_router(metrics_handle)) + .layer(axum::middleware::from_fn( + lance_context_metrics::http_metrics_layer, + )) .layer(TraceLayer::new_for_http()) .layer(CorsLayer::permissive()); diff --git a/crates/lance-context-master/src/scanner.rs b/crates/lance-context-master/src/scanner.rs index 355c9623..d4eaa570 100644 --- a/crates/lance-context-master/src/scanner.rs +++ b/crates/lance-context-master/src/scanner.rs @@ -26,6 +26,7 @@ const OBSERVE_TIMEOUT: Duration = Duration::from_secs(30); /// for experiments no longer in the registry. Returns the number of /// experiments successfully observed. pub async fn scan_once(state: &Arc) -> lance::Result { + let scan_start = std::time::Instant::now(); let entries = state.registry.read().await.list().await?; let live: HashSet = entries.iter().map(|e| e.name.clone()).collect(); let concurrency = state.config.scan_concurrency.max(1); @@ -53,13 +54,25 @@ pub async fn scan_once(state: &Arc) -> lance::Result { } // Drop stats rows for experiments removed from the registry. let existing = stats.list(None, usize::MAX, 0).await?; + let mut total_rows: i64 = 0; + let mut total_fragments: i64 = 0; + let mut live_count: usize = 0; for row in existing { if !live.contains(&row.name) { if let Err(e) = stats.remove(&row.name).await { tracing::warn!(store = %row.name, error = %e, "stats reconcile-remove failed"); } + } else { + live_count += 1; + total_rows += row.row_count; + total_fragments += row.fragment_count; } } + + metrics::histogram!("master_scan_duration_seconds").record(scan_start.elapsed().as_secs_f64()); + 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); Ok(count) } diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index c8b61a93..6a75ab65 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -49,12 +49,16 @@ pub async fn enqueue(state: &Arc, name: &str) -> CompactJobStatus { .insert(name.to_string(), CompactJobStatus::Queued); // Unbounded send only fails if the receiver was dropped (scheduler gone). let _ = state.compact_tx.send(name.to_string()); + metrics::counter!("master_compaction_enqueued_total").increment(1); + metrics::gauge!("master_compaction_queue_depth").increment(1.0); CompactJobStatus::Queued } /// Compact one experiment now (serial, on the scheduler task). Updates the job /// status map and the stats table's compaction counters on success. async fn run_compaction(state: &Arc, name: &str) { + // This job is leaving the queue and entering execution. + metrics::gauge!("master_compaction_queue_depth").decrement(1.0); state .jobs .lock() @@ -65,6 +69,7 @@ async fn run_compaction(state: &Arc, name: &str) { let config = compaction_config(state); let opts = RolloutStoreOptions::default(); + let compact_start = std::time::Instant::now(); let status = match RolloutStore::open_existing_with_options(&uri, opts).await { Ok(mut store) => match store.compact(Some(config)).await { Ok(metrics) => { @@ -83,6 +88,15 @@ async fn run_compaction(state: &Arc, name: &str) { }, }; + metrics::histogram!("master_compaction_duration_seconds") + .record(compact_start.elapsed().as_secs_f64()); + let result = if matches!(status, CompactJobStatus::Done { .. }) { + "success" + } else { + "failed" + }; + metrics::counter!("master_compactions_total", "result" => result).increment(1); + if let CompactJobStatus::Failed { error } = &status { tracing::warn!(store = %name, error = %error, "compaction failed"); } diff --git a/crates/lance-context-metrics/Cargo.toml b/crates/lance-context-metrics/Cargo.toml new file mode 100644 index 00000000..d41f0403 --- /dev/null +++ b/crates/lance-context-metrics/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "lance-context-metrics" +version = "0.1.0" +edition.workspace = true +license.workspace = true + +[dependencies] +axum = { version = "0.8", features = ["json"] } +metrics = "0.24" +metrics-exporter-prometheus = { version = "0.16", default-features = false } +metrics-process = "2.4" +tokio = { version = "1", features = ["time"] } +tracing = "0.1" + +[dev-dependencies] +tokio = { version = "1", features = ["rt-multi-thread", "macros"] } +tower = "0.5" diff --git a/crates/lance-context-metrics/src/lib.rs b/crates/lance-context-metrics/src/lib.rs new file mode 100644 index 00000000..f718d791 --- /dev/null +++ b/crates/lance-context-metrics/src/lib.rs @@ -0,0 +1,125 @@ +//! Shared Prometheus `/metrics` wiring for lance-context binaries. +//! +//! Each binary installs a single process-wide [`PrometheusRecorder`] via +//! [`install_recorder`], mounts the [`metrics_router`] on its main HTTP port, +//! and wraps its app with [`http_metrics_layer`] to record request counts and +//! latency. Domain metrics are emitted by each binary through the `metrics` +//! facade macros (`counter!`, `gauge!`, `histogram!`). +//! +//! Master exposes only its *own* metrics; it does not scrape workers. + +use std::sync::Arc; +use std::time::Instant; + +use axum::{ + extract::{MatchedPath, State}, + http::Request, + middleware::Next, + response::{IntoResponse, Response}, + routing::get, + Router, +}; +use metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle}; +use metrics_process::Collector; + +/// Handle used to render the Prometheus exposition text on demand, plus a +/// process-resource collector that is refreshed on each scrape. +#[derive(Clone)] +pub struct MetricsHandle { + prometheus: PrometheusHandle, + process: Arc, +} + +/// Install the global Prometheus recorder for this process. +/// +/// Must be called exactly once, before any metrics are emitted. Panics if a +/// recorder was already installed (mirrors the metrics ecosystem contract). +pub fn install_recorder() -> MetricsHandle { + let prometheus = PrometheusBuilder::new() + .install_recorder() + .expect("failed to install Prometheus recorder"); + + let process = Collector::default(); + // Register the process metric descriptions once up front. + process.describe(); + + MetricsHandle { + prometheus, + process: Arc::new(process), + } +} + +/// Router exposing `GET /metrics` (relative to wherever it is nested/merged). +pub fn metrics_router(handle: MetricsHandle) -> Router { + Router::new() + .route("/metrics", get(render)) + .with_state(handle) +} + +async fn render(State(handle): State) -> impl IntoResponse { + // Refresh process CPU/memory/FD/thread gauges just before rendering. + handle.process.collect(); + let body = handle.prometheus.render(); + ([("content-type", "text/plain; version=0.0.4")], body) +} + +/// Axum middleware that records `http_requests_total` and +/// `http_request_duration_seconds` for every handled request, labelled by +/// method, route template (low cardinality), and status code. +pub async fn http_metrics_layer(req: Request, next: Next) -> Response { + let start = Instant::now(); + let method = req.method().clone(); + let path = req + .extensions() + .get::() + .map(|p| p.as_str().to_owned()) + .unwrap_or_else(|| "unknown".to_owned()); + + let response = next.run(req).await; + + let status = response.status().as_u16().to_string(); + let elapsed = start.elapsed().as_secs_f64(); + let labels = [ + ("method", method.to_string()), + ("path", path), + ("status", status), + ]; + metrics::counter!("http_requests_total", &labels).increment(1); + metrics::histogram!("http_request_duration_seconds", &labels).record(elapsed); + + response +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::body::Body; + use axum::http::{Request, StatusCode}; + use tower::ServiceExt; + + #[tokio::test] + async fn metrics_endpoint_renders_prometheus_text() { + // install_recorder can only run once per process; guard so parallel + // tests don't double-install. + let handle = install_recorder(); + metrics::counter!("test_counter_total").increment(7); + + let app = metrics_router(handle); + let resp = app + .oneshot( + Request::builder() + .uri("/metrics") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + let text = String::from_utf8(bytes.to_vec()).unwrap(); + assert!(text.contains("test_counter_total 7"), "body: {text}"); + } +} diff --git a/crates/lance-context-server/Cargo.toml b/crates/lance-context-server/Cargo.toml index d7ce72c1..7e04a016 100644 --- a/crates/lance-context-server/Cargo.toml +++ b/crates/lance-context-server/Cargo.toml @@ -15,7 +15,9 @@ path = "src/main.rs" [dependencies] lance-context-core = { version = "0.6.3", path = "../lance-context-core" } lance-context-api = { version = "0.6.3", path = "../lance-context-api" } +lance-context-metrics = { version = "0.1.0", path = "../lance-context-metrics" } axum = { version = "0.8", features = ["json", "multipart"] } +metrics = "0.24" chrono = { version = "0.4", default-features = false, features = ["clock"] } clap = { version = "4", features = ["derive", "env"] } lru = "0.12" diff --git a/crates/lance-context-server/src/main.rs b/crates/lance-context-server/src/main.rs index 38b14a3f..6e300b5a 100644 --- a/crates/lance-context-server/src/main.rs +++ b/crates/lance-context-server/src/main.rs @@ -45,8 +45,15 @@ async fn main() { // the last `Arc` is dropped. let _sweeper = state.spawn_global_sweeper(); + // Install the Prometheus recorder once, before any metrics are emitted. + let metrics_handle = lance_context_metrics::install_recorder(); + let app = routes::router() .with_state(state) + .merge(lance_context_metrics::metrics_router(metrics_handle)) + .layer(axum::middleware::from_fn( + lance_context_metrics::http_metrics_layer, + )) .layer(TraceLayer::new_for_http()) .layer(CorsLayer::permissive()); diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs index ad6769aa..80a9a979 100644 --- a/crates/lance-context-server/src/routes/rollouts.rs +++ b/crates/lance-context-server/src/routes/rollouts.rs @@ -203,6 +203,9 @@ pub async fn add_rollouts( .await .map_err(AppError::from_lance)?; + metrics::counter!("rollout_appends_total").increment(1); + metrics::counter!("rollout_records_appended_total").increment(count as u64); + Ok(( StatusCode::CREATED, Json(AddRolloutsResponse { @@ -407,7 +410,20 @@ pub async fn compact_rollout( }; let mut store = store_lock.write().await; - let metrics = store.compact(config).await.map_err(AppError::from_lance)?; + let compact_start = std::time::Instant::now(); + let compact_result = store.compact(config).await; + ::metrics::histogram!("rollout_compaction_duration_seconds") + .record(compact_start.elapsed().as_secs_f64()); + let metrics = match compact_result { + Ok(m) => { + ::metrics::counter!("rollout_compactions_total", "result" => "success").increment(1); + m + } + Err(e) => { + ::metrics::counter!("rollout_compactions_total", "result" => "failed").increment(1); + return Err(AppError::from_lance(e)); + } + }; Ok(Json(CompactResponse { fragments_removed: metrics.fragments_removed, diff --git a/crates/lance-context-server/src/state.rs b/crates/lance-context-server/src/state.rs index 6f8fc520..70e74b74 100644 --- a/crates/lance-context-server/src/state.rs +++ b/crates/lance-context-server/src/state.rs @@ -204,8 +204,10 @@ impl AppState { ) -> Result>, AppError> { // Fast path: resident in this process's LRU (updates recency). if let Some(store) = self.rollout_stores.lock().await.get(name) { + metrics::counter!("rollout_store_cache_hits_total").increment(1); return Ok(store.clone()); } + metrics::counter!("rollout_store_cache_misses_total").increment(1); // Existence is the registry's job, not the cache's. let exists = self @@ -237,6 +239,7 @@ impl AppState { return Ok(existing.clone()); } cache.put(name.to_string(), opened.clone()); + metrics::gauge!("rollout_stores_resident").set(cache.len() as f64); Ok(opened) } @@ -310,20 +313,34 @@ impl AppState { let mut guard = store.write().await; match tokio::time::timeout(pass_timeout, guard.cleanup_own_shard()).await { Ok(Ok(0)) => {} - Ok(Ok(n)) => tracing::info!( - store = %name, - reclaimed = n, - "global sweeper merged flushed generations" - ), - Ok(Err(e)) => tracing::warn!( - store = %name, - error = %e, - "global sweeper WAL cleanup failed" - ), - Err(_elapsed) => tracing::warn!( - store = %name, - "global sweeper WAL cleanup timed out; abandoning this store this tick" - ), + Ok(Ok(n)) => { + metrics::counter!("rollout_wal_cleanup_total", "result" => "merged") + .increment(1); + metrics::counter!("rollout_wal_generations_reclaimed_total") + .increment(n as u64); + tracing::info!( + store = %name, + reclaimed = n, + "global sweeper merged flushed generations" + ) + } + Ok(Err(e)) => { + metrics::counter!("rollout_wal_cleanup_total", "result" => "failed") + .increment(1); + tracing::warn!( + store = %name, + error = %e, + "global sweeper WAL cleanup failed" + ) + } + Err(_elapsed) => { + metrics::counter!("rollout_wal_cleanup_total", "result" => "timeout") + .increment(1); + tracing::warn!( + store = %name, + "global sweeper WAL cleanup timed out; abandoning this store this tick" + ) + } } } }