实用脚本能批量监控吗?

wen 实用脚本 12

本文目录导读:

实用脚本能批量监控吗?

  1. 脚本批量监控的核心能力
  2. 实用脚本示例(Python / Shell)
  3. 批量监控脚本的常见应用场景
  4. 进阶与优化:让监控更“实用”

是的,实用脚本当然可以批量监控,脚本化的批量监控正是运维和开发中提高效率的核心手段之一。

关键点在于:“批量”意味着自动化、可重复、可扩展,脚本(Shell、Python、PowerShell等)非常适合做这件事,可以代替人工一个个去检查服务器、服务或指标的重复劳动。

下面分几个层面来介绍,并给出一些通用的思路和脚本示例。

脚本批量监控的核心能力

一个好的批量监控脚本通常具备以下能力:

  1. 输入管理:从文件、数据库、API或命令行参数中读取要监控的“目标列表”(IP地址列表、URL列表、服务名称列表)。
  2. 并行执行:同时对多个目标进行检查,而不是串行等待,极大提升效率。
  3. 统一判断:对每个目标执行相同的检查逻辑(如Ping、端口、HTTP状态码、磁盘空间等)。
  4. 结果聚合:收集所有目标的结果,并输出清晰的报告(成功、失败、异常列表)。
  5. 报警集成:当发现异常时,能够触发报警(如发送邮件、钉钉/企业微信消息、写入日志等)。
  6. 灵活定制:根据需求调整检查项、阈值、超时时间等。

实用脚本示例(Python / Shell)

以下提供两个脚本框架,可以根据实际环境修改扩展。

示例 1: Python版 - 批量HTTP状态码监控(带并发)

这个脚本会从一个文本文件中读取URL列表,并发检查每个URL的HTTP状态码。

monitor_urls.py

import subprocess
import concurrent.futures
import requests
from datetime import datetime
import csv
import sys
# 配置
URL_FILE = "urls.txt"           # 每行一个URL,如 http://example.com
OUTPUT_CSV = "monitor_report.csv"
TIMEOUT = 5                     # 超时秒数
MAX_WORKERS = 10                # 并发线程数
# 通知/报警函数(示例打印到屏幕,可改为邮件、钉钉等)
def send_alert(url, status_code, error=""):
    msg = f"[ALERT] {datetime.now()} - {url} 异常! 状态码: {status_code}, 错误: {error}"
    print(msg)
    # 这里可以加入发邮件、写入日志等逻辑
def check_url(url):
    """检查单个URL的状态码"""
    url = url.strip()
    if not url:
        return None
    try:
        # 忽略SSL证书验证,实际生产需注意
        resp = requests.get(url, timeout=TIMEOUT, verify=False)
        status_code = resp.status_code
        error = ""
        # 定义成功规则:状态码<400算成功,否则报警
        if status_code >= 400:
            send_alert(url, status_code, f"HTTP {status_code}")
            return (url, status_code, "FAIL", datetime.now())
        return (url, status_code, "PASS", datetime.now())
    except requests.exceptions.RequestException as e:
        send_alert(url, "N/A", str(e))
        return (url, "N/A", "FAIL", datetime.now())
def main():
    # 读取URL列表
    with open(URL_FILE, "r") as f:
        urls = [line.strip() for line in f if line.strip()]
    print(f"开始监控 {len(urls)} 个URL...")
    results = []
    # 使用线程池并发检查
    with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        future_to_url = {executor.submit(check_url, url): url for url in urls}
        for future in concurrent.futures.as_completed(future_to_url):
            result = future.result()
            if result:
                results.append(result)
    # 写入CSV报告
    with open(OUTPUT_CSV, "w", newline="") as csvfile:
        writer = csv.writer(csvfile)
        writer.writerow(["URL", "状态码", "结果", "检查时间"])
        writer.writerows(results)
    # 汇总统计
    pass_count = sum(1 for r in results if r[2] == "PASS")
    fail_count = sum(1 for r in results if r[2] == "FAIL")
    print(f"监控完成: 成功 {pass_count}, 失败 {fail_count}, 总 {len(results)}")
    print(f"报告已保存至 {OUTPUT_CSV}")
if __name__ == "__main__":
    main()

使用方式:

  1. 创建 urls.txt,每行一个URL(如 http://www.baidu.com)。
  2. 安装依赖:pip install requests
  3. 运行:python monitor_urls.py

示例 2: 纯Shell版 - 批量Ping监控(简单、原生)

不需要安装任何额外工具,适合在Linux服务器上对一批IP进行存活检测。

monitor_ping.sh

#!/bin/bash
# 配置
IP_FILE="ip_list.txt"           # 每行一个IP地址
PING_COUNT=2                   # 每个IP ping 2次
PING_TIMEOUT=3                 # 超时秒数
ALERT_LOG="ping_alert.log"
# 检查IP是否可达
check_ip() {
    local ip=$1
    # `ping -c $PING_COUNT -W $PING_TIMEOUT $ip` 检查
    # 丢弃输出,只通过返回值判断:0表示成功,非0表示失败
    if ping -c "$PING_COUNT" -W "$PING_TIMEOUT" "$ip" &>/dev/null; then
        echo "$(date '+%Y-%m-%d %H:%M:%S') [PASS] $ip 可达"
        return 0
    else
        echo "$(date '+%Y-%m-%d %H:%M:%S') [FAIL] $ip 不可达" | tee -a "$ALERT_LOG"
        # 这里可加入发送报警命令,如 curl 调用Webhook
        # curl -X POST -d "ip=$ip 不可达" http://your-alert-api/...
        return 1
    fi
}
# 主循环:串行检查(也可以使用 & 实现简单并行)
main() {
    local pass_count=0
    local fail_count=0
    local total=0
    echo "开始批量Ping监控..."
    echo "目标文件: $IP_FILE"
    echo "------------------------"
    # 读取IP列表并检查
    while IFS= read -r ip || [[ -n "$ip" ]]; do
        # 跳过空行和注释行
        [[ -z "$ip" || "$ip" =~ ^# ]] && continue
        total=$((total + 1))
        if check_ip "$ip"; then
            pass_count=$((pass_count + 1))
        else
            fail_count=$((fail_count + 1))
        fi
    done < "$IP_FILE"
    echo "------------------------"
    echo "总检查数: $total"
    echo "成功: $pass_count"
    echo "失败: $fail_count (详情见 $ALERT_LOG)"
}
main

使用方式:

  1. 创建 ip_list.txt,每行一个IP(如 8.8.8)。
  2. 赋予脚本执行权限:chmod +x monitor_ping.sh
  3. 运行:./monitor_ping.sh

批量监控脚本的常见应用场景

监控类型 脚本检查方式 工具/命令 关键输出
主机存活 Ping ping, fping 丢包率、延迟
端口监听 TCP连接测试 nc, telnet, socket 端口是否开放
HTTP/HTTPS服务 HTTP请求 curl, requests (Python) 状态码、响应时间、内容
磁盘使用率 远程执行命令 ssh, df -h 使用率百分比超过阈值
CPU/内存占用 SSH + top / free ssh, mpstat, free 负载、占用率百分比
日志关键字监控 远程grep ssh, grep, tail 错误或异常行数
SSL证书过期 检查证书有效期 openssl, ssl_expiry 剩余天数

进阶与优化:让监控更“实用”

  1. 并行化

    • Shell:使用 &wait 在后台并发执行,或使用 xargs -P N 指定并行数。
    • Python:使用 concurrent.futures.ThreadPoolExecutor(I/O密集型)或 multiprocessing.Pool (CPU密集型)。
    • 示例(Shell版并行pings):cat ip_list.txt | xargs -P 5 -I {} sh -c 'ping -c1 -W1 {} >/dev/null 2>&1 && echo "{} OK" || echo "{} FAIL"'
  2. 配置化:将监控目标、阈值、报警方式、超时时间写入配置文件(YAML、JSON、INI),让脚本更通用。

  3. 报警集成:在发现异常时,通过API发送通知:

    • 钉钉/企业微信机器人:curl -X POST -H "Content-Type: application/json" -d '{"msgtype":"text","text":{"content":"警报: 服务器A宕机"}}' your_webhook_url
    • 邮件:使用 mailxsendmail 或 Python smtplib
    • PagerDuty/云监控API。
  4. 结果持久化:不仅打印到屏幕,还写入日志文件(日志名称_日期.log)、CSV、数据库,便于后续分析和追踪。

  5. 定时执行:使用 Crontab(Linux)或 Windows 任务计划程序 定期运行脚本。

    • 示例(每5分钟运行一次Python脚本):
      */5 * * * * /usr/bin/python3 /path/to/monitor_urls.py >> /var/log/monitor_cron.log 2>&1
  6. 健康检查 vs. 性能基线:简单的监控检查是否在线/正常;更高级的可以记录历史数据并设定动态阈值(如响应时间突然异常增加)。

实用脚本完全可以、而且非常适合做批量监控,它比商业监控软件更轻量、灵活、低成本,尤其适合:

  • 数十到数百台服务器的中小规模环境。
  • 需要快速定制检查逻辑的场景。
  • 作为更大型监控系统(Zabbix、Prometheus、Datadog)的有益补充或前期验证。

核心是 “自动化检查 + 结果汇总 + 异常报警” 三角,脚本让这个三角可以很方便地在不同场景下落地(比如上面的 Ping 监控和 HTTP 监控样例)。

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