[PATCH 03/23] rteval: Add log file access tools to MCP server
John Kacur <[email protected]>
| Newsgroups | org.kernel.vger.linux-rt-users |
|---|---|
| Message-ID | <[email protected]> |
Add two new tools for accessing rteval log files: 1. list_logs - Lists all log files in an rteval result directory - Groups by stdout and stderr - Shows file sizes and paths - Helps discover what logs are available 2. read_log - Reads content from log files with filtering options - head: Read first N lines - tail: Read last N lines - grep: Search for pattern in log - Default: Shows first 100 lines - Handles large files gracefully (timerlat.stdout can be 1.5MB+) These tools are useful for debugging issues like the timerlat warnings we encountered, allowing users to examine load and measurement logs without manual file access. Tested with rteval-20260605-2 logs directory containing 8 log files (hackbench, kcompile, timerlat stdout/stderr). Assisted-by: Claude:claude-sonnet-4-5 Signed-off-by: John Kacur <[email protected]> --- mcp-server/server.py | 157 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 157 insertions(+) diff --git a/mcp-server/server.py b/mcp-server/server.py index 40c11dfecb34..58de7fc660d5 100755 --- a/mcp-server/server.py +++ b/mcp-server/server.py @@ -145,6 +145,46 @@ async def list_tools() -> list[Tool]: "required": ["file1", "file2"], }, ), + Tool( + name="list_logs", + description="List available log files in an rteval result directory", + inputSchema={ + "type": "object", + "properties": { + "result_dir": { + "type": "string", + "description": "Path to rteval result directory (e.g., rteval-20260605-1)", + }, + }, + "required": ["result_dir"], + }, + ), + Tool( + name="read_log", + description="Read content from an rteval log file", + inputSchema={ + "type": "object", + "properties": { + "log_path": { + "type": "string", + "description": "Path to the log file", + }, + "tail": { + "type": "integer", + "description": "Read last N lines (optional)", + }, + "head": { + "type": "integer", + "description": "Read first N lines (optional)", + }, + "grep": { + "type": "string", + "description": "Search for pattern in log (optional)", + }, + }, + "required": ["log_path"], + }, + ), ] @@ -422,6 +462,123 @@ async def call_tool(name: str, arguments: Any) -> list[TextContent]: text=f"Error comparing results: {str(e)}" )] + elif name == "list_logs": + result_dir = arguments["result_dir"] + + try: + dir_path = Path(result_dir) + if not dir_path.exists(): + return [TextContent( + type="text", + text=f"Error: Directory '{result_dir}' does not exist" + )] + + logs_dir = dir_path / "logs" + if not logs_dir.exists(): + return [TextContent( + type="text", + text=f"Error: Logs directory not found in '{result_dir}'" + )] + + log_files = sorted(logs_dir.glob("*")) + + if not log_files: + return [TextContent( + type="text", + text=f"No log files found in '{logs_dir}'" + )] + + result = f"Log files in {logs_dir}:\n\n" + + # Group by type (stdout vs stderr) + stdout_files = [f for f in log_files if f.name.endswith('.stdout')] + stderr_files = [f for f in log_files if f.name.endswith('.stderr')] + + if stdout_files: + result += "Standard Output Logs:\n" + for f in stdout_files: + size = f.stat().st_size + result += f" {f.name}\n" + result += f" Size: {size:,} bytes\n" + result += f" Path: {f}\n" + result += "\n" + + if stderr_files: + result += "Standard Error Logs:\n" + for f in stderr_files: + size = f.stat().st_size + result += f" {f.name}\n" + result += f" Size: {size:,} bytes\n" + result += f" Path: {f}\n" + result += "\n" + + return [TextContent(type="text", text=result)] + + except Exception as e: + return [TextContent( + type="text", + text=f"Error listing logs: {str(e)}" + )] + + elif name == "read_log": + log_path = arguments["log_path"] + tail = arguments.get("tail") + head = arguments.get("head") + grep = arguments.get("grep") + + try: + path = Path(log_path) + if not path.exists(): + return [TextContent( + type="text", + text=f"Error: Log file '{log_path}' does not exist" + )] + + # Read the file + with open(path, 'r', errors='replace') as f: + lines = f.readlines() + + total_lines = len(lines) + + # Apply filters + if grep: + lines = [line for line in lines if grep in line] + result = f"Log file: {path.name}\n" + result += f"Filtered by: '{grep}'\n" + result += f"Matching lines: {len(lines)} / {total_lines}\n" + result += "=" * 60 + "\n" + else: + result = f"Log file: {path.name}\n" + result += f"Total lines: {total_lines}\n" + result += "=" * 60 + "\n" + + if tail: + lines = lines[-tail:] + result += f"(showing last {tail} lines)\n\n" + elif head: + lines = lines[:head] + result += f"(showing first {head} lines)\n\n" + elif not grep: + # Default: show first 100 lines if no filter specified + if len(lines) > 100: + lines = lines[:100] + result += f"(showing first 100 of {total_lines} lines)\n\n" + + result += "".join(lines) + + # Warn if file was very large + if total_lines > 10000 and not (tail or head or grep): + result += f"\n\n[Note: File has {total_lines} lines. " + result += "Consider using --tail, --head, or --grep to filter]" + + return [TextContent(type="text", text=result)] + + except Exception as e: + return [TextContent( + type="text", + text=f"Error reading log file: {str(e)}" + )] + else: return [TextContent( type="text", -- 2.55.0