## 背景：群里喊人派单，终于喊崩了

我在一家做IT运维外包（MSP）的公司负责交付体系搭建。去年底客户数从8家涨到32家，出问题了。

具体怎么崩的：12月某天晚上，3个客户同时出故障——一个是华东门店断网、一个是华南客户的ERP响应超时、还有一个是备份任务失败。值班的同事在企微群里发了3条消息叫人处理。结果网络组的小张以为华南的事情有人接了（实际没人接），那条消息被后面的讨论刷上去了。等客户自己打电话过来催，已经过去47分钟，SLA早就超了。

复盘的时候大家都觉得"不是我的问题"——确实也不是谁的问题，是没有系统。

当时我们面临的状况：

- 腾讯云监控CM已经部署了，告警能收到，但告警之后全靠人肉
- 工单记在飞书文档里，格式不统一，谁写的只有谁看得懂
- 派单靠值班长在群里@人，响应没人跟踪，升级全凭感觉
- 月底跟客户汇报SLA数据，要从聊天记录里一条一条扒时间

这篇文章记录我们花了3周，用腾讯云现有的产品组合把这件事解决到什么程度。

![Gemini_Generated_Image_yl29kvyl29kvyl29.png](https://developer.qcloudimg.com/http-save/yehe-12466791/78f71dbc6e00cf4c188fc7eb377cc731.png)

---

## 一、整体架构：四段闭环，每段都得有系统兜底

先说结论，最后跑通的架构长这样：

```
腾讯云监控CM告警 → 告警回调URL → 云函数SCF（事件聚合+研判）
    → 写入工单表（腾讯云数据库MySQL） 
    → 企微群机器人通知派单
    → 超时未响应 → SCF定时触发器检查 → 升级通知
    → 工单关闭 → 复盘数据自动归档
```

用到的腾讯云产品：
- **云监控CM**：基础指标采集 + 告警策略 + 告警回调
- **云函数SCF**：事件聚合逻辑 + 超时检查 + 所有胶水逻辑
- **云数据库MySQL**：工单存储（用的最小规格，够了）
- **企业微信Webhook**：派单通知 + 升级通知

成本：SCF按调用次数计费，我们月均调用不到5万次，几乎不花钱。MySQL用的1核1G基础版，月费大概30多块。

---

## 二、第一步：告警不等于事件——用SCF做聚合降噪

这是我们踩的第一个坑。

一开始直接把CM告警回调连到企微机器人——告警来了就发群里。结果第一天晚上，某客户的一台交换机抖动，5分钟内触发了23条告警（端口up/down反复），群里直接炸了。第二天早上大家打开手机全是红点，值班同事说"我以为是告警风暴就没管，结果里面混了一条真的断网"。

教训：**告警必须先聚合成"事件"，事件才值得开工单。**

### 聚合逻辑

我们在SCF里实现了一个简单但够用的聚合：

```python
# SCF函数：接收CM告警回调，聚合后决定是否开工单
import json
import time
import pymysql

# 聚合窗口：同一聚合键在5分钟内不重复开单
AGGREGATE_WINDOW = 300  # 秒

def main_handler(event, context):
    """接收腾讯云监控CM告警回调"""
    body = json.loads(event.get("body", "{}"))
    
    # CM告警回调的数据结构
    alarm_status = body.get("alarmStatus", "")  # "1"=告警触发, "0"=告警恢复
    alarm_type = body.get("alarmType", "")
    dimensions = body.get("dimensions", {})
    
    # 提取聚合键：客户ID + 实例ID + 告警指标
    customer_id = extract_customer_from_tag(dimensions)
    instance_id = dimensions.get("unInstanceId", "")
    metric_name = body.get("metricName", "")
    
    aggregate_key = f"{customer_id}:{instance_id}:{metric_name}"
    
    if alarm_status != "1":
        # 告警恢复，更新工单状态
        resolve_ticket(aggregate_key)
        return {"code": 0, "message": "resolved"}
    
    # 检查聚合窗口内是否已有未关闭事件
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute(
        """SELECT id, created_at FROM tickets 
           WHERE aggregate_key = %s AND status != 'closed'
           AND created_at > NOW() - INTERVAL %s SECOND""",
        (aggregate_key, AGGREGATE_WINDOW)
    )
    existing = cursor.fetchone()
    
    if existing:
        # 聚合窗口内已有事件，追加证据不新开工单
        append_evidence(existing[0], body)
        return {"code": 0, "message": "aggregated"}
    
    # 新事件，创建工单 + 派单
    ticket_id = create_ticket(conn, body, aggregate_key, customer_id)
    dispatch_ticket(ticket_id, customer_id, metric_name)
    
    return {"code": 0, "message": f"ticket created: {ticket_id}"}


def extract_customer_from_tag(dimensions):
    """从实例标签中提取客户ID
    我们给每个客户的CVM/CLB都打了customer_id标签"""
    tags = dimensions.get("tags", [])
    for tag in tags:
        if tag.get("tagKey") == "customer_id":
            return tag.get("tagValue", "unknown")
    return "unknown"
```

### 聚合效果

上线第一周做了个统计：7天内CM推了1847条原始告警，经过聚合后实际开了143张工单。降噪率92%。

其中效果最明显的是网络抖动类告警——之前一次交换机重启能产生几十条告警轰炸群里，现在合并成1张工单，里面附了所有相关的告警快照作为证据。

---

## 三、第二步：工单字段标准化——MySQL表设计

说实话这步花的时间最短但争论最多。团队里有人想用JIRA，有人想用飞书多维表格，最后我拍板用了最简单的方案——一张MySQL表，字段尽量少。

为什么？因为MSP工单最核心的需求是**统计SLA**。只要能回答"什么时间收到的、什么时间响应的、什么时间恢复的"，其他花哨功能都可以后面再加。

```sql
CREATE TABLE tickets (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    aggregate_key VARCHAR(256) NOT NULL,
    customer_id VARCHAR(64) NOT NULL,
    site_name VARCHAR(128) DEFAULT '',
    instance_id VARCHAR(128) NOT NULL,
    
    -- 事件信息
    event_type ENUM('unavailable', 'degraded', 'backup_fail') NOT NULL,
    priority ENUM('P0', 'P1', 'P2') NOT NULL,
    description TEXT,
    
    -- 派单信息
    assigned_group VARCHAR(64),
    assigned_engineer VARCHAR(64),
    
    -- 状态机
    status ENUM('new', 'acknowledged', 'in_progress', 
                'resolved', 'reviewing', 'closed') 
        NOT NULL DEFAULT 'new',
    
    -- SLA时间戳（这几个字段是核心）
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    ack_at DATETIME DEFAULT NULL,          -- 首次响应时间
    resolved_at DATETIME DEFAULT NULL,     -- 恢复时间
    closed_at DATETIME DEFAULT NULL,
    
    -- SLA配置（不同客户/优先级的SLA不同）
    sla_ack_minutes INT NOT NULL DEFAULT 15,
    sla_resolve_minutes INT NOT NULL DEFAULT 120,
    
    -- 复盘字段
    root_cause TEXT,
    impact_scope VARCHAR(256),
    action_taken TEXT,
    prevention TEXT,
    
    INDEX idx_status (status),
    INDEX idx_customer_status (customer_id, status),
    INDEX idx_aggregate (aggregate_key),
    INDEX idx_created_status (created_at, status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 工单时间线表：每次状态变更都记录
CREATE TABLE ticket_timeline (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    ticket_id BIGINT NOT NULL,
    operator VARCHAR(64) NOT NULL,
    action VARCHAR(32) NOT NULL,
    old_status VARCHAR(32),
    new_status VARCHAR(32),
    comment TEXT,
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_ticket (ticket_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
```

### 状态机

状态流转很简单，一共就6个状态：

```
new → acknowledged → in_progress → resolved → reviewing → closed
```

每次状态变更都往`ticket_timeline`表里插一行，记录谁、什么时间、改成了什么状态。这样月底统计SLA的时候，直接查时间差就行。

我们踩的坑：一开始没有`reviewing`（待复盘）状态，工程师恢复故障后直接关单。结果月底发现很多工单的根因是空的，复盘全靠记忆。加了这个状态后强制要求"恢复后必须填根因才能关单"，根因填写率从不到20%涨到了85%。

---

## 四、第三步：派单规则化——3条规则够用了

一开始也想做智能调度，后来发现根本不需要。我们团队总共就8个工程师、分3个组，用3条if-else就把派单覆盖了：

```python
def dispatch_ticket(ticket_id, customer_id, metric_name):
    """根据规则自动派单，并通过企微机器人通知"""
    
    # 规则1：按客户归属区域
    region = get_customer_region(customer_id)
    candidates = get_engineers_by_region(region)
    
    # 规则2：按技能匹配
    skill_required = classify_skill(metric_name)
    # network: 端口/带宽/丢包类指标
    # system: CPU/内存/磁盘类指标  
    # application: 响应时间/错误率类指标
    candidates = [e for e in candidates if skill_required in e["skills"]]
    
    # 规则3：按当前负载（未关闭工单数最少的优先）
    candidates.sort(key=lambda e: e["open_ticket_count"])
    
    if not candidates:
        # 兜底：没有合适人选时派给值班长
        assigned = get_duty_leader()
        notify_wecom_escalation(ticket_id, "无合适工程师，请人工分配")
    else:
        assigned = candidates[0]
    
    update_ticket_assignment(ticket_id, assigned["group"], assigned["name"])
    notify_wecom_new_ticket(ticket_id, assigned)
```

### 企微机器人通知——这是工程师真正在看的入口

老实说，工程师不会主动去刷工单系统。他们看的是企微群消息。所以通知格式我们迭代了3版才定下来：

```python
def notify_wecom_new_ticket(ticket_id, engineer):
    """新工单派发通知——发到对应组的企微群"""
    import requests
    
    ticket = get_ticket(ticket_id)
    
    # 根据优先级决定通知强度
    if ticket["priority"] == "P0":
        prefix = "🔴 P0 紧急"
    elif ticket["priority"] == "P1":
        prefix = "🟡 P1"
    else:
        prefix = "🟢 P2"
    
    # 注意：要@人必须用text类型，markdown类型不支持@
    content = (
        f"{prefix} | 新工单 #{ticket_id}\n"
        f"客户：{ticket['customer_id']}\n"  
        f"实例：{ticket['instance_id']}\n"
        f"事件：{ticket['description'][:50]}\n"
        f"SLA响应时限：{ticket['sla_ack_minutes']}分钟\n"
        f"责任人：{engineer['name']}\n\n"
        f"回复「接{ticket_id}」确认响应"
    )
    
    webhook_url = get_group_webhook(engineer["group"])
    
    # 用text类型才能@人
    payload = {
        "msgtype": "text",
        "text": {
            "content": content,
            "mentioned_mobile_list": [engineer["mobile"]]
        }
    }
    
    resp = requests.post(webhook_url, json=payload, timeout=5)
    if resp.status_code != 200:
        print(f"wecom notify failed: {resp.text}")
```

这里踩的坑：一开始用markdown消息类型发@人，结果发现markdown类型不支持mentioned_list。腾讯企微的文档里这个写得不太明显，我们试了好几次才搞对。

---

## 五、超时升级：用SCF定时触发器兜底

这一步是整个闭环里最容易被忽略、但最关键的。

如果派了单没人响应怎么办？靠值班长自己盯着看？那值班长就变成人肉监控了。

我们用了SCF的定时触发器，每分钟跑一次检查：

```python
# SCF函数：定时触发器，每分钟检查超时工单
import pymysql

def timeout_checker(event, context):
    """检查未响应和未恢复的工单是否超时"""
    conn = get_db_connection()
    cursor = conn.cursor(pymysql.cursors.DictCursor)
    
    # 检查1：超过SLA响应时限仍未确认的工单
    cursor.execute("""
        SELECT id, customer_id, assigned_engineer, priority,
               sla_ack_minutes, created_at,
               TIMESTAMPDIFF(MINUTE, created_at, NOW()) as elapsed
        FROM tickets
        WHERE status = 'new'
        AND TIMESTAMPDIFF(MINUTE, created_at, NOW()) > sla_ack_minutes
    """)
    unacked = cursor.fetchall()
    
    for ticket in unacked:
        elapsed = ticket["elapsed"]
        sla = ticket["sla_ack_minutes"]
        
        if elapsed > sla * 2:
            # 超时200%：升级到技术经理
            escalate_to_manager(ticket)
        elif elapsed > sla:
            # 超时100%：升级到值班长
            escalate_to_leader(ticket)
    
    # 检查2：处理中但长时间无更新的工单
    cursor.execute("""
        SELECT t.id, t.assigned_engineer,
               TIMESTAMPDIFF(MINUTE, 
                   COALESCE(tl.last_update, t.ack_at), NOW()) as idle
        FROM tickets t
        LEFT JOIN (
            SELECT ticket_id, MAX(created_at) as last_update
            FROM ticket_timeline GROUP BY ticket_id
        ) tl ON t.id = tl.ticket_id
        WHERE t.status = 'in_progress'
        AND TIMESTAMPDIFF(MINUTE, 
            COALESCE(tl.last_update, t.ack_at), NOW()) > 60
    """)
    idle_tickets = cursor.fetchall()
    
    for ticket in idle_tickets:
        notify_wecom_reminder(ticket["id"], 
            f"工单已{ticket['idle']}分钟无更新，请补充进展")
    
    conn.close()
    return {"checked": len(unacked) + len(idle_tickets)}
```

### SCF定时触发器配置

在腾讯云SCF控制台配置：

- **触发方式**：定时触发
- **触发周期**：每1分钟（Cron表达式 `0 */1 * * * * *`）
- **函数超时**：设成15秒（默认3秒不够，数据库查询会超）
- **VPC配置**：要和MySQL在同一个VPC，否则连不上

---

## 六、CM告警回调配置细节

在腾讯云监控控制台配置：

1. 进入**告警管理** → **通知模板** → 新建模板
2. 接收渠道里选**接口回调**
3. URL填SCF函数的API网关触发器地址：
   ```
   https://service-xxxxx-xxxxx.gz.apigw.tencentcs.com/release/alarm_callback
   ```
4. 在对应告警策略里引用这个通知模板

CM会在保存时发一次验证请求，SCF函数需要返回HTTP 200。

我们踩的坑：CM告警回调的请求体格式在去年底做了一次调整，`dimensions`字段结构变了。如果你发现回调数据和文档对不上，直接在SCF日志里打印`event["body"]`看实际结构最靠谱：

```python
# 调试用：打印CM回调的实际数据结构
def debug_handler(event, context):
    import json
    print("=== CM callback body ===")
    print(json.dumps(json.loads(event.get("body", "{}")), 
                     ensure_ascii=False, indent=2))
    return {"code": 0}
```

---

## 七、3周落地时间线和踩坑记录

| 时间 | 做了什么 | 踩的坑 |
|------|------|------|
| 第1周前3天 | MySQL建表 + SCF聚合函数 + CM告警回调对接 | SCF访问VPC内MySQL要单独配网络；CM回调格式与文档不一致 |
| 第1周后2天 | 企微机器人通知 + 派单规则 | markdown类型不支持@人，改用text类型 |
| 第2周 | 超时检查函数 + 定时触发器 + 升级通知 | SCF默认超时3秒不够用；Cron表达式写法和Linux不完全一样 |
| 第3周 | 试运行 + 修bug + 补复盘流程 | 工程师忘记确认响应（加了超时自动升级兜底） |

第3周结束后的数据（对比改造前）：

| 指标 | 改造前 | 改造后 |
|------|--------|--------|
| 日均协调派单耗时 | ~2小时 | <10分钟（大部分自动派） |
| SLA响应超时率 | 无数据（没记录） | 8%（第1月）→ 3%（第2月） |
| 月底SLA报告制作时间 | 2天（从聊天记录扒） | 15分钟（SQL直接查） |
| 告警风暴导致遗漏真实事件 | 月均2-3次 | 0次 |
| 工单根因填写率 | ~20% | 85% |

---

## 八、几个补充踩坑

**1. SCF冷启动**

Python运行时冷启动大概800ms-1.5s。对告警回调来说完全可接受（CM本身就是分钟级采集）。如果对延迟敏感可以开预置并发。

**2. 企微机器人频率限制**

每个Webhook每分钟最多发20条。告警风暴时如果聚合没做好会触发限流。这也是为什么聚合逻辑一定要放在通知前面。

**3. CM告警恢复通知可能延迟**

CM判定恢复需要连续N个周期正常。工程师手动恢复故障后，CM可能5-10分钟后才推恢复回调。我们的做法是：工程师在群里回复「复#{工单号}」手动标记恢复，CM恢复回调到了再做二次确认。

**4. 多客户共用实例的标签问题**

我们给每个客户的资源都打了`customer_id`标签。但有几台共用的跳板机漏打了，导致告警时解析为"unknown"。上线前一定要检查标签覆盖率。

**5. 工单量上来后要补索引**

第二周日均工单50+后，超时检查函数执行时间从2秒涨到8秒。加了`(created_at, status)`联合索引后降回1秒内。建表时就该加的，算是偷懒的代价。

---

## 小结

整套方案用到的腾讯云产品就4个：云监控CM、云函数SCF、云数据库MySQL、企业微信Webhook。没有引入任何第三方ITSM工具，总计代码量不到500行Python。

核心思路是把每段的"人工判断"替换成"规则执行"：
- 告警来了不靠人看，靠聚合规则判断是否值得开工单
- 工单不靠群里喊人，靠派单规则自动分配+机器人通知
- 超时不靠值班长盯，靠定时触发器自动检查+升级

下一步准备接入腾讯云日志服务CLS做告警关联分析——告警触发时自动捞相关时间段的日志附到工单里，减少工程师排查时的上下文切换。但那是后面的事了，先把最小闭环跑稳再说。

如果你也在做MSP交付体系搭建，欢迎评论区交流你们的方案和踩过的坑。

我在一家做IT运维外包（MSP）的公司负责交付体系搭建。去年底客户数从8家涨到32家，出问题了。

腾讯云监控+云函数+企微机器人：我们如何用3周搭起MSP工单派单自动化闭环

管5家客户时靠群里喊一嗓子就行，管到30家时每天光协调派单就吃掉2小时。本文记录我们用腾讯云监控CM告警回调+云函数SCF做事件聚合+企业微信机器人派单通知，3周跑通"监控→工单→派单→复盘"闭环的完整过程，含SCF函数代码、告警策略配置、企微Webhook对接细节，以及踩过的5个坑。

产品经理

云计算

后端

运维

IT运维外包公司通过腾讯云产品组合（云监控CM、云函数SCF、云数据库MySQL、企微Webhook）构建自动化工单系统，实现告警聚合降噪、工单标准化派发、超时自动升级闭环。3周落地后SLA响应超时率从8%降至3%，告警风暴遗漏事件归零，月底报告制作时间从2天缩短至15分钟。

数据结构

云数据库

日志服务

数据库

云函数

云监控

Python

MySQL

Linux

2026采购季 | AI焕新·智启新局

tcpm

腾讯云数据库特惠，新用户首单0.5折起


tencentdb-catalog

4核4G3M云服务器 新用户低至38元/年！

cngw

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

腾讯云监控+云函数+企微机器人：我们如何用3周搭起MSP工单派单自动化闭环-腾讯云开发者社区-腾讯云

腾讯云监控+云函数+企微机器人：我们如何用3周搭起MSP工单派单自动化闭环

腾讯云监控+云函数+企微机器人：我们如何用3周搭起MSP工单派单自动化闭环

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐