v0.34.0_8

src/logging/error_runtime.rs

@@ -48,53 +48,61 @@ fn redact_regex(pattern: &'static str) -> &'static regex::Regex {
 }
 
 fn is_secret_field(name: &str) -> bool {
-    matches!(
-        name.to_ascii_lowercase().as_str(),
-        "authorization"
-            | "proxy_authorization"
-            | "api_token"
-            | "access_token"
-            | "application_secret"
-            | "consumer_key"
-            | "key_secret"
-            | "backup_password"
-            | "password"
-            | "credential"
-            | "credentials"
-            | "secret"
-            | "token"
-            | "private_key"
-            | "master_key"
-    )
+    const FIELDS: &[&str] = &[
+        "authorization",
+        "proxy_authorization",
+        "api_token",
+        "access_token",
+        "application_secret",
+        "consumer_key",
+        "key_secret",
+        "backup_password",
+        "password",
+        "credential",
+        "credentials",
+        "secret",
+        "token",
+        "private_key",
+        "master_key",
+    ];
+    // ASCII-insensitive matching has the same semantics as to_ascii_lowercase,
+    // without allocating a new key for every field in every JSON access log.
+    FIELDS.iter().any(|field| name.eq_ignore_ascii_case(field))
+}
+
+fn redact_text_cow(input: &str) -> std::borrow::Cow<'_, str> {
+    let mut value = std::borrow::Cow::Borrowed(input);
+    // Keep the original order: a replacement can expose a pattern handled by
+    // the next rule. Ordinary, non-secret values stay borrowed (no six copies).
+    for (pattern, replacement) in [
+        ("pem", "[REDACTED PRIVATE KEY]"),
+        ("auth", "$1[REDACTED]"),
+        ("query", "$1[REDACTED]"),
+        ("escaped_json", "$1[REDACTED]$2"),
+        ("json", "$1[REDACTED]$2"),
+        ("kv", "$1[REDACTED]"),
+    ] {
+        let regex = redact_regex(pattern);
+        if regex.is_match(value.as_ref()) {
+            value = std::borrow::Cow::Owned(
+                regex.replace_all(value.as_ref(), replacement).into_owned(),
+            );
+        }
+    }
+    value
 }
 
 pub fn redact_text(input: &str) -> String {
-    let mut value = redact_regex("pem")
-        .replace_all(input, "[REDACTED PRIVATE KEY]")
-        .into_owned();
-    value = redact_regex("auth")
-        .replace_all(&value, "$1[REDACTED]")
-        .into_owned();
-    value = redact_regex("query")
-        .replace_all(&value, "$1[REDACTED]")
-        .into_owned();
-    // Debug-formatted fields can contain a JSON string with escaped quote
-    // delimiters. Redact that form before plain JSON, retaining quote escapes
-    // and consuming escaped quotes inside a secret instead of leaking its tail.
-    value = redact_regex("escaped_json")
-        .replace_all(&value, "$1[REDACTED]$2")
-        .into_owned();
-    value = redact_regex("json")
-        .replace_all(&value, "$1[REDACTED]$2")
-        .into_owned();
-    redact_regex("kv")
-        .replace_all(&value, "$1[REDACTED]")
-        .into_owned()
+    redact_text_cow(input).into_owned()
 }
 
 fn redact_value(value: &mut Value) {
     match value {
-        Value::String(text) => *text = redact_text(text),
+        Value::String(text) => {
+            if let std::borrow::Cow::Owned(redacted) = redact_text_cow(text) {
+                *text = redacted;
+            }
+        }
         Value::Array(items) => {
             for item in items {
                 redact_value(item);

src/logging/format.rs

@@ -138,11 +138,21 @@ impl Logs {
     pub fn push(&self, entry: LogEntry) {
         let has_viewers = self.live.receiver_count() != 0;
         let has_sink = self.settings.load().enabled();
-        // The usual history-only configuration needs no deep copy at all.
-        // When external consumers are present, clone exactly for history and
-        // keep the original entry available for streaming/sink delivery.
+        // 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 {
+            match self.sink.try_reserve() {
+                Ok(permit) => Some(permit),
+                Err(_) => {
+                    self.sink_dropped.fetch_add(1, Ordering::Relaxed);
+                    None
+                }
+            }
+        } else {
+            None
+        };
         let mut external = Some(entry);
-        let history_entry = if has_viewers || has_sink {
+        let history_entry = if has_viewers || sink_permit.is_some() {
             external.as_ref().expect("entry").clone()
         } else {
             external.take().expect("entry")
@@ -165,10 +175,8 @@ impl Logs {
             if has_viewers {
                 let _ = self.live.send(entry.clone());
             }
-            // Snapshot the sink choice once. This preserves sink drop
-            // accounting and avoids a second ArcSwap load per request.
-            if has_sink && self.sink.try_send(entry).is_err() {
-                self.sink_dropped.fetch_add(1, Ordering::Relaxed);
+            if let Some(permit) = sink_permit {
+                permit.send(entry);
             }
         }
     }

src/logging/runtime.rs

@@ -41,6 +41,10 @@ fn format_entry(entry: &LogEntry, format: &JsonFormat) -> Value {
     Value::Object(object)
 }
 
+// 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;
+
 type LokiClient = Client<hyper_rustls::HttpsConnector<HttpConnector>, Body>;
 fn loki_client() -> Result<LokiClient> {
     let mut roots = RootCertStore::empty();
@@ -114,12 +118,74 @@ 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 file_batch = Vec::with_capacity(8 * 1024);
     loop {
-        let entry = tokio::select! {
-            _ = shutdown.cancelled() => break,
-            entry = rx.recv() => match entry { Some(entry) => entry, None => break },
+        let entry = if let Some(entry) = deferred_entry.take() {
+            entry
+        } else {
+            tokio::select! {
+                _ = shutdown.cancelled() => break,
+                entry = rx.recv() => match entry { Some(entry) => entry, None => break },
+            }
         };
         let config = settings.load_full();
+        // The common local-file-only sink has no per-event network or routing
+        // dependency. Coalesce consecutive JSONL records into one bounded write.
+        // Other sink combinations retain their original per-event semantics.
+        if config.file.enabled && !config.file.per_vhost
+            && !config.logstash.enabled && !config.loki.enabled
+        {
+            vhost_files.clear();
+            vhost_directory = None;
+            let path = PathBuf::from(&config.file.path);
+            if file.as_ref().is_none_or(|state| state.path != path) {
+                file = match open_access_file(path.clone()).await {
+                    Ok(opened) => Some(FileState { path: path.clone(), file: opened }),
+                    Err(error) => {
+                        tracing::warn!(%error, path=%path.display(), "access log file unavailable");
+                        None
+                    }
+                };
+            }
+            file_batch.clear();
+            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);
+                count += 1;
+                if count >= FILE_BATCH_EVENTS || file_batch.len() >= FILE_BATCH_BYTES {
+                    break;
+                }
+                match rx.try_recv() {
+                    Ok(item) if Arc::ptr_eq(&settings.load_full(), &config) => {
+                        next = Some(item);
+                    }
+                    Ok(item) => {
+                        // A pending entry must be handled with the new snapshot;
+                        // a batch never silently crosses a configuration change.
+                        deferred_entry = Some(item);
+                        break;
+                    }
+                    Err(_) => break,
+                }
+            }
+            let written = match file.as_mut() {
+                Some(state) => state.file.write_all(&file_batch).await.is_ok(),
+                None => false,
+            };
+            if written {
+                sent.fetch_add(count, Ordering::Relaxed);
+            } else {
+                errors.fetch_add(count, Ordering::Relaxed);
+                file = None;
+            }
+            continue;
+        }
         let mut payload = format_entry(&entry, &config.format);
         redact_value(&mut payload);
         // File and Logstash consume the same JSONL record. Serialize once and

src/logging/tests.rs

@@ -458,4 +458,67 @@ mod tests {
         assert_eq!(logs.stats().queue_dropped, 1);
     }
 
+    #[test]
+    fn redaction_leaves_ordinary_strings_borrowed_and_preserves_secret_matching() {
+        use std::borrow::Cow;
+        assert!(matches!(redact_text_cow("GET /index.html"), Cow::Borrowed(_)));
+        assert!(matches!(redact_text_cow("hello € world"), Cow::Borrowed(_)));
+        assert!(matches!(redact_text_cow("?api_token=topsecret"), Cow::Owned(_)));
+        let mut value = json!({
+            "ToKeN": "secret-value",
+            "ordinary": "hello € world",
+            "nested": {"AUTHORIZATION": "Bearer private"},
+            "path": "/index?password=topsecret"
+        });
+        redact_value(&mut value);
+        assert_eq!(value["ToKeN"], "[REDACTED]");
+        assert_eq!(value["ordinary"], "hello € world");
+        assert_eq!(value["nested"]["AUTHORIZATION"], "[REDACTED]");
+        assert_eq!(value["path"], "/index?password=[REDACTED]");
+    }
+
+    #[tokio::test]
+    async fn local_file_log_batch_keeps_every_jsonl_record_and_counters() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("access.jsonl");
+        let mut settings = Settings::default();
+        settings.file.enabled = true;
+        settings.file.path = path.to_string_lossy().into_owned();
+        settings.format.fields.insert("request_number".into(), "id".into());
+        settings.format.fields.insert("request_path".into(), "path".into());
+        settings.format.static_fields.insert("service".into(), "rustproxy".into());
+        let stop = CancellationToken::new();
+        let logs = Logs::start(settings, stop.clone());
+        const COUNT: usize = 180;
+        for id in 0..COUNT {
+            let mut item = entry();
+            item.id = id as u64;
+            item.path = format!("/test/{id}?token=secret");
+            logs.push(item);
+        }
+        let lines = tokio::time::timeout(Duration::from_secs(10), async {
+            loop {
+                if logs.stats().sent == COUNT as u64 {
+                    if let Ok(text) = tokio::fs::read_to_string(&path).await {
+                        if text.lines().count() == COUNT {
+                            break text;
+                        }
+                    }
+                }
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            }
+        }).await.expect("all accepted local file records must be written");
+        stop.cancel();
+        assert_eq!(logs.stats().queue_dropped, 0);
+        assert_eq!(logs.stats().errors, 0);
+        assert_eq!(logs.stats().sent, COUNT as u64);
+        for (id, line) in lines.lines().enumerate() {
+            let record: Value = serde_json::from_str(line).unwrap();
+            assert_eq!(record["request_number"], id as u64);
+            assert_eq!(record["request_path"], format!("/test/{id}?token=[REDACTED]"));
+            assert_eq!(record["service"], "rustproxy");
+            assert_eq!(record.as_object().unwrap().len(), 3);
+        }
+    }
+
 }