diff --git a/SHA256SUMS b/SHA256SUMS index 708e3ae..1e07b47 100644 --- a/SHA256SUMS +++ b/SHA256SUMS @@ -346,10 +346,10 @@ a31db7eac12b578b1eb5a74fae98bbe36312da95b7d7173135dbc41fc454da21 src/http3/test 68fb5ab6a889995a66ad12cb6acf88c5d04f0aca572e2ae15f1a9cd171a5f24b src/lib.rs 2551189a3e5701371026c9abd2c5d44d37495f28fb67e220180421bfd315e3dd src/logging.rs 21ddb01b8f9c7fab60840d18103ef8de4716ff0c890cc1525b472b69af7916ec src/logging/error_runtime.rs -42b88cefef603fafa9eec3ba83a53069f01f571e98cf923210a6fe605e008e90 src/logging/format.rs +add1293410e87ddae85ba4b8bec57b56ff54415b8112576c450f85920ab6c38c src/logging/format.rs f5815d778cc1785723faed4abb15f9880f22b6524b5289fe1e7158905aac0986 src/logging/model.rs -b90ee5c7badacd0141fba99a8109fced785a11abe11675c8687fa35e68f5e7b1 src/logging/runtime.rs -c83a7b84a9d7226c200394c351c4b3d9c417869828c01a35d4b1297670ab50f2 src/logging/tests.rs +b9bef61d332df0dec47f15d864a7aaea240935c432bf935e40144916e37553bc src/logging/runtime.rs +106f769f45a832079b170e1cf11d7e6a78adba354bb0a88bfb8e3aff14f8cce6 src/logging/tests.rs 3836ed14b6710e183f048ee4d7807a35366c27a27338057fe4a2811129b63302 src/main.rs fc8e99eac83c083d65101f6402aacbcc8392d66508c31566a50de3ebcb12e7da src/main/admin_runtime.rs 42a9197475823f67c1ca762951ede5d7a2f10124bd6ed4d1a3365f4297be894c src/main/bootstrap.rs @@ -400,7 +400,7 @@ b9a2307c6aa7f16613d935508a61574dfee964020b1b5be86293df2d684615d8 src/proxy/requ 0f2eacadf1447db8aa1c1a78005aa262e0f747096bfe7e0a0e41a2cd31a0d17f src/proxy/tests/h3_navigation_tests.rs d134b0eb617f82f9971b3f3461379ea37c57bd0079bf54748d0c434b2ef1b475 src/proxy/tests/http_invalidation_tests.rs a74815a0b46d1c94ffbf8cc3a8bbbc967daa4aec0981139d2d92624c10553026 src/proxy/types.rs -b45cf5baf45e37915c8cfac43e845976f874f0aa5474f167fa943c0f38decade src/proxy/upstream_helpers.rs +1db11efc7c35b8654f88f3176d9bca341c3e625d055e6c7e888f953c04b5f573 src/proxy/upstream_helpers.rs 2d9142e3ae95895b5271b4009ab48236a3a2074ce51179c021193b35777560e4 src/proxy/web_cache_revalidation.rs d7b4b809b12ee89293df45e9a64b1c9981fb3f89deaab3762f4a3ad3992dd7b1 src/proxy/web_cache_revalidation/tests.rs 69f775f6746b4c6d729dc10961dbef54168846f4ce9cf7fe453ab5a2b879c9b8 src/proxy/websocket.rs diff --git a/src/logging/format.rs b/src/logging/format.rs index f958b47..6832861 100644 --- a/src/logging/format.rs +++ b/src/logging/format.rs @@ -49,12 +49,28 @@ pub struct SinkStats { pub errors: u64, } +// History owns entries only when there is no external sink. With a sink, +// both the queue and history share one immutable record instead of copying +// every allocated string on each request. +enum HistoryEntry { + Owned(LogEntry), + Shared(Arc), +} +impl HistoryEntry { + fn entry(&self) -> &LogEntry { + match self { + Self::Owned(value) => value, + Self::Shared(value) => value.as_ref(), + } + } +} + pub struct Logs { - entries: Mutex>, + entries: Mutex>, pub dropped: AtomicU64, settings: Arc>, live: broadcast::Sender, - sink: mpsc::Sender, + sink: mpsc::Sender>, sink_dropped: Arc, sink_sent: Arc, sink_errors: Arc, @@ -135,12 +151,25 @@ impl Logs { pub fn capture_country(&self) -> bool { self.settings.load().capture_country } + fn retain(&self, entry: HistoryEntry) { + let evicted = if let Ok(mut entries) = self.entries.try_lock() { + let evicted = if entries.len() == HISTORY_CAPACITY { + entries.pop_front() + } else { + None + }; + entries.push_back(entry); + evicted + } else { + self.dropped.fetch_add(1, Ordering::Relaxed); + None + }; + // Evicted record destruction must happen outside the history mutex. + drop(evicted); + } pub fn push(&self, entry: LogEntry) { let has_viewers = self.live.receiver_count() != 0; - let has_sink = self.settings.load().enabled(); - // Reserve the bounded sink slot BEFORE cloning a log entry. Under - // saturation, the failed enqueue must not copy 30+ owned fields. - let sink_permit = if has_sink { + let sink_permit = if self.settings.load().enabled() { match self.sink.try_reserve() { Ok(permit) => Some(permit), Err(_) => { @@ -151,38 +180,31 @@ impl Logs { } else { None }; - let mut external = Some(entry); - let history_entry = if has_viewers || sink_permit.is_some() { - external.as_ref().expect("entry").clone() - } else { - external.take().expect("entry") - }; - let evicted = if let Ok(mut entries) = self.entries.try_lock() { - let evicted = if entries.len() == HISTORY_CAPACITY { - entries.pop_front() - } else { - None - }; - entries.push_back(history_entry); - evicted - } else { - self.dropped.fetch_add(1, Ordering::Relaxed); - None - }; - // Destroy evicted strings outside of the history lock. - drop(evicted); - if let Some(entry) = external { - if has_viewers { - let _ = self.live.send(entry.clone()); + if let Some(permit) = sink_permit { + // One Arc allocation replaces a full LogEntry clone: no per-field + // String allocations for the active file/network logging path. + let shared = Arc::new(entry); + let live = has_viewers.then(|| shared.as_ref().clone()); + self.retain(HistoryEntry::Shared(shared.clone())); + if let Some(live) = live { + let _ = self.live.send(live); } - if let Some(permit) = sink_permit { - permit.send(entry); + permit.send(shared); + } else { + // History-only operation retains the original no-Arc/no-copy path. + let live = has_viewers.then(|| entry.clone()); + self.retain(HistoryEntry::Owned(entry)); + if let Some(live) = live { + let _ = self.live.send(live); } } } fn push_sink(&self, entry: LogEntry) { - if self.settings.load().enabled() && self.sink.try_send(entry).is_err() { - self.sink_dropped.fetch_add(1, Ordering::Relaxed); + if self.settings.load().enabled() { + match self.sink.try_reserve() { + Ok(permit) => permit.send(Arc::new(entry)), + Err(_) => { self.sink_dropped.fetch_add(1, Ordering::Relaxed); } + } } } /// Persist transport/protocol failures without polluting the bounded Live Logs buffer. @@ -281,7 +303,7 @@ impl Logs { .iter() .rev() .take(limit.min(HISTORY_CAPACITY)) - .cloned() + .map(|item| item.entry().clone()) .collect() }) .unwrap_or_default() diff --git a/src/logging/runtime.rs b/src/logging/runtime.rs index a68a1a6..cece702 100644 --- a/src/logging/runtime.rs +++ b/src/logging/runtime.rs @@ -41,6 +41,63 @@ fn format_entry(entry: &LogEntry, format: &JsonFormat) -> Value { Value::Object(object) } +// A request with ordinary values cannot match any redaction rule. Every +// supported rule (including private PEM and escaped JSON) contains one of +// these case-insensitive ASCII markers. If any marker is present, fall back +// to the original full recursive redaction, even for a false positive. +fn may_require_redaction(text: &str) -> bool { + let data = text.as_bytes(); + for (i, byte) in data.iter().enumerate() { + let marker: &[u8] = match byte.to_ascii_lowercase() { + b'k' => b"key", + b'a' => b"auth", + b'p' => b"pass", + b't' => b"token", + b's' => b"secret", + b'c' => b"credential", + _ => continue, + }; + if data.get(i..i + marker.len()).is_some_and(|value| value.eq_ignore_ascii_case(marker)) { + return true; + } + } + false +} + +fn entry_requires_redaction(entry: &LogEntry) -> bool { + [ + entry.event, entry.protocol_error.unwrap_or(""), entry.reason.unwrap_or(""), + entry.error_detail.as_deref().unwrap_or(""), entry.error.as_deref().unwrap_or(""), + entry.local_addr.as_deref().unwrap_or(""), entry.tls_sni.as_deref().unwrap_or(""), + entry.sni.as_deref().unwrap_or(""), entry.authority.as_deref().unwrap_or(""), + entry.matched_by.unwrap_or(""), entry.vhost.as_deref().unwrap_or(""), + entry.route.as_deref().unwrap_or(""), entry.backend.as_deref().unwrap_or(""), + &entry.request_id, &entry.method, &entry.host, &entry.path, + entry.http_version, entry.transport, entry.websocket_bootstrap.unwrap_or(""), + entry.websocket_phase.unwrap_or(""), entry.websocket_error.unwrap_or(""), + entry.h3_reset_code.unwrap_or(""), entry.upstream_error.as_deref().unwrap_or(""), + entry.client_ip.as_deref().unwrap_or(""), entry.user_agent.as_deref().unwrap_or(""), + entry.country.as_deref().unwrap_or(""), entry.outcome, entry.cache, entry.cache_reason, + ].iter().any(|value| may_require_redaction(value)) +} + +// Serialize into the existing batch buffer: no temporary Vec per record. +// The ordinary default format also skips the intermediate serde_json::Value +// tree and its many field allocations. Sensitive/custom records retain the +// exact previous formatting and recursive redaction behavior. +fn append_jsonl(entry: &LogEntry, format: &JsonFormat, buffer: &mut Vec) { + if format.fields.is_empty() && format.static_fields.is_empty() + && !entry_requires_redaction(entry) + { + serde_json::to_writer(&mut *buffer, entry).expect("LogEntry JSON serialization"); + } else { + let mut payload = format_entry(entry, format); + redact_value(&mut payload); + serde_json::to_writer(&mut *buffer, &payload).expect("log Value JSON serialization"); + } + buffer.push(b'\n'); +} + // Bound both write size and event count; never wait for the batch to fill. const FILE_BATCH_EVENTS: u64 = 64; const FILE_BATCH_BYTES: usize = 128 * 1024; @@ -107,7 +164,7 @@ async fn open_access_file(path: PathBuf) -> Result { Ok(File::from_std(file)) } async fn sink_loop( - mut rx: mpsc::Receiver, + mut rx: mpsc::Receiver>, settings: Arc>, sent: Arc, errors: Arc, @@ -118,7 +175,7 @@ async fn sink_loop( let mut vhost_files: BTreeMap = BTreeMap::new(); let mut vhost_directory: Option = None; let mut logstash: Option = None; - let mut deferred_entry: Option = None; + let mut deferred_entry: Option> = None; let mut file_batch = Vec::with_capacity(8 * 1024); loop { let entry = if let Some(entry) = deferred_entry.take() { @@ -152,11 +209,7 @@ async fn sink_loop( let mut count = 0u64; let mut next = Some(entry); while let Some(item) = next.take() { - let mut payload = format_entry(&item, &config.format); - redact_value(&mut payload); - let mut line = serde_json::to_vec(&payload).unwrap_or_default(); - line.push(b'\n'); - file_batch.extend_from_slice(&line); + append_jsonl(&item, &config.format, &mut file_batch); count += 1; if count >= FILE_BATCH_EVENTS || file_batch.len() >= FILE_BATCH_BYTES { break; @@ -186,14 +239,11 @@ async fn sink_loop( } continue; } - let mut payload = format_entry(&entry, &config.format); - redact_value(&mut payload); - // File and Logstash consume the same JSONL record. Serialize once and - // keep no JSONL buffer at all for Loki-only configurations. + // File/Logstash share one record; Loki-only needs a JSON object only + // when the HTTP request is actually built. let mut jsonl_line = Vec::new(); if config.file.enabled || config.logstash.enabled { - jsonl_line = serde_json::to_vec(&payload).unwrap_or_default(); - jsonl_line.push(b'\n'); + append_jsonl(&entry, &config.format, &mut jsonl_line); } let mut attempted = false; let mut failed = false; @@ -308,6 +358,8 @@ async fn sink_loop( // serialized record when another sink is active; avoid a second // full serde_json pass for every high-volume access event. let event_json = if jsonl_line.is_empty() { + let mut payload = format_entry(&entry, &config.format); + redact_value(&mut payload); payload.to_string() } else { String::from_utf8_lossy(&jsonl_line[..jsonl_line.len() - 1]).into_owned() diff --git a/src/logging/tests.rs b/src/logging/tests.rs index 0dbc6e5..b32f488 100644 --- a/src/logging/tests.rs +++ b/src/logging/tests.rs @@ -378,7 +378,7 @@ mod tests { let mut sentinel = entry(); sentinel.id = 6; let mut history = logs.entries.lock().unwrap(); - history.push_back(sentinel.clone()); + history.push_back(HistoryEntry::Owned(sentinel.clone())); let mut item = entry(); item.path = "/contention/€\"quoted\"".into(); @@ -521,4 +521,72 @@ mod tests { } } + #[test] + fn jsonl_fast_serialization_preserves_values_and_redaction() { + let clean = entry(); + assert!(!entry_requires_redaction(&clean)); + let mut token = clean.clone(); + token.path = "/download?access_token=private&language=pl".into(); + token.user_agent = Some("Authorization: Bearer example".into()); + assert!(entry_requires_redaction(&token)); + let mut error = clean.clone(); + error.event = "protocol_error"; + error.error_detail = Some("-----BEGIN PRIVATE KEY-----abc-----END PRIVATE KEY-----".into()); + error.error = Some("{\"password\":\"secret\"}".into()); + assert!(entry_requires_redaction(&error)); + let mut mapped = JsonFormat::default(); + mapped.fields.insert("request_path".into(), "path".into()); + mapped.fields.insert("http_status".into(), "status".into()); + mapped.static_fields.insert("service".into(), "edge".into()); + let mut nested = JsonFormat::default(); + nested.static_fields.insert("token".into(), "secret-value".into()); + let defaults = JsonFormat::default(); + for (record, format) in [ + (&clean, &defaults), + (&token, &defaults), + (&error, &defaults), + (&token, &mapped), + (&clean, &nested), + ] { + let mut previous = format_entry(record, format); + redact_value(&mut previous); + let mut jsonl = Vec::new(); + append_jsonl(record, format, &mut jsonl); + assert_eq!(jsonl.last(), Some(&b'\n')); + let got: Value = serde_json::from_slice(&jsonl[..jsonl.len() - 1]).unwrap(); + assert_eq!(got, previous); + } + assert!(may_require_redaction("?key_secret=value")); + assert!(may_require_redaction("\"PASSWORD\":\"value\"")); + assert!(!may_require_redaction("/assets/home.html")); + } + + #[tokio::test] + async fn active_sink_shares_immutable_record_with_history() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("access.jsonl"); + let mut settings = Settings::default(); + settings.file.enabled = true; + settings.file.path = path.to_string_lossy().into_owned(); + let shutdown = CancellationToken::new(); + let logs = Logs::start(settings, shutdown.clone()); + let sample = entry(); + logs.push(sample.clone()); + { + let history = logs.entries.lock().unwrap(); + assert!(matches!(history.back(), Some(HistoryEntry::Shared(_)))); + assert_eq!(history.back().unwrap().entry().id, sample.id); + } + assert_eq!(serde_json::to_value(logs.recent()).unwrap(), json!([sample])); + tokio::time::timeout(Duration::from_secs(10), async { + while logs.stats().sent == 0 { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }).await.unwrap(); + shutdown.cancel(); + assert_eq!(logs.stats().queue_dropped, 0); + let jsonl = tokio::fs::read_to_string(path).await.unwrap(); + assert_eq!(jsonl.lines().count(), 1); + } + } diff --git a/src/proxy/upstream_helpers.rs b/src/proxy/upstream_helpers.rs index d17ca09..9885972 100644 --- a/src/proxy/upstream_helpers.rs +++ b/src/proxy/upstream_helpers.rs @@ -150,7 +150,7 @@ async fn observe(State(state): State>, mut request: Request, next: Next .headers() .get(USER_AGENT) .and_then(|value| value.to_str().ok()) - .map(|value| value.chars().take(512).collect::()) + .map(|value| log_prefix(value, 512)) }) .flatten(); let (http_version, transport) = match request.version() {