From 37a89fd56895f970ae49dc7d1e0cfb3dd2bcc223 Mon Sep 17 00:00:00 2001 From: Beinan Date: Tue, 8 Sep 2026 20:27:27 +0000 Subject: [PATCH] feat(server): add blob-size histogram and oversized-blob warn log Inline binary_payload size drives memory pressure on both the worker (WAL merge buffers, resident manifests) and the master (compaction rewrite buffer), but there was no way to see it: the only blob metric was rollout_blob_budget_rejections_total, which fires solely when ROLLOUT_MAX_INFLIGHT_BLOB_BYTES is set (default 0 = disabled). Capacity problems were invisible until a pod died. Emit a rollout_blob_bytes histogram per inline blob on the add path, and warn when a single blob crosses ROLLOUT_LARGE_BLOB_LOG_BYTES (default 16 MiB, 0 disables) with the experiment and record id attached. Instrumentation is placed where the multipart and JSON parse paths converge rather than inside parse_multipart_rollouts' blob loop as the issue suggested: a JSON body populates binary_payload directly and never enters that loop, so instrumenting there would silently miss inline-JSON uploads. The converged site also already has the experiment name in scope, so no threading through the parser is needed. The histogram is intentionally unlabelled. Tagging by experiment is the more useful signal but each label value is its own time series, and a deployment can carry thousands of experiments; the warn log supplies the identifying detail for the outliers that actually matter. Tests, both verified to fail when the behavior is reverted: - blob_sizes_are_recorded_for_records_with_payloads asserts via a DebuggingRecorder that exactly the payload-carrying records produce samples (a payload-less record contributes nothing), covering the inline-JSON shape the multipart loop would have missed. - large_blob_log_threshold_is_inclusive_and_zero_disables pins the boundary and the disable switch. Co-Authored-By: Claude Opus 5 --- Cargo.lock | 1 + crates/lance-context-server/Cargo.toml | 5 + crates/lance-context-server/src/config.rs | 15 +++ .../src/routes/rollouts.rs | 103 ++++++++++++++++++ crates/lance-context-server/src/state.rs | 5 + 5 files changed, 129 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index 69288527..79283840 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6439,6 +6439,7 @@ dependencies = [ "lance-context-metrics", "lru", "metrics", + "metrics-util", "serde", "serde_json", "tempfile", diff --git a/crates/lance-context-server/Cargo.toml b/crates/lance-context-server/Cargo.toml index bcb7e950..c4fa6783 100644 --- a/crates/lance-context-server/Cargo.toml +++ b/crates/lance-context-server/Cargo.toml @@ -32,4 +32,9 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] } uuid = { version = "1", features = ["v4"] } [dev-dependencies] +# Snapshotting recorder so tests can assert which metric series an operation +# emitted, without installing a process-global Prometheus exporter. +metrics-util = { version = "0.19", default-features = false, features = [ + "debugging", +] } tempfile = "3" diff --git a/crates/lance-context-server/src/config.rs b/crates/lance-context-server/src/config.rs index 2d368184..8642086e 100644 --- a/crates/lance-context-server/src/config.rs +++ b/crates/lance-context-server/src/config.rs @@ -125,6 +125,21 @@ pub struct ServerConfig { #[arg(long, env = "ROLLOUT_MAX_INFLIGHT_BLOB_BYTES", default_value = "0")] pub rollout_max_inflight_blob_bytes: usize, + /// Size, in bytes, at or above which a single inline artifact blob is + /// logged at `WARN` with its experiment and record id. + /// + /// Blob size drives memory pressure on both the worker (WAL merge buffers + /// flushed generations, and every resident store keeps its manifest) and + /// the master (the compaction rewrite buffer). The `rollout_blob_bytes` + /// histogram shows the *distribution* but not which experiment produced an + /// outlier; thousands of experiments make a per-experiment histogram label + /// too high-cardinality to carry by default. This log closes that gap: the + /// histogram says a large blob was written, and the log says who wrote it. + /// + /// `0` disables the log; the histogram is always emitted. + #[arg(long, env = "ROLLOUT_LARGE_BLOB_LOG_BYTES", default_value = "16777216")] + pub rollout_large_blob_log_bytes: usize, + /// Total byte budget for the Lance metadata/index caches, shared across /// **all** resident rollout stores on this instance. /// diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs index f9c7c5af..64ad0903 100644 --- a/crates/lance-context-server/src/routes/rollouts.rs +++ b/crates/lance-context-server/src/routes/rollouts.rs @@ -88,6 +88,45 @@ fn flush_requested(query: Option<&str>) -> bool { }) } +/// Whether a blob of `len` bytes crosses the large-blob log threshold. +/// +/// Split out from [`observe_blob_sizes`] so the boundary is testable without a +/// tracing-capture harness: `0` disables, and the comparison is inclusive so a +/// blob exactly at the threshold logs. +fn should_log_large_blob(len: usize, threshold: usize) -> bool { + threshold > 0 && len >= threshold +} + +/// Record the size of every inline artifact blob in `records`, and warn for any +/// that crosses `ROLLOUT_LARGE_BLOB_LOG_BYTES`. +/// +/// Called once after parsing, where the multipart and JSON paths converge, so +/// both content types are covered by a single site: a record can arrive with +/// `binary_payload` already populated from an inline-JSON body, never passing +/// through the multipart blob loop. +/// +/// The histogram is deliberately unlabelled. Tagging by experiment would be the +/// more useful signal, but a deployment can carry thousands of experiments and +/// each label value is a separate time series; the warn log supplies the +/// identifying detail for the outliers that actually matter instead. +fn observe_blob_sizes(state: &AppState, experiment: &str, records: &[AddRolloutRequest]) { + let threshold = state.rollout_large_blob_log_bytes; + for record in records { + let Some(payload) = record.binary_payload.as_ref() else { + continue; + }; + metrics::histogram!("rollout_blob_bytes").record(payload.len() as f64); + if should_log_large_blob(payload.len(), threshold) { + tracing::warn!( + experiment = %experiment, + record_id = %record.id, + blob_bytes = payload.len(), + "large inline blob written" + ); + } + } +} + /// Reserve `bytes` from the instance's in-flight blob budget (if configured), /// returning the RAII guard to hold for the request's duration. Returns /// `503 Overloaded` when the budget cannot currently admit the request. When no @@ -296,6 +335,8 @@ pub async fn add_rollouts( )); } + observe_blob_sizes(&state, &name, &records); + let store_lock = state.get_or_open_rollout_store(&name).await?; let ids: Vec = records.iter().map(|r| r.id.clone()).collect(); @@ -791,6 +832,68 @@ mod tests { assert_eq!(content_length(&headers), 4096); } + /// The threshold is inclusive and `0` disables the log entirely, so an + /// operator who leaves the knob unset is never spammed. + #[test] + fn large_blob_log_threshold_is_inclusive_and_zero_disables() { + assert!(!should_log_large_blob(1023, 1024)); + assert!(should_log_large_blob(1024, 1024), "threshold is inclusive"); + assert!(should_log_large_blob(4096, 1024)); + // Disabled: not even an enormous blob logs. + assert!(!should_log_large_blob(usize::MAX, 0)); + } + + /// Every inline blob is measured, including one supplied through the + /// inline-JSON path rather than a multipart part. + /// + /// This is why the instrumentation sits after the two parse paths converge + /// instead of inside the multipart blob loop: a JSON body populates + /// `binary_payload` directly and never enters that loop, so instrumenting + /// there would silently miss it. + #[tokio::test] + async fn blob_sizes_are_recorded_for_records_with_payloads() { + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + use metrics_util::MetricKind; + + let dir = TempDir::new().unwrap(); + let state = AppState::new_for_test(dir.path().to_path_buf()).await; + + let record = |id: &str, payload: Option>| AddRolloutRequest { + id: id.to_string(), + binary_payload: payload, + ..Default::default() + }; + let records = vec![ + record("with-blob", Some(vec![0u8; 512])), + record("no-blob", None), + record("bigger-blob", Some(vec![0u8; 2048])), + ]; + + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, || { + observe_blob_sizes(&state, "exp", &records); + }); + + let samples = snapshotter + .snapshot() + .into_vec() + .into_iter() + .find_map(|(key, _, _, value)| { + (key.kind() == MetricKind::Histogram && key.key().name() == "rollout_blob_bytes") + .then_some(value) + }) + .expect("rollout_blob_bytes histogram must be emitted"); + + let DebugValue::Histogram(values) = samples else { + panic!("rollout_blob_bytes must be a histogram"); + }; + let mut observed: Vec = values.into_iter().map(|v| v.into_inner()).collect(); + observed.sort_by(f64::total_cmp); + // The payload-less record contributes nothing: two samples, not three. + assert_eq!(observed, vec![512.0, 2048.0]); + } + async fn rollout_state() -> (Arc, TempDir) { let dir = TempDir::new().unwrap(); let state = Arc::new(AppState::new_for_test(dir.path().to_path_buf()).await); diff --git a/crates/lance-context-server/src/state.rs b/crates/lance-context-server/src/state.rs index dce945e0..19ab7af4 100644 --- a/crates/lance-context-server/src/state.rs +++ b/crates/lance-context-server/src/state.rs @@ -76,6 +76,9 @@ pub struct AppState { /// uploads/downloads. `None` disables the budget (unbounded). See /// [`BlobBudget`]. pub blob_budget: Option>, + /// Byte size at or above which a single inline blob is logged at `WARN`. + /// `0` disables the log. See `Config::rollout_large_blob_log_bytes`. + pub rollout_large_blob_log_bytes: usize, /// Shared Lance cache session attached to every resident rollout store, so /// the process's total metadata/index cache is bounded by one budget rather /// than Lance's default 6 GiB *per store*. `None` restores the per-store @@ -232,6 +235,7 @@ impl AppState { rollout_cleanup_interval_secs: config.rollout_cleanup_interval_secs, rollout_flush_interval_secs: config.rollout_flush_interval_secs, blob_budget, + rollout_large_blob_log_bytes: config.rollout_large_blob_log_bytes, rollout_session, datagen_stores: Mutex::new(LruCache::new(capacity)), datagen_registry: RwLock::new(datagen_registry), @@ -292,6 +296,7 @@ impl AppState { rollout_cleanup_interval_secs: 0, rollout_flush_interval_secs: 0, blob_budget: None, + rollout_large_blob_log_bytes: 16 * 1024 * 1024, rollout_session: build_rollout_session(2 * 1024 * 1024 * 1024), generic_stores: Mutex::new(LruCache::new( std::num::NonZeroUsize::new(DEFAULT_ROLLOUT_CACHE_CAPACITY).unwrap(),