实战基于开源模型构建可生产落地的磁盘预测感知运维Agent

注：五层架构=系统部署分层结构；七段闭环=Agent单次运行生命周期，两个概念做口径区分。系统核心目标：采集/var/log分区磁盘使用率时序指标，基于近24h指标数据预判未来3h磁盘占用增长趋势，当预测使用率≥85%时自动执行过期日志安全清理+多渠道告警推送，低于阈值静默运行。全文从时序理论、分层架构设计、工程落地规范、SRE安全风控、多模型容错、Agent自省机制、场景拓展与行业发展趋势完整阐述落地路径，配套经过漏洞修复、生产加固、代码对齐文档的Python源码；方案经生产级容错改造，微调配置即可安全部署于生产服务器集群，并可横向复用至内存、inode、数据库日志、带宽等全资源预测自愈场景。

Plain Text(1 - node_filesystem_avail_bytes{mountpoint="/var/log",node_filesystem_size_bytes>0,device!~"tmpfs|overlay|nsfs"} / node_filesystem_size_bytes{mountpoint="/var/log",node_filesystem_size_bytes>0,device!~"tmpfs|overlay|nsfs"}) * 100

Bashpip install prometheus-api-client ollama pandas numpy prophet requests configparser

新增configparser用于外置配置文件解析。

Plain Text你是专业SRE磁盘预测智能体，严格遵守规则，禁止多余文字、解释、分析，只允许输出ACTION或NO_ACTION。规则：根据输入的过去24小时/var/log磁盘使用率百分比时序序列，预测未来3小时磁盘最大使用率。1.预测最大值≥85% → 仅输出ACTION2.预测最大值＜85% → 仅输出NO_ACTION时序数据列表：{{time_series_data}}

代码包含正则匹配，自动剔除输出首尾空格、换行字符，避免合法判定误判ERR。

TOML[BASE]PROM_URL = http://127.0.0.1:9090OLLAMA_HOST = http://127.0.0.1:11434MODEL_NAME = qwen:1.8bMOUNT_POINT = /var/logTHRESHOLD = 85CLEAR_COOL_SEC = 3600LOG_SAVE_PATH = /var/log/disk_agent.logAUDIT_LOG_PATH = /var/log/disk_agent_audit.logDING_WEBHOOK =FAIL_CLEAR_THRESHOLD = 3EXPECTED_DATA_POINTS = 288MIN_VALID_POINTS = 280ERR_THRESHOLD = 12[PROTECT]exclude=/var/log/secure,/var/log/messages,/var/log/maillog,/var/log/cron[SWITCH]agent_disabled = false

TOML[Unit]Description=Disk Predict Agent ServiceAfter=network.target prometheus.target[Service]User=rootExecStart=/usr/bin/python3 /opt/disk_agent/main.pyRestart=on-failureRestartSec=30[Install]WantedBy=multi-user.target

Pythonimport timeimport loggingimport jsonimport reimport osfrom pathlib import Pathfrom configparser import ConfigParserimport pandas as pdimport requestsfrom prometheus_api_client import PrometheusConnectimport ollamafrom prophet import Prophetfrom datetime import datetime, timedelta# 1、加载外置配置文件cfg = ConfigParser()cfg.read("agent.ini", encoding="utf-8")# 基础配置PROM_URL = cfg.get("BASE", "PROM_URL")OLLAMA_HOST = cfg.get("BASE", "OLLAMA_HOST")MODEL_NAME = cfg.get("BASE", "MODEL_NAME")MOUNT_POINT = cfg.get("BASE", "MOUNT_POINT")THRESHOLD = cfg.getint("BASE", "THRESHOLD")CLEAR_COOL_SEC = cfg.getint("BASE", "CLEAR_COOL_SEC")LOG_SAVE_PATH = cfg.get("BASE", "LOG_SAVE_PATH")AUDIT_LOG_PATH = cfg.get("BASE", "AUDIT_LOG_PATH")DING_WEBHOOK = cfg.get("BASE", "DING_WEBHOOK")FAIL_CLEAR_THRESHOLD = cfg.getint("BASE", "FAIL_CLEAR_THRESHOLD")EXPECTED_DATA_POINTS = cfg.getint("BASE", "EXPECTED_DATA_POINTS")MIN_VALID_POINTS = cfg.getint("BASE", "MIN_VALID_POINTS")ERR_THRESHOLD = cfg.getint("BASE", "ERR_THRESHOLD")# 保护路径PROTECTED_PATTERNS = cfg.get("PROTECT", "exclude").split(",")# 运行开关agent_disabled = cfg.getboolean("SWITCH", "agent_disabled")# 日志初始化logging.basicConfig(filename=LOG_SAVE_PATH, level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")audit_logger = logging.getLogger("audit")audit_handler = logging.FileHandler(AUDIT_LOG_PATH, encoding="utf-8")audit_handler.setFormatter(logging.Formatter("%(asctime)s | %(message)s"))audit_logger.addHandler(audit_handler)audit_logger.setLevel(logging.INFO)last_clear_ts = 0consecutive_fail_clear = 0# 存储历史预测&真实值用于自省误差计算history_reflect = []# 异常值剔除：3σ规则清洗时序毛刺def drop_outlier(data_list): if len(data_list) < 5: return data_list s = pd.Series(data_list) mean = s.mean() std = s.std() upper = mean + 3*std lower = mean - 3*std new_data = [x for x in data_list if lower <= x <= upper] return new_data# 1.从Prometheus拉取24h磁盘时序数据（本地时区，修复UTC时差）def get_disk_metric(): prom = PrometheusConnect(url=PROM_URL, disable_ssl=True) end = datetime.now() start = end - timedelta(hours=24) promql = ( f'(1 - node_filesystem_avail_bytes{{mountpoint="{MOUNT_POINT}",node_filesystem_size_bytes>0,device!~"tmpfs|overlay|nsfs"}} / ' f'node_filesystem_size_bytes{{mountpoint="{MOUNT_POINT}",node_filesystem_size_bytes>0,device!~"tmpfs|overlay|nsfs"}}) * 100' ) try: data = prom.custom_query_range(query=promql, start_time=start, end_time=end, step="5m") if not data or len(data) == 0 or "values" not in data[0]: logging.error("Prometheus查询无数据返回") return [] ts_list = [float(res[1]) for res in data[0]["values"] if res[1] is not None] clean_ts = drop_outlier(ts_list) return clean_ts except Exception as e: logging.error(f"Prometheus查询异常: {str(e)}") return []# 2.LLM预测+正则过滤输出空格换行def llm_predict(ts_data): prompt = """你是专业SRE磁盘预测智能体，严格遵守规则，禁止多余文字、解释、分析，只允许输出ACTION或NO_ACTION。规则：根据输入的过去24小时/var/log磁盘使用率百分比时序序列，预测未来3小时磁盘最大使用率。1.预测最大值≥85% → 仅输出ACTION2.预测最大值＜85% → 仅输出NO_ACTION时序数据列表：""" + str(ts_data) try: resp = ollama.chat(model=MODEL_NAME, messages=[{"role": "user", "content": prompt}], options={"temperature": 0}) res = resp["message"]["content"].strip() res = re.sub(r"\s+", "", res) if res in ["ACTION", "NO_ACTION"]: return res else: logging.warning(f"LLM输出非法: {res}") return "ERR" except Exception as e: logging.error(f"LLM推理异常: {str(e)}") return "ERR"# 3.Prophet兜底预测def prophet_predict(ts_data): try: df = pd.DataFrame({"y": ts_data}) df["ds"] = pd.date_range(end=datetime.now(), periods=len(ts_data), freq="5min") m = Prophet(growth="linear", yearly_seasonality=False, weekly_seasonality=False, daily_seasonality=True) m.fit(df) future = m.make_future_dataframe(periods=36, freq="5min") pred = m.predict(future) max_pred = pred["yhat"].iloc[-36:].max() return "ACTION" if max_pred >= THRESHOLD else "NO_ACTION" except Exception as e: logging.error(f"Prophet预测异常: {str(e)}") return "ERR"# 4.Python原生安全遍历清理（消除shell注入漏洞，支持log.gz/log.*）def safe_clear_log(): global last_clear_ts, consecutive_fail_clear, agent_disabled now = time.time() if now - last_clear_ts < CLEAR_COOL_SEC: logging.info("处于清理冷却期，跳过执行") return False if agent_disabled: logging.error("Agent已被熔断禁用，请修改agent.ini开关人工恢复") return False base_path = Path(MOUNT_POINT) del_exts = ("*.log", "*.log.*", "*.log.gz") del_files = [] protect_set = set([Path(p).resolve() for p in PROTECTED_PATTERNS]) # 遍历匹配文件 for ext in del_exts: for f in base_path.rglob(ext): f_res = f.resolve() # 排除保护文件 if any(str(f_res).startswith(p) for p in protect_set): continue # mtime>3天 mtime_sec = f.stat().st_mtime if now - mtime_sec > 3*24*3600: del_files.append(str(f_res)) if not del_files: logging.info("未找到符合清理条件的日志文件") return False # 审计记录 audit_logger.info(json.dumps({ "event": "pre_clear", "files": del_files, "count": len(del_files), "timestamp": datetime.now().isoformat() }, ensure_ascii=False)) deleted_count = 0 for fp in del_files: try: os.remove(fp) deleted_count += 1 except Exception as rm_e: logging.error(f"删除失败 {fp}: {rm_e}") last_clear_ts = now logging.info(f"日志清理完成，共删除 {deleted_count} 个文件") audit_logger.info(json.dumps({ "event": "post_clear", "deleted_count": deleted_count, "timestamp": datetime.now().isoformat() }, ensure_ascii=False)) consecutive_fail_clear = 0 return True# 5.钉钉告警def send_alert(msg): if not DING_WEBHOOK: return payload = {"msgtype": "text", "text": {"content": msg}} try: requests.post(DING_WEBHOOK, json=payload, timeout=5) except Exception as e: logging.error(f"告警发送失败: {e}")# 6.清理后校验磁盘空间def check_after_clear(): prom = PrometheusConnect(url=PROM_URL, disable_ssl=True) promql = f'(1 - node_filesystem_avail_bytes{{mountpoint="{MOUNT_POINT}",node_filesystem_size_bytes>0,device!~"tmpfs|overlay|nsfs"}} / node_filesystem_size_bytes{{mountpoint="{MOUNT_POINT}",node_filesystem_size_bytes>0,device!~"tmpfs|overlay|nsfs"}}) * 100' try: result = prom.custom_query(query=promql) if result and len(result) > 0 and "value" in result[0]: val = float(result[0]["value"][1]) logging.info(f"清理后磁盘使用率: {val:.2f}%") return val else: logging.error("校验查询无数据") return -1 except Exception as e: logging.error(f"校验异常: {e}") return -1# 7.自省误差计算（补齐原文档缺失的误差统计逻辑）def record_self_reflection(pred_decision, pre_pred_val, real_usage): info = { "dt": datetime.now().isoformat(), "decision": pred_decision, "pred_val": pre_pred_val, "real_val": real_usage } history_reflect.append(info) audit_logger.info(json.dumps({"event":"reflect", "data":info}, ensure_ascii=False)) # 误差超限告警 if real_usage > 0 and pre_pred_val > 0: err_rate = abs(pre_pred_val - real_usage)/pre_pred_val*100 if err_rate > ERR_THRESHOLD: send_alert(f"【模型精度告警】本次预测误差{err_rate:.2f}%，超出阈值{ERR_THRESHOLD}%，请优化Prompt或时序参数")# 8.主循环def agent_main(): global agent_disabled, consecutive_fail_clear while True: # 实时读取配置，支持不停服开关解禁 cfg.read("agent.ini", encoding="utf-8") agent_disabled = cfg.getboolean("SWITCH", "agent_disabled") if agent_disabled: logging.error("Agent已禁用，等待人工修改配置恢复") time.sleep(600) continue logging.info("===== 启动一轮Agent检测 =====") try: ts_data = get_disk_metric() if len(ts_data) < MIN_VALID_POINTS: send_alert(f"时序数据不足（{len(ts_data)}/{EXPECTED_DATA_POINTS}），终止AI预测，请检查采集！") time.sleep(600) continue llm_res = llm_predict(ts_data) prophet_res = prophet_predict(ts_data) pre_estimate = max(ts_data[-1]*1.1, THRESHOLD-5) # 冲突仲裁 if llm_res != "ERR" and prophet_res != "ERR": if llm_res != prophet_res: send_alert(f"【冲突】LLM={llm_res}, Prophet={prophet_res}，触发保守策略：NO_ACTION，请人工核验") logging.warning(f"双模型决策冲突: LLM={llm_res}, Prophet={prophet_res}") decision = "NO_ACTION" else: decision = llm_res logging.info(f"双模型一致: {decision}") elif llm_res != "ERR": decision = llm_res logging.info(f"LLM决策（Prophet异常）: {decision}") elif prophet_res != "ERR": decision = prophet_res logging.info(f"Prophet兜底决策（LLM异常）: {decision}") else: decision = "NO_ACTION" send_alert("双模型均异常，本次跳过执行，请检查Ollama和Prophet") logging.error("双模型均失败") logging.info(f"最终决策: {decision}") real_after = -1 if decision == "ACTION": send_alert(f"预测3h内{MOUNT_POINT}磁盘使用率≥{THRESHOLD}%，即将执行自动清理") ok = safe_clear_log() if ok: real_after = check_after_clear() if real_after >= 0: send_alert(f"清理完成，当前磁盘使用率: {real_after:.2f}%") else: consecutive_fail_clear +=1 if consecutive_fail_clear >= FAIL_CLEAR_THRESHOLD: agent_disabled = True send_alert(f"【熔断】连续{consecutive_fail_clear}次清理失败，Agent已禁用，请人工介入") send_alert(f"清理执行失败或无可清理文件，请检查") else: logging.info("预测无溢出风险，无需操作") record_self_reflection(decision, pre_estimate, real_after) except Exception as e: err_msg = f"Agent主循环异常: {str(e)}" logging.error(err_msg) send_alert(err_msg) time.sleep(600)if __name__ == "__main__": agent_main()

补充说明：内存、带宽时序特征与磁盘存量数据不同，复用框架时需单独调整Prophet周期参数与预测步长。

备注：不同资源时序特征存在差异，复用需要微调Prophet季节参数、预测时长：