脚本概述
- 功能描述：
通过拉取 node_exporter 的 /metrics 指标，周期性监控 Linux 服务器的 CPU 使用率、内存使用率、磁盘分区使用率，并记录日志。当 CPU 或内存超过设定阈值时自动触发邮件告警。同时采集并记录当前系统中资源占用最高的进程列表（用于定位问题），支持网络请求/告警发送自动重试与失败告警。
- 适用环境：
- 执行权限：
- 普通用户即可运行脚本；
- 如需写入默认日志目录 /var/log/ops，需具备相应写权限（通常为 root 或调整目录权限到运行用户）。
完整代码
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Linux系统监控脚本（基于 node_exporter）
- 指标：CPU、内存、磁盘、进程Top
- 告警：CPU/内存阈值触发邮件通知
- 日志：RotatingFileHandler 至指定目录
- 安全：不包含破坏性操作，不硬编码敏感信息
作者：原创
"""
import argparse
import os
import sys
import time
import ssl
import smtplib
import socket
import signal
import subprocess
import re
import threading
from email.mime.text import MIMEText
from email.utils import formatdate, make_msgid
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
from logging.handlers import RotatingFileHandler
import logging
STOP_EVENT = threading.Event()
DEFAULT_EXCLUDE_FSTYPE = [
"tmpfs", "devtmpfs", "overlay", "squashfs", "aufs", "ramfs", "iso9660",
"proc", "sysfs", "cgroup", "cgroup2", "pstore", "debugfs", "bpf", "nsfs"
]
DEFAULT_EXCLUDE_MOUNTPOINT_RE = r'^/(proc|sys|dev|run)($|/)'
def setup_logging(log_dir: str, level: str = "INFO") -> logging.Logger:
os.makedirs(log_dir, exist_ok=True)
log_file = os.path.join(log_dir, "ops_monitor.log")
logger = logging.getLogger("ops_monitor")
logger.setLevel(getattr(logging, level.upper(), logging.INFO))
fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
handler = RotatingFileHandler(log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8")
handler.setFormatter(fmt)
logger.addHandler(handler)
# 同时输出到控制台（可选）
console = logging.StreamHandler(sys.stdout)
console.setFormatter(fmt)
console.setLevel(getattr(logging, level.upper(), logging.INFO))
logger.addHandler(console)
return logger
def graceful_exit(signum, frame):
STOP_EVENT.set()
def http_get_with_retries(url: str, retries: int, timeout: int, logger: logging.Logger) -> str:
last_err = None
for attempt in range(1, retries + 1):
try:
req = Request(url, headers={"Accept": "text/plain"})
with urlopen(req, timeout=timeout) as resp:
charset = resp.headers.get_content_charset() or "utf-8"
data = resp.read().decode(charset, errors="replace")
return data
except (HTTPError, URLError, TimeoutError) as e:
last_err = e
logger.warning("HTTP GET failed (attempt %d/%d): %s", attempt, retries, str(e))
time.sleep(min(2 * attempt, 10))
raise RuntimeError(f"Failed to fetch metrics from {url}: {last_err}")
def parse_prometheus_text(text: str):
"""
解析 Prometheus 文本指标为 dict[name] = list of (labels_dict, value_float)
仅处理简单的 name{labels} value 或 name value 格式。
"""
metrics = {}
for line in text.splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
# name{l1="v1",l2="v2"} value 或 name value
m = re.match(r'^([a-zA-Z_:][a-zA-Z0-9_:]*)(\{[^}]*\})?\s+(-?\d+\.?\d*(?:e[+-]?\d+)?|NaN|Inf|-Inf)\s*$', line)
if not m:
# 跳过无法解析的行
continue
name, label_str, value_str = m.groups()
labels = {}
if label_str:
# 去掉花括号
label_content = label_str[1:-1]
# 简单解析 k="v"，允许有逗号分隔
# 注意：不处理复杂转义场景，仅处理常见格式
parts = re.findall(r'([a-zA-Z_][a-zA-Z0-9_]*)="((?:\\.|[^"\\])*)"', label_content)
for k, v in parts:
labels[k] = bytes(v, "utf-8").decode("unicode_escape")
try:
if value_str in ("NaN", "Inf", "-Inf"):
value = float("nan") if value_str == "NaN" else float(value_str)
else:
value = float(value_str)
except Exception:
continue
metrics.setdefault(name, []).append((labels, value))
return metrics
def compute_cpu_usage(prev_metrics: dict, cur_metrics: dict, logger: logging.Logger) -> float:
"""
依据 node_cpu_seconds_total 两次快照计算总CPU使用率（百分比）
使用：usage = 1 - (delta_idle / delta_total)
"""
def sum_metric(mdict, mode_filter=None):
total = 0.0
for labels, val in mdict.get("node_cpu_seconds_total", []):
if "cpu" in labels and labels.get("cpu") == "cpu":
# 某些系统可能有聚合CPU标签 'cpu="cpu"', 忽略，以各核心求和
continue
if mode_filter is None or labels.get("mode") == mode_filter:
total += val
return total
idle_prev = sum_metric(prev_metrics, mode_filter="idle")
idle_cur = sum_metric(cur_metrics, mode_filter="idle")
total_prev = sum_metric(prev_metrics, mode_filter=None)
total_cur = sum_metric(cur_metrics, mode_filter=None)
delta_idle = idle_cur - idle_prev
delta_total = total_cur - total_prev
if delta_total <= 0:
logger.debug("CPU delta_total <= 0, skip calculation")
return float("nan")
usage = max(0.0, min(100.0, (1.0 - (delta_idle / delta_total)) * 100.0))
return usage
def compute_memory_usage(cur_metrics: dict) -> float:
"""
使用：1 - MemAvailable / MemTotal
"""
avail = None
total = None
for labels, val in cur_metrics.get("node_memory_MemAvailable_bytes", []):
avail = val
break
for labels, val in cur_metrics.get("node_memory_MemTotal_bytes", []):
total = val
break
if avail is None or total is None or total <= 0:
return float("nan")
usage = max(0.0, min(100.0, (1.0 - (avail / total)) * 100.0))
return usage
def compute_disk_usages(cur_metrics: dict, exclude_fstypes, exclude_mount_re: str):
"""
返回每个挂载点的使用率字典：{mountpoint: (usage_percent, device, fstype)}
使用：1 - avail / size
"""
size_entries = cur_metrics.get("node_filesystem_size_bytes", [])
avail_entries = cur_metrics.get("node_filesystem_avail_bytes", [])
# 构建查找表 (device, fstype, mountpoint) -> value
def key(lbl):
return (lbl.get("device", ""), lbl.get("fstype", ""), lbl.get("mountpoint", ""))
size_map = {key(lbl): val for lbl, val in size_entries}
avail_map = {key(lbl): val for lbl, val in avail_entries}
result = {}
mp_ex_re = re.compile(exclude_mount_re) if exclude_mount_re else None
for k, size in size_map.items():
device, fstype, mount = k
if not mount or not fstype:
continue
if fstype in exclude_fstypes:
continue
if mp_ex_re and mp_ex_re.search(mount):
continue
avail = avail_map.get(k)
if avail is None or size <= 0:
continue
usage = max(0.0, min(100.0, (1.0 - (avail / size)) * 100.0))
result[mount] = (usage, device, fstype)
return result
def get_top_processes(n: int, logger: logging.Logger):
"""
通过 'ps' 获取前N个CPU占用最高的进程及其内存占用
返回列表：[{'pid':..., 'comm':..., 'cpu':..., 'mem':...}, ...]
注：此项非 node_exporter 指标，为辅助定位问题。
"""
try:
# -eo: 指定列，--sort=-%cpu 按CPU降序
cmd = ["ps", "-eo", "pid,comm,%cpu,%mem", "--sort=-%cpu"]
out = subprocess.check_output(cmd, text=True)
lines = out.strip().splitlines()
# 跳过标题
data = []
for ln in lines[1:n+1]:
parts = ln.split(None, 3)
if len(parts) < 4:
continue
pid, comm, cpu, mem = parts
data.append({
"pid": int(pid),
"comm": comm,
"cpu": float(cpu),
"mem": float(mem),
})
return data
except Exception as e:
logger.warning("Failed to get top processes: %s", e)
return []
def send_mail(
smtp_host: str,
smtp_port: int,
use_tls: bool,
mail_from: str,
mail_to: list,
subject: str,
body: str,
retries: int,
smtp_user: str = None,
smtp_pass: str = None,
logger: logging.Logger = None
):
if not mail_to:
if logger:
logger.warning("No mail recipients specified; skip sending.")
return
msg = MIMEText(body, _charset="utf-8")
msg["Subject"] = subject
msg["From"] = mail_from
msg["To"] = ", ".join(mail_to)
msg["Date"] = formatdate(localtime=True)
msg["Message-Id"] = make_msgid()
last_err = None
for attempt in range(1, retries + 1):
try:
if use_tls:
context = ssl.create_default_context()
with smtplib.SMTP(smtp_host, smtp_port, timeout=10) as server:
server.ehlo()
server.starttls(context=context)
server.ehlo()
if smtp_user and smtp_pass:
server.login(smtp_user, smtp_pass)
server.sendmail(mail_from, mail_to, msg.as_string())
else:
with smtplib.SMTP(smtp_host, smtp_port, timeout=10) as server:
server.ehlo()
if smtp_user and smtp_pass:
server.login(smtp_user, smtp_pass)
server.sendmail(mail_from, mail_to, msg.as_string())
if logger:
logger.info("Alert mail sent: %s -> %s", subject, mail_to)
return
except Exception as e:
last_err = e
if logger:
logger.warning("Send mail failed (attempt %d/%d): %s", attempt, retries, e)
time.sleep(min(2 * attempt, 10))
# 发送失败记录错误
if logger:
logger.error("Failed to send alert mail after %d attempts: %s", retries, last_err)
def build_alert_body(hostname: str, cpu_usage: float, mem_usage: float, disks_alert: dict, top_procs: list) -> str:
lines = []
lines.append(f"Host: {hostname}")
lines.append(f"CPU Usage: {cpu_usage:.2f}%")
lines.append(f"Memory Usage: {mem_usage:.2f}%")
if disks_alert:
lines.append("Disk Partitions Exceeded Threshold:")
for mp, (usage, device, fstype) in disks_alert.items():
lines.append(f" - {mp} ({device}, {fstype}): {usage:.2f}%")
if top_procs:
lines.append("Top Processes by CPU:")
for p in top_procs:
lines.append(f" PID {p['pid']} {p['comm']} - CPU {p['cpu']:.2f}% MEM {p['mem']:.2f}%")
lines.append("")
lines.append("This is an automated message.")
return "\n".join(lines)
def run_monitor(args):
logger = setup_logging(args.log_dir, args.log_level)
hostname = socket.gethostname()
logger.info("Starting ops_monitor on host: %s", hostname)
logger.info("Node exporter endpoint: %s", args.node_exporter_url)
logger.info("Monitoring interval: %ds | thresholds: CPU>%s%%, MEM>%s%%, DISK>%s%%",
args.interval, args.cpu_threshold, args.mem_threshold, args.disk_threshold)
# 初次抓取，作为 CPU 计算的前快照
try:
prev_text = http_get_with_retries(args.node_exporter_url, args.retry, args.timeout, logger)
prev_metrics = parse_prometheus_text(prev_text)
prev_time = time.time()
except Exception as e:
logger.error("Initial metrics fetch failed: %s", e)
# 不终止，等待下一轮
prev_metrics = None
prev_time = None
# 主循环
while not STOP_EVENT.is_set():
start_ts = time.time()
# 睡眠到下次周期
remaining = args.interval - (start_ts - (prev_time or start_ts))
sleep_sec = max(1.0, remaining)
STOP_EVENT.wait(sleep_sec)
if STOP_EVENT.is_set():
break
# 抓取当前指标
try:
cur_text = http_get_with_retries(args.node_exporter_url, args.retry, args.timeout, logger)
cur_metrics = parse_prometheus_text(cur_text)
cur_time = time.time()
except Exception as e:
logger.error("Metrics fetch failed: %s", e)
# 记录失败告警但不中止
if args.fail_alert_email:
send_mail(
args.smtp_host, args.smtp_port, args.smtp_tls,
args.mail_from, args.mail_to,
subject=f"[OPS ALERT] Metrics fetch failed on {hostname}",
body=f"Failed to fetch metrics from {args.node_exporter_url}: {e}",
retries=args.retry,
smtp_user=args.smtp_user, smtp_pass=args.smtp_pass,
logger=logger
)
continue
# 计算 CPU（需前后快照）
cpu_usage = float("nan")
if prev_metrics:
cpu_usage = compute_cpu_usage(prev_metrics, cur_metrics, logger)
else:
logger.debug("No previous CPU snapshot; will compute from next iteration.")
# 计算内存
mem_usage = compute_memory_usage(cur_metrics)
# 计算磁盘
disk_usages = compute_disk_usages(
cur_metrics,
exclude_fstypes=args.exclude_fstype,
exclude_mount_re=args.exclude_mount_re
)
disks_exceeded = {mp: info for mp, info in disk_usages.items() if info[0] >= args.disk_threshold}
# 获取Top进程（仅记录，不触发阈值告警）
top_procs = get_top_processes(args.process_top, logger)
# 日志记录
cpu_str = "N/A" if cpu_usage != cpu_usage else f"{cpu_usage:.2f}%"
mem_str = "N/A" if mem_usage != mem_usage else f"{mem_usage:.2f}%"
logger.info("CPU=%s MEM=%s", cpu_str, mem_str)
for mp, (usage, device, fstype) in disk_usages.items():
logger.info("DISK %s (%s,%s)=%.2f%%", mp, device, fstype, usage)
if top_procs:
for p in top_procs:
logger.info("TOP PID=%d COMM=%s CPU=%.2f%% MEM=%.2f%%",
p["pid"], p["comm"], p["cpu"], p["mem"])
# 告警判断（CPU/内存阈值，磁盘阈值独立）
should_alert = False
alert_reasons = []
if cpu_usage == cpu_usage and cpu_usage >= args.cpu_threshold:
should_alert = True
alert_reasons.append(f"CPU>{args.cpu_threshold}% (current {cpu_usage:.2f}%)")
if mem_usage == mem_usage and mem_usage >= args.mem_threshold:
should_alert = True
alert_reasons.append(f"MEM>{args.mem_threshold}% (current {mem_usage:.2f}%)")
if disks_exceeded:
should_alert = True
for mp, (usage, _, _) in disks_exceeded.items():
alert_reasons.append(f"DISK {mp}>{args.disk_threshold}% (current {usage:.2f}%)")
if should_alert and args.mail_to:
subject = f"[OPS ALERT] {hostname}: " + "; ".join(alert_reasons)
body = build_alert_body(hostname, cpu_usage, mem_usage, disks_exceeded, top_procs)
send_mail(
args.smtp_host, args.smtp_port, args.smtp_tls,
args.mail_from, args.mail_to, subject, body,
retries=args.retry,
smtp_user=args.smtp_user, smtp_pass=args.smtp_pass,
logger=logger
)
# 更新前快照用于下一轮CPU计算
prev_metrics = cur_metrics
prev_time = cur_time
def parse_args():
parser = argparse.ArgumentParser(description="Linux 系统监控脚本（基于 node_exporter）")
parser.add_argument("--node-exporter-url", default="http://127.0.0.1:9100/metrics",
help="node_exporter /metrics 地址")
parser.add_argument("--interval", type=int, default=60, help="监控间隔（秒）")
parser.add_argument("--cpu-threshold", type=float, default=80.0, help="CPU使用率阈值（百分比）")
parser.add_argument("--mem-threshold", type=float, default=75.0, help="内存使用率阈值（百分比）")
parser.add_argument("--disk-threshold", type=float, default=85.0, help="磁盘分区使用率阈值（百分比）")
parser.add_argument("--process-top", type=int, default=5, help="记录前N个CPU最高的进程数量")
parser.add_argument("--smtp-host", default=os.environ.get("OPS_SMTP_HOST", "smtp.example.com"),
help="SMTP服务器地址（默认读取环境变量 OPS_SMTP_HOST）")
parser.add_argument("--smtp-port", type=int, default=int(os.environ.get("OPS_SMTP_PORT", "587")),
help="SMTP服务器端口（默认读取环境变量 OPS_SMTP_PORT 或 587）")
parser.add_argument("--smtp-tls", action="store_true", default=True,
help="使用STARTTLS（默认启用）")
parser.add_argument("--smtp-user", default=os.environ.get("OPS_SMTP_USER"),
help="SMTP用户名（可使用环境变量 OPS_SMTP_USER）")
parser.add_argument("--smtp-pass", default=os.environ.get("OPS_SMTP_PASS"),
help="SMTP密码（可使用环境变量 OPS_SMTP_PASS）")
parser.add_argument("--mail-from", default=None,
help="告警发件人（默认 ops-monitor@<hostname>）")
parser.add_argument("--mail-to", nargs="*", default=[],
help="告警收件人列表（空则不发送邮件）")
parser.add_argument("--retry", type=int, default=3, help="自动重试次数（网络/邮件）")
parser.add_argument("--timeout", type=int, default=10, help="HTTP请求超时（秒）")
parser.add_argument("--log-dir", default="/var/log/ops", help="日志目录")
parser.add_argument("--log-level", default="INFO", help="日志级别（DEBUG/INFO/WARN/ERROR）")
parser.add_argument("--exclude-fstype", nargs="*", default=DEFAULT_EXCLUDE_FSTYPE,
help="磁盘过滤的文件系统类型列表（不统计）")
parser.add_argument("--exclude-mount-re", dest="exclude_mount_re", default=DEFAULT_EXCLUDE_MOUNTPOINT_RE,
help="磁盘过滤的挂载点正则（不统计）")
parser.add_argument("--fail-alert-email", action="store_true", default=False,
help="抓取失败时发送故障告警邮件")
args = parser.parse_args()
if not args.mail_from:
args.mail_from = f"ops-monitor@{socket.gethostname()}"
return args
def main():
# 信号处理（优雅退出）
signal.signal(signal.SIGINT, graceful_exit)
signal.signal(signal.SIGTERM, graceful_exit)
args = parse_args()
try:
run_monitor(args)
except Exception as e:
# 顶层兜底日志
logger = setup_logging(args.log_dir, args.log_level)
logger.exception("Monitor terminated with error: %s", e)
sys.exit(1)
if __name__ == "__main__":
main()
参数说明
- --node-exporter-url
node_exporter 的 /metrics 地址，默认 http://127.0.0.1:9100/metrics
- --interval
监控采集间隔，单位秒。默认 60
- --cpu-threshold
CPU 使用率告警阈值（百分比），默认 80.0
- --mem-threshold
内存使用率告警阈值（百分比），默认 75.0
- --disk-threshold
磁盘分区使用率告警阈值（百分比），默认 85.0
- --process-top
记录前 N 个 CPU 占用最高的进程数量，默认 5（仅日志记录，不触发告警）
- --smtp-host
SMTP 服务器地址，默认读取环境变量 OPS_SMTP_HOST，否则使用 smtp.example.com
- --smtp-port
SMTP 服务器端口，默认读取环境变量 OPS_SMTP_PORT，否则 587
- --smtp-tls
是否使用 STARTTLS，默认启用
- --smtp-user
SMTP 用户名，默认读取环境变量 OPS_SMTP_USER（不建议通过参数传递敏感信息）
- --smtp-pass
SMTP 密码，默认读取环境变量 OPS_SMTP_PASS（不建议通过参数传递敏感信息）
- --mail-from
告警邮件发件人地址，默认 ops-monitor@
- --mail-to
告警邮件收件人列表（空列表则不发送邮件）
- --retry
网络请求和邮件发送的自动重试次数，默认 3
- --timeout
HTTP 请求超时时间（秒），默认 10
- --log-dir
日志目录，默认 /var/log/ops
- --log-level
日志级别（DEBUG/INFO/WARN/ERROR），默认 INFO
- --exclude-fstype
不统计的文件系统类型列表（用于过滤临时/伪文件系统），默认包含 tmpfs/devtmpfs/overlay 等
- --exclude-mount-re
不统计的挂载点正则表达式，默认过滤 /proc /sys /dev /run 等
- --fail-alert-email
抓取 node_exporter 指标失败时是否发送故障告警邮件，默认 False
使用示例
- 按默认配置运行（60秒间隔，阈值：CPU>80%、内存>75%，磁盘>85%），发送告警到指定邮箱：
python3 ops_monitor.py --mail-to ops-alerts@example.com
- 指定 node_exporter 地址与自定义日志目录，启用抓取失败告警：
python3 ops_monitor.py \
--node-exporter-url http://127.0.0.1:9100/metrics \
--log-dir /var/log/ops \
--fail-alert-email \
--mail-to ops-alerts@example.com ops-backup@example.com
- 自定义阈值与SMTP参数（通过环境变量传入凭据）：
export OPS_SMTP_HOST=smtp.example.com
export OPS_SMTP_PORT=587
export OPS_SMTP_USER=monitor_user
export OPS_SMTP_PASS='your_app_password'
python3 ops_monitor.py \
--cpu-threshold 85 --mem-threshold 80 --disk-threshold 90 \
--mail-to ops-alerts@example.com
注意事项
- 执行前的准备工作
- 确保 node_exporter 已在本机运行且可访问（默认端口 9100）。可检查：curl http://127.0.0.1:9100/metrics
- 保障日志目录（默认 /var/log/ops）存在且对运行用户可写：例如 mkdir -p /var/log/ops && chown : /var/log/ops
- 若需邮件告警，配置 SMTP（主机、端口、TLS、账号与密码）；建议通过环境变量传递凭据，避免硬编码。
- 可能的风险和防范措施
- 邮件轰炸风险：在持续超阈值情况下会每个周期发送一次告警。可通过外部MTA限流，或在脚本上层（如 systemd timer）调整周期；需要更精细的抑制策略可在后续版本加入告警冷却时间与恢复通知。
- 指标解析依赖 node_exporter 的标准指标名称；若使用非标准构建或启用了额外采集器，可能导致解析失败或数据偏差。
- 进程信息使用 ps 命令采集，不依赖 node_exporter；若要求严格仅来自 node_exporter，请关闭该功能或忽略该日志项。
- 性能影响和资源消耗说明
- 每个周期进行一次 HTTP 拉取与文本解析，计算量较小；默认间隔60秒对系统资源影响极低。
- 日志采用大小滚动，默认最大10MB*5个文件；长期运行磁盘占用可控。
- 进程列表采集使用 ps，开销低；如在超大进程数环境中，可将 --process-top 调小或禁用（设为0则不采集）。
