一个事故可能产生很多告警，但一组告警并不能证明它们具有同一个原因。Alertmanager 根据配置的标签与路由上下文组织告警，便于一起处理通知，组内仍然保留各条告警的独立身份。静默则是另一种操作：在指定时间窗口内抑制匹配通知，并不会替应用解决故障。理解这些差异，才能准确解释告警页面与通知数量的变化。

分组键决定哪些细节放在一起

配置 group_by: [alertname, service] 时，两条属于结算服务的 QueueHigh 告警可以进入同组，即使实例标签不同。搜索服务的 QueueHigh 属于另一组，结算服务的 DiskFull 也因为告警名不同而单独成组。分组依据被选择的标签值，不要求所有成员具有完全相同的指纹。通知被合并后，实例级细节仍然可以从成员列表查看。

把实例加入分组键会增加组数，也可能抵消跨副本合并的目的；省略服务又可能把不同负责人或工作负载混在一起。应围绕排查方式和责任归属选择分组键，同时保留查看成员的入口。特殊值 ... 表示使用所有标签分组，实际效果接近关闭更大范围的聚合，不能把它当作自动寻找合理分组方式的通配符。

路由上下文同样重要。发往不同接收器的告警，即使所选分组标签相同，也不一定组成同一份通知。应验证路由与分组的组合行为。负责整个服务的接收器可能适合合并多个实例，另一个集成则可能需要不同分组契约。任何一种设置都不会自动证明这些告警具有相同根因，因果关系仍然需要从事件与运行证据判断。

初次等待、组内变化和重复提醒使用不同计时

group_wait 延迟新组的第一份通知，让相近时间到来的告警可以加入。某条告警若在初次等待结束前已经恢复，可能完全不产生通知。group_interval 决定之后检查组内更新的节奏，包括新增触发成员或恢复成员；它不是承诺每隔这么久无条件重发所有尚未结束的组。

在没有其他通知内容变化时，repeat_interval 控制重复提醒。是否到达重复时机是在组周期上检查的，因此应选择兼容的整数倍，而不是把它理解为完全独立且精确的闹钟。发送处理与失败还可能增加延迟。应依据响应目标设置这些参数，并将规则等待时长与通知调度分别考虑，避免只看其中一个参数就承诺发现到送达的时间。

坏掉的样子

分组太细，会让一组副本故障产生很多通知；分组太宽，又可能看不出具体服务需要处理。页面只显示一个组，不意味着只剩一条告警。判断事故是否改善前，应展开成员并核对分组键，不能把通知变少直接当成故障减少的证据。通知数量也可能只是由策略变更或抑制行为造成。

静默条件过宽，可能抑制计划维护范围之外的告警。应限制时间窗口，并检查哪些活动告警实际匹配。静默到期只是恢复通知资格，不会修复应用，也不保证立即发出一条消息。抑制规则则是独立的配置关系，由活动的源告警抑制匹配目标通知，与人工设置有时限静默不同，不能把两者的原因和结束条件混为一谈。

快速验证

把程序保存为 grouping.py，在 Docker 可用时执行 python3 grouping.py。使用 Alertmanager 0.28.1，必要时获取镜像。真实接口接收四条合成告警。接收器没有发送集成，初次通知等待也故意设置得很长，因此实验只观察分组和抑制状态，不会向外部发送消息，也不接触已有告警实例。

import http.client
import io
import json
import subprocess
import time
import uuid

name = "xbcatm-monitoring-" + uuid.uuid4().hex[:12]


class ResponseSocket:
    def __init__(self, data):
        self.data = data

    def makefile(self, mode):
        return io.BytesIO(self.data)


def request(method, path, body="", content_type="application/json"):
    body = body.encode()
    headers = (f"{method} {path} HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\n"
               f"Connection: close\r\nContent-Type: {content_type}\r\n"
               f"Content-Length: {len(body)}\r\n\r\n").encode()
    result = subprocess.run(
        ["docker", "exec", "-i", name, "/bin/nc", "-w", "3", "127.0.0.1", str(port)],
        input=headers + body, capture_output=True, timeout=5,
    )
    if result.returncode:
        raise RuntimeError(result.stderr.decode())
    response = http.client.HTTPResponse(ResponseSocket(result.stdout))
    response.begin()
    return response.status, response.read().decode()


def get_json(path):
    status, body = request("GET", path)
    assert status == 200, (status, body)
    return json.loads(body)


def eventually(check):
    for attempt in range(150):
        try:
            value = check()
            if value:
                return value
        except (RuntimeError, AssertionError, http.client.HTTPException):
            pass
        time.sleep(0.1)
    raise AssertionError("Expected local server state did not appear")


from datetime import datetime, timedelta, timezone

port = 9093
script = r"""set -eu
cat > /tmp/alertmanager.yml <<'ALERT_CONFIG'
route:
  receiver: preview
  group_by: [alertname, service]
  group_wait: 1h
  group_interval: 5m
  repeat_interval: 4h
receivers:
  - name: preview
ALERT_CONFIG
exec /bin/alertmanager --config.file=/tmp/alertmanager.yml --storage.path=/tmp/data \
  --web.listen-address=127.0.0.1:9093 --cluster.listen-address=
"""
try:
    subprocess.run(
        ["docker", "run", "-d", "--rm", "--name", name, "--network", "none",
         "--entrypoint", "/bin/sh", "prom/alertmanager:v0.28.1", "-c", script],
        check=True, capture_output=True, text=True, timeout=20,
    )
    eventually(lambda: request("GET", "/-/ready")[0] == 200)
    labels = [
        {"alertname": "QueueHigh", "service": "checkout", "instance": "api-a"},
        {"alertname": "QueueHigh", "service": "checkout", "instance": "api-b"},
        {"alertname": "QueueHigh", "service": "search", "instance": "api-c"},
        {"alertname": "DiskFull", "service": "checkout", "instance": "api-a"},
    ]
    assert request("POST", "/api/v2/alerts", json.dumps([{"labels": x} for x in labels]))[0] == 200
    groups = eventually(lambda: (groups if len(groups := get_json("/api/v2/alerts/groups")) == 3 else False))
    sizes = {g["labels"]["service"] + "/" + g["labels"]["alertname"]: len(g["alerts"]) for g in groups}
    assert sizes == {"checkout/QueueHigh": 2, "checkout/DiskFull": 1, "search/QueueHigh": 1}
    now = datetime.now(timezone.utc)
    silence = {"matchers": [{"name": "service", "value": "checkout", "isRegex": False, "isEqual": True}],
               "startsAt": (now - timedelta(seconds=1)).isoformat(),
               "endsAt": (now + timedelta(minutes=10)).isoformat(),
               "createdBy": "local-fixture", "comment": "Disposable grouping test"}
    status, body = request("POST", "/api/v2/silences", json.dumps(silence))
    assert status == 200
    silence_id = json.loads(body)["silenceID"]
    def silenced_state():
        alerts = get_json("/api/v2/alerts")
        count = sum(silence_id in x["status"]["silencedBy"] for x in alerts)
        return alerts if count == 3 else None
    alerts = eventually(silenced_state)
    assert len(alerts) == 4
    assert all((silence_id in x["status"]["silencedBy"]) == (x["labels"]["service"] == "checkout") for x in alerts)
    assert request("DELETE", "/api/v2/silence/" + silence_id)[0] == 200
    def unsilenced_state():
        alerts = get_json("/api/v2/alerts")
        return len(alerts) == 4 and all(silence_id not in x["status"]["silencedBy"] for x in alerts)
    eventually(unsilenced_state)
    print(json.dumps({"group_sizes": sizes, "silenced_alerts": 3,
                      "alerts_remain_visible": 4, "silence_expired": True}, indent=2))
finally:
    subprocess.run(["docker", "rm", "-f", name],
                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

输出应显示三个组：结算服务的 QueueHigh 有两个成员，结算服务的 DiskFull 有一个，搜索服务的 QueueHigh 也有一个。随后，临时静默匹配结算服务，抑制三条告警，但四条告警仍然都能通过接口看到；搜索服务告警不在静默范围内。程序再主动结束静默，并确认各告警的静默标识列表中不再包含该标识。

所有请求只经过容器回环地址，不发布宿主机端口，也不挂载目录，清理仅删除本次唯一名称的容器。这些断言验证成员关系与静默生命周期，没有测量通知计时精度，没有验证抑制规则，也没有证明下游送达。相关行为仍需使用实际路由、调度与接收器配置另行测试，才能得到对应的完成证据。

延伸阅读

来源