11.3 看板与告警

看板解决「该看什么」,告警解决「什么时候该被吵醒」。本节给出一份看板分区加告警规则配置,用种子化序列仿真告警:连续窗口触发、冷却去重、真阳性 / 假阳性 / 漏报判定,再跑 B 层基准看阈值浮动 10% 时触发次数会不会翻盘——把「告警定太松会漏、定太紧会吵」变成可比较的数字。

指标树(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)——数据分析体系的三根桩全部立好。下一章进入运营本身:活动怎么设计、赛季怎么排、节点日历怎么定。

阅读导航:上一节:埋点规范与口径统一 · 下一节:活动模板与奖励预算 。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「minigame」更多文章

  1. 附录 F:全书公式、参数与数据集索引
  2. 9.3 同意管理与数据出境
  3. 9.2 COPPA 与儿童向游戏的限制