AICoding代码运行异常实时监控告警实现方案

一、方案概述

在AI辅助编码平台AICoding的日常运维场景中，用户在线编译、代码执行、模型调用过程会出现内存溢出、脚本报错、接口超时、进程崩溃等异常问题。若缺少实时监控告警机制，故障会持续影响多用户编码体验，运维人员无法第一时间定位问题。本文搭建一套轻量化实时监控告警系统，采集AICoding运行日志、进程指标、接口返回状态，触发阈值后推送告警信息，覆盖代码运行全链路异常识别。

本方案基于Python实现日志采集、指标监控、告警推送三大核心模块，适配AICoding后端服务，支持钉钉/企业微信告警渠道，低侵入、易部署，无需重构原有编码服务架构。

二、核心设计思路

指标采集层：监听AICoding代码执行日志文件，实时解析报错堆栈、执行耗时、内存占用、进程退出码四类核心数据；
规则判断层：预设异常阈值，包含代码执行失败率＞5%、单任务内存占用超800MB、单次执行超时10s、进程非正常退出四类触发规则；
告警推送层：规则命中后封装告警标题、故障时间、用户ID、报错详情，通过Webhook推送至运维告警群；
持久缓存层：临时存储近5分钟异常记录，避免短时间重复刷屏告警，设置告警冷却时间300s。

三、完整代码实现

3.1 依赖安装

pip install watchdog requests python-dotenv

3.2 监控告警核心代码 `aicoding_monitor.py`

import time
import requests
import re
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from collections import deque

# 配置项
LOG_PATH = "/var/log/aicoding/code_run.log"  # AICoding运行日志路径
WEBHOOK_URL = "https://oapi.dingtalk.com/robot/send?access_token=xxx"
ALERT_COOLDOWN = 300  # 告警冷却秒数
MEMORY_THRESHOLD = 800  # 内存阈值MB
TIMEOUT_THRESHOLD = 10  # 执行超时阈值s
recent_alert_time = 0
error_record_queue = deque(maxlen=100)

class LogMonitorHandler(FileSystemEventHandler):
    def on_modified(self, event):
        global recent_alert_time
        if event.is_directory:
            return
        try:
            with open(LOG_PATH, "r", encoding="utf-8") as f:
                lines = f.readlines()[-20:]
            for line in lines:
                # 匹配代码运行报错、内存超限、超时、进程崩溃日志
                err_match = re.search(r"ERROR.*code_run|MemoryOverLimit|ExecuteTimeout|ProcessExitCode=[1-9]", line)
                mem_match = re.search(r"MemoryUsage:(\d+)MB", line)
                time_match = re.search(r"RunCost:(\d+\.\d+)s", line)
                if not err_match:
                    continue
                # 指标阈值校验
                mem_val = int(mem_match.group(1)) if mem_match else 0
                run_time = float(time_match.group(1)) if time_match else 0
                if mem_val > MEMORY_THRESHOLD or run_time > TIMEOUT_THRESHOLD or "ERROR" in line:
                    now = time.time()
                    if now - recent_alert_time < ALERT_COOLDOWN:
                        continue
                    # 构造告警标题与内容
                    alert_title = "AICoding代码运行异常实时监控告警"
                    alert_content = f"""
【AICoding实时告警】
告警标题：{alert_title}
故障日志：{line.strip()}
检测时间：{time.strftime('%Y-%m-%d %H:%M:%S')}
内存占用阈值：{MEMORY_THRESHOLD}MB | 执行超时阈值：{TIMEOUT_THRESHOLD}s
                    """
                    self.send_alert(alert_title, alert_content)
                    recent_alert_time = now
                    error_record_queue.append(line)
        except Exception as e:
            print(f"日志读取异常：{str(e)}")

    def send_alert(self, title, content):
        send_data = {
            "msgtype": "text",
            "text": {"content": f"{title}\n{content}"}
        }
        resp = requests.post(WEBHOOK_URL, json=send_data, timeout=5)
        print(f"告警推送结果：{resp.status_code}, {resp.text}")

if __name__ == "__main__":
    event_handler = LogMonitorHandler()
    observer = Observer()
    observer.schedule(event_handler, path="/var/log/aicoding", recursive=False)
    print("AICoding代码运行监控服务已启动，等待日志异常...")
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

3.3 日志模拟测试脚本 `log_simulate.py`

用于模拟AICoding代码执行异常日志，验证监控告警功能：

import time
log_file = "/var/log/aicoding/code_run.log"
# 模拟内存超限异常
err_log = "2026-07-05 10:20:30 ERROR code_run user_id=10086 MemoryOverLimit MemoryUsage:920MB RunCost:3.2s\n"
# 模拟执行超时异常
timeout_log = "2026-07-05 10:21:15 ERROR code_run user_id=10099 ExecuteTimeout RunCost:12.8s MemoryUsage:256MB\n"
with open(log_file, "a", encoding="utf-8") as f:
    f.write(err_log)
    f.write(timeout_log)
print("模拟异常日志已写入，监控程序将触发告警")

四、部署与使用步骤

环境准备：安装Python3.8及以上版本，执行依赖安装命令；
日志路径配置：确认AICoding服务日志输出路径与LOG_PATH保持一致；
告警渠道配置：替换WEBHOOK_URL为企业钉钉/微信机器人真实地址；
启动监控：python aicoding_monitor.py，后台常驻运行可搭配systemd托管；
功能验证：执行python log_simulate.py写入异常日志，查看告警群接收消息。

五、扩展优化方案

多渠道告警：新增邮件、短信推送接口，实现分级告警，严重故障触发短信通知；
告警分级：区分普通报错、内存溢出、服务崩溃三类等级，展示不同告警颜色；
数据可视化：对接Prometheus+Grafana，展示代码异常次数、平均内存占用趋势图；
日志溯源：对接ELK栈，告警附带日志检索链接，运维一键查看完整上下文。

六、总结

本套AICoding代码运行异常实时监控告警系统以日志监听为核心，轻量无侵入，通过预设指标阈值精准捕获编码服务各类运行故障，自动化推送告警信息，大幅缩短故障发现与处理时长。代码可直接部署使用，适配私有化部署的AICoding平台，同时支持灵活扩展监控规则与告警渠道，满足不同规模研发团队运维需求。

海量精选技术文档和实战案例持续更新，敬请关注【风骏时光少年】

AICoding代码运行异常实时监控告警