Capstone: Incident Response Tool

incident_response/
  __init__.py
  main.py           # CLI entry point
  alertmanager.py   # Alertmanager API client
  prometheus.py     # Prometheus query client
  kubernetes.py     # Kubernetes diagnostics collector
  slack.py          # Slack posting
  report.py         # Report generation
  config.py         # Configuration
requirements.txt
pyproject.toml

# config.py
import os
from dataclasses import dataclass

@dataclass
class Config:
    alertmanager_url: str
    prometheus_url: str
    slack_token: str
    slack_channel: str
    silence_duration_hours: int = 2

    @classmethod
    def from_env(cls) -> "Config":
        return cls(
            alertmanager_url=os.environ["ALERTMANAGER_URL"],
            prometheus_url=os.environ["PROMETHEUS_URL"],
            slack_token=os.environ["SLACK_BOT_TOKEN"],
            slack_channel=os.environ.get("SLACK_CHANNEL", "#incidents"),
            silence_duration_hours=int(os.environ.get("SILENCE_DURATION", "2")),
        )

import httpx
from datetime import datetime, timedelta

class AlertmanagerClient:
    def __init__(self, base_url: str):
        self.base_url = base_url
        self.client = httpx.Client(timeout=30)

    def get_firing_alerts(self, service: str | None = None) -> list[dict]:
        resp = self.client.get(f"{self.base_url}/api/v2/alerts")
        resp.raise_for_status()
        alerts = [a for a in resp.json() if a["status"]["state"] == "active"]
        if service:
            alerts = [a for a in alerts if a["labels"].get("job") == service]
        return alerts

    def silence_service(self, service: str, duration_hours: int, comment: str) -> str:
        now = datetime.utcnow()
        payload = {
            "matchers": [{"name": "job", "value": service, "isRegex": False}],
            "startsAt": now.isoformat() + "Z",
            "endsAt": (now + timedelta(hours=duration_hours)).isoformat() + "Z",
            "createdBy": "incident-response-bot",
            "comment": comment,
        }
        resp = self.client.post(f"{self.base_url}/api/v2/silences", json=payload)
        resp.raise_for_status()
        return resp.json()["silenceID"]

    def get_active_silences(self) -> list[dict]:
        resp = self.client.get(f"{self.base_url}/api/v2/silences")
        resp.raise_for_status()
        return [s for s in resp.json() if s["status"]["state"] == "active"]

from kubernetes import client, config
from collections import defaultdict

class KubeDiagnostics:
    def __init__(self):
        config.load_kube_config()
        self.v1 = client.CoreV1Api()
        self.apps_v1 = client.AppsV1Api()

    def get_pod_status(self, namespace: str, label_selector: str = "") -> list[dict]:
        pods = self.v1.list_namespaced_pod(namespace, label_selector=label_selector)
        results = []
        for pod in pods.items:
            results.append({
                "name": pod.metadata.name,
                "phase": pod.status.phase,
                "restarts": sum(
                    cs.restart_count
                    for cs in (pod.status.container_statuses or [])
                ),
                "ready": all(
                    cs.ready
                    for cs in (pod.status.container_statuses or [])
                ),
                "conditions": [
                    {"type": c.type, "status": c.status, "reason": c.reason}
                    for c in (pod.status.conditions or [])
                ],
            })
        return results

    def get_recent_events(self, namespace: str, field_selector: str = "") -> list[dict]:
        events = self.v1.list_namespaced_event(namespace, field_selector=field_selector)
        return [
            {
                "type": e.type,
                "reason": e.reason,
                "message": e.message,
                "count": e.count,
                "last_seen": e.last_timestamp.isoformat() if e.last_timestamp else "",
            }
            for e in events.items
            if e.type in ["Warning", "Error"]
        ]

    def get_deployment_status(self, namespace: str) -> list[dict]:
        deployments = self.apps_v1.list_namespaced_deployment(namespace)
        return [
            {
                "name": d.metadata.name,
                "replicas": d.spec.replicas,
                "ready_replicas": d.status.ready_replicas or 0,
                "updated_replicas": d.status.updated_replicas or 0,
                "available_replicas": d.status.available_replicas or 0,
            }
            for d in deployments.items
        ]

from datetime import datetime

def generate_incident_report(
    service: str,
    alerts: list[dict],
    pods: list[dict],
    events: list[dict],
    metrics: dict,
) -> str:
    """Generate a Markdown incident report."""
    lines = [
        f"# Incident Report: {service}",
        f"",
        f"**Generated:** {datetime.utcnow().isoformat()}Z",
        f"**Service:** {service}",
        f"**Firing alerts:** {len(alerts)}",
        f"",
    ]

    # Alert summary
    if alerts:
        lines.append("## Firing Alerts")
        lines.append("")
        for alert in alerts:
            name = alert["labels"].get("alertname", "unknown")
            severity = alert["labels"].get("severity", "unknown")
            summary = alert["annotations"].get("summary", "")
            lines.append(f"- **{name}** ({severity}): {summary}")
        lines.append("")

    # Pod status
    lines.append("## Pod Status")
    lines.append("")
    unhealthy = [p for p in pods if not p["ready"] or p["restarts"] > 0]
    if unhealthy:
        for pod in unhealthy:
            lines.append(f"- **{pod['name']}**: phase={pod['phase']}, restarts={pod['restarts']}")
    else:
        lines.append("All pods are healthy.")
    lines.append("")

    # Recent events
    if events:
        lines.append("## Recent Warning Events")
        lines.append("")
        for event in events[:10]:
            lines.append(f"- [{event['type']}] {event['reason']}: {event['message'][:100]}")
        lines.append("")

    # Metrics
    if metrics:
        lines.append("## Key Metrics")
        lines.append("")
        if "error_rate" in metrics:
            lines.append(f"- Error rate: {metrics['error_rate']:.2f}%")
        if "p99_latency" in metrics:
            lines.append(f"- P99 latency: {metrics['p99_latency']:.0f}ms")
        if "request_rate" in metrics:
            lines.append(f"- Request rate: {metrics['request_rate']:.1f} req/s")
        lines.append("")

    return "\n".join(lines)

import argparse
import sys
from config import Config
from alertmanager import AlertmanagerClient
from prometheus import PrometheusClient
from kubernetes import KubeDiagnostics
from report import generate_incident_report
from slack import post_to_slack

def respond_to_incident(service: str, config: Config):
    """Main incident response workflow."""
    print(f"Starting incident response for {service}...")

    # 1. Silence alerts
    am = AlertmanagerClient(config.alertmanager_url)
    silence_id = am.silence_service(
        service,
        config.silence_duration_hours,
        f"Incident response for {service}",
    )
    print(f"Silenced alerts: {silence_id}")

    # 2. Collect diagnostics
    kube = KubeDiagnostics()
    pods = kube.get_pod_status("production", label_selector=f"app={service}")
    events = kube.get_recent_events("production", field_selector="type=Warning")

    # 3. Collect metrics
    prom = PrometheusClient(config.prometheus_url)
    error_rate = prom.query(f'sum(rate(http_requests_total{{job="{service}",status=~"5.."}}[5m])) / sum(rate(http_requests_total{{job="{service}"}}[5m])) * 100')
    p99 = prom.query(f'histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket{{job="{service}"}}[5m])) by (le))')

    metrics = {}
    if error_rate["result"]:
        metrics["error_rate"] = float(error_rate["result"][0]["value"][1])
    if p99["result"]:
        metrics["p99_latency"] = float(p99["result"][0]["value"][1]) * 1000

    # 4. Generate report
    alerts = am.get_firing_alerts(service)
    report = generate_incident_report(service, alerts, pods, events, metrics)

    # 5. Post to Slack
    post_to_slack(config.slack_token, config.slack_channel, report)

    print(f"Incident report posted to {config.slack_channel}")
    print(f"Silence ID: {silence_id}")
    print(f"Silence expires in {config.silence_duration_hours} hours")

def main():
    parser = argparse.ArgumentParser(description="Incident Response Tool")
    parser.add_argument("service", help="Service name to respond to")
    args = parser.parse_args()

    config = Config.from_env()
    respond_to_incident(args.service, config)

if __name__ == "__main__":
    main()

# Set environment variables
export ALERTMANAGER_URL=http://alertmanager:9093
export PROMETHEUS_URL=http://prometheus:9090
export SLACK_BOT_TOKEN=xoxb-your-token
export SLACK_CHANNEL=#incidents

# Run incident response
python -m incident_response api-server

# Or invoke from Slack bot
/ops respond api-server