feat: add per-repo fetch/serve latency histograms

README.md

@@ -150,7 +150,9 @@ Every flag has an environment-variable equivalent.
 | `--cache-max-mb`         | `GITCACHEPROXY_CACHE_MAX_MB`         | `0`                          | Cap on total on-disk mirror cache, in MiB; evicts least-recently-used idle mirrors when exceeded (`0` = unlimited, no eviction) |
 | `--git-binary`           | `GITCACHEPROXY_GIT_BINARY`           | `git`                        | Path to git                                                                    |
 
-Endpoints: `/healthz`, `/readyz`, `/metrics` (Prometheus).
+Endpoints: `/healthz`, `/readyz`, `/metrics` (Prometheus - per-repo request and
+upstream counters, cache-size gauges, and `*_duration_seconds` fetch/serve
+latency histograms).
 
 ## Auth model
 
@@ -235,7 +237,8 @@ rely on it.
 
 Not yet implemented, in rough priority order:
 
-- Per-repo latency histograms (fetch/serve durations); per-repo counters exist.
+- A Helm chart for Kubernetes deployment (liveness/readiness probes,
+  single-writer RWO PVC, metrics scrape), shipped in-repo.
 - A background/scheduled refresh option (today every `info/refs` triggers an
   on-demand, TTL-coalesced fetch).
 - No external `git` binary: move the plumbing in-process to a Rust library

src/git.rs

@@ -15,13 +15,15 @@
 
 use std::collections::HashMap;
 use std::path::Path;
+use std::pin::Pin;
 use std::process::Stdio;
 use std::sync::Arc;
+use std::task::Poll;
 use std::time::{Duration, Instant};
 
 use anyhow::{Context, Result, bail};
 use bytes::Bytes;
-use tokio::io::AsyncWriteExt;
+use tokio::io::{AsyncRead, AsyncWriteExt, ReadBuf};
 use tokio::process::{ChildStdout, Command};
 use tokio::sync::Mutex;
 use tokio_util::io::ReaderStream;
@@ -182,6 +184,7 @@ impl GitCache {
             .arg("--stateless-rpc")
             .arg("--advertise-refs")
             .arg(&repo.cache_dir);
+        let started = Instant::now();
         let out = cmd
             .output()
             .await
@@ -193,6 +196,8 @@ impl GitCache {
                 String::from_utf8_lossy(&out.stderr)
             );
         }
+        self.metrics
+            .observe_serve("info_refs", &repo.name, started.elapsed().as_secs_f64());
         let mut body = pkt_line("# service=git-upload-pack\n");
         body.extend_from_slice(b"0000"); // flush-pkt
         body.extend_from_slice(&out.stdout);
@@ -208,7 +213,10 @@ impl GitCache {
         repo: &RepoRef,
         git_protocol: Option<&str>,
         body: Bytes,
-    ) -> Result<ReaderStream<ChildStdout>> {
+    ) -> Result<ReaderStream<TimedReader<ChildStdout>>> {
+        // Serve duration spans the whole RPC: spawn, negotiation write, and the
+        // streamed packfile, recorded when the stream reaches EOF (see `TimedReader`).
+        let started = Instant::now();
         let mut cmd = self.local_cmd(git_protocol);
         cmd.arg("upload-pack")
             .arg("--stateless-rpc")
@@ -245,7 +253,12 @@ impl GitCache {
                 _ => {}
             }
         });
-        Ok(ReaderStream::new(stdout))
+        let timed = TimedReader {
+            inner: stdout,
+            repo: repo.name.clone(),
+            recorder: Some((self.metrics.clone(), started)),
+        };
+        Ok(ReaderStream::new(timed))
     }
 
     async fn clone_mirror(&self, repo: &RepoRef) -> Result<()> {
@@ -271,6 +284,7 @@ impl GitCache {
         // repo, not just HEAD, and maps them 1:1 so a later `fetch` keeps them in
         // sync. The client then negotiates whatever ref it wants via upload-pack, so
         // the mirror can serve any branch/tag/sha the origin has - never HEAD-only.
+        let started = Instant::now();
         let status = self
             .fetch_cmd()
             .arg("clone")
@@ -289,10 +303,12 @@ impl GitCache {
             self.metrics.record_upstream("clone", "error", "-");
             bail!("git clone --mirror failed for {}", repo.name);
         }
+        let elapsed = started.elapsed().as_secs_f64();
         tokio::fs::rename(&tmp, &repo.cache_dir)
             .await
             .context("rename mirror into place")?;
         self.metrics.record_upstream("clone", "ok", &repo.name);
+        self.metrics.observe_upstream("clone", &repo.name, elapsed);
         self.mark_changed(repo);
         Ok(())
     }
@@ -303,6 +319,7 @@ impl GitCache {
         // up exactly one remote named `origin` (git's default remote name) pointing at
         // the upstream URL, with a mirror refspec that updates all refs. So `origin`
         // is not an assumption about the client - it is the remote this proxy created.
+        let started = Instant::now();
         let status = self
             .fetch_cmd()
             .current_dir(&repo.cache_dir)
@@ -318,6 +335,8 @@ impl GitCache {
             bail!("git fetch failed for {}", repo.name);
         }
         self.metrics.record_upstream("fetch", "ok", &repo.name);
+        self.metrics
+            .observe_upstream("fetch", &repo.name, started.elapsed().as_secs_f64());
         self.mark_changed(repo);
         Ok(())
     }
@@ -346,6 +365,51 @@ impl GitCache {
     }
 }
 
+/// Wraps `upload-pack`'s stdout to record how long the packfile took to serve.
+/// The duration spans from the RPC starting to the stream reaching EOF - or the
+/// client disconnecting, caught by `Drop` - so it includes the client's read
+/// speed: this is serve latency, not pure generation time. Recorded exactly once
+/// (the `Option` is `take`n on the first of EOF or drop).
+pub struct TimedReader<R> {
+    inner: R,
+    repo: String,
+    recorder: Option<(Arc<Metrics>, Instant)>,
+}
+
+impl<R> TimedReader<R> {
+    fn record(&mut self) {
+        if let Some((metrics, started)) = self.recorder.take() {
+            metrics.observe_serve("upload_pack", &self.repo, started.elapsed().as_secs_f64());
+        }
+    }
+}
+
+impl<R: AsyncRead + Unpin> AsyncRead for TimedReader<R> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<std::io::Result<()>> {
+        let this = self.get_mut();
+        let before = buf.filled().len();
+        let poll = Pin::new(&mut this.inner).poll_read(cx, buf);
+        // A ready read that produced no bytes is EOF: the packfile is fully served.
+        if let Poll::Ready(Ok(())) = &poll
+            && buf.filled().len() == before
+        {
+            this.record();
+        }
+        poll
+    }
+}
+
+impl<R> Drop for TimedReader<R> {
+    fn drop(&mut self) {
+        // Covers a client that hung up before EOF; a no-op if EOF already recorded.
+        self.record();
+    }
+}
+
 /// Encode a string as a single pkt-line (4-hex length prefix + payload).
 fn pkt_line(s: &str) -> Vec<u8> {
     let mut v = format!("{:04x}", s.len() + 4).into_bytes();
@@ -362,4 +426,22 @@ mod tests {
         assert_eq!(pkt_line("a"), b"0005a");
         assert_eq!(&pkt_line("# service=git-upload-pack\n")[..4], b"001e");
     }
+
+    #[tokio::test]
+    async fn timed_reader_records_serve_duration_at_eof() {
+        use tokio::io::AsyncReadExt;
+
+        let metrics = Arc::new(Metrics::new());
+        let mut reader = TimedReader {
+            inner: &b"packfile bytes"[..],
+            repo: "group/foo.git".into(),
+            recorder: Some((metrics.clone(), Instant::now())),
+        };
+        // Reading to EOF drives the final zero-byte read, which records once.
+        let mut sink = Vec::new();
+        reader.read_to_end(&mut sink).await.unwrap();
+        assert!(metrics.gather().contains(
+            r#"gitcacheproxy_serve_duration_seconds_count{kind="upload_pack",repo="group/foo.git"} 1"#
+        ));
+    }
 }

src/metrics.rs

@@ -11,7 +11,16 @@
 //! clones), so a flood of distinct but doomed repo paths cannot inflate the
 //! series count.
 
-use prometheus::{Encoder, IntCounter, IntCounterVec, IntGauge, Opts, Registry, TextEncoder};
+use prometheus::{
+    Encoder, HistogramOpts, HistogramVec, IntCounter, IntCounterVec, IntGauge, Opts, Registry,
+    TextEncoder,
+};
+
+/// Histogram buckets, in seconds, for git operation latency: from a fast cached
+/// advertisement (tens of ms) to a large clone over a slow WAN (minutes).
+const DURATION_BUCKETS: &[f64] = &[
+    0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0,
+];
 
 pub struct Metrics {
     pub registry: Registry,
@@ -30,6 +39,13 @@ pub struct Metrics {
     cache_mirrors: IntGauge,
     /// `evictions_total` - idle mirrors evicted to keep the cache under the cap.
     evictions: IntCounter,
+    /// `upstream_duration_seconds{op, repo}` - clone/fetch wall-clock, observed
+    /// only on success (same bounded-`repo` discipline as the counters).
+    upstream_duration: HistogramVec,
+    /// `serve_duration_seconds{kind, repo}` - kind = info_refs (the buffered
+    /// advertisement) | upload_pack (the packfile stream, timed to EOF). Observed
+    /// only on success.
+    serve_duration: HistogramVec,
 }
 
 impl Metrics {
@@ -63,6 +79,24 @@ impl Metrics {
             "Idle mirrors evicted to keep the cache under the configured cap",
         )
         .expect("valid metric");
+        let upstream_duration = HistogramVec::new(
+            HistogramOpts::new(
+                "gitcacheproxy_upstream_duration_seconds",
+                "Upstream clone/fetch duration in seconds",
+            )
+            .buckets(DURATION_BUCKETS.to_vec()),
+            &["op", "repo"],
+        )
+        .expect("valid metric");
+        let serve_duration = HistogramVec::new(
+            HistogramOpts::new(
+                "gitcacheproxy_serve_duration_seconds",
+                "Client serve duration in seconds (info/refs advertisement, upload-pack stream)",
+            )
+            .buckets(DURATION_BUCKETS.to_vec()),
+            &["kind", "repo"],
+        )
+        .expect("valid metric");
         registry
             .register(Box::new(requests.clone()))
             .expect("register requests");
@@ -78,13 +112,21 @@ impl Metrics {
         registry
             .register(Box::new(evictions.clone()))
             .expect("register evictions");
+        registry
+            .register(Box::new(upstream_duration.clone()))
+            .expect("register upstream_duration");
+        registry
+            .register(Box::new(serve_duration.clone()))
+            .expect("register serve_duration");
         Self {
             registry,
             requests,
             upstream,
             cache_bytes,
             cache_mirrors,
             evictions,
+            upstream_duration,
+            serve_duration,
         }
     }
 
@@ -115,6 +157,22 @@ impl Metrics {
         self.evictions.inc();
     }
 
+    /// Observe an upstream op's duration. Call only on success with the real repo,
+    /// matching the counters' bounded-`repo` cardinality discipline.
+    pub fn observe_upstream(&self, op: &str, repo: &str, seconds: f64) {
+        self.upstream_duration
+            .with_label_values(&[op, repo])
+            .observe(seconds);
+    }
+
+    /// Observe a client serve duration (`kind` = `info_refs` | `upload_pack`),
+    /// same cardinality discipline as `observe_upstream`.
+    pub fn observe_serve(&self, kind: &str, repo: &str, seconds: f64) {
+        self.serve_duration
+            .with_label_values(&[kind, repo])
+            .observe(seconds);
+    }
+
     pub fn gather(&self) -> String {
         let mut buf = Vec::new();
         let enc = TextEncoder::new();
@@ -144,11 +202,19 @@ mod tests {
         m.set_cache_size(2048, 3);
         m.record_eviction();
         m.record_eviction();
+        m.observe_upstream("clone", "group/foo.git", 1.5);
+        m.observe_serve("upload_pack", "group/foo.git", 2.0);
 
         let out = m.gather();
         assert!(out.contains("gitcacheproxy_cache_bytes 2048"));
         assert!(out.contains("gitcacheproxy_cache_mirrors 3"));
         assert!(out.contains("gitcacheproxy_evictions_total 2"));
+        assert!(out.contains(
+            r#"gitcacheproxy_upstream_duration_seconds_count{op="clone",repo="group/foo.git"} 1"#
+        ));
+        assert!(out.contains(
+            r#"gitcacheproxy_serve_duration_seconds_count{kind="upload_pack",repo="group/foo.git"} 1"#
+        ));
         assert!(out.contains(
             r#"gitcacheproxy_requests_total{kind="info_refs",repo="group/foo.git",result="ok"} 1"#
         ));

tests/e2e.rs

@@ -273,6 +273,20 @@ async fn clones_through_proxy_serves_all_refs_and_rejects_push() {
         "cache index should track the cloned repo"
     );
 
+    // Latency histograms were recorded per repo for the synchronous ops: the
+    // upstream clone + fetch, and the info/refs advertisement serve. (The streamed
+    // upload-pack serve is timed on EOF and covered by a unit test in `git.rs`.)
+    for series in [
+        r#"gitcacheproxy_upstream_duration_seconds_count{op="clone",repo="repo.git"}"#,
+        r#"gitcacheproxy_upstream_duration_seconds_count{op="fetch",repo="repo.git"}"#,
+        r#"gitcacheproxy_serve_duration_seconds_count{kind="info_refs",repo="repo.git"}"#,
+    ] {
+        assert!(
+            scraped.contains(series),
+            "missing latency histogram {series}:\n{scraped}"
+        );
+    }
+
     // --- A push attempt is rejected over the wire (403). ---
     let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap();
     let req = format!(