★ FEATURED ARTICLE 刚把 PLFM_RADAR 这套东西调稳趁热写一篇记录。PLFM_RADAR 说白了就是一套平台雷达系统。名字的拆法是 PLFMPlatform RADAR核心职责是把分布在各个环境里的服务、接口、云资源当成一个个需要持续追踪的信号源定时主动扫一圈发现新增的、消失的、异常的、漂移的及时把告警推到钉钉或者企业微信。它在我们团队里的角色有点像小区门口的保安加监控探头组合保安定时巡逻探头全天盯着哪个角落多了陌生人、哪扇窗户没关、哪条过道异常拥堵都能第一时间发现。这篇文章我会把手感、坑点、核心参数和代码骨架都摊开讲适合正在搭服务监控、做资产梳理、或者被割裂环境搞到头大的运维和开发同学。不是广告贴纯项目复盘。1. 项目概述PLFM_RADAR 要解决什么问题1.1 为什么需要一台“平台雷达”先说背景。我们团队从单体应用拆成微服务之后遇到的第一个麻烦不是代码而是“搞不清线上到底长了什么”。服务一多环境一多配置一多光靠脑子记根本不现实。举几个真实场景有人上线了一个新接口但忘了在文档里登记直到别的团队来问“这个端口是哪个服务的”才发现漏了某个测试环境的 Redis 突然对外暴露了一个非默认端口没人记得是谁改的凌晨流量高峰时有个实例的响应时间翻了三倍但监控大盘上并没有破阈值因为阈值是拿正常时期的数据写的压根没覆盖这种“缓慢性劣化”。PLFM_RADAR 就是冲这些痛点去的。它不做全链路追踪也不替代 Prometheus它只做一件事持续感知平台上的“信号变化”。新增服务、消失端口、异常访问、配置漂移、隐藏依赖这些东西如果等人工去发现黄花菜都凉了。我的设计出发点很简单与其在问题发生后花两小时查日志不如让系统每五分钟主动扫一遍把所有变化统一收敛到一张变动清单里。这个思路后来被验证是对的很多问题在用户反馈之前就被雷达提前拍下来了。1.2 它到底是什么以及不做什么PLFM_RADAR 的核心定位是“平台侧的健康雷达”不是业务监控也不是调用链追踪。它更像一个巡检员走固定的路线看固定的点位记录每次看到的状态把不寻常的变化拎出来。具体而言扫描对象分成三类基础设施各环境的主机存活、端口开放、关键进程状态。服务层服务注册中心里的实例数量、健康检查接口返回码、响应耗时。边界层对外暴露的 API 入口、网关路由规则、DNS 解析记录。举个例子服务注册中心里的实例数从 6 变成 3Prometheus 的告警可能要等一段时间才触发但 PLFM_RADAR 在下一轮巡检时就能发现并且还能主动把“变化前后”的清单打出来直接定位是哪个节点的哪个服务掉了。那些常规监控系统做不了的事它就用最简单的方式做扫记录对比告警。我必须强调它不碰业务日志内容也不分析用户敏感数据只做端口连通性、版本标识、响应状态这类“元信息”级别的采集。这样既能控制权限范围也避免了很多合规上的麻烦。我们内部叫做“只看包装盒不拆里面”。2. 核心架构与设计取舍2.1 三个核心模块一条流水线PLFM_RADAR 的架构拆开来看就三层采集器、分析器、通知器。采集器负责定时执行扫描任务。它拿着一个配置好的“探针清单”去各环境检查 TCP 端口、HTTP 接口、进程状态等把原始结果写进存储层。这里不搞 agent 模式用的是无代理扫描只从一台跳板机发请求减少部署面。分析器是核心。它会把本次采集结果和上一次的结果做 diff同时结合配置的基线规则做判断。举个例子“端口 3306 从扫描结果里消失了”这种情况分析器会判断是服务正常下架还是异常宕机依据是服务注册表里的状态。通知器负责把分析结果推送出去。支持钉钉、企业微信、邮件三种通道告警内容包含变化的资源标识、变化前状态、变化后状态、持续时间等字段。为了不炸群每个告警有 15 分钟的静默窗口防止重复提醒。三条模块之间的数据流转我用了一个很朴素的模型采集器把结果落库分析器定时捞取比对通知器只在“差异集”上工作。这个模型没有用消息队列也没搞流式计算因为扫描任务本身是分钟级别的低频操作引入 Kafka 纯粹是给自己挖坑。2.2 关键选型与妥协点为什么用 Python 而不是 Go团队里 Python 基础好开发迭代快。说实话单纯做端口扫描性能差距在这个量级根本看不出来我们单轮扫描也就 2000 多个目标点Python 并发跑完不到 20 秒。如果目标是十万级资产那我肯定劝你上 Go 或 Rust但中小规模团队用 Python 完全够了。为什么用推模式而不是拉模式最早考虑过所有目标主动上报的模式后来放弃了。因为很多老系统根本不会给你加 agent 的机会推模式要求目标端配合这在跨团队场景下推进阻力很大。拉模式只需要扫描端有权限访问目标端口就行省去了大量说服工作。为什么没用现成的开源扫描器我调研过几个比如 Nmap 的定时任务方案还有某些资产测绘工具。问题在于它们要么太重要么太偏安全扫描。我需要的是一个能自定义基线、能对接内部告警渠道、能留存历史状态的工具这种“平台管理视角”的扫描器开源社区还真没有完全贴合的。技术选型上的取舍列表如下模块选型理由采集器Python asyncio aiohttp异步并发效率高代码量少配置管理YAML 文件运维同学改起来方便不用重启编译状态存储SQLite RedisSQLite 存历史快照Redis 存去重缓存任务调度APScheduler轻量级支持 cron 表达式通知Webhook SMTP钉钉、企业微信、邮件三通道这套组合的总体感觉是“够用且好修”。没有任何炫技的成分每加一个组件我都要先问自己一句这玩意儿挂了我能不能五分钟内修好如果答案是不确定就不加。3. 实操过程从零搭一台能用的 PLFM_RADAR3.1 环境准备与目录结构部署环境我用了一台 2C4G 的 Linux 虚拟机系统是 CentOS 7.9Python 版本 3.9。其实资源很宽裕因为扫描任务本身不占多少 CPU真正耗内存的是后续数据积累后的 SQLite 查询不过这是后话。目录结构我按模块拆的没搞花活plfm_radar/ ├── config/ │ ├── targets.yaml # 扫描目标配置 │ ├── policies.yaml # 告警策略配置 │ └── settings.py # 全局参数 ├── core/ │ ├── scanner.py # 采集器模块 │ ├── analyzer.py # 分析器模块 │ ├── notifier.py # 通知器模块 │ └── scheduler.py # 调度模块 ├── storage/ │ ├── snapshot.db # SQLite 快照库 │ └── redis_client.py # Redis 封装 ├── logs/ │ └── radar.log # 运行日志 ├── requirements.txt └── main.py # 入口依赖安装没什么特殊之处。requirements.txt 里核心就四个库aiohttp、apscheduler、redis、pydantic。不需要装 Nmap因为扫描逻辑我直接用 Python 的 asyncio 写省掉外部工具依赖降低部署成本。装依赖这一步给新手提个醒pip install前先建虚拟环境别图省事往系统 Python 里装。踩过一次坑系统自带的 Python 环境被搞坏了后来 rebuild 了一次服务器才恢复。3.2 目标配置的四个关键参数targets.yaml 是整个系统的输入源头它的设计直接决定雷达扫得准不准。先看一个精简示例version: 1 checkpoints: - name: main-gateway-prod group: api-gateway type: http host: gw.internal.example.com path: /healthz port: 8080 timeout: 3 interval: 60 expected_status: 200 - name: mysql-core-prod group: database type: tcp host: 10.10.2.15 port: 3306 timeout: 2 interval: 120 - name: user-service-registered-check group: service-discovery type: consul url: http://consul.internal.example.com/v1/health/service/user-service timeout: 5 interval: 60 expected_count: 4每个检查点有四个关键参数type、interval、timeout、expected_*。type决定采集器用什么方式探测。http 类型会发起 GET 请求检查返回状态码和响应时间tcp 类型只做端口连通性检测consul 类型会调服务注册中心的 API数注册实例数量。第一次做的时候我把所有目标都配成 tcp 类型结果很多 HTTP 服务的 503 状态码根本没被发现后来才补齐了类型分类。interval是扫描间隔单位秒。这不是一个可以随便填的参数它直接决定发现异常的延迟。我们的经验值核心入口 60 秒数据库 120 秒边缘服务 300 秒。不要全都设成 30 秒扫描频率提上去Redis 连接数和 SQLite 写入量都会上涨但异常发现速度并不会因此有质的提升。timeout是单次探测超时时间。这个参数一定要小于interval否则任务会堆积。如果扫描目标之间存在依赖关系——比如一个服务在等待目标 A 的响应——超时设置不合理会把整个扫描链路拖垮。expected_*是基线参数不同类型含义不同。HTTP 类型对应expected_statusTCP 类型没有额外期望值Consul 类型对应expected_count。我在上线初期犯过一个错把一个预期实例数为 4 的服务配成 2结果每次扫描都误报“实例数减少”排查半天才发现是基线值写错了。3.3 采集器代码骨架采集器的核心就是并发探测。我用 asyncio 写了一个轻量并发池控制每秒最大发出的探测请求数避免对老旧的中间件造成冲击。import asyncio import aiohttp import socket import time from typing import List, Dict SEMAPHORE asyncio.Semaphore(50) # 并发控制 TIMEOUT_AIOHTTP 3 async def check_http(checkpoint: Dict) - Dict: url fhttp://{checkpoint[host]}:{checkpoint[port]}{checkpoint.get(path, /)} start time.time() async with SEMAPHORE: try: async with aiohttp.ClientSession( timeoutaiohttp.ClientTimeout(totalTIMEOUT_AIOHTTP) ) as session: async with session.get(url) as resp: return { name: checkpoint[name], status: up, http_status: resp.status, latency_ms: round((time.time() - start) * 1000, 2), ts: int(time.time()), } except Exception as e: return { name: checkpoint[name], status: down, error: str(e), ts: int(time.time()), } async def check_tcp(checkpoint: Dict) - Dict: start time.time() async with SEMAPHORE: loop asyncio.get_event_loop() try: observer socket.socket(socket.AF_INET, socket.SOCK_STREAM) observer.settimeout(2) # 利用线程池做阻塞式 connect await loop.run_in_executor(None, observer.connect, (checkpoint[host], checkpoint[port])) observer.close() return { name: checkpoint[name], status: up, latency_ms: round((time.time() - start) * 1000, 2), ts: int(time.time()), } except Exception as e: return { name: checkpoint[name], status: down, error: str(e), ts: int(time.time()), }这段代码有几个地方值得说明。用asyncio.Semaphore(50)做并发限制是为了防止扫描目标中的某个网段突然全部不可达时扫描器自身被拖垮。如果不限制并发当某个机房交换机挂了所有 TCP 探测都会阻塞到超时并发协程数量会飙升到几千那台跳板机大概率直接 OOM。TCP 探测用loop.run_in_executor而不是直接发非阻塞 connect是因为 asyncio 本身没有内置纯异步的 socket connect 操作用线程池是最省事的办法。虽然每个连接会占一个线程但并发 50 的场景完全没有问题。HTTP 探测用 aiohttp 是顺手的事团队本来就熟。你也可以用 httpx也是一样的效果。这里的核心思路就一句话所有的探测只输出一个结果字典不做任何业务判断。判断全部丢给分析器去做这样采集器就变成一个完全无状态的无脑探测器出问题最好排查。3.4 分析器差异就是问题分析器比采集器难写难点在于“如何判断一个变化是异常还是正常操作”。我的方案是三层判断第一层是硬性存活判断。只要 TCP 连不上、HTTP 返回非预期状态码直接标记为故障。第二层是状态差异判断。和上一次快照比如果实例数量减少、端口消失、配置发生变化进入待观察队列。第三层是基线判断。只有在配置了明确预期值时才触发比如 Consul 服务实例数应该为 4实际扫到 2直接告警。import sqlite3 import json from datetime import datetime class Analyzer: def __init__(self, db_path: str): self.db sqlite3.connect(db_path, check_same_threadFalse) self._init_db() def _init_db(self): with self.db: self.db.execute( CREATE TABLE IF NOT EXISTS checkpoint_state ( name TEXT PRIMARY KEY, snapshot TEXT, updated_at INTEGER ) ) def analyze_result(self, checkpoints: list) - list: 返回需要告警的差异项列表 alerts [] for cp in checkpoints: name cp[name] snapshot json.dumps(cp, sort_keysTrue) row self.db.execute( SELECT snapshot FROM checkpoint_state WHERE name ?, (name,) ).fetchone() if row is None: # 首次采集只记录不告警 with self.db: self.db.execute( INSERT INTO checkpoint_state (name, snapshot, updated_at) VALUES (?, ?, ?), (name, snapshot, int(datetime.now().timestamp())), ) continue prev json.loads(row[0]) if self._is_significant_change(prev, cp): alerts.append({ checkpoint: name, old: prev, new: cp, time: int(datetime.now().timestamp()), }) with self.db: self.db.execute( UPDATE checkpoint_state SET snapshot ?, updated_at ? WHERE name ?, (snapshot, int(datetime.now().timestamp()), name), ) return alerts def _is_significant_change(self, prev: dict, new: dict) - bool: if prev.get(status) ! new.get(status): return True if new.get(http_status) and prev.get(http_status) ! new.get(http_status): return True if new.get(latency_ms) and prev.get(latency_ms): # 响应耗时翻倍且超过 500ms才告警 if new[latency_ms] 500 and new[latency_ms] prev[latency_ms] * 2: return True return False这里最重要的一点是首次采集不告警。一定要加这个逻辑否则系统刚上线时所有检查点都是第一次写入没有快照作对比会把所有目标都报一遍“异常”。我们当时真遇到过这个问题上线当天群里被刷屏了。另一个细节是_is_significant_change里对延时变化的判断不光是平均值要看翻倍情况。因为网络抖动导致单次延时从 200ms 变到 400ms 很常见如果 400ms 超过阈值就告警那么一天能收到几十条。加上“翻倍且大于绝对阈值”的条件噪音就小很多。3.5 调度模块与启动调度用的 APScheduler配置简单明了from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger def load_checkpoints(): # 读取 YAML config/targets.yaml返回检查点列表 pass def run_scan_job(): checkpoints load_checkpoints() loop asyncio.new_event_loop() asyncio.set_event_loop(loop) results loop.run_until_complete(scan_all(checkpoints)) alerts Analyzer(storage/snapshot.db).analyze_result(results) if alerts: Notifier().send(alerts) scheduler BlockingScheduler() scheduler.add_job( run_scan_job, CronTrigger.from_crontab(*/1 * * * *), # 每分钟跑一次 max_instances1, coalesceTrue, ) scheduler.start()这里我踩过一个重要的坑max_instances1和coalesceTrue必须同时设置。max_instances1保证上一个扫描任务没结束时下一个任务不会启动避免同一时间出现多个扫描实例互相推挤coalesce保证如果扫描任务在上一个周期积压了跳过的周期就直接合并掉不搞补跑。如果没有这两个参数当某次全链路扫描因为某个目标 IP 卡到超时任务执行时间超过 60 秒下一轮任务就会叠加启动。我实测过一次最严重的情况现场堆了 8 个扫描实例Redis 连接数直接打满数据库被锁。后来加了这两个参数就再也没出现过。启动命令很简单# Python 3.9 python main.py日志输出到 logs/radar.log默认按天切割。运营初期建议开着终端前台跑观察几轮扫描日志的节奏确认没有异常再丢到 systemd 或 supervisor 托管。4. 告警策略与噪音调优4.1 控制台和告警分级一开始我把所有异常都设为同一等级结果就是告警疲劳几分钟不刷群就一堆。后来我加了分级不同等级对应不同的提醒方式和处理时效。等级触发条件提醒方式预期响应时间P0核心服务实例数降为 0、主数据库端口失联钉钉电话群消息立即P1核心接口 HTTP 5xx、实例数减半群消息 责任人5 分钟P2响应延时翻倍超阈值、非核心端口变化群消息工作日处理P3配置漂移、信息变更记录到日报、不实时推送有空再看分级实现不复杂在policies.yaml里给每个检查点打标签例如policies: main-gateway-prod: level: P0 channels: [dingtalk, phone] mysql-core-prod: level: P0 channels: [dingtalk] user-service-registered-check: level: P1 channels: [dingtalk]channels指定告警通道。P0 可以额外接电话提醒P1 只发群消息P2/P3 则进入日报汇总。分级这一套做完之后最大的效果是群里 P0 告警刷屏的问题彻底消失了值班的同学终于能睡得着了。4.2 噪音来源与降噪三板斧告警噪音主要来自三个地方首次采样的基线噪声、瞬时故障误报、重复提醒轰炸。对应的三个处理方法是首采静默。之前已经说过首次采集只建快照不告警。这个逻辑必须在系统上线前就写好。连续失败 N 次才告警。单次探测失败不代表服务真的挂了也许是网络闪断。我在分析器里加了“连续失败次数”字段默认 N3。也就是说一个检查点连续三轮扫描都失败才真正进入告警清单。这样能过滤掉大量的偶发网络抖动。去重静默窗口。同一个检查点告警后15 分钟内部再重复推送。这是因为服务可能持续处于异常状态每分钟扫描都发现故障每分钟都给群里发一条消息只会添乱。静默窗口期内系统只记录状态不推送。这三个方法叠加起来告警量能下降 70% 左右。我拿真实数据对比过调优前一周告警 286 条调优后一周 43 条其中还包含了不少真实故障没有被滤掉。唯一要小心的是降噪措施不能掩盖真实问题所以连续失败到 P0 级别的告警我特意不做静默去重。宁可多一条消息也不能让真正严重的问题被静默吞掉。4.3 基线漂移检测的思路最后一类告警不是故障而是“变化”。这类告警检查的不是服务是否可用而是系统配置、开放端口、依赖关系是否和预期一致。举个例子一个内部管理后台原本只监听内网 IP突然有一天发现它的端口对外开放了。这在平台层面可能是一个严重的安全隐患。PLFM_RADAR 的思路是把端口开放状态记录到快照里如果新增了一个从未见过的对外端口就会触发“未登记端口出现”的告警。再看配置漂移一个服务在多个环境里的版本号不一致可以通过执行version接口去比对。雷达定时抓取版本号如果 prod 环境和 staging 环境的版本号漂移过大也会给发布负责人发一条提醒。这些不是纯技术故障而是偏管理侧的异常。实际用下来这一类告警反而最受领导认可因为很多隐患都藏在看不到的变化里雷达能把这些变化用最朴素的方式翻出来。5. 常见问题与排查速查5.1 五个高频问题实录问题 1扫描任务卡死不动了。现象是所有检查点都显示“上次扫描时间”停留在几分钟甚至几小时前日志里没有任何异常输出。排查后一般是某个 TCP 探测阻塞在 connect 上且超时时间设置过长了。解决方法检查超时参数不能让连接一直卡住把socket.settimeout和aiohttp.ClientTimeout都显式设置不要依赖系统默认值。问题 2Redis 连接数飙高。现象是 Redis 的 connected_clients 超过了 100。原因是每次扫描结果都新建一个 Redis 连接没有复用。后来改成全局维护一个 Redis 连接池问题就解决了。问题 3SQLite 数据库锁死。排查发现分析器和采集器同时写入了数据库SQLite 不支持并发的写锁导致长时间等待。解决方法是用单一线程写入分析器只在主调度流程里执行不额外开线程。问题 4告警延迟了 10 多分钟才发现。从配置看interval是 60 秒但实际大约每 5 分钟才收到告警。后来发现是因为扫描目标太多单轮扫描一次要 3 分钟加上调度器等待下一个周期窗口实际发现延迟被拉长了。解决方案是把检查点按组拆开三个组并行扫描每组配置独立调度器。问题 5升级后配置格式不兼容启动报错。这纯粹是配置版本演进的问题。给 targets.yaml 加了version字段启动时校验版本号和代码要求的版本不一致就直接拒绝启动。这个保护机制看起来简单但能省掉无数排查“为什么配置生效了又像没生效”的痛苦。5.2 排查思路从日志反推卡点这可能是全文最实用的一段。当我收到“雷达不工作了”的反馈第一件事永远不是看配置而是去 logs/radar.log 尾部看最近几行。我用一个极简的日志约定每条扫描任务开始输出SCAN_START任务结束输出SCAN_DONE每次告警输出ALERT_TRIGGERED每次异常输出EXCEPTION_TRACE。不管是什么问题只要顺着这四个关键词去 grep总能很快定位到弱点在哪了。特别是在排查扫描卡死时如果日志里一直有SCAN_START但迟迟没有SCAN_DONE那答案基本锁定在采集器某次探测上。把这个检查点单独拿出来跑一遍问题就暴露了。5.3 怎么扩展成平台资产地图写到这里分享一个扩展思路PLFM_RADAR 长期运行后快照库里会积累大量“资源状态的历史时间线”。这些数据本身就是一份活资产地图完全可以拿来二次开发。比如可以做一个简单的前端页面按资源组展示最新扫描状态、历史变动时间线、告警记录。这比任何 CMDB 都准因为它是通过主动扫描得到的“实际状态”而不是人工填写的“以为的状态”。还可以接入发布系统每次发布前自动对比发布前后的服务版本、实例数量、开放端口生成一条变更摘要。这样发布验收就不再依赖人工去点“检查通过”雷达已经替你把变化量算好了。这个工具最大的价值不在于告警本身而在于它把“平台状态”从一堆离散的、需要到处问人的黑盒变成了一张看得见的变化地图。每一条快照都是一个脚印历史回溯、变更追溯都变得有据可依。工具会持续打磨但核心思路基本定了扫描、记录、对比、发现差异这就是一个雷达该做的事。
阅读完成 · 觉得有帮助？
