如何编写磁盘空间检测脚本

wen 实用脚本 24

本文目录导读:

如何编写磁盘空间检测脚本

  1. Linux Shell 脚本(最常用)
  2. Python 脚本(更灵活)
  3. Windows PowerShell 脚本
  4. 定时执行设置
  5. Docker 容器化部署

Linux Shell 脚本(最常用)

#!/bin/bash
# 磁盘空间检测脚本
# 支持自定义阈值和告警
# 配置
WARNING_THRESHOLD=80  # 告警阈值(百分比)
CRITICAL_THRESHOLD=90 # 严重阈值(百分比)
LOG_FILE="/var/log/disk_monitor.log"
EMAIL_ALERT="admin@example.com"  # 可选
# 日志写入函数
write_log() {
    echo "$(date '+%Y-%m-%d %H:%M:%S') - $1" >> "$LOG_FILE"
}
# 发送邮件告警
send_alert() {
    local subject="磁盘空间告警 - $HOSTNAME"
    local body="$1"
    echo "$body" | mail -s "$subject" "$EMAIL_ALERT"
}
# 主检测逻辑
check_disk_space() {
    local alert_messages=""
    df -h | grep -v "^Filesystem" | while read line; do
        # 获取挂载点和使用率
        MOUNT_POINT=$(echo $line | awk '{print $6}')
        USAGE_PERCENT=$(echo $line | awk '{print $5}' | sed 's/%//')
        TOTAL=$(echo $line | awk '{print $2}')
        USED=$(echo $line | awk '{print $3}')
        AVAIL=$(echo $line | awk '{print $4}')
        # 跳过虚拟文件系统
        if [[ "$MOUNT_POINT" == "/dev"* ]] || [[ "$MOUNT_POINT" == "/sys"* ]] || [[ "$MOUNT_POINT" == "/proc"* ]]; then
            continue
        fi
        # 检测使用率
        if [ "$USAGE_PERCENT" -ge "$CRITICAL_THRESHOLD" ] 2>/dev/null; then
            message="[CRITICAL] 磁盘 $MOUNT_POINT 使用率 ${USAGE_PERCENT}% (已用: $USED, 可用: $AVAIL, 总容量: $TOTAL)"
            echo "$message"
            write_log "$message"
            alert_messages="$alert_messages\n$message"
        elif [ "$USAGE_PERCENT" -ge "$WARNING_THRESHOLD" ] 2>/dev/null; then
            message="[WARNING] 磁盘 $MOUNT_POINT 使用率 ${USAGE_PERCENT}% (已用: $USED, 可用: $AVAIL, 总容量: $TOTAL)"
            echo "$message"
            write_log "$message"
        else
            echo "[OK] 磁盘 $MOUNT_POINT 使用率 ${USAGE_PERCENT}%"
        fi
    done
    # 如果有告警,发送邮件
    if [ -n "$alert_messages" ]; then
        send_alert "磁盘空间告警详情:\n$alert_messages"
    fi
}
# 保存当前磁盘状态到文件(用于趋势分析)
save_disk_stats() {
    df -h | grep -v "^Filesystem" | awk '{print $6","$5","$3","$4","$2}' >> /tmp/disk_stats.csv
}
# 清理超过30天的日志
cleanup_old_logs() {
    find "$LOG_DIR" -name "*.log" -mtime +30 -delete
}
# 执行检测
main() {
    echo "=== 磁盘空间检测报告 - $(date) ==="
    echo "服务器: $HOSTNAME"
    echo "----------------------------------------"
    check_disk_space
    save_disk_stats
    # 每小时清理一次旧日志
    if [ $(( $(date +%H) % 6 )) -eq 0 ]; then
        cleanup_old_logs
    fi
}
main

Python 脚本(更灵活)

#!/usr/bin/env python3
"""
磁盘空间检测脚本
支持多平台、邮件告警、Webhook通知
"""
import psutil
import smtplib
import json
import requests
from datetime import datetime
from email.mime.text import MIMEText
import logging
class DiskMonitor:
    def __init__(self, config=None):
        self.config = config or {
            'warning_threshold': 80,
            'critical_threshold': 90,
            'exclude_mounts': ['/snap', '/loop'],
            'log_file': '/var/log/disk_monitor.log',
            'email': {
                'enabled': False,
                'smtp_server': 'smtp.gmail.com',
                'smtp_port': 587,
                'username': 'your-email@gmail.com',
                'password': 'your-password',
                'from_addr': 'monitor@example.com',
                'to_addrs': ['admin@example.com']
            },
            'webhook': {
                'enabled': False,
                'url': 'https://hooks.slack.com/services/xxx',
                'headers': {'Content-Type': 'application/json'}
            }
        }
        self.setup_logging()
    def setup_logging(self):
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(self.config['log_file']),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def get_disk_info(self):
        """获取磁盘信息"""
        disks = []
        for partition in psutil.disk_partitions():
            # 跳过要排除的挂载点
            if any(partition.mountpoint.startswith(exclude) 
                   for exclude in self.config['exclude_mounts']):
                continue
            try:
                usage = psutil.disk_usage(partition.mountpoint)
                disk_info = {
                    'device': partition.device,
                    'mountpoint': partition.mountpoint,
                    'fstype': partition.fstype,
                    'total': usage.total,
                    'used': usage.used,
                    'free': usage.free,
                    'percent': usage.percent,
                    'total_gb': round(usage.total / (1024**3), 2),
                    'used_gb': round(usage.used / (1024**3), 2),
                    'free_gb': round(usage.free / (1024**3), 2)
                }
                disks.append(disk_info)
            except PermissionError:
                continue
        return disks
    def check_threshold(self, disk_info):
        """检查磁盘使用阈值"""
        percent = disk_info['percent']
        mountpoint = disk_info['mountpoint']
        if percent >= self.config['critical_threshold']:
            self.logger.critical(
                f"磁盘 {mountpoint} 使用率 {percent:.1f}% "
                f"(已用: {disk_info['used_gb']}GB, "
                f"可用: {disk_info['free_gb']}GB)"
            )
            return 'CRITICAL'
        elif percent >= self.config['warning_threshold']:
            self.logger.warning(
                f"磁盘 {mountpoint} 使用率 {percent:.1f}% "
                f"(已用: {disk_info['used_gb']}GB, "
                f"可用: {disk_info['free_gb']}GB)"
            )
            return 'WARNING'
        else:
            self.logger.info(
                f"磁盘 {mountpoint} 使用率 {percent:.1f}% (正常)"
            )
            return 'OK'
    def send_email_alert(self, message):
        """发送邮件告警"""
        if not self.config['email']['enabled']:
            return
        try:
            msg = MIMEText(message)
            msg['Subject'] = f'磁盘空间告警 - {datetime.now().strftime("%Y-%m-%d %H:%M")}'
            msg['From'] = self.config['email']['from_addr']
            msg['To'] = ', '.join(self.config['email']['to_addrs'])
            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)
            self.logger.info("邮件告警已发送")
        except Exception as e:
            self.logger.error(f"发送邮件失败: {e}")
    def send_webhook_alert(self, message):
        """发送Webhook通知(例如Slack、钉钉等)"""
        if not self.config['webhook']['enabled']:
            return
        try:
            payload = {
                'text': message,
                'username': '磁盘监控',
                'icon_emoji': ':floppy_disk:'
            }
            response = requests.post(
                self.config['webhook']['url'],
                headers=self.config['webhook']['headers'],
                data=json.dumps(payload)
            )
            if response.status_code == 200:
                self.logger.info("Webhook通知已发送")
            else:
                self.logger.warning(f"Webhook发送失败: {response.status_code}")
        except Exception as e:
            self.logger.error(f"发送Webhook失败: {e}")
    def collect_metrics(self):
        """收集并返回指标数据(用于监控系统)"""
        disks = self.get_disk_info()
        metrics = []
        for disk in disks:
            metrics.append({
                'metric': 'disk_usage_percent',
                'value': disk['percent'],
                'tags': {
                    'mountpoint': disk['mountpoint'],
                    'device': disk['device']
                }
            })
            metrics.append({
                'metric': 'disk_free_gb',
                'value': disk['free_gb'],
                'tags': {
                    'mountpoint': disk['mountpoint']
                }
            })
        return metrics
    def run(self):
        """主运行逻辑"""
        print(f"\n=== 磁盘空间检测报告 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ===")
        print(f"主机名: {psutil.users()[0].name if psutil.users() else 'Unknown'}")
        print("-" * 50)
        alerts = []
        disks = self.get_disk_info()
        for disk in disks:
            status = self.check_threshold(disk)
            # 格式化输出
            print(f"{'[' + status + ']':<10} {disk['mountpoint']:<20} "
                  f"使用率: {disk['percent']:.1f}% "
                  f"(已用: {disk['used_gb']:.1f}GB / {disk['total_gb']:.1f}GB)")
            if status != 'OK':
                alerts.append(
                    f"[{status}] {disk['mountpoint']}: "
                    f"{disk['percent']:.1f}% "
                    f"({disk['used_gb']:.1f}GB/{disk['total_gb']:.1f}GB)"
                )
        # 发送告警
        if alerts:
            alert_message = "\n".join(alerts)
            self.send_email_alert(alert_message)
            self.send_webhook_alert(alert_message)
        # 返回兼容监控系统的数据
        return self.collect_metrics()
# 使用示例
if __name__ == "__main__":
    # 创建监控器实例
    monitor = DiskMonitor()
    # 运行检测
    metrics = monitor.run()
    # 如果需要,可以将指标输出为JSON
    print("\n指标数据:")
    print(json.dumps(metrics, indent=2))

Windows PowerShell 脚本

# 磁盘空间检测脚本 (PowerShell)
# 配置
$WarningThreshold = 80
$CriticalThreshold = 90
$LogPath = "C:\Logs\disk_monitor.log"
$AlertEmail = @{
    To = "admin@company.com"
    From = "monitor@company.com"
    SmtpServer = "smtp.company.com"
}
function Write-Log {
    param([string]$Message, [string]$Level="INFO")
    $Timestamp = Get-Date -Format "yyyy-MM-dd HH:mm:ss"
    $LogMessage = "$Timestamp [$Level] $Message"
    # 写入文件
    Add-Content -Path $LogPath -Value $LogMessage
    # 控制台输出
    Write-Host $LogMessage
}
function Get-DiskInfo {
    $Disks = Get-WmiObject Win32_LogicalDisk -Filter "DriveType=3" |
             Where-Object { $_.DeviceID -notlike "A:*" }
    $DiskInfo = @()
    foreach ($Disk in $Disks) {
        $TotalGB = [math]::Round($Disk.Size / 1GB, 2)
        $FreeGB = [math]::Round($Disk.FreeSpace / 1GB, 2)
        $UsedGB = $TotalGB - $FreeGB
        $UsagePercent = [math]::Round(($UsedGB / $TotalGB) * 100, 1)
        $Info = [PSCustomObject]@{
            Drive = $Disk.DeviceID
            TotalGB = $TotalGB
            UsedGB = $UsedGB
            FreeGB = $FreeGB
            UsagePercent = $UsagePercent
            Status = if ($UsagePercent -ge $CriticalThreshold) { "CRITICAL" }
                     elseif ($UsagePercent -ge $WarningThreshold) { "WARNING" }
                     else { "OK" }
        }
        $DiskInfo += $Info
    }
    return $DiskInfo
}
function Send-EmailAlert {
    param([string[]]$AlertMessages)
    $Body = @"
磁盘空间告警报告
服务器: $env:COMPUTERNAME
时间: $(Get-Date -Format "yyyy-MM-dd HH:mm:ss")
告警详情:
$($AlertMessages -join "`n")
"@
    try {
        Send-MailMessage @AlertEmail `
            -Subject "磁盘空间告警 - $env:COMPUTERNAME" `
            -Body $Body `
            -ErrorAction Stop
        Write-Log "邮件告警已发送" "INFO"
    }
    catch {
        Write-Log "发送邮件失败: $_" "ERROR"
    }
}
# 主程序
Write-Host "=== 磁盘空间检测报告 - $(Get-Date -Format 'yyyy-MM-dd HH:mm:ss') ==="
Write-Host "服务器: $env:COMPUTERNAME"
Write-Host ("-" * 50)
$Alerts = @()
$DiskInfo = Get-DiskInfo
foreach ($Disk in $DiskInfo) {
    # 格式化输出
    $StatusSymbol = if ($Disk.Status -eq "OK") { "✓" } else { "✗" }
    Write-Host "[$StatusSymbol $($Disk.Status)] Drive $($Disk.Drive): "
    Write-Host "    使用率: $($Disk.UsagePercent)% (已用: $($Disk.UsedGB)GB / $($Disk.TotalGB)GB)"
    # 记录日志
    Write-Log "Drive $($Disk.Drive): $($Disk.UsagePercent)% 使用率" $Disk.Status
    # 收集告警
    if ($Disk.Status -ne "OK") {
        $Alerts += "$($Disk.Drive): $($Disk.UsagePercent)% ($($Disk.UsedGB)GB/$($Disk.TotalGB)GB)"
    }
}
# 发送告警
if ($Alerts.Count -gt 0) {
    Send-EmailAlert -AlertMessages $Alerts
}
else {
    Write-Host "`n所有磁盘使用率正常"
}

定时执行设置

Linux Crontab

# 每5分钟检测一次
*/5 * * * * /usr/local/bin/disk_monitor.sh
# 每小时检测一次并记录
0 * * * * /usr/local/bin/disk_monitor.sh >> /var/log/disk_cron.log 2>&1

Windows 任务计划程序

# 创建定时任务
$Action = New-ScheduledTaskAction -Execute "PowerShell.exe" -Argument "-File C:\Scripts\disk_monitor.ps1"
$Trigger = New-ScheduledTaskTrigger -Daily -At "09:00" -RepetitionInterval (New-TimeSpan -Minutes 30) -RepetitionDuration ([TimeSpan]::MaxValue)
Register-ScheduledTask -TaskName "Disk Space Monitor" -Action $Action -Trigger $Trigger -RunLevel Highest

Docker 容器化部署

# Dockerfile
FROM alpine:latest
RUN apk add --no-cache bash curl
COPY disk_monitor.sh /usr/local/bin/
RUN chmod +x /usr/local/bin/disk_monitor.sh
# 配置定时任务
RUN echo "*/5 * * * * /usr/local/bin/disk_monitor.sh" > /etc/crontabs/root
CMD ["crond", "-f", "-l", "2"]

这些脚本可以根据具体需求进行调整,

  1. 添加钉钉/企业微信通知
  2. 集成到监控系统(Prometheus、Zabbix等)
  3. 自动清理临时文件
  4. 生成可视化报告
  5. 支持多服务器批量监控

需要我详细说明某个特定功能的实现吗?

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