基于 SLS + GitLab + Jenkins 的微服务发布监控系统

用 AI Agent 自动监控微服务发版全流程：pom 版本盘点 → 线上版本对比 → 启动日志确认 → ERROR 关联分析。无需人工盯守，发版完成自动通知。

---

目录

1. 背景与痛点
2. 整体架构
3. 核心流程详解
4. 技术选型
5. 配置数据结构
6. 核心模块代码
7. 定时任务部署
8. 通知模板设计
9. 踩坑记录
10. 扩展与改进方向

---

1. 背景与痛点

在微服务架构下，一个业务系统可能有十几个甚至几十个服务同时迭代。每晚发版时，运维和开发团队面临以下痛点：

• 人工盯守成本高：需要持续关注 Jenkins 构建状态、检查各服务是否部署成功
• 启动确认滞后：服务多、机器多，逐台检查启动日志耗时耗力
• ERROR 关联困难：发版后出现的 ERROR，需要人工判断是否与当天代码变更相关
• 信息碎片化：pom 版本、线上版本、启动日志、错误日志分散在不同系统

目标：构建一套自动化监控系统，覆盖"版本变更检测 → 发版确认 → 启动验证 → ERROR 分析"全链路，检测到异常自动推送通知。

---

2. 整体架构

┌─────────────────────────────────────────────────────────┐
│                    定时任务调度器（Cron）                    │
│              每 3 分钟执行一次 Python 脚本                    │
└──────────────────────┬──────────────────────────────────┘
                       │
         ┌─────────────┼─────────────┐
         ▼             ▼             ▼
   ┌──────────┐  ┌──────────┐  ┌──────────┐
   │  GitLab  │  │  Jenkins │  │ 阿里云 SLS │
   │  API     │  │  API     │  │  API      │
   └────┬─────┘  └────┬─────┘  └────┬─────┘
        │             │             │
        ▼             ▼             ▼
   ┌──────────────────────────────────────┐
   │           监控脚本核心逻辑             │
   │  ┌─────────────────────────────────┐ │
   │  │ 阶段1: pom 版本盘点（每日一次）   │ │
   │  │ 阶段2: 线上版本对比（每3分钟）    │ │
   │  │ 阶段3: 启动日志检测              │ │
   │  │ 阶段4: ERROR 关联分析           │ │
   │  └─────────────────────────────────┘ │
   └──────────────────┬───────────────────┘
                      │
                      ▼
              ┌──────────────┐
              │  消息推送      │
              │ (钉钉/飞书等)  │
              └──────────────┘

数据源

数据源	用途	接口方式
GitLab	pom.xml 版本号、master 分支提交记录	REST API + Token
Jenkins	Tomcat 类老项目的构建版本（无 HTTP version 接口时）	REST API（匿名）
阿里云 SLS	启动日志、ERROR 日志	aliyun-log-python-sdk（AccessKey）
线上 version 接口	各服务当前部署版本	HTTP GET（返回 JSON）

---

3. 核心流程详解

阶段 1：每日 pom 版本盘点（每天 18:30 执行一次）

拉取所有项目的 pom.xml
    │
    ▼
解析版本号（支持三种 pom 写法）
    │
    ▼
对比昨天记录 → 找出【今天版本号发生变化】的项目
    │
    ▼
输出："📦 今日待升级候选：service-a (1.0.49→1.0.50), service-b (2.1.10→2.1.11)"

为什么只监控 pom 变化？ 不是每次提交都会发版，只有 pom.xml 中的版本号变更才代表正式的版本升级。

阶段 2：线上版本对比（19:00 后每 3 分钟）

遍历【今日待升级候选】
    │
    ▼
调用各服务的 /version 接口
    │
    ├── 线上版本 == pom 新版本 → ✅ 发版完成！
    │       └── 触发阶段 3（启动日志确认）
    │
    └── 线上版本 != pom 新版本 → ⏳ 未发版，继续等待

关键优化：pom 没改的项目根本不调 version 接口，节省请求量。

阶段 3：启动日志检测（发版完成后立即执行）

查询 SLS 日志：message: "Tomcat started on port" AND __topic__: service-a
    │
    ▼
按 __topic__（服务名）+ __source__（IP）分组
    │
    ▼
统计：service-a 启动了 N 台机器（时间: HH:MM）
    │
    ▼
触发阶段 4（15 分钟后执行 ERROR 分析）

阶段 4：ERROR 关联分析（启动后 15 分钟）

查询启动后 15 分钟窗口内的 ERROR 日志
    │
    ▼
统计 ERROR 类型分布（按消息前 60 字分类）
    │
    ▼
提取堆栈类名，与当天 master 提交的改动文件对比
    │
    ├── 命中 → ⚠️ ERROR 可能与当天迭代相关
    └── 未命中 → ✅ ERROR 与当天迭代无关
    │
    ▼
输出："📊 service-a 启动后15分钟检查：ERROR 100条，均与当天迭代无关"

关联逻辑：ERROR 堆栈中的类名 vs 当天提交改动的文件名。例如 ERROR 抛自 CardNotifyConsumer，当天只改了 GoodsPoolManager.java → 无关。

---

4. 技术选型

组件	选型	理由
监控脚本	Python 3.11+	生态丰富，SLS SDK 原生支持
日志查询	aliyun-log-python-sdk	阿里云官方 SDK，支持 AccessKey 直连
代码仓库	GitLab REST API	读 pom.xml、查提交记录
构建系统	Jenkins REST API	Tomcat 类老项目读构建版本
定时调度	Cron（或 Hermes Agent cronjob）	简单可靠
通知推送	钉钉 Webhook / 飞书 Webhook	一行代码推送群消息
状态持久化	JSON 文件	记录 pom 版本基线、已通知批次

---

5. 配置数据结构

5.1 服务注册表

每个服务需要配置以下信息：

# 服务 topic → (GitLab项目名, 项目ID, 分支, 网关名)
# 网关名为空字符串 = 无 HTTP version 接口，走 Jenkins
SERVICE_PROJECT = {
    "service-a": ("service-a-project", 101, "master", None),
    "service-b": ("service-b-project", 102, "master", None),
    "service-c": ("service-c-project", 103, "master", "service-c-gateway"),
    # Tomcat 类老项目：无 HTTP version 接口
    "service-legacy": ("service-legacy-project", 201, "master", ""),
}

字段	说明
key	SLS 日志中的 __topic__（服务名）
GitLab 项目名	用于读 pom.xml
项目 ID	GitLab API 查询用
分支	统一读 master（避免 default_branch 不一致问题）
网关名	version 接口的 URL 路径；空字符串 = 无接口，走 Jenkins

5.2 特殊服务日志库

# 默认查主日志库，特殊服务用专属库
SERVICE_LOGSTORE = {
    "service-legacy": ["legacy-log-store"],
}

5.3 特殊启动条件

# 默认: message: "Tomcat started on port"
# 多 topic 服务需要各自独立的启动条件（不能合并查询）
SERVICE_STARTUP_QUERY = {
    "service-legacy": [
        ("legacy-thread", 'message: "Thread start success"'),
        ("legacy-core", "message: started"),
    ],
}

5.4 Jenkins Job 映射

# 无 HTTP version 接口的服务，走 Jenkins 构建版本对比
JENKINS_JOBS = {
    "service-legacy": ["build-service-legacy-main", "build-service-legacy-thread"],
}

---

6. 核心模块代码

6.1 SLS 日志查询（AccessKey 直连）

import os
import sys
from datetime import datetime, timezone, timedelta

# SDK 路径（cron 环境可能需要手动指定）
# sys.path.insert(0, "/path/to/site-packages")

from aliyun.log import LogClient, GetLogsRequest

# ===== 配置（从环境变量读取） =====
SLS_PROJECT = os.environ.get("SLS_PROJECT", "my-project")
SLS_ENDPOINT = os.environ.get("SLS_ENDPOINT", "cn-hangzhou.log.aliyuncs.com")
SLS_AK_ID = os.environ.get("SLS_ACCESS_KEY_ID", "")
SLS_AK_SECRET = os.environ.get("SLS_ACCESS_KEY_SECRET", "")
DEFAULT_LOGSTORE = os.environ.get("SLS_LOGSTORE", "my-logstore")

BJ = timezone(timedelta(hours=8))

_client = None

def get_sls_client():
    global _client
    if _client is None:
        _client = LogClient(SLS_ENDPOINT, SLS_AK_ID, SLS_AK_SECRET)
    return _client

def query_sls(query, from_ts, to_ts, size=100, logstore=None):
    """查询 SLS 日志，返回 (ok, data_dict)"""
    logstore = logstore or DEFAULT_LOGSTORE
    try:
        req = GetLogsRequest(
            SLS_PROJECT, logstore, int(from_ts), int(to_ts), "",
            query=query, line=min(size, 1000), offset=0, reverse=True
        )
        resp = get_sls_client().get_logs(req)
    except Exception as e:
        return False, {"error": str(e)[:300]}
    
    logs = []
    for item in resp.get_logs():
        d = item.get_contents()
        d["__source__"] = item.get_source() or ""  # IP 地址
        logs.append(d)
    return True, {"data": logs}

6.2 GitLab pom 版本读取

import json
import re
import urllib.request
import urllib.parse

GITLAB_URL = os.environ.get("GITLAB_URL", "https://gitlab.example.com")
GITLAB_TOKEN = os.environ.get("GITLAB_TOKEN", "")

def gitlab_get(path, params=None):
    """GitLab REST API 调用"""
    url = f"{GITLAB_URL}/api/v4{path}"
    if params:
        url += "?" + urllib.parse.urlencode(params)
    req = urllib.request.Request(url, headers={"PRIVATE-TOKEN": GITLAB_TOKEN})
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            return json.loads(r.read())
    except Exception as e:
        return {"_error": str(e)}

def get_pom_version(project_id, branch="master"):
    """从 GitLab 读取 pom.xml 的版本号
    
    支持三种 pom 写法：
    1. <revision>xxx</revision>（flatten 插件）
    2. 根 <artifactId>xxx</artifactId><version>xxx</version>
    3. 老式 <artifactId>xxx</artifactId><packaging>xxx</packaging><version>xxx</version>
    """
    data = gitlab_get(
        f"/projects/{project_id}/repository/files/pom.xml/raw",
        {"ref": branch}
    )
    if isinstance(data, dict) and data.get("_error"):
        return None
    
    # raw 接口返回纯文本（可能是 bytes 或 str）
    if isinstance(data, bytes):
        text = data.decode("utf-8", errors="replace")
    elif isinstance(data, str):
        text = data
    else:
        text = str(data)
    
    # 截取 <dependencies> 之前头部（避免匹配到依赖版本）
    head = text.split("<dependencies>")[0] if "<dependencies>" in text else text[:3000]
    
    # 写法1: <revision>xxx</revision>
    m = re.search(r"<revision>([^<]+)</revision>", head)
    if m:
        return m.group(1).strip()
    
    # 写法2/3: <artifactId>xxx</artifactId> ... <version>xxx</version>
    # 取第一个 artifactId 之后紧跟的 version
    m = re.search(
        r"<artifactId>[^<]+</artifactId>\s*"
        r"(?:<packaging>[^<]+</packaging>\s*)?"
        r"<version>([^<]+)</version>",
        head
    )
    if m:
        return m.group(1).strip()
    
    return None

6.3 Jenkins 构建版本读取

def get_jenkins_version(job_name):
    """从 Jenkins 最近一次成功构建的控制台输出提取 POM_VERSION"""
    jenkins_url = os.environ.get("JENKINS_URL", "http://jenkins.example.com:8080")
    url = f"{jenkins_url}/job/{job_name}/lastSuccessfulBuild/consoleText"
    try:
        req = urllib.request.Request(url)
        with urllib.request.urlopen(req, timeout=30) as r:
            # 控制台输出可能很大，只读前 50KB
            text = r.read(50000).decode("utf-8", errors="replace")
    except Exception:
        return None
    
    # 提取 POM_VERSION=xxx（构建脚本中 grep pom 版本的输出）
    m = re.search(r"^\s*POM_VERSION=([^\s]+)\s*$", text, re.MULTILINE)
    if m:
        return m.group(1).strip()
    
    # 兜底: [INFO] Building xxx 1.2.3
    m = re.search(r"\[INFO\] Building \S+ ([0-9][\w.\-]+)", text)
    if m:
        return m.group(1).strip()
    
    return None

6.4 线上版本查询

def get_online_version(service_name):
    """查询线上服务版本（HTTP version 接口）"""
    version_base = os.environ.get("VERSION_BASE_URL", "https://api.example.com")
    url = f"{version_base}/{service_name}/version"
    try:
        req = urllib.request.Request(url)
        with urllib.request.urlopen(req, timeout=10) as r:
            data = json.loads(r.read())
            return data.get("data") or data.get("version")
    except Exception:
        return None

6.5 通知推送

import hashlib
import hmac
import base64
import time

def send_dingtalk(webhook_url, title, content, at_all=False):
    """钉钉 Webhook 推送 Markdown 消息"""
    payload = {
        "msgtype": "markdown",
        "markdown": {"title": title, "text": content},
        "at": {"isAtAll": at_all},
    }
    data = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(
        webhook_url,
        data=data,
        headers={"Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as r:
            return json.loads(r.read())
    except Exception as e:
        return {"error": str(e)}

def send_feishu(webhook_url, title, content):
    """飞书 Webhook 推送"""
    payload = {
        "msg_type": "interactive",
        "card": {
            "header": {"title": {"tag": "plain_text", "content": title}},
            "elements": [{"tag": "markdown", "content": content}],
        },
    }
    data = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(
        webhook_url,
        data=data,
        headers={"Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as r:
            return json.loads(r.read())
    except Exception as e:
        return {"error": str(e)}

6.6 启动日志检测与 ERROR 关联

from collections import Counter

def detect_startup(service_key, from_ts, to_ts, candidates):
    """检测服务启动，返回 {topic: {ips, earliest_ts, count}}"""
    queries = get_project_startup_queries(service_key)
    logstores = get_project_logstores(service_key)
    results = {}
    
    for logstore in logstores:
        for topic, query in queries:
            full_query = f"__topic__: {topic} and {query}"
            ok, data = query_sls(full_query, from_ts, to_ts, 300, logstore=logstore)
            if not ok or not data.get("data"):
                continue
            
            # 按 IP 分组
            ips = set()
            earliest = None
            for log in data["data"]:
                ip = log.get("__source__", "")
                ts = parse_sls_time(log.get("time", ""))
                if ip:
                    ips.add(ip)
                if ts and (earliest is None or ts < earliest):
                    earliest = ts
            
            results[topic] = {
                "ips": ips,
                "count": len(data["data"]),
                "earliest_ts": earliest,
                "ip_count": len(ips),
            }
    
    return results

def analyze_errors(service_key, start_ts, window_seconds=900):
    """启动后 ERROR 分析（默认 15 分钟窗口）"""
    from_ts = start_ts
    to_ts = start_ts + window_seconds
    queries = get_project_startup_queries(service_key)
    logstores = get_project_logstores(service_key)
    
    error_types = Counter()
    stack_classes = []
    
    for logstore in logstores:
        for topic, _ in queries:
            query = f"__topic__: {topic} and level: ERROR"
            ok, data = query_sls(query, from_ts, to_ts, 100, logstore=logstore)
            if not ok or not data.get("data"):
                continue
            
            for log in data["data"]:
                msg = log.get("message", log.get("msg", ""))
                # 按消息前 60 字分类
                key = msg[:60].strip()
                error_types[key] += 1
                # 提取堆栈类名
                classes = re.findall(r"at (com\.[\w.]+)\.", log.get("throwable", msg))
                stack_classes.extend(classes)
    
    return {
        "total": sum(error_types.values()),
        "top_types": error_types.most_common(5),
        "stack_classes": list(set(stack_classes)),
    }

def correlate_with_commits(project_id, branch, since_iso, until_iso, stack_classes):
    """ERROR 堆栈类名 vs 当天提交改动文件关联"""
    commits = get_master_commits(project_id, branch, since_iso, until_iso)
    if not commits:
        return False, []
    
    for commit in commits:
        files = get_commit_files(project_id, commit["id"])
        file_names = {f.split("/")[-1].replace(".java", "") for f in files}
        
        for cls in stack_classes:
            cls_name = cls.split(".")[-1]
            if cls_name in file_names:
                return True, [commit]
    
    return False, commits

6.7 状态持久化

import os

STATE_FILE = os.environ.get("MONITOR_STATE_FILE", "/tmp/release_monitor_state.json")

def load_state():
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE) as f:
            return json.load(f)
    return {
        "pom_versions": {},      # {project: version} 昨天的 pom 版本基线
        "candidates": {},        # {topic: {project, old, new, status}} 今日待升级候选
        "daily_date": "",        # 今日盘点日期（避免重复盘点）
        "seen": {},              # {batch_key: status} 已通知批次
    }

def save_state(state):
    with open(STATE_FILE, "w") as f:
        json.dump(state, f, ensure_ascii=False, indent=2)

---

7. 定时任务部署

7.1 Cron 表达式

# 每 3 分钟执行一次（UTC 时区）
# 对应北京时间 18:00 ~ 次日 00:59
*/3 10-16 * * * /usr/bin/python3 /path/to/release_monitor.py

⚠️ 时区坑：Cron 表达式按 UTC 解释。北京 19:00 = UTC 11:00，所以 10-16 覆盖北京 18:00~次日 00:59。

7.2 脚本内时间窗口控制

def should_run():
    """判断当前是否在监控时间窗口内"""
    now_bj = datetime.now(BJ)
    today_19 = now_bj.replace(hour=19, minute=0, second=0, microsecond=0)
    today_1830 = now_bj.replace(hour=18, minute=30, second=0, microsecond=0)
    
    # 18:00-18:29 静默等待
    if now_bj.hour == 18 and now_bj.minute < 30:
        return False
    # 18:30 首跑盘点
    if now_bj.hour == 18 and now_bj.minute >= 30:
        return "inventory"
    # 19:00 后轮询
    if now_bj >= today_19:
        return "polling"
    return False

7.3 部署检查清单

# 1. 确认 Python 环境
python3 --version  # 需要 3.11+

# 2. 安装依赖
pip3 install aliyun-log-python-sdk

# 3. 配置环境变量（或 .env 文件）
export SLS_PROJECT="your-project"
export SLS_ACCESS_KEY_ID="YOUR_KEY_ID"
export SLS_ACCESS_KEY_SECRET="YOUR_KEY_SECRET"
export GITLAB_URL="https://gitlab.example.com"
export GITLAB_TOKEN="YOUR_GITLAB_TOKEN"
export VERSION_BASE_URL="https://api.example.com"
export DINGTALK_WEBHOOK="https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN"

# 4. 测试运行
python3 release_monitor.py

# 5. 注册 cron
crontab -e
# 添加: */3 10-16 * * * /usr/bin/python3 /path/to/release_monitor.py

# 6. 查看日志
tail -f /tmp/release_monitor.log

---

8. 通知模板设计

8.1 每日盘点通知（18:30）

📦 **今日待升级项目**（2026-08-24）

| 服务 | 版本变更 |
|:---|:---|
| service-a | 1.0.49 → 1.0.50 |
| service-b | 2.1.10 → 2.1.11 |

共 2 个候选，19:00 后开始轮询线上版本。

8.2 发版完成通知（实时）

✅ **service-a** 发版完成！
- 线上版本：1.0.50 == pom 版本 1.0.50
- 正在检查启动日志...

8.3 启动通知（两段式，第一段）

🚀 **service-a** 启动：47台机器（08-24 20:58 北京）
- 版本：1.0.50（今日升级已发版）
- 15分钟后查看 ERROR 是否与当天迭代相关

8.4 ERROR 分析通知（两段式，第二段，15 分钟后）

📊 **service-a** 启动后15分钟检查（47台，20:58 启动）

- ERROR：100条
- Top 错误：
  1. consume message error (CardNotifyConsumer) × 45
  2. Redis connection timeout × 30
  3. 子订单状态通知失败 × 25
- 堆栈类：CardNotifyConsumer, ProcessDataService
- 当天迭代：3 个提交（service-a-project）
  - abc1234 张三 修复商品池权限
  - def5678 李四 新增导出功能
- 结论：✅ ERROR 与当天迭代无关（堆栈类不在提交文件中）

---

9. 踩坑记录

9.1 SLS 查询不带 topic 会拿到全量日志

问题：查询 message: "Tomcat started on port" 不带 __topic__ 过滤，返回所有服务的启动日志，导致未启动的服务被误报。

解决：所有启动查询必须带 __topic__: {服务名} 前缀。

9.2 topic 名 ≠ 网关名 ≠ 项目名

问题：SLS 日志的 __topic__ 是采集时配置的服务名，可能与 version 接口的网关名、GitLab 项目名不一致（例如 topic 无连字符，网关名带连字符）。

解决：在配置中显式映射三者关系。

9.3 多 topic 服务不能合并查询

问题：一个服务有多个线程（如 jifen-thread、core），合并成一条 OR 查询会导致结果混乱。

解决：每个 topic 独立查询，结果归一到同一项目。

9.4 老式 pom 版本解析会匹配到依赖版本

问题：<artifactId>xxx</artifactId><version>1.0.2.1</version> 可能匹配到依赖 open-sdk 的版本号。

解决：截取 <dependencies> 之前的头部再正则。

9.5 GitLab default_branch 不可靠

问题：某些项目默认分支显示 main，但代码实际在 master。

解决：统一指定分支，不依赖 API 返回的 default_branch。

9.6 Cron 时区问题

问题：WSL 系统是 UTC，cron 表达式按 UTC 解释。写 19-23 会在北京时间凌晨 3 点跑。

解决：换算成 UTC 时间写 cron 表达式。

9.7 Cron 环境 Python 路径不同

问题：cron 使用的 Python 环境可能未安装 aliyun-log-python-sdk。

解决：脚本头部手动 sys.path.insert 添加 site-packages 路径。

9.8 启动后 15 分钟窗口的批次聚合

问题：47 台机器分批启动（间隔几秒），不应报 47 次。

解决：按时间聚类（间隔 ≤15 分钟 = 同一批），每批只报一次。

---

10. 扩展与改进方向

10.1 当前局限

• 单机运行：脚本依赖本地 cron，无高可用
• 状态文件：JSON 文件存储，无并发保护
• ERROR 关联：基于类名匹配，准确率有限
• 通知渠道：目前只支持钉钉/飞书 Webhook

10.2 改进方向

方向	方案
高可用	部署为 K8s CronJob，或用分布式锁（Redis）
更精确的 ERROR 关联	结合 traceId 链路追踪，或用 AI 分析 ERROR 上下文
可视化	接入 Grafana，展示发版时间线和 ERROR 趋势
回滚建议	ERROR 数超阈值时自动建议回滚版本
多环境支持	区分 staging/production 环境的发版监控
AI Agent 集成	通过 MCP 协议让 AI Agent 直接查询监控状态

10.3 与 AI Agent 结合

本系统已通过 MCP 协议与 Hermes Agent 集成，支持：

• 自然语言查询：“今天有发版吗？”、“service-a 最近一次启动是什么时候？”
• 自动分析：AI 读取 ERROR 日志后自动判断是否与当天迭代相关
• 定时巡检：Agent 按计划执行监控脚本，有异常才推送通知

---

附录：环境变量模板

# ===== SLS 日志服务 =====
SLS_PROJECT=my-sls-project
SLS_ENDPOINT=cn-hangzhou.log.aliyuncs.com
SLS_ACCESS_KEY_ID=YOUR_SLS_AK_ID
SLS_ACCESS_KEY_SECRET=YOUR_SLS_AK_SECRET
SLS_LOGSTORE=my-main-logstore

# ===== GitLab =====
GITLAB_URL=https://gitlab.example.com
GITLAB_TOKEN=YOUR_GITLAB_TOKEN

# ===== Jenkins =====
JENKINS_URL=http://jenkins.example.com:8080

# ===== 线上版本接口 =====
VERSION_BASE_URL=https://api.example.com

# ===== 通知 =====
DINGTALK_WEBHOOK=https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN

# ===== 监控配置 =====
MONITOR_STATE_FILE=/tmp/release_monitor_state.json

---

最后更新：2026-08-24
适用场景：Spring Boot 微服务 + GitLab + Jenkins + 阿里云 SLS 技术栈
协议：MIT