海量指标告警杂乱难维护？基于时序 AI 自动校准 Prometheus 动态阈值落地实战

# 导入所需依赖库
import requests
import pandas as pd
from prophet import Prophet
import paramiko
import yaml
import time
from datetime import datetime
# 全局配置，根据实际环境修改
PROM_URL = "http://192.168.40.160:9090"
SSH_HOST = "192.168.40.160"
SSH_USER = "root"
SSH_PASS = "111111"
RULES_PATH = "/usr/local/prometheus-2.37.6.linux-amd64/rules.yml"
LOG_FILE = "ai_threshold.log"
# 日志打印函数
def log(msg):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with open(LOG_FILE, "a", encoding="utf-8") as f:
        f.write(f"[{timestamp}] {msg}\n")
    print(f"[{timestamp}] {msg}")
# 1、从Prometheus拉取24小时CPU使用率时序数据
def fetch_cpu_data():
    # PromQL计算整机CPU使用率
    query = '100 - (avg by(instance) (irate(node_cpu_seconds_total{mode="idle"}[5m])) * 100)'
    url = f"{PROM_URL}/api/v1/query_range"
    params = {
        "query": query,
        "start": datetime.now().timestamp() - 3600 * 24,
        "end": datetime.now().timestamp(),
        "step": "5m"
    }
    resp = requests.get(url, params=params)
    data = resp.json()
    if data["status"] != "success" or not data["data"]["result"]:
        raise Exception("拉取Prometheus CPU指标失败，请检查服务连通性")
    # 格式化Prophet标准输入格式ds(时间)、y(指标值)
    values = data["data"]["result"][0]["values"]
    df = pd.DataFrame(values, columns=["timestamp", "cpu"])
    df["ds"] = pd.to_datetime(df["timestamp"], unit="s")
    df["y"] = df["cpu"].astype(float)
    df = df[["ds", "y"]]
    log(f"成功获取{len(df)}条CPU历史时序数据")
    return df
# 2、Prophet模型训练，学习业务周期趋势
def train_and_forecast(df):
    # 开启日、周周期拟合，适配早晚高峰、工作日周末差异
    model = Prophet(daily_seasonality=True, weekly_seasonality=True)
    model.fit(df)
    # 预测未来24小时趋势
    future = model.make_future_dataframe(periods=24, freq='H')
    forecast = model.predict(future)
    return forecast
# 3、分时段动态阈值计算（核心逻辑）
def get_dynamic_threshold(hour):
    # 8:00-20:00业务高峰，阈值50%；其余夜间低峰阈值20%
    return 50 if 8 <= hour <= 20 else 20
# 4、SSH远程更新Prometheus告警规则文件
def update_rules(threshold):
    local_temp = "rules.yml.tmp"
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(SSH_HOST, username=SSH_USER, password=SSH_PASS)
    sftp = ssh.open_sftp()
    # 拉取原规则文件
    sftp.get(RULES_PATH, local_temp)
    with open(local_temp, "r", encoding="utf-8") as f:
        rules = yaml.safe_load(f)
    update_flag = False
    # 匹配CPU告警规则，动态替换阈值
    for group in rules["groups"]:
        for rule in group.get("rules", []):
            if rule.get("alert") == "物理节点cpu使用率":
                rule["expr"] = f"(100 - avg(irate(node_cpu_seconds_total{{mode='idle'}}[5m])) by(instance)*100) > {threshold}"
                rule["annotations"]["description"] = f"{{{{ $labels.instance }}}} CPU使用率超过动态阈值{threshold}%，当前值{{{{$value}}}}，请排查负载"
                update_flag = True
    # 回写规则文件并重启Prometheus
    if update_flag:
        with open(local_temp, "w", encoding="utf-8") as f:
            yaml.dump(rules, f, allow_unicode=True)
        sftp.put(local_temp, RULES_PATH)
        log(f"告警规则更新完成，当前动态阈值：{threshold}%")
    else:
        log("未匹配到CPU告警规则，跳过更新")
    sftp.close()
    # 远程执行重启命令
    ssh.exec_command("systemctl restart prometheus")
    ssh.close()
    log("Prometheus服务重载完成，新阈值生效")
# 5、主循环，每小时自动执行一次
def main_loop():
    log("===== AI动态阈值自动校准任务启动 =====")
    while True:
        try:
            current_hour = datetime.now().hour
            df = fetch_cpu_data()
            forecast = train_and_forecast(df)
            threshold = get_dynamic_threshold(current_hour)
            log(f"当前时段{current_hour}点，设定告警阈值{threshold}%")
            update_rules(threshold)
        except Exception as err:
            log(f"任务执行异常：{str(err)}")
        log("等待1小时后重新计算阈值\n")
        time.sleep(3600)
if __name__ == "__main__":
    main_loop()

histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m]))