求Python2.7/Shell脚本：多服务器日志收集合并及错误检测终止部署

Alright, let's work through your requirement step by step. You need to collect logs from multiple servers, merge them into a single file, search for error messages, and halt the deployment workflow if any errors are detected. Since you prefer Python 2.7, I'll start with that solution, then share a Shell script alternative which is also straightforward.

This script uses paramiko to securely fetch logs from remote servers, merges them locally, and checks for error patterns. If errors are found, it exits with a non-zero code to trigger workflow termination.

import paramiko

# Configuration - adjust these values to match your environment
SERVERS = [
    {"host": "server1.example.com", "user": "your_username", "log_paths": ["/var/app/logs/app1.log", "/var/app/logs/app2.log"]},
    {"host": "server2.example.com", "user": "your_username", "log_paths": ["/var/app/logs/app1.log"]},
]
MERGED_LOG_FILE = "merged_app_logs.log"
ERROR_PATTERNS = ["ERROR:", "Exception:", "Failed to initialize", "Critical failure"]

def fetch_and_merge_logs():
    """Fetch logs from all servers and merge into a single file"""
    with open(MERGED_LOG_FILE, "w") as merged_file:
        for server in SERVERS:
            ssh_client = paramiko.SSHClient()
            ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            
            try:
                # Connect to the server (ensure passwordless SSH is set up)
                ssh_client.connect(server["host"], username=server["user"])
                
                # Fetch each log file from the server
                for log_path in server["log_paths"]:
                    stdin, stdout, stderr = ssh_client.exec_command(f"cat {log_path}")
                    log_content = stdout.read()
                    
                    # Write server and log path header for clarity
                    merged_file.write(f"=== Log from {server['host']}: {log_path} ===\n")
                    merged_file.write(log_content)
                    merged_file.write("\n\n")
                    
            except Exception as e:
                print(f"Error fetching logs from {server['host']}: {str(e)}")
                exit(1)
            finally:
                ssh_client.close()

def check_for_errors():
    """Search merged log file for error patterns"""
    errors_found = False
    with open(MERGED_LOG_FILE, "r") as log_file:
        for line in log_file:
            if any(pattern in line for pattern in ERROR_PATTERNS):
                print(f"Detected error: {line.strip()}")
                errors_found = True
    return errors_found

if __name__ == "__main__":
    print("Starting log collection and error check...")
    fetch_and_merge_logs()
    print(f"Merged logs saved to {MERGED_LOG_FILE}")
    
    if check_for_errors():
        print("Errors detected! Terminating deployment workflow.")
        exit(1)
    else:
        print("No errors found. Proceeding with deployment.")
        exit(0)

Setup Notes: #

• Install paramiko first: pip install paramiko (make sure this is for Python 2.7)
• Set up passwordless SSH access to all target servers using ssh-keygen and ssh-copy-id your_username@server_host
• Update the SERVERS, ERROR_PATTERNS, and log path values to match your infrastructure

If you prefer a Shell script (no Python dependencies), this uses ssh to fetch logs, cat to merge, and grep to detect errors. It also exits with a non-zero code on failure.

#!/bin/bash

# Configuration - adjust these values
SERVERS=("your_username@server1.example.com" "your_username@server2.example.com")
LOG_PATHS=("/var/app/logs/app1.log" "/var/app/logs/app2.log")
MERGED_LOG_FILE="merged_app_logs.log"
ERROR_PATTERNS="ERROR:|Exception:|Failed to|Critical failure"

# Step 1: Fetch and merge logs from all servers
echo "Fetching logs from remote servers..."
> "$MERGED_LOG_FILE"  # Clear existing merged file

for server in "${SERVERS[@]}"; do
    for log_path in "${LOG_PATHS[@]}"; do
        echo "=== Log from $server: $log_path ===" >> "$MERGED_LOG_FILE"
        # Fetch log via SSH; exit if fetch fails
        ssh "$server" "cat $log_path" >> "$MERGED_LOG_FILE" 2>/dev/null
        if [ $? -ne 0 ]; then
            echo "Failed to fetch log $log_path from $server"
            exit 1
        fi
        echo -e "\n\n" >> "$MERGED_LOG_FILE"
    done
done

echo "Merged logs saved to $MERGED_LOG_FILE"

# Step 2: Check for errors in merged log
echo "Scanning for errors..."
grep -E "$ERROR_PATTERNS" "$MERGED_LOG_FILE"

if [ $? -eq 0 ]; then
    echo "Errors detected! Terminating deployment workflow."
    exit 1
else
    echo "No errors found. Proceeding with deployment."
    exit 0
fi

Setup Notes: #

• Ensure passwordless SSH access is configured for all servers
• Make the script executable: chmod +x log_checker.sh
• Adjust the error patterns, server list, and log paths as needed

Key Workflow Integration Tip: #

Most deployment tools (Jenkins, GitLab CI, GitHub Actions) automatically terminate workflows when a script exits with a non-zero code. So running either of these scripts as part of your deployment pipeline will handle the termination automatically if errors are found.

内容的提问来源于stack exchange，提问作者Nikhilesh