注解解释告警现场，同时保留稳定身份

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")


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)
    alerts = [{"labels": {"alertname": "QueueHigh", "service": "checkout", "instance": instance},
               "annotations": {"summary": "queue=" + str(depth), "runbook": "https://example.invalid/queue"}}
              for instance, depth in [("api-a", 6), ("api-b", 9)]]
    assert request("POST", "/api/v2/alerts", json.dumps(alerts))[0] == 200
    before = eventually(lambda: (a if len(a := get_json("/api/v2/alerts")) == 2 else False))
    fingerprints = {a["labels"]["instance"]: a["fingerprint"] for a in before}
    alerts[0]["annotations"]["summary"] = "queue=8"
    assert request("POST", "/api/v2/alerts", json.dumps([alerts[0]]))[0] == 200
    def updated_annotations():
        items = get_json("/api/v2/alerts")
        changed = any(x["labels"]["instance"] == "api-a" and
                      x["annotations"]["summary"] == "queue=8" for x in items)
        return items if changed else None
    updated = eventually(updated_annotations)
    assert len(updated) == 2
    assert {a["labels"]["instance"]: a["fingerprint"] for a in updated} == fingerprints
    # Supply the documented notification Data shape to the real template renderer.
    data = {"receiver": "preview", "status": "firing",
            "groupLabels": {"alertname": "QueueHigh", "service": "checkout"},
            "commonLabels": {"alertname": "QueueHigh", "service": "checkout"},
            "commonAnnotations": {"runbook": "https://example.invalid/queue"},
            "alerts": [{"status": "firing", "labels": a["labels"], "annotations": a["annotations"]}
                       for a in sorted(updated, key=lambda a: a["labels"]["instance"])]}
    template = '{{define "preview"}}{{.GroupLabels.service}}|common-summary={{with .CommonAnnotations.summary}}{{.}}{{else}}none{{end}}{{range .Alerts}}\n{{.Labels.instance}}:{{.Annotations.summary}}{{end}}{{end}}'
    for path, content in [("/tmp/preview.tmpl", template), ("/tmp/data.json", json.dumps(data))]:
        subprocess.run(["docker", "exec", "-i", name, "/bin/sh", "-c", "cat > " + path],
                       input=content, text=True, capture_output=True, check=True, timeout=5)
    rendered = subprocess.run(["docker", "exec", name, "/bin/amtool", "template", "render",
                               "--template.glob=/tmp/preview.tmpl", "--template.data=/tmp/data.json",
                               '--template.text={{ template "preview" . }}'],
                              text=True, capture_output=True, check=True, timeout=5).stdout.strip()
    assert rendered == "checkout|common-summary=none\napi-a:queue=8\napi-b:queue=9", rendered
    print(json.dumps({"alert_count": 2, "fingerprints_unchanged": True, "rendered": rendered}, indent=2))
finally:
    subprocess.run(["docker", "rm", "-f", name],
                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)