：
-
运维复杂性增加，传统响应式运维成本高、效率低。
-
数据驱动的运维（AIOps）成为趋势，机器学习是实现智能预测的核心。
目标：
-
构建预测模型，提前发现潜在故障、性能瓶颈或资源需求。
-
实现从“被动响应”到“主动预防”的运维模式转变，提升系统稳定性与资源利用率。
运维磁盘预测
import asyncio
import aiohttp
import time
import numpy as np
from typing import Dict, List
from dataclasses import dataclass
# ================= 配置区域 =================
VM_URL = "http://192.168.100.130:8428" # VictoriaMetrics 地址
LOOKBACK_WINDOW = "5m" # 历史数据窗口，支持 s/m/h/d
PREDICT_WINDOW_HOURS = 1 # 预测未来多久 按天计算
THRESHOLD_PERCENT = 80 # 告警阈值 %
STEP = "15s" # 数据步长
# ================= 数据类 =================
@dataclass
class DiskMetric:
instance: str
device: str
mountpoint: str
total_bytes: float
history_timestamps: np.ndarray
history_values: np.ndarray
# ================= 核心类 =================
class DiskPredictor:
def __init__(self, vm_url: str):
self.vm_url = vm_url
self.query_range_url = f"{vm_url}/api/v1/query_range"
self.query_url = f"{vm_url}/api/v1/query"
async def fetch_json(self, session: aiohttp.ClientSession, url: str, params: Dict) -> Dict:
async with session.get(url, params=params) as response:
if response.status != 200:
print(f"Error fetching data: {response.status}")
return {}
return await response.json()
@staticmethod
def parse_duration(duration: str) -> int:
"""将 Prometheus 风格时间字符串转换为秒"""
unit_map = {"s": 1, "m": 60, "h": 3600, "d": 86400}
unit = duration[-1]
value = int(duration[:-1])
if unit not in unit_map:
raise ValueError(f"Unsupported duration unit: {unit}")
return value * unit_map[unit]
async def get_data(self):
"""
并行获取两组数据：
1. 磁盘总大小 (Instant Query)
2. 磁盘剩余空间历史趋势 (Range Query)
"""
now = int(time.time())
lookback_seconds = self.parse_duration(LOOKBACK_WINDOW)
start = now - lookback_seconds
q_total = 'node_filesystem_size_bytes{fstype!~"tmpfs|overlay"}'
q_free_history = 'node_filesystem_free_bytes{fstype!~"tmpfs|overlay"}'
async with aiohttp.ClientSession() as session:
task_total = self.fetch_json(session, self.query_url, {'query': q_total})
task_history = self.fetch_json(session, self.query_range_url, {
'query': q_free_history,
'start': start,
'end': now,
'step': STEP
})
return await asyncio.gather(task_total, task_history)
def parse_and_merge(self, res_total: Dict, res_history: Dict) -> List[DiskMetric]:
"""将总大小和历史趋势数据合并"""
disks = {}
# 解析总大小
if 'data' in res_total and 'result' in res_total['data']:
for item in res_total['data']['result']:
m = item['metric']
key = f"{m.get('instance')}|{m.get('device')}|{m.get('mountpoint')}"
disks[key] = {'meta': m, 'total': float(item['value'][1]), 'history': None}
# 解析历史数据并合并
metrics_list = []
if 'data' in res_history and 'result' in res_history['data']:
for item in res_history['data']['result']:
m = item['metric']
key = f"{m.get('instance')}|{m.get('device')}|{m.get('mountpoint')}"
if key in disks and disks[key]['total'] > 0:
values = np.array(item['values'], dtype=float)
metric_obj = DiskMetric(
instance=m.get('instance'),
device=m.get('device'),
mountpoint=m.get('mountpoint'),
total_bytes=disks[key]['total'],
history_timestamps=values[:, 0],
history_values=values[:, 1]
)
metrics_list.append(metric_obj)
return metrics_list
def predict_and_alert(self, metrics: List[DiskMetric]):
"""线性回归预测未来磁盘使用率，并打印告警"""
print(f"Analyzed {len(metrics)} disk devices across infrastructure.")
print(f"{'Instance':<30} {'Mount':<10} {'Curr Usage%':<12} {'Pred Usage%':<15} {'Status'}")
print("-" * 85)
for m in metrics:
if len(m.history_timestamps) < 2:
continue # 数据点不足，无法预测
x = m.history_timestamps - m.history_timestamps[0] # 时间归一化
y = m.history_values # free bytes
# 线性回归
slope, intercept = np.polyfit(x, y, deg=1)
# 预测未来
current_offset = x[-1]
future_offset = current_offset + (PREDICT_WINDOW_HOURS * 3600)
predicted_free_bytes = max(0, slope * future_offset + intercept)
current_free = y[-1]
curr_usage_pct = (1 - (current_free / m.total_bytes)) * 100
pred_usage_pct = (1 - (predicted_free_bytes / m.total_bytes)) * 100
status = "🔴 ALERT" if pred_usage_pct >= THRESHOLD_PERCENT else "🟢 OK"
print(f"{m.instance:<30} {m.mountpoint:<10} {curr_usage_pct:>6.2f}% {pred_usage_pct:>6.2f}% {status}")
# ================= 执行逻辑 =================
async def main():
predictor = DiskPredictor(vm_url=VM_URL)
res_total, res_history = await predictor.get_data()
if not res_total or not res_history:
print("Failed to fetch data from VictoriaMetrics")
return
metrics = predictor.parse_and_merge(res_total, res_history)
predictor.predict_and_alert(metrics)
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
pass
结果
Analyzed 6 disk devices across infrastructure.
Instance Mount Curr Usage% Pred Usage% Status
-------------------------------------------------------------------------------------
192.168.100.130:9100 / 37.13% 37.16% 🟢 OK
192.168.100.131:9100 / 76.60% 100.00% 🔴 ALERT
192.168.100.132:9100 / 33.13% 33.13% 🟢 OK
192.168.100.130:9100 /boot 9.86% 9.86% 🟢 OK
192.168.100.131:9100 /boot 9.86% 9.86% 🟢 OK
192.168.100.132:9100 /boot 9.86% 9.86% 🟢 OK
文章评论