传统运维依托人工巡检、固定阈值告警处理业务故障,随着云原生、微服务架构普及,服务器、容器、中间件、业务链路指标呈指数级增长,海量监控数据带来告警风暴、根因定位慢、故障预测滞后等痛点。AIOps融合大数据、机器学习、时序预测算法,实现告警收敛、异常检测、故障根因分析、容量智能预测全流程自动化运维,降低运维人力成本,提升系统稳定性。本文围绕企业级AIOps平台架构、核心算法、实战代码落地展开说明。
整体分为五层,自上而下依次为:
核心技术栈:Python3.9、Scikit-learn、Prometheus Client、FastAPI、InfluxDB2、Pandas。
采用3σ离群值检测算法实现实时指标异常识别,适用于CPU、内存、接口耗时等时序监控指标,代码基于Python实现,可直接集成至AIOps算法引擎。
import pandas as pd
import numpy as np
from prometheus_client import CollectorRegistry, Gauge, push_to_gateway
import time
# 3σ时序异常检测核心函数
def ts_anomaly_detect(metric_data: list, threshold=3):
df = pd.DataFrame(metric_data, columns=["timestamp", "value"])
data_mean = df["value"].mean()
data_std = df["value"].std()
upper_bound = data_mean + threshold * data_std
lower_bound = data_mean - threshold * data_std
# 筛选异常点
anomaly_points = df[(df["value"] > upper_bound) | (df["value"] < lower_bound)]
if len(anomaly_points) > 0:
return True, anomaly_points.to_dict("records")
return False, []
# 模拟采集服务器CPU时序指标
def mock_get_cpu_metrics():
metrics = []
base_cpu = 30
for t in range(60):
# 随机生成正常数据,每50个点插入异常值模拟突增
if t == 50:
cpu = 95
else:
cpu = base_cpu + np.random.randint(-8, 8)
metrics.append([int(time.time()) + t, cpu])
return metrics
# AIOps算法调度主流程
if __name__ == "__main__":
cpu_metrics = mock_get_cpu_metrics()
is_anomaly, anomaly_info = ts_anomaly_detect(cpu_metrics)
if is_anomaly:
print(f"检测到CPU指标异常:{anomaly_info}")
# 推送异常告警至Prometheus网关
registry = CollectorRegistry()
anomaly_gauge = Gauge('server_cpu_anomaly', 'CPU异常标记', registry=registry)
anomaly_gauge.set(1)
push_to_gateway('127.0.0.1:9091', job='aiops-alarm', registry=registry)
else:
print("当前CPU指标无异常")代码说明:通过统计学3σ规则识别偏离正常区间的指标,一旦检测异常自动推送告警指标至监控网关,联动AIOps告警中心生成故障工单。
海量指标会产生重复、关联告警,平台采用相似度匹配算法聚合同源告警:
AIOps通过AI算法替代传统人工运维模式,从被动响应故障转向主动预测、自动处置。本文提供的时序异常检测代码可快速接入企业现有监控体系,搭配告警收敛、故障自愈模块完整搭建轻量化AIOps平台,适配中小微企业至大型集团云原生运维场景,有效提升运维自动化水平与业务稳定性。
海量精选技术文档和实战案例持续更新,敬请关注【风骏时光少年】
https://edu.51cto.com/surl=dcqQj1