From 1c9c7755a41d70df1e19d6e6bf3ef45fb7a7593c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tobi=20L=C3=BCtke?= Date: Sun, 13 Sep 2026 02:08:46 +0000 Subject: [PATCH] feat(lfs): record handler outcomes and honest byte metrics Adapt PR #51 with explicit offered-download and validated-upload counters. Exclude HEAD/304/error/offloaded responses; count HTTP errors even when returned as Ok(Response). A recorder regression covers range and dropped-response semantics. (cherry picked from commit e21a07841b3df40db960090b3cbabdee05e07058) Signed-off-by: Yordis Prieto --- crates/walgit-server/src/lfs.rs | 159 ++++++++++++++++++++++++++++++++ docs/LFS.md | 16 ++++ 2 files changed, 175 insertions(+) diff --git a/crates/walgit-server/src/lfs.rs b/crates/walgit-server/src/lfs.rs index eefd6567..c87e8b19 100644 --- a/crates/walgit-server/src/lfs.rs +++ b/crates/walgit-server/src/lfs.rs @@ -76,12 +76,71 @@ struct LfsError { message: String, } +/// Operation labels for the LFS metrics below. +const OP_BATCH: &str = "batch"; +const OP_GET: &str = "get"; +const OP_PUT: &str = "put"; +const OP_VERIFY: &str = "verify"; + +/// Handler completion, not socket transfer completion. Downloads stream after +/// this returns; successful responses may still fail or be cancelled downstream. +fn record_lfs(op: &'static str, t0: std::time::Instant, result: &Result) { + metrics::histogram!("walgit_lfs_handler_seconds", "op" => op) + .record(t0.elapsed().as_secs_f64()); + let outcome = match result { + Ok(response) if response.status() == StatusCode::NOT_FOUND => "not_found", + Ok(response) + if response.status().is_client_error() || response.status().is_server_error() => + { + "error" + } + Ok(_) => "ok", + Err(ApiError::NotFound(_)) => "not_found", + Err(_) => "error", + }; + metrics::counter!("walgit_lfs_requests_total", "op" => op, "result" => outcome).increment(1); +} + +/// Offered body bytes: includes an interrupted GET's offered range, excludes +/// HEAD, errors, 304 and edge offload. This is not delivered network throughput. +fn record_download_offer(method: &axum::http::Method, response: &Response) { + if method != axum::http::Method::GET + || !matches!( + response.status(), + StatusCode::OK | StatusCode::PARTIAL_CONTENT + ) + || response.headers().contains_key("x-accel-redirect") + { + return; + } + if let Some(bytes) = response + .headers() + .get(axum::http::header::CONTENT_LENGTH) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()) + { + metrics::counter!("walgit_lfs_response_bytes_offered_total").increment(bytes); + } +} + /// `POST /{repo}/info/lfs/objects/batch` pub async fn batch( st: &AppState, route: &RepoRoute, headers: &HeaderMap, body_bytes: Bytes, +) -> Result { + let t0 = std::time::Instant::now(); + let r = batch_inner(st, route, headers, body_bytes).await; + record_lfs(OP_BATCH, t0, &r); + r +} + +async fn batch_inner( + st: &AppState, + route: &RepoRoute, + headers: &HeaderMap, + body_bytes: Bytes, ) -> Result { let body: BatchRequest = serde_json::from_slice(&body_bytes) .map_err(|e| ApiError::BadRequest(format!("invalid lfs batch: {e}")))?; @@ -243,6 +302,23 @@ pub async fn get_object( headers: &HeaderMap, query: &str, peer: Option, +) -> Result { + let t0 = std::time::Instant::now(); + let r = get_object_inner(st, route, method, headers, query, peer).await; + if let Ok(response) = &r { + record_download_offer(method, response); + } + record_lfs(OP_GET, t0, &r); + r +} + +async fn get_object_inner( + st: &AppState, + route: &RepoRoute, + method: &axum::http::Method, + headers: &HeaderMap, + query: &str, + peer: Option, ) -> Result { if !st.cfg.lfs.enabled { return Err(ApiError::NotFound("lfs disabled".into())); @@ -411,6 +487,18 @@ pub async fn put_object( route: &RepoRoute, headers: &HeaderMap, body: Body, +) -> Result { + let t0 = std::time::Instant::now(); + let r = put_object_inner(st, route, headers, body).await; + record_lfs(OP_PUT, t0, &r); + r +} + +async fn put_object_inner( + st: &AppState, + route: &RepoRoute, + headers: &HeaderMap, + body: Body, ) -> Result { use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -461,6 +549,9 @@ pub async fn put_object( if hex::encode(hasher.finalize()) != oid { return Err(ApiError::BadRequest("lfs object sha256 mismatch".into())); } + // Input has been fully consumed and its hash validated; the store PUT may fail. + metrics::counter!("walgit_lfs_upload_bytes_validated_total").increment(n); + metrics::histogram!("walgit_lfs_upload_object_bytes").record(n as f64); store .put( &key, @@ -478,6 +569,18 @@ pub async fn verify( route: &RepoRoute, headers: &HeaderMap, body_bytes: Bytes, +) -> Result { + let t0 = std::time::Instant::now(); + let r = verify_inner(st, route, headers, body_bytes).await; + record_lfs(OP_VERIFY, t0, &r); + r +} + +async fn verify_inner( + st: &AppState, + route: &RepoRoute, + headers: &HeaderMap, + body_bytes: Bytes, ) -> Result { if !st.cfg.lfs.enabled { return Err(ApiError::NotFound("lfs disabled".into())); @@ -548,3 +651,59 @@ fn auth_err(e: crate::auth::AuthError) -> ApiError { fn store_err(e: walgit_store::StoreError) -> ApiError { e.into() } + +#[cfg(test)] +mod metric_tests { + use super::*; + + #[test] + fn offers_exclude_head_errors_not_modified_and_edge_offload() { + let recorder = metrics_exporter_prometheus::PrometheusBuilder::new().build_recorder(); + let handle = recorder.handle(); + metrics::with_local_recorder(&recorder, || { + for (method, status, accel) in [ + ("HEAD", 200, false), + ("GET", 304, false), + ("GET", 404, false), + ("GET", 416, false), + ("GET", 200, true), + ] { + let mut builder = Response::builder() + .status(status) + .header("content-length", "100"); + if accel { + builder = builder.header("x-accel-redirect", "/internal"); + } + let response = builder.body(Body::empty()).unwrap(); + record_download_offer(&method.parse().unwrap(), &response); + } + assert!( + !handle + .render() + .contains("walgit_lfs_response_bytes_offered_total") + ); + for (status, bytes) in [(200, 100), (206, 7)] { + let response = Response::builder() + .status(status) + .header("content-length", bytes) + .body(Body::empty()) + .unwrap(); + record_download_offer(&axum::http::Method::GET, &response); + // Dropping a response does not turn offered bytes into delivered bytes. + drop(response); + } + let response = Response::builder().status(503).body(Body::empty()).unwrap(); + record_lfs(OP_GET, std::time::Instant::now(), &Ok(response)); + }); + let text = handle.render(); + assert!( + text.contains("walgit_lfs_response_bytes_offered_total 107"), + "{text}" + ); + assert!( + text.contains("walgit_lfs_requests_total{op=\"get\",result=\"error\"} 1"), + "{text}" + ); + assert!(!text.contains("walgit_lfs_bytes_total")); + } +} diff --git a/docs/LFS.md b/docs/LFS.md index 41f4ae08..63f5462b 100644 --- a/docs/LFS.md +++ b/docs/LFS.md @@ -59,3 +59,19 @@ from a machine with disk and bandwidth: bare clone/fetch of the upstream, `git l - `lfs.serve_via = "signed_url"` hands out presigned store URLs (S3, or GCS with a signer); the default `proxy` streams through walgit or the edge. - Size accounting of LFS bytes per repository in the overview. + +## 5. Metrics + +`walgit_lfs_handler_seconds{op}` measures batch/get/put/verify handler completion; +a GET body streams afterward. `walgit_lfs_requests_total{op,result}` records +handler outcomes (including HTTP error responses). These do not measure stream +completion, client cancellation or completed downloads; per-object batch errors +still travel in a successful HTTP response. + +`walgit_lfs_response_bytes_offered_total` counts Content-Length for successful +GET/Range responses prepared by walgit, excluding HEAD, 304, errors and edge +offload. Interrupted clients may receive less: this is offered volume, not network +throughput. `walgit_lfs_upload_bytes_validated_total` and +`walgit_lfs_upload_object_bytes` count consumed, SHA-256-validated upload bodies, +including ones whose subsequent store write fails. Direct store/edge transfers +need their own byte instrumentation.