[Elastic] Add the log collection and diagnosis function

flagscale/runner/auto_tuner/tuner.py

@@ -239,7 +239,8 @@ def tune(self):
             best_task = self.generator.gen_best_task(best_strategy, self.orig_config)
             best_task.action = "run"
             runner = SSHTrainRunner(best_task)
-            runner.run(monitor=True, interval=60)
+            enable_monitoring = best_task.experiment.runner.get("enable_monitoring", False)
+            runner.run(monitor=True, interval=60, enable_monitoring=enable_monitoring)
 
     def need_stop(self):
         """Judge whether need to stop tuning."""
@@ -296,7 +297,8 @@ def run(self, task=None):
         if task is None:
             task = self.cur_task
         self.runner = SSHTrainRunner(task)
-        self.runner.run()
+        enable_monitoring = task.experiment.runner.get("enable_monitoring", False)
+        self.runner.run(enable_monitoring=enable_monitoring)
         # set start time
         self.task_start_time = time.time()
 

flagscale/runner/elastic/diagnostic.py

@@ -0,0 +1,175 @@
+import os
+import time
+
+from flagscale.runner.utils import logger
+
+# Record the number of rows last diagnosed and analyzed by each host
+_diagnostic_offsets = {}
+
+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
+    "rendezvousconnectionerror": "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 (most recent call last)": "CodeError: Python exception occurred during execution.",
+    "fatal error": "FatalError: Fatal error occurred during training.",
+    "exception": "Exception: An exception occurred during training.",
+    # Process errors
+    "process killed": "ProcessKilled: Training process was killed.",
+    "killed by signal": "ProcessKilled: Process was killed by signal.",
+    "terminated by signal": "ProcessKilled: Process was terminated by signal.",
+    "keyboardinterrupt": "ProcessKilled: MANUAL KILL - Keyboard interrupt detected",
+    "sigint": "ProcessKilled: MANUAL KILL - SIGINT signal received",
+    "sigterm": "ProcessKilled: MANUAL KILL - SIGTERM signal received",
+    "segmentation fault": "SegmentationFault: Process crashed due to memory access error.",
+    "core dumped": "CoreDump: Process crashed and dumped core.",
+    # CUDA errors
+    "cuda error": "CUDAError: CUDA-related error occurred.",
+    "cudnn error": "CUDNNError: CuDNN library error occurred.",
+    "gpu error": "GPUError: GPU-related error occurred.",
+    # File and storage errors
+    "no such file or directory": "FileNotFound: Required file or directory not found.",
+    "permission denied": "PermissionError: File permission denied.",
+    "no space left on device": "StorageError: Insufficient disk space.",
+    # Timeout errors
+    "operation timed out": "TimeoutError: Operation timed out.",
+    "connection timeout": "TimeoutError: Connection timed out.",
+    "hanging": "HangError: Process appears to be hanging.",
+}
+
+
+def find_error_lines(lines, error_key, start_line=0):
+    """
+    Search for the error keyword in the log line and return a list of line numbers
+    Args:
+        lines (list): All lines of the log file
+        error_key (str): errors keyword
+        start_line (int): The line number where the search began
+    Returns:
+        list: A list of errors line numbers
+    """
+    matches = []
+    for i in range(start_line, len(lines)):
+        if error_key.lower() in lines[i].lower():
+            matches.append(i + 1)  # Line numbers start from 1
+    return matches
+
+
+def format_line_range(line_numbers):
+    """
+    Format the line number range display
+    Args:
+        line_numbers (list): List of line numbers
+    Returns:
+        str: Formatted line number range, such as "111-112" or "111"
+    """
+    if not line_numbers:
+        return "unknown"
+    elif len(line_numbers) == 1:
+        return str(line_numbers[0])
+    else:
+        return f"{min(line_numbers)}-{max(line_numbers)}"
+
+
+def generate_diagnostic_report(config, host, node_rank, log_file, return_content=False):
+    """
+    Generate an incremental 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 diagnostic file path.
+    """
+    global _diagnostic_offsets
+
+    # Generate the path of the diagnostic file
+    log_dir = os.path.dirname(log_file)
+    diagnostic_file = os.path.join(log_dir, f"host_{node_rank}_{host}_diagnostic.txt")
+    host_key = f"{host}_{node_rank}"
+
+    try:
+        if not os.path.exists(log_file) or os.path.getsize(log_file) == 0:
+            logger.debug(f"Log file {log_file} is empty or does not exist")
+            return diagnostic_file if not return_content else ""
+        # Read all lines of the log file
+        with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
+            all_lines = f.readlines()
+
+        # Get the number of rows analyzed last time
+        last_analyzed_line = _diagnostic_offsets.get(host_key, 0)
+
+        # If it is the first analysis, create the header of the diagnostic file
+        if last_analyzed_line == 0 and not os.path.exists(diagnostic_file):
+            os.makedirs(os.path.dirname(diagnostic_file), exist_ok=True)
+            header_content = f"Diagnostic Report for {host} (node {node_rank})\n"
+            header_content += (
+                f"Generated at {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}\n"
+            )
+            header_content += "Analysis:\n"
+
+            with open(diagnostic_file, 'w', encoding='utf-8') as f:
+                f.write(header_content)
+
+        # Only analyze the newly added rows
+        new_lines = all_lines[last_analyzed_line:]
+        if not new_lines:
+            logger.debug(f"No new lines to analyze in {log_file}")
+            return diagnostic_file if not return_content else ""
+
+        # Analyze the errors in the new lines
+        current_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())
+        new_errors = []
+        error_lines_by_type = {key: [] for key in error_types}
+
+        # Iterate through new lines once and check against all error types
+        for i, line in enumerate(all_lines[last_analyzed_line:], start=last_analyzed_line):
+            for key, desc in error_types.items():
+                if find_error_lines([line], key, 0):  # Check single line
+                    error_lines_by_type[key].append(i)
+
+        # Process error lines for each type
+        for key, desc in error_types.items():
+            error_lines = error_lines_by_type[key]
+            if error_lines:
+                line_range = format_line_range(error_lines)
+                log_filename = os.path.basename(log_file)
+                error_entry = f"[{current_time}] {desc} Check {log_filename} line:{line_range}"
+                new_errors.append(error_entry)
+
+        # Update Analysis Location
+        _diagnostic_offsets[host_key] = len(all_lines)
+
+        # If there are new errors, append them to the diagnostic file
+        if new_errors:
+            os.makedirs(os.path.dirname(diagnostic_file), exist_ok=True)
+            with open(diagnostic_file, 'a', encoding='utf-8') as f:
+                for error in new_errors:
+                    f.write(f"{error}\n")
+
+            logger.debug(
+                f"Added {len(new_errors)} new errors to diagnostic report for {host} (node {node_rank})"
+            )
+
+        if return_content:
+            return '\n'.join(new_errors) if new_errors else ""
+        else:
+            return diagnostic_file
+    except Exception as e:
+        logger.error(f"Failed to generate diagnostic report for {host} (node {node_rank}): {e}")
+        if return_content:
+            return f"Error analyzing log file: {e}"
+        else:
+            return None

flagscale/runner/elastic/log_collector.py

@@ -0,0 +1,175 @@
+import glob
+import os
+import shlex
+import subprocess
+
+from datetime import datetime
+
+from flagscale.runner.utils import logger, run_local_command
+
+_log_offsets = {}
+
+
+def get_remote_file_size(host, filepath):
+    """
+    Retrieve the size of a file on a remote host (in bytes).
+
+    Parameters:
+        host (str): The address of the remote host (e.g., 'user@hostname').
+        filepath (str): The path to the file on the remote host.
+
+    Returns:
+        int: The size of the file in bytes if successful; returns -1 if an error occurs.
+
+    Exception Handling:
+        subprocess.CalledProcessError: Caught when the SSH command fails.
+    """
+    try:
+        result = subprocess.run(
+            ["ssh", host, f"stat -c%s {filepath}"],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            check=True,
+        )
+        return int(result.stdout.strip())
+    except subprocess.CalledProcessError:
+        return -1
+
+
+def get_file_size(host, filepath):
+    """
+    Retrieve the size of a file, either locally or on a remote host (in bytes).
+
+    Parameters:
+        host (str): The address of the host (e.g., 'localhost' or 'user@hostname').
+        filepath (str): The path to the file, either local or on the remote host.
+
+    Returns:
+        int: The size of the file in bytes if successful; returns -1 if the file does not exist
+             or an error occurs during remote access.
+
+    Notes:
+        - For local files ('localhost'), uses os.path to check existence and get size.
+    """
+    if host == "localhost":
+        if os.path.exists(filepath):
+            return os.path.getsize(filepath)
+        return -1
+    else:
+        return get_remote_file_size(host, filepath)
+
+
+def find_actual_log_file(log_dir, node_rank, host, no_shared_fs=False):
+    """
+    Smart file discovery for log files that handles hostname/IP mismatches.
+
+    Args:
+        log_dir (str): Directory containing log files
+        node_rank (int): Rank of the node
+        host (str): Expected hostname or IP
+        no_shared_fs (bool): Whether shared filesystem is disabled
+
+    Returns:
+        str: Path to the actual log file found, or expected path if not found
+    """
+    # Construct expected filename based on original logic
+    if no_shared_fs:
+        expected_file = os.path.join(log_dir, "host.output")
+    else:
+        expected_file = os.path.join(log_dir, f"host_{node_rank}_{host}.output")
+
+    # Try exact match first
+    if os.path.exists(expected_file):
+        return expected_file
+
+    # Use glob pattern to find any matching node_rank file
+    if not no_shared_fs:
+        pattern = os.path.join(log_dir, f"host_{node_rank}_*.output")
+        matches = glob.glob(pattern)
+
+        if matches:
+            # Return the first match found
+            logger.debug(
+                f"Smart discovery found log file: {matches[0]} (expected: {expected_file})"
+            )
+            return matches[0]
+
+    # Return original expected path for error handling
+    return expected_file
+
+
+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 = find_actual_log_file(log_dir, node_rank, host, no_shared_fs)
+    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"ssh -p {ssh_port} {host} 'tail -c +{offset + 1} {shlex.quote(src_log_file)}' > {shlex.quote(dest_log_file)}"
+            run_local_command(command, dryrun)
+            logger.debug(
+                f"Collected incremental log from {host} (node {node_rank}) to {dest_log_file}"
+            )
+        else:
+            command = (
+                f"tail -c +{offset + 1} {shlex.quote(src_log_file)} > {shlex.quote(dest_log_file)}"
+            )
+            run_local_command(command, dryrun)
+            logger.debug(f"Collected incremental local log to {dest_log_file}")
+
+        # Check if the source file exists and update the offset
+        if os.path.exists(src_log_file):
+            current_src_size = get_file_size(host, src_log_file)
+            if current_src_size > 0:
+                if current_src_size > offset:  # There is new content in the source file
+                    _log_offsets[log_key] = current_src_size
+
+                    if os.path.exists(dest_log_file) and os.path.getsize(dest_log_file) > 0:
+                        logger.debug(
+                            f"Collected {current_src_size - offset} bytes from {src_log_file}"
+                        )
+                        return dest_log_file
+                    else:
+                        logger.debug(f"No new content extracted from {src_log_file}")
+                        if os.path.exists(dest_log_file):
+                            os.remove(dest_log_file)
+                        return None
+            else:
+                logger.debug(f"No new content in source file {src_log_file}")
+                if os.path.exists(dest_log_file):
+                    os.remove(dest_log_file)
+                return None
+        else:
+            logger.debug(f"Source log file {src_log_file} not found")
+            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