Add the corresponding log line to the diagnostic file, and fix some bugs

Author: wanglei19991004

.github/workflows/unit-tests-elastic.yml

@@ -0,0 +1,155 @@
+name: Elastic Module Tests
+
+on:
+  push:
+    branches: [ main, develop ]
+    paths:
+      - 'flagscale/runner/elastic/**'
+      - 'tests/unit_tests/runner/elastic/**'
+      - '.github/workflows/unit-tests-elastic.yml'
+  pull_request:
+    branches: [ main, develop ]
+    paths:
+      - 'flagscale/runner/elastic/**'
+      - 'tests/unit_tests/runner/elastic/**'
+      - '.github/workflows/unit-tests-elastic.yml'
+
+jobs:
+  lint-and-test:
+    runs-on: [self-hosted, Linux, X64, nvidia-0, gpus-8]
+    container:
+      image: flagscale/flagscale:dev-megatron
+      ports:
+        - 80
+      volumes:
+        - /home/flagscale_cicd/flask/static:/workspace/report
+        - /home/flagscale_cicd/flask/config:/workspace/config
+        - /home/flagscale_cicd/docker/docker_build/docker_data:/home/gitlab-runner/data
+        - /home/flagscale_cicd/docker/docker_build/docker_tokenizers:/home/gitlab-runner/tokenizers
+      options: --gpus all --shm-size=500g --hostname flagscale_cicd --user root --ulimit nofile=65535:65535
+    
+    steps:
+      - name: Checkout Code
+        uses: actions/checkout@v4
+        with:
+          repository: ${{ github.event.pull_request.head.repo.full_name || github.repository }}
+          ref: ${{ github.event.pull_request.head.ref || github.ref }}
+          ssh-strict: true
+          ssh-user: git
+          persist-credentials: true
+          clean: true
+          sparse-checkout-cone-mode: true
+          fetch-tags: false
+          show-progress: true
+          lfs: false
+          submodules: false
+          set-safe-directory: true
+
+      - name: Setup Environment
+        run: |
+          echo "USER: $USER"
+          echo "UID: $(id -u)"
+          echo "GID: $(id -g)"
+          echo "Home: $HOME"
+          whoami
+          git config --global --add safe.directory /__w/FlagScale/FlagScale || true
+
+      - name: Install Dependencies
+        run: |
+          source /root/miniconda3/etc/profile.d/conda.sh
+          conda activate flagscale-train
+          pip install pylint pytest pytest-cov
+          python tools/patch/unpatch.py --backend Megatron-LM || true
+          export PYTHONPATH=./third_party/Megatron-LM:$PYTHONPATH
+          export NVTE_FLASH_ATTN=0
+          export NVTE_FUSED_ATTN=0
+          ulimit -n 65535
+
+      - name: Pylint Check - Elastic Module
+        run: |
+          source /root/miniconda3/etc/profile.d/conda.sh
+          conda activate flagscale-train
+          export PYTHONPATH=./third_party/Megatron-LM:$PYTHONPATH
+          
+          echo "Running pylint on elastic module..."
+          pylint flagscale/runner/elastic/ --output-format=text --reports=yes --exit-zero > pylint_report.txt
+          
+          # Display pylint results
+          cat pylint_report.txt
+          
+          # Extract pylint score
+          PYLINT_SCORE=$(grep "Your code has been rated at" pylint_report.txt | sed 's/.*rated at \([0-9.]*\).*/\1/' || echo "0")
+          echo "Pylint Score: $PYLINT_SCORE"
+          
+          # Fail if score is below 8.0
+          if (( $(echo "$PYLINT_SCORE < 8.0" | bc -l) )); then
+            echo "Pylint score $PYLINT_SCORE is below required threshold of 8.0"
+            exit 1
+          fi
+
+      - name: Run Elastic Unit Tests
+        run: |
+          source /root/miniconda3/etc/profile.d/conda.sh
+          conda activate flagscale-train
+          export PYTHONPATH=./third_party/Megatron-LM:$PYTHONPATH
+          export NVTE_FLASH_ATTN=0
+          export NVTE_FUSED_ATTN=0
+          ulimit -n 65535
+          
+          echo "Running elastic module unit tests..."
+          pytest tests/unit_tests/runner/elastic/ \
+            --cov=flagscale/runner/elastic \
+            --cov-report=xml:coverage_elastic.xml \
+            --cov-report=html:coverage_elastic_html \
+            --cov-report=term \
+            -v \
+            --tb=short \
+            -x
+
+      - name: Coverage Report
+        run: |
+          source /root/miniconda3/etc/profile.d/conda.sh
+          conda activate flagscale-train
+          
+          echo "Coverage Summary for Elastic Module:"
+          if [ -f coverage_elastic.xml ]; then
+            echo "Coverage XML report generated successfully"
+            python -c "
+import xml.etree.ElementTree as ET
+try:
+    tree = ET.parse('coverage_elastic.xml')
+    root = tree.getroot()
+    coverage = root.attrib.get('line-rate', '0')
+    percentage = float(coverage) * 100
+    print(f'Line Coverage: {percentage:.1f}%')
+    if percentage < 80:
+        print(f'Warning: Coverage {percentage:.1f}% is below recommended 80%')
+        exit(1)
+except Exception as e:
+    print(f'Error parsing coverage: {e}')
+    exit(1)
+"
+          else
+            echo "Coverage XML report not found"
+            exit 1
+          fi
+
+      - name: Archive Test Results
+        uses: actions/upload-artifact@v3
+        if: always()
+        with:
+          name: elastic-test-results
+          path: |
+            pylint_report.txt
+            coverage_elastic.xml
+            coverage_elastic_html/
+          retention-days: 30
+
+      - name: Summary
+        if: always()
+        run: |
+          echo "=== Elastic Module CI/CD Summary ==="
+          echo "[INFO] Pylint check completed"
+          echo "[INFO] Unit tests executed" 
+          echo "[INFO] Coverage report generated"
+          echo "=== Test artifacts uploaded to GitHub Actions ==="

flagscale/runner/elastic/diagnostic.py

@@ -1,9 +1,11 @@
 import os
-
-from datetime import datetime
+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.",
@@ -23,6 +25,12 @@
     "error": "GeneralError: An error occurred during training.",
     # Process errors
     "killed": "ProcessKilled: Training process was killed.",
+    "kill": "ProcessKilled: Process was manually killed.",
+    "sigterm": "ProcessKilled: Process received SIGTERM signal.",
+    "sigkill": "ProcessKilled: Process received SIGKILL signal.",
+    "signal": "ProcessKilled: Process received termination signal.",
+    "terminated": "ProcessKilled: Process was terminated.",
+    "sigint": "ProcessKilled: Process received SIGINT signal (Ctrl+C).",
     "segmentation fault": "SegmentationFault: Process crashed due to memory access error.",
     "core dumped": "CoreDump: Process crashed and dumped core.",
     # CUDA errors
@@ -39,59 +47,120 @@
 }
 
 
+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 a diagnostic report from a log file.
+    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 None.
+        str: Diagnostic report content if return_content=True, else diagnostic file path.
     """
-    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"
+    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:
-            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"
+            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()
 
-    if return_content:
-        return report_content
-    else:
-        try:
-            diagnostic_file = log_file.replace("temp", "diagnostic").replace(".log", ".txt")
+        # 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 line
+        current_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())
+        new_errors = []
+        for key, desc in error_types.items():
+            # Look for errors in the newly added lines
+            error_lines = find_error_lines(all_lines, key, last_analyzed_line)
+            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, 'w') as f:
-                f.write(report_content)
+            with open(diagnostic_file, 'a', encoding='utf-8') as f:
+                for error in new_errors:
+                    f.write(f"{error}\n")
+
             logger.debug(
-                f"Generated diagnostic report for {host} (node {node_rank}) at {diagnostic_file}"
+                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 write diagnostic report for {host} (node {node_rank}): {e}")
+    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

@@ -49,12 +49,27 @@ def collect_logs(config, host, node_rank, destination_dir, dryrun=False):
             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
+        # Check if the source file exists and update the offset
+        if os.path.exists(src_log_file):
+            current_src_size = os.path.getsize(src_log_file)
+            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.warning(f"Log file {src_log_file} not found or empty")
+            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