刚把 PLFM_RADAR 这套东西调稳，趁热写一篇记录。

PLFM_RADAR 说白了就是一套平台雷达系统。名字的拆法是 PLFM（Platform）+ RADAR，核心职责是把分布在各个环境里的服务、接口、云资源当成一个个需要持续追踪的信号源，定时主动扫一圈，发现新增的、消失的、异常的、漂移的，及时把告警推到钉钉或者企业微信。

它在我们团队里的角色，有点像小区门口的保安加监控探头组合：保安定时巡逻，探头全天盯着，哪个角落多了陌生人、哪扇窗户没关、哪条过道异常拥堵，都能第一时间发现。

这篇文章我会把手感、坑点、核心参数和代码骨架都摊开讲，适合正在搭服务监控、做资产梳理、或者被割裂环境搞到头大的运维和开发同学。不是广告贴，纯项目复盘。

1. 项目概述：PLFM_RADAR 要解决什么问题

1.1 为什么需要一台“平台雷达”

先说背景。我们团队从单体应用拆成微服务之后，遇到的第一个麻烦不是代码，而是“搞不清线上到底长了什么”。服务一多，环境一多，配置一多，光靠脑子记根本不现实。

举几个真实场景：有人上线了一个新接口，但忘了在文档里登记，直到别的团队来问“这个端口是哪个服务的”才发现漏了；某个测试环境的 Redis 突然对外暴露了一个非默认端口，没人记得是谁改的；凌晨流量高峰时有个实例的响应时间翻了三倍，但监控大盘上并没有破阈值，因为阈值是拿正常时期的数据写的，压根没覆盖这种“缓慢性劣化”。

PLFM_RADAR 就是冲这些痛点去的。它不做全链路追踪，也不替代 Prometheus，它只做一件事：持续感知平台上的“信号变化”。新增服务、消失端口、异常访问、配置漂移、隐藏依赖，这些东西如果等人工去发现，黄花菜都凉了。

我的设计出发点很简单：与其在问题发生后花两小时查日志，不如让系统每五分钟主动扫一遍，把所有变化统一收敛到一张变动清单里。这个思路后来被验证是对的，很多问题在用户反馈之前就被雷达提前拍下来了。

1.2 它到底是什么，以及不做什么

PLFM_RADAR 的核心定位是“平台侧的健康雷达”，不是业务监控，也不是调用链追踪。它更像一个巡检员：走固定的路线，看固定的点位，记录每次看到的状态，把不寻常的变化拎出来。

具体而言，扫描对象分成三类：

• 基础设施：各环境的主机存活、端口开放、关键进程状态。
• 服务层：服务注册中心里的实例数量、健康检查接口返回码、响应耗时。
• 边界层：对外暴露的 API 入口、网关路由规则、DNS 解析记录。

举个例子，服务注册中心里的实例数从 6 变成 3，Prometheus 的告警可能要等一段时间才触发，但 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 的定时任务方案，还有某些资产测绘工具。问题在于它们要么太重，要么太偏安全扫描。我需要的是一个能自定义基线、能对接内部告警渠道、能留存历史状态的工具，这种“平台管理视角”的扫描器，开源社区还真没有完全贴合的。

技术选型上的取舍列表如下：

这套组合的总体感觉是“够用且好修”。没有任何炫技的成分，每加一个组件我都要先问自己一句：这玩意儿挂了，我能不能五分钟内修好？如果答案是不确定，就不加。

3. 实操过程：从零搭一台能用的 PLFM_RADAR

3.1 环境准备与目录结构

部署环境我用了一台 2C4G 的 Linux 虚拟机，系统是 CentOS 7.9，Python 版本 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_status，TCP 类型没有额外期望值，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 = f"http://{checkpoint['host']}:{checkpoint['port']}{checkpoint.get('path', '/')}" start = time.time() async with SEMAPHORE: try: async with aiohttp.ClientSession( timeout=aiohttp.ClientTimeout(total=TIMEOUT_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_thread=False) 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_keys=True) 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_instances=1, coalesce=True, ) scheduler.start()

这里我踩过一个重要的坑：max_instances=1和coalesce=True必须同时设置。

max_instances=1保证上一个扫描任务没结束时，下一个任务不会启动，避免同一时间出现多个扫描实例互相推挤；coalesce保证如果扫描任务在上一个周期积压了，跳过的周期就直接合并掉，不搞补跑。

如果没有这两个参数，当某次全链路扫描因为某个目标 IP 卡到超时，任务执行时间超过 60 秒，下一轮任务就会叠加启动。我实测过一次最严重的情况：现场堆了 8 个扫描实例，Redis 连接数直接打满，数据库被锁。后来加了这两个参数，就再也没出现过。

启动命令很简单：

# Python 3.9 python main.py

日志输出到 logs/radar.log，默认按天切割。运营初期建议开着终端前台跑，观察几轮扫描日志的节奏，确认没有异常再丢到 systemd 或 supervisor 托管。

4. 告警策略与噪音调优

4.1 控制台和告警分级

一开始我把所有异常都设为同一等级，结果就是告警疲劳，几分钟不刷群就一堆。后来我加了分级，不同等级对应不同的提醒方式和处理时效。

分级实现不复杂，在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 次才告警。单次探测失败不代表服务真的挂了，也许是网络闪断。我在分析器里加了“连续失败次数”字段，默认 N=3。也就是说，一个检查点连续三轮扫描都失败，才真正进入告警清单。这样能过滤掉大量的偶发网络抖动。

去重静默窗口。同一个检查点告警后，15 分钟内部再重复推送。这是因为服务可能持续处于异常状态，每分钟扫描都发现故障，每分钟都给群里发一条消息只会添乱。静默窗口期内，系统只记录状态，不推送。

这三个方法叠加起来，告警量能下降 70% 左右。我拿真实数据对比过：调优前一周告警 286 条，调优后一周 43 条，其中还包含了不少真实故障，没有被滤掉。

唯一要小心的是，降噪措施不能掩盖真实问题，所以连续失败到 P0 级别的告警，我特意不做静默去重。宁可多一条消息，也不能让真正严重的问题被静默吞掉。

4.3 基线漂移检测的思路

最后一类告警不是故障，而是“变化”。这类告警检查的不是服务是否可用，而是系统配置、开放端口、依赖关系是否和预期一致。

举个例子：一个内部管理后台原本只监听内网 IP，突然有一天发现它的端口对外开放了。这在平台层面可能是一个严重的安全隐患。PLFM_RADAR 的思路是把端口开放状态记录到快照里，如果新增了一个从未见过的对外端口，就会触发“未登记端口出现”的告警。

再看配置漂移：一个服务在多个环境里的版本号不一致，可以通过执行version接口去比对。雷达定时抓取版本号，如果 prod 环境和 staging 环境的版本号漂移过大，也会给发布负责人发一条提醒。

这些不是纯技术故障，而是偏管理侧的异常。实际用下来，这一类告警反而最受领导认可，因为很多隐患都藏在看不到的变化里，雷达能把这些变化用最朴素的方式翻出来。

5. 常见问题与排查速查

5.1 五个高频问题实录

问题 1：扫描任务卡死，不动了。现象是所有检查点都显示“上次扫描时间”停留在几分钟甚至几小时前，日志里没有任何异常输出。排查后一般是某个 TCP 探测阻塞在 connect 上，且超时时间设置过长了。解决方法：检查超时参数不能让连接一直卡住，把socket.settimeout和aiohttp.ClientTimeout都显式设置，不要依赖系统默认值。

问题 2：Redis 连接数飙高。现象是 Redis 的 connected_clients 超过了 100。原因是每次扫描结果都新建一个 Redis 连接，没有复用。后来改成全局维护一个 Redis 连接池，问题就解决了。

问题 3：SQLite 数据库锁死。排查发现分析器和采集器同时写入了数据库，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 都准，因为它是通过主动扫描得到的“实际状态”，而不是人工填写的“以为的状态”。

还可以接入发布系统：每次发布前自动对比发布前后的服务版本、实例数量、开放端口，生成一条变更摘要。这样发布验收就不再依赖人工去点“检查通过”，雷达已经替你把变化量算好了。

这个工具最大的价值不在于告警本身，而在于它把“平台状态”从一堆离散的、需要到处问人的黑盒，变成了一张看得见的变化地图。每一条快照都是一个脚印，历史回溯、变更追溯都变得有据可依。工具会持续打磨，但核心思路基本定了：扫描、记录、对比、发现差异，这就是一个雷达该做的事。