diff --git a/Cargo.lock b/Cargo.lock index b5864fdaba..f36ac0173c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4523,6 +4523,7 @@ dependencies = [ "openshell-core", "openshell-ocsf", "openshell-policy", + "prost", "rand 0.10.2", "russh", "rustix 1.1.4", @@ -7341,6 +7342,7 @@ dependencies = [ "axum", "base64 0.22.1", "bytes", + "flate2", "h2", "http 1.4.0", "http-body 1.0.1", diff --git a/crates/openshell-sandbox/src/main.rs b/crates/openshell-sandbox/src/main.rs index 7108378654..5e2d415ab8 100644 --- a/crates/openshell-sandbox/src/main.rs +++ b/crates/openshell-sandbox/src/main.rs @@ -583,12 +583,15 @@ fn main() -> Result<()> { let log_push_state = if let (Some(sandbox_id), Some(endpoint)) = (&args.sandbox_id, &args.openshell_endpoint) { - let (tx, handle) = openshell_supervisor_process::log_push::spawn_log_push_task( + let (tx, drops, handle) = openshell_supervisor_process::log_push::spawn_log_push_task( endpoint.clone(), sandbox_id.clone(), ); - let layer = - openshell_supervisor_process::log_push::LogPushLayer::new(sandbox_id.clone(), tx); + let layer = openshell_supervisor_process::log_push::LogPushLayer::new( + sandbox_id.clone(), + tx, + drops, + ); Some((layer, handle)) } else { None diff --git a/crates/openshell-server/Cargo.toml b/crates/openshell-server/Cargo.toml index 898eef334d..27d3603b83 100644 --- a/crates/openshell-server/Cargo.toml +++ b/crates/openshell-server/Cargo.toml @@ -38,7 +38,7 @@ socket2 = { workspace = true } nix = { workspace = true } # gRPC -tonic = { workspace = true, features = ["channel", "tls-native-roots"] } +tonic = { workspace = true, features = ["channel", "gzip", "tls-native-roots"] } prost = { workspace = true } prost-reflect = { workspace = true } prost-types = { workspace = true } diff --git a/crates/openshell-server/src/grpc/policy.rs b/crates/openshell-server/src/grpc/policy.rs index 681a4fd128..e5aee7fa6f 100644 --- a/crates/openshell-server/src/grpc/policy.rs +++ b/crates/openshell-server/src/grpc/policy.rs @@ -4088,7 +4088,32 @@ pub(super) async fn handle_push_sandbox_logs( ) .await?; - for log in batch.logs.into_iter().take(100) { + let sandbox_dropped = state + .tracing_log_bus + .sandbox_drop_delta(&batch.sandbox_id, batch.dropped_total); + if sandbox_dropped > 0 { + metrics::counter!("openshell_sandbox_log_push_dropped_total") + .increment(sandbox_dropped); + warn!( + sandbox_id = %batch.sandbox_id, + dropped = sandbox_dropped, + "sandbox dropped log lines before delivery" + ); + } + + let dropped = log_batch_overflow(batch.logs.len()); + if dropped > 0 { + metrics::counter!("openshell_sandbox_log_ingest_dropped_total") + .increment(dropped as u64); + warn!( + sandbox_id = %batch.sandbox_id, + dropped, + cap = MAX_LOG_LINES_PER_BATCH, + "sandbox log batch exceeded the per-batch cap; lines discarded" + ); + } + + for log in batch.logs.into_iter().take(MAX_LOG_LINES_PER_BATCH) { let mut log = log; log.source = "sandbox".to_string(); log.sandbox_id.clone_from(&batch.sandbox_id); @@ -4099,6 +4124,17 @@ pub(super) async fn handle_push_sandbox_logs( Ok(Response::new(PushSandboxLogsResponse {})) } +/// Maximum log lines accepted from a single `PushSandboxLogs` batch. +/// +/// The supervisor flushes at 50 but can carry up to 200 after a reconnect, so +/// this is headroom rather than a limit it hits in normal operation. +const MAX_LOG_LINES_PER_BATCH: usize = 200; + +/// Lines a batch exceeds the per-batch cap by, and so loses. +const fn log_batch_overflow(len: usize) -> usize { + len.saturating_sub(MAX_LOG_LINES_PER_BATCH) +} + async fn ensure_log_stream_sandbox_scope( state: &Arc, principal: &Principal, @@ -15465,6 +15501,16 @@ mod tests { assert_eq!(undo_err.code(), Code::NotFound); } + #[test] + fn log_batch_overflow_counts_only_lines_past_the_cap() { + assert_eq!(log_batch_overflow(0), 0); + assert_eq!(log_batch_overflow(50), 0); + // The supervisor's post-reconnect flush carries up to 200, which must + // fit without loss. + assert_eq!(log_batch_overflow(200), 0); + assert_eq!(log_batch_overflow(250), 50); + } + #[test] fn build_gateway_policy_audit_event_formats_ocsf_config_line() { let message = build_gateway_policy_audit_event( diff --git a/crates/openshell-server/src/multiplex.rs b/crates/openshell-server/src/multiplex.rs index 3ef774e4cb..557b5951ab 100644 --- a/crates/openshell-server/src/multiplex.rs +++ b/crates/openshell-server/src/multiplex.rs @@ -271,7 +271,8 @@ impl MultiplexService { S: AsyncRead + AsyncWrite + Unpin + Send + 'static, { let openshell = OpenShellServer::new(OpenShellService::new(self.state.clone())) - .max_decoding_message_size(MAX_GRPC_DECODE_SIZE); + .max_decoding_message_size(MAX_GRPC_DECODE_SIZE) + .accept_compressed(tonic::codec::CompressionEncoding::Gzip); let openshell = GatewayInterceptorGrpcService::new( openshell, self.state.gateway_interceptors.clone(), diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index 7c81bbe5bb..c59bb3e102 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -24,6 +24,7 @@ pub struct TracingLogBus { struct Inner { per_id: HashMap>, tails: HashMap>, + dropped_totals: HashMap, /// Recently removed sandbox ids, in eviction order. removed: VecDeque, removed_set: HashSet, @@ -42,6 +43,7 @@ impl TracingLogBus { inner: Arc::new(Mutex::new(Inner { per_id: HashMap::new(), tails: HashMap::new(), + dropped_totals: HashMap::new(), removed: VecDeque::new(), removed_set: HashSet::new(), })), @@ -81,6 +83,7 @@ impl TracingLogBus { let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); inner.per_id.remove(sandbox_id); inner.tails.remove(sandbox_id); + inner.dropped_totals.remove(sandbox_id); if inner.removed_set.insert(sandbox_id.to_string()) { inner.removed.push_back(sandbox_id.to_string()); @@ -104,6 +107,22 @@ impl TracingLogBus { .collect() } + /// Return newly reported sandbox-side drops since the previous push. + pub(crate) fn sandbox_drop_delta(&self, sandbox_id: &str, reported_total: u64) -> u64 { + let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); + if inner.removed_set.contains(sandbox_id) { + return 0; + } + + let seen_total = inner + .dropped_totals + .entry(sandbox_id.to_string()) + .or_default(); + let delta = reported_total.saturating_sub(*seen_total); + *seen_total = reported_total; + delta + } + /// Publish a log line from an external source (e.g., sandbox push). /// /// Injects the line into the same broadcast channel and tail buffer @@ -311,6 +330,7 @@ mod tests { // Create entries via subscribe and publish let _rx = bus.subscribe(sandbox_id); bus.publish_external(make_log_event(sandbox_id, "hello")); + assert_eq!(bus.sandbox_drop_delta(sandbox_id, 3), 3); // Verify entries exist assert_eq!(bus.tail(sandbox_id, 10).len(), 1); @@ -320,6 +340,33 @@ mod tests { // Verify entries are gone assert!(bus.tail(sandbox_id, 10).is_empty()); + assert!( + !bus.inner + .lock() + .unwrap() + .dropped_totals + .contains_key(sandbox_id) + ); + } + + #[test] + fn sandbox_drop_totals_survive_stream_reconnects() { + let bus = TracingLogBus::new(); + + assert_eq!(bus.sandbox_drop_delta("sb-reconnect", 0), 0); + assert_eq!(bus.sandbox_drop_delta("sb-reconnect", 5), 5); + assert_eq!( + bus.sandbox_drop_delta("sb-reconnect", 5), + 0, + "reconnecting must not recount the cumulative total" + ); + assert_eq!(bus.sandbox_drop_delta("sb-reconnect", 9), 4); + assert_eq!( + bus.sandbox_drop_delta("sb-reconnect", 2), + 0, + "a supervisor restart must not underflow the counter" + ); + assert_eq!(bus.sandbox_drop_delta("sb-reconnect", 4), 2); } #[test] diff --git a/crates/openshell-supervisor-process/Cargo.toml b/crates/openshell-supervisor-process/Cargo.toml index 2e2120f1d0..50d58a8333 100644 --- a/crates/openshell-supervisor-process/Cargo.toml +++ b/crates/openshell-supervisor-process/Cargo.toml @@ -26,6 +26,7 @@ rand = "0.10" russh = "0.62" serde_json = { workspace = true } sha2 = { workspace = true } +prost = { workspace = true } tokio = { workspace = true } tokio-stream = { workspace = true } tonic = { workspace = true, features = ["channel", "tls-native-roots"] } diff --git a/crates/openshell-supervisor-process/src/log_push.rs b/crates/openshell-supervisor-process/src/log_push.rs index 2651ee147e..2cdef42551 100644 --- a/crates/openshell-supervisor-process/src/log_push.rs +++ b/crates/openshell-supervisor-process/src/log_push.rs @@ -7,6 +7,9 @@ //! channel to a background task. The task batches lines and streams them to //! the server using the `PushSandboxLogs` client-streaming RPC. +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + use openshell_core::grpc_client::CachedOpenShellClient; use openshell_core::proto::{PushSandboxLogsRequest, SandboxLogLine}; use tokio::sync::mpsc; @@ -23,19 +26,82 @@ pub struct LogPushLayer { sandbox_id: String, tx: mpsc::Sender, max_level: tracing::Level, + drops: Arc, } impl LogPushLayer { - pub fn new(sandbox_id: String, tx: mpsc::Sender) -> Self { + pub fn new( + sandbox_id: String, + tx: mpsc::Sender, + drops: Arc, + ) -> Self { let max_level = parse_max_level(std::env::var("OPENSHELL_LOG_PUSH_LEVEL").ok().as_deref()); Self { sandbox_id, tx, max_level, + drops, } } } +/// Counts log lines the sandbox could not deliver to the gateway. +#[derive(Debug, Default)] +pub struct LogPushDrops { + channel_full: AtomicU64, + backoff_overflow: AtomicU64, + handoff_failed: AtomicU64, + oversized: AtomicU64, +} + +impl LogPushDrops { + /// Lines dropped because the layer's channel was full. + #[must_use] + pub fn channel_full(&self) -> u64 { + self.channel_full.load(Ordering::Relaxed) + } + + /// Lines dropped because the reconnect buffer was full. + #[must_use] + pub fn backoff_overflow(&self) -> u64 { + self.backoff_overflow.load(Ordering::Relaxed) + } + + /// Lines dropped when shutdown prevented retrying a failed stream handoff. + #[must_use] + pub fn handoff_failed(&self) -> u64 { + self.handoff_failed.load(Ordering::Relaxed) + } + + /// Total lines dropped on the sandbox-to-gateway hop. + #[must_use] + pub fn total(&self) -> u64 { + self.channel_full() + self.backoff_overflow() + self.handoff_failed() + self.oversized() + } + + /// Lines dropped because one line alone exceeds the gateway's decode limit. + #[must_use] + pub fn oversized(&self) -> u64 { + self.oversized.load(Ordering::Relaxed) + } + + fn record_channel_full(&self) { + self.channel_full.fetch_add(1, Ordering::Relaxed); + } + + fn record_backoff_overflow(&self) { + self.backoff_overflow.fetch_add(1, Ordering::Relaxed); + } + + fn record_handoff_failed(&self, count: u64) { + self.handoff_failed.fetch_add(count, Ordering::Relaxed); + } + + fn record_oversized(&self, count: u64) { + self.oversized.fetch_add(count, Ordering::Relaxed); + } +} + /// Resolve the push level filter, defaulting to `INFO` when unset or unparseable. fn parse_max_level(raw: Option<&str>) -> tracing::Level { raw.and_then(|s| s.parse().ok()) @@ -93,24 +159,128 @@ impl Layer for LogPushLayer { ocsf_json, }; - // Best-effort: drop if the channel is full (don't block tracing). - let _ = self.tx.try_send(log); + // Best-effort: drop if the channel is full (don't block tracing), but + // count the loss so it is visible rather than silent. + if self.tx.try_send(log).is_err() { + self.drops.record_channel_full(); + } } } /// Spawn a background task that batches and pushes log lines to the server. /// -/// Returns the sender half of the channel (for the [`LogPushLayer`]) and the -/// task handle. The task runs until the sender is dropped or the gRPC stream -/// breaks. +/// Returns the channel sender, shared drop counters, and task handle. pub fn spawn_log_push_task( endpoint: String, sandbox_id: String, -) -> (mpsc::Sender, tokio::task::JoinHandle<()>) { +) -> ( + mpsc::Sender, + Arc, + tokio::task::JoinHandle<()>, +) { let (tx, rx) = mpsc::channel::(1024); - let handle = tokio::spawn(run_push_loop(endpoint, sandbox_id, rx)); + let drops = Arc::new(LogPushDrops::default()); + + let handle = tokio::spawn(run_push_loop(endpoint, sandbox_id, rx, Arc::clone(&drops))); + + (tx, drops, handle) +} + +/// Build a push request, stamping the running drop total for gap accounting. +fn push_request( + sandbox_id: &str, + logs: Vec, + drops: &LogPushDrops, +) -> PushSandboxLogsRequest { + PushSandboxLogsRequest { + sandbox_id: sandbox_id.to_string(), + logs, + dropped_total: drops.total(), + } +} + +/// Keep each request clear of the gateway's 1 MiB gRPC decode cap. +/// +/// Exceeding the cap does not reject one request: it kills the stream, and the +/// reconnect loop would resend the same batch forever. +const MAX_PUSH_REQUEST_BYTES: usize = 512 * 1024; - (tx, handle) +/// Bytes a request costs before any log lines. +const ENVELOPE_BYTES: usize = 64; + +/// Tag and length prefix prost writes per repeated `logs` entry. +const FIELD_OVERHEAD: usize = 8; + +/// Group `logs` so each group encodes below `limit`. +/// +/// Returns the groups and the count of lines no request could carry. +fn split_to_fit(logs: Vec, limit: usize) -> (Vec>, u64) { + let mut groups: Vec> = Vec::new(); + let mut current: Vec = Vec::new(); + let mut current_len = ENVELOPE_BYTES; + let mut oversized = 0; + + for line in logs { + let line_len = prost::Message::encoded_len(&line) + FIELD_OVERHEAD; + if ENVELOPE_BYTES + line_len > limit { + oversized += 1; + continue; + } + if !current.is_empty() && current_len + line_len > limit { + groups.push(std::mem::take(&mut current)); + current_len = ENVELOPE_BYTES; + } + current_len += line_len; + current.push(line); + } + if !current.is_empty() { + groups.push(current); + } + + (groups, oversized) +} + +/// Send `logs`, splitting to fit and counting anything undeliverable. +/// +/// Returns the rejected group and any groups not yet handed off when the stream +/// is gone, preserving their original order for retry after reconnecting. +async fn send_logs( + push_tx: &mpsc::Sender, + sandbox_id: &str, + logs: Vec, + drops: &LogPushDrops, +) -> Result<(), Vec> { + let (mut groups, oversized) = split_to_fit(logs, MAX_PUSH_REQUEST_BYTES); + if oversized > 0 { + drops.record_oversized(oversized); + eprintln!("openshell: dropped {oversized} log line(s) too large to deliver"); + if groups.is_empty() { + // Report the loss even when the batch has no deliverable line. + groups.push(Vec::new()); + } + } + let mut groups = groups.into_iter(); + while let Some(group) = groups.next() { + if let Err(rejected) = push_tx.send(push_request(sandbox_id, group, drops)).await { + let mut unsent = rejected.0.logs; + unsent.extend(groups.flatten()); + return Err(unsent); + } + } + Ok(()) +} + +/// Account for lines that cannot be retried because the log source closed. +/// +/// A rejected synthetic request used only to report oversized drops contains +/// no lines, so it must not produce a second, zero-line drop report. +fn record_shutdown_handoff_failure(unsent: &[SandboxLogLine], drops: &LogPushDrops) -> Option { + let count = u64::try_from(unsent.len()).unwrap_or(u64::MAX); + if count == 0 { + return None; + } + drops.record_handoff_failed(count); + Some(count) } /// Maximum backoff delay between reconnection attempts. @@ -118,10 +288,36 @@ const MAX_BACKOFF: tokio::time::Duration = tokio::time::Duration::from_secs(30); /// Initial backoff delay after a connection failure. const INITIAL_BACKOFF: tokio::time::Duration = tokio::time::Duration::from_secs(1); +/// Request compression for sandbox log pushes. +/// +/// Keep this disabled until all supported gateways accept gzip. Enabling it in +/// the same release as server-side acceptance breaks newer supervisors that +/// connect to an older gateway or an older replica during a rolling upgrade. +const fn log_push_request_compression() -> Option { + None +} + +/// Observe why an immediately rejected handoff ended, then drain new lines +/// during the reconnect delay unless authentication made retrying futile. +async fn back_off_after_handoff_failure( + rpc_done_rx: &mut mpsc::Receiver, + rx: &mut mpsc::Receiver, + batch: &mut Vec, + delay: tokio::time::Duration, + drops: &LogPushDrops, +) -> bool { + let fatal_auth = rpc_done_rx.recv().await.unwrap_or(false); + if !fatal_auth { + drain_during_backoff(rx, batch, delay, drops).await; + } + fatal_auth +} + async fn run_push_loop( endpoint: String, sandbox_id: String, mut rx: mpsc::Receiver, + drops: Arc, ) { let mut batch = Vec::with_capacity(50); let mut backoff = INITIAL_BACKOFF; @@ -143,8 +339,8 @@ async fn run_push_loop( Err(e) => { eprintln!("openshell: log push connect failed: {e}"); // Drain the channel during backoff so the tracing layer doesn't - // block, but discard lines we can't deliver. - drain_during_backoff(&mut rx, &mut batch, backoff).await; + // fill while retaining the newest lines up to the buffer cap. + drain_during_backoff(&mut rx, &mut batch, backoff, &drops).await; backoff = (backoff * 2).min(MAX_BACKOFF); continue; } @@ -159,6 +355,9 @@ async fn run_push_loop( let (rpc_done_tx, mut rpc_done_rx) = mpsc::channel::(1); tokio::spawn({ let mut nav_client = client.raw_client(); + if let Some(encoding) = log_push_request_compression() { + nav_client = nav_client.send_compressed(encoding); + } async move { let fatal_auth = match nav_client.push_sandbox_logs(stream).await { Ok(_) => false, @@ -175,16 +374,24 @@ async fn run_push_loop( // --- Flush any lines buffered during reconnect --- if !batch.is_empty() { let lines = std::mem::take(&mut batch); - if push_tx - .send(PushSandboxLogsRequest { - sandbox_id: sandbox_id.clone(), - logs: lines, - }) + if let Err(unsent) = send_logs(&push_tx, &sandbox_id, lines, &drops).await { + // RPC died immediately. Retain the rejected requests, observe + // fatal authentication, and drain during backoff before retry. + batch = unsent; + if back_off_after_handoff_failure( + &mut rpc_done_rx, + &mut rx, + &mut batch, + backoff, + &drops, + ) .await - .is_err() - { - // RPC died immediately — go back to reconnect. - backoff = INITIAL_BACKOFF; + { + eprintln!("openshell: log push disabled after authentication failure"); + return; + } + eprintln!("openshell: log push stream lost, reconnecting after backoff..."); + backoff = (backoff * 2).min(MAX_BACKOFF); continue; } } @@ -203,20 +410,21 @@ async fn run_push_loop( // Flush remaining and exit entirely. if !batch.is_empty() { let lines = std::mem::take(&mut batch); - let _ = push_tx.send(PushSandboxLogsRequest { - sandbox_id: sandbox_id.clone(), - logs: lines, - }).await; + if let Err(unsent) = send_logs(&push_tx, &sandbox_id, lines, &drops).await + && let Some(count) = record_shutdown_handoff_failure(&unsent, &drops) + { + eprintln!( + "openshell: dropped {count} log line(s) after stream handoff failed during shutdown" + ); + } } return; }; batch.push(line); if batch.len() >= 50 { let lines = std::mem::take(&mut batch); - if push_tx.send(PushSandboxLogsRequest { - sandbox_id: sandbox_id.clone(), - logs: lines, - }).await.is_err() { + if let Err(unsent) = send_logs(&push_tx, &sandbox_id, lines, &drops).await { + batch = unsent; break true; } } @@ -224,10 +432,8 @@ async fn run_push_loop( _ = timer.tick() => { if !batch.is_empty() { let lines = std::mem::take(&mut batch); - if push_tx.send(PushSandboxLogsRequest { - sandbox_id: sandbox_id.clone(), - logs: lines, - }).await.is_err() { + if let Err(unsent) = send_logs(&push_tx, &sandbox_id, lines, &drops).await { + batch = unsent; break true; } } @@ -247,7 +453,7 @@ async fn run_push_loop( if stream_broken { eprintln!("openshell: log push stream lost, reconnecting after backoff..."); - drain_during_backoff(&mut rx, &mut batch, backoff).await; + drain_during_backoff(&mut rx, &mut batch, backoff, &drops).await; backoff = (backoff * 2).min(MAX_BACKOFF); } } @@ -260,6 +466,7 @@ async fn drain_during_backoff( rx: &mut mpsc::Receiver, batch: &mut Vec, delay: tokio::time::Duration, + drops: &LogPushDrops, ) { // Keep at most 200 lines across reconnect attempts to bound memory. const MAX_BUFFERED: usize = 200; @@ -271,10 +478,13 @@ async fn drain_during_backoff( line = rx.recv() => { match line { Some(l) => { - if batch.len() < MAX_BUFFERED { - batch.push(l); + if batch.len() >= MAX_BUFFERED { + // Prefer current sandbox activity over the oldest + // retry while keeping loss explicitly accounted. + batch.remove(0); + drops.record_backoff_overflow(); } - // else: drop — we're over the reconnect buffer limit + batch.push(l); } None => return, // channel closed, sandbox shutting down } @@ -347,6 +557,7 @@ mod tests { sandbox_id: "sb-test".to_string(), tx, max_level: tracing::Level::INFO, + drops: Arc::new(LogPushDrops::default()), }; let subscriber = tracing_subscriber::registry().with(layer); tracing::subscriber::with_default(subscriber, f); @@ -409,6 +620,195 @@ mod tests { ); } + #[test] + fn a_batch_that_fits_is_sent_as_one_request() { + let logs = vec![test_line("a"), test_line("b")]; + + let (groups, oversized) = split_to_fit(logs, MAX_PUSH_REQUEST_BYTES); + + assert_eq!(groups.len(), 1); + assert_eq!(groups[0].len(), 2); + assert_eq!(oversized, 0); + } + + #[test] + fn an_empty_batch_produces_no_requests() { + let (groups, oversized) = split_to_fit(Vec::new(), MAX_PUSH_REQUEST_BYTES); + + assert!(groups.is_empty()); + assert_eq!(oversized, 0); + } + + #[test] + fn an_oversized_batch_is_split_rather_than_dropped() { + // Each line carries a 1 KiB payload; a 4 KiB limit cannot hold all six. + let logs: Vec = (0..6).map(|i| padded_line(i, 1024)).collect(); + + let (groups, oversized) = split_to_fit(logs, 4096); + + assert!(groups.len() > 1, "the batch should have been split"); + assert_eq!(oversized, 0, "nothing should be lost to a split"); + let total: usize = groups.iter().map(Vec::len).sum(); + assert_eq!(total, 6); + } + + #[test] + fn every_group_encodes_below_the_limit() { + let logs: Vec = (0..20).map(|i| padded_line(i, 1024)).collect(); + + let (groups, _) = split_to_fit(logs, 4096); + + for group in &groups { + let request = push_request("sb", group.clone(), &LogPushDrops::default()); + let encoded = prost::Message::encoded_len(&request); + assert!( + encoded <= 4096, + "a group encoded to {encoded} bytes, over the 4096 limit" + ); + } + } + + #[test] + fn splitting_preserves_line_order() { + let logs: Vec = (0..6).map(|i| padded_line(i, 1024)).collect(); + + let (groups, _) = split_to_fit(logs, 4096); + + let messages: Vec = groups + .into_iter() + .flatten() + .map(|line| line.message) + .collect(); + let expected: Vec = (0..6).map(|i| padded_line(i, 1024).message).collect(); + assert_eq!(messages, expected); + } + + #[test] + fn a_single_line_too_large_to_send_is_dropped_and_counted() { + let logs = vec![test_line("small"), padded_line(1, 8192)]; + + let (groups, oversized) = split_to_fit(logs, 4096); + + assert_eq!(oversized, 1); + let kept: Vec = groups + .into_iter() + .flatten() + .map(|line| line.message) + .collect(); + assert_eq!(kept, vec!["small"], "the deliverable line still goes"); + } + + #[tokio::test] + async fn an_oversized_only_batch_reports_its_drop() { + let (push_tx, mut push_rx) = mpsc::channel(1); + let drops = LogPushDrops::default(); + + send_logs( + &push_tx, + "sb", + vec![padded_line(1, MAX_PUSH_REQUEST_BYTES)], + &drops, + ) + .await + .expect("drop report should be handed off"); + + let request = push_rx + .try_recv() + .expect("the gateway should receive the updated drop total"); + assert!(request.logs.is_empty()); + assert_eq!(request.dropped_total, 1); + } + + #[tokio::test] + async fn a_rejected_request_and_remaining_split_groups_are_returned_for_retry() { + let (push_tx, mut push_rx) = mpsc::channel(1); + let logs: Vec = (0..3) + .map(|i| padded_line(i, MAX_PUSH_REQUEST_BYTES / 2 + 1)) + .collect(); + + let send = + tokio::spawn( + async move { send_logs(&push_tx, "sb", logs, &LogPushDrops::default()).await }, + ); + + let accepted = push_rx + .recv() + .await + .expect("first group should be accepted"); + assert_eq!(accepted.logs.len(), 1); + assert_eq!( + accepted.logs[0].message, + padded_line(0, MAX_PUSH_REQUEST_BYTES / 2 + 1).message + ); + drop(push_rx); + + let unsent = send + .await + .expect("send task should finish") + .expect_err("closed request channel should reject the next group"); + let messages: Vec<_> = unsent.into_iter().map(|line| line.message).collect(); + assert_eq!( + messages, + vec![ + padded_line(1, MAX_PUSH_REQUEST_BYTES / 2 + 1).message, + padded_line(2, MAX_PUSH_REQUEST_BYTES / 2 + 1).message, + ] + ); + } + + #[tokio::test] + async fn a_rejected_oversized_drop_report_is_not_reported_as_a_zero_line_drop() { + let (push_tx, push_rx) = mpsc::channel(1); + drop(push_rx); + let drops = LogPushDrops::default(); + let unsent = send_logs( + &push_tx, + "sb", + vec![padded_line(1, MAX_PUSH_REQUEST_BYTES)], + &drops, + ) + .await + .expect_err("closed request channel should reject the drop report"); + + assert!(unsent.is_empty()); + assert_eq!(record_shutdown_handoff_failure(&unsent, &drops), None); + assert_eq!(drops.handoff_failed(), 0); + assert_eq!(drops.oversized(), 1); + } + + #[test] + fn oversized_drops_join_the_running_drop_total() { + let drops = LogPushDrops::default(); + drops.record_oversized(2); + + assert_eq!(drops.oversized(), 2); + assert_eq!(drops.total(), 2); + } + + #[test] + fn push_requests_report_the_running_drop_total() { + let drops = LogPushDrops::default(); + assert_eq!(push_request("sb", Vec::new(), &drops).dropped_total, 0); + + drops.record_channel_full(); + drops.record_backoff_overflow(); + drops.record_backoff_overflow(); + drops.record_handoff_failed(4); + + let request = push_request("sb", vec![test_line("x")], &drops); + assert_eq!(request.dropped_total, 7); + assert_eq!(request.sandbox_id, "sb"); + assert_eq!(request.logs.len(), 1); + } + + #[test] + fn log_push_requests_remain_uncompressed_for_legacy_gateways() { + assert!( + log_push_request_compression().is_none(), + "supervisors must not require gzip until every supported gateway accepts it" + ); + } + #[test] fn non_ocsf_lines_carry_no_ocsf_payload() { let lines = capture(16, || { @@ -498,6 +898,55 @@ mod tests { assert_eq!(lines[1].message, "line 1"); } + #[test] + fn lines_dropped_on_a_full_channel_are_counted() { + use tracing_subscriber::layer::SubscriberExt; + + let (tx, _rx) = mpsc::channel::(2); + let drops = Arc::new(LogPushDrops::default()); + let layer = LogPushLayer { + sandbox_id: "sb-test".to_string(), + tx, + max_level: tracing::Level::INFO, + drops: Arc::clone(&drops), + }; + let subscriber = tracing_subscriber::registry().with(layer); + + tracing::subscriber::with_default(subscriber, || { + for i in 0..5 { + tracing::info!(target: "test_target", "line {i}"); + } + }); + + // Two fit in the channel; the other three are dropped and counted, so + // the loss is a countable gap rather than silence. + assert_eq!(drops.channel_full(), 3); + assert_eq!(drops.backoff_overflow(), 0); + } + + #[tokio::test] + async fn lines_dropped_during_backoff_are_counted() { + let (tx, mut rx) = mpsc::channel::(1024); + for i in 0..250 { + tx.try_send(test_line(&format!("line {i}"))).unwrap(); + } + drop(tx); + + let drops = LogPushDrops::default(); + let mut batch = Vec::new(); + drain_during_backoff( + &mut rx, + &mut batch, + tokio::time::Duration::from_secs(30), + &drops, + ) + .await; + + assert_eq!(batch.len(), 200); + assert_eq!(drops.backoff_overflow(), 50); + assert_eq!(drops.channel_full(), 0); + } + #[test] fn parse_max_level_defaults_to_info() { assert_eq!(parse_max_level(None), tracing::Level::INFO); @@ -520,8 +969,15 @@ mod tests { } } + /// A line whose message is `pad` bytes long, for size-limit tests. + fn padded_line(index: usize, pad: usize) -> SandboxLogLine { + let mut line = test_line(&format!("line-{index}")); + line.message = format!("line-{index}-{}", "x".repeat(pad)); + line + } + #[tokio::test] - async fn drain_during_backoff_buffers_up_to_the_cap_and_drops_the_rest() { + async fn drain_during_backoff_keeps_the_newest_lines_at_the_cap() { let (tx, mut rx) = mpsc::channel::(1024); for i in 0..250 { tx.try_send(test_line(&format!("line {i}"))).unwrap(); @@ -529,11 +985,99 @@ mod tests { drop(tx); let mut batch = Vec::new(); - drain_during_backoff(&mut rx, &mut batch, tokio::time::Duration::from_secs(30)).await; + let drops = LogPushDrops::default(); + drain_during_backoff( + &mut rx, + &mut batch, + tokio::time::Duration::from_secs(30), + &drops, + ) + .await; assert_eq!(batch.len(), 200); - assert_eq!(batch[0].message, "line 0"); - assert_eq!(batch[199].message, "line 199"); + assert_eq!(batch[0].message, "line 50"); + assert_eq!(batch[199].message, "line 249"); + assert_eq!(drops.backoff_overflow(), 50); + } + + #[tokio::test] + async fn immediate_handoff_failure_observes_auth_failure_without_draining() { + let (rpc_done_tx, mut rpc_done_rx) = mpsc::channel(1); + rpc_done_tx.send(true).await.unwrap(); + drop(rpc_done_tx); + + let (tx, mut rx) = mpsc::channel(1); + tx.send(test_line("fresh")).await.unwrap(); + drop(tx); + + let drops = LogPushDrops::default(); + let mut batch = vec![test_line("retry")]; + let fatal_auth = back_off_after_handoff_failure( + &mut rpc_done_rx, + &mut rx, + &mut batch, + tokio::time::Duration::from_secs(30), + &drops, + ) + .await; + + assert!(fatal_auth); + assert_eq!(rx.try_recv().unwrap().message, "fresh"); + } + + #[tokio::test] + async fn immediate_handoff_failure_drains_new_lines_during_backoff() { + let (rpc_done_tx, mut rpc_done_rx) = mpsc::channel(1); + rpc_done_tx.send(false).await.unwrap(); + drop(rpc_done_tx); + + let (tx, mut rx) = mpsc::channel(1); + tx.send(test_line("fresh")).await.unwrap(); + drop(tx); + + let drops = LogPushDrops::default(); + let mut batch = vec![test_line("retry")]; + let fatal_auth = back_off_after_handoff_failure( + &mut rpc_done_rx, + &mut rx, + &mut batch, + tokio::time::Duration::from_secs(30), + &drops, + ) + .await; + + assert!(!fatal_auth); + assert_eq!(batch.len(), 2); + assert_eq!(batch[0].message, "retry"); + assert_eq!(batch[1].message, "fresh"); + } + + #[tokio::test] + async fn immediate_handoff_failure_waits_for_the_backoff_deadline() { + let (rpc_done_tx, mut rpc_done_rx) = mpsc::channel(1); + rpc_done_tx.send(false).await.unwrap(); + drop(rpc_done_tx); + + let (_input_tx, mut rx) = mpsc::channel(1); + let drops = LogPushDrops::default(); + let mut batch = vec![test_line("retry")]; + + let result = tokio::time::timeout( + tokio::time::Duration::from_millis(10), + back_off_after_handoff_failure( + &mut rpc_done_rx, + &mut rx, + &mut batch, + tokio::time::Duration::from_secs(1), + &drops, + ), + ) + .await; + + assert!( + result.is_err(), + "retry must not spin before backoff elapses" + ); } #[tokio::test] @@ -543,7 +1087,14 @@ mod tests { drop(tx); let mut batch = vec![test_line("buffered")]; - drain_during_backoff(&mut rx, &mut batch, tokio::time::Duration::from_secs(30)).await; + let drops = LogPushDrops::default(); + drain_during_backoff( + &mut rx, + &mut batch, + tokio::time::Duration::from_secs(30), + &drops, + ) + .await; assert_eq!(batch.len(), 2); assert_eq!(batch[0].message, "buffered"); @@ -557,9 +1108,15 @@ mod tests { drop(tx); let mut batch = Vec::new(); + let drops = LogPushDrops::default(); tokio::time::timeout( tokio::time::Duration::from_secs(5), - drain_during_backoff(&mut rx, &mut batch, tokio::time::Duration::from_secs(30)), + drain_during_backoff( + &mut rx, + &mut batch, + tokio::time::Duration::from_secs(30), + &drops, + ), ) .await .expect("closed channel should end the backoff drain"); diff --git a/proto/openshell.proto b/proto/openshell.proto index 272ee85893..7016599e12 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -2335,6 +2335,8 @@ message PushSandboxLogsRequest { string sandbox_id = 1; // Log lines to ingest. repeated SandboxLogLine logs = 2; + // Cumulative count of lines the sandbox dropped, for gap accounting. + uint64 dropped_total = 3; } // Push sandbox logs response. diff --git a/sdk/go/proto/openshellv1/openshell.pb.go b/sdk/go/proto/openshellv1/openshell.pb.go index 3e16ec0e82..4831979d25 100644 --- a/sdk/go/proto/openshellv1/openshell.pb.go +++ b/sdk/go/proto/openshellv1/openshell.pb.go @@ -10192,7 +10192,9 @@ type PushSandboxLogsRequest struct { // The sandbox ID. SandboxId string `protobuf:"bytes,1,opt,name=sandbox_id,json=sandboxId,proto3" json:"sandbox_id,omitempty"` // Log lines to ingest. - Logs []*SandboxLogLine `protobuf:"bytes,2,rep,name=logs,proto3" json:"logs,omitempty"` + Logs []*SandboxLogLine `protobuf:"bytes,2,rep,name=logs,proto3" json:"logs,omitempty"` + // Cumulative count of lines the sandbox dropped, for gap accounting. + DroppedTotal uint64 `protobuf:"varint,3,opt,name=dropped_total,json=droppedTotal,proto3" json:"dropped_total,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -10241,6 +10243,13 @@ func (x *PushSandboxLogsRequest) GetLogs() []*SandboxLogLine { return nil } +func (x *PushSandboxLogsRequest) GetDroppedTotal() uint64 { + if x != nil { + return x.DroppedTotal + } + return 0 +} + // Push sandbox logs response. type PushSandboxLogsResponse struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -15639,11 +15648,12 @@ const file_openshell_proto_rawDesc = "" + "\bsince_ms\x18\x03 \x01(\x03R\asinceMs\x12\x18\n" + "\asources\x18\x04 \x03(\tR\asources\x12\x1b\n" + "\tmin_level\x18\x05 \x01(\tR\bminLevel\x12\x1c\n" + - "\tworkspace\x18\x06 \x01(\tR\tworkspace\"i\n" + + "\tworkspace\x18\x06 \x01(\tR\tworkspace\"\x8e\x01\n" + "\x16PushSandboxLogsRequest\x12\x1d\n" + "\n" + "sandbox_id\x18\x01 \x01(\tR\tsandboxId\x120\n" + - "\x04logs\x18\x02 \x03(\v2\x1c.openshell.v1.SandboxLogLineR\x04logs\"\x19\n" + + "\x04logs\x18\x02 \x03(\v2\x1c.openshell.v1.SandboxLogLineR\x04logs\x12#\n" + + "\rdropped_total\x18\x03 \x01(\x04R\fdroppedTotal\"\x19\n" + "\x17PushSandboxLogsResponse\"m\n" + "\x16GetSandboxLogsResponse\x120\n" + "\x04logs\x18\x01 \x03(\v2\x1c.openshell.v1.SandboxLogLineR\x04logs\x12!\n" +