运维提效利器:实战自动化脚本分享与解析

在日常的运维工作中,我们经常被繁琐、重复的任务所包围:每天巡检几十台服务器的状态、清理不断膨胀的日志文件、在告警风暴中人工过滤有效信息……这些不仅耗费大量精力,还容易因人为疏忽导致线上故障。自动化是运维走向现代化的必经之路,而编写脚本则是实现自动化最轻量、最高效的手段。

本文将结合实际工作场景,分享三款实用的 Shell 与 Python 运维自动化脚本,并解析其核心逻辑,希望能为你的日常运维工作提效。


一、 Shell 脚本:多维度服务器健康巡检

服务器巡检是运维的日常,面对多台主机,手动执行 topdf 等命令效率极低。下面这个 Shell 脚本可以快速抓取 CPU、内存、磁盘的关键指标,并在发现异常时输出告警。


#!/bin/bash
# 描述:服务器基础健康巡检脚本
# 用法:./health_check.sh

# 定义阈值
CPU_THRESHOLD=80    # CPU使用率告警阈值(%)
MEM_THRESHOLD=80    # 内存使用率告警阈值(%)
DISK_THRESHOLD=85   # 磁盘使用率告警阈值(%)

echo "========== 服务器健康巡检报告 =========="
echo "主机名: $(hostname)"
echo "巡检时间: $(date '+%Y-%m-%d %H:%M:%S')"
echo "----------------------------------------"

# 1. CPU使用率检查 (取idle值,用100减去)
cpu_idle=$(top -bn1 | grep "Cpu(s)" | awk '{print $8}')
cpu_usage=$(echo "100 - $cpu_idle" | bc)
echo "【CPU使用率】: ${cpu_usage}%"
if (( $(echo "$cpu_usage > $CPU_THRESHOLD" | bc -l) )); then
    echo "⚠️  警告: CPU使用率超过 ${CPU_THRESHOLD}%!"
fi

# 2. 内存使用率检查
mem_total=$(free -m | awk '/Mem/{print $2}')
mem_used=$(free -m | awk '/Mem/{print $3}')
mem_usage=$(echo "scale=2; $mem_used/$mem_total * 100" | bc)
echo "【内存使用率】: ${mem_usage}% (已用: ${mem_used}MB / 总计: ${mem_total}MB)"
if (( $(echo "$mem_usage > $MEM_THRESHOLD" | bc -l) )); then
    echo "⚠️  警告: 内存使用率超过 ${MEM_THRESHOLD}%!"
fi

# 3. 磁盘使用率检查
echo "【磁盘使用率】:"
df -h | grep -vE '^Filesystem|tmpfs|cdrom' | awk '{print $1, $5, $6}' | while read partition usage mountpoint; do
    usage_num=${usage%\%}
    echo "  挂载点 $mountpoint ($partition): 已用 $usage"
    if [ "$usage_num" -gt "$DISK_THRESHOLD" ]; then
        echo "  ⚠️  警告: 磁盘 $mountpoint 使用率超过 ${DISK_THRESHOLD}%!"
    fi
done

echo "========================================="

解析与亮点:


二、 Python 脚本:日志异常关键字监控与钉钉告警

当系统发生故障时,日志是最直接的线索。此 Python 脚本用于实时监控应用日志,一旦捕获到 ERRORException 等关键字,立即通过 Webhook 推送钉钉告警,实现从“人找日志”到“日志找人”。


#!/usr/bin/env python3
# 描述:日志异常监控与钉钉告警脚本

import time
import requests
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# 配置项
LOG_FILE = "/var/log/myapp/app.log"
KEYWORDS = ["ERROR", "Exception", "OutOfMemory"]
DINGTALK_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN"

def send_dingtalk_alert(message):
    """发送钉钉机器人告警"""
    headers = {'Content-Type': 'application/json'}
    data = {
        "msgtype": "text",
        "text": {
            "content": f"