多门店运维闭环全景架构监控告警工单SLA复盘一套最小可用系统怎么串起来如果你是一个连锁奶茶品牌的运维工程师每天面对全国 200 家门店的监控告警你会怎么做 今天我们不聊大厂那套复杂的 Prometheus Grafana PagerDuty 全栈方案就讲一个最小可用、能跑通闭环的系统怎么搭起来。 我会用 Python 和简单技术栈把“监控 → 告警 → 工单 → SLA → 复盘”这条链完整串给你看。—## 一、闭环是什么为什么需要闭环先想一个场景 门店 A 的冰柜温度报警了监控系统发了消息运维看一眼哦好了。 但第二天同样的问题又来了第三天还来。没人知道到底修没修好也没人知道维修花了多久。这就是没有闭环——监控只管告警没人跟踪处理结果。而闭环架构要解决的是1.监控发现异常温度、门禁、网络2.告警通知到人钉钉/微信/短信3.工单自动创建任务谁处理什么优先级4.SLA承诺多长时间内必须响应/修复5.复盘事后统计哪里频繁出问题哪里响应慢下面我用一个最小系统来演示这个流程。—## 二、架构总览四层模型┌─────────────────────────────────────────┐│ 复盘层 (复盘报表) │├─────────────────────────────────────────┤│ SLA层 (超时计算升级告警) │├─────────────────────────────────────────┤│ 工单层 (创建/分配/状态流转) │├─────────────────────────────────────────┤│ 监控告警层 (采集/规则/通知) │└─────────────────────────────────────────┘每层之间通过事件总线简单点就用 Redis 队列或 Kafka串联数据流方向是单向的监控 → 工单 → SLA → 复盘。—## 三、第一步监控 告警层最小实现我们用 Python 模拟一个门店温度监控器。 实际生产中可能是 IoT 设备上报这里为演示写一个模拟脚本。python# monitor_simulator.py# 模拟20家门店的温度数据超过阈值则触发告警事件import randomimport timeimport jsonimport redis# 连接Redis作为事件总线r redis.Redis(hostlocalhost, port6379, db0)STORES [fstore_{i:03d} for i in range(1, 21)] # 20家门店TEMP_THRESHOLD 8.0 # 温度告警阈值摄氏度while True: store_id random.choice(STORES) # 模拟温度波动大部分正常偶尔超标 temperature round(random.uniform(2.0, 12.0), 1) if temperature TEMP_THRESHOLD: event { type: temperature_alert, store_id: store_id, value: temperature, timestamp: time.time() } # 发布到redis频道 r.publish(monitor_events, json.dumps(event)) print(f[ALERT] {store_id} 温度 {temperature}°C 超过阈值 {TEMP_THRESHOLD}°C) else: print(f[OK] {store_id} 温度 {temperature}°C) time.sleep(random.uniform(0.5, 2.0)) # 随机间隔这段代码做了三件事- 模拟门店温度- 判断是否超过阈值- 超过则发布告警事件到 Redis 频道—## 四、第二步告警 → 工单自动创建有了告警事件我们需要自动创建工单。 工单需要包含门店、问题描述、优先级、创建时间、处理人。python# ticket_creator.py# 监听Redis事件自动创建工单并推送到SLA检查队列import jsonimport timeimport redisfrom datetime import datetime, timedeltar redis.Redis(hostlocalhost, port6379, db0)pubsub r.pubsub()pubsub.subscribe(monitor_events)# 模拟一个简单的工单存储实际可用SQLite或MySQLtickets {}def create_ticket(event): 根据告警事件创建工单 ticket_id fTICKET-{int(time.time())} ticket { id: ticket_id, store_id: event[store_id], problem: f温度异常: {event[value]}°C, priority: high if event[value] 10 else medium, status: open, created_at: datetime.now().isoformat(), sla_deadline: (datetime.now() timedelta(hours2)).isoformat(), # 2小时SLA assigned_to: None } tickets[ticket_id] ticket # 将工单推送到SLA检查队列 r.lpush(sla_check_queue, json.dumps(ticket)) print(f[工单创建] {ticket_id} for {event[store_id]}) return ticketfor message in pubsub.listen(): if message[type] message: event json.loads(message[data]) if event[type] temperature_alert: create_ticket(event)这里的关键设计 - 工单创建后立即推送到sla_check_queue供后续SLA检查模块消费 - 工单优先级根据异常严重程度动态调整10°C 为 high - SLA 期限设为2小时超时未关闭则触发升级—## 五、第三步SLA 监控与升级机制SLA 不是只设一个死线而是要主动检查工单是否超时。 如果超时需要升级告警比如通知运维经理。python# sla_checker.py# 从队列中取出工单检查是否超时超时则升级import jsonimport timeimport redisfrom datetime import datetimer redis.Redis(hostlocalhost, port6379, db0)def check_sla(): 持续检查SLA队列中的工单是否超时 while True: # 阻塞地从队列中取工单 _, ticket_data r.brpop(sla_check_queue, timeout5) if ticket_data: ticket json.loads(ticket_data) deadline datetime.fromisoformat(ticket[sla_deadline]) now datetime.now() if now deadline and ticket[status] open: # 超时升级告警 print(f[SLA超时] {ticket[id]} 已超时! 原定 {deadline}) # 这里可以发送钉钉/邮件给经理 # 同时标记工单为 escalated ticket[status] escalated ticket[escalated_at] now.isoformat() # 重新入队下次检查还能看到 r.lpush(sla_check_queue, json.dumps(ticket)) else: # 未超时重新放回队列或放回延迟队列 # 简单起见我们放回队列并sleep一会儿 time.sleep(30) r.lpush(sla_check_queue, json.dumps(ticket)) time.sleep(1)if __name__ __main__: check_sla()这个模块的难点在于如何避免死循环——如果工单未超时不能立刻再检查否则 CPU 会跑满。 生产环境会使用延迟队列Redis ZSET 或 RabbitMQ 的 TTL这里简化处理。—## 六、第四步工单流转与状态管理工单需要有人来关闭。 我们模拟一个简单的处理流程python# ticket_handler.py# 模拟运维人员处理工单实际通过API或前端操作import jsonimport timeimport redisr redis.Redis(hostlocalhost, port6379, db0)def close_ticket(ticket_id): 关闭工单模拟操作 # 简单起见我们假设工单存储在Redis Hash里 ticket_key fticket:{ticket_id} ticket r.hgetall(ticket_key) if ticket: ticket[status] closed ticket[closed_at] time.time() r.hmset(ticket_key, ticket) print(f[工单关闭] {ticket_id}) else: print(f[错误] 未找到工单 {ticket_id})# 模拟每10秒随机关闭一个工单while True: # 获取所有工单 all_tickets r.keys(ticket:*) if all_tickets: ticket_id random.choice(all_tickets).decode().split(:)[1] close_ticket(ticket_id) time.sleep(10)实际系统中这里应该是运维人员通过 Web 界面点击“处理完成”或者 IOT 设备自动上报修复信号。—## 七、第五步复盘统计有了完整的工单数据复盘就很简单了。 我们可以统计- 每个门店的告警次数- 平均响应时间从告警到工单被认领- 平均修复时间从工单创建到关闭- 按门店/问题类型的分布python# review_report.py# 生成复盘报表示例按门店统计告警次数和平均修复时间import jsonimport redisfrom datetime import datetimer redis.Redis(hostlocalhost, port6379, db0)def generate_report(): 生成简单的复盘报表 report {} all_tickets r.keys(ticket:*) for key in all_tickets: ticket r.hgetall(key) ticket {k.decode(): v.decode() for k, v in ticket.items()} store ticket[store_id] if store not in report: report[store] {count: 0, total_duration: 0, closed_count: 0} report[store][count] 1 if ticket[status] closed and closed_at in ticket and created_at in ticket: created datetime.fromisoformat(ticket[created_at]) closed datetime.fromisoformat(ticket[closed_at]) duration (closed - created).total_seconds() / 3600 # 小时 report[store][total_duration] duration report[store][closed_count] 1 print( 门店运维复盘报表 ) print(f{门店:12} {告警次数:10} {平均修复时间(h):15}) for store, data in sorted(report.items(), keylambda x: x[1][count], reverseTrue): avg data[total_duration] / data[closed_count] if data[closed_count] else 0 print(f{store:12} {data[count]:10} {avg:15.2f})if __name__ __main__: generate_report()这个报表可以直接发给管理层或者集成到 Grafana 面板中。—## 八、总结最小可用系统的核心设计原则| 环节 | 关键设计要点 | 技术选型参考 ||------|-------------|-------------|| 监控 | 阈值可配置、数据采集频率可调 | Redis Pub/Sub, MQTT || 告警 | 分级通知、去重、抑制 | 钉钉/飞书 Webhook || 工单 | 自动创建、分配规则、状态机 | Redis Python dict || SLA | 超时检测、升级机制、延迟队列 | Redis ZSET / RabbitMQ || 复盘 | 聚合统计、趋势分析 | SQLite / Pandas |这套系统的优点- 全部用 Python Redis 实现没有复杂组件- 每层解耦可以单独替换比如把 Redis Pub/Sub 换成 Kafka- 从监控到复盘 5 个环节完整覆盖局限性也就是未来可以加的功能- 缺少持久化存储建议上 SQLite 或 MySQL- 没有 Web 界面可以加 Flask Vue- 告警去重需要额外逻辑如果你正在管理几十到几百家门店这个最小系统完全可以跑起来帮你把“出了事没人管”变成“每个问题都有记录、有跟踪、有结果”。 下次老板问“上周哪家门店问题最多”你直接甩一张复盘报表就完事了。
