Build an AI Kubernetes Event Analyzer with Claude API

pip install anthropic slack-sdk kubernetes

#!/usr/bin/env python3
"""
Kubernetes Event Analyzer using Claude API
Collects cluster events, analyzes with AI, alerts on Slack
"""
 
import subprocess
import json
import os
import sys
from datetime import datetime
import anthropic
from slack_sdk.webhook import WebhookClient
 
ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY")
SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL", "")
CLUSTER_NAME = os.environ.get("CLUSTER_NAME", "my-cluster")
 
client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)
 
 
def run_kubectl(args: list[str]) -> str:
    """Run a kubectl command and return stdout."""
    try:
        result = subprocess.run(
            ["kubectl"] + args,
            capture_output=True,
            text=True,
            timeout=30
        )
        if result.returncode != 0:
            print(f"kubectl error: {result.stderr}", file=sys.stderr)
            return ""
        return result.stdout
    except subprocess.TimeoutExpired:
        print(f"kubectl timed out: {args}", file=sys.stderr)
        return ""
 
 
def collect_cluster_state() -> dict:
    """Collect events and pod status from the cluster."""
    print("Collecting cluster state...")
 
    # Get Warning events only (last 1 hour via field selector is not supported well,
    # so we filter in Python)
    events_raw = run_kubectl([
        "get", "events",
        "--all-namespaces",
        "--field-selector", "type=Warning",
        "--sort-by=.lastTimestamp",
        "-o", "wide"
    ])
 
    # Get all pods that are not Running or Completed
    pods_raw = run_kubectl([
        "get", "pods",
        "--all-namespaces",
        "-o", "wide"
    ])
 
    # Filter pods to problem states
    problem_pods = []
    for line in pods_raw.splitlines():
        if any(status in line for status in [
            "CrashLoopBackOff", "Error", "OOMKilled",
            "Pending", "ImagePullBackOff", "ErrImagePull",
            "Terminating", "Unknown", "Init:Error"
        ]):
            problem_pods.append(line)
 
    # Get recent events count for context
    all_events = run_kubectl(["get", "events", "--all-namespaces", "--sort-by=.lastTimestamp"])
    event_lines = [l for l in all_events.splitlines() if l.strip()]
 
    return {
        "warning_events": events_raw,
        "problem_pods": "\n".join(problem_pods) if problem_pods else "None",
        "total_events_seen": len(event_lines),
        "timestamp": datetime.utcnow().isoformat() + "Z"
    }
 
 
def analyze_with_claude(cluster_state: dict) -> str:
    """Send cluster state to Claude for analysis."""
 
    # Build the prompt
    prompt = f"""You are a Kubernetes SRE expert. Analyze these cluster events and pod issues.
 
CLUSTER: {CLUSTER_NAME}
TIMESTAMP: {cluster_state['timestamp']}
TOTAL EVENTS IN CLUSTER: {cluster_state['total_events_seen']}
 
=== WARNING EVENTS ===
{cluster_state['warning_events'] or 'No warning events found'}
 
=== PROBLEM PODS (non-Running/Completed) ===
{cluster_state['problem_pods']}
 
Provide your analysis in this exact format:
 
SEVERITY: [CRITICAL / WARNING / OK]
 
SUMMARY:
One sentence summary of the cluster health.
 
ISSUES FOUND:
- Issue 1: [What is happening] → [Which pods/namespaces affected]
- Issue 2: ...
(If no issues, write "None detected")
 
ROOT CAUSES:
- [Most likely cause for each issue]
 
RECOMMENDED ACTIONS:
1. [Most urgent action with exact kubectl command if applicable]
2. [Next action]
3. ...
 
Keep the response concise and actionable. Focus on patterns, not individual events."""
 
    print("Sending to Claude API for analysis...")
    message = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": prompt}
        ]
    )
 
    return message.content[0].text
 
 
def send_slack_alert(analysis: str, cluster_state: dict) -> None:
    """Post analysis to Slack if issues found and webhook is configured."""
    if not SLACK_WEBHOOK_URL:
        print("No Slack webhook configured, skipping alert")
        return
 
    # Only alert if severity is not OK
    if "SEVERITY: OK" in analysis:
        print("Cluster is healthy, no Slack alert needed")
        return
 
    severity = "CRITICAL" if "SEVERITY: CRITICAL" in analysis else "WARNING"
    emoji = ":red_circle:" if severity == "CRITICAL" else ":large_yellow_circle:"
 
    webhook = WebhookClient(SLACK_WEBHOOK_URL)
    response = webhook.send(
        text=f"{emoji} *K8s Cluster Alert — {CLUSTER_NAME}*",
        blocks=[
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"{emoji} Kubernetes Alert: {CLUSTER_NAME}"
                }
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Timestamp:* {cluster_state['timestamp']}\n*Severity:* {severity}"
                }
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"```{analysis[:2800]}```"
                }
            }
        ]
    )
 
    if response.status_code == 200:
        print("Slack alert sent successfully")
    else:
        print(f"Slack alert failed: {response.status_code} {response.body}")
 
 
def main():
    print(f"=== Kubernetes Event Analyzer ===")
    print(f"Cluster: {CLUSTER_NAME}")
    print(f"Time: {datetime.utcnow().isoformat()}Z")
    print()
 
    # Step 1: Collect cluster state
    cluster_state = collect_cluster_state()
 
    # Step 2: If no warning events and no problem pods, skip API call
    if (not cluster_state["warning_events"].strip() or
            cluster_state["warning_events"].strip() == "No resources found." or
            cluster_state["warning_events"].strip() == ""):
        if cluster_state["problem_pods"] == "None":
            print("Cluster is clean — no warning events, no problem pods. Exiting.")
            return
 
    # Step 3: Analyze with Claude
    analysis = analyze_with_claude(cluster_state)
 
    # Step 4: Print analysis
    print("\n=== AI ANALYSIS ===")
    print(analysis)
    print("==================\n")
 
    # Step 5: Send Slack alert if needed
    send_slack_alert(analysis, cluster_state)
 
 
if __name__ == "__main__":
    main()

FROM python:3.12-slim
RUN pip install anthropic slack-sdk kubernetes
COPY k8s_event_analyzer.py /app/
WORKDIR /app
CMD ["python", "k8s_event_analyzer.py"]

docker build -t your-registry/k8s-analyzer:latest .
docker push your-registry/k8s-analyzer:latest

apiVersion: v1
kind: ServiceAccount
metadata:
  name: k8s-analyzer
  namespace: monitoring
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: k8s-analyzer-reader
rules:
- apiGroups: [""]
  resources: ["pods", "events", "namespaces", "nodes"]
  verbs: ["get", "list"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: k8s-analyzer-binding
subjects:
- kind: ServiceAccount
  name: k8s-analyzer
  namespace: monitoring
roleRef:
  kind: ClusterRole
  name: k8s-analyzer-reader
  apiGroup: rbac.authorization.k8s.io

kubectl create secret generic k8s-analyzer-secrets \
  --from-literal=anthropic-api-key=sk-ant-your-key \
  --from-literal=slack-webhook-url=https://hooks.slack.com/your-webhook \
  -n monitoring

apiVersion: batch/v1
kind: CronJob
metadata:
  name: k8s-event-analyzer
  namespace: monitoring
spec:
  schedule: "*/15 * * * *"
  concurrencyPolicy: Forbid
  successfulJobsHistoryLimit: 3
  failedJobsHistoryLimit: 3
  jobTemplate:
    spec:
      template:
        spec:
          serviceAccountName: k8s-analyzer
          restartPolicy: OnFailure
          containers:
          - name: analyzer
            image: your-registry/k8s-analyzer:latest
            env:
            - name: CLUSTER_NAME
              value: "production-eks"
            - name: ANTHROPIC_API_KEY
              valueFrom:
                secretKeyRef:
                  name: k8s-analyzer-secrets
                  key: anthropic-api-key
            - name: SLACK_WEBHOOK_URL
              valueFrom:
                secretKeyRef:
                  name: k8s-analyzer-secrets
                  key: slack-webhook-url
            resources:
              requests:
                memory: "128Mi"
                cpu: "100m"
              limits:
                memory: "256Mi"
                cpu: "500m"

kubectl apply -f serviceaccount.yaml
kubectl apply -f cronjob.yaml
 
# Trigger a manual run to test
kubectl create job --from=cronjob/k8s-event-analyzer test-run -n monitoring
kubectl logs -l job-name=test-run -n monitoring -f