监控覆盖率怎么算：多工具、多集群下的本体层抽象

先看三行数字。它们来自三个监控工具各自的概览页（这是拼出来的一个典型场景，数字是示意）：

Prometheus   集群 prod-sh、prod-bj    抓取目标 1,284   up=1 的 1,284   覆盖率 100%
Zabbix       主机组 IDC-物理机        主机 312         agent 可用 309  覆盖率 99.0%
Flink 平台   实时作业                 作业 86          已自动接入 86   覆盖率 100%

同一天早上，数仓的人发现订单宽表 dwd_order_wide 从凌晨 02:10 起就没再写入，下游的报表空了 6 个小时，没有任何告警。写这张表的 Flink 作业 ods_order_etl 一切正常：没有重启，checkpoint 按时完成，消费积压是 0。

三个工具都说自己接近 100%，事故却恰好落在它们之外。这三个 100% 到底是什么意思？它们能加起来，或者取个平均，得到一个“全公司的监控覆盖率”吗？Flink 作业接上源头就自动有了监控，这算不算覆盖了那张表？

本文顺着这几个问题往下追：先弄清覆盖率的分子分母各是什么，再看“被监控”这件事怎么脱离具体模板来定义，然后落到一个工具无关的抽象层上，最后看怎样不靠手工登记，用两套规则和一个求解器把覆盖率“求”出来。

一、覆盖率的分母

现象：每个工具都接近 100%

先看 Prometheus 的 100% 是怎么来的。Prometheus 通过服务发现得到一批抓取目标，每抓一次，就给这个目标写一条 up：抓到了是 1，没抓到是 0。所谓覆盖率，就是 up=1 的目标数除以目标总数。Zabbix 的算法类似，分母是登记在它里面的主机，分子是 agent.ping 还通的主机¹。Flink 平台更直接：作业是平台自己创建的，每个作业创建时就挂上了指标上报，分母和分子天然相等。

问题出在分母上。三个工具的分母，都是它自己已经知道的东西。一台从没装过 Zabbix agent 的机器，不在 Zabbix 的分母里；一个新建的集群 prod-gz，只要没人把它加进 Prometheus 的服务发现配置，就不会有一条 up 序列，自然也不会让覆盖率下降。

第一次尝试：对缺失的指标告警

Prometheus 社区对“东西不见了”有一个常用办法：absent(up{job="node"})，当整个 job 的 up 都消失时触发告警²。它能抓住“曾经有、后来没了”的一类问题，比如整个 exporter 挂了。

但它抓不住另外两类。一类是从服务发现里被移除的目标：目标一移除，旧的 up 序列就被标记为过期，直接从查询结果里消失，absent() 看到的只是“少了一个”，不会报警³。另一类更彻底，是从来没被发现过的东西，比如那个新集群和那张表，它们连“曾经有”都没有。说到底，absent() 拿来比对的仍是 Prometheus 自己的历史，没有跳出它的分母。

图 1 里还藏着第二个问题：同一台数据库机器，Prometheus 记成 10.2.3.17:9100，Zabbix 记成 sh-db-017。两个工具各把它算进自己的分母，各报一次“已覆盖”。如果我们把几个工具的数字简单相加，这台机器被数了两次，而那张表一次也没被数到。

解决：分母从工具之外来

所以第一条结论很朴素：覆盖率的分母不能由监控工具自己提供，必须来自一份独立的“应该被监控的东西”清单。 这份清单的来源是各类资产的权威系统：主机来自 CMDB 或云厂商 API，集群和工作负载来自 Kubernetes API，Flink 作业来自各 Flink 集群的 REST 接口（GET /jobs/overview）⁴，Kafka 主题来自集群元数据，数据表来自数据目录。

覆盖率于是变成两份集合的比对：清单里有哪些东西（期望），工具里看得到哪些东西（观测），前者减去后者就是缺口。

可是，“工具里看得到”就算被监控了吗？ Prometheus 抓得到 sh-db-017 的指标，不代表它的磁盘快满时有人收到告警。接下来要回答的，是“被监控”到底指什么。

二、模板之上的“能力”

现象：模板之间没法直接比较

同样是“监控一台 Linux 主机”，两个工具给出的东西长得完全不一样。Zabbix 的官方模板 “Linux by Zabbix agent” 里是一组 item 和 trigger：agent.ping 检查存活，system.cpu.util 采 CPU，vfs.fs.dependent.size[{#FSNAME},pused] 采每个分区的使用率，配上 Linux: Zabbix agent is not available、Linux: High CPU utilization 这类触发器¹。Prometheus 那边是 node_exporter 暴露的 node_filesystem_avail_bytes 之类的指标，再由 kube-prometheus 的规则集给出 TargetDown、NodeFilesystemAlmostOutOfSpace 这类告警⁵⁶。

如果按“模板里有多少条规则”来衡量，两边的数字毫无可比性：一个模板可能有 60 个 item，却没有一条磁盘告警；另一个只有 10 条规则，却条条都有人接。数规则、数仪表盘，量的都是工具的形态，而不是我们关心的问题有没有被盯住。

原因：我们真正想问的是“这件事有没有人盯”

换个问法会清楚很多。对一台主机，我们想知道的是：它挂了会不会有人知道？CPU、内存、磁盘快满了会不会有人知道？至于用哪个工具、哪个指标名、哪条规则，是实现细节。

把这些“想知道的事”叫作能力（capability）。能力属于实体类型，而不属于工具：主机需要“存活”和“资源”两项能力，一个对外提供接口的服务需要“存活”和“请求”（速率、错误、耗时）能力，一个 Flink 作业需要“运行状态”“checkpoint”“消费积压”，一张数据表需要“新鲜度”和“量级”。

这份清单不必从零想。几套流传很广的监控方法论，本身就是在回答“某类东西要看哪几项”：Brendan Gregg 的 USE 方法针对资源类对象，看利用率、饱和度、错误⁷；Tom Wilkie 的 RED 方法针对请求驱动的服务，看速率、错误、耗时⁸；Google SRE 书里的四个黄金信号是延迟、流量、错误、饱和度⁹。它们正好可以作为不同实体类型的默认能力集。

解决：把每个工具的规则映射到能力上

有了能力这一层，模板差异就被消化了：每个工具里的每一条 item 或规则，都映射到“哪类实体的哪项能力”。比较两个工具，比的是它们各自覆盖了哪些能力，而不是它们有多少条规则。

表里 Prometheus 一列的 Flink 指标，是 Flink 的 Prometheus reporter 把指标作用域拼成名字、把作业名等变量转成 label 之后的样子¹⁰，所以同一个 numRestarts 会在两个工具里各出现一次，名字不同，能力相同。

能力表还顺带让“覆盖”有了层级。同一项能力，可以只是有采集（指标在库里，出事后能查），也可以是有告警（有一条规则会触发），还可以是告警有人接（规则路由到了这个实体的负责团队）。三者对“出事时有没有人知道”的意义完全不同，统计时应该分开。

图 2 最后一行三个工具都是空的，那正是开头那张表。可 Flink 平台明明说那个作业“已自动接入监控”。作业被监控了，为什么它写的表没有？

三、Flink 自动监控覆盖了什么

Flink 的指标是怎么“自动”来的

Flink 自带一套指标系统。作业一提交，JobManager 和各个 TaskManager 就按固定的作用域（scope）注册指标：作业级的 numRestarts、lastCheckpointDuration，任务级的 backPressuredTimeMsPerSecond，算子级的 numRecordsOutPerSecond、currentOutputWatermark。算子指标的默认名字是 <host>.taskmanager.<tm_id>.<job_name>.<operator_name>.<subtask_index>，作业名、算子名都带在里面¹¹。连接器还会补上自己的指标，比如 KafkaSource 会报告还没拉取的记录数 pendingRecords、距离上次处理记录的时间 sourceIdleTime¹²。

平台只要在集群配置里挂一个 reporter，所有作业的这些指标就会源源不断地送到监控系统，再配一套按作业名展开的通用告警模板，就做到了“源头接上就有监控”。这是真实的便利：没有人需要为每个新作业单独配监控。

现象：作业全绿，表停了 6 小时

回到开头的事故。凌晨 02:10，上游的订单服务因为一次配置变更停止往 order_events 写消息。对 ods_order_etl 来说，这意味着什么？

作业没有重启，checkpoint 照常完成（没有新数据，状态很小，反而更快），pendingRecords 是 0，因为主题里确实没有待读的消息。平台模板里每一条告警都是绿的，而且它们没有说错：作业本身确实健康。

唯一变了的是 sourceIdleTime，它从 02:10 起一路涨到 6 小时。这个指标一直在采集，但平台的通用模板不会给它设阈值。原因不难理解：有的主题每秒几千条，空闲 5 分钟就不正常；有的主题只在夜里批量灌一次数，空闲 20 小时也正常。“多久没数据算异常”取决于数据本身的业务节奏，平台在作业这一层根本不知道这个数。

原因：自动监控盯的是载体，不是货物

这就是 Flink 自动监控的边界：它覆盖的是“作业”这个实体的能力，也就是运行状态、checkpoint、积压。作业是搬运数据的载体，数据本身（主题里的消息、表里的行）是另外的实体，它们的能力（生产速率、新鲜度、量级）需要另外的期望值，而这些期望值只有数据的负责人知道。

但作业的指标并非和数据实体毫无关系。sourceIdleTime 说的其实是主题这一侧的事：上游多久没生产了。numRecordsOutPerSecond 说的是表这一侧的事：每秒写进去多少行。也就是说，一个实体上的信号，可以顺着实体之间的关系，给相邻的实体提供一部分覆盖。只是这种“派生覆盖”通常只到“有采集”这一级，要升到“有告警”，还得有人在数据实体上把阈值定下来。

这样，问题就从“Flink 有没有监控”变成了更一般的三件事：实体有哪些类型、它们之间有什么关系、每类实体要求哪些能力。能不能把这三件事做成一个独立的层，让所有工具都往上对？

四、本体层：实体、关系与能力

它应该长什么样

“本体”（ontology）这个词在这里的意思很朴素：对我们关心的世界，约定有哪些类型的东西、它们之间有哪些关系。Palantir 的 Foundry 把这套东西叫 Ontology，用对象类型、属性、链接类型来描述业务世界¹³；我们借这个词，描述的是“需要被监控的世界”。

把前三节的结论拼起来，这个层至少要有五样东西：

实体类型和身份键。 每类实体（主机、集群、Flink 作业、Kafka 主题、数据表、服务）约定一个身份键：主机用主机名，集群用集群名，作业用“Flink 集群 + 作业名”，表用“库.表”。身份键必须稳定，并且不依赖任何一个监控工具的命名。

实体清单。 每类实体的全体实例，从第一节说的权威来源同步过来，这是覆盖率的分母。

关系。 作业读哪个主题、写哪张表，Pod 跑在哪个集群，实体归哪个团队负责。关系既用来做派生覆盖，也用来回答“告警该发给谁”。

能力要求。 每类实体需要哪些能力，按重要度分级：核心表要求新鲜度告警，临时表也许只要求有采集。

监控绑定。 每个工具里的每一条规则或采集项，落到“哪个实体的哪项能力、到哪一级”。这一步要做身份解析：Prometheus 的 instance="10.2.3.17:9100" 和 Zabbix 的 sh-db-017，都要映射到身份键 host.name = sh-db-017。

业界已有的几块拼图

这五样东西，业界其实各有现成的做法，只是很少有人把它们拼成“覆盖率”这一个用途。

OpenTelemetry 的实体模型解决身份。它规定一个实体由类型、身份属性和描述属性组成：身份属性在实体的生命周期内不能变，描述属性可以变¹⁴。它的语义约定还要求身份属性的选择能让不同的观察者（应用里的 SDK、旁边的 Collector）独立推出同一个身份。这正是我们需要的：host.name、k8s.cluster.name、service.name 这些属性名可以直接拿来做身份键。不过这个模型目前还处在开发状态，关系的建模也写明留待后续。

Backstage 的软件目录解决“声明的清单”和负责人。它用 catalog-info.yaml 声明 Component、API、Resource、System 等实体，每个实体有 spec.owner¹⁵。它的清单来自开发者主动声明，而不是从监控里推出来，所以天然适合当分母；owner 则直接决定告警路由。ServiceNow 的 CSDM 在 CMDB 里做的是类似的事，把业务服务、应用服务、技术服务分层管理¹⁶。

**Grafana 的 Asserts（现在叫 Knowledge Graph）**解决关系。它从 Prometheus 指标的 label 里抽取实体和关系，比如节点 HOSTS Pod、服务 CALLS 服务，还允许用 YAML 补充自定义的实体类型¹⁷。要注意它的图是从观测侧长出来的：没有被采集的东西不会出现在图里。所以它适合补关系和做身份解析，却不能当分母用，这一点和第一节的结论一致。

OpenLineage 的 Flink 集成解决数据链路上的关系。它在作业提交时通过 Flink 的 JobListener 读取 Source 和 Sink，生成“作业读了哪些数据集、写了哪些数据集”的血缘事件¹⁸。有意思的是，它挂的位置和平台的“自动监控”是同一个时刻：作业一提交，监控和血缘边可以一起自动产生。目前它的支持范围有限，文档写明只支持 Application Mode，连接器以 Kafka 为主。

这四块分别对应身份、清单、关系三样东西，再加上我们自己定义的能力要求和监控绑定，本体层就齐了。剩下的问题是：有了这一层，覆盖率具体怎么算、怎么呈现？

五、从枚举到求解

现象：枚举出来的覆盖表会腐化

照第四节的写法，最直接的落地方式是一张对照表：每个工具里的每个目标、每条规则，手工登记成“它覆盖了哪个实体的哪项能力”。规模小的时候这能用，可它很快就会烂掉。

设想机房新上架一台数据库机器 gz-db-001。它进了 CMDB，清单那一侧自动就有了；对照表那一侧却没有人想起来加一行，于是它在报表里要么不出现，要么永远显示“没有任何监控”，没人分得清这是真缺口还是表没更新。反过来，一台下线的机器从 CMDB 删掉了，对照表里那几行还在，指向一个已经不存在的东西。实体每天在变，对照表靠人维护，它最后会变成第二份 CMDB，而且是更旧的那份。

原因：工具本身就是按规则覆盖的

如果回头看各个工具实际是怎么决定“监控谁”的，会发现它们几乎都不按名单，而是按规则。

Zabbix 的自动注册可以写一个动作：主机元数据里含 Linux，就加入某个主机组、挂上 Linux 模板¹⁹。Prometheus Operator 的 ServiceMonitor 不列服务名，只写 selector.matchLabels，选中所有带某个 label 的 Service；Prometheus 再用 serviceMonitorSelector 选中它要用的 ServiceMonitor²⁰。Flink 的 reporter 写在集群配置里，这个集群上提交的每个作业都会上报。这和 Kubernetes 调度的亲和性（affinity）是一个思路：Pod 不写“我要去 node-17”，只写“我要去带 disktype=ssd 的节点”，由调度器去匹配²¹。

也就是说，真实世界里的覆盖关系本来就是“属性和关系上的规则”，是我们在建模时把它摊平成了名单。名单会腐化，是因为它把规则的结果抄了一遍，而规则的输入每天都在变。

解决：两套规则，一个求解器

那就不抄结果，直接把规则写下来。整个系统拆成三样东西：

事实，由同步程序从权威来源和工具配置里读出来，不经人手：CMDB 里有哪些主机、各自属于哪个组、重要度是几级；血缘里谁读谁、谁写谁；Zabbix 的模板挂在哪些主机组上；Prometheus 当前有哪些抓取目标。

规则一：哪些需要被覆盖。 它只用类型、属性和关系来说话，比如“主机都要有存活”“一级主机要有资源”“一级表要有新鲜度”。这里正好用上关系：一张一级表的上游，不管隔了几层作业和主题，都继承它的重要度。这是一条递归规则，写一次，任意深度的血缘都适用。

规则二：当前覆盖了什么。 它把每个工具的作用域翻译成能力：主机组在 Linux 模板的作用域里，组里的主机就有存活和资源；主机的 IP 出现在 Prometheus 抓取目标里，它就有存活（身份解析在这里也是一条规则，而不是一张别名表）；平台托管的作业有运行；被有监控的作业读着的主题，派生出“生产”的 L1。

求解器拿事实和两套规则，算出“要求的”和“做到的”，两者之差就是缺口。规则里不出现任何一个主机名、作业名，新东西一进事实，就自动落进两套规则里被判定一次。

这类“事实加规则、推出结论”的问题，有一族专门的工具：逻辑编程，最常用的是 Datalog。它天然支持递归（血缘多少层都行），一定会算完，每个结论都能追溯到是哪几条事实和规则推出来的。它在工程里并不陌生：GitHub 的 CodeQL 查询语言，语义就建立在 Datalog 上²²；OPA 的策略语言 Rego 也是受 Datalog 启发设计的²³；Soufflé 则是一个可以直接拿来跑大规模 Datalog 的引擎²⁴。

直接写求解器的语法，表达力够了，可读性却不好：:-、大写变量、#max，对大多数写监控规则的人来说是一门新语言。好在求解是引擎的事，写法可以另选。下面两小节给出两种写法，一种借 Kubernetes 的亲和性，一种用 SQL，它们表达的是同一套语义，跑出来的缺口也完全一样。

规则用什么写：亲缘性风格的声明

这样分层的先例就在 Kubernetes 生态里。OPA Gatekeeper 把策略拆成两层：ConstraintTemplate 里是 Rego 写的逻辑，Constraint 是一个普通的 YAML 资源，用 match 字段（kinds、labelSelector、namespaceSelector）指定作用在哪些对象上，用 parameters 传参²⁵。写约束的人只碰 YAML，求值交给底下的引擎。

覆盖规则也可以这么做。Kubernetes 用户已经熟悉的几种选择语法，正好对得上求解器里的几种基本操作：

• matchLabels / matchExpressions（In、NotIn、Exists、DoesNotExist）是对实体自身属性的条件²⁶。
• affinity 加 topologyKey 在 Kubernetes 里的意思是“存在另一个满足条件的对象，并且和我在这个 label 上取值相同”²¹。放到覆盖率里，“主机所在的组挂了 Linux 模板”“主机的 IP 出现在抓取目标里”都是这个形状，身份解析也就成了一条亲和性规则。
• antiAffinity 是“不存在这样的对象”，正好用来写反方向的检查：抓取目标找不到任何 IP 相同的主机。
• 再补两个 Kubernetes 里没有、覆盖率需要的：relatedFrom 沿血缘关系找邻居，LabelPropagation 把一个 label 沿血缘传给所有上游。

于是规则写出来是这样的（节选，完整的 11 条在下面的折叠块里）：

kind: LabelPropagation
metadata: {name: tier-follows-lineage}
spec:
  label: {tier: "1"}
  towards: upstream          # 沿血缘传给上游，深度不限
---
kind: CoverageRequirement
metadata: {name: critical-host}
spec:
  selector: {kind: host, matchLabels: {tier: "1"}}
  require: [{capability: resource, level: 3}]
---
kind: CoverageProvider
metadata: {name: zabbix-linux-template}
spec:
  selector:
    kind: host
    affinity:                # 存在一个挂了 Linux 模板的主机组，和它的 group 相同
      - selector: {kind: hostgroup, matchLabels: {zabbix-template: linux}}
        topologyKey: group
  provide: [{capability: alive, level: 3}, {capability: resource, level: 3}]
---
kind: CoverageProvider
metadata: {name: source-idle-derived}
spec:
  selector:
    kind: topic
    relatedFrom:             # 被一个有“运行”覆盖的作业读着
      - type: reads
        selector: {kind: flink_job, covered: running}
  provide: [{capability: produce, level: 1}]
---
kind: OrphanCheck
metadata: {name: prometheus-orphan}
spec:
  selector:
    kind: prom_target
    antiAffinity:            # 找不到任何一台 ip 相同的主机
      - selector: {kind: host}
        topologyKey: ip

规则分成三种资源：CoverageRequirement 是第一套逻辑（哪些需要覆盖），CoverageProvider 是第二套（当前覆盖了什么），OrphanCheck 是反方向的检查。每条都只说“选中哪些实体、要求或提供哪项能力到哪一级”，没有一个主机名。事实也换成同样的形状：实体带 kind、name 和 labels，关系是 from、type、to。

一个八十行左右的编译器把 YAML 翻译成 ASP（回答集编程，Datalog 的一个扩展），再交给求解器 clingo²⁷。比如上面那条 zabbix-linux-template 会变成：

has(E, alive, 3) :- entity(host, E), entity(hostgroup, O0),
    label(O0, "zabbix-template", "linux"),
    label(E, "group", V1), label(O0, "group", V1).

affinity 变成了对另一个实体 O0 的连接，topologyKey: group 变成了两边 group 取同一个值 V1。LabelPropagation 则编译成一条递归规则：

label(U, "tier", "1") :- upstream(U, X), label(X, "tier", "1").

python3 compile.py facts.yaml rules.yaml > coverage.lp && python3 -m clingo coverage.lp 的结果（每个结论一行）：

orphan("10.2.9.40:9100")
gap("gz-db-001",alive,3,0)
gap("gz-db-001",resource,3,0)
gap("dws_daily_job",running,3,0)
gap("order_events",produce,2,1)
gap("dwd_order_wide",fresh,3,0)

# 事实：由同步程序从 CMDB、各工具配置和血缘生成
entities:
  - {kind: host, name: sh-db-017, labels: {group: IDC-SH, tier: "1", ip: 10.2.3.17}}
  - {kind: host, name: sh-db-018, labels: {group: IDC-SH, tier: "2"}}
  - {kind: host, name: gz-db-001, labels: {group: IDC-GZ, tier: "1"}}
  - {kind: hostgroup, name: IDC-SH, labels: {group: IDC-SH, zabbix-template: linux}}
  - {kind: hostgroup, name: IDC-GZ, labels: {group: IDC-GZ}}
  - {kind: prom_target, name: "10.2.3.17:9100", labels: {ip: 10.2.3.17}}
  - {kind: prom_target, name: "10.2.9.40:9100", labels: {ip: 10.2.9.40}}
  - {kind: flink_job, name: ods_order_etl, labels: {platform: managed}}
  - {kind: flink_job, name: dws_daily_job}
  - {kind: topic, name: order_events}
  - {kind: table, name: dwd_order_wide, labels: {tier: "1"}}
  - {kind: table, name: dws_order_daily, labels: {tier: "1", freshness-sla: 1h}}
relations:
  - {from: ods_order_etl, type: reads, to: order_events}
  - {from: ods_order_etl, type: writes, to: dwd_order_wide}
  - {from: dws_daily_job, type: reads, to: dwd_order_wide}
  - {from: dws_daily_job, type: writes, to: dws_order_daily}

# ---- 规则一：哪些需要被覆盖 ----
kind: LabelPropagation
metadata: {name: tier-follows-lineage}
spec:
  label: {tier: "1"}
  towards: upstream          # 沿血缘传给上游，深度不限
---
kind: CoverageRequirement
metadata: {name: host-baseline}
spec:
  selector: {kind: host}
  require: [{capability: alive, level: 3}]
---
kind: CoverageRequirement
metadata: {name: critical-host}
spec:
  selector: {kind: host, matchLabels: {tier: "1"}}
  require: [{capability: resource, level: 3}]
---
kind: CoverageRequirement
metadata: {name: critical-job}
spec:
  selector: {kind: flink_job, matchLabels: {tier: "1"}}
  require: [{capability: running, level: 3}]
---
kind: CoverageRequirement
metadata: {name: critical-topic}
spec:
  selector: {kind: topic, matchLabels: {tier: "1"}}
  require: [{capability: produce, level: 2}]
---
kind: CoverageRequirement
metadata: {name: critical-table}
spec:
  selector: {kind: table, matchLabels: {tier: "1"}}
  require: [{capability: fresh, level: 3}]
---
# ---- 规则二：当前覆盖了什么 ----
kind: CoverageProvider
metadata: {name: zabbix-linux-template}
spec:
  selector:
    kind: host
    affinity:                # 存在一个挂了 Linux 模板的主机组，和它的 group 相同
      - selector: {kind: hostgroup, matchLabels: {zabbix-template: linux}}
        topologyKey: group
  provide: [{capability: alive, level: 3}, {capability: resource, level: 3}]
---
kind: CoverageProvider
metadata: {name: prometheus-scrape}
spec:
  selector:
    kind: host
    affinity:                # 存在一个抓取目标，和它的 ip 相同
      - selector: {kind: prom_target}
        topologyKey: ip
  provide: [{capability: alive, level: 3}]
---
kind: CoverageProvider
metadata: {name: flink-platform}
spec:
  selector: {kind: flink_job, matchLabels: {platform: managed}}
  provide: [{capability: running, level: 3}]
---
kind: CoverageProvider
metadata: {name: source-idle-derived}
spec:
  selector:
    kind: topic
    relatedFrom:             # 被一个有“运行”覆盖的作业读着
      - type: reads
        selector: {kind: flink_job, covered: running}
  provide: [{capability: produce, level: 1}]
---
kind: CoverageProvider
metadata: {name: freshness-sla}
spec:
  selector:
    kind: table
    matchExpressions: [{key: freshness-sla, operator: Exists}]
  provide: [{capability: fresh, level: 3}]
---
# ---- 反方向：工具里有、清单里没有 ----
kind: OrphanCheck
metadata: {name: prometheus-orphan}
spec:
  selector:
    kind: prom_target
    antiAffinity:            # 找不到任何一台 ip 相同的主机
      - selector: {kind: host}
        topologyKey: ip

"""把亲缘性风格的 YAML 规则编译成 ASP，交给 clingo 求解。"""
import itertools, sys, yaml

def q(s): return '"%s"' % s
fresh = itertools.count()

def sel(spec, var):
    """把一个 selector 编译成一串 ASP 条件，var 是被选中的实体。"""
    body = [f"entity({spec['kind']}, {var})"]
    for k, v in spec.get("matchLabels", {}).items():
        body.append(f"label({var}, {q(k)}, {q(v)})")
    for e in spec.get("matchExpressions", []):
        op, k = e["operator"], q(e["key"])
        if op == "Exists":        body.append(f"label({var}, {k}, _)")
        elif op == "DoesNotExist": body.append(f"not label({var}, {k}, _)")
        elif op == "In":
            v = f"V{next(fresh)}"
            body.append(f"label({var}, {k}, {v}), {v} = ({'; '.join(q(x) for x in e['values'])})")
        elif op == "NotIn":
            body += [f"not label({var}, {k}, {q(x)})" for x in e["values"]]
    if "covered" in spec:
        body.append(f"has({var}, {spec['covered']}, _)")
    for a in spec.get("affinity", []):
        o, v = f"O{next(fresh)}", f"V{next(fresh)}"
        body += sel(a["selector"], o) + [f"label({var}, {q(a['topologyKey'])}, {v})",
                                         f"label({o}, {q(a['topologyKey'])}, {v})"]
    for a in spec.get("antiAffinity", []):
        aux = f"aff_{next(fresh)}"
        o, v = f"O{next(fresh)}", f"V{next(fresh)}"
        aux_body = sel(a["selector"], o) + [f"label(X, {q(a['topologyKey'])}, {v})",
                                            f"label({o}, {q(a['topologyKey'])}, {v})"]
        out.append(f"{aux}(X) :- entity(_, X), {', '.join(aux_body)}.")
        body.append(f"not {aux}({var})")
    for r in spec.get("relatedFrom", []):
        o = f"O{next(fresh)}"
        body += [f"edge({r['type']}, {o}, {var})"] + sel(r["selector"], o)
    return body

out = []
facts = yaml.safe_load(open(sys.argv[1]))
for e in facts["entities"]:
    out.append(f"entity({e['kind']}, {q(e['name'])}).")
    out += [f"label({q(e['name'])}, {q(k)}, {q(v)})." for k, v in (e.get("labels") or {}).items()]
for r in facts["relations"]:
    out.append(f"edge({r['type']}, {q(r['from'])}, {q(r['to'])}).")

out.append('upstream(U, X) :- edge(reads, X, U).  upstream(U, X) :- edge(writes, U, X).')
for doc in yaml.safe_load_all(open(sys.argv[2])):
    s = doc["spec"]
    if doc["kind"] == "LabelPropagation":
        (k, v), = s["label"].items()
        out.append(f"label(U, {q(k)}, {q(v)}) :- upstream(U, X), label(X, {q(k)}, {q(v)}).")
    elif doc["kind"] in ("CoverageRequirement", "CoverageProvider"):
        head = "need" if doc["kind"] == "CoverageRequirement" else "has"
        body = ", ".join(sel(s["selector"], "E"))
        for c in s.get("require", s.get("provide")):
            out.append(f"{head}(E, {c['capability']}, {c['level']}) :- {body}.  % {doc['metadata']['name']}")
    elif doc["kind"] == "OrphanCheck":
        out.append(f"orphan(E) :- {', '.join(sel(s['selector'], 'E'))}.")
out += ["has(E, C, 0) :- need(E, C, _).",
        "gap(E, C, N, G) :- need(E, C, N), G = #max { L : has(E, C, L) }, G < N.",
        "#show gap/4.  #show orphan/1."]
print("\n".join(out))

逐条看，每一条都是规则自己推出来的，没有人登记过：

• gz-db-001 是新机器，它所在的 IDC-GZ 组没挂 Linux 模板，IP 也不在抓取目标里，于是存活和资源都是 0。
• dws_daily_job 是一个跑在自建集群上的作业，不归平台托管，没有自动监控。它之所以被要求覆盖，是因为它写的 dws_order_daily 是一级表，tier: "1" 顺着血缘传了上来。
• order_events 只有从 ods_order_etl 派生来的 L1，没到要求的 L2。
• dwd_order_wide 是一级表，却没有在数据目录里登记新鲜度 SLA，也就没有新鲜度检查。它的下游 dws_order_daily 登记了 SLA，所以不在缺口里。
• 第一行是反方向的检查：10.2.9.40:9100 在 Prometheus 里被抓着，清单里却没有这台机器，多半是下线后没清理的“僵尸”目标，或者是清单漏登记了。

也可以用 SQL 写

如果团队更熟 SQL，同一套逻辑也能直接写成视图。Datalog 和 SQL 的距离其实不远：Datalog 的规则就是带连接的查询，递归规则对应 SQL 的 WITH RECURSIVE²⁸；Google 的 Logica 就是一门把 Datalog 风格的程序翻译成 SQL、交给 SQLite、DuckDB、BigQuery 执行的语言²⁹。

事实用普通的表来存：

-- 事实：由同步程序写入
CREATE TABLE host        (name TEXT PRIMARY KEY, grp TEXT, tier INT, ip TEXT);
CREATE TABLE hostgroup   (name TEXT PRIMARY KEY, zabbix_template TEXT);
CREATE TABLE prom_target (address TEXT PRIMARY KEY, ip TEXT);
CREATE TABLE flink_job   (name TEXT PRIMARY KEY, platform TEXT);
CREATE TABLE topic       (name TEXT PRIMARY KEY);
CREATE TABLE tbl         (name TEXT PRIMARY KEY, tier INT, freshness_sla TEXT);
CREATE TABLE reads       (job TEXT, topic TEXT);
CREATE TABLE writes      (job TEXT, tbl TEXT);

INSERT INTO host VALUES ('sh-db-017','IDC-SH',1,'10.2.3.17'),
                        ('sh-db-018','IDC-SH',2,NULL),
                        ('gz-db-001','IDC-GZ',1,NULL);
INSERT INTO hostgroup   VALUES ('IDC-SH','linux'), ('IDC-GZ',NULL);
INSERT INTO prom_target VALUES ('10.2.3.17:9100','10.2.3.17'), ('10.2.9.40:9100','10.2.9.40');
INSERT INTO flink_job   VALUES ('ods_order_etl','managed'), ('dws_daily_job',NULL);
INSERT INTO topic       VALUES ('order_events');
INSERT INTO tbl         VALUES ('dwd_order_wide',1,NULL), ('dws_order_daily',1,'1h');
INSERT INTO reads       VALUES ('ods_order_etl','order_events'), ('dws_daily_job','dwd_order_wide');
INSERT INTO writes      VALUES ('ods_order_etl','dwd_order_wide'), ('dws_daily_job','dws_order_daily');

两套规则是两个视图，一级实体的递归传播是一个递归 CTE，求解是一次左连接：

-- 血缘：谁是谁的上游
CREATE VIEW upstream AS
  SELECT job AS child, topic AS parent FROM reads
  UNION SELECT tbl, job FROM writes;

-- 一级实体：自己标了 tier 1，或者是某个一级实体的上游（递归，深度不限）
CREATE VIEW critical AS
  WITH RECURSIVE c(name) AS (
    SELECT name FROM host WHERE tier = 1
    UNION SELECT name FROM tbl WHERE tier = 1
    UNION SELECT u.parent FROM upstream u JOIN c ON u.child = c.name
  ) SELECT name FROM c;

-- 规则一：哪些需要被覆盖
CREATE VIEW need AS
            SELECT name, 'alive'    AS cap, 3 AS lvl FROM host
  UNION ALL SELECT name, 'resource', 3 FROM host      WHERE name IN (SELECT name FROM critical)
  UNION ALL SELECT name, 'running',  3 FROM flink_job WHERE name IN (SELECT name FROM critical)
  UNION ALL SELECT name, 'produce',  2 FROM topic     WHERE name IN (SELECT name FROM critical)
  UNION ALL SELECT name, 'fresh',    3 FROM tbl       WHERE name IN (SELECT name FROM critical);

-- 规则二：当前覆盖了什么
CREATE VIEW has AS
            SELECT h.name, c.cap, 3 AS lvl            -- 主机组挂了 Linux 模板
              FROM host h JOIN hostgroup g ON g.name = h.grp AND g.zabbix_template = 'linux'
              CROSS JOIN (SELECT 'alive' AS cap UNION ALL SELECT 'resource') c
  UNION ALL SELECT h.name, 'alive', 3                 -- ip 在抓取目标里
              FROM host h JOIN prom_target p ON p.ip = h.ip
  UNION ALL SELECT name, 'running', 3 FROM flink_job WHERE platform = 'managed'
  UNION ALL SELECT r.topic, 'produce', 1              -- 被平台托管的作业读着
              FROM reads r JOIN flink_job j ON j.name = r.job AND j.platform = 'managed'
  UNION ALL SELECT name, 'fresh', 3 FROM tbl WHERE freshness_sla IS NOT NULL;

-- 求解：要求的层级减去达到的层级
SELECT n.name, n.cap, n.lvl AS need, COALESCE(MAX(h.lvl), 0) AS got
  FROM need n LEFT JOIN has h ON h.name = n.name AND h.cap = n.cap
 GROUP BY n.name, n.cap, n.lvl
HAVING got < n.lvl;

-- 反方向：抓取目标里有、清单里没有
SELECT address AS orphan FROM prom_target
 WHERE ip NOT IN (SELECT ip FROM host WHERE ip IS NOT NULL);

cat facts.sql coverage.sql | sqlite3 -header -column 的结果和 YAML 那一版逐行一致：

name            cap       need  got
--------------  --------  ----  ---
dwd_order_wide  fresh     3     0
dws_daily_job   running   3     0
gz-db-001       alive     3     0
gz-db-001       resource  3     0
order_events    produce   2     1
orphan
--------------
10.2.9.40:9100

两种写法各有长处。亲缘性风格的 YAML 每条规则是一个独立的资源，可以有自己的名字、负责人和评审记录，适合由各个团队分别提交；SQL 一眼能看到全部逻辑，能直接用现有的数据库和 BI 工具出报表，适合数据团队维护。它们背后的思想是同一个：

求解器还能反过来问：最少补哪几项

表格最后一行是两种写法都替代不了的地方。既然缺口是求出来的，就可以反过来问：从一组候选补法里，最少挑哪几项，能让缺口清零？这是一个搜索问题，SQL 只能一行行地查，表达不了“在所有组合里找最好的那个”，求解器却是专门干这个的。把候选补法写成“可选”的事实，加一条“不许有缺口”的约束和一个“补的项越少越好”的目标，接在编译出来的程序后面：

% 候选补法：求解器从里面挑，要求补完之后没有缺口，且补的项最少
{ fix(link(G))       : entity(hostgroup, G) }.
{ fix(onboard(J))    : entity(flink_job, J) }.
{ fix(idle_alert(J)) : edge(reads, J, _) }.
{ fix(sla(T))        : entity(table, T) }.
label(G, "zabbix-template", "linux") :- fix(link(G)).
label(J, "platform", "managed")      :- fix(onboard(J)).
has(T, produce, 3)                   :- edge(reads, J, T), fix(idle_alert(J)).
label(T, "freshness-sla", "待定")     :- fix(sla(T)).
:- gap(_, _, _, _).
#minimize { 1, F : fix(F) }.
#show fix/1.

python3 -m clingo coverage.lp repair.lp 给出的最优解是 4 项：

fix(link("IDC-GZ"))
fix(onboard("dws_daily_job"))
fix(idle_alert("ods_order_etl"))
fix(sla("dwd_order_wide"))

gz-db-001 的两个缺口被一项“给 IDC-GZ 组挂上模板”同时补上，这也是求解器选它、而不是给这台机器单独配监控的原因：挂在组上，这个组以后上架的机器也一并有了。sla("dwd_order_wide") 只说明这张表需要一个新鲜度期望，具体是 1 小时还是 1 天，仍然要由表的负责人来定。候选补法本身也可以照前面的办法包成一种 YAML 资源（比如 kind: FixCandidate），由编译器生成这几行，这里为了看清楚求解器在做什么，直接写了出来。

腐化去了哪里

规则也会过时，但它腐化的方式和名单不一样。名单是每来一个实体就要动一次；规则只在“我们的要求变了”或“工具的作用域变了”时才要动，而后者本身也是同步来的事实。剩下的风险集中在属性上：一台机器的 tier 标错了，规则就会给出错误的要求。所以属性要来自它的权威来源，而不是在覆盖率系统里另抄一份；规则则应当像代码一样放进仓库、走评审，改一条规则的影响，重新求解一次就能看到。

六、覆盖率的呈现

求解出来的是 need 和 has 两张关系，呈现时把它们展开就行。如果一定要一个数字，可以算“一级实体的必需能力里，达到 L3 的比例”。但单个百分比会把最重要的信息平均掉：一千台测试机的资源监控和一张核心表的新鲜度，在百分比里权重一样。更有用的呈现是两样东西：

矩阵用来看全貌，缺口清单用来分派工作。缺口清单的每一行都天然有负责人（实体的 owner），也天然有“补到哪一级算完”的标准，这比一句“覆盖率要提升到 95%”好执行得多。

最后还要补一条检查：规则判定有覆盖，不等于信号还活着。 告警规则写在那里，指标却因为 exporter 升级改了名字、label 变了而再也没有数据，这条规则就永远不会触发。所以同步事实时，还应该顺便查一下每条告警规则引用的序列最近有没有数据，把“信号失效”也作为一条事实交给求解器，命中的降回 L0。这一步把 absent() 用回了它擅长的地方：它不适合找分母之外的东西，但很适合检查已知的规则有没有失效。

七、这套方法的边界

本体层不是免费的。实体类型、能力要求、覆盖规则都要有人维护，清单和属性的来源一旦失真，求解的结论跟着失真，求解器只保证推导正确，不保证输入正确。比较稳妥的起步方式是只建两三类最关键的实体（比如一级服务和核心表），每套规则先写最少的几条，先让缺口清单跑起来，再逐步扩大。

派生覆盖是一种推断，需要谨慎。作业上的 sourceIdleTime 能说明主题“多久没被读到新消息”，但如果作业本身停了，这个信号也跟着消失，它对主题的覆盖也就不存在了。所以派生覆盖应当依赖“提供信号的实体本身是健康的”这个前提，计算时要一起检查。

能力要求是一种约定，也就可能被“刷”。给每张表挂一条永远不会触发的新鲜度告警（阈值设成 30 天），矩阵会全部变绿，却没有任何意义。这类问题本体层管不了，只能靠告警质量本身的度量（触发次数、误报率、处理时长）来约束，那是另一个话题。

小结

1. 三个工具的 100% 是什么意思？ 每个工具的分母都是它自己发现的目标，所以自报覆盖率只能说明“我看见的我都看着”。不同工具的数字既不能相加，也不能平均，同一台机器还会被重复计算。
2. 分母从哪来？ 从工具之外的权威清单来：CMDB、Kubernetes API、Flink REST、Kafka 元数据、数据目录。覆盖率是“期望”和“观测”两个集合的比对。
3. 模板不同怎么比较？ 不比模板，比能力。给每类实体定义必需能力（可以借 USE、RED、黄金信号作默认），把每个工具的规则映射到能力上，并区分有采集、有告警、告警有人接三个层级。
4. Flink 的自动监控覆盖了什么？ 覆盖的是作业这个载体的能力。数据实体的新鲜度、量级需要数据负责人给出期望值；作业上的信号可以顺着读写关系给数据实体提供派生覆盖，但通常只到“有采集”。
5. 本体层长什么样？ 实体类型与身份键、实体清单、关系、能力要求、监控绑定五样东西。身份可以借 OpenTelemetry 的实体模型，清单和负责人可以借 Backstage 或 CMDB，关系可以借 Asserts 和 OpenLineage。
6. 为什么不能手工登记覆盖关系？ 名单是规则结果的抄本，实体一变就腐化。工具本身就按规则覆盖，所以改成两套规则：一套说哪些需要覆盖，一套说当前覆盖了什么，事实自动同步，用 Datalog 或 ASP 一类的求解器求出两者之差，还能反过来求最少的补法。规则不必直接用求解器的语法写：可以写成 Kubernetes 亲和性风格的 YAML 再编译过去，也可以直接写成 SQL 视图和递归 CTE。
7. 结果怎么呈现？ 一张“实体 × 能力 × 层级”的矩阵，加一份按重要度排序、每行都有负责人的缺口清单，再配上对每条告警规则的信号存活检查。

参考资料