这个项目背后其实是我的一个执念数据平台最怕的不是数据不准而是不知道数据什么时候开始不准。之前我维护过好几个业务线的数据看板每天最怕看到的消息就是XX平台的数据怎么停在昨天了等我去查的时候业务同学已经基于旧数据做了好几个决策。后来我干脆自己写了一个叫PLFM_RADAR的轻量级感知系统核心思路就一句话像雷达一样主动发射探测波束去测每一个数据通道而不是等上游出问题之后被动挨打。PLFM_RADAR 全称可以理解为 Platform Radar即平台数据感知雷达。它做的事情不复杂按照事先配置的节奏主动探测所有关联平台的接口连通性、关键指标新鲜度和数据质量然后把探测结果归一化、判定状态、做告警收敛最终推送到对应的通知渠道。这套系统我只用了不到两周就搭出了可用版本后续三个月里不断打磨现在它稳定扛着 12 个平台、36 个数据源的巡检任务日探测量在十万次级别。如果你也维护着多平台数据采集、指标平台或者数据中台这篇文章值得看完。1. 为什么叫雷达从被动等告警到主动测通道1.1 雷达式感知的触发场景先还原一下我最初的痛点。当时公司内部有多个数据看板底层数据来自不同的业务平台有的是直接连对方数据库只读账号有的是调 HTTP 接口拉 JSON有的则是靠对方定时推送文件到 FTP。表面上每个链路都有监控但实际情况是数据库只读账号权限被回收不会有人通知你只有定时任务开始报错你才发现HTTP 接口偶尔返回 200 但 body 里是错误提示连通性一切正常数据其实是脏的对方推送文件经常晚点看板依然展示昨天的数据看起来一切正常上游做了字段类型升级老解析代码直接崩但监控告警只盯着任务是否执行成功没盯着数据内容是否正确。这些都是典型的被动等告警模式的盲区。传统监控关注的是我的任务跑没跑而业务真正关心的其实是数据通道是否健康、数据是否新鲜、数据内容是否正确。PLFM_RADAR 的思路就是把这三件事拆成可探测的指标用主动探测的方式周期性测量。1.2 目标读者与适用边界这套方案最适合的人群是数据平台/中台工程师、数据采集链路的维护者以及需要同时对接多个外部数据源的业务系统负责人。它不追求替代 Prometheus、Zabbix 这类基础设施监控而是专注在业务数据通道的感知这一层。适用边界我也得说清楚对于那些需要毫秒级响应的实时风控场景这个工具不合适它是分钟级巡检的对于已经建设了完善的指标监控、链路追踪的团队它可以作为补充层而不是重复造轮子。PLFM_RADAR 真正的价值是把那些没人愿意天天盯、出了问题影响又很大的数据链路用极低的成本纳入统一的巡检体系。2. 系统整体骨架五层结构与关键选型2.1 核心模块划分PLFM_RADAR 的整体设计是五层结构每层职责单一层与层之间只通过标准数据结构通信。我最初也想过搞复杂的编排引擎规则中心工作流那一套但仔细想想这种场景根本不需要那么重轻量可靠才是第一诉求。层级模块名核心职责调度层scheduler按配置的周期和时区触发生成探测任务采集层collectors适配不同平台协议执行具体探测动作归一化层normalizers把不同来源的原始结果转成统一字段模型判定层rules基于归一化数据判定健康状态分发层dispatchers去重、聚合、限频后推送通知渠道调度层不直接调用采集层而是通过一个内部任务队列解耦。这样做的原因很直接调度层不关心单个探测耗时多长采集层也不关心下一秒要跑哪些任务两边各自演进互不干扰。目录结构方面我采用的是按功能拆包的方案实际项目里长这样plfm_radar/ ├── core/ # 数据结构、配置加载、日志 ├── scheduler/ # 任务编排与调度 ├── collectors/ # 各平台采集适配器 ├── normalizers/ # 归一化与清洗逻辑 ├── rules/ # 状态判定规则 ├── dispatchers/ # 告警收敛与通知分发 ├── storage/ # SQLite/PostgreSQL 存储 └── api/ # 内部查询 API便于看板展示2.2 调度选型与存储设计调度这块我一开始考虑用 APScheduler因为支持 cron 表达式、内存/数据库两种存储方式。但实际用下来发现当探测任务数量上升到几百个之后APScheduler 的默认线程池容易出现任务堆积——某个慢接口把线程占住了后面的任务全部延迟。所以我最后改成了自研的轻量调度器核心是每分钟轮询一次任务到期表到期的任务扔进 asyncio 队列由采集层的协程池消费。如果你不打算维护太多代码APScheduler 依然够用但建议把 executor 换成 ThreadPoolExecutor 并单独设置 max_workes别用默认值。存储方面用小任务量的场景用 SQLite 完全够但我是直接上了 PostgreSQL。原因不是性能而是告警收敛需要频繁做窗口内去重查询、状态查询SQLite 在并发读写下容易出现锁等待。另外探测历史数据我设置了 30 天滚动清理用一张probe_result表记录每次探测的原始返回码、耗时、校验结果方便事后复盘。3. 采集层的实现细节adapter、调度与重试策略3.1 采集适配器基类采集层是 PLFM_RADAR 里工作量最大的部分每接入一个新平台本质上就是写一个 adapter。我定义了一个BaseAdapter基类把通用的生命周期都收敛进去新 adapter 只需要关心三个方法pre_check在探测前做参数校验、probe执行探测、normalize把原始响应转成统一结构。基类代码大概长这样# collectors/base.py from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Any dataclass class ProbeContext: source: str target: str config: dict timeout: float 10.0 dataclass class ProbeResult: source: str target: str ok: bool cost_ms: int status: str data: dict field(default_factorydict) error: str class BaseAdapter(ABC): name base abstractmethod async def probe(self, ctx: ProbeContext) - ProbeResult: pass async def pre_check(self, ctx: ProbeContext) - None: return None async def run(self, ctx: ProbeContext) - ProbeResult: try: await self.pre_check(ctx) start time_monotonic() result await self.probe(ctx) result.cost_ms int((time_monotonic() - start) * 1000) return result except Exception as exc: return ProbeResult( sourcectx.source, targetctx.target, okFalse, cost_ms0, errorf{exc.__class__.__name__}: {exc}, )这套设计的好处是异常兜底、耗时统计、统一状态码转换都在基类完成adapter 里不会散落一堆 try/except。我接入第一个 HTTP 平台用时半天后面再接入同类平台基本一两个小时就搞定。3.2 HTTP 探测与业务探测的两种模式实际操作下来我发现采集适配器可以粗分为两类一类是纯连通性探测就是发一个 HEAD/GET 请求看返回码。这类探测成本低适合高频执行比如每 1 分钟跑一次。但它的缺陷很明显——只能证明接口通着证明不了数据是对的。另一类是业务探测会拉取平台的关键指标比如最新数据的写入时间、今天的订单量、最近一条记录的状态。这类探测能真正反映数据链路是否健康但成本高而且需要针对每个平台写解析逻辑。我的建议是先用高频连通性探测兜底再用低频业务探测校验质量两级搭配。举个具体例子对于外部广告平台的报表接口连通性探测每 2 分钟跑一次业务探测每 30 分钟拉一次最近同步时间如果最近同步时间距离当前时间超过 2 小时就判定为 WARNING超过 4 小时判定为 CRITICAL。这个时间阈值不是拍脑袋定的是根据对方 SLA 承诺的 T1 数据延迟反推出来的。3.3 重试与限频策略采集层最容易翻车的不是写代码而是重试策略。我最早天真地给所有探测任务加了失败重试 3 次结果某天某平台接口抖动所有任务同时重试直接把对方接口打挂了还把自己的出口 IP 搞进了对方的风控名单。那次事故之后我总结了几个原则所有重试必须带指数退避且加随机抖动比如delay min(backoff_base * 2 ** retry_count, 60)再加上random.uniform(0, 5)秒重试只用于瞬时故障超时、连接拒绝、5xx业务校验失败不允许重试因为重试多少次结果都一样每个平台单独配置最大 QPS用信号量控制并发绝不能一股脑全量放行。下面是信号量控制的简化示例# collectors/limiter.py import asyncio class PlatformLimiter: def __init__(self, max_qps: float): self.interval 1.0 / max_qps self._lock asyncio.Lock() self._next_slot 0.0 async def acquire(self): async with self._lock: now loop_time() if now self._next_slot: await asyncio.sleep(self._next_slot - now) self._next_slot max(loop_time(), self._next_slot) self.interval这东西本质上就是一个很朴素的令牌桶不用引入第三方库十几行代码就能把对外的礼貌程度约束住。4. 归一化与实体对齐最容易被低估的一层4.1 统一字段模型刚开始搭 PLFM_RADAR 的时候我心想反正每个 adapter 返回的数据我自己看字段随便定就行。结果第一个平台接入完第二个平台接入到一半我就发现不对了A 平台的data_time是字符串2024-06-11 10:00:00B 平台返回的是毫秒时间戳1718085600000C 平台干脆就不返回时间得从业务含义里自己推断。如果不在采集层之后加一道强制的归一化后面所有规则都要写两套判断逻辑维护成本完全失控。后来我设计了统一字段模型所有 adapter 的原始响应在经过normalize之后都必须转成下面的结构dataclass class NormalizedProbe: source: str target: str probe_type: str # connectivity | business observed_at: int # 统一毫秒时间戳 metric_name: str metric_value: float status_raw: str extra: dict这个模型看起来很死板但它保证了两个关键能力第一判定层可以写完全通用、与具体平台无关的规则第二查询 API 可以统一按source、probe_type、metric_name做聚合分析不需要知道每个平台的原始结构。我甚至把observed_at的时区转换也放在这一层所有内部存储一律用 UTC 毫秒时间戳只有展示层才转本地时区。4.2 解析异常兜底与字段质量检验归一化层还有一项重要任务把数据格式异常和业务异常明确分开。举个例子接口返回{code: 0, data: {value: null}}很多人的第一反应是解析失败其实不然——code0说明接口本身是正常的valuenull是业务数据缺失这在语义上是完全不同的两件事。PLFM_RADAR 的做法是接口层的错误直接标记为UNREACHABLE业务层的字段异常标记为DATA_ABNORMAL两者走不同的判定规则和告警级别。另外字符串数字清洗也是坑非常多的地方。平台返回的数字可能有千分位分隔符1,234,567、全角数字、带单位45ms等各种形态。我写了一个to_float()的兜底函数专门做清洗并且在清洗失败时把原始值原样留在extra.raw_value里方便事后定位。# normalizers/cleaning.py import re def to_float(value): if value is None: return None if isinstance(value, (int, float)): return float(value) if isinstance(value, str): s value.strip() s s.replace(,, ).replace(, ) s s.replace(, %).replace(, ) # 全角数字转半角 s .join(chr(ord(c) - 0xFEE0) if c else c for c in s) # 去掉常见单位 s s.replace(ms, ).replace(毫秒, ).replace(%, ) try: return float(s) except ValueError: return None return None4.3 实体对齐多平台 ID 的映射难题如果说字段模型解决的是格式不统一那实体对齐解决的就是业务主体不统一。举个我实际遇到的例子公司内部商品中心用一套自增 ID外部电商平台的商品 ID 是字符串编码广告平台的 campaign ID 又是另一套体系。如果不做映射即使探测到平台数据异常也很难直接定位到内部业务负责人。我的解法是引入一个轻量的实体映射表本质上就是一张 PostgreSQL 表维护source_type source_id - internal_entity_id的关系。归一化层在数据处理完成后会自动做一次实体映射把target字段转换为内部实体 ID。这样告警推送的时候可以直接按内部组织架构找到责任人。5. 告警收敛的算法细节雷达的准心在抑制误报5.1 三层收敛机制探测系统刚上线那阵告警风暴是我最头疼的问题。某个平台接口抖了 10 分钟因为配置了多级阈值同一时间可能触发 20 条告警直接刷屏。后来我参考了监控领域的常见做法在分发层实现了三层收敛第一层是事件去重。每一条探测结果都会生成一个指纹指纹由source target probe_type 状态码组成。滑动窗口默认 10 分钟内相同指纹只允许生成一条告警事件。指纹不是简单地把几个字段拼起来而是先做归一化再做拼接避免因为空格、大小写、时区格式不同导致误判。第二层是告警聚合。可以理解为把同一时间段内、同一实体或同一平台的多条告警打包成一条摘要。比如某平台 3 个接口同时超时最终只推一条XX平台 3 个接口在 16:00-16:10 期间超时而不是三条单独告警。聚合规则是可配置的可以按source聚合也可以按internal_entity_id聚合。第三层是通知限频。即便经过了去重和聚合某些持续告警依然会高频到达。我实现了两个简单但有效的限频策略静默期同一指纹告警后 30 分钟内不再重复通知和指数退避如果告警持续未恢复通知间隔从 5 分钟逐步拉长到 60 分钟。# dispatchers/convergence.py import time from hashlib import sha1 class EventDeduplicator: def __init__(self, window_seconds: int 600): self.window window_seconds self._seen {} def _fingerprint(self, event: dict) - str: raw f{event[source]}|{event[target]}|{event[probe_type]}|{event[status]} return sha1(raw.encode(utf-8)).hexdigest() def is_duplicate(self, event: dict) - bool: fp self._fingerprint(event) now time.time() last self._seen.get(fp) if last is None or now - last self.window: self._seen[fp] now return False return True这套三层机制上线后告警量直接降了一个数量级而真正的关键告警一条都没漏。我个人的经验是告警系统的设计目标不是多告警、全覆盖而是关键时刻一定响平时绝不乱响——雷达准心对准的靶子比雷达的扫描范围更重要。5.2 状态机判定减少抖动还有一个容易被忽略的细节状态变化抖动。假设某个探测点状态在 WARNING 和 OK 之间反复横跳如果每次变化都发告警接收人很快就会把告警渠道静音。PLFM_RADAR 引入了状态机机制每个探测目标维护一个迷你状态机UNKNOWN - OK - WARNING - CRITICAL ^ | --------- | RECOVERED关键规则是从 OK 到 WARNING 必须连续两次探测异常才迁移避免单次抖动触发从 WARNING/FATAL 恢复到 OK 时要发一条恢复通知但恢复通知的优先级低于异常告警。这个连续两次的阈值不建议调成一次就切实测下来一次切换太敏感半夜经常被很小的毛刺吵醒。6. 一次典型的排障复盘数据停了但没收到告警6.1 排查链路完整回顾讲一个让我印象深刻的真实案例某平台看板数据停留在昨天但 PLFM_RADAR 没有发出任何告警。当时我的第一反应是探测系统是不是也挂了于是按下面的链路逐步排查第一步确认探测任务本身在跑。我查了probe_result表发现该平台每 10 分钟都有新的记录说明调度和采集都正常。第二步看探测结果的状态码全是 OK没有超时和 5xx。这就奇了怪了数据通道明明是通的。第三步我看了归一化层的metric_value发现一直不为空但是数值和昨天一模一样——不是数据停了是数据没有更新而我没有对数据是否变化这个特征做判断。根因其实在这里这个平台的数据是每天整点由上游任务生成某天上游任务静默失败平台侧没有任何报错接口依然返回昨天的数据。我的业务探测只校验了接口能通、返回值存在、格式正确并没有校验数据新鲜度和数据是否变化。所以数据停在昨天从探测系统视角看完全是健康的。6.2 修复方案与后续加固那次事故之后我给业务探测加了一个新指标data_staleness也就是最近一次数据时间与当前时间的间隔。同时给所有数据每天应该更新的探测点配置了新鲜度规则数据时间晚于调度时间 1 小时以上判为 WARNING超过 2 小时未更新判为 CRITICAL。另外我还加了一条更隐蔽的检查对连续两次探测到的关键指标进行 diff如果间隔周期内应该变化但数值完全相同即便没有显式的数据时间字段也要触发告警。这就是那次事故带给我的实际经验——监控系统真正要盯的不是系统有没有报告而是业务数据有没有如约流转。7. 压测数据、最终效果与后续演进7.1 实际运行指标PLFM_RADAR 上线运行半年多之后我统计过一组实际数据。接入规模是 12 个平台、36 个数据源配置了 130 多个探测任务。其中连通性探测占 60%业务探测占 40%。高峰期每 10 秒会产生约 8 个并发探测请求。进程本身跑在一台 2C4G 的容器里内存稳定在 350MB 左右CPU 平均使用率不到 10%。整体效果方面我重点看两个指标漏报率和误报率。漏报率从原来有问题靠人发现的不可量化状态降到了测试期 23 个故障全部有告警误报率经过告警收敛优化后从最开始的每天几十条降到了平均每天不超过 3 条。这个指标对于告警接收人至关重要——误报一多再响的雷达也没人听。值得特别说的是整体维护成本并不高。新接入一个标准 HTTP 平台的耗时基本控制在 2 小时内其中大部分时间是花在写业务探测的字段解析上连通性探测基本 20 分钟就能接入。7.2 后续演进方向这个项目目前还在迭代我自己的演进计划大致有三个方向第一个方向是配置热更新。现在添加新探测任务要改 YAML 配置并重启进程虽然频率不高但每次重启都会丢若干秒的探测窗口。后续打算用配置中心比如 Nacos 或 Consul做配置动态下发并且保留配置版本管理改完能回滚。第二个方向是自定义判定脚本。现在规则层是内置的未来打算允许用户写一段简单的 Python 脚本作为判定逻辑这样遇到需要结合多指标判断的场景就不用改主程序了。第三个方向是联邦部署。也就是每个业务线各自部署一套 PLFM_RADAR数据统一汇聚到上层做全局视图这样既保证了各团队的自主性又保留了整体的可观测性。7.3 最后分享一点个人的实操体会整个项目做下来如果说只能总结一条经验那就是做数据感知系统最难的不是写探测器而是决定什么才算异常。字段格式、时区、ID 映射、去重阈值这些都有标准答案唯独异常的定义没有它完全取决于你对业务链路的理解深度。而只有当你对每条链路的正常形态足够了解才有可能设计出既不吵人、又不漏报的告警规则。我最后还留了一个小技巧给看到这里的朋友在写任何探测任务的 YAML 配置时一定要加一个description字段写清楚这个探测任务为什么存在、当时遇到了什么问题才加的。三个月后你回来维护配置的时候会感激当初写下这句话的自己。