[Elastic] Add the log collection and diagnosis function

flagscale/runner/elastic/diagnostic.py

@@ -0,0 +1,97 @@
+import os
+
+from datetime import datetime
+
+from flagscale.runner.utils import logger
+
+error_types = {
+    # Success indicators
+    "completed": "Completed: The task finished successfully with no errors.",
+    # Memory errors
+    "out of memory": "OutOfMemoryError: The training process ran out of GPU memory.",
+    "outofmemoryerror": "OutOfMemoryError: The training process ran out of GPU memory.",
+    "cuda out of memory": "OutOfMemoryError: CUDA out of memory error occurred.",
+    # Connection and network errors
+    "rendezvousconne": "RendezvousConnectionError: Connection to rendezvous backend failed.",
+    "rendezvous": "RendezvousError: Rendezvous coordination failed between nodes.",
+    "connection refused": "ConnectionError: Network connection refused.",
+    "connection timeout": "ConnectionTimeout: Network connection timeout.",
+    # Import and code errors
+    "importerror": "ImportError: Failed to import required modules.",
+    "modulenotfounderror": "ModuleNotFoundError: Required Python module not found.",
+    "traceback": "CodeError: Python exception occurred during execution.",
+    "error": "GeneralError: An error occurred during training.",
+    # Process errors
+    "killed": "ProcessKilled: Training process was killed.",
+    "segmentation fault": "SegmentationFault: Process crashed due to memory access error.",
+    "core dumped": "CoreDump: Process crashed and dumped core.",
+    # CUDA errors
+    "cuda": "CUDAError: CUDA-related error occurred.",
+    "cudnn": "CUDNNError: CuDNN library error occurred.",
+    "gpu": "GPUError: GPU-related error occurred.",
+    # File and storage errors
+    "no such file": "FileNotFound: Required file or directory not found.",
+    "permission denied": "PermissionError: File permission denied.",
+    "disk space": "StorageError: Insufficient disk space.",
+    # Timeout errors
+    "timeout": "TimeoutError: Operation timed out.",
+    "hanging": "HangError: Process appears to be hanging.",
+}
+
+
+def generate_diagnostic_report(config, host, node_rank, log_file, return_content=False):
+    """
+    Generate a diagnostic report from a log file.
+    Args:
+        config (DictConfig): Configuration object.
+        host (str): Hostname or IP.
+        node_rank (int): Node rank.
+        log_file (str): Path to the log file.
+        return_content (bool): If True, return report as string instead of writing to file.
+    Returns:
+        str: Diagnostic report content if return_content=True, else None.
+    """
+    report_content = f"Diagnostic Report for {host} (node {node_rank})\n"
+    report_content += f"Generated at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
+    # report_content += f"Log file analyzed: {log_file}\n"
+    report_content += "Analysis:\n"
+
+    try:
+        if not os.path.exists(log_file) or os.path.getsize(log_file) == 0:
+            report_content += "- Log file is empty or does not exist, no analysis possible.\n"
+            return report_content if return_content else None
+        else:
+            with open(log_file, 'r') as f:
+                log_content = f.read()
+            if not log_content.strip():
+                report_content += "- Log file is empty, no analysis possible.\n"
+            else:
+                matched_errors = []
+                for key, desc in error_types.items():
+                    if key in log_content.lower():
+                        matched_errors.append(desc)
+
+                if matched_errors:
+                    for err in matched_errors:
+                        report_content += f"- {err}\n"
+                else:
+                    report_content += "- No errors or unknown error detected in logs.\n"
+    except Exception as e:
+        logger.error(f"Failed to read log file {log_file} for {host} (node {node_rank}): {e}")
+        report_content += f"- Error reading log file: {e}\n"
+
+    if return_content:
+        return report_content
+    else:
+        try:
+            diagnostic_file = log_file.replace("temp", "diagnostic").replace(".log", ".txt")
+            os.makedirs(os.path.dirname(diagnostic_file), exist_ok=True)
+            with open(diagnostic_file, 'w') as f:
+                f.write(report_content)
+            logger.debug(
+                f"Generated diagnostic report for {host} (node {node_rank}) at {diagnostic_file}"
+            )
+            return diagnostic_file
+        except Exception as e:
+            logger.error(f"Failed to write diagnostic report for {host} (node {node_rank}): {e}")
+            return None

flagscale/runner/elastic/log_collector.py

@@ -0,0 +1,66 @@
+import os
+import shlex
+
+from datetime import datetime
+
+from flagscale.runner.utils import logger, run_local, run_scp
+
+_log_offsets = {}
+
+
+def collect_logs(config, host, node_rank, destination_dir, dryrun=False):
+    """
+    Collect logs incrementally from a specified host and node rank, saving to destination_dir.
+    Args:
+        config (DictConfig): Configuration object containing experiment and logging details.
+        host (str): Hostname or IP of the node.
+        node_rank (int): Rank of the node.
+        destination_dir (str): Directory to store collected logs.
+        dryrun (bool): If True, simulate the collection without executing commands.
+    Returns:
+        str: Path to the collected log file.
+    """
+    logging_config = config.train.system.logging
+    no_shared_fs = config.experiment.runner.get("no_shared_fs", False)
+    log_dir = logging_config.log_dir
+    src_log_file = os.path.join(
+        log_dir, f"host{'_' + str(node_rank) + '_' + host if not no_shared_fs else ''}.output"
+    )
+    dest_log_file = os.path.join(
+        destination_dir,
+        f"host_{node_rank}_{host}_temp_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log",
+    )
+
+    os.makedirs(destination_dir, exist_ok=True)
+
+    log_key = f"{host}_{node_rank}"
+    offset = _log_offsets.get(log_key, 0)
+
+    try:
+        if host != "localhost":
+            ssh_port = config.experiment.runner.get("ssh_port", 22)
+            command = f"tail -c +{offset + 1} {src_log_file} > {dest_log_file}"
+            run_scp(host, src_log_file, dest_log_file, ssh_port, dryrun, incremental=True)
+            logger.debug(
+                f"Collected incremental log from {host} (node {node_rank}) to {dest_log_file}"
+            )
+        else:
+            command = f"tail -c +{offset + 1} {src_log_file} > {dest_log_file}"
+            run_local(command, dryrun)
+            logger.debug(f"Collected incremental local log to {dest_log_file}")
+
+        if os.path.exists(dest_log_file) and os.path.getsize(dest_log_file) > 0:
+            new_offset = os.path.getsize(src_log_file)
+            _log_offsets[log_key] = new_offset
+            return dest_log_file
+        else:
+            logger.warning(f"Log file {src_log_file} not found or empty")
+            if os.path.exists(dest_log_file):
+                os.remove(dest_log_file)
+            return None
+
+    except Exception as e:
+        logger.error(f"Failed to collect logs from {host} (node {node_rank}): {e}")
+        if os.path.exists(dest_log_file):
+            os.remove(dest_log_file)
+        return None

flagscale/runner/elastic/monitor_service.py

@@ -0,0 +1,230 @@
+import os
+import signal
+import sys
+import threading
+import time
+
+from datetime import datetime
+from typing import Any, Dict, Optional
+
+from flagscale.runner.elastic.diagnostic import generate_diagnostic_report
+from flagscale.runner.elastic.log_collector import collect_logs
+from flagscale.runner.runner_base import JobStatus
+from flagscale.runner.utils import logger
+
+
+class MonitorService:
+    """
+    An independent monitoring service for background monitoring of training task status, log collection, and diagnostic report generation.
+    """
+
+    def __init__(self, config, runner_instance, interval=10):
+        """
+        Initializing service
+
+        Args:
+            config: Configuration object
+            runner_instance: runner instance
+            interval: interval time
+        """
+        self.config = config
+        self.runner = runner_instance
+        self.interval = interval
+        self.is_running = False
+        self.monitor_thread = None
+        self.log_collection_enabled = True
+        self.diagnostic_enabled = True
+
+        self.monitor_log_dir = os.path.join(config.train.system.logging.log_dir, "monitor")
+        os.makedirs(self.monitor_log_dir, exist_ok=True)
+
+        signal.signal(signal.SIGINT, self._signal_handler)
+        signal.signal(signal.SIGTERM, self._signal_handler)
+
+    def _signal_handler(self, signum, frame):
+        logger.info(f"Received signal {signum}, stopping monitor service...")
+        self.stop()
+        sys.exit(0)
+
+    def start_monitoring(self, enable_log_collection=True, enable_diagnostic=True):
+        """
+        Start monitoring service (non-blocking)
+
+        Args:
+            enable_log_collection: Whether to enable log collection
+            enable_diagnostic: Whether to enable diagnostic report generation
+        """
+        if self.is_running:
+            logger.warning("Monitor service is already running")
+            return
+
+        self.log_collection_enabled = enable_log_collection
+        self.diagnostic_enabled = enable_diagnostic
+        self.is_running = True
+
+        self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
+        self.monitor_thread.start()
+
+        logger.info(f"Monitor service started with interval={self.interval}s")
+        logger.info(f"Log collection enabled: {enable_log_collection}")
+        logger.info(f"Diagnostic enabled: {enable_diagnostic}")
+        logger.info(f"Monitor logs will be saved to: {self.monitor_log_dir}")
+
+    def stop(self):
+        if not self.is_running:
+            return
+
+        self.is_running = False
+        if self.monitor_thread and self.monitor_thread.is_alive():
+            self.monitor_thread.join(timeout=5)
+
+        logger.info("Monitor service stopped")
+
+    def _monitor_loop(self):
+        logger.info("Starting monitoring loop...")
+
+        time.sleep(self.interval)
+
+        try:
+            while self.is_running:
+                start_time = time.time()
+
+                try:
+                    job_status = self._get_job_status()
+                    logger.info(f"Job Status: {job_status.name}")
+
+                    self._log_status(job_status)
+
+                    if job_status == JobStatus.COMPLETED_OR_IDLE:
+                        logger.info("Job completed, stopping monitoring")
+                        break
+
+                    if self.log_collection_enabled:
+                        self._collect_logs()
+
+                    if self.diagnostic_enabled:
+                        self._generate_diagnostics()
+
+                except Exception as e:
+                    logger.error(f"Error in monitoring loop: {e}")
+
+                elapsed = time.time() - start_time
+                sleep_time = max(0, self.interval - elapsed)
+
+                if self.is_running:
+                    time.sleep(sleep_time)
+
+        except Exception as e:
+            logger.error(f"Monitor loop crashed: {e}")
+        finally:
+            logger.info("Monitor loop ended")
+            self.is_running = False
+
+    def _get_job_status(self) -> JobStatus:
+        return self.runner._query_status()
+
+    def _log_status(self, status: JobStatus):
+        status_log_file = os.path.join(self.monitor_log_dir, "status.log")
+
+        try:
+            with open(status_log_file, "a", encoding="utf-8") as f:
+                timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+                f.write(f"[{timestamp}] Status: {status.name}\n")
+        except Exception as e:
+            logger.error(f"Failed to write status log: {e}")
+
+    def _collect_logs(self):
+        if not hasattr(self.runner, 'resources') or self.runner.resources is None:
+            self._collect_logs_for_host("localhost", 0)
+        else:
+            for node_rank, (host, _) in enumerate(self.runner.resources.items()):
+                self._collect_logs_for_host(host, node_rank)
+
+    def _collect_logs_for_host(self, host: str, node_rank: int):
+        try:
+            log_file = collect_logs(
+                self.config, host, node_rank, self.monitor_log_dir, dryrun=False
+            )
+
+            if log_file:
+                logger.debug(f"Collected logs for {host} (node {node_rank}): {log_file}")
+        except Exception as e:
+            logger.error(f"Failed to collect logs for {host} (node {node_rank}): {e}")
+
+    def _generate_diagnostics(self):
+        if not hasattr(self.runner, 'resources') or self.runner.resources is None:
+            self._generate_diagnostic_for_host("localhost", 0)
+        else:
+            for node_rank, (host, _) in enumerate(self.runner.resources.items()):
+                self._generate_diagnostic_for_host(host, node_rank)
+
+    def _generate_diagnostic_for_host(self, host: str, node_rank: int):
+        try:
+            log_files = [
+                f
+                for f in os.listdir(self.monitor_log_dir)
+                if f.startswith(f"host_{node_rank}_{host}_temp_") and f.endswith(".log")
+            ]
+
+            if log_files:
+                latest_log = max(
+                    log_files, key=lambda f: os.path.getmtime(os.path.join(self.monitor_log_dir, f))
+                )
+                log_file_path = os.path.join(self.monitor_log_dir, latest_log)
+
+                diagnostic_file = generate_diagnostic_report(
+                    self.config, host, node_rank, log_file_path, return_content=False
+                )
+
+                if diagnostic_file:
+                    logger.debug(
+                        f"Generated diagnostic for {host} (node {node_rank}): {diagnostic_file}"
+                    )
+        except Exception as e:
+            logger.error(f"Failed to generate diagnostic for {host} (node {node_rank}): {e}")
+
+    def get_status_summary(self) -> Dict[str, Any]:
+        return {
+            "is_running": self.is_running,
+            "interval": self.interval,
+            "log_collection_enabled": self.log_collection_enabled,
+            "diagnostic_enabled": self.diagnostic_enabled,
+            "monitor_log_dir": self.monitor_log_dir,
+            "thread_alive": self.monitor_thread.is_alive() if self.monitor_thread else False,
+        }
+
+
+def main():
+    """
+    python monitor_service.py [config_file] [interval]
+    """
+    import argparse
+
+    from omegaconf import OmegaConf
+
+    parser = argparse.ArgumentParser(description="Run FlagScale monitor service")
+    parser.add_argument("--config", type=str, help="Config file path")
+    parser.add_argument("--interval", type=int, default=10, help="Monitor interval in seconds")
+    parser.add_argument("--no-log-collection", action="store_true", help="Disable log collection")
+    parser.add_argument("--no-diagnostic", action="store_true", help="Disable diagnostic reports")
+
+    args = parser.parse_args()
+
+    if not args.config:
+        logger.error("Config file is required")
+        sys.exit(1)
+
+    try:
+        config = OmegaConf.load(args.config)
+
+        # Here needs to create a runner instance according to the actual situation
+        logger.info("Monitor service is designed to be integrated with runner_train.py")
+        logger.info("For standalone usage, additional runner initialization is needed")
+
+    except Exception as e:
+        logger.error(f"Failed to start monitor service: {e}")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()