feat(metrics): expose job queue saturation

Author: haasonsaas

TODO.md

@@ -128,7 +128,7 @@ This roadmap is derived from deep research into Greptile's public docs, blog, MC
 
 81. [ ] Split `src/server/api.rs` by domain so the growing platform API stays maintainable.
 82. [ ] Split `src/server/state.rs` into session lifecycle, persistence, progress, and GitHub coordination modules.
-83. [ ] Add queue depth and worker saturation metrics for long-running review and eval jobs.
+83. [x] Add queue depth and worker saturation metrics for long-running review and eval jobs.
 84. [ ] Add retention policies for review artifacts, eval artifacts, and trend histories.
 85. [x] Add storage migrations for richer comment lifecycle and reinforcement schemas.
 86. [ ] Add deployment docs for self-hosted review + analytics + trend retention setups.

src/server/metrics.rs

@@ -1,10 +1,45 @@
 use axum::extract::State;
 use axum::response::IntoResponse;
 use std::collections::HashMap;
-use std::fmt::Write;
+use std::fmt::{Display, Write};
 use std::sync::Arc;
 
-use super::state::{AppState, ReviewStatus};
+use super::state::{AppState, ReviewStatus, MAX_CONCURRENT_REVIEWS};
+
+#[derive(Clone, Copy)]
+struct LongRunningJobMetrics {
+    job_type: &'static str,
+    queue_depth: u64,
+    worker_capacity: u64,
+    workers_active: u64,
+    workers_available: u64,
+}
+
+impl LongRunningJobMetrics {
+    const fn new(
+        job_type: &'static str,
+        queue_depth: u64,
+        worker_capacity: u64,
+        workers_active: u64,
+        workers_available: u64,
+    ) -> Self {
+        Self {
+            job_type,
+            queue_depth,
+            worker_capacity,
+            workers_active,
+            workers_available,
+        }
+    }
+
+    fn worker_saturation_ratio(self) -> f64 {
+        if self.worker_capacity == 0 {
+            0.0
+        } else {
+            self.workers_active as f64 / self.worker_capacity as f64
+        }
+    }
+}
 
 pub async fn get_metrics(State(state): State<Arc<AppState>>) -> impl IntoResponse {
     let reviews = state.reviews.read().await;
@@ -69,6 +104,23 @@ pub async fn get_metrics(State(state): State<Arc<AppState>>) -> impl IntoRespons
 
     drop(reviews);
 
+    let review_worker_capacity = MAX_CONCURRENT_REVIEWS as u64;
+    let review_workers_available = state
+        .review_semaphore
+        .available_permits()
+        .min(MAX_CONCURRENT_REVIEWS) as u64;
+    let review_workers_active = review_worker_capacity.saturating_sub(review_workers_available);
+    let long_running_job_metrics = [
+        LongRunningJobMetrics::new(
+            "review",
+            pending,
+            review_worker_capacity,
+            review_workers_active,
+            review_workers_available,
+        ),
+        LongRunningJobMetrics::new("eval", 0, 0, 0, 0),
+    ];
+
     let mut buf = String::with_capacity(4096);
 
     write_metric(
@@ -170,6 +222,81 @@ pub async fn get_metrics(State(state): State<Arc<AppState>>) -> impl IntoRespons
         total_completion_tokens,
     );
 
+    write_metric_header(
+        &mut buf,
+        "diffscope_job_queue_depth",
+        "Current queue depth for long-running jobs by job type",
+        "gauge",
+    );
+    for metric in &long_running_job_metrics {
+        write_labeled_metric(
+            &mut buf,
+            "diffscope_job_queue_depth",
+            &[("job_type", metric.job_type)],
+            metric.queue_depth,
+        );
+    }
+
+    write_metric_header(
+        &mut buf,
+        "diffscope_job_worker_capacity",
+        "Worker capacity for long-running jobs by job type",
+        "gauge",
+    );
+    for metric in &long_running_job_metrics {
+        write_labeled_metric(
+            &mut buf,
+            "diffscope_job_worker_capacity",
+            &[("job_type", metric.job_type)],
+            metric.worker_capacity,
+        );
+    }
+
+    write_metric_header(
+        &mut buf,
+        "diffscope_job_workers_active",
+        "Active workers handling long-running jobs by job type",
+        "gauge",
+    );
+    for metric in &long_running_job_metrics {
+        write_labeled_metric(
+            &mut buf,
+            "diffscope_job_workers_active",
+            &[("job_type", metric.job_type)],
+            metric.workers_active,
+        );
+    }
+
+    write_metric_header(
+        &mut buf,
+        "diffscope_job_workers_available",
+        "Available workers for long-running jobs by job type",
+        "gauge",
+    );
+    for metric in &long_running_job_metrics {
+        write_labeled_metric(
+            &mut buf,
+            "diffscope_job_workers_available",
+            &[("job_type", metric.job_type)],
+            metric.workers_available,
+        );
+    }
+
+    write_metric_header(
+        &mut buf,
+        "diffscope_job_worker_saturation_ratio",
+        "Worker saturation ratio for long-running jobs by job type",
+        "gauge",
+    );
+    for metric in &long_running_job_metrics {
+        write_labeled_metric(
+            &mut buf,
+            "diffscope_job_worker_saturation_ratio",
+            &[("job_type", metric.job_type)],
+            metric.worker_saturation_ratio(),
+        );
+    }
+
     // Per-severity comment counts
     let _ = writeln!(
         buf,
@@ -217,15 +344,38 @@ pub async fn get_metrics(State(state): State<Arc<AppState>>) -> impl IntoRespons
     )
 }
 
-fn write_metric(buf: &mut String, name: &str, help: &str, metric_type: &str, value: u64) {
+fn write_metric_header(buf: &mut String, name: &str, help: &str, metric_type: &str) {
     let _ = writeln!(buf, "# HELP {name} {help}");
     let _ = writeln!(buf, "# TYPE {name} {metric_type}");
+}
+
+fn write_metric<T: Display>(buf: &mut String, name: &str, help: &str, metric_type: &str, value: T) {
+    write_metric_header(buf, name, help, metric_type);
     let _ = writeln!(buf, "{name} {value}");
 }
 
+fn write_labeled_metric<T: Display>(
+    buf: &mut String,
+    name: &str,
+    labels: &[(&str, &str)],
+    value: T,
+) {
+    let mut label_buf = String::new();
+    for (index, (key, raw_value)) in labels.iter().enumerate() {
+        if index > 0 {
+            label_buf.push(',');
+        }
+        let _ = write!(label_buf, "{key}=\"{raw_value}\"");
+    }
+
+    let _ = writeln!(buf, "{name}{{{label_buf}}} {value}");
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::server::state::ReviewSession;
+    use axum::body::to_bytes;
 
     #[test]
     fn test_write_metric_format() {
@@ -235,4 +385,85 @@ mod tests {
         assert!(buf.contains("# TYPE test_metric gauge\n"));
         assert!(buf.contains("test_metric 42\n"));
     }
+
+    #[test]
+    fn test_write_labeled_metric_format() {
+        let mut buf = String::new();
+        write_metric_header(&mut buf, "test_metric", "A labeled metric", "gauge");
+        write_labeled_metric(&mut buf, "test_metric", &[("job_type", "review")], 0.4);
+
+        assert!(buf.contains("# HELP test_metric A labeled metric\n"));
+        assert!(buf.contains("# TYPE test_metric gauge\n"));
+        assert!(buf.contains("test_metric{job_type=\"review\"} 0.4\n"));
+    }
+
+    fn make_session(id: &str, status: ReviewStatus) -> ReviewSession {
+        ReviewSession {
+            id: id.to_string(),
+            status,
+            diff_source: "head".to_string(),
+            github_head_sha: None,
+            github_post_results_requested: None,
+            started_at: 1,
+            completed_at: None,
+            comments: Vec::new(),
+            summary: None,
+            files_reviewed: 0,
+            error: None,
+            pr_summary_text: None,
+            diff_content: None,
+            event: None,
+            progress: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn metrics_include_queue_depth_and_worker_saturation_gauges() {
+        let storage_path = std::path::PathBuf::from("metrics-test-reviews.json");
+        let config_path = std::path::PathBuf::from("metrics-test-config.json");
+        let storage = crate::server::storage_json::JsonStorageBackend::new(&storage_path);
+
+        let mut reviews = HashMap::new();
+        reviews.insert(
+            "running".to_string(),
+            make_session("running", ReviewStatus::Running),
+        );
+        reviews.insert(
+            "pending-1".to_string(),
+            make_session("pending-1", ReviewStatus::Pending),
+        );
+        reviews.insert(
+            "pending-2".to_string(),
+            make_session("pending-2", ReviewStatus::Pending),
+        );
+
+        let state = Arc::new(AppState {
+            config: Arc::new(tokio::sync::RwLock::new(crate::config::Config::default())),
+            repo_path: std::path::PathBuf::from("."),
+            reviews: Arc::new(tokio::sync::RwLock::new(reviews)),
+            storage: Arc::new(storage),
+            storage_path,
+            config_path,
+            http_client: reqwest::Client::new(),
+            review_semaphore: Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_REVIEWS)),
+            last_reviewed_shas: Arc::new(tokio::sync::RwLock::new(HashMap::new())),
+        });
+        let _permit = state
+            .review_semaphore
+            .clone()
+            .acquire_owned()
+            .await
+            .unwrap();
+
+        let response = get_metrics(State(state)).await.into_response();
+        let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+        let text = String::from_utf8(body.to_vec()).unwrap();
+
+        assert!(text.contains("diffscope_job_queue_depth{job_type=\"review\"} 2\n"));
+        assert!(text.contains("diffscope_job_queue_depth{job_type=\"eval\"} 0\n"));
+        assert!(text.contains("diffscope_job_worker_capacity{job_type=\"review\"} 5\n"));
+        assert!(text.contains("diffscope_job_workers_active{job_type=\"review\"} 1\n"));
+        assert!(text.contains("diffscope_job_workers_available{job_type=\"review\"} 4\n"));
+        assert!(text.contains("diffscope_job_worker_saturation_ratio{job_type=\"review\"} 0.2\n"));
+    }
 }

src/server/state.rs

@@ -172,7 +172,7 @@ pub enum ReviewStatus {
 }
 
 /// Maximum number of concurrent reviews.
-const MAX_CONCURRENT_REVIEWS: usize = 5;
+pub(crate) const MAX_CONCURRENT_REVIEWS: usize = 5;
 
 pub struct AppState {
     pub config: Arc<RwLock<Config>>,