diff --git a/README.md b/README.md index 998c63a..d5143aa 100644 --- a/README.md +++ b/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 diff --git a/src/git.rs b/src/git.rs index b9ffc48..a1985b5 100644 --- a/src/git.rs +++ b/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> { + ) -> Result>> { + // 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 { + inner: R, + repo: String, + recorder: Option<(Arc, Instant)>, +} + +impl TimedReader { + 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 AsyncRead for TimedReader { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + 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 Drop for TimedReader { + 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 { 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"# + )); + } } diff --git a/src/metrics.rs b/src/metrics.rs index 845b4f4..a57cb27 100644 --- a/src/metrics.rs +++ b/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,6 +112,12 @@ 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, @@ -85,6 +125,8 @@ impl Metrics { 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"# )); diff --git a/tests/e2e.rs b/tests/e2e.rs index a206328..81ee038 100644 --- a/tests/e2e.rs +++ b/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!(