一个事故可能产生很多告警，但一组告警并不能证明它们具有同一个原因。Alertmanager 根据配置的标签与路由上下文组织告警，便于一起处理通知，组内仍然保留各条告警的独立身份。静默则是另一种操作：在指定时间窗口内抑制匹配通知，并不会替应用解决故障。理解这些差异，才能准确解释告警页面与通知数量的变化。
分组键决定哪些细节放在一起 配置 group_by: [alertname, service] 时，两条属于结算服务的 QueueHigh 告警可以进入同组，即使实例标签不同。搜索服务的 QueueHigh 属于另一组，结算服务的 DiskFull 也因为告警名不同而单独成组。分组依据被选择的标签值，不要求所有成员具有完全相同的指纹。通知被合并后，实例级细节仍然可以从成员列表查看。
把实例加入分组键会增加组数，也可能抵消跨副本合并的目的；省略服务又可能把不同负责人或工作负载混在一起。应围绕排查方式和责任归属选择分组键，同时保留查看成员的入口。特殊值 ... 表示使用所有标签分组，实际效果接近关闭更大范围的聚合，不能把它当作自动寻找合理分组方式的通配符。
路由上下文同样重要。发往不同接收器的告警，即使所选分组标签相同，也不一定组成同一份通知。应验证路由与分组的组合行为。负责整个服务的接收器可能适合合并多个实例，另一个集成则可能需要不同分组契约。任何一种设置都不会自动证明这些告警具有相同根因，因果关系仍然需要从事件与运行证据判断。
初次等待、组内变化和重复提醒使用不同计时 group_wait 延迟新组的第一份通知，让相近时间到来的告警可以加入。某条告警若在初次等待结束前已经恢复，可能完全不产生通知。group_interval 决定之后检查组内更新的节奏，包括新增触发成员或恢复成员；它不是承诺每隔这么久无条件重发所有尚未结束的组。
没有新增触发或恢复的告警身份需要更新时，repeat_interval 控制重复提醒；单独改变注解文字并不一定触发新通知。是否到达重复时机是在组周期上检查的，因此应选择兼容的整数倍，而不是把它理解为完全独立且精确的闹钟。发送处理与失败还可能增加延迟。应依据响应目标设置这些参数，并将规则等待时长与通知调度分别考虑，避免只看其中一个参数就承诺发现到送达的时间。
通知历史同样存在保留边界。如果 repeat_interval 超过 Alertmanager 的数据保留时间，文档说明可能按保留边界重复发送。因此，把提醒间隔设置得很长，不保证这段时间里绝无重复消息。通知日志的保留期限，与具有明确结束时间的静默，是不同生命周期，应分别纳入运行检查。
坏掉的样子 分组太细，会让一组副本故障产生很多通知；分组太宽，又可能看不出具体服务需要处理。页面只显示一个组，不意味着只剩一条告警。判断事故是否改善前，应展开成员并核对分组键，不能把通知变少直接当成故障减少的证据。通知数量也可能只是由策略变更或抑制行为造成。
静默条件过宽，可能抑制计划维护范围之外的告警。应限制时间窗口，并检查哪些活动告警实际匹配。静默到期只是恢复通知资格，不会修复应用，也不保证立即发出一条消息。抑制规则则是独立的配置关系，由活动的源告警抑制匹配目标通知，与人工设置有时限静默不同，不能把两者的原因和结束条件混为一谈。
快速验证 把程序保存为 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 也有一个。随后，临时静默匹配结算服务，抑制三条告警，但四条告警仍然都能通过接口看到；搜索服务告警不在静默范围内。程序再主动结束静默，并确认各告警的静默标识列表中不再包含该标识。
...
