v0.34.0_9

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<LogEntry>),
+}
+impl HistoryEntry {
+    fn entry(&self) -> &LogEntry {
+        match self {
+            Self::Owned(value) => value,
+            Self::Shared(value) => value.as_ref(),
+        }
+    }
+}
+
 pub struct Logs {
-    entries: Mutex<VecDeque<LogEntry>>,
+    entries: Mutex<VecDeque<HistoryEntry>>,
     pub dropped: AtomicU64,
     settings: Arc<ArcSwap<Settings>>,
     live: broadcast::Sender<LogEntry>,
-    sink: mpsc::Sender<LogEntry>,
+    sink: mpsc::Sender<Arc<LogEntry>>,
     sink_dropped: Arc<AtomicU64>,
     sink_sent: Arc<AtomicU64>,
     sink_errors: Arc<AtomicU64>,
@@ -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()

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<u8>) {
+    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<File> {
     Ok(File::from_std(file))
 }
 async fn sink_loop(
-    mut rx: mpsc::Receiver<LogEntry>,
+    mut rx: mpsc::Receiver<Arc<LogEntry>>,
     settings: Arc<ArcSwap<Settings>>,
     sent: Arc<AtomicU64>,
     errors: Arc<AtomicU64>,
@@ -118,7 +175,7 @@ async fn sink_loop(
     let mut vhost_files: BTreeMap<PathBuf, File> = BTreeMap::new();
     let mut vhost_directory: Option<PathBuf> = None;
     let mut logstash: Option<LogstashState> = None;
-    let mut deferred_entry: Option<LogEntry> = None;
+    let mut deferred_entry: Option<Arc<LogEntry>> = 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()

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);
+    }
+
 }

src/proxy/upstream_helpers.rs

@@ -150,7 +150,7 @@ async fn observe(State(state): State<Arc<App>>, mut request: Request, next: Next
                 .headers()
                 .get(USER_AGENT)
                 .and_then(|value| value.to_str().ok())
-                .map(|value| value.chars().take(512).collect::<String>())
+                .map(|value| log_prefix(value, 512))
         })
         .flatten();
     let (http_version, transport) = match request.version() {