IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

怎样实现服务自动修复脚本

wen 实用脚本 27

本文目录导读：

怎样实现服务自动修复脚本

  1. 基础版 - Shell脚本实现
  2. Python版本 - 功能更强大
  3. 高级实现 - 使用Docker健康检查
  4. Kubernetes实现 - 使用Liveness探针
  5. 使用指南

我来提供一个完整的服务自动修复脚本实现方案,包含多种实现方式：

基础版 - Shell脚本实现

#!/bin/bash
# 服务自动修复脚本 - 基础版
# 文件名: service_auto_repair.sh
# 配置
SERVICE_NAME="your-service-name"
LOG_FILE="/var/log/service_self_repair.log"
MAX_RETRIES=3
RETRY_INTERVAL=5  # 秒
# 日志函数
log_message() {
    local timestamp=$(date '+%Y-%m-%d %H:%M:%S')
    echo "[$timestamp] $1" | tee -a $LOG_FILE
}
# 健康检查函数
check_service_health() {
    # 方法1：检查进程是否在运行
    if pgrep -x "$SERVICE_NAME" > /dev/null 2>&1; then
        return 0
    fi
    # 方法2：通过systemd检查
    if systemctl is-active --quiet "$SERVICE_NAME" 2>/dev/null; then
        return 0
    fi
    # 方法3：通过curl检查HTTP服务
    if curl -f http://localhost:8080/health > /dev/null 2>&1; then
        return 0
    fi
    return 1
}
# 修复服务函数
repair_service() {
    local attempt=$1
    log_message "开始第 $attempt 次修复尝试..."
    # 停止服务
    systemctl stop "$SERVICE_NAME" 2>/dev/null || killall "$SERVICE_NAME" 2>/dev/null
    sleep 2
    # 清理状态
    rm -f /var/run/"$SERVICE_NAME".pid 2>/dev/null
    # 启动服务
    systemctl start "$SERVICE_NAME" 2>/dev/null || {
        # 备用启动方式
        /usr/bin/$SERVICE_NAME &
    }
    sleep $RETRY_INTERVAL
    # 验证服务是否正常运行
    if check_service_health; then
        log_message "第 $attempt 次修复成功！"
        return 0
    fi
    return 1
}
# 主修复流程
main() {
    log_message "服务修复脚本启动 - 检查: $SERVICE_NAME"
    # 检查服务是否正常运行
    if check_service_health; then
        log_message "服务正常运行，无需修复"
        exit 0
    fi
    log_message "服务异常，开始修复..."
    # 重试修复
    for ((i=1; i<=MAX_RETRIES; i++)); do
        if repair_service $i; then
            exit 0
        fi
        if [ $i -lt $MAX_RETRIES ]; then
            log_message "等待 $RETRY_INTERVAL 秒后重试..."
            sleep $RETRY_INTERVAL
        fi
    done
    # 所有重试都失败
    log_message "修复失败！开始深度修复..."
    deep_repair_service
}
# 深度修复
deep_repair_service() {
    log_message "执行深度修复方案..."
    # 重启依赖服务
    systemctl restart network 2>/dev/null
    systemctl restart mysql 2>/dev/null
    # 清理日志和缓存
    journalctl --vacuum-time=1d 2>/dev/null
    sync && echo 3 > /proc/sys/vm/drop_caches 2>/dev/null
    # 重新安装服务（示例）
    # yum reinstall -y $SERVICE_NAME 2>/dev/null
    # 最后尝试重启
    systemctl restart "$SERVICE_NAME"
    if check_service_health; then
        log_message "深度修复成功！"
    else
        log_message "深度修复失败，需要人工介入"
        send_alert "服务 $SERVICE_NAME 修复失败"
    fi
}
# 告警函数
send_alert() {
    local message="$1"
    # 发送邮件告警
    echo "$message" | mail -s "[ALERT] 服务修复失败" admin@example.com
    # 发送Slack通知
    curl -X POST -H 'Content-type: application/json' \
        --data "{\"text\":\"$message\"}" \
        https://hooks.slack.com/services/YOUR/WEBHOOK/URL
}
# 设置crontab定时任务
setup_cron() {
    local cron_job="*/5 * * * * /path/to/$0"
    (crontab -l 2>/dev/null | grep -v "$0"; echo "$cron_job") | crontab -
    log_message "已添加定时任务：每5分钟执行一次"
}
# 执行主函数
main "$@"

Python版本 - 功能更强大

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
服务自动修复脚本 - Python版本
文件名: service_auto_repair.py
"""
import os
import sys
import time
import logging
import subprocess
from datetime import datetime
import requests
import json
import smtplib
from email.mime.text import MIMEText
import argparse
class ServiceAutoRepair:
    """服务自动修复类"""
    def __init__(self, service_name, config_file=None):
        self.service_name = service_name
        self.max_retries = 3
        self.retry_interval = 5
        self.log_file = "/var/log/service_self_repair.log"
        self.setup_logging()
        # 配置邮箱告警
        self.email_config = {
            'smtp_server': 'smtp.gmail.com',
            'smtp_port': 587,
            'username': 'your-email@gmail.com',
            'password': 'your-password',
            'to_email': 'admin@example.com'
        }
        # 配置Slack告警
        self.slack_webhook_url = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
    def setup_logging(self):
        """设置日志"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(self.log_file),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def check_health(self):
        """检查服务健康状态"""
        try:
            # 方法1：systemd检查
            result = subprocess.run(
                ['systemctl', 'is-active', '--quiet', self.service_name],
                capture_output=True
            )
            if result.returncode == 0:
                return True
            # 方法2：进程检查
            result = subprocess.run(
                ['pgrep', '-x', self.service_name],
                capture_output=True
            )
            if result.returncode == 0:
                return True
            # 方法3：HTTP检查
            try:
                response = requests.get(
                    'http://localhost:8080/health',
                    timeout=5
                )
                if response.status_code == 200:
                    return True
            except:
                pass
        except Exception as e:
            self.logger.error(f"健康检查失败: {str(e)}")
        return False
    def repair_service(self, attempt):
        """修复服务"""
        self.logger.info(f"开始第 {attempt} 次修复尝试...")
        try:
            # 停止服务
            subprocess.run(['systemctl', 'stop', self.service_name], 
                          capture_output=True, check=False)
            time.sleep(2)
            # 清理PID文件
            pid_file = f"/var/run/{self.service_name}.pid"
            if os.path.exists(pid_file):
                os.remove(pid_file)
            # 启动服务
            subprocess.run(['systemctl', 'start', self.service_name], 
                          capture_output=True, check=False)
            time.sleep(self.retry_interval)
            if self.check_health():
                self.logger.info(f"第 {attempt} 次修复成功！")
                return True
        except Exception as e:
            self.logger.error(f"修复失败: {str(e)}")
        return False
    def deep_repair(self):
        """深度修复"""
        self.logger.info("执行深度修复...")
        try:
            # 重启依赖服务
            for dep_service in ['network', 'mysql', 'nginx']:
                subprocess.run(['systemctl', 'restart', dep_service],
                              capture_output=True, check=False)
            # 清理缓存
            with open('/proc/sys/vm/drop_caches', 'w') as f:
                f.write('3')
            # 清理日志
            subprocess.run(['journalctl', '--vacuum-time=1d'],
                          capture_output=True, check=False)
            # 重新启动服务
            subprocess.run(['systemctl', 'restart', self.service_name],
                          capture_output=True, check=False)
            time.sleep(5)
            if self.check_health():
                self.logger.info("深度修复成功！")
                return True
        except Exception as e:
            self.logger.error(f"深度修复失败: {str(e)}")
        return False
    def send_alert(self, message):
        """发送告警通知"""
        self.logger.warning(f"发送告警: {message}")
        # 发送邮件
        try:
            msg = MIMEText(message)
            msg['Subject'] = f'[ALERT] 服务修复失败 - {self.service_name}'
            msg['From'] = self.email_config['username']
            msg['To'] = self.email_config['to_email']
            with smtplib.SMTP(
                self.email_config['smtp_server'],
                self.email_config['smtp_port']
            ) as server:
                server.starttls()
                server.login(
                    self.email_config['username'],
                    self.email_config['password']
                )
                server.send_message(msg)
        except Exception as e:
            self.logger.error(f"发送邮件失败: {str(e)}")
        # 发送Slack通知
        try:
            requests.post(
                self.slack_webhook_url,
                json={'text': message},
                timeout=5
            )
        except Exception as e:
            self.logger.error(f"发送Slack通知失败: {str(e)}")
    def run(self):
        """主修复流程"""
        self.logger.info(f"服务修复脚本启动 - 检查: {self.service_name}")
        # 检查服务是否正常运行
        if self.check_health():
            self.logger.info("服务正常运行，无需修复")
            return True
        self.logger.info("服务异常，开始修复...")
        # 重试修复
        for attempt in range(1, self.max_retries + 1):
            if self.repair_service(attempt):
                return True
            if attempt < self.max_retries:
                self.logger.info(f"等待 {self.retry_interval} 秒后重试...")
                time.sleep(self.retry_interval)
        # 所有重试都失败
        self.logger.error("标准修复失败，尝试深度修复...")
        if self.deep_repair():
            return True
        # 完全失败
        error_msg = f"服务 {self.service_name} 自动修复失败，需要人工介入"
        self.logger.error(error_msg)
        self.send_alert(error_msg)
        return False
    def add_cron_job(self, interval=5):
        """添加定时任务"""
        script_path = os.path.abspath(__file__)
        cron_job = f"*/{interval} * * * * /usr/bin/python3 {script_path} --service {self.service_name}"
        # 获取当前crontab
        result = subprocess.run(['crontab', '-l'], capture_output=True, text=True)
        current_cron = result.stdout
        # 移除旧的定时任务
        new_cron = "\n".join(
            [line for line in current_cron.split('\n') 
             if self.service_name not in line]
        )
        # 添加新任务
        if new_cron:
            new_cron += "\n" + cron_job
        else:
            new_cron = cron_job
        # 写入crontab
        with subprocess.Popen(['crontab'], stdin=subprocess.PIPE) as proc:
            proc.communicate(new_cron.encode())
        self.logger.info(f"已添加定时任务：每{interval}分钟执行一次")
    def generate_report(self):
        """生成修复报告"""
        report = {
            'timestamp': datetime.now().isoformat(),
            'service': self.service_name,
            'status': 'healthy' if self.check_health() else 'unhealthy',
            'repair_attempts': self.max_retries,
            'last_check': datetime.now().isoformat()
        }
        # 保存报告到文件
        report_file = f"/var/log/{self.service_name}_repair_report.json"
        with open(report_file, 'w') as f:
            json.dump(report, f, indent=2)
        self.logger.info(f"报告已保存到: {report_file}")
        return report
def main():
    """主函数"""
    parser = argparse.ArgumentParser(description='服务自动修复脚本')
    parser.add_argument('--service', required=True, help='服务名称')
    parser.add_argument('--setup-cron', action='store_true', help='设置定时任务')
    parser.add_argument('--interval', type=int, default=5, help='检查间隔（分钟）')
    args = parser.parse_args()
    repair = ServiceAutoRepair(args.service)
    if args.setup_cron:
        repair.add_cron_job(args.interval)
    else:
        repair.run()
        repair.generate_report()
if __name__ == '__main__':
    main()

高级实现 - 使用Docker健康检查

# docker-compose.yml
version: '3.8'
services:
  app:
    image: your-app:latest
    container_name: your-app
    restart: always
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 40s
    environment:
      - AUTO_REPAIR=true
    volumes:
      - ./repair.sh:/usr/local/bin/repair.sh
# Docker容器内的修复脚本
#!/bin/bash
while true; do
    # 检查健康状态
    if ! curl -f http://localhost:8080/health; then
        echo "Service unhealthy, attempting repair..."
        # 清理和重启
        rm -rf /tmp/cache/*
        kill -HUP 1  # 向init进程发送SIGHUP
        # 记录日志
        logger "Service auto-repair triggered"
    fi
    sleep 60
done

Kubernetes实现 - 使用Liveness探针

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: app-deployment
spec:
  replicas: 3
  selector:
    matchLabels:
      app: myapp
  template:
    metadata:
      labels:
        app: myapp
    spec:
      containers:
      - name: app
        image: your-app:latest
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
          timeoutSeconds: 5
          failureThreshold: 3
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 5
        lifecycle:
          preStop:
            exec:
              command: ["/bin/sh", "-c", "sleep 5"]

使用指南

安装和配置

# 1. 下载脚本
curl -O https://your-server.com/service_auto_repair.sh
chmod +x service_auto_repair.sh
# 2. 配置服务
SERVICE_NAME="nginx"
MAX_RETRIES=5
RETRY_INTERVAL=10
# 3. 设置定时任务
./service_auto_repair.sh --setup-cron
# 或手动添加crontab
crontab -e
*/5 * * * * /path/to/service_auto_repair.sh --service nginx
# 4. 日志查看
tail -f /var/log/service_self_repair.log
# 5. 测试
./service_auto_repair.sh --dry-run

最佳实践

  1. 完善健康检查：

    • 实现自定义的健康检查端点
    • 检查数据库连接
    • 验证缓存状态
    • 监控资源使用情况

  2. 分级修复策略：

    • 第1级：重启服务
    • 第2级：清理缓存和临时文件
    • 第3级：重启依赖服务
    • 第4级：重新部署

  3. 告警机制：

    • 邮件通知
    • Slack/企业微信
    • PagerDuty集成
    • 短信告警

  4. 日志和监控：

    • 详细日志记录
    • Prometheus指标导出
    • Grafana仪表板

  5. 安全考虑：

    • 限制脚本权限
    • 使用非root用户运行
    • 加密敏感信息

这个解决方案提供了从简单到复杂的多种实现方式,可以根据实际需求选择适合的方案。

上一篇脚本如何巡检集群节点状态

下一篇脚本如何检测服务运行状态

相关文章

抱歉，评论功能暂时关闭!