diff --git a/cvs/monitors/cluster-mon/Dockerfile b/cvs/monitors/cluster-mon/Dockerfile index 602b1ba4d..bff419355 100644 --- a/cvs/monitors/cluster-mon/Dockerfile +++ b/cvs/monitors/cluster-mon/Dockerfile @@ -1,5 +1,15 @@ # Multi-stage build for CVS Cluster Monitor -# Stage 1: Build React frontend + +# Stage 1: Build Go GPU collector binary +FROM golang:1.22-alpine AS go-builder +WORKDIR /go-src +COPY go-collector/go.mod ./ +RUN go mod download golang.org/x/crypto +COPY go-collector/ ./ +RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 \ + go build -ldflags="-s -w" -o /gpu-collector ./cmd/gpu-collector/ + +# Stage 2: Build React frontend FROM node:18-slim AS frontend-builder WORKDIR /app/frontend @@ -17,7 +27,7 @@ COPY frontend/ ./ RUN npm run build -# Stage 2: Main application image +# Stage 3: Main application image FROM python:3.10-slim # Install system dependencies @@ -43,6 +53,9 @@ COPY backend/app/ ./app/ # Copy frontend build from builder stage COPY --from=frontend-builder /app/frontend/dist ./static +# Copy Go binary from go-builder stage +COPY --from=go-builder /gpu-collector /usr/local/bin/gpu-collector + # Create config directory RUN mkdir -p /app/config diff --git a/cvs/monitors/cluster-mon/backend/app/api/logs.py b/cvs/monitors/cluster-mon/backend/app/api/logs.py index 7f97f5560..378a70041 100644 --- a/cvs/monitors/cluster-mon/backend/app/api/logs.py +++ b/cvs/monitors/cluster-mon/backend/app/api/logs.py @@ -5,12 +5,44 @@ from fastapi import APIRouter, HTTPException, Query from typing import Dict, Any import logging +import re +import time from datetime import datetime logger = logging.getLogger(__name__) router = APIRouter() +# Matches bracketed timestamps: [Thu Jun 6 01:23:45 2026] or [123456.789] +_TIMESTAMP_RE = re.compile(r'\[[\d\s\w:\.]+\]') + + +def _deduplicate_log_section(section: Dict[str, Any], max_occurrences: int = 2) -> Dict[str, Any]: + """ + For each node's log string, strip timestamps and keep at most + max_occurrences of each unique message. Timestamps are preserved + in the output lines — only used for grouping, not removed. + """ + result = {} + for node, log_output in section.items(): + if not isinstance(log_output, str) or not log_output.strip(): + result[node] = log_output + continue + + seen: Dict[str, int] = {} + kept: list = [] + for line in log_output.split('\n'): + if not line.strip(): + continue + normalized = _TIMESTAMP_RE.sub('', line).strip() + count = seen.get(normalized, 0) + if count < max_occurrences: + seen[normalized] = count + 1 + kept.append(line) + + result[node] = '\n'.join(kept) + return result + def validate_grep_command(grep_cmd: str) -> tuple[bool, str]: """ @@ -87,12 +119,31 @@ async def get_dmesg_errors() -> Dict[str, Any]: raise HTTPException(status_code=503, detail="SSH manager not initialized") try: + current_time = time.time() + cache_age = current_time - app_state.logs_cache_time + + # Return cached data if still fresh (180s TTL, same as software caches) + if cache_age < app_state.software_cache_ttl and app_state.cached_logs: + logger.info(f"API: Returning cached logs (age: {cache_age:.0f}s)") + return app_state.cached_logs + from app.collectors.logs_collector import LogsCollector collector = LogsCollector() logs_data = await collector.collect_all_logs(app_state.ssh_manager) - # Count nodes with actual data + # Deduplicate recurring lines (strip timestamps, keep max 2 per unique message) + logs_data = { + **logs_data, + "amd_logs": _deduplicate_log_section(logs_data.get("amd_logs", {})), + "dmesg_errors": _deduplicate_log_section(logs_data.get("dmesg_errors", {})), + "userspace_errors": _deduplicate_log_section(logs_data.get("userspace_errors", {})), + } + + # Update cache + app_state.cached_logs = logs_data + app_state.logs_cache_time = current_time + amd_with_data = sum(1 for v in logs_data.get("amd_logs", {}).values() if isinstance(v, str) and v.strip()) dmesg_with_data = sum(1 for v in logs_data.get("dmesg_errors", {}).values() if isinstance(v, str) and v.strip()) userspace_with_data = sum( @@ -100,12 +151,17 @@ async def get_dmesg_errors() -> Dict[str, Any]: ) logger.info( - f"API: Returning logs - {amd_with_data} nodes with AMD logs, {dmesg_with_data} with dmesg errors, {userspace_with_data} with userspace errors" + f"API: Returning fresh logs - {amd_with_data} nodes with AMD logs, " + f"{dmesg_with_data} with dmesg errors, {userspace_with_data} with userspace errors" ) return logs_data except Exception as e: + # If collection fails but we have cached data, return it + if app_state.cached_logs: + logger.warning(f"API: Log collection failed, returning stale cache: {e}") + return app_state.cached_logs logger.error(f"API: Failed to collect logs: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Failed to collect logs: {str(e)}") @@ -155,24 +211,31 @@ async def search_dmesg_logs( logger.info(f"Grep command validated successfully: {grep_command}") try: - # Build safe command: sudo dmesg -T | | head -5 - # Add head -5 to limit output per node - # IMPORTANT: Use single quotes for outer bash -c, escape single quotes in grep_command - # Bash single quote escaping: replace ' with '\'' (end quote, escaped quote, start quote) + import asyncio + from app.core.go_collector import collect_parallel + + # Build safe command escaped_grep_cmd = grep_command.replace("'", "'\\''") cmd = f"bash -c 'sudo dmesg -T 2>/dev/null | {escaped_grep_cmd} | head -5 || echo \"\"'" - logger.info(f"Executing search on {len(app_state.ssh_manager.get_reachable_hosts())} nodes") + logger.info(f"Executing search on {len(app_state.ssh_manager.get_reachable_hosts())} nodes via Go binary") logger.info(f"Command: {cmd[:200]}...") - # Execute with 60 second timeout - results = await app_state.ssh_manager.exec_async(cmd, timeout=60) + # Run via Go binary (all nodes simultaneously) + go_results = await asyncio.to_thread(collect_parallel, app_state.ssh_manager, {"search": cmd}, 60) + + if go_results is None: + # Fallback to parallel-ssh + logger.info("Go binary unavailable, falling back to parallel-ssh for search") + raw = await app_state.ssh_manager.exec_async(cmd, timeout=60) + else: + raw = go_results.get("search", {}) # Filter out empty results and errors search_results = {} nodes_with_results = 0 - for node, output in results.items(): + for node, output in raw.items(): if output and not output.startswith("ERROR") and not output.startswith("ABORT") and output.strip(): search_results[node] = output.strip() nodes_with_results += 1 @@ -183,7 +246,7 @@ async def search_dmesg_logs( "timestamp": datetime.utcnow().isoformat() + "Z", "grep_command": grep_command, "results": search_results, - "total_nodes_searched": len(results), + "total_nodes_searched": len(raw), "nodes_with_results": nodes_with_results, } diff --git a/cvs/monitors/cluster-mon/backend/app/api/ssh_keys.py b/cvs/monitors/cluster-mon/backend/app/api/ssh_keys.py index 84fc7bd5f..9f8893a99 100644 --- a/cvs/monitors/cluster-mon/backend/app/api/ssh_keys.py +++ b/cvs/monitors/cluster-mon/backend/app/api/ssh_keys.py @@ -20,12 +20,20 @@ async def upload_ssh_key(file: UploadFile = File(...)) -> Dict[str, Any]: try: # Validate file if not file.filename: + logger.error("SSH key upload rejected: no filename provided") raise HTTPException(status_code=400, detail="No filename provided") - # Only allow common SSH key filenames for security - allowed_names = ["id_rsa", "id_ed25519", "id_ecdsa", "cluster_id_ed25519", "known_hosts", "config"] - if file.filename not in allowed_names: - raise HTTPException(status_code=400, detail=f"Invalid key filename. Allowed: {', '.join(allowed_names)}") + # Only allow safe SSH key filenames (alphanumeric, underscore, hyphen, dot) + import re + + if not re.match(r'^[a-zA-Z0-9._-]+$', file.filename) or '/' in file.filename or '..' in file.filename: + logger.error( + f"SSH key upload rejected: invalid filename '{file.filename}'. Use only alphanumeric characters, underscores, hyphens, and dots." + ) + raise HTTPException( + status_code=400, + detail="Invalid key filename. Use only alphanumeric characters, underscores, hyphens, and dots.", + ) # Create .ssh directory if it doesn't exist ssh_dir = Path("/root/.ssh") @@ -77,6 +85,8 @@ async def upload_ssh_key(file: UploadFile = File(...)) -> Dict[str, Any]: "path": str(key_path), } + except HTTPException: + raise except Exception as e: logger.error(f"Failed to upload SSH key: {e}") raise HTTPException(status_code=500, detail=f"Failed to upload SSH key: {str(e)}") @@ -118,9 +128,11 @@ async def delete_ssh_key(filename: str) -> Dict[str, Any]: Delete an SSH key from the container. """ try: - # Security: only allow deleting SSH key files - allowed_names = ["id_rsa", "id_ed25519", "id_ecdsa", "cluster_id_ed25519", "known_hosts", "config"] - if filename not in allowed_names: + # Security: only allow deleting SSH key files with safe filenames + import re + + if not re.match(r'^[a-zA-Z0-9._-]+$', filename) or '/' in filename or '..' in filename: + logger.error(f"SSH key delete rejected: invalid filename '{filename}'") raise HTTPException(status_code=400, detail="Invalid key filename") key_path = Path(f"/root/.ssh/{filename}") diff --git a/cvs/monitors/cluster-mon/backend/app/collectors/gpu_collector.py b/cvs/monitors/cluster-mon/backend/app/collectors/gpu_collector.py index 1a56c8f49..798fe2a78 100644 --- a/cvs/monitors/cluster-mon/backend/app/collectors/gpu_collector.py +++ b/cvs/monitors/cluster-mon/backend/app/collectors/gpu_collector.py @@ -5,7 +5,7 @@ import json import logging -from typing import Dict, Any +from typing import Dict, Any, Optional from datetime import datetime logger = logging.getLogger(__name__) @@ -311,40 +311,75 @@ async def collect_pcie_info(self, ssh_manager) -> Dict[str, Any]: return pcie_info + _GPU_COMMANDS = { + "metric": "amd-smi metric --json", + "pcie": "amd-smi metric --pcie --json", + "xgmi": "amd-smi metric --xgmi-err --json", + "ecc": "amd-smi metric --ecc --json", + } + + def _collect_via_go_binary(self, ssh_manager) -> Optional[Dict[str, Any]]: + """ + Collect all GPU metrics via the Go binary (all nodes + all commands in parallel). + Returns the metrics dict or None if unavailable (falls back to parallel-ssh). + """ + from app.core.go_collector import collect_parallel + + results = collect_parallel(ssh_manager, self._GPU_COMMANDS, timeout=90) + if results is None: + return None + + amd_smi_data = self.parse_json_output(results.get("metric", {})) + utilization = self._parse_utilization_from_amd_smi(amd_smi_data) + memory = self._parse_memory_from_amd_smi(amd_smi_data) + temperature = self._parse_temperature_from_amd_smi(amd_smi_data) + pcie_data = self.parse_json_output(results.get("pcie", {})) + xgmi_data = self.parse_json_output(results.get("xgmi", {})) + ecc_data = self.parse_json_output(results.get("ecc", {})) + pcie_info = self._parse_pcie_metrics_from_amd_smi(pcie_data) + + return { + "timestamp": datetime.utcnow().isoformat() + "Z", + "utilization": utilization, + "memory": memory, + "temperature": temperature, + "power": amd_smi_data, + "pcie": pcie_info, + "xgmi": xgmi_data, + "ras_errors": ecc_data, + "pcie_link_status": pcie_info, + "info": amd_smi_data, + } + async def collect_all_metrics(self, ssh_manager) -> Dict[str, Any]: """ Collect all GPU metrics. - Optimized to call amd-smi metric --json once and parse all data from it. - Returns: - { - "timestamp": "2025-02-11T12:00:00Z", - "utilization": {...}, - "memory": {...}, - "temperature": {...}, - "power": {...}, - "pcie": {...}, - "xgmi": {...}, - "ras_errors": {...}, - "info": {...} - } + Fast path: Go binary SSHes all nodes simultaneously, all commands + in parallel per node (~60-90s for 165 nodes). + + Fallback: sequential parallel-ssh calls if binary unavailable. """ import asyncio logger.info("Collecting all GPU metrics") - # OPTIMIZATION: Call amd-smi metric --json ONCE to get ALL data - # This single command includes: utilization, memory, temperature, PCIe, XGMI, and ECC metrics + # Fast path via Go binary + result = await asyncio.to_thread(self._collect_via_go_binary, ssh_manager) + if result is not None: + logger.info("GPU metrics collected via Go binary") + return result + + # Fallback: original parallel-ssh sequential path + logger.info("Falling back to parallel-ssh for GPU metrics collection") logger.info("Calling amd-smi metric --json for comprehensive GPU data") amd_smi_output = await asyncio.to_thread(ssh_manager.exec, "amd-smi metric --json") amd_smi_data = self.parse_json_output(amd_smi_output) - # Parse all metrics from single amd-smi output utilization = self._parse_utilization_from_amd_smi(amd_smi_data) memory = self._parse_memory_from_amd_smi(amd_smi_data) temperature = self._parse_temperature_from_amd_smi(amd_smi_data) - # Call dedicated commands for PCIe and ECC for cleaner data logger.info("Collecting PCIe metrics with dedicated command") pcie_output = await asyncio.to_thread(ssh_manager.exec, "amd-smi metric --pcie --json") pcie_data = self.parse_json_output(pcie_output) @@ -357,28 +392,24 @@ async def collect_all_metrics(self, ssh_manager) -> Dict[str, Any]: ecc_output = await asyncio.to_thread(ssh_manager.exec, "amd-smi metric --ecc --json") ecc_data = self.parse_json_output(ecc_output) - # Parse for frontend display pcie_info = self._parse_pcie_metrics_from_amd_smi(pcie_data) logger.info(f"Parsed PCIE data: {len(pcie_info)} nodes") logger.info(f"ECC data (raw): {len(ecc_data)} nodes") - # Package results - metrics = { + return { "timestamp": datetime.utcnow().isoformat() + "Z", "utilization": utilization, "memory": memory, "temperature": temperature, - "power": amd_smi_data, # Power is in the main amd-smi output - "pcie": pcie_info, # Parsed PCIE data + "power": amd_smi_data, + "pcie": pcie_info, "xgmi": xgmi_data, - "ras_errors": ecc_data, # Raw ECC data from dedicated command - "pcie_link_status": pcie_info, # For backward compatibility - "info": amd_smi_data, # GPU info also in amd-smi output + "ras_errors": ecc_data, + "pcie_link_status": pcie_info, + "info": amd_smi_data, } - return metrics - def _parse_utilization_from_amd_smi(self, amd_smi_data: Dict) -> Dict: """Parse utilization from amd-smi metric output.""" util_data = {} diff --git a/cvs/monitors/cluster-mon/backend/app/collectors/gpu_software_collector.py b/cvs/monitors/cluster-mon/backend/app/collectors/gpu_software_collector.py index 89de9de1e..cd27558b2 100644 --- a/cvs/monitors/cluster-mon/backend/app/collectors/gpu_software_collector.py +++ b/cvs/monitors/cluster-mon/backend/app/collectors/gpu_software_collector.py @@ -183,37 +183,10 @@ async def collect_rocm_libraries(self, ssh_manager) -> Dict[str, Any]: return lib_info - async def collect_all_software_info(self, ssh_manager) -> Dict[str, Any]: - """ - Collect all GPU software information. - - OPTIMIZATION: Use minimal commands: - - amd-smi version --json (for ROCm, AMDSMI, and amdgpu driver versions) - - amd-smi firmware --json (for firmware versions per GPU) - - amd-smi version --json output format: - [{ - "tool": "AMDSMI Tool", - "version": "26.2.0+021c61fc", - "amdsmi_library_version": "26.2.0", - "rocm_version": "7.0.2", - "amdgpu_version": "6.16.6", - "amd_hsmp_driver_version": "N/A" - }] - - Returns consolidated software info for all nodes. - """ - - logger.info("Collecting all GPU software information (optimized)") - - # IMPORTANT: Run commands SEQUENTIALLY to avoid parallel-ssh thread safety issues - # asyncio.gather() was causing "munmap_chunk(): invalid pointer" crashes - version_output = await ssh_manager.exec_async("amd-smi version --json", timeout=60) - firmware_output = await ssh_manager.exec_async("amd-smi firmware --json", timeout=120) - - # Parse amd-smi version --json output + def _parse_version_output(self, version_output: Dict[str, str]) -> Dict[str, Any]: + """Parse amd-smi version --json output into rocm_version_info dict.""" rocm_version_info = {} - for host, out_str in (version_output if isinstance(version_output, dict) else {}).items(): + for host, out_str in version_output.items(): if not out_str.startswith("ERROR") and not out_str.startswith("ABORT"): try: version_data = json.loads(out_str.strip()) @@ -233,14 +206,40 @@ async def collect_all_software_info(self, ssh_manager) -> Dict[str, Any]: rocm_version_info[host] = {'rocm_version': 'N/A', 'amdgpu_version': 'N/A'} else: rocm_version_info[host] = {'rocm_version': 'N/A', 'amdgpu_version': 'N/A'} + return rocm_version_info + + async def collect_all_software_info(self, ssh_manager) -> Dict[str, Any]: + """ + Collect all GPU software information. - # Parse firmware - gpu_firmware = self.parse_json_output(firmware_output) if isinstance(firmware_output, dict) else {} + Fast path: Go binary collects amd-smi version + firmware from all nodes + simultaneously in parallel. - software_info = { - "timestamp": datetime.utcnow().isoformat() + "Z", - "rocm_version": rocm_version_info, - "gpu_firmware": gpu_firmware, + Fallback: sequential parallel-ssh calls. + """ + import asyncio + from app.core.go_collector import collect_parallel + + logger.info("Collecting all GPU software information (optimized)") + + commands = { + "version": "amd-smi version --json", + "firmware": "amd-smi firmware --json", } - return software_info + go_results = await asyncio.to_thread(collect_parallel, ssh_manager, commands, 60) + + if go_results is not None: + logger.info("GPU software collected via Go binary") + version_output = go_results.get("version", {}) + firmware_output = go_results.get("firmware", {}) + else: + logger.info("Falling back to sequential parallel-ssh for GPU software") + version_output = await ssh_manager.exec_async("amd-smi version --json", timeout=60) + firmware_output = await ssh_manager.exec_async("amd-smi firmware --json", timeout=120) + + return { + "timestamp": datetime.utcnow().isoformat() + "Z", + "rocm_version": self._parse_version_output(version_output), + "gpu_firmware": self.parse_json_output(firmware_output), + } diff --git a/cvs/monitors/cluster-mon/backend/app/collectors/logs_collector.py b/cvs/monitors/cluster-mon/backend/app/collectors/logs_collector.py index c1d88490e..6b079b686 100644 --- a/cvs/monitors/cluster-mon/backend/app/collectors/logs_collector.py +++ b/cvs/monitors/cluster-mon/backend/app/collectors/logs_collector.py @@ -12,7 +12,7 @@ class LogsCollector: """Collects system error logs from dmesg.""" - async def collect_dmesg_errors(self, ssh_manager) -> Dict[str, Any]: + async def collect_dmesg_errors(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect critical system errors from dmesg. @@ -29,7 +29,7 @@ async def collect_dmesg_errors(self, ssh_manager) -> Dict[str, Any]: cmd = "bash -c 'sudo dmesg --decode -T -l emerg,alert,crit,err 2>/dev/null || echo \"\"'" logger.info(f"Running command: {cmd}") - output = await ssh_manager.exec_async(cmd, timeout=60) + output = preloaded_output if preloaded_output is not None else await ssh_manager.exec_async(cmd, timeout=60) logger.info(f"Received output from {len(output)} nodes") @@ -58,7 +58,7 @@ async def collect_dmesg_errors(self, ssh_manager) -> Dict[str, Any]: ) return logs - async def collect_amd_logs(self, ssh_manager) -> Dict[str, Any]: + async def collect_amd_logs(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect AMD-specific hardware and driver error logs from dmesg. @@ -82,7 +82,7 @@ async def collect_amd_logs(self, ssh_manager) -> Dict[str, Any]: # cmd = """bash -c 'sudo dmesg --decode -T -l emerg,alert,crit,err,warn 2>/dev/null | grep -iE "PCIe|XGMI|amdgpu|epyc|cpu|ionic|bnxt|mlnx|mellanox|Link" 2>/dev/null || echo ""'""" cmd = """bash -c 'sudo dmesg --decode -T -l emerg,alert,crit,err,warn 2>/dev/null | grep -iE "PCIe|XGMI|amdgpu|epyc|cpu|ionic|bnxt|mlnx|mellanox|Link|error|fail" 2>/dev/null | grep -iv "vital buffer" 2>/dev/null || echo ""'""" logger.info(f"Running command: {cmd[:150]}...") - output = await ssh_manager.exec_async(cmd, timeout=60) + output = preloaded_output if preloaded_output is not None else await ssh_manager.exec_async(cmd, timeout=60) logger.info(f"Received output from {len(output)} nodes") @@ -111,7 +111,7 @@ async def collect_amd_logs(self, ssh_manager) -> Dict[str, Any]: ) return logs - async def collect_userspace_errors(self, ssh_manager) -> Dict[str, Any]: + async def collect_userspace_errors(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect userspace errors including OOM, segfaults, crashes, and ML framework errors. @@ -134,7 +134,7 @@ async def collect_userspace_errors(self, ssh_manager) -> Dict[str, Any]: # Use -l to filter levels first, then egrep for userspace patterns cmd = """bash -c 'sudo dmesg --decode -T -l emerg,alert,crit,err,warn 2>/dev/null | egrep -i "oom|out of memory|killed process|segfault|general protection|call trace|bug:|hardware error|mce|stack trace|pytorch|torch|tensorflow|megatron|jax|vllm|sglang|triton.*error|triton.*exception|triton.*failed" 2>/dev/null || echo ""'""" logger.info(f"Running command: {cmd[:150]}...") - output = await ssh_manager.exec_async(cmd, timeout=60) + output = preloaded_output if preloaded_output is not None else await ssh_manager.exec_async(cmd, timeout=60) logger.info(f"Received output from {len(output)} nodes") @@ -167,11 +167,29 @@ async def collect_all_logs(self, ssh_manager) -> Dict[str, Any]: """ Collect all system logs including AMD-specific hardware/driver logs. """ + import asyncio + from app.core.go_collector import collect_parallel + logger.info("Collecting all system logs") - amd_logs = await self.collect_amd_logs(ssh_manager) - dmesg_logs = await self.collect_dmesg_errors(ssh_manager) - userspace_logs = await self.collect_userspace_errors(ssh_manager) + commands = { + "amd_logs": """bash -c 'sudo dmesg --decode -T -l emerg,alert,crit,err,warn 2>/dev/null | grep -iE "PCIe|XGMI|amdgpu|epyc|cpu|ionic|bnxt|mlnx|mellanox|Link|error|fail" 2>/dev/null | grep -iv "vital buffer" 2>/dev/null || echo ""'""", + "dmesg_errors": "bash -c 'sudo dmesg --decode -T -l emerg,alert,crit,err 2>/dev/null || echo \"\"'", + "userspace_logs": """bash -c 'sudo dmesg --decode -T -l emerg,alert,crit,err,warn 2>/dev/null | egrep -i "oom|out of memory|killed process|segfault|general protection|call trace|bug:|hardware error|mce|stack trace|pytorch|torch|tensorflow|megatron|jax|vllm|sglang|triton.*error|triton.*exception|triton.*failed" 2>/dev/null || echo ""'""", + } + + go_results = await asyncio.to_thread(collect_parallel, ssh_manager, commands, 60) + + if go_results is not None: + logger.info("System logs collected via Go binary") + amd_logs = await self.collect_amd_logs(preloaded_output=go_results.get("amd_logs", {})) + dmesg_logs = await self.collect_dmesg_errors(preloaded_output=go_results.get("dmesg_errors", {})) + userspace_logs = await self.collect_userspace_errors(preloaded_output=go_results.get("userspace_logs", {})) + else: + logger.info("Falling back to sequential parallel-ssh for system logs") + amd_logs = await self.collect_amd_logs(ssh_manager) + dmesg_logs = await self.collect_dmesg_errors(ssh_manager) + userspace_logs = await self.collect_userspace_errors(ssh_manager) return { "timestamp": datetime.utcnow().isoformat() + "Z", diff --git a/cvs/monitors/cluster-mon/backend/app/collectors/nic_advanced_collector.py b/cvs/monitors/cluster-mon/backend/app/collectors/nic_advanced_collector.py index 09edeffc5..3a355a5d2 100644 --- a/cvs/monitors/cluster-mon/backend/app/collectors/nic_advanced_collector.py +++ b/cvs/monitors/cluster-mon/backend/app/collectors/nic_advanced_collector.py @@ -15,7 +15,7 @@ class NICAdvancedCollector: """Collects vendor-specific NIC information and congestion metrics.""" - async def collect_nic_pcie_info(self, ssh_manager) -> Dict[str, Any]: + async def collect_nic_pcie_info(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect PCIe information for all NICs using lspci. Optimized: single lspci call per node instead of per NIC. @@ -24,8 +24,11 @@ async def collect_nic_pcie_info(self, ssh_manager) -> Dict[str, Any]: # Get ALL NIC PCIe info in one command per node # cmd = "sudo lspci -vvv 2>/dev/null | grep -A 30 -i 'ethernet\\|network' | grep -E '^[0-9a-f]{2}:|Ethernet|Network|LnkCap:|LnkSta:'" - cmd = "sudo lspci -vvv 2>/dev/null | egrep -A 30 -i 'ethernet\\|network' | egrep '^[0-9a-f]{2}:|Ethernet|Network|LnkCap:|LnkSta:'" - result = await ssh_manager.exec_async(cmd, timeout=120) + if preloaded_output is not None: + result = preloaded_output + else: + cmd = "sudo lspci -vvv 2>/dev/null | egrep -A 30 -i 'ethernet\\|network' | egrep '^[0-9a-f]{2}:|Ethernet|Network|LnkCap:|LnkSta:'" + result = await ssh_manager.exec_async(cmd, timeout=120) logger.info(f"NIC PCIe lspci returned results from {len(result)} nodes") @@ -134,7 +137,7 @@ async def collect_nic_pcie_info(self, ssh_manager) -> Dict[str, Any]: ) return pcie_info - async def collect_congestion_info(self, ssh_manager) -> Dict[str, Any]: + async def collect_congestion_info(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect congestion control information (PFC, ECN, CNP). @@ -153,11 +156,12 @@ async def collect_congestion_info(self, ssh_manager) -> Dict[str, Any]: Looks for fields matching patterns: pfc, pause, ecn, cnp, drop, err, timeout """ logger.info("Collecting congestion control information from rdma statistic (optimized)") - - # Use rdma statistic show with JSON for reliable parsing - output = await ssh_manager.exec_async( - "bash -c 'rdma statistic show --json 2>/dev/null || echo \"[]\"'", timeout=60 - ) + if preloaded_output is not None: + output = preloaded_output + else: + output = await ssh_manager.exec_async( + "bash -c 'rdma statistic show --json 2>/dev/null || echo \"[]\"'", timeout=60 + ) congestion_info = {} @@ -234,12 +238,27 @@ async def collect_all_nic_advanced_info(self, ssh_manager) -> Dict[str, Any]: Collect all advanced NIC information. """ + import asyncio + from app.core.go_collector import collect_parallel + logger.info("Collecting all advanced NIC information") - # IMPORTANT: Run commands SEQUENTIALLY to avoid parallel-ssh thread safety issues - # asyncio.gather() was causing "munmap_chunk(): invalid pointer" crashes - nic_pcie = await self.collect_nic_pcie_info(ssh_manager) - congestion = await self.collect_congestion_info(ssh_manager) + commands = { + "nic_pcie": "sudo lspci -vvv 2>/dev/null | egrep -A 30 -i 'ethernet\\|network' | egrep '^[0-9a-f]{2}:|Ethernet|Network|LnkCap:|LnkSta:'", + "congestion": "bash -c 'rdma statistic show --json 2>/dev/null || echo \"[]\"'", + } + + go_results = await asyncio.to_thread(collect_parallel, ssh_manager, commands, 90) + + if go_results is not None: + logger.info("NIC advanced info collected via Go binary") + nic_pcie = await self.collect_nic_pcie_info(preloaded_output=go_results.get("nic_pcie", {})) + congestion = await self.collect_congestion_info(preloaded_output=go_results.get("congestion", {})) + else: + logger.info("Falling back to sequential parallel-ssh for NIC advanced info") + nic_pcie = await self.collect_nic_pcie_info(ssh_manager) + congestion = await self.collect_congestion_info(ssh_manager) + mellanox = await self.collect_mellanox_info(ssh_manager) broadcom = await self.collect_broadcom_info(ssh_manager) diff --git a/cvs/monitors/cluster-mon/backend/app/collectors/nic_collector.py b/cvs/monitors/cluster-mon/backend/app/collectors/nic_collector.py index c0a5e0795..0c233add0 100644 --- a/cvs/monitors/cluster-mon/backend/app/collectors/nic_collector.py +++ b/cvs/monitors/cluster-mon/backend/app/collectors/nic_collector.py @@ -15,7 +15,7 @@ class NICMetricsCollector: """Collects NIC metrics via rdma, ethtool, and ip commands.""" - async def collect_rdma_links(self, ssh_manager) -> Dict[str, Any]: + async def collect_rdma_links(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect RDMA link information. @@ -33,7 +33,7 @@ async def collect_rdma_links(self, ssh_manager) -> Dict[str, Any]: } """ logger.info("Collecting RDMA link info") - output = ssh_manager.exec("rdma link", timeout=60) + output = preloaded_output if preloaded_output is not None else ssh_manager.exec("rdma link", timeout=60) rdma_dict = {} for node, out_str in output.items(): @@ -59,7 +59,7 @@ async def collect_rdma_links(self, ssh_manager) -> Dict[str, Any]: return rdma_dict - async def collect_rdma_stats(self, ssh_manager) -> Dict[str, Any]: + async def collect_rdma_stats(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect RDMA statistics including congestion control metrics. @@ -81,8 +81,11 @@ async def collect_rdma_stats(self, ssh_manager) -> Dict[str, Any]: } """ logger.info("Collecting RDMA statistics (includes congestion control metrics)") - # Use bash -c to properly handle shell redirection and || operator - output = ssh_manager.exec("bash -c 'rdma statistic show --json 2>/dev/null || echo \"{}\"'", timeout=60) + output = ( + preloaded_output + if preloaded_output is not None + else ssh_manager.exec("bash -c 'rdma statistic show --json 2>/dev/null || echo \"{}\"'", timeout=60) + ) logger.info(f"RDMA stats output received from {len(output)} nodes") @@ -135,7 +138,9 @@ async def collect_rdma_stats(self, ssh_manager) -> Dict[str, Any]: logger.info(f"RDMA stats collection complete: {len(rdma_stats)} nodes with data") return rdma_stats - async def collect_ethtool_stats(self, ssh_manager, interfaces: Dict[str, list] = None) -> Dict[str, Any]: + async def collect_ethtool_stats( + self, ssh_manager=None, interfaces: Dict[str, list] = None, preloaded_output=None + ) -> Dict[str, Any]: """ Collect network interface statistics using 'ip -s link' (optimized). @@ -162,10 +167,7 @@ async def collect_ethtool_stats(self, ssh_manager, interfaces: Dict[str, list] = } """ logger.info("Collecting network statistics via 'ip -s link' (optimized)") - - # Run 'ip -s link' once per node to get all interface stats - cmd = "ip -s link show" - output = ssh_manager.exec(cmd, timeout=60) + output = preloaded_output if preloaded_output is not None else ssh_manager.exec("ip -s link show", timeout=60) eth_stats = {} @@ -243,7 +245,7 @@ async def collect_ethtool_stats(self, ssh_manager, interfaces: Dict[str, list] = return eth_stats - async def collect_ip_addr(self, ssh_manager) -> Dict[str, Any]: + async def collect_ip_addr(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect IP address information. @@ -263,7 +265,11 @@ async def collect_ip_addr(self, ssh_manager) -> Dict[str, Any]: } """ logger.info("Collecting IP address info") - output = ssh_manager.exec("bash -c 'ip addr show | grep -A 5 mtu --color=never'", timeout=60) + output = ( + preloaded_output + if preloaded_output is not None + else ssh_manager.exec("bash -c 'ip addr show | grep -A 5 mtu --color=never'", timeout=60) + ) ip_dict = {} @@ -320,7 +326,7 @@ async def collect_ip_addr(self, ssh_manager) -> Dict[str, Any]: return ip_dict - async def collect_lldp(self, ssh_manager) -> Dict[str, Any]: + async def collect_lldp(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect LLDP neighbor information (if lldpctl is available). @@ -338,8 +344,11 @@ async def collect_lldp(self, ssh_manager) -> Dict[str, Any]: } """ logger.info("Collecting LLDP info") - # Use bash -c to properly handle shell redirection and || operator - output = ssh_manager.exec("bash -c 'sudo lldpctl -f json 2>/dev/null || echo \"{}\"'", timeout=60) + output = ( + preloaded_output + if preloaded_output is not None + else ssh_manager.exec("bash -c 'sudo lldpctl -f json 2>/dev/null || echo \"{}\"'", timeout=60) + ) lldp_dict = {} for node, out_str in output.items(): @@ -435,48 +444,57 @@ async def collect_all_metrics(self, ssh_manager) -> Dict[str, Any]: } """ - logger.info("Collecting all NIC metrics") - - # Collect SEQUENTIALLY (one command completes before next starts) - rdma_links = await self.collect_rdma_links(ssh_manager) - rdma_stats = await self.collect_rdma_stats(ssh_manager) - ip_addr = await self.collect_ip_addr(ssh_manager) - lldp = await self.collect_lldp(ssh_manager) - results = [rdma_links, rdma_stats, ip_addr, lldp] - - # Get interface info first for ethtool - ip_data = results[2] if not isinstance(results[2], Exception) else {} + import asyncio + from app.core.go_collector import collect_parallel - # Build interface list for ethtool - interfaces = {} - for node, iface_data in ip_data.items(): - if isinstance(iface_data, dict) and "error" not in iface_data: - interfaces[node] = list(iface_data.keys()) + logger.info("Collecting all NIC metrics") - # Collect ethtool stats with interface list - ethtool_stats = await self.collect_ethtool_stats(ssh_manager, interfaces) + commands = { + "rdma_links": "rdma link", + "rdma_stats": "bash -c 'rdma statistic show --json 2>/dev/null || echo \"{}\"'", + "ip_addr": "bash -c 'ip addr show | grep -A 5 mtu --color=never'", + "lldp": "bash -c 'sudo lldpctl -f json 2>/dev/null || echo \"{}\"'", + "ip_stats": "ip -s link show", + "rdma_res": "rdma res", + } - # Collect RDMA resources - rdma_res = await self.collect_rdma_resources(ssh_manager) + go_results = await asyncio.to_thread(collect_parallel, ssh_manager, commands, 60) + + if go_results is not None: + logger.info("NIC metrics collected via Go binary") + rdma_links = await self.collect_rdma_links(preloaded_output=go_results.get("rdma_links", {})) + rdma_stats = await self.collect_rdma_stats(preloaded_output=go_results.get("rdma_stats", {})) + ip_addr = await self.collect_ip_addr(preloaded_output=go_results.get("ip_addr", {})) + lldp = await self.collect_lldp(preloaded_output=go_results.get("lldp", {})) + ethtool_stats = await self.collect_ethtool_stats(preloaded_output=go_results.get("ip_stats", {})) + rdma_res = await self.collect_rdma_resources(preloaded_output=go_results.get("rdma_res", {})) + else: + logger.info("Falling back to sequential parallel-ssh for NIC metrics") + rdma_links = await self.collect_rdma_links(ssh_manager) + rdma_stats = await self.collect_rdma_stats(ssh_manager) + ip_addr = await self.collect_ip_addr(ssh_manager) + lldp = await self.collect_lldp(ssh_manager) + interfaces = { + node: list(iface_data.keys()) + for node, iface_data in ip_addr.items() + if isinstance(iface_data, dict) and "error" not in iface_data + } + ethtool_stats = await self.collect_ethtool_stats(ssh_manager, interfaces) + rdma_res = await self.collect_rdma_resources(ssh_manager) - # Filter LLDP data to only include RDMA interfaces - rdma_links_data = results[0] if not isinstance(results[0], Exception) else {} - lldp_data = results[3] if not isinstance(results[3], Exception) else {} - filtered_lldp = self._filter_lldp_by_rdma(lldp_data, rdma_links_data) + filtered_lldp = self._filter_lldp_by_rdma(lldp, rdma_links) - metrics = { + return { "timestamp": datetime.utcnow().isoformat() + "Z", - "rdma_links": rdma_links_data, - "rdma_stats": results[1] if not isinstance(results[1], Exception) else {}, + "rdma_links": rdma_links, + "rdma_stats": rdma_stats, "rdma_resources": rdma_res, - "ip_addr": ip_data, + "ip_addr": ip_addr, "lldp": filtered_lldp, "ethtool_stats": ethtool_stats, } - return metrics - - async def collect_rdma_resources(self, ssh_manager) -> Dict[str, Any]: + async def collect_rdma_resources(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect RDMA resources (pd, cq, qp, mr, etc.). @@ -490,7 +508,7 @@ async def collect_rdma_resources(self, ssh_manager) -> Dict[str, Any]: } """ logger.info("Collecting RDMA resources") - output = ssh_manager.exec("rdma res", timeout=60) + output = preloaded_output if preloaded_output is not None else ssh_manager.exec("rdma res", timeout=60) rdma_res = {} for node, out_str in output.items(): diff --git a/cvs/monitors/cluster-mon/backend/app/collectors/nic_devlink_collector.py b/cvs/monitors/cluster-mon/backend/app/collectors/nic_devlink_collector.py index e0e4e17c6..49787bac7 100644 --- a/cvs/monitors/cluster-mon/backend/app/collectors/nic_devlink_collector.py +++ b/cvs/monitors/cluster-mon/backend/app/collectors/nic_devlink_collector.py @@ -15,7 +15,7 @@ class NICDevlinkCollector: """Collects NIC information via devlink.""" - async def collect_devlink_info(self, ssh_manager) -> Dict[str, Any]: + async def collect_devlink_info(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect NIC device information using 'devlink dev info --json'. @@ -46,9 +46,12 @@ async def collect_devlink_info(self, ssh_manager) -> Dict[str, Any]: } """ logger.info("Collecting NIC devlink information") - output = await ssh_manager.exec_async( - "bash -c 'devlink dev info --json 2>/dev/null || echo \"{}\"'", timeout=60 - ) + if preloaded_output is not None: + output = preloaded_output + else: + output = await ssh_manager.exec_async( + "bash -c 'devlink dev info --json 2>/dev/null || echo \"{}\"'", timeout=60 + ) devlink_info = {} @@ -141,9 +144,23 @@ async def collect_all_devlink_info(self, ssh_manager) -> Dict[str, Any]: """ Collect all NIC devlink information. """ + import asyncio + from app.core.go_collector import collect_parallel + logger.info("Collecting all NIC devlink information") - devlink_info = await self.collect_devlink_info(ssh_manager) + commands = { + "devlink": "bash -c 'devlink dev info --json 2>/dev/null || echo \"{}\"'", + } + + go_results = await asyncio.to_thread(collect_parallel, ssh_manager, commands, 60) + + if go_results is not None: + logger.info("NIC devlink info collected via Go binary") + devlink_info = await self.collect_devlink_info(preloaded_output=go_results.get("devlink", {})) + else: + logger.info("Falling back to sequential parallel-ssh for NIC devlink info") + devlink_info = await self.collect_devlink_info(ssh_manager) return { "timestamp": datetime.utcnow().isoformat() + "Z", diff --git a/cvs/monitors/cluster-mon/backend/app/collectors/nic_software_collector.py b/cvs/monitors/cluster-mon/backend/app/collectors/nic_software_collector.py index 7ca489fd2..7311764bd 100644 --- a/cvs/monitors/cluster-mon/backend/app/collectors/nic_software_collector.py +++ b/cvs/monitors/cluster-mon/backend/app/collectors/nic_software_collector.py @@ -16,7 +16,49 @@ class NICSoftwareCollector: """Collects NIC software, firmware, and detailed statistics.""" - async def collect_nic_firmware_version(self, ssh_manager) -> Dict[str, Any]: + # Combined single-command versions of each collection (used by Go binary fast path) + _CMD_FIRMWARE = ( + r"bash -c '" + r"for iface in $(ip -o link show | awk -F\": \" \"{print \$2}\" | grep -v lo | grep -v @ | head -10); do " + r"printf \"===IFACE:%s===\n\" \"$iface\"; " + r"sudo ethtool -i \"$iface\" 2>/dev/null; " + r"done'" + ) + _CMD_DRIVERS = ( + r"bash -c '" + r"printf \"===DRIVER:mlx5_core===\n\"; modinfo mlx5_core 2>/dev/null | grep -E \"^version|^firmware\" | head -3; " + r"printf \"===DRIVER:bnxt_en===\n\"; modinfo bnxt_en 2>/dev/null | grep -E \"^version|^firmware\" | head -3; " + r"printf \"===DRIVER:amd-ainic===\n\"; modinfo amd-ainic 2>/dev/null | grep -E \"^version|^firmware\" | head -3 || printf \"not loaded\n\"'" + ) + _CMD_RDMA = "bash -c 'rdma statistic show --json 2>/dev/null || echo \"[]\"'" + _CMD_ETHTOOL = ( + r"bash -c '" + r"for iface in $(ip -o link show | awk -F\": \" \"{print \$2}\" | grep -v lo | grep -v @ | head -10); do " + r"printf \"===IFACE:%s===\n\" \"$iface\"; " + r"sudo ethtool -S \"$iface\" 2>/dev/null; " + r"done'" + ) + _CMD_PCI = "bash -c \"lspci -nn | grep -i 'network\\|ethernet'\"" + + @staticmethod + def _parse_iface_sections(output_str: str) -> Dict[str, str]: + """Split combined per-interface output into {iface: raw_block} dict.""" + sections: Dict[str, str] = {} + current: str | None = None + lines: list = [] + for line in output_str.split('\n'): + if line.startswith('===IFACE:') and line.endswith('==='): + if current is not None: + sections[current] = '\n'.join(lines) + current = line[9:-3] + lines = [] + elif current is not None: + lines.append(line) + if current is not None: + sections[current] = '\n'.join(lines) + return sections + + async def collect_nic_firmware_version(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect NIC firmware versions. @@ -24,25 +66,39 @@ async def collect_nic_firmware_version(self, ssh_manager) -> Dict[str, Any]: """ logger.info("Collecting NIC firmware versions") - # First get list of interfaces + if preloaded_output is not None: + # Combined output: one string per host containing all interfaces + firmware_info = {} + for host, combined_str in preloaded_output.items(): + if not combined_str or combined_str.startswith("ERROR") or combined_str.startswith("ABORT"): + firmware_info[host] = {"error": combined_str} + continue + firmware_info[host] = {} + for iface, block in self._parse_iface_sections(combined_str).items(): + info = {} + for line in block.split("\n"): + if ":" in line: + key, value = line.split(":", 1) + key = key.strip().lower().replace(" ", "_") + info[key] = value.strip() + if info: + firmware_info[host][iface] = info + return firmware_info + + # Fallback: original sequential parallel-ssh approach ip_output = await ssh_manager.exec_async( "bash -c \"ip -o link show | awk -F': ' '{print \\$2}' | grep -v lo\"", timeout=60 ) - firmware_info = {} - for host, ifaces_str in ip_output.items(): if ifaces_str.startswith("ERROR") or ifaces_str.startswith("ABORT"): firmware_info[host] = {"error": ifaces_str} continue - firmware_info[host] = {} interfaces = [i.strip() for i in ifaces_str.split("\n") if i.strip() and "@" not in i] - - for iface in interfaces[:10]: # Limit to first 10 interfaces + for iface in interfaces[:10]: cmd = f"sudo ethtool -i {iface} 2>/dev/null" output = await ssh_manager.exec_async(cmd, timeout=60) - if host in output and output[host]: info = {} for line in output[host].split("\n"): @@ -50,13 +106,11 @@ async def collect_nic_firmware_version(self, ssh_manager) -> Dict[str, Any]: key, value = line.split(":", 1) key = key.strip().lower().replace(" ", "_") info[key] = value.strip() - if info: firmware_info[host][iface] = info - return firmware_info - async def collect_nic_driver_version(self, ssh_manager) -> Dict[str, Any]: + async def collect_nic_driver_version(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect NIC driver versions for different vendors. @@ -67,61 +121,79 @@ async def collect_nic_driver_version(self, ssh_manager) -> Dict[str, Any]: """ logger.info("Collecting NIC driver versions") + def _parse_driver_block(block: str, driver_name: str) -> dict: + info = {} + for line in block.split("\n"): + if ":" in line and "modinfo" not in line and "not loaded" not in line.lower(): + key, value = line.split(":", 1) + info[key.strip()] = value.strip() + return info + + if preloaded_output is not None: + driver_info = {} + for host, combined_str in preloaded_output.items(): + if not combined_str or combined_str.startswith("ERROR") or combined_str.startswith("ABORT"): + driver_info[host] = {"error": combined_str} + continue + driver_info[host] = {} + # Split on ===DRIVER:name=== markers + current_driver = None + lines = [] + for line in combined_str.split("\n"): + if line.startswith("===DRIVER:") and line.endswith("==="): + if current_driver and lines: + info = _parse_driver_block("\n".join(lines), current_driver) + if info: + driver_info[host][current_driver] = info + current_driver = line[10:-3] + lines = [] + elif current_driver is not None: + lines.append(line) + if current_driver and lines: + info = _parse_driver_block("\n".join(lines), current_driver) + if info: + driver_info[host][current_driver] = info + return driver_info + + # Fallback: original sequential parallel-ssh approach commands = [ "modinfo mlx5_core 2>/dev/null | grep -E '^version|^firmware' | head -3", "modinfo bnxt_en 2>/dev/null | grep -E '^version|^firmware' | head -3", "modinfo amd-ainic 2>/dev/null | grep -E '^version|^firmware' | head -3 || echo 'Not loaded'", ] - driver_info = {} - for host in ssh_manager.reachable_hosts: driver_info[host] = {} - - # Check Mellanox (NVIDIA CX7) - output = await ssh_manager.exec_async(commands[0], timeout=60) - if host in output and output[host] and "modinfo" not in output[host]: - mlx_info = {} - for line in output[host].split("\n"): - if ":" in line: - key, value = line.split(":", 1) - mlx_info[key.strip()] = value.strip() - if mlx_info: - driver_info[host]["mlx5_core"] = mlx_info - - # Check Broadcom (Thor2) - output = await ssh_manager.exec_async(commands[1], timeout=60) - if host in output and output[host] and "modinfo" not in output[host]: - bnxt_info = {} - for line in output[host].split("\n"): - if ":" in line: - key, value = line.split(":", 1) - bnxt_info[key.strip()] = value.strip() - if bnxt_info: - driver_info[host]["bnxt_en"] = bnxt_info - - # Check AMD AINIC - output = await ssh_manager.exec_async(commands[2], timeout=60) - if host in output and output[host] and "Not loaded" not in output[host]: - amd_info = {} - for line in output[host].split("\n"): - if ":" in line: - key, value = line.split(":", 1) - amd_info[key.strip()] = value.strip() - if amd_info: - driver_info[host]["amd-ainic"] = amd_info - + for driver_name, cmd in zip(["mlx5_core", "bnxt_en", "amd-ainic"], commands): + output = await ssh_manager.exec_async(cmd, timeout=60) + if ( + host in output + and output[host] + and "Not loaded" not in output[host] + and "modinfo" not in output[host] + ): + info = {} + for line in output[host].split("\n"): + if ":" in line: + key, value = line.split(":", 1) + info[key.strip()] = value.strip() + if info: + driver_info[host][driver_name] = info return driver_info - async def collect_rdma_statistics_detailed(self, ssh_manager) -> Dict[str, Any]: + async def collect_rdma_statistics_detailed(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect detailed RDMA statistics from 'rdma statistic show --json'. Returns comprehensive RDMA counter statistics. """ logger.info("Collecting detailed RDMA statistics") - output = await ssh_manager.exec_async( - "bash -c 'rdma statistic show --json 2>/dev/null || echo \"[]\"'", timeout=60 + output = ( + preloaded_output + if preloaded_output is not None + else await ssh_manager.exec_async( + "bash -c 'rdma statistic show --json 2>/dev/null || echo \"[]\"'", timeout=60 + ) ) rdma_stats = {} @@ -165,7 +237,7 @@ async def collect_rdma_statistics_detailed(self, ssh_manager) -> Dict[str, Any]: return rdma_stats - async def collect_ethtool_statistics_detailed(self, ssh_manager) -> Dict[str, Any]: + async def collect_ethtool_statistics_detailed(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect detailed ethtool statistics for all interfaces. @@ -174,46 +246,59 @@ async def collect_ethtool_statistics_detailed(self, ssh_manager) -> Dict[str, An """ logger.info("Collecting detailed ethtool statistics") - # Get list of interfaces first + if preloaded_output is not None: + eth_stats = {} + for host, combined_str in preloaded_output.items(): + if not combined_str or combined_str.startswith("ERROR") or combined_str.startswith("ABORT"): + eth_stats[host] = {"error": combined_str} + continue + eth_stats[host] = {} + for iface, block in self._parse_iface_sections(combined_str).items(): + stats = {} + for line in block.split("\n"): + match = re.search(r"^\s+([\w_]+):\s+(\d+)", line) + if match: + stats[match.group(1)] = int(match.group(2)) + if stats: + eth_stats[host][iface] = stats + return eth_stats + + # Fallback: original sequential parallel-ssh approach ip_output = await ssh_manager.exec_async( "bash -c \"ip -o link show | awk -F': ' '{print \\$2}' | grep -v lo\"", timeout=60 ) - eth_stats = {} - for host, ifaces_str in ip_output.items(): if ifaces_str.startswith("ERROR") or ifaces_str.startswith("ABORT"): eth_stats[host] = {"error": ifaces_str} continue - eth_stats[host] = {} interfaces = [i.strip() for i in ifaces_str.split("\n") if i.strip() and "@" not in i] - - for iface in interfaces[:10]: # Limit to first 10 + for iface in interfaces[:10]: cmd = f"sudo ethtool -S {iface} 2>/dev/null" output = await ssh_manager.exec_async(cmd, timeout=60) - if host in output and output[host] and "NOT_AVAILABLE" not in output[host]: stats = {} for line in output[host].split("\n"): - # Parse " stat_name: value" match = re.search(r"^\s+([\w_]+):\s+(\d+)", line) if match: stats[match.group(1)] = int(match.group(2)) - if stats: eth_stats[host][iface] = stats - return eth_stats - async def collect_pci_device_info(self, ssh_manager) -> Dict[str, Any]: + async def collect_pci_device_info(self, ssh_manager=None, preloaded_output=None) -> Dict[str, Any]: """ Collect PCI device information for NICs. Command: lspci -nn | grep -i network """ logger.info("Collecting PCI device info for NICs") - output = await ssh_manager.exec_async("bash -c \"lspci -nn | grep -i 'network\\|ethernet'\"", timeout=60) + output = ( + preloaded_output + if preloaded_output is not None + else await ssh_manager.exec_async("bash -c \"lspci -nn | grep -i 'network\\|ethernet'\"", timeout=60) + ) pci_info = {} for host, out_str in output.items(): @@ -240,15 +325,37 @@ async def collect_all_software_info(self, ssh_manager) -> Dict[str, Any]: Returns consolidated NIC software info. """ + import asyncio + from app.core.go_collector import collect_parallel + logger.info("Collecting all NIC software information") - # IMPORTANT: Run commands SEQUENTIALLY to avoid parallel-ssh thread safety issues - # asyncio.gather() was causing "munmap_chunk(): invalid pointer" crashes - nic_firmware = await self.collect_nic_firmware_version(ssh_manager) - nic_drivers = await self.collect_nic_driver_version(ssh_manager) - rdma_statistics = await self.collect_rdma_statistics_detailed(ssh_manager) - ethtool_statistics = await self.collect_ethtool_statistics_detailed(ssh_manager) - pci_devices = await self.collect_pci_device_info(ssh_manager) + commands = { + "firmware": self._CMD_FIRMWARE, + "drivers": self._CMD_DRIVERS, + "rdma": self._CMD_RDMA, + "ethtool": self._CMD_ETHTOOL, + "pci": self._CMD_PCI, + } + + go_results = await asyncio.to_thread(collect_parallel, ssh_manager, commands, 60) + + if go_results is not None: + logger.info("NIC software collected via Go binary") + nic_firmware = await self.collect_nic_firmware_version(preloaded_output=go_results.get("firmware", {})) + nic_drivers = await self.collect_nic_driver_version(preloaded_output=go_results.get("drivers", {})) + rdma_statistics = await self.collect_rdma_statistics_detailed(preloaded_output=go_results.get("rdma", {})) + ethtool_statistics = await self.collect_ethtool_statistics_detailed( + preloaded_output=go_results.get("ethtool", {}) + ) + pci_devices = await self.collect_pci_device_info(preloaded_output=go_results.get("pci", {})) + else: + logger.info("Falling back to sequential parallel-ssh for NIC software") + nic_firmware = await self.collect_nic_firmware_version(ssh_manager) + nic_drivers = await self.collect_nic_driver_version(ssh_manager) + rdma_statistics = await self.collect_rdma_statistics_detailed(ssh_manager) + ethtool_statistics = await self.collect_ethtool_statistics_detailed(ssh_manager) + pci_devices = await self.collect_pci_device_info(ssh_manager) software_info = { "timestamp": datetime.utcnow().isoformat() + "Z", diff --git a/cvs/monitors/cluster-mon/backend/app/core/cvs_parallel_ssh_reliable.py b/cvs/monitors/cluster-mon/backend/app/core/cvs_parallel_ssh_reliable.py index f440152c6..286aa234a 100644 --- a/cvs/monitors/cluster-mon/backend/app/core/cvs_parallel_ssh_reliable.py +++ b/cvs/monitors/cluster-mon/backend/app/core/cvs_parallel_ssh_reliable.py @@ -9,6 +9,7 @@ from pssh.clients import ParallelSSHClient from pssh.exceptions import Timeout, ConnectionError +import os import time import logging import threading @@ -21,6 +22,13 @@ # TCP probe for fast reachability detection from app.core.host_probe import discover_reachable_hosts +# Read once at module load time. +# true → log first 5 lines of each host's output (set via --debug in full-rebuild.sh) +# false → log nothing (default, much faster for large clusters) +_DEBUG_SSH_OUTPUT = os.environ.get("DEBUG_SSH_OUTPUT", "false").lower() == "true" +_DEBUG_MAX_LINES = 5 # lines to show per host in debug mode + + # Module-level logger logger = logging.getLogger(__name__) @@ -305,13 +313,23 @@ def _process_output(self, output, cmd=None, cmd_list=None, print_console=True): else: logger.debug(cmd) try: + log_line_count = 0 + do_log = print_console and _DEBUG_SSH_OUTPUT for line in item.stdout or []: - if print_console: + if do_log and log_line_count < _DEBUG_MAX_LINES: logger.info(line) + log_line_count += 1 + elif do_log and log_line_count == _DEBUG_MAX_LINES: + logger.info(f"... (truncated, showing first {_DEBUG_MAX_LINES} lines only)") + log_line_count += 1 cmd_out_str += line.replace('\t', ' ') + '\n' for line in item.stderr or []: - if print_console: + if do_log and log_line_count < _DEBUG_MAX_LINES: logger.info(line) + log_line_count += 1 + elif do_log and log_line_count == _DEBUG_MAX_LINES: + logger.info(f"... (truncated, showing first {_DEBUG_MAX_LINES} lines only)") + log_line_count += 1 cmd_out_str += line.replace('\t', ' ') + '\n' except Timeout as e: if not self.stop_on_errors: diff --git a/cvs/monitors/cluster-mon/backend/app/core/go_collector.py b/cvs/monitors/cluster-mon/backend/app/core/go_collector.py new file mode 100644 index 000000000..e08741c96 --- /dev/null +++ b/cvs/monitors/cluster-mon/backend/app/core/go_collector.py @@ -0,0 +1,100 @@ +""" +Shared utility for parallel SSH command collection via the Go gpu-collector binary. +All nodes run simultaneously; all commands run in parallel per node. +Falls back gracefully when the binary is unavailable or when using JumpHostPssh. +""" + +import json +import logging +import os +import subprocess +from typing import Dict, Optional + +logger = logging.getLogger(__name__) + +_GO_BINARY = os.environ.get("GPU_COLLECTOR_BIN", "/usr/local/bin/gpu-collector") + + +def collect_parallel( + ssh_manager, + commands: Dict[str, str], + timeout: int = 60, +) -> Optional[Dict[str, Dict[str, str]]]: + """ + Run multiple shell commands across all nodes using the Go binary. + + Args: + ssh_manager: Pssh instance with .reachable_hosts, .user, .pkey, .password + commands: {name: shell_command} — all commands run in parallel per node + timeout: per-command timeout in seconds (global timeout = timeout + 20s) + + Returns: + {name: {host: output_str}} — same format as ssh_manager.exec() per command, + or None if the binary is unavailable or errored (caller should fall back). + """ + from app.core.jump_host_pssh import JumpHostPssh + + if not os.path.isfile(_GO_BINARY): + logger.debug(f"Go binary not found at {_GO_BINARY}, using parallel-ssh fallback") + return None + + if isinstance(ssh_manager, JumpHostPssh): + logger.debug("Jump host mode detected, using parallel-ssh fallback") + return None + + if not ssh_manager.reachable_hosts: + # No hosts — return empty result dicts, no fallback needed + return {name: {} for name in commands} + + inp = { + "hosts": ssh_manager.reachable_hosts, + "ssh_user": ssh_manager.user, + "ssh_key_path": ssh_manager.pkey or "", + "ssh_password": ssh_manager.password or "", + "ssh_port": 22, + "commands": commands, + "per_host_timeout_s": timeout, + "global_timeout_s": timeout + 20, + } + + try: + proc = subprocess.run( + [_GO_BINARY], + input=json.dumps(inp).encode(), + capture_output=True, + timeout=timeout + 30, + ) + except subprocess.TimeoutExpired: + logger.error(f"Go binary timed out after {timeout + 30}s") + return None + except Exception as e: + logger.error(f"Go binary execution failed: {e}") + return None + + if proc.returncode != 0: + logger.error(f"Go binary exited {proc.returncode}: {proc.stderr.decode()[:300]}") + return None + + try: + go_out = json.loads(proc.stdout) + except json.JSONDecodeError as e: + logger.error(f"Go binary output parse error: {e}") + return None + + raw_results = go_out.get("results", {}) + duration_ms = go_out.get("collection_duration_ms", 0) + unreachable = go_out.get("unreachable", []) + + logger.info(f"Go binary: {len(raw_results)} nodes in {duration_ms}ms, {len(unreachable)} unreachable") + + # Convert {host: {cmd_name: {status, raw}}} → {cmd_name: {host: output_str}} + by_cmd: Dict[str, Dict[str, str]] = {name: {} for name in commands} + for host, host_results in raw_results.items(): + for name, result in host_results.items(): + if name in by_cmd: + if result.get("status") == "ok": + by_cmd[name][host] = result.get("raw", "") + else: + by_cmd[name][host] = f"ERROR: {result.get('error', 'unknown error')}" + + return by_cmd diff --git a/cvs/monitors/cluster-mon/backend/app/main.py b/cvs/monitors/cluster-mon/backend/app/main.py index 31f748ada..740e9d337 100644 --- a/cvs/monitors/cluster-mon/backend/app/main.py +++ b/cvs/monitors/cluster-mon/backend/app/main.py @@ -79,6 +79,9 @@ def __init__(self): self.nic_software_cache_time: float = 0 self.nic_advanced_cache_time: float = 0 self.software_cache_ttl: int = 180 # 3 minutes + # Logs cache (TTL: 180 seconds — logs change slowly and collection is expensive) + self.cached_logs: dict = {} + self.logs_cache_time: float = 0 # SECURITY: Passwords stored in memory only (never persisted to disk) self.ssh_password: str = None # Direct SSH password self.jump_host_password: str = None # Jump host password @@ -132,9 +135,11 @@ async def reload_configuration(): app_state.cached_gpu_software = {} app_state.cached_nic_software = {} app_state.cached_nic_advanced = {} + app_state.cached_logs = {} app_state.gpu_software_cache_time = 0 app_state.nic_software_cache_time = 0 app_state.nic_advanced_cache_time = 0 + app_state.logs_cache_time = 0 # 4. Reload configuration from files logger.info("Reloading configuration from cluster.yaml and nodes.txt...") diff --git a/cvs/monitors/cluster-mon/docker-compose.yml b/cvs/monitors/cluster-mon/docker-compose.yml index c4481d1c9..95d797c22 100644 --- a/cvs/monitors/cluster-mon/docker-compose.yml +++ b/cvs/monitors/cluster-mon/docker-compose.yml @@ -30,6 +30,10 @@ services: - POLLING__FAILURE_THRESHOLD=${POLLING__FAILURE_THRESHOLD:-5} # Debug mode - DEBUG=${DEBUG:-false} + # SSH output logging: set to true via --debug flag in full-rebuild.sh + # When false (default): no SSH command output in logs (faster, cleaner) + # When true: first 5 lines of each host's output logged (for debugging) + - DEBUG_SSH_OUTPUT=${DEBUG_SSH_OUTPUT:-false} # Cluster monitor home - CLUSTER_MONITOR_HOME=/app restart: unless-stopped diff --git a/cvs/monitors/cluster-mon/frontend/src/components/ui/DataTable.tsx b/cvs/monitors/cluster-mon/frontend/src/components/ui/DataTable.tsx index 12541c66c..782c21e53 100644 --- a/cvs/monitors/cluster-mon/frontend/src/components/ui/DataTable.tsx +++ b/cvs/monitors/cluster-mon/frontend/src/components/ui/DataTable.tsx @@ -38,8 +38,8 @@ export function CustomDataTable({ pageLength: defaultPageLength, lengthMenu: pageLengthOptions, order: [[0, 'asc']], // Default sort by first column - responsive: true, - autoWidth: false, + scrollX: true, // Enable horizontal scrolling + autoWidth: true, // Let DataTables size columns to content language: { search: 'Search:', lengthMenu: 'Show _MENU_ entries per page', @@ -69,8 +69,8 @@ export function CustomDataTable({ }, [data, columns, defaultPageLength, pageLengthOptions]) return ( -
- +
+
{columns.map((col, idx) => ( @@ -175,14 +175,30 @@ export function CustomDataTable({ font-size: 0.875rem; } + /* Horizontal scroll */ + .datatable-wrapper .dataTables_wrapper .dataTables_scroll { + overflow-x: auto; + } + + .datatable-wrapper .dataTables_wrapper .dataTables_scrollBody { + overflow-x: auto !important; + } + /* Text alignment classes */ - .datatable-wrapper table.dataTable td.dt-right { + .datatable-wrapper table.dataTable td.dt-right, + .datatable-wrapper table.dataTable th.dt-right { text-align: right; } - .datatable-wrapper table.dataTable td.dt-center { + .datatable-wrapper table.dataTable td.dt-center, + .datatable-wrapper table.dataTable th.dt-center { text-align: center; } + + .datatable-wrapper table.dataTable td.dt-left, + .datatable-wrapper table.dataTable th.dt-left { + text-align: left; + } `} ) diff --git a/cvs/monitors/cluster-mon/frontend/src/pages/GPUMetricsPage.tsx b/cvs/monitors/cluster-mon/frontend/src/pages/GPUMetricsPage.tsx index 53b945b43..02ac8c4ef 100644 --- a/cvs/monitors/cluster-mon/frontend/src/pages/GPUMetricsPage.tsx +++ b/cvs/monitors/cluster-mon/frontend/src/pages/GPUMetricsPage.tsx @@ -33,7 +33,20 @@ export function GPUMetricsPage() { const powerData = latestMetrics.gpu.power || {} Object.entries(utilData).forEach(([node, gpuData]: [string, any]) => { - if (typeof gpuData === 'object' && !gpuData.error) { + if (typeof gpuData === 'object' && gpuData.error) { + // Node had a collection error — show one row with dashes so it's visible in the table + data.push({ + node, + gpu: '-', + utilization: null, + memUsed: null, + memTotal: null, + memPercent: null, + temperature: null, + power: null, + error: gpuData.error, + }) + } else if (typeof gpuData === 'object' && !gpuData.error) { Object.entries(gpuData).forEach(([gpuId, metrics]: [string, any]) => { if (typeof metrics === 'object') { const memMetrics = memData[node]?.[gpuId] || {} @@ -55,6 +68,7 @@ export function GPUMetricsPage() { memPercent, temperature: temp, power: 0, + error: null, }) } }) @@ -90,16 +104,15 @@ export function GPUMetricsPage() { // Use pcie_link_status for actual link info, pcie for error counters Object.entries(pcieLinkData).forEach(([node, gpuData]: [string, any]) => { - if (typeof gpuData === 'object') { + if (typeof gpuData === 'object' && gpuData.error) { + data.push({ + node, gpu: '-', width: '-', speed: '-', bandwidth: '-', + replay_count: '-', l0_to_recovery_count: '-', nak_sent: '-', nak_received: '-', + error: gpuData.error, + }) + } else if (typeof gpuData === 'object') { Object.entries(gpuData).forEach(([gpuId, linkInfo]: [string, any]) => { - // Get error metrics from pcie data - const pcieErrors = pcieMetricData[node]?.gpu_data?.find((g: any) => - `card${g.gpu}` === gpuId - )?.pcie || {} - - // Use pcie data directly (has all fields) const pcieData = pcieMetricData[node]?.[gpuId] || linkInfo - data.push({ node, gpu: gpuId, @@ -110,6 +123,7 @@ export function GPUMetricsPage() { l0_to_recovery_count: pcieData.l0_to_recovery_count || 0, nak_sent: pcieData.nak_sent_count || 0, nak_received: pcieData.nak_received_count || 0, + error: null, }) }) } @@ -138,26 +152,18 @@ export function GPUMetricsPage() { }) Object.entries(latestMetrics.gpu.xgmi).forEach(([node, nodeData]: [string, any]) => { - // Handle both formats: direct array or wrapped in gpu_data + if (typeof nodeData === 'object' && nodeData.error) { + data.push({ node, gpu: '-', error_count: '-', status: '⚠ Error', error: nodeData.error }) + return + } const gpuArray = Array.isArray(nodeData) ? nodeData : (nodeData.gpu_data || []) - if (Array.isArray(gpuArray)) { gpuArray.forEach((gpu: any) => { const xgmi = gpu.xgmi_err || gpu.xgmi if (xgmi && xgmi !== 'N/A' && typeof xgmi === 'object') { - data.push({ - node, - gpu: `card${gpu.gpu || 0}`, - error_count: xgmi.error_count || 0, - status: xgmi.status || 'N/A', - }) + data.push({ node, gpu: `card${gpu.gpu || 0}`, error_count: xgmi.error_count || 0, status: xgmi.status || 'N/A', error: null }) } else { - data.push({ - node, - gpu: `card${gpu.gpu || 0}`, - error_count: 0, - status: xgmi === 'N/A' ? 'N/A' : 'No errors', - }) + data.push({ node, gpu: `card${gpu.gpu || 0}`, error_count: 0, status: xgmi === 'N/A' ? 'N/A' : 'No errors', error: null }) } }) } @@ -171,14 +177,22 @@ export function GPUMetricsPage() { if (!latestMetrics?.gpu?.ras_errors) return data Object.entries(latestMetrics.gpu.ras_errors).forEach(([node, nodeData]: [string, any]) => { - // Handle both formats: direct array or wrapped in gpu_data + if (typeof nodeData === 'object' && nodeData.error) { + data.push({ + node, gpu: '-', + total_correctable: '-', total_uncorrectable: '-', + cache_correctable: '-', cache_uncorrectable: '-', + umc_correctable: '-', umc_uncorrectable: '-', + gfx_correctable: '-', gfx_uncorrectable: '-', + error: nodeData.error, + }) + return + } const gpuArray = Array.isArray(nodeData) ? nodeData : (nodeData.gpu_data || []) - if (Array.isArray(gpuArray)) { gpuArray.forEach((gpu: any) => { const ecc = gpu.ecc || {} const ecc_blocks = gpu.ecc_blocks || {} - data.push({ node, gpu: `card${gpu.gpu || 0}`, @@ -190,6 +204,7 @@ export function GPUMetricsPage() { umc_uncorrectable: ecc_blocks.UMC?.uncorrectable_count || 0, gfx_correctable: ecc_blocks.GFX?.correctable_count || 0, gfx_uncorrectable: ecc_blocks.GFX?.uncorrectable_count || 0, + error: null, }) }) } @@ -211,7 +226,8 @@ export function GPUMetricsPage() { title: 'Utilization', data: 'utilization', className: 'dt-right', - render: (data: number) => { + render: (data: number | null) => { + if (data === null) return '-' const color = data > 90 ? 'text-red-600' : data > 70 ? 'text-yellow-600' : 'text-green-600' return `${formatPercentage(data)}` }, @@ -220,25 +236,26 @@ export function GPUMetricsPage() { title: 'Memory Used', data: 'memUsed', className: 'dt-right', - render: (data: number) => `${formatBytes(data)}`, + render: (data: number | null) => data === null ? '-' : `${formatBytes(data)}`, }, { title: 'Memory Total', data: 'memTotal', className: 'dt-right', - render: (data: number) => `${formatBytes(data)}`, + render: (data: number | null) => data === null ? '-' : `${formatBytes(data)}`, }, { title: 'Memory %', data: 'memPercent', className: 'dt-right', - render: (data: number) => formatPercentage(data), + render: (data: number | null) => data === null ? '-' : formatPercentage(data), }, { title: 'Temperature', data: 'temperature', className: 'dt-right', - render: (data: number) => { + render: (data: number | null) => { + if (data === null) return '-' const color = data > 85 ? 'text-red-600 font-medium' : data > 70 ? 'text-yellow-600' : 'text-gray-900' return `${formatTemperature(data)}` }, @@ -247,7 +264,15 @@ export function GPUMetricsPage() { title: 'Power', data: 'power', className: 'dt-right', - render: (data: number) => data > 0 ? formatPower(data) : '-', + render: (data: number | null) => (data === null || data === 0) ? '-' : formatPower(data), + }, + { + title: 'Status', + data: 'error', + className: 'dt-left', + render: (data: string | null) => data + ? `⚠ Error` + : 'OK', }, ] @@ -406,9 +431,12 @@ export function GPUMetricsPage() {
- {formatPercentage( - gpuData.reduce((sum, g) => sum + g.utilization, 0) / gpuData.length - )} + {(() => { + const valid = gpuData.filter(g => typeof g.utilization === 'number') + return valid.length > 0 + ? formatPercentage(valid.reduce((sum, g) => sum + g.utilization, 0) / valid.length) + : '-' + })()}

Avg Utilization

 @@ -416,9 +444,12 @@ export function GPUMetricsPage() {
- {formatTemperature( - gpuData.reduce((sum, g) => sum + g.temperature, 0) / gpuData.length - )} + {(() => { + const valid = gpuData.filter(g => typeof g.temperature === 'number' && g.temperature > 0) + return valid.length > 0 + ? formatTemperature(valid.reduce((sum, g) => sum + g.temperature, 0) / valid.length) + : '-' + })()}

Avg Temperature

 @@ -426,7 +457,7 @@ export function GPUMetricsPage() {
- {eccData.reduce((sum, e) => sum + e.total_uncorrectable, 0)} + {eccData.reduce((sum, e) => sum + (typeof e.total_uncorrectable === 'number' ? e.total_uncorrectable : 0), 0)}

Total ECC Errors

 diff --git a/cvs/monitors/cluster-mon/frontend/src/pages/LogsPage.tsx b/cvs/monitors/cluster-mon/frontend/src/pages/LogsPage.tsx index 08d5dad8e..5ac8ea37c 100644 --- a/cvs/monitors/cluster-mon/frontend/src/pages/LogsPage.tsx +++ b/cvs/monitors/cluster-mon/frontend/src/pages/LogsPage.tsx @@ -3,6 +3,16 @@ import { RefreshCw, FileText, AlertCircle } from 'lucide-react' import { Card, CardContent, CardHeader, CardTitle } from '@/components/ui/Card' import { CustomDataTable } from '@/components/ui/DataTable' +// Module-level cache — survives React Router navigation (component unmount/remount) +const LOGS_CACHE_TTL_MS = 180_000 // 3 minutes, matches backend TTL +let _logsCache: { + amdLogs: any[] + systemLogs: any[] + userspaceLogs: any[] + lastUpdate: string + fetchedAt: number +} | null = null + export function LogsPage() { const [amdLogs, setAmdLogs] = useState([]) const [systemLogs, setSystemLogs] = useState([]) @@ -17,56 +27,56 @@ export function LogsPage() { const [isSearching, setIsSearching] = useState(false) const [searchError, setSearchError] = useState(null) - const fetchLogs = async () => { + const applyCache = (cache: NonNullable) => { + setAmdLogs(cache.amdLogs) + setSystemLogs(cache.systemLogs) + setUserspaceLogs(cache.userspaceLogs) + setLastUpdate(cache.lastUpdate) + } + + const fetchLogs = async (forceRefresh = false) => { + // Return immediately from module-level cache if still fresh + if (!forceRefresh && _logsCache && (Date.now() - _logsCache.fetchedAt) < LOGS_CACHE_TTL_MS) { + applyCache(_logsCache) + return + } + setLoading(true) setError(null) try { const response = await fetch('/api/logs/dmesg') const data = await response.json() - // Transform AMD logs data into table format - const amdLogsArray: any[] = [] - if (data.amd_logs) { - Object.entries(data.amd_logs).forEach(([node, logOutput]: [string, any]) => { - if (typeof logOutput === 'string' && logOutput.trim()) { - amdLogsArray.push({ - node, - logs: logOutput, - }) - } - }) + const toRows = (section: any) => { + const rows: any[] = [] + if (section) { + Object.entries(section).forEach(([node, logOutput]: [string, any]) => { + if (typeof logOutput === 'string' && logOutput.trim()) { + rows.push({ node, logs: logOutput }) + } + }) + } + return rows } - // Transform system errors data into table format - const systemLogsArray: any[] = [] - if (data.dmesg_errors) { - Object.entries(data.dmesg_errors).forEach(([node, logOutput]: [string, any]) => { - if (typeof logOutput === 'string' && logOutput.trim()) { - systemLogsArray.push({ - node, - logs: logOutput, - }) - } - }) - } + const amdLogsArray = toRows(data.amd_logs) + const systemLogsArray = toRows(data.dmesg_errors) + const userspaceLogsArray = toRows(data.userspace_errors) + const update = new Date().toLocaleString() - // Transform userspace errors data into table format - const userspaceLogsArray: any[] = [] - if (data.userspace_errors) { - Object.entries(data.userspace_errors).forEach(([node, logOutput]: [string, any]) => { - if (typeof logOutput === 'string' && logOutput.trim()) { - userspaceLogsArray.push({ - node, - logs: logOutput, - }) - } - }) + // Store in module-level cache + _logsCache = { + amdLogs: amdLogsArray, + systemLogs: systemLogsArray, + userspaceLogs: userspaceLogsArray, + lastUpdate: update, + fetchedAt: Date.now(), } setAmdLogs(amdLogsArray) setSystemLogs(systemLogsArray) setUserspaceLogs(userspaceLogsArray) - setLastUpdate(new Date().toLocaleString()) + setLastUpdate(update) } catch (err: any) { console.error('Failed to fetch logs:', err) setError(err.message || 'Failed to fetch logs') @@ -129,12 +139,10 @@ export function LogsPage() { } useEffect(() => { - // Clear search results when page loads setGrepCommand('') setSearchResults([]) setSearchError(null) - - fetchLogs() + fetchLogs() // uses cache if fresh, fetches otherwise }, []) return ( @@ -146,7 +154,7 @@ export function LogsPage() {

Critical system errors from dmesg (emerg, alert, crit, err)