Alertmanager & PagerDuty API

import httpx
from datetime import datetime, timedelta

class AlertmanagerClient:
    def __init__(self, base_url: str = "http://localhost:9093"):
        self.base_url = base_url
        self.client = httpx.Client(timeout=30)

    def get_alerts(self) -> list[dict]:
        """Get all active alerts."""
        resp = self.client.get(f"{self.base_url}/api/v2/alerts")
        resp.raise_for_status()
        return resp.json()

    def get_silences(self) -> list[dict]:
        """Get all active silences."""
        resp = self.client.get(f"{self.base_url}/api/v2/silences")
        resp.raise_for_status()
        return resp.json()

    def create_silence(self, matchers: list[dict], duration_hours: int = 2, comment: str = "") -> str:
        """Create a silence for matching alerts."""
        now = datetime.utcnow()
        payload = {
            "matchers": matchers,
            "startsAt": now.isoformat() + "Z",
            "endsAt": (now + timedelta(hours=duration_hours)).isoformat() + "Z",
            "createdBy": "python-automation",
            "comment": comment,
        }
        resp = self.client.post(f"{self.base_url}/api/v2/silences", json=payload)
        resp.raise_for_status()
        return resp.json()["silenceID"]

    def delete_silence(self, silence_id: str):
        """Delete (expire) a silence."""
        resp = self.client.delete(f"{self.base_url}/api/v2/silence/{silence_id}")
        resp.raise_for_status()

am = AlertmanagerClient("http://alertmanager:9093")

def get_alerts_by_severity(client: AlertmanagerClient) -> dict:
    """Group alerts by severity."""
    alerts = client.get_alerts()
    by_severity = {"critical": [], "warning": [], "info": []}

    for alert in alerts:
        severity = alert["labels"].get("severity", "info")
        by_severity.setdefault(severity, []).append({
            "name": alert["labels"].get("alertname", "unknown"),
            "instance": alert["labels"].get("instance", "unknown"),
            "summary": alert["annotations"].get("summary", ""),
            "started": alert["startsAt"],
            "fingers": alert.get("fingerprint", ""),
        })

    return by_severity

def count_firing_by_service(client: AlertmanagerClient) -> dict:
    """Count alerts grouped by service."""
    alerts = client.get_alerts()
    counts = {}
    for alert in alerts:
        service = alert["labels"].get("job", "unknown")
        counts[service] = counts.get(service, 0) + 1
    return counts

# Usage
alerts = get_alerts_by_severity(am)
print(f"Critical: {len(alerts['critical'])}")
print(f"Warning: {len(alerts['warning'])}")

for alert in alerts["critical"]:
    print(f"  CRITICAL: {alert['name']} on {alert['instance']}")

def silence_maintenance(
    client: AlertmanagerClient,
    service: str,
    duration_hours: int = 2,
) -> str:
    """Silence all alerts for a service during maintenance."""
    matchers = [
        {"name": "job", "value": service, "isRegex": False},
    ]
    return client.create_silence(
        matchers=matchers,
        duration_hours=duration_hours,
        comment=f"Silenced for maintenance: {service}",
    )

def silence_specific_alert(
    client: AlertmanagerClient,
    alert_name: str,
    instance: str,
    hours: int = 1,
) -> str:
    """Silence a specific alert on a specific instance."""
    matchers = [
        {"name": "alertname", "value": alert_name, "isRegex": False},
        {"name": "instance", "value": instance, "isRegex": False},
    ]
    return client.create_silence(
        matchers=matchers,
        duration_hours=hours,
        comment=f"Silenced {alert_name} on {instance}",
    )

# Silence during deployment
silence_id = silence_maintenance(am, "api-server", duration_hours=1)
print(f"Created silence: {silence_id}")

# ... perform deployment ...

# Expire the silence early
am.delete_silence(silence_id)

import httpx

class PagerDutyClient:
    def __init__(self, token: str):
        self.client = httpx.Client(
            base_url="https://api.pagerduty.com",
            headers={
                "Authorization": f"Token token={token}",
                "Accept": "application/vnd.pagerduty+json;version=2",
            },
            timeout=30,
        )

    def get_incidents(self, status: str = "triggered") -> list[dict]:
        """Get incidents by status."""
        resp = self.client.get("/incidents", params={
            "statuses[]": status,
            "sort_by": "created_at:desc",
        })
        resp.raise_for_status()
        return resp.json()["incidents"]

    def acknowledge_incident(self, incident_id: str, requester: str):
        """Acknowledge an incident."""
        resp = self.client.put(f"/incidents/{incident_id}", json={
            "incident": {
                "type": "incident_reference",
                "status": "acknowledged",
            },
        }, headers={"From": requester})
        resp.raise_for_status()

    def resolve_incident(self, incident_id: str, requester: str):
        """Resolve an incident."""
        resp = self.client.put(f"/incidents/{incident_id}", json={
            "incident": {
                "type": "incident_reference",
                "status": "resolved",
            },
        }, headers={"From": requester})
        resp.raise_for_status()

    def create_incident(self, title: str, service_id: str, urgency: str = "high") -> dict:
        """Create a new incident."""
        resp = self.client.post("/incidents", json={
            "incident": {
                "type": "incident",
                "title": title,
                "service": {"id": service_id},
                "urgency": urgency,
            },
        })
        resp.raise_for_status()
        return resp.json()["incident"]

# Usage
pd = PagerDutyClient(token="your-api-token")
incidents = pd.get_incidents(status="triggered")
for inc in incidents:
    print(f"{inc['title']}: {inc['urgency']} ({inc['created_at']})")

from datetime import datetime

def incident_response(alertmanager: AlertmanagerClient, pagerduty: PagerDutyClient):
    """Automated response to critical alerts."""
    alerts = alertmanager.get_alerts()
    critical = [a for a in alerts if a["labels"].get("severity") == "critical"]

    if not critical:
        return

    # Group by service
    by_service = {}
    for alert in critical:
        service = alert["labels"].get("job", "unknown")
        by_service.setdefault(service, []).append(alert)

    for service, alerts in by_service.items():
        # Create PagerDuty incident
        incident = pagerduty.create_incident(
            title=f"Critical alerts firing for {service} ({len(alerts)} alerts)",
            service_id="PXXXXXX",
            urgency="high",
        )

        # Silence the Alertmanager alerts (they are being handled by PagerDuty)
        for alert in alerts:
            matchers = [
                {"name": "alertname", "value": alert["labels"]["alertname"]},
                {"name": "instance", "value": alert["labels"].get("instance", "")},
            ]
            alertmanager.create_silence(
                matchers=matchers,
                duration_hours=4,
                comment=f"Incident {incident['id']} created",
            )

        print(f"Incident created: {incident['html_url']}")

def auto_resolve_incidents(
    alertmanager: AlertmanagerClient,
    pagerduty: PagerDutyClient,
):
    """Resolve PagerDuty incidents when Alertmanager alerts resolve."""
    # Get currently firing alerts
    firing = alertmanager.get_alerts()
    firing_names = {a["labels"].get("alertname") for a in firing}

    # Get open PagerDuty incidents
    incidents = pagerduty.get_incidents(status="triggered")

    for incident in incidents:
        # Extract alert name from incident title
        # Assuming title format: "Alert: AlertName firing"
        title = incident.get("title", "")
        if "Alert:" in title:
            alert_name = title.split("Alert:")[1].split("firing")[0].strip()
            if alert_name not in firing_names:
                # Alert has resolved, resolve the incident
                pagerduty.resolve_incident(incident["id"], requester="auto-resolve")
                print(f"Resolved incident: {incident['title']}")