refactor: update log handling to support UTF-8 text and enhance related tests

Author: halajohn

core/src/ten_manager/src/designer/log_watcher/mod.rs

@@ -32,7 +32,7 @@ pub struct SetAppBaseDir {
 
 #[derive(Message, Debug, Serialize, Deserialize)]
 #[rtype(result = "()")]
-pub struct FileContent(pub Vec<u8>);
+pub struct FileContent(pub String);
 
 #[derive(Message, Debug, Serialize, Deserialize)]
 #[rtype(result = "()")]
@@ -160,8 +160,8 @@ impl Handler<FileContent> for WsLogWatcher {
         msg: FileContent,
         ctx: &mut Self::Context,
     ) -> Self::Result {
-        // Send the file content to the WebSocket client.
-        ctx.binary(msg.0);
+        // Send the file content as text to the WebSocket client.
+        ctx.text(msg.0);
     }
 }
 

core/src/ten_manager/src/fs/file_watcher.rs

@@ -23,10 +23,10 @@ const DEFAULT_TIMEOUT: Duration = Duration::from_secs(60); // 1 minute timeout.
 const DEFAULT_BUFFER_SIZE: usize = 4096; // Default read buffer size.
 const DEFAULT_CHECK_INTERVAL: Duration = Duration::from_millis(100);
 
-/// Stream of file content changes.
+/// Stream of UTF-8 text file content changes.
 pub struct FileContentStream {
-    // Channel for receiving file content.
-    content_rx: Receiver<Result<Vec<u8>>>,
+    // Channel for receiving file content as UTF-8 text.
+    content_rx: Receiver<Result<String>>,
 
     // Sender to signal stop request.
     stop_tx: Option<oneshot::Sender<()>>,
@@ -35,14 +35,14 @@ pub struct FileContentStream {
 impl FileContentStream {
     /// Create a new FileContentStream.
     fn new(
-        content_rx: Receiver<Result<Vec<u8>>>,
+        content_rx: Receiver<Result<String>>,
         stop_tx: oneshot::Sender<()>,
     ) -> Self {
         Self { content_rx, stop_tx: Some(stop_tx) }
     }
 
-    /// Get the next chunk of data from the file.
-    pub async fn next(&mut self) -> Option<Result<Vec<u8>>> {
+    /// Get the next chunk of text from the file.
+    pub async fn next(&mut self) -> Option<Result<String>> {
         self.content_rx.recv().await
     }
 
@@ -97,7 +97,7 @@ fn is_same_file(a: &Metadata, b: &Metadata) -> bool {
     }
 }
 
-/// Watch a file for changes and stream its content.
+/// Watch a UTF-8 text file for changes and stream its content.
 ///
 /// Returns a FileContentStream that can be used to read the content of the file
 /// as it changes. The stream will end when either:
@@ -132,7 +132,7 @@ pub async fn watch_file<P: AsRef<Path>>(
 /// Actual file watch task running in the background.
 async fn watch_file_task(
     path: PathBuf,
-    content_tx: Sender<Result<Vec<u8>>>,
+    content_tx: Sender<Result<String>>,
     mut stop_rx: oneshot::Receiver<()>,
     options: FileWatchOptions,
 ) {
@@ -157,15 +157,18 @@ async fn watch_file_task(
         let _ = content_tx.send(Err(anyhow!(e))).await;
         return;
     }
-    let mut init_buf = Vec::new();
-    match file.read_to_end(&mut init_buf) {
+
+    let mut init_content = String::new();
+    match file.read_to_string(&mut init_content) {
         Ok(n) => {
             if n > 0 {
-                let _ = content_tx.send(Ok(init_buf)).await;
+                let _ = content_tx.send(Ok(init_content)).await;
             }
         }
         Err(e) => {
-            let _ = content_tx.send(Err(anyhow!(e))).await;
+            let _ = content_tx
+                .send(Err(anyhow!("Failed to read file as UTF-8 text: {}", e)))
+                .await;
             return;
         }
     }
@@ -226,17 +229,25 @@ async fn watch_file_task(
                     }
 
                     let mut reader = BufReader::with_capacity(options.buffer_size, &file);
-                    let mut buf = Vec::with_capacity(options.buffer_size);
-                    match reader.read_until(0, &mut buf) {
+                    let mut line = String::with_capacity(options.buffer_size);
+                    match reader.read_line(&mut line) {
                         Ok(n) if n > 0 => {
                             last_pos += n as u64;
-                            let _ = content_tx.send(Ok(buf)).await;
+                            let _ = content_tx.send(Ok(line)).await;
                             last_activity = Instant::now();
                         }
                         Ok(_) => {}
                         Err(e) => {
-                            eprintln!("Error reading from file: {}", e);
-                            let _ = content_tx.send(Err(anyhow::anyhow!(e))).await;
+                            // Specific error for UTF-8 decoding failures.
+                            if e.kind() == std::io::ErrorKind::InvalidData {
+                                eprintln!("Error: Invalid UTF-8 data in file");
+                                let _ = content_tx
+                                    .send(Err(anyhow!("Invalid UTF-8 data in file")))
+                                    .await;
+                            } else {
+                                eprintln!("Error reading from file: {}", e);
+                                let _ = content_tx.send(Err(anyhow::anyhow!(e))).await;
+                            }
                             break;
                         }
                     }

core/src/ten_manager/src/log/mod.rs

@@ -4,23 +4,100 @@
 // Licensed under the Apache License, Version 2.0, with certain conditions.
 // Refer to the "LICENSE" file in the root directory for more information.
 //
-use anyhow::Result;
+use anyhow::{anyhow, Result};
+use serde_json::Value;
 use std::collections::HashMap;
 
-pub struct ExtensionInfo {
-    pub thread_id: u64,
+pub struct ExtensionThreadInfo {
+    pub extensions: Vec<String>,
+}
+
+pub struct AppInfo {
+    pub extension_threads: HashMap<String, ExtensionThreadInfo>,
 }
 
 pub struct GraphResourcesLog {
-    pub app_uri: Option<String>,
     pub graph_id: String,
     pub graph_name: String,
-    pub extensions: HashMap<String, ExtensionInfo>,
+    pub apps: HashMap<Option<String>, AppInfo>,
 }
 
 pub fn parse_graph_resources_log(
     log_message: &str,
     graph_resources_log: &mut GraphResourcesLog,
 ) -> Result<()> {
+    // Check if the log level is 'M'.
+    let parts: Vec<&str> = log_message.split_whitespace().collect();
+    if parts.len() < 4 {
+        return Ok(());
+    }
+
+    // Check for log level 'M' - it should be in the fourth position after the
+    // timestamp and process/thread IDs.
+    let log_level_pos = 3;
+    if parts.len() <= log_level_pos || parts[log_level_pos] != "M" {
+        return Ok(());
+    }
+
+    // Find the "[graph resources]" marker.
+    if !log_message.contains("[graph resources]") {
+        return Ok(());
+    }
+
+    // Extract the JSON content after "[graph resources]".
+    let json_content = log_message
+        .split("[graph resources]")
+        .nth(1)
+        .ok_or_else(|| anyhow!("Failed to extract JSON content"))?
+        .trim();
+
+    // Parse the JSON content.
+    let json_value: Value = serde_json::from_str(json_content)?;
+
+    // Extract data from the JSON.
+    let app_uri = json_value.get("app_uri").and_then(|v| v.as_str());
+    let graph_id = json_value["graph id"]
+        .as_str()
+        .ok_or_else(|| anyhow!("Missing graph id"))?;
+    let graph_name = json_value["graph name"]
+        .as_str()
+        .ok_or_else(|| anyhow!("Missing graph name"))?;
+
+    // Update graph_resources_log with graph ID and name.
+    graph_resources_log.graph_id = graph_id.to_string();
+    graph_resources_log.graph_name = graph_name.to_string();
+
+    // Create or get the AppInfo for this app_uri.
+    let app_key = app_uri.map(|uri| uri.to_string());
+    let app_info = graph_resources_log
+        .apps
+        .entry(app_key)
+        .or_insert_with(|| AppInfo { extension_threads: HashMap::new() });
+
+    // Process extension_threads if present.
+    if let Some(extension_threads) = json_value.get("extension_threads") {
+        if let Some(extension_threads_obj) = extension_threads.as_object() {
+            for (thread_id, thread_info) in extension_threads_obj {
+                if let Some(extensions_array) = thread_info.get("extensions") {
+                    if let Some(extensions) = extensions_array.as_array() {
+                        let mut extension_names = Vec::new();
+                        for ext in extensions {
+                            if let Some(ext_name) = ext.as_str() {
+                                extension_names.push(ext_name.to_string());
+                            }
+                        }
+
+                        let thread_info =
+                            ExtensionThreadInfo { extensions: extension_names };
+
+                        app_info
+                            .extension_threads
+                            .insert(thread_id.to_string(), thread_info);
+                    }
+                }
+            }
+        }
+    }
+
     Ok(())
 }

core/src/ten_manager/tests/test_case/designer/log_watcher.rs

@@ -93,11 +93,10 @@ async fn test_ws_log_watcher_endpoint() {
     let mut received_content = false;
     while let Some(msg) = read.next().await {
         let msg = msg.unwrap();
-        if msg.is_binary() {
-            let binary = msg.into_data();
-            let content = String::from_utf8_lossy(&binary);
-            println!("Received binary: {}", content);
-            if content.contains(test_content) {
+        if msg.is_text() {
+            let text = msg.to_text().unwrap();
+            println!("Received text: {}", text);
+            if text.contains(test_content) {
                 received_content = true;
                 break;
             }

core/src/ten_manager/tests/test_case/fs/mod.rs

@@ -23,8 +23,8 @@ mod tests {
         rt.block_on(async {
             // Create a temporary file for testing
             let mut temp_file = NamedTempFile::new()?;
-            let test_content = b"Hello, World!";
-            temp_file.write_all(test_content)?;
+            let test_content = "Hello, World!";
+            temp_file.write_all(test_content.as_bytes())?;
             temp_file.flush()?;
 
             // Create options with shorter timeout for testing.
@@ -43,8 +43,8 @@ mod tests {
             assert_eq!(chunk, test_content);
 
             // Write more content to the file.
-            let more_content = b"More content!";
-            temp_file.write_all(more_content)?;
+            let more_content = "More content!";
+            temp_file.write_all(more_content.as_bytes())?;
             temp_file.flush()?;
 
             // Get the second chunk.
@@ -82,7 +82,7 @@ mod tests {
                     .create(true)
                     .truncate(true)
                     .open(&path_str)?;
-                file.write_all(b"Initial content")?;
+                file.write_all("Initial content".as_bytes())?;
                 file.flush()?;
             }
 
@@ -98,7 +98,7 @@ mod tests {
 
             // Get the first chunk
             let chunk = stream.next().await.expect("Should receive data")?;
-            assert_eq!(chunk, b"Initial content");
+            assert_eq!(chunk, "Initial content");
 
             // Simulate log rotation - delete and recreate file
             std::fs::remove_file(&path_str)?;
@@ -112,14 +112,14 @@ mod tests {
                     .create(true)
                     .truncate(true)
                     .open(&path_str)?;
-                file.write_all(b"Rotated content")?;
+                file.write_all("Rotated content".as_bytes())?;
                 file.flush()?;
             }
 
             // Get the content after rotation
             let chunk =
                 stream.next().await.expect("Should receive rotated data")?;
-            assert_eq!(chunk, b"Rotated content");
+            assert_eq!(chunk, "Rotated content");
 
             // Stop watching
             stream.stop();
@@ -134,7 +134,7 @@ mod tests {
         rt.block_on(async {
             // Create a temporary file for testing.
             let mut temp_file = NamedTempFile::new()?;
-            temp_file.write_all(b"Test content")?;
+            temp_file.write_all("Test content".as_bytes())?;
             temp_file.flush()?;
 
             // Create options with very short timeout for testing.
@@ -150,7 +150,7 @@ mod tests {
 
             // Get the first chunk.
             let chunk = stream.next().await.expect("Should receive data")?;
-            assert_eq!(chunk, b"Test content");
+            assert_eq!(chunk, "Test content");
 
             // Wait for the timeout to occur (no new content being written).
             let next_result = stream.next().await;