如何编写自动处理错误脚本

本文目录导读：

1. 目录导读
2. 总结与下一步行动

如何编写自动处理错误脚本,打造零故障运维体系

目录导读

1. 为什么需要自动错误处理脚本——解决人工排查的三大痛点
2. 自动错误处理的核心设计原则——少写代码，多留后路
3. 脚本语言选型与基础框架——Bash vs Python vs PowerShell
4. 五种常见异常场景的自动捕获逻辑——日志、API、进程、网络、磁盘
5. 错误分级与智能重试策略——别让脚本变成“崩溃雪崩”
6. 日志记录与通知机制的黄金组合——在哪里写日志，往哪里发告警
7. 实战案例：自动修复Web服务502错误的完整脚本
8. 常见问题与避坑指南——那些让你凌晨3点起床的脚本错误

---

为什么需要自动错误处理脚本

据某运维社区2024年统计,企业IT团队平均每天要处理3.2次非计划性故障，其中73%的故障可以通过脚本自动恢复，当你还在熬夜查日志时，隔壁团队已经用自动错误处理脚本完成了“自愈”——坏掉的Nginx自动重启，磁盘满了自动清理缓存，崩溃的数据库进程自动拉起。

自动错误处理脚本不是“写个try-catch就完事”，而是一套检测→判断→决策→执行→验证→记录的完整流程，它的核心价值在于：

• 减少MTTR（平均修复时间）：从人工30分钟下降到脚本10秒
• 降低人为误操作：凌晨两点的手抖打错命令是宕机的主要诱因
• 实现故障闭环：从“看到错误”到“修复验证”全自动化

自动错误处理的核心设计原则

幂等性
脚本无论执行一次还是十次，对系统的影响必须一致，例如重启服务前先检查进程是否存在，不要重复启动造成资源争抢。

防御性编程
假设所有外部调用都可能会失败，读取文件前先检测是否存在，执行命令前先检查用户权限，连接数据库前先超时设置。

渐进式处理
错误越严重，处理力度越小——别因为一次502就把整台服务器重启了，先重试一次，不行再重启服务，再不行才重启服务器，最后才是报警通知人类。

脚本语言选型与基础框架

语言	适用场景	错误处理特点
Bash	系统级故障、进程管理、日志搜索	$?判断退出码，trap捕获信号
Python	复杂逻辑、API调用、多条件判断	try-except-else-finally，自定义异常类
PowerShell	Windows系统、Exchange、Azure	$Error变量，Try-Catch-Finally

推荐组合方案：
核心逻辑用Python（可读性强、库丰富），系统命令调用用subprocess模块，定时任务用crontab或systemd timer。

基础框架模板（Python）：

import sys
import logging
import subprocess
logging.basicConfig(filename='auto_heal.log', level=logging.INFO, 
                    format='%(asctime)s - %(levelname)s - %(message)s')
def check_service_status(service_name):
    try:
        result = subprocess.run(['systemctl', 'is-active', service_name],
                                capture_output=True, text=True, timeout=10)
        return result.stdout.strip() == 'active'
    except Exception as e:
        logging.error(f"检查服务状态失败: {e}")
        return False

五种常见异常场景的自动捕获逻辑

日志异常

# 检测最近30秒是否有ERROR级别日志
import time
def detect_log_error(log_path, keywords=['ERROR', 'FATAL'], window_seconds=30):
    with open(log_path, 'r') as f:
        f.seek(0, 2)  # 移到文件末尾
        current_size = f.tell()
        # 读取最近写入的日志块
        f.seek(max(0, current_size - 10240))  # 读取10KB
        recent_logs = f.readlines()
    # 过滤出时间窗口内的错误
    for line in recent_logs:
        if any(k in line for k in keywords):
            return True
    return False

API返回状态码非200

import requests
import time
def check_api_health(url, retries=3, timeout=5):
    for attempt in range(retries):
        try:
            resp = requests.get(url, timeout=timeout)
            if resp.status_code == 200:
                return True
            else:
                logging.warning(f"API返回{resp.status_code}，尝试第{attempt+1}次")
        except requests.exceptions.Timeout:
            logging.error("API响应超时")
        except requests.exceptions.ConnectionError:
            logging.error("API连接失败")
        time.sleep(2 ** attempt)  # 指数退避
    return False

进程异常退出

#!/bin/bash
PROCESS_NAME="java"
if ! pgrep -x "$PROCESS_NAME" > /dev/null; then
    echo "进程未运行，尝试重启"
    systemctl restart my_java_app
    sleep 3
    if pgrep -x "$PROCESS_NAME" > /dev/null; then
        echo "重启成功"
        # 发送成功通知
    else
        echo "重启失败，需要人工介入"
        # 发送告警通知
    fi
fi

网络不通

import socket
def check_network(host="8.8.8.8", port=53, timeout=3):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        result = sock.connect_ex((host, port))
        sock.close()
        return result == 0
    except Exception:
        return False

磁盘空间不足

#!/bin/bash
THRESHOLD=85
CURRENT=$(df / | grep / | awk '{ print $5}' | sed 's/%//g')
if [ "$CURRENT" -gt "$THRESHOLD" ]; then
    echo "磁盘使用率达${CURRENT}%，清理/tmp下3天前的文件"
    find /tmp -type f -mtime +3 -delete
    # 或者清理docker日志
    docker system prune -f --volumes
fi

错误分级与智能重试策略

错误分级模型：

级别	描述	处理策略
INFO	瞬态错误，如网络抖动	重试3次，间隔1-5秒
WARNING	临时资源不足，如磁盘IO高	重试+扩容/清理
ERROR	服务异常，如进程挂掉	自动重启+日志采集
CRITICAL	硬件故障，如磁盘损坏	立即报警，停止自动操作

智能重试策略示例（Python）：

import random
def retry_with_backoff(func, max_retries=3, base_delay=1):
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise  # 最后一次失败直接抛出
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
            logging.warning(f"重试第{attempt+1}次，等待{delay:.2f}秒")
            time.sleep(delay)

日志记录与通知机制的黄金组合

日志记录要点：

• 记录时间戳、脚本名称、错误类型、上下文变量
• 区分ERROR和WARN级别,ERROR触发告警
• 日志文件自动轮转（logrotate设置）

通知机制推荐：

• 即时通知：企业微信机器人webhook、钉钉机器人、Slack
• 短信告警（仅CRITICAL）：集成阿里云短信、Twilio
• 工单系统：自建或对接Jira、Zendesk

Python通知示例：

def send_alert(message, level='WARN'):
    if level == 'CRITICAL':
        # 发短信：twilio
        pass
    # 发送到企业微信
    webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx"
    data = {
        "msgtype": "text",
        "text": {"content": f"[{level}] {message}"}
    }
    requests.post(webhook_url, json=data)

实战案例：自动修复Web服务502错误的完整脚本

场景：Nginx反向代理时，上游应用偶尔返回502，需要自动重启应用服务并验证。

完整脚本（Python）：

#!/usr/bin/env python3
import subprocess, time, requests, logging
# 配置
UPSTREAM_URL = "http://localhost:8080/health"
SERVICE_NAME = "my_app"
NGINX_SERVICE = "nginx"
LOG_FILE = "/var/log/auto_heal.log"
ALERT_WEBHOOK = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx"
logging.basicConfig(filename=LOG_FILE, level=logging.INFO,
                    format='%(asctime)s %(levelname)s %(message)s')
def check_upstream():
    """检查上游服务健康状态"""
    try:
        resp = requests.get(UPSTREAM_URL, timeout=5)
        return resp.status_code in [200, 204]
    except Exception:
        return False
def restart_service(service):
    """重启系统服务"""
    try:
        subprocess.run(['systemctl', 'restart', service], check=True, timeout=30)
        return True
    except subprocess.CalledProcessError:
        return False
def verify_service():
    """验证服务是否正常"""
    time.sleep(5)  # 等待服务启动
    return check_upstream()
def main():
    if check_upstream():
        logging.info("上游服务正常")
        return
    logging.warning("检测到上游服务异常，尝试重启应用...")
    if restart_service(SERVICE_NAME):
        if verify_service():
            logging.info("应用重启成功，服务恢复正常")
            # 可选：发送恢复通知
        else:
            logging.error("应用重启后仍有问题，尝试重启Nginx")
            if restart_service(NGINX_SERVICE):
                time.sleep(3)
                if verify_service():
                    logging.info("Nginx重启后服务恢复")
                else:
                    logging.critical("所有自动恢复策略均失败，需要人工介入")
                    # 发送CRITICAL告警
                    requests.post(ALERT_WEBHOOK, json={"msgtype":"text",
                        "text":{"content":"[CRITICAL] Web服务502自动修复失败，请立即处理"}})
            else:
                logging.critical("Nginx重启失败，立即报警")
    else:
        logging.critical("应用重启失败，立即报警")
if __name__ == "__main__":
    main()

配置到crontab（每分钟执行一次）：

* * * * * /usr/bin/python3 /opt/scripts/auto_heal.py

常见问题与避坑指南

Q1：脚本自己崩溃了怎么办？
A：设置Supervisor或systemd守护进程，使脚本具备自愈能力；关键脚本增加看门狗（watchdog）监控。

Q2：重试导致雪崩式重启怎么办？
A：加入“熔断机制”——连续失败3次后等待10分钟再重试，或者使用Redis记录失败次数（setex设置过期时间）。

Q3：如何防止脚本误判？
A：避免单点检测，检测服务健康时同时检查端口、进程、HTTP状态码三个维度，两票通过才视为正常。

Q4：日志增长太快怎么办？
A：配置日志轮转（logrotate每天切割，保留7天）；核心逻辑只记录事件摘要，详细调试日志用单独的debug.log文件。

Q5：需要处理哪些边界情况？
A：脚本运行期间服务器重启、磁盘IO满导致日志写不进去、crontab执行时间重叠等，建议在脚本开头检查是否为单实例运行（文件锁/端口锁）。

---

总结与下一步行动

编写自动错误处理脚本的关键在于设计思维——不是“出现错误就修”，而是“如何优雅地不犯错、犯错后快速恢复、恢复后记录原因”，建议从最简单的“检测单服务+重启”开始，逐步增加网络检测、日志分析、资源清理、分级告警等功能。

立即开始：

1. 盘点你系统中最频繁出现的3类错误
2. 为每个错误编写一个10行以内的检测脚本
3. 加入指数退避重试
4. 配置crontab定时执行
5. 观察一周,根据false positive率调整阈值

当你不再被凌晨的报警电话惊醒时,你会感谢现在动手写脚本的自己。

（全文约2200字）