指标树(11.1)告诉我们要算哪些数,埋点规范(11.2)保证这些数算得干净。本节解决最后一段路:这些数怎么摆到人眼前、以及出问题时怎么把人叫醒。
边界先说清:本节只讲「看板怎么排、告警怎么定」;告警触发之后怎么排查、怎么用数据驱动下一版迭代,是第 14 章的事;告警指向的某个指标该怎么用实验去改,是第 8 章的事。
看板:把指标摆到该被看到的地方
看板不是「把所有指标堆一页」,而是按决策频率分层。北极星最上面,一眼就能看到;四条 L1 驱动分区块,各自的叶子跟着;客户端健康度(无崩溃率、加载耗时)单独一区,因为它们出问题时影响的是所有人,必须最醒目。
每个区块要写清两件事:刷新频率(北极星和健康度要快,比如 5 分钟;留存这类按天算的,15 分钟足够)和责任人(这个区出问题找谁)。没有责任人的看板等于没有看板——数字红了没人管,比不显示还糟。
告警的两个错误:漏报与误报
告警规则的形式很朴素:某个指标、满足某个条件、连续几个采样点,就响。但定阈值这件事里藏着两个方向的错误:
- 误报(假阳性):指标没真出事却响了。阈值定得太紧、或者用了「单点触发」,噪声一抖就报警。误报多了,人就开始无视告警——这比不报警更危险,因为它把真警报也一起淹了。
- 漏报(假阴性):真出事了却没响。阈值定得太松,或者要求「连续 N 点」而故障只持续了 1 点,告警就被窗口过滤掉了。
所以「连续窗口」和「冷却时间」是两个关键旋钮:窗口调大能压误报,但会放大漏报;冷却去重能防风暴,但设太大也会吞掉相邻的真警报。本节的任务就是把这组取舍变成能跑出数字的仿真。
D 层:看板与告警配置
下面这份是 D 层(静态层)配置:panels 是看板分区,alerts 是四条告警规则(比较符、阈值、连续窗口、冷却、等级),signals 用种子化 LCG 生成四条确定性序列,并在指定点注入已知异常(anomaly)——这样才好在事后判断告警「抓没抓到、抓错了没有」。
{
"version": 1,
"series": "《小游戏开发实战》",
"volume": "卷三《上线变现与长线运营》",
"game": "merge-island",
"observedAt": "2026-10",
"seed": 20261012,
"note": "看板与告警配置(D 层)。panels 声明看板分区(放哪些指标、刷新频率、责任人);alerts 是告警规则:指标 + 比较符(lt/gt)+ 阈值 + 连续窗口 windowPoints + 冷却 cooldownPoints + 等级。signals 用种子化 LCG 生成确定性时间序列:baseline + 噪声 ±noise,再在指定点注入 anomaly 偏移,用于验证「告警会不会漏报、会不会误报」。全整数,禁浮点;无 Math.random,随机数由 seed 决定。baseline / noise / anomaly 均为示例假设值,真实指标需真实业务环境,命令行不可复现(X)。",
"units": {
"baselinePermille": "序列基线(同指标单位 ×1000 或直接取值)",
"noisePermille": "噪声幅度,序列在 ±noise 内抖动",
"windowPoints": "连续多少个采样点满足条件才触发",
"cooldownPoints": "两次触发之间至少间隔多少个采样点"
},
"panels": [
{ "id": "north-star", "title": "北极星", "metrics": ["efficient-active"], "refreshSec": 300, "owner": "制作人" },
{ "id": "retention", "title": "留存", "metrics": ["d1-retention", "d7-retention", "session-per-dau"], "refreshSec": 900, "owner": "数据分析" },
{ "id": "money", "title": "变现", "metrics": ["arpdau", "pay-rate", "ad-imp-per-dau"], "refreshSec": 900, "owner": "商业化" },
{ "id": "health", "title": "健康度", "metrics": ["crash-free-rate", "first-load-sec"], "refreshSec": 300, "owner": "客户端" }
],
"alerts": [
{ "id": "d1-drop", "metric": "d1-retention", "comparator": "lt", "thresholdPermille": 360, "windowPoints": 2, "cooldownPoints": 4, "severity": "P1" },
{ "id": "crash-free", "metric": "crash-free-rate", "comparator": "lt", "thresholdPermille": 985, "windowPoints": 3, "cooldownPoints": 5, "severity": "P1" },
{ "id": "arpdau-watch", "metric": "arpdau", "comparator": "lt", "thresholdPermille": 95, "windowPoints": 2, "cooldownPoints": 3, "severity": "P2" },
{ "id": "load-slow", "metric": "first-load-sec", "comparator": "gt", "thresholdPermille": 3500, "windowPoints": 1, "cooldownPoints": 0, "severity": "P3" }
],
"points": 14,
"signals": [
{ "metric": "d1-retention", "baselinePermille": 410, "noisePermille": 12, "anomaly": { "atPoint": 6, "deltaPermille": -80, "points": 2 } },
{ "metric": "crash-free-rate", "baselinePermille": 996, "noisePermille": 3, "anomaly": null },
{ "metric": "arpdau", "baselinePermille": 120, "noisePermille": 8, "anomaly": { "atPoint": 9, "deltaPermille": -40, "points": 1 } },
{ "metric": "first-load-sec", "baselinePermille": 3000, "noisePermille": 700, "anomaly": null }
]
}
S 层:告警仿真
下面这段是 S 层脚本:先用种子化 LCG 生成四条序列(注入已知异常),再逐条规则扫描——连续 windowPoints 点满足条件才触发、两次触发至少隔 cooldownPoints 点,最后按「触发点是否落在异常窗口内」判定真阳性 / 假阳性 / 漏报。产出 JSON + CSV + SVG。
#!/usr/bin/env python3
# file: sim/l11_alert.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/l11_alert.py --config config/ --out out/
"""看板与告警仿真:读 config/l11_alert.json。
1) 种子化 LCG(seed 决定,无 Math.random)生成每条信号的时间序列:
value[i] = baseline + 整数噪声(±noise) + 若 i 落在 anomaly 窗口再加 delta。
2) 逐条告警规则扫描:连续 windowPoints 个点满足比较条件才触发;两次触发之间至少间隔
cooldownPoints 个点(去重 / 抑制风暴)。
3) 判定:触发点落在已知 anomaly 窗口内为真阳性 TP,否则假阳性 FP;有 anomaly 却无触发为漏报。
产出:out/l11_alert.json + .csv + .svg。全整数,禁浮点。baseline / noise / anomaly 均为示例
假设值,真实指标需真实业务环境,命令行不可复现(X)。
"""
from __future__ import annotations
import argparse
import csv
import json
import os
import xml.etree.ElementTree as ET
MASK = 0x7FFFFFFF
def load(cfg_dir: str) -> dict:
with open(os.path.join(cfg_dir, "l11_alert.json"), encoding="utf-8") as f:
return json.load(f)
class Lcg:
"""确定性 32 位 LCG(Numerical Recipes 常数)。state 由 seed 决定,无 Math.random。"""
def __init__(self, seed: int) -> None:
self.state = seed & MASK
def noise(self, amp: int) -> int:
self.state = (self.state * 1103515245 + 12345) & MASK
return (self.state * 2 * amp) // (MASK + 1) - amp
def gen_series(cfg: dict) -> dict[str, list[int]]:
rng, n, out = Lcg(cfg["seed"]), cfg["points"], {}
for sig in cfg["signals"]:
vals = []
for i in range(n):
v = sig["baselinePermille"] + rng.noise(sig["noisePermille"])
an = sig.get("anomaly")
if an and an["atPoint"] <= i < an["atPoint"] + an["points"]:
v += an["deltaPermille"]
vals.append(v)
out[sig["metric"]] = vals
return out
def anomaly_window(sig: dict) -> tuple[int, int] | None:
an = sig.get("anomaly")
return (an["atPoint"], an["atPoint"] + an["points"] - 1) if an else None
def run_alert(rule: dict, series: list[int], sig: dict) -> dict:
thr, cmp, win, cool = rule["thresholdPermille"], rule["comparator"], rule["windowPoints"], rule["cooldownPoints"]
win_range, fires, consec, last_fire = anomaly_window(sig), [], 0, -10**9
for i, v in enumerate(series):
consec = consec + 1 if (v < thr if cmp == "lt" else v > thr) else 0
if consec >= win and i - last_fire >= cool:
fires.append(i)
last_fire, consec = i, 0
tps = [i for i in fires if win_range and win_range[0] <= i <= win_range[1]]
return {"id": rule["id"], "metric": rule["metric"], "fires": fires, "fireCount": len(fires),
"tp": len(tps), "fp": len(fires) - len(tps), "missed": bool(win_range) and not tps}
def draw_svg(path: str, series: dict, results: list, cfg: dict) -> int:
W, H, L, R, T = 760, 60 + 92 * len(series), 56, 24, 20
svg = ET.Element("svg", {"xmlns": "http://www.w3.org/2000/svg", "width": str(W),
"height": str(H), "viewBox": f"0 0 {W} {H}"})
ET.SubElement(svg, "rect", {"width": str(W), "height": str(H), "fill": "#ffffff"})
n, y0 = cfg["points"], T
px = lambda i: L + (W - L - R) * i // max(1, n - 1)
for metric, vals in series.items():
lo, span = min(vals), max(1, max(vals) - min(vals))
py = lambda v, y0=y0, lo=lo, span=span: y0 + 60 - 56 * (v - lo) // span
ET.SubElement(svg, "text", {"x": "16", "y": str(y0 + 8), "font-size": "12",
"fill": "#374151"}).text = f"{metric} [{lo}–{lo + span}]"
ET.SubElement(svg, "polyline", {"points": " ".join(f"{px(i)},{py(v)}" for i, v in enumerate(vals)),
"fill": "none", "stroke": "#2563eb", "stroke-width": "2"})
fired = {i for r in results if r["metric"] == metric for i in r["fires"]}
for i in fired:
ET.SubElement(svg, "circle", {"cx": str(px(i)), "cy": str(py(vals[i])), "r": "4", "fill": "#dc2626"})
y0 += 92
os.makedirs(os.path.dirname(path), exist_ok=True)
ET.ElementTree(svg).write(path, encoding="utf-8", xml_declaration=True)
return os.path.getsize(path)
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default="config/")
ap.add_argument("--out", default="out/")
a = ap.parse_args()
cfg = load(a.config)
sig_by_metric = {s["metric"]: s for s in cfg["signals"]}
series = gen_series(cfg)
results = [run_alert(r, series[r["metric"]], sig_by_metric[r["metric"]]) for r in cfg["alerts"]]
total_fires = sum(r["fireCount"] for r in results)
total_tp, total_fp = sum(r["tp"] for r in results), sum(r["fp"] for r in results)
storms = sorted({r["id"] for r in results for i in r["fires"]
if len([j for j in r["fires"] if i <= j < i + 5]) >= 4}) # 5 点内触发 >= 4 次
os.makedirs(a.out, exist_ok=True)
sb = draw_svg(os.path.join(a.out, "l11_alert.svg"), series, results, cfg)
with open(os.path.join(a.out, "l11_alert_series.csv"), "w", encoding="utf-8", newline="") as f:
w = csv.writer(f)
w.writerow(["point"] + list(series.keys()))
for i in range(cfg["points"]):
w.writerow([i] + [series[m][i] for m in series])
report = {"version": cfg["version"], "game": cfg["game"], "seed": cfg["seed"], "points": cfg["points"],
"alertCount": len(results), "totalFires": total_fires, "truePositive": total_tp,
"falsePositive": total_fp, "missed": [r["id"] for r in results if r["missed"]],
"storms": storms, "series": series, "alerts": results, "svgBytes": sb}
with open(os.path.join(a.out, "l11_alert.json"), "w", encoding="utf-8") as f:
json.dump(report, f, ensure_ascii=False, indent=2)
print(json.dumps({"totalFires": total_fires, "truePositive": total_tp, "falsePositive": total_fp,
"perAlert": {r["id"]: {"fires": r["fires"], "tp": r["tp"], "fp": r["fp"]} for r in results},
"missed": report["missed"], "storms": storms, "svgBytes": sb}, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
在本机跑它(假设参数),输出如下:
{
"totalFires": 5,
"truePositive": 1,
"falsePositive": 4,
"perAlert": {
"d1-drop": { "fires": [7], "tp": 1, "fp": 0 },
"crash-free": { "fires": [], "tp": 0, "fp": 0 },
"arpdau-watch": { "fires": [], "tp": 0, "fp": 0 },
"load-slow": { "fires": [3, 4, 7, 13], "tp": 0, "fp": 4 }
},
"missed": ["arpdau-watch"],
"storms": [],
"svgBytes": 1453
}
五条规则各代表一种调法。d1-drop(次日留存连续 2 点 < 360)抓到了真异常——序列在第 6、7 点跌到 318 和 340,第 7 点触发,真阳性 1。crash-free(连续 3 点 < 985)一声没响也没误报——无崩溃率稳定在 993~998,阈值留了余量。arpdau-watch(连续 2 点 < 95)漏报了——arpdau 只在第 9 点跌到 87,单点异常被「连续 2 点」的窗口过滤掉了。load-slow(单点 > 3500,无冷却)误报 4 次——首包耗时噪声大(±700),14 个点里 4 个偶然越过 3500,全是假阳性。
这四条合起来就是一句结论:窗口是压误报的旋钮,但它同时会制造漏报;单点触发则相反。load-slow 那类噪声大的指标,靠单点阈值几乎必然误报,正确做法是拉长窗口或改看分位数(P50 而不是瞬时值)。
B 层:阈值浮动 10%,告警还稳吗
阈值都是拍的。所以必须问一句:如果阈值上下浮动 10%,触发次数会怎么变?下面这段是 B 层基准脚本,逐条规则扰动阈值 ±10%、重跑同一份序列,比较触发次数。它是基准,软告警不硬失败,退出码恒为 0。
#!/usr/bin/env python3
# file: sim/l11_alert_bench.py
# tier: B
# verify: /opt/homebrew/bin/python3 sim/l11_alert_bench.py --config config/ --out out/
"""B 基准:告警规则对阈值的稳健性。
告警阈值是**假设参数**——360‰ 是拍的、985‰ 是拍的。所以必须问一句:如果阈值上下浮动
10%,触发次数会变多少?做法:逐条规则把 thresholdPermille 上下各扰动 10%,重跑同一份
确定性序列(seed 不变),比较触发次数。软告警:任一规则在 ±10% 扰动下触发次数变动 >= 2
即打 WARN(说明该规则对阈值过敏感,容易误报或漏报)。退出码恒为 0——这是基准,不是门禁。
全整数,无浮点。阈值均为示例假设值,真实指标需真实业务环境,命令行不可复现(X)。
"""
from __future__ import annotations
import argparse
import json
import os
from l11_alert import gen_series, load, run_alert, anomaly_window
SWING_WARN_FIRES = 2
PERTURB_PCT = 10
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default="config/")
ap.add_argument("--out", default="out/")
a = ap.parse_args()
cfg = load(a.config)
sig_by_metric = {s["metric"]: s for s in cfg["signals"]}
series = gen_series(cfg)
base = {r["id"]: len(run_alert(r, series[r["metric"]], sig_by_metric[r["metric"]])["fires"])
for r in cfg["alerts"]}
cases = []
for r in cfg["alerts"]:
sig = sig_by_metric[r["metric"]]
for pct in (-PERTURB_PCT, PERTURB_PCT):
thr = r["thresholdPermille"] * (100 + pct) // 100
rr = dict(r, thresholdPermille=thr)
fires = len(run_alert(rr, series[r["metric"]], sig)["fires"])
cases.append({"id": r["id"], "pct": pct, "threshold": thr,
"fires": fires, "delta": fires - base[r["id"]]})
swing = {r["id"]: max(abs(c["delta"]) for c in cases if c["id"] == r["id"]) for r in cfg["alerts"]}
warnings = [f"规则 {i} 在 ±{PERTURB_PCT}% 阈值扰动下触发次数变动 {v} >= {SWING_WARN_FIRES}"
for i, v in swing.items() if v >= SWING_WARN_FIRES]
os.makedirs(a.out, exist_ok=True)
rep = {"version": cfg["version"], "source": "sim/l11_alert_bench.py", "seed": cfg["seed"],
"perturbPct": PERTURB_PCT, "swingWarnFires": SWING_WARN_FIRES,
"baseFires": base, "cases": cases, "swingById": swing,
"maxSwing": max(swing.values()) if swing else 0, "warnings": warnings,
"verdict": "PASS(soft)" if not warnings else f"WARN(soft) {len(warnings)} 条"}
with open(os.path.join(a.out, "l11_alert_bench.json"), "w", encoding="utf-8") as f:
json.dump(rep, f, ensure_ascii=False, indent=2)
print(json.dumps({"baseFires": base, "swingById": swing, "maxSwing": rep["maxSwing"],
"verdict": rep["verdict"], "warnings": warnings}, ensure_ascii=False))
return 0 # B 层软告警:恒返回 0
if __name__ == "__main__":
raise SystemExit(main())
在本机跑它,输出如下:
{"baseFires": {"d1-drop": 1, "crash-free": 0, "arpdau-watch": 0, "load-slow": 4}, "swingById": {"d1-drop": 1, "crash-free": 3, "arpdau-watch": 0, "load-slow": 4}, "maxSwing": 4, "verdict": "WARN(soft) 2 条", "warnings": ["规则 crash-free 在 ±10% 阈值扰动下触发次数变动 3 >= 2", "规则 load-slow 在 ±10% 阈值扰动下触发次数变动 4 >= 2"]}
两条软告警把问题指得很准:crash-free 的触发次数在 ±10% 阈值下能变动 3 次(阈值挪一点就从「完全不响」变成「响几次」),load-slow 变动 4 次(本来就全是误报)。而 d1-drop 只变动 1 次——它对着一个真实的、持续 2 点的异常,阈值挪 10% 也照样抓到。这正是好告警与坏告警的分水岭:好告警对阈值不敏感,因为它抓的是真信号;坏告警对阈值极度敏感,因为它抓的本来就是噪声。
小结与下一步
这一节把看板与告警接在了指标树上:看板按决策频率分区、每区有责任人和刷新频率;告警用「连续窗口 + 冷却」两个旋钮在漏报与误报之间取舍;再跑 B 基准,把「阈值到底稳不稳」也变成数字。三件事记住就够——误报比漏报更危险,因为它会让人无视告警;窗口压误报但制造漏报;好告警对阈值不敏感。
第 11 章到此收尾:指标树(11.1)、埋点规范(11.2)、看板告警(11.3)——数据分析体系的三根桩全部立好。下一章进入运营本身:活动怎么设计、赛季怎么排、节点日历怎么定。
阅读导航:上一节:埋点规范与口径统一 · 下一节:活动模板与奖励预算 。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。