Migrating to Golang based collectors instead of parallel-ssh and also…

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
 

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,25 +119,49 @@ 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(
             1 for v in logs_data.get("userspace_errors", {}).values() if isinstance(v, str) and v.strip()
         )
 
         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 | <validated_grep_command> | 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,
         }
 

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}")

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 = {}