feat(metrics): Prometheus /metrics on worker and master

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",

crates/lance-context-master/Cargo.toml

@@ -15,13 +15,15 @@ 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"] }
 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"] }

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());
 

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<MasterState>) -> lance::Result<usize> {
+    let scan_start = std::time::Instant::now();
     let entries = state.registry.read().await.list().await?;
     let live: HashSet<String> = 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<MasterState>) -> lance::Result<usize> {
     }
     // 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)
 }