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(),