refactor: update log handling to support UTF-8 text and enhance related tests (#777)

Author: halajohn

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

@@ -16,7 +16,7 @@ use serde_json::json;
 use tokio::sync::Mutex;
 
 use crate::designer::DesignerState;
-use crate::fs::file_watcher::{FileContentStream, FileWatchOptions};
+use crate::fs::log_file_watcher::{LogFileContentStream, LogFileWatchOptions};
 use crate::pkg_info::property::get_log_file_path;
 
 // Message types for WebSocket communication
@@ -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 = "()")]
@@ -48,12 +48,12 @@ pub struct InfoMessage(pub String);
 
 #[derive(Message)]
 #[rtype(result = "()")]
-pub struct StoreWatcher(pub FileContentStream);
+pub struct StoreWatcher(pub LogFileContentStream);
 
 // WebSocket actor for log file watching.
 struct WsLogWatcher {
     app_base_dir: Option<String>,
-    file_watcher: Option<Arc<Mutex<FileContentStream>>>,
+    file_watcher: Option<Arc<Mutex<LogFileContentStream>>>,
 }
 
 impl WsLogWatcher {
@@ -125,10 +125,10 @@ impl Handler<SetAppBaseDir> for WsLogWatcher {
             };
 
             // Create file watch options.
-            let options = FileWatchOptions::default();
+            let options = LogFileWatchOptions::default();
 
             // Start watching the file.
-            match crate::fs::file_watcher::watch_file(
+            match crate::fs::log_file_watcher::watch_log_file(
                 &log_file_path,
                 Some(options),
             )
@@ -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 renamed to core/src/ten_manager/src/fs/log_file_watcher.rs

@@ -23,26 +23,26 @@ 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.
-pub struct FileContentStream {
-    // Channel for receiving file content.
-    content_rx: Receiver<Result<Vec<u8>>>,
+/// Stream of UTF-8 text file content changes.
+pub struct LogFileContentStream {
+    // Channel for receiving file content as UTF-8 text.
+    content_rx: Receiver<Result<String>>,
 
     // Sender to signal stop request.
     stop_tx: Option<oneshot::Sender<()>>,
 }
 
-impl FileContentStream {
+impl LogFileContentStream {
     /// 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
     }
 
@@ -54,15 +54,15 @@ impl FileContentStream {
     }
 }
 
-impl Drop for FileContentStream {
+impl Drop for LogFileContentStream {
     fn drop(&mut self) {
         self.stop();
     }
 }
 
 /// Options for watching a file.
 #[derive(Clone)]
-pub struct FileWatchOptions {
+pub struct LogFileWatchOptions {
     /// Timeout for waiting for new content after reaching EOF.
     pub timeout: Duration,
 
@@ -73,7 +73,7 @@ pub struct FileWatchOptions {
     pub check_interval: Duration,
 }
 
-impl Default for FileWatchOptions {
+impl Default for LogFileWatchOptions {
     fn default() -> Self {
         Self {
             timeout: DEFAULT_TIMEOUT,
@@ -97,17 +97,17 @@ 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:
 /// 1. The caller stops it by calling `stop()` or dropping the stream.
 /// 2. No new content is available after reaching EOF and the timeout is
 ///    reached.
-pub async fn watch_file<P: AsRef<Path>>(
+pub async fn watch_log_file<P: AsRef<Path>>(
     path: P,
-    options: Option<FileWatchOptions>,
-) -> Result<FileContentStream> {
+    options: Option<LogFileWatchOptions>,
+) -> Result<LogFileContentStream> {
     let path = path.as_ref().to_path_buf();
 
     // Ensure the file exists before we start watching it.
@@ -123,18 +123,18 @@ pub async fn watch_file<P: AsRef<Path>>(
 
     // Spawn a task to watch the file.
     tokio::spawn(async move {
-        watch_file_task(path, content_tx, stop_rx, options).await;
+        watch_log_file_task(path, content_tx, stop_rx, options).await;
     });
 
-    Ok(FileContentStream::new(content_rx, stop_tx))
+    Ok(LogFileContentStream::new(content_rx, stop_tx))
 }
 
 /// Actual file watch task running in the background.
-async fn watch_file_task(
+async fn watch_log_file_task(
     path: PathBuf,
-    content_tx: Sender<Result<Vec<u8>>>,
+    content_tx: Sender<Result<String>>,
     mut stop_rx: oneshot::Receiver<()>,
-    options: FileWatchOptions,
+    options: LogFileWatchOptions,
 ) {
     // Open the file.
     let mut file = match File::open(&path) {
@@ -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/fs/mod.rs

@@ -5,8 +5,8 @@
 // Refer to the "LICENSE" file in the root directory for more information.
 //
 pub mod file_type;
-pub mod file_watcher;
 pub mod json;
+pub mod log_file_watcher;
 
 use std::env;
 use std::path::{Path, PathBuf};

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