alertmanager

feishu.integrations.alertmanager

AlertmanagerMessageStore

Bases: EventMessageStore, Protocol

Store Feishu message IDs and per-alert lifecycle revisions.

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

class AlertmanagerMessageStore(EventMessageStore, Protocol):
    r"""Store Feishu message IDs and per-alert lifecycle revisions."""

    def get_alert_revisions(self, event_id: str) -> _AlertRevisions:
        r"""Return lifecycle revisions keyed by Alertmanager fingerprint."""
        ...

    def set_alert_revisions(
        self,
        event_id: str,
        revisions: _AlertRevisions,
    ) -> None:
        r"""Persist lifecycle revisions for an Alertmanager notification group."""
        ...

get_alert_revisions

Python

get_alert_revisions(event_id: str) -> _AlertRevisions

Return lifecycle revisions keyed by Alertmanager fingerprint.

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

def get_alert_revisions(self, event_id: str) -> _AlertRevisions:
    r"""Return lifecycle revisions keyed by Alertmanager fingerprint."""
    ...

set_alert_revisions

Python

set_alert_revisions(event_id: str, revisions: _AlertRevisions) -> None

Persist lifecycle revisions for an Alertmanager notification group.

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

def set_alert_revisions(
    self,
    event_id: str,
    revisions: _AlertRevisions,
) -> None:
    r"""Persist lifecycle revisions for an Alertmanager notification group."""
    ...

InMemoryAlertmanagerStore

Bases: InMemoryEventMessageStore

Process-local Alertmanager message and alert lifecycle state.

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

class InMemoryAlertmanagerStore(InMemoryEventMessageStore):
    r"""Process-local Alertmanager message and alert lifecycle state."""

    def __init__(self) -> None:
        super().__init__()
        self._alert_revisions: dict[str, _AlertRevisions] = {}

    def get_alert_revisions(self, event_id: str) -> _AlertRevisions:
        return dict(self._alert_revisions.get(event_id, {}))

    def set_alert_revisions(
        self,
        event_id: str,
        revisions: _AlertRevisions,
    ) -> None:
        self._alert_revisions[event_id] = dict(revisions)

JsonFileAlertmanagerStore

Bases: JsonFileEventMessageStore

Single-process JSON-file Alertmanager message and lifecycle state.

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

class JsonFileAlertmanagerStore(JsonFileEventMessageStore):
    r"""Single-process JSON-file Alertmanager message and lifecycle state."""

    def __init__(self, path: str | Path) -> None:
        super().__init__(path)
        self._alerts_path = Path(f"{self.path}.alerts").resolve()
        self._alerts_lock = _json_path_lock(self._alerts_path)

    def get_alert_revisions(self, event_id: str) -> _AlertRevisions:
        with self._alerts_lock:
            return dict(self._read_alert_revisions().get(event_id, {}))

    def set_alert_revisions(
        self,
        event_id: str,
        revisions: _AlertRevisions,
    ) -> None:
        with self._alerts_lock:
            state = self._read_alert_revisions()
            state[event_id] = dict(revisions)
            serializable = {
                group: {fingerprint: [revision[0], revision[1]] for fingerprint, revision in alerts.items()}
                for group, alerts in state.items()
            }
            _write_json(self._alerts_path, serializable)

    def _read_alert_revisions(self) -> dict[str, _AlertRevisions]:
        data = _read_json_object(self._alerts_path)
        if not isinstance(data, Mapping):
            return {}
        state: dict[str, _AlertRevisions] = {}
        for event_id, raw_alerts in data.items():
            if not isinstance(raw_alerts, Mapping):
                continue
            alerts: _AlertRevisions = {}
            for fingerprint, raw_revision in raw_alerts.items():
                if (
                    isinstance(raw_revision, list)
                    and len(raw_revision) == 2
                    and isinstance(raw_revision[0], (int, float))
                    and isinstance(raw_revision[1], int)
                ):
                    alerts[str(fingerprint)] = (
                        float(raw_revision[0]),
                        raw_revision[1],
                    )
            state[str(event_id)] = alerts
        return state

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
    card_builder: AlertmanagerCardBuilder | 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,
                card_builder=self.card_builder,
            )
        ]

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, card_builder: AlertmanagerCardBuilder | None = None) -> Route

Create a service-authenticated Alertmanager webhook route.

The route converts the standard Alertmanager webhook payload into a Feishu interactive card. Repeated notifications for one incident update the original card, while a single-alert incident with a new startsAt posts a new card so operators receive a fresh notification. The built-in stores serialize delivery within one process; multi-worker deployments require a gateway-level distributed delivery lock.

源代码位于： 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,
    card_builder: AlertmanagerCardBuilder | None = None,
) -> Route:
    r"""Create a service-authenticated Alertmanager webhook route.

    The route converts the standard Alertmanager webhook payload into a Feishu
    interactive card. Repeated notifications for one incident update the
    original card, while a single-alert incident with a new ``startsAt`` posts
    a new card so operators receive a fresh notification. The built-in stores
    serialize delivery within one process; multi-worker deployments require a
    gateway-level distributed delivery lock.
    """
    return Route(
        path,
        _alertmanager_endpoint(
            config,
            client,
            receive_id,
            receive_id_type=receive_id_type,
            store=store,
            card_builder=card_builder,
        ),
        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 and cluster.casefold() not in title.casefold():
        title_parts.append(cluster)

    metadata = [
        f"**ID**: `{display_id}`",
        f"**Status**: {status}",
    ]
    if priority:
        metadata.append(f"**Priority**: {priority}")
    if cluster:
        metadata.append(f"**Cluster**: {cluster}")
    if severity:
        metadata.append(f"**Severity**: {severity}")
    if summary and summary != title:
        metadata.append(f"**Summary**: {summary}")

    card = Card().header(" - ".join(title_parts), template=_template(status, priority)).markdown("\n".join(metadata))
    if alerts:
        instances = ["**Instances**:"]
        instances.extend(f"- {_alert_instance_line(payload, alert)}" for alert in alerts[:8])
        card.markdown("\n".join(instances))
        for block in _alert_detail_blocks(payload, alerts):
            card.markdown(block)
    external_url = _text(payload.get("externalURL"))
    if external_url:
        card.markdown(f"[Open Alertmanager]({external_url})")
    return card.to_dict()

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:
        alert = alerts[0]
        labels = {**labels, **_dict(alert.get("labels"))}
        cluster = _text(labels.get("cluster"))
        resource = _text(labels.get("node") or labels.get("service") or labels.get("instance") or labels.get("job"))
        device = _text(labels.get("device"))
        started_at = _display_timestamp(alert.get("startsAt"))
        concise = [part for part in (cluster, resource, device) if part]
        if concise and started_at:
            return f"{'/'.join(concise)}@{started_at}"

    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)