AIOps 实战：智能运维在 AI 应用中的实践
前言
随着 AI 应用规模的扩大，传统的运维方式已经无法满足需求。AIOps 利用 AI 技术来提升运维效率，实现故障快速发现、预测和自愈。
我在多个项目中实践过 AIOps，今天分享一些实用经验。
监控指标体系
核心指标采集
import time
import psutil
import threading
from collections import defaultdict
class MetricsCollector:
"""指标采集器"""
def __init__(self):
self.metrics = defaultdict(list)
self.running = False
self.thread = None
def start(self):
"""启动采集"""
self.running = True
self.thread = threading.Thread(target=self._collect_loop)
self.thread.start()
def stop(self):
"""停止采集"""
self.running = False
if self.thread:
self.thread.join()
def _collect_loop(self):
"""采集循环"""
while self.running:
self._collect_system_metrics()
self._collect_ai_metrics()
time.sleep(30)
def _collect_system_metrics(self):
"""采集系统指标"""
cpu_percent = psutil.cpu_percent()
memory = psutil.virtual_memory()
disk = psutil.disk_usage('/')
self.metrics['cpu_percent'].append((time.time(), cpu_percent))
self.metrics['memory_percent'].append((time.time(), memory.percent))
self.metrics['disk_percent'].append((time.time(), disk.percent))
def _collect_ai_metrics(self):
"""采集 AI 相关指标"""
# 这里根据实际应用实现
pass
def get_latest_metrics(self):
"""获取最新指标"""
return {
k: v[-1] if v else None
for k, v in self.metrics.items()
}
指标分析与告警
import numpy as np
from scipy import stats
class AnomalyDetector:
"""异常检测器"""
def __init__(self, window_size=60):
self.window_size = window_size
def detect_zscore(self, metric_name, values, threshold=3):
"""Z-score 异常检测"""
if len(values) < 10:
return False, 0
recent_values = values[-self.window_size:]
mean = np.mean([v[1] for v in recent_values])
std = np.std([v[1] for v in recent_values])
if std == 0:
return False, 0
z_score = abs((values[-1][1] – mean) / std)
return z_score > threshold, z_score
def detect_trend(self, metric_name, values, window=30):
"""趋势检测"""
if len(values) < window:
return None
recent = values[-window:]
x = list(range(window))
y = [v[1] for v in recent]
slope, _, _, p_value, _ = stats.linregress(x, y)
trend = "increasing" if slope > 0 else "decreasing" if slope < 0 else "stable"
return {
"trend": trend,
"slope": slope,
"p_value": p_value
}
class AlertManager:
"""告警管理器"""
def __init__(self):
self.detector = AnomalyDetector()
self.alert_history = []
self.notifiers = []
def add_notifier(self, notifier):
"""添加通知渠道"""
self.notifiers.append(notifier)
def check_and_alert(self, metrics):
"""检查并告警"""
for metric_name, values in metrics.items():
if len(values) < 5:
continue
is_anomaly, z_score = self.detector.detect_zscore(metric_name, values)
if is_anomaly:
alert = {
"type": "anomaly",
"metric": metric_name,
"z_score": z_score,
"value": values[-1][1],
"timestamp": time.time()
}
self.alert_history.append(alert)
self._notify(alert)
def _notify(self, alert):
"""发送通知"""
for notifier in self.notifiers:
try:
notifier.send(alert)
except Exception as e:
print(f"Notification failed: {e}")
日志分析
日志解析与聚合
import re
from collections import Counter
import json
class LogParser:
"""日志解析器"""
def __init__(self):
self.patterns = {
"error": re.compile(r'ERROR|Error|error|Exception|Traceback'),
"warning": re.compile(r'WARN|Warning|warning'),
"performance": re.compile(r'took\\s+(\\d+)ms|latency.*(\\d+)ms')
}
def parse_line(self, line):
"""解析单行日志"""
result = {
"original": line,
"level": "info",
"tags": []
}
if self.patterns["error"].search(line):
result["level"] = "error"
result["tags"].append("error")
elif self.patterns["warning"].search(line):
result["level"] = "warning"
result["tags"].append("warning")
perf_match = self.patterns["performance"].search(line)
if perf_match:
result["tags"].append("performance")
result["latency"] = int(perf_match.group(1) or perf_match.group(2) or 0)
return result
def aggregate_logs(self, log_lines):
"""聚合分析日志"""
parsed = [self.parse_line(line) for line in log_lines]
levels = Counter(p["level"] for p in parsed)
errors = [p for p in parsed if p["level"] == "error"]
latencies = [p["latency"] for p in parsed if "latency" in p]
return {
"total": len(parsed),
"levels": dict(levels),
"errors": errors,
"avg_latency": np.mean(latencies) if latencies else 0,
"p95_latency": np.percentile(latencies, 95) if latencies else 0
}
智能日志分析
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import KMeans
class LogAnalyzer:
"""智能日志分析器"""
def __init__(self):
self.vectorizer = TfidfVectorizer(max_features=1000)
self.clusterer = KMeans(n_clusters=10)
self.is_trained = False
def cluster_errors(self, error_logs):
"""错误日志聚类"""
if not error_logs:
return []
texts = [e["original"] for e in error_logs]
if len(texts) < 10:
return [{"cluster": 0, "logs": error_logs}]
tfidf_matrix = self.vectorizer.fit_transform(texts)
if not self.is_trained:
self.clusterer.fit(tfidf_matrix)
self.is_trained = True
clusters = self.clusterer.predict(tfidf_matrix)
clusters_dict = defaultdict(list)
for i, cluster in enumerate(clusters):
clusters_dict[cluster].append(error_logs[i])
return [
{"cluster": k, "logs": v, "size": len(v)}
for k, v in clusters_dict.items()
]
自动修复
自愈能力实现
class SelfHealingSystem:
"""自愈系统"""
def __init__(self):
self.healing_strategies = {
"high_cpu": self._handle_high_cpu,
"high_memory": self._handle_high_memory,
"service_down": self._handle_service_down
}
def detect_issue(self, metrics, logs):
"""检测问题"""
issues = []
if metrics.get("cpu_percent", (0, 0))[1] > 80:
issues.append("high_cpu")
if metrics.get("memory_percent", (0, 0))[1] > 90:
issues.append("high_memory")
return issues
def heal(self, issue):
"""执行自愈"""
if issue in self.healing_strategies:
return self.healing_strategies[issue]()
return {"status": "no_strategy", "issue": issue}
def _handle_high_cpu(self):
"""处理高 CPU"""
return {
"action": "scale_up",
"status": "in_progress"
}
def _handle_high_memory(self):
"""处理高内存"""
return {
"action": "gc_and_clear_cache",
"status": "in_progress"
}
def _handle_service_down(self):
"""处理服务宕机"""
return {
"action": "restart_service",
"status": "in_progress"
}
自动化运维工作流
class AIOpsWorkflow:
"""AIOps 工作流"""
def __init__(self):
self.collector = MetricsCollector()
self.alert_manager = AlertManager()
self.healing_system = SelfHealingSystem()
self.log_parser = LogParser()
def run(self, log_lines):
"""运行工作流"""
# 1. 采集指标
metrics = self.collector.get_latest_metrics()
# 2. 分析日志
log_analysis = self.log_parser.aggregate_logs(log_lines)
# 3. 检测问题
issues = self.healing_system.detect_issue(
self.collector.metrics,
log_analysis["errors"]
)
# 4. 检查告警
self.alert_manager.check_and_alert(self.collector.metrics)
# 5. 尝试自愈
results = []
for issue in issues:
result = self.healing_system.heal(issue)
results.append(result)
return {
"metrics": metrics,
"log_analysis": log_analysis,
"issues": issues,
"healing_results": results
}
总结
AIOps 实战要点：
实践建议：
- 从基础监控开始
- 逐步引入智能分析
- 先人工验证，再自动执行
- 重视反馈和持续改进
