如何编写磁盘空间预警脚本

本文目录导读：

1. 基础Shell脚本版本
2. 增强版脚本（支持多个阈值）
3. Python版本（更灵活）
4. 配置定时执行
5. 安装依赖
6. 使用方法

我来为您编写一个磁盘空间预警脚本,提供多种实现方式：

基础Shell脚本版本

#!/bin/bash
# 磁盘空间预警脚本
# 设置阈值（百分比）
THRESHOLD=80
# 获取磁盘使用情况
df -h | grep -vE '^Filesystem|tmpfs|cdrom' | awk '{ print $5 " " $1 }' | while read output;
do
  # 提取使用百分比
  usage=$(echo $output | awk '{ print $1}' | cut -d'%' -f1)
  filesystem=$(echo $output | awk '{ print $2 }')
  # 检查是否超过阈值
  if [ $usage -ge $THRESHOLD ]; then
    echo "警告: 磁盘 $filesystem 使用率达到 ${usage}%，已超过阈值 ${THRESHOLD}%"
    # 可选：发送邮件告警
    # echo "磁盘 $filesystem 使用率达到 ${usage}%" | mail -s "磁盘空间告警" admin@example.com
    # 可选：记录日志
    echo "$(date): 警告 - 磁盘 $filesystem 使用率达到 ${usage}%" >> /var/log/disk_alert.log
  fi
done

增强版脚本（支持多个阈值）

#!/bin/bash
# 增强版磁盘空间预警脚本
WARNING_THRESHOLD=75
CRITICAL_THRESHOLD=85
EMERGENCY_THRESHOLD=95
# 日志文件
LOG_FILE="/var/log/disk_space_monitor.log"
# 检查日志目录是否存在
if [ ! -d "$(dirname $LOG_FILE)" ]; then
    mkdir -p "$(dirname $LOG_FILE)"
fi
# 获取磁盘信息
df -h | grep -vE '^Filesystem|tmpfs|cdrom|overlay' | while read line;
do
    # 提取必要信息
    filesystem=$(echo $line | awk '{print $1}')
    size=$(echo $line | awk '{print $2}')
    used=$(echo $line | awk '{print $3}')
    available=$(echo $line | awk '{print $4}')
    usage_percent=$(echo $line | awk '{print $5}' | sed 's/%//')
    mount_point=$(echo $line | awk '{print $6}')
    # 跳过根文件系统（如果不需要监控）
    # if [ "$mount_point" == "/" ]; then
    #     continue
    # fi
    # 判断告警级别
    if [ $usage_percent -ge $EMERGENCY_THRESHOLD ]; then
        level="紧急"
        message="EMERGENCY: 磁盘 $filesystem 使用率已达 ${usage_percent}%，挂载点: $mount_point"
    elif [ $usage_percent -ge $CRITICAL_THRESHOLD ]; then
        level="严重"
        message="CRITICAL: 磁盘 $filesystem 使用率为 ${usage_percent}%，挂载点: $mount_point"
    elif [ $usage_percent -ge $WARNING_THRESHOLD ]; then
        level="警告"
        message="WARNING: 磁盘 $filesystem 使用率为 ${usage_percent}%，挂载点: $mount_point"
    else
        continue  # 正常状态，跳过
    fi
    # 记录日志
    echo "$(date '+%Y-%m-%d %H:%M:%S') [$level] $message" >> $LOG_FILE
    # 控制台输出
    echo "$message"
    # 可选：发送通知（根据告警级别）
    # case $level in
    #     "紧急")
    #         # 发送短信或电话告警
    #         curl -X POST "https://api.example.com/alert" -d "message=$message"
    #         ;;
    #     "严重")
    #         # 发送邮件告警
    #         echo "$message" | mail -s "磁盘空间严重告警" admin@example.com
    #         ;;
    #     "警告")
    #         # 发送即时通知
    #         echo "$message" | mail -s "磁盘空间警告" devops@example.com
    #         ;;
    # esac
done

Python版本（更灵活）

#!/usr/bin/env python3
import os
import shutil
import logging
from datetime import datetime
import json
import smtplib
from email.mime.text import MIMEText
class DiskMonitor:
    def __init__(self, config_file='disk_monitor_config.json'):
        self.config = self.load_config(config_file)
        self.setup_logging()
    def load_config(self, config_file):
        """加载配置文件"""
        default_config = {
            "warning_threshold": 75,
            "critical_threshold": 85,
            "emergency_threshold": 95,
            "exclude_filesystems": ["tmpfs", "devtmpfs", "overlay"],
            "exclude_mounts": ["/boot", "/sys", "/proc"],
            "log_file": "/var/log/disk_monitor.log",
            "email": {
                "enabled": False,
                "smtp_server": "smtp.example.com",
                "smtp_port": 587,
                "sender": "monitor@example.com",
                "password": "your_password",
                "recipients": ["admin@example.com"]
            },
            "webhook": {
                "enabled": False,
                "url": "https://hooks.example.com/alert"
            }
        }
        try:
            with open(config_file, 'r') as f:
                config = json.load(f)
                return {**default_config, **config}
        except FileNotFoundError:
            return default_config
    def setup_logging(self):
        """设置日志"""
        logging.basicConfig(
            filename=self.config['log_file'],
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(__name__)
    def check_disk_usage(self):
        """检查所有磁盘使用情况"""
        alerts = []
        # 获取所有分区信息
        partitions = psutil.disk_partitions()
        for partition in partitions:
            # 跳过排除的文件系统
            if partition.fstype in self.config['exclude_filesystems']:
                continue
            # 跳过排除的挂载点
            if partition.mountpoint in self.config['exclude_mounts']:
                continue
            try:
                usage = shutil.disk_usage(partition.mountpoint)
                usage_percent = (usage.used / usage.total) * 100
                # 检查是否超过阈值
                if usage_percent >= self.config['emergency_threshold']:
                    alert = {
                        'level': 'EMERGENCY',
                        'mount': partition.mountpoint,
                        'usage': usage_percent,
                        'total': usage.total,
                        'used': usage.used,
                        'free': usage.free
                    }
                    alerts.append(alert)
                elif usage_percent >= self.config['critical_threshold']:
                    alert = {
                        'level': 'CRITICAL',
                        'mount': partition.mountpoint,
                        'usage': usage_percent,
                        'total': usage.total,
                        'used': usage.used,
                        'free': usage.free
                    }
                    alerts.append(alert)
                elif usage_percent >= self.config['warning_threshold']:
                    alert = {
                        'level': 'WARNING',
                        'mount': partition.mountpoint,
                        'usage': usage_percent,
                        'total': usage.total,
                        'used': usage.used,
                        'free': usage.free
                    }
                    alerts.append(alert)
            except PermissionError:
                self.logger.warning(f"无权访问 {partition.mountpoint}")
                continue
        return alerts
    def format_alert_message(self, alert):
        """格式化告警信息"""
        size_gb = lambda x: f"{x / (1024**3):.2f}GB"
        message = f"""
磁盘空间告警 - {alert['level']}
时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
挂载点: {alert['mount']}
使用率: {alert['usage']:.1f}%
总容量: {size_gb(alert['total'])}
已使用: {size_gb(alert['used'])}
可用空间: {size_gb(alert['free'])}
"""
        return message.strip()
    def send_email_alert(self, alert):
        """发送邮件告警"""
        if not self.config['email']['enabled']:
            return
        message = self.format_alert_message(alert)
        msg = MIMEText(message)
        msg['Subject'] = f"[{alert['level']}] 磁盘空间告警 - {alert['mount']}"
        msg['From'] = self.config['email']['sender']
        msg['To'] = ', '.join(self.config['email']['recipients'])
        try:
            server = smtplib.SMTP(
                self.config['email']['smtp_server'],
                self.config['email']['smtp_port']
            )
            server.starttls()
            server.login(
                self.config['email']['sender'],
                self.config['email']['password']
            )
            server.send_message(msg)
            server.quit()
            self.logger.info(f"邮件告警发送成功: {alert['mount']}")
        except Exception as e:
            self.logger.error(f"邮件发送失败: {str(e)}")
    def run_check(self):
        """执行检查"""
        alerts = self.check_disk_usage()
        for alert in alerts:
            # 记录日志
            if alert['level'] == 'EMERGENCY':
                self.logger.error(self.format_alert_message(alert))
            elif alert['level'] == 'CRITICAL':
                self.logger.warning(self.format_alert_message(alert))
            else:
                self.logger.info(self.format_alert_message(alert))
            # 打印到控制台
            print(self.format_alert_message(alert))
            # 发送邮件告警
            self.send_email_alert(alert)
        return alerts
if __name__ == "__main__":
    monitor = DiskMonitor()
    alerts = monitor.run_check()

配置定时执行

Crontab配置（每5分钟检查一次）

# 编辑crontab
crontab -e
# 添加以下行（每5分钟执行一次）
*/5 * * * * /path/to/disk_monitor.sh
# 或者使用Python版本
*/5 * * * * /usr/bin/python3 /path/to/disk_monitor.py >> /var/log/disk_monitor_cron.log 2>&1

Systemd Service配置

创建服务文件 /etc/systemd/system/disk-monitor.service：

[Unit]
Description=Disk Space Monitor Service
After=network.target
[Service]
Type=simple
ExecStart=/usr/bin/python3 /path/to/disk_monitor.py
Restart=always
RestartSec=60
User=root
[Install]
WantedBy=multi-user.target

创建定时器文件 /etc/systemd/system/disk-monitor.timer：

[Unit]
Description=Run disk monitor every 5 minutes
[Timer]
OnBootSec=1min
OnCalendar=*:0/5
Persistent=true
[Install]
WantedBy=timers.target

启用服务：

sudo systemctl enable disk-monitor.timer
sudo systemctl start disk-monitor.timer

安装依赖

# Python版本需要安装依赖
pip install psutil
# 如果需要发送邮件，需要配置sendmail或postfix
sudo apt-get install sendmail  # Ubuntu/Debian
sudo yum install sendmail      # CentOS/RHEL

使用方法

1. 保存脚本到适当位置
2. 添加执行权限：chmod +x disk_monitor.sh
3. 测试运行：./disk_monitor.sh
4. 配置定时任务

选择适合您需求的版本即可,脚本会自动检测磁盘使用情况并发出告警。