TelemetryEventsRefactor: Move redact_secrets to the sequential background consumer

proxy_agent_shared/src/misc_helpers.rs

@@ -680,6 +680,24 @@ pub fn xml_escape(s: String) -> String {
         .replace('>', ">")
 }
 
+/// Truncate the given string to the specified maximum number of bytes, ensuring that it does not cut off in the middle of a character.
+/// # Arguments
+/// * `text` - The string to be truncated
+/// * `max_bytes_size` - The maximum number of bytes the string should occupy
+/// # Notes
+/// This function ensures that the resulting string is valid UTF-8 by truncating at character boundaries.
+pub fn truncate_to_char_boundary(text: &mut String, max_bytes_size: usize) {
+    if text.len() <= max_bytes_size {
+        return;
+    }
+
+    let mut end = max_bytes_size;
+    while !text.is_char_boundary(end) {
+        end -= 1;
+    }
+    text.truncate(end);
+}
+
 #[cfg(test)]
 mod tests {
     use regex::Regex;

proxy_agent_shared/src/telemetry/event_logger.rs

@@ -13,7 +13,6 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
-const MAX_MESSAGE_LENGTH: usize = 1024 * 4; // 4KB
 static EVENT_QUEUE: Lazy<ConcurrentQueue<Event>> =
     Lazy::new(|| ConcurrentQueue::<Event>::bounded(1000));
 static SHUT_DOWN: Lazy<Arc<AtomicBool>> = Lazy::new(|| Arc::new(AtomicBool::new(false)));
@@ -300,11 +299,13 @@ pub fn write_event(
 /// Write event only without logging to file
 /// This event will send out as `TelemetryGenericLogsEvent`
 pub fn write_event_only(level: Level, message: String, method_name: &str, module_name: &str) {
-    let event_message = if message.len() > MAX_MESSAGE_LENGTH {
-        message[..MAX_MESSAGE_LENGTH].to_string()
-    } else {
-        message.to_string()
-    };
+    // Truncate the event message to the maximum allowed telemetry message length before sending it to the memory queue.
+    let mut event_message = message;
+    misc_helpers::truncate_to_char_boundary(
+        &mut event_message,
+        super::telemetry_event::MAX_TELEMETRY_MESSAGE_LENGTH,
+    );
+
     match EVENT_QUEUE.push(Event::new(
         level.to_string(),
         event_message,
@@ -327,14 +328,12 @@ pub fn push_windows_event(windows_event: crate::windows_events::models::WindowsE
     if let (Some(provider_name), Some(task_name)) =
         (&windows_event.provider_name, &windows_event.task_name)
     {
-        let event_message = {
-            let message = windows_event.get_message();
-            if message.len() > MAX_MESSAGE_LENGTH {
-                message[..MAX_MESSAGE_LENGTH].to_string()
-            } else {
-                message
-            }
-        };
+        // Truncate the event message to the maximum allowed telemetry message length before sending it to the memory queue.
+        let mut event_message = windows_event.get_message();
+        misc_helpers::truncate_to_char_boundary(
+            &mut event_message,
+            super::telemetry_event::MAX_TELEMETRY_MESSAGE_LENGTH,
+        );
 
         match EVENT_QUEUE.push(Event {
             EventLevel: windows_event.get_level_string(),

proxy_agent_shared/src/telemetry/event_sender.rs

@@ -108,7 +108,11 @@ impl EventSender {
             let mut add_more_events = true;
             while !TELEMETRY_EVENT_QUEUE.is_empty() && add_more_events {
                 match TELEMETRY_EVENT_QUEUE.pop() {
-                    Ok(event) => {
+                    Ok(mut event) => {
+                        // Redact only in this sequential background consumer.
+                        // Producers stay off the blocking regex path.
+                        event.redact_secrets();
+
                         telemetry_data.add_event(event.clone());
 
                         if telemetry_data.get_size() >= MAX_MESSAGE_SIZE {

proxy_agent_shared/src/telemetry/telemetry_event.rs

@@ -8,6 +8,10 @@ use crate::{current_info, misc_helpers};
 use once_cell::sync::Lazy;
 use serde_derive::{Deserialize, Serialize};
 
+// Keep telemetry messages bounded before secret redaction. Besides limiting the wire payload, this
+// prevents externally supplied extension status from causing disproportionate regex work or memory use.
+pub const MAX_TELEMETRY_MESSAGE_LENGTH: usize = 4 * 1024;
+
 const METRICS_PROVIDER_ID: &str = "FFF0196F-EE4C-4EAF-9AA5-776F622DEB4F";
 const STATUS_PROVIDER_ID: &str = "69B669B9-4AF8-4C50-BDC4-6006FA76E975";
 
@@ -279,6 +283,22 @@ impl TelemetryEvent {
             TelemetryEvent::ExtensionEvent(event) => event.to_xml_event(vm_data),
         }
     }
+
+    /// Redact an event in the background telemetry-consumer path.
+    pub(crate) fn redact_secrets(&mut self) {
+        match self {
+            TelemetryEvent::GenericLogsEvent(event) => {
+                event.context1 = crate::secrets_redactor::redact_secrets_string(std::mem::take(
+                    &mut event.context1,
+                ));
+            }
+            TelemetryEvent::ExtensionEvent(event) => {
+                event.message = crate::secrets_redactor::redact_secrets_string(std::mem::take(
+                    &mut event.message,
+                ));
+            }
+        }
+    }
 }
 
 /// Struct to hold Generic Logs telemetry event data without VM metadata.
@@ -311,9 +331,10 @@ impl TelemetryGenericLogsEvent {
             Some(version) => (version, format!("{}-{}", event_name, event_log.Version)),
             None => (event_log.Version.clone(), event_name),
         };
-        // redact secrets in the message before sending to telemetry
-        let message = event_log.Message.clone();
-        let message = crate::secrets_redactor::redact_secrets_string(message);
+        // Bound producer work and queue memory here; redaction runs later in the background sender.
+        let mut message = event_log.Message.clone();
+        misc_helpers::truncate_to_char_boundary(&mut message, MAX_TELEMETRY_MESSAGE_LENGTH);
+
         TelemetryGenericLogsEvent {
             event_name,
             ga_version,
@@ -416,9 +437,10 @@ impl TelemetryExtensionEventsEvent {
         execution_mode: String,
         ga_version: String,
     ) -> Self {
-        // redact secrets in the message before sending to telemetry
-        let message = event.operation_status.message.clone();
-        let message = crate::secrets_redactor::redact_secrets_string(message);
+        // Bound producer work and queue memory here; redaction runs later in the background sender.
+        let mut message = event.operation_status.message.clone();
+        misc_helpers::truncate_to_char_boundary(&mut message, MAX_TELEMETRY_MESSAGE_LENGTH);
+
         TelemetryExtensionEventsEvent {
             ga_version,
             execution_mode,
@@ -832,6 +854,47 @@ mod tests {
         assert!(xml.contains("500"));
     }
 
+    #[test]
+    fn test_extension_event_bounds_message_before_redaction() {
+        let mut extension_status_event = create_test_extension_status_event();
+        extension_status_event.operation_status.message = format!(
+            "Authorization: Bearer secret\n{}",
+            "x".repeat(MAX_TELEMETRY_MESSAGE_LENGTH * 16)
+        );
+
+        let telemetry_event = TelemetryExtensionEventsEvent::from_extension_status_event(
+            &extension_status_event,
+            "production".to_string(),
+            "1.0.0".to_string(),
+        );
+        let mut telemetry_event = TelemetryEvent::ExtensionEvent(telemetry_event);
+        telemetry_event.redact_secrets();
+
+        let TelemetryEvent::ExtensionEvent(telemetry_event) = telemetry_event else {
+            unreachable!();
+        };
+        assert!(telemetry_event.message.len() <= MAX_TELEMETRY_MESSAGE_LENGTH);
+        assert!(telemetry_event.message.starts_with("[REDACTED]\n"));
+        assert!(!telemetry_event.message.contains("secret"));
+    }
+
+    #[test]
+    fn test_extension_event_truncates_at_utf8_boundary() {
+        let mut extension_status_event = create_test_extension_status_event();
+        extension_status_event.operation_status.message = "é".repeat(MAX_TELEMETRY_MESSAGE_LENGTH);
+
+        let telemetry_event = TelemetryExtensionEventsEvent::from_extension_status_event(
+            &extension_status_event,
+            "production".to_string(),
+            "1.0.0".to_string(),
+        );
+
+        assert_eq!(telemetry_event.message.len(), MAX_TELEMETRY_MESSAGE_LENGTH);
+        assert!(telemetry_event
+            .message
+            .is_char_boundary(telemetry_event.message.len()));
+    }
+
     /// Tests TelemetryExtensionEventsEvent with operation failure
     #[test]
     fn test_telemetry_extension_events_event_failure() {