alertmanager

feishu.integrations.alertmanager

AlertmanagerIntegration dataclass

Mount Alertmanager webhooks on a Feishu gateway.

源代码位于： feishu/integrations/alertmanager.py

@dataclass(frozen=True)
class AlertmanagerIntegration:
    r"""Mount Alertmanager webhooks on a Feishu gateway."""

    receive_id: str
    receive_id_type: str = "chat_id"
    path: str = "/alerts/alertmanager"
    store: AlertmanagerMessageStore | None = None

    def routes(self, context: GatewayContext) -> list[Route]:
        return [
            create_alertmanager_route(
                context.config,
                context.client,
                self.receive_id,
                receive_id_type=self.receive_id_type,
                path=self.path,
                store=self.store,
            )
        ]

create_alertmanager_route

Python

create_alertmanager_route(config: GatewayConfig, client: Any, receive_id: str, *, receive_id_type: str = 'chat_id', path: str = '/alerts/alertmanager', store: AlertmanagerMessageStore | None = None) -> Route

Create a service-authenticated Alertmanager webhook route.

The route converts the standard Alertmanager webhook payload into a Feishu interactive card. It uses the Alertmanager groupKey or alert fingerprint as a stable event ID, so repeated notifications update the original Feishu card instead of creating a new post.

源代码位于： feishu/integrations/alertmanager.py

def create_alertmanager_route(
    config: GatewayConfig,
    client: Any,
    receive_id: str,
    *,
    receive_id_type: str = "chat_id",
    path: str = "/alerts/alertmanager",
    store: AlertmanagerMessageStore | None = None,
) -> Route:
    r"""Create a service-authenticated Alertmanager webhook route.

    The route converts the standard Alertmanager webhook payload into a Feishu
    interactive card. It uses the Alertmanager ``groupKey`` or alert
    ``fingerprint`` as a stable event ID, so repeated notifications update the
    original Feishu card instead of creating a new post.
    """
    return Route(
        path,
        _alertmanager_endpoint(config, client, receive_id, receive_id_type=receive_id_type, store=store),
        methods=["POST"],
    )

build_alertmanager_card

Python

build_alertmanager_card(payload: dict[str, Any]) -> dict[str, Any]

Render an Alertmanager webhook payload as a Feishu alert card.

源代码位于： feishu/integrations/alertmanager.py

def build_alertmanager_card(payload: dict[str, Any]) -> dict[str, Any]:
    r"""Render an Alertmanager webhook payload as a Feishu alert card."""
    status = _status(payload)
    priority = _priority(payload)
    labels = _dict(payload.get("commonLabels"))
    annotations = _dict(payload.get("commonAnnotations"))
    alerts = [alert for alert in payload.get("alerts", []) if isinstance(alert, dict)]
    title = _alert_title(payload)
    cluster = _text(labels.get("cluster"))
    severity = _text(labels.get("severity"))
    summary = _text(annotations.get("summary") or annotations.get("description"))
    display_id = alertmanager_display_id(payload)

    title_parts = [status.upper()]
    if priority:
        title_parts.append(priority)
    title_parts.append(title)
    if cluster:
        title_parts.append(cluster)

    lines = [
        f"**ID**: `{display_id}`",
        f"**Status**: {status}",
    ]
    if priority:
        lines.append(f"**Priority**: {priority}")
    if cluster:
        lines.append(f"**Cluster**: {cluster}")
    if severity:
        lines.append(f"**Severity**: {severity}")
    if summary and summary != title:
        lines.append(f"**Summary**: {summary}")
    if alerts:
        lines.append("")
        lines.append("**Instances**:")
        for alert in alerts[:8]:
            lines.append(f"- {_alert_line(alert)}")
    external_url = _text(payload.get("externalURL"))
    if external_url:
        lines.append("")
        lines.append(f"[Open Alertmanager]({external_url})")

    return alert_card(
        "\n".join(lines),
        title=" - ".join(title_parts),
        template=_template(status, priority),
    )

alertmanager_event_id

Python

alertmanager_event_id(payload: Mapping[str, Any]) -> str

Return the stable event ID for an Alertmanager webhook payload.

源代码位于： feishu/integrations/alertmanager.py

def alertmanager_event_id(payload: Mapping[str, Any]) -> str:
    r"""Return the stable event ID for an Alertmanager webhook payload."""
    group_key = _text(payload.get("groupKey"))
    if group_key:
        return group_key

    alerts = [alert for alert in payload.get("alerts", []) if isinstance(alert, Mapping)]
    fingerprints = sorted(_text(alert.get("fingerprint")) for alert in alerts if _text(alert.get("fingerprint")))
    if len(fingerprints) == 1:
        return fingerprints[0]
    if fingerprints:
        return "fingerprints:" + ",".join(fingerprints)

    stable = {
        "receiver": _text(payload.get("receiver")),
        "commonLabels": _dict(payload.get("commonLabels")),
        "commonAnnotations": _dict(payload.get("commonAnnotations")),
        "alerts": [_dict(alert.get("labels")) for alert in alerts],
    }
    digest = hashlib.sha256(json.dumps(stable, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
    return f"sha256:{digest}"

alertmanager_display_id

Python

alertmanager_display_id(payload: Mapping[str, Any]) -> str

Return a user-readable Alertmanager identifier for cards.

源代码位于： feishu/integrations/alertmanager.py

def alertmanager_display_id(payload: Mapping[str, Any]) -> str:
    r"""Return a user-readable Alertmanager identifier for cards."""
    labels = _dict(payload.get("commonLabels"))
    alerts = [alert for alert in payload.get("alerts", []) if isinstance(alert, Mapping)]
    if len(alerts) == 1:
        labels = {**_dict(alerts[0].get("labels")), **labels}

    parts = [
        _text(labels.get("alertname")),
        _text(labels.get("cluster")),
        _text(labels.get("node")),
        _text(labels.get("device")),
        _text(labels.get("service")),
        _text(labels.get("job")),
        _text(labels.get("instance")),
    ]
    readable = [part for part in parts if part]
    if readable:
        return "/".join(readable)

    fingerprints = sorted(_text(alert.get("fingerprint")) for alert in alerts if _text(alert.get("fingerprint")))
    if len(fingerprints) == 1:
        return fingerprints[0]
    if fingerprints:
        return "fingerprints:" + ",".join(fingerprints)
    return alertmanager_event_id(payload)