2026年运维实战：提升十倍效率的自动化脚本分享

作者：ai | 发布时间：2026-05-22 07:00

2026年运维实战：提升十倍效率的自动化脚本分享

在2026年的IT运维领域，随着云原生架构的全面普及和微服务规模的持续膨胀，传统的人工干预模式早已无法满足海量节点与高频变更的需求。运维工程师的核心竞争力，已经从“熟练敲击命令行”转变为“编写高质量自动化代码”。在AIOps与自动化并行的当下，掌握实用的脚本能力依然是构建智能运维底座的基石。本文将分享三个在2026年依然极具实战价值的Shell与Python自动化脚本，帮助大家从繁琐的日常巡检与清理中解放出来。

一、 Shell脚本：智能磁盘巡检与日志动态清理

在分布式系统中，因为突发流量导致日志暴增而撑爆磁盘的“黑色星期五”事件屡见不鲜。下面的Shell脚本不仅能自动巡检各挂载点的磁盘使用率，还能在达到阈值时，智能清理过期日志，并通过企业微信/钉钉机器人发送告警。


#!/bin/bash
# 2026年智能磁盘巡检与清理脚本

THRESHOLD=80
LOG_DIR="/var/log/app"
ALERT_WEBHOOK="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"

# 遍历所有磁盘分区
df -h | awk 'NR>1 {print $5, $6}' | while read -r perc mount; do
    # 去除百分号并比较
    usage=${perc%\%}
    if [ "$usage" -ge "$THRESHOLD" ]; then
        # 1. 执行日志清理：删除7天前的压缩日志，清空超过500M的活动日志
        find "$LOG_DIR" -name "*.gz" -mtime +7 -exec rm -f {} \;
        for log in "$LOG_DIR"/*.log; do
            size=$(du -m "$log" | awk '{print $1}')
            if [ "$size" -gt 500 ]; then
                > "$log" # 安全清空而非删除
            fi
        done
        
        # 2. 构造告警消息
        msg="⚠️ 磁盘告警: $mount 使用率达 $perc%。已自动执行日志清理。"
        
        # 3. 发送Webhook告警
        curl -s -X POST "$ALERT_WEBHOOK" \
            -H 'Content-Type: application/json' \
            -d "{\"msgtype\": \"text\", \"text\": {\"content\": \"$msg\"}}"
    fi
done

实战解析： 该脚本的精妙之处在于“先清理，后告警”的闭环逻辑。相较于传统的只报错不处理，它能在深夜突发流量时第一时间自救，避免业务因磁盘满而宕机，极大降低了P0级故障的发生率。

二、 Python脚本：微服务API异步健康探针

2026年的微服务动辄上百个，依赖传统的同步监控脚本逐个拨测不仅效率低下，且容易因超时导致误报。利用Python的asyncio与aiohttp，我们可以实现毫秒级的并发健康检查。


import asyncio
import aiohttp
import json

# 2026年微服务集群API清单
SERVICES = {
    "user-service": "https://api.internal/v1/health",
    "payment-gateway": "https://pay.internal/status",
    "inventory-core": "https://inv.internal/ping"
}

ALERT_WEBHOOK = "https://hooks.dingtalk.com/robot/send?token=YOUR_TOKEN"

async def check_service(session, name, url):
    try:
        # 设置极短的超市时间，快速失败
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=3)) as resp:
            if resp.status == 200:
                return name, "✅ Healthy"
            return name, f"❌ Unhealthy (HTTP {resp.status})"
    except Exception as e:
        return name, f"❌ Down ({type(e).__name__})"

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [check_service(session, name, url) for name, url in SERVICES.items()]
        results = await asyncio.gather(*tasks)
        
        failures = [r for r in results if "❌" in r[1]]
        if failures:
            msg = "


    ← 返回AI专栏