怎样实现容器资源超限告警脚本

wen 实用脚本 2026-06-26 250

本文目录导读：

怎样实现容器资源超限告警脚本

方案一：基于 Docker 的简单监控脚本
方案二：可配置的告警脚本 (支持多种告警方式)
方案三：高级监控脚本 (含历史数据和趋势分析)
配置文件示例 (config.json)
使用方法

我来为你提供几种实现容器资源超限告警的脚本方案。

基于 Docker 的简单监控脚本

#!/bin/bash
# docker_resource_monitor.sh
# 配置参数
CPU_THRESHOLD=80          # CPU使用率阈值(%)
MEM_THRESHOLD=80          # 内存使用率阈值(%)
DISK_THRESHOLD=80         # 磁盘使用率阈值(%)
INTERVAL=60               # 检测间隔(秒)
LOG_FILE="/var/log/docker_monitor.log"
WEBHOOK_URL="https://hooks.slack.com/services/YOUR/WEBHOOK/URL"  # 可选
# 日志函数
log() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a $LOG_FILE
}
# 发送告警 (可扩展为微信、钉钉等)
send_alert() {
    local message="$1"
    log "ALERT: $message"
    # 发送到Slack (可选)
    if [ ! -z "$WEBHOOK_URL" ]; then
        curl -s -X POST -H 'Content-type: application/json' \
            --data "{\"text\":\"$message\"}" \
            $WEBHOOK_URL > /dev/null 2>&1
    fi
    # 发送到系统日志
    logger -t docker-monitor "ALERT: $message"
}
# 主监控逻辑
monitor_containers() {
    while true; do
        # 获取所有运行中的容器
        containers=$(docker ps --format "{{.Names}}")
        for container in $containers; do
            # 获取容器资源使用情况
            stats=$(docker stats --no-stream --format \
                "{{.Name}},{{.CPUPerc}},{{.MemPerc}},{{.MemUsage}},{{.NetIO}},{{.BlockIO}}" \
                $container 2>/dev/null)
            if [ -z "$stats" ]; then
                continue
            fi
            # 解析数据
            IFS=',' read -r name cpu mem mem_usage net_io block_io <<< "$stats"
            # 清理数据 (去除%符号)
            cpu_value=${cpu%\%}
            mem_value=${mem%\%}
            # 检查CPU使用率
            if (( $(echo "$cpu_value > $CPU_THRESHOLD" | bc -l) )); then
                send_alert "Container '$name' CPU usage high: ${cpu} (threshold: ${CPU_THRESHOLD}%)"
            fi
            # 检查内存使用率
            if (( $(echo "$mem_value > $MEM_THRESHOLD" | bc -l) )); then
                send_alert "Container '$name' Memory usage high: ${mem} (threshold: ${MEM_THRESHOLD}%)"
            fi
        done
        sleep $INTERVAL
    done
}
# 启动监控
monitor_containers

可配置的告警脚本 (支持多种告警方式)

#!/bin/bash
# container_alarm_system.sh
# 配置文件
CONFIG_FILE="/etc/container_monitor/config.conf"
ALERT_SCRIPT="/usr/local/bin/send_alert.sh"
# 默认配置
DEFAULT_CPU_LIMIT=85
DEFAULT_MEM_LIMIT=85
DEFAULT_DISK_LIMIT=85
DEFAULT_INTERVAL=60
# 读取配置
load_config() {
    if [ -f "$CONFIG_FILE" ]; then
        source "$CONFIG_FILE"
    else
        CPU_LIMIT=$DEFAULT_CPU_LIMIT
        MEM_LIMIT=$DEFAULT_MEM_LIMIT
        DISK_LIMIT=$DEFAULT_DISK_LIMIT
        INTERVAL=$DEFAULT_INTERVAL
    fi
}
# 告警发送函数 (支持多种方式)
send_alert() {
    local severity=$1
    local message=$2
    local timestamp=$(date +%s)
    # 1. 写入日志
    echo "$timestamp|$severity|$message" >> /var/log/container_alerts.log
    # 2. 系统日志
    logger -t "container-alarm" -p $([[ "$severity" == "CRITICAL" ]] && echo "user.crit" || echo "user.warning") "$message"
    # 3. HTTP回调 (支持自定义webhook)
    if [ ! -z "$ALARM_WEBHOOK" ]; then
        curl -s -X POST -H "Content-Type: application/json" \
            -d "{\"severity\":\"$severity\",\"message\":\"$message\",\"timestamp\":$timestamp}" \
            "$ALARM_WEBHOOK" > /dev/null 2>&1 &
    fi
    # 4. 发送邮件 (如果配置了)
    if [ ! -z "$ALARM_EMAIL" ]; then
        echo "$message" | mail -s "[$severity] Container Resource Alert" "$ALARM_EMAIL" 2>/dev/null &
    fi
}
# 检查单个容器
check_container() {
    local container_id=$1
    local container_name=$2
    # 获取详细资源使用情况
    local stats=$(docker stats $container_id --no-stream --format \
        '{"cpu":"{{.CPUPerc}}","mem":"{{.MemPerc}}","memUsage":"{{.MemUsage}}","netIO":"{{.NetIO}}","blockIO":"{{.BlockIO}}"}' 2>/dev/null)
    if [ -z "$stats" ]; then
        return 1
    fi
    local cpu_percent=$(echo "$stats" | jq -r '.cpu' | sed 's/%//')
    local mem_percent=$(echo "$stats" | jq -r '.mem' | sed 's/%//')
    local mem_usage=$(echo "$stats" | jq -r '.memUsage')
    local alerts=""
    # CPU告警
    if (( $(echo "$cpu_percent > $CPU_LIMIT" | bc -l) )); then
        alerts="$alerts\nCPU: ${cpu_percent}% (limit: ${CPU_LIMIT}%)"
    fi
    # 内存告警
    if (( $(echo "$mem_percent > $MEM_LIMIT" | bc -l) )); then
        alerts="$alerts\nMemory: ${mem_percent}% - ${mem_usage} (limit: ${MEM_LIMIT}%)"
    fi
    # 发送告警
    if [ ! -z "$alerts" ]; then
        local alert_message="Container: $container_name ($container_id)\nAlerts:$alerts"
        send_alert "WARNING" "$alert_message"
    fi
    return 0
}
# 主监控循环
main() {
    load_config
    while true; do
        local timestamp=$(date '+%Y-%m-%d %H:%M:%S')
        # 获取所有容器
        docker ps --format "{{.ID}}|{{.Names}}" | while IFS='|' read -r id name; do
            check_container "$id" "$name"
        done
        # 休眠
        sleep $INTERVAL
    done
}
# 启动
main

高级监控脚本 (含历史数据和趋势分析)

#!/usr/bin/env python3
# container_advanced_monitor.py
import docker
import json
import time
import logging
import os
from datetime import datetime, timedelta
from collections import defaultdict
import smtplib
from email.mime.text import MIMEText
import requests
class ContainerMonitor:
    def __init__(self, config_path='/etc/container_monitor/config.json'):
        self.load_config(config_path)
        self.client = docker.from_env()
        self.setup_logging()
        self.history = defaultdict(list)
    def load_config(self, config_path):
        """加载配置"""
        default_config = {
            'cpu_threshold': 80,
            'mem_threshold': 80,
            'disk_threshold': 80,
            'check_interval': 60,
            'alert_cooldown': 300,  # 告警冷却时间
            'history_retention': 3600,  # 历史数据保留时间
            'email': {
                'enabled': False,
                'smtp_server': 'smtp.gmail.com',
                'smtp_port': 587,
                'username': '',
                'password': '',
                'from_addr': '',
                'to_addr': ''
            },
            'webhook': {
                'enabled': False,
                'url': '',
                'headers': {}
            },
            'containers': []  # 空列表表示监控所有容器
        }
        if os.path.exists(config_path):
            with open(config_path, 'r') as f:
                user_config = json.load(f)
                default_config.update(user_config)
        self.config = default_config
    def setup_logging(self):
        """设置日志"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('/var/log/container_monitor.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def get_container_stats(self, container):
        """获取容器资源统计"""
        stats = container.stats(stream=False)
        name = container.name
        # CPU计算
        cpu_delta = stats['cpu_stats']['cpu_usage']['total_usage'] - \
                    stats['precpu_stats']['cpu_usage']['total_usage']
        system_delta = stats['cpu_stats']['system_cpu_usage'] - \
                      stats['precpu_stats']['system_cpu_usage']
        cpu_percent = 0.0
        if system_delta > 0:
            cpu_percent = (cpu_delta / system_delta) * 100.0
        # 内存计算
        mem_usage = stats['memory_stats']['usage']
        mem_limit = stats['memory_stats']['limit']
        mem_percent = (mem_usage / mem_limit) * 100.0
        # 网络统计
        networks = stats['networks']
        total_rx = sum(n['rx_bytes'] for n in networks.values())
        total_tx = sum(n['tx_bytes'] for n in networks.values())
        return {
            'name': name,
            'cpu_percent': round(cpu_percent, 2),
            'mem_percent': round(mem_percent, 2),
            'mem_usage': self.format_bytes(mem_usage),
            'mem_limit': self.format_bytes(mem_limit),
            'network_rx': self.format_bytes(total_rx),
            'network_tx': self.format_bytes(total_tx),
            'timestamp': datetime.now().isoformat()
        }
    def format_bytes(self, bytes_value):
        """格式化字节"""
        for unit in ['B', 'KB', 'MB', 'GB']:
            if bytes_value < 1024.0:
                return f"{bytes_value:.2f}{unit}"
            bytes_value /= 1024.0
        return f"{bytes_value:.2f}TB"
    def check_thresholds(self, stats):
        """检查阈值"""
        alerts = []
        if stats['cpu_percent'] > self.config['cpu_threshold']:
            alerts.append({
                'type': 'CPU',
                'current': stats['cpu_percent'],
                'threshold': self.config['cpu_threshold'],
                'message': f"CPU usage {stats['cpu_percent']}% exceeds {self.config['cpu_threshold']}%"
            })
        if stats['mem_percent'] > self.config['mem_threshold']:
            alerts.append({
                'type': 'Memory',
                'current': stats['mem_percent'],
                'threshold': self.config['mem_threshold'],
                'message': f"Memory usage {stats['mem_percent']}% exceeds {self.config['mem_threshold']}%"
            })
        return alerts
    def send_alert(self, container_name, alerts):
        """发送告警"""
        subject = f"[Container Alert] {container_name}"
        body = f"Container: {container_name}\nTime: {datetime.now().isoformat()}\n\n"
        for alert in alerts:
            body += f"- {alert['message']}\n"
        # 日志
        self.logger.warning(body)
        # 邮件告警
        if self.config['email']['enabled']:
            self.send_email(subject, body)
        # Webhook告警
        if self.config['webhook']['enabled']:
            self.send_webhook(subject, body)
    def send_email(self, subject, body):
        """发送邮件"""
        try:
            msg = MIMEText(body)
            msg['Subject'] = subject
            msg['From'] = self.config['email']['from_addr']
            msg['To'] = self.config['email']['to_addr']
            with smtplib.SMTP(
                self.config['email']['smtp_server'],
                self.config['email']['smtp_port']
            ) as server:
                server.starttls()
                server.login(
                    self.config['email']['username'],
                    self.config['email']['password']
                )
                server.send_message(msg)
        except Exception as e:
            self.logger.error(f"Failed to send email: {e}")
    def send_webhook(self, subject, body):
        """发送Webhook"""
        try:
            payload = {
                'subject': subject,
                'body': body,
                'timestamp': datetime.now().isoformat()
            }
            requests.post(
                self.config['webhook']['url'],
                json=payload,
                headers=self.config['webhook']['headers'],
                timeout=10
            )
        except Exception as e:
            self.logger.error(f"Failed to send webhook: {e}")
    def run(self):
        """主运行循环"""
        self.logger.info("Container Monitor Started")
        last_alert_time = defaultdict(lambda: 0)
        while True:
            try:
                # 获取容器列表
                containers = self.client.containers.list()
                # 过滤特定容器
                if self.config['containers']:
                    containers = [
                        c for c in containers 
                        if c.name in self.config['containers']
                    ]
                for container in containers:
                    stats = self.get_container_stats(container)
                    alerts = self.check_thresholds(stats)
                    if alerts:
                        # 检查冷却时间
                        current_time = time.time()
                        if current_time - last_alert_time[container.name] > \
                           self.config['alert_cooldown']:
                            self.send_alert(container.name, alerts)
                            last_alert_time[container.name] = current_time
                    # 保存历史数据
                    self.history[container.name].append(stats)
                    # 清理过期数据
                    cutoff = datetime.now() - timedelta(
                        seconds=self.config['history_retention']
                    )
                    self.history[container.name] = [
                        h for h in self.history[container.name]
                        if datetime.fromisoformat(h['timestamp']) > cutoff
                    ]
                    # 记录正常状态
                    self.logger.info(
                        f"Container: {stats['name']} | "
                        f"CPU: {stats['cpu_percent']}% | "
                        f"Mem: {stats['mem_percent']}% | "
                        f"Net: RX={stats['network_rx']} TX={stats['network_tx']}"
                    )
                time.sleep(self.config['check_interval'])
            except KeyboardInterrupt:
                self.logger.info("Monitor stopped by user")
                break
            except Exception as e:
                self.logger.error(f"Error in monitoring loop: {e}")
                time.sleep(10)
if __name__ == "__main__":
    monitor = ContainerMonitor()
    monitor.run()

配置文件示例 (config.json)

{
  "cpu_threshold": 80,
  "mem_threshold": 80,
  "disk_threshold": 80,
  "check_interval": 60,
  "alert_cooldown": 300,
  "history_retention": 3600,
  "email": {
    "enabled": true,
    "smtp_server": "smtp.gmail.com",
    "smtp_port": 587,
    "username": "your_email@gmail.com",
    "password": "your_app_password",
    "from_addr": "your_email@gmail.com",
    "to_addr": "admin@example.com"
  },
  "webhook": {
    "enabled": true,
    "url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK",
    "headers": {
      "Content-Type": "application/json"
    }
  },
  "containers": ["nginx", "mysql", "redis"]
}

使用方法

基础脚本使用

# 赋予执行权限
chmod +x docker_resource_monitor.sh
# 后台运行
nohup ./docker_resource_monitor.sh > /dev/null 2>&1 &
# 添加到crontab (每分钟检查)
*/1 * * * * /path/to/docker_resource_monitor.sh

Python版本使用

# 安装依赖
pip3 install docker requests
# 创建配置目录
mkdir -p /etc/container_monitor
# 运行监控脚本
python3 container_advanced_monitor.py
# 作为服务运行
cat > /etc/systemd/system/container-monitor.service << EOF
[Unit]
Description=Container Resource Monitor
After=docker.service
[Service]
ExecStart=/usr/bin/python3 /path/to/container_advanced_monitor.py
Restart=always
User=root
[Install]
WantedBy=multi-user.target
EOF
systemctl enable container-monitor
systemctl start container-monitor

容器化部署

FROM python:3.9-alpine
RUN pip install docker requests
COPY container_advanced_monitor.py /app/monitor.py
COPY config.json /etc/container_monitor/config.json
CMD ["python", "/app/monitor.py"]

这些脚本可以根据你的具体需求进行调整和扩展。

本文地址： https://www.bmjup.com/post/8693.html