Build an AI Kubernetes Pod Debugger with Claude API

pip install anthropic subprocess-run rich
# kubectl must be configured and pointed at your cluster

import anthropic
import subprocess
import json
import sys
from rich.console import Console
from rich.panel import Panel
from rich.markdown import Markdown
 
client = anthropic.Anthropic()
console = Console()
 
 
def run_kubectl(args: list[str]) -> tuple[str, str]:
    """Run kubectl and return (stdout, stderr)."""
    result = subprocess.run(
        ["kubectl"] + args,
        capture_output=True,
        text=True,
        timeout=30
    )
    return result.stdout, result.stderr
 
 
def collect_pod_context(pod_name: str, namespace: str) -> dict:
    """Collect all diagnostic info for a pod."""
    console.print(f"[dim]Collecting data for pod: {pod_name} in {namespace}[/dim]")
    context = {}
 
    # Pod describe
    stdout, _ = run_kubectl(["describe", "pod", pod_name, "-n", namespace])
    context["describe"] = stdout[:5000]    # Limit to avoid token waste
 
    # Pod logs (current)
    stdout, stderr = run_kubectl(["logs", pod_name, "-n", namespace, "--tail=100"])
    context["logs_current"] = stdout or stderr
 
    # Pod logs (previous container if crashed)
    stdout, stderr = run_kubectl(["logs", pod_name, "-n", namespace, "--previous", "--tail=100"])
    context["logs_previous"] = stdout or "No previous container logs"
 
    # Pod events
    stdout, _ = run_kubectl([
        "get", "events", "-n", namespace,
        "--field-selector", f"involvedObject.name={pod_name}",
        "--sort-by=.metadata.creationTimestamp"
    ])
    context["events"] = stdout
 
    # Pod resource usage (requires metrics-server)
    stdout, stderr = run_kubectl(["top", "pod", pod_name, "-n", namespace, "--containers"])
    context["resource_usage"] = stdout or "Metrics server not available"
 
    # Pod status JSON (structured data)
    stdout, _ = run_kubectl(["get", "pod", pod_name, "-n", namespace, "-o", "json"])
    try:
        pod_json = json.loads(stdout)
        # Extract just the status fields we need
        context["status"] = {
            "phase": pod_json.get("status", {}).get("phase"),
            "conditions": pod_json.get("status", {}).get("conditions", []),
            "containerStatuses": pod_json.get("status", {}).get("containerStatuses", []),
            "initContainerStatuses": pod_json.get("status", {}).get("initContainerStatuses", []),
        }
        # Also get resource requests/limits
        containers = pod_json.get("spec", {}).get("containers", [])
        context["resource_config"] = [
            {
                "name": c["name"],
                "resources": c.get("resources", {}),
                "readinessProbe": c.get("readinessProbe"),
                "livenessProbe": c.get("livenessProbe"),
            }
            for c in containers
        ]
    except json.JSONDecodeError:
        context["status"] = "Could not parse pod JSON"
 
    return context
 
 
def analyze_with_claude(pod_name: str, namespace: str, context: dict) -> str:
    """Send pod context to Claude for root cause analysis."""
 
    prompt = f"""You are a senior Kubernetes SRE. Analyze this broken pod and identify the root cause.
 
## Pod: {pod_name} in namespace: {namespace}
 
## Current Status
{json.dumps(context.get('status', {}), indent=2)}
 
## Resource Configuration
{json.dumps(context.get('resource_config', {}), indent=2)}
 
## Pod Events
{context.get('events', 'No events')}
 
## Current Container Logs (last 100 lines)
~~~
{context.get('logs_current', 'No logs')}
~~~
 
## Previous Container Logs (if crashed)
~~~
{context.get('logs_previous', 'No previous logs')}
~~~
 
## Resource Usage
{context.get('resource_usage', 'Unknown')}
 
## Pod Describe (partial)
{context.get('describe', '')[:2000]}
 
Provide:
1. **Root Cause** (1-2 sentences, be specific — e.g., "OOMKilled: container exceeded 256Mi memory limit" not just "memory issue")
2. **Evidence** (which log line or event proves this)
3. **Fix Commands** (exact kubectl or YAML changes to resolve, copy-paste ready)
4. **Prevention** (what change prevents this in future)
 
Be direct. If you see a crash, name the error. If it's an OOMKill, say the memory numbers. If it's a readiness probe failing, show the exact path and what it returned."""
 
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=2000,
        messages=[{"role": "user", "content": prompt}]
    )
 
    return response.content[0].text
 
 
def debug_pod(pod_name: str, namespace: str = "default"):
    """Main debug function."""
    console.print(Panel(f"[bold]AI Pod Debugger[/bold]\nAnalyzing: [cyan]{pod_name}[/cyan] in [green]{namespace}[/green]"))
 
    # Verify pod exists
    _, stderr = run_kubectl(["get", "pod", pod_name, "-n", namespace])
    if stderr and "not found" in stderr:
        console.print(f"[red]Pod not found: {pod_name} in {namespace}[/red]")
        sys.exit(1)
 
    # Collect data
    with console.status("Collecting pod diagnostics..."):
        context = collect_pod_context(pod_name, namespace)
 
    console.print("[green]Data collected. Analyzing with Claude...[/green]\n")
 
    # Analyze
    analysis = analyze_with_claude(pod_name, namespace, context)
 
    console.print(Panel(Markdown(analysis), title="[bold green]Root Cause Analysis[/bold green]", border_style="green"))
 
 
def find_broken_pods(namespace: str = "default") -> list[str]:
    """Find pods that are not Running or Succeeded."""
    stdout, _ = run_kubectl([
        "get", "pods", "-n", namespace,
        "--field-selector=status.phase!=Running,status.phase!=Succeeded",
        "-o", "jsonpath={.items[*].metadata.name}"
    ])
    return stdout.split() if stdout.strip() else []
 
 
if __name__ == "__main__":
    if len(sys.argv) >= 2:
        pod = sys.argv[1]
        ns = sys.argv[2] if len(sys.argv) >= 3 else "default"
        debug_pod(pod, ns)
    else:
        # Auto-discover broken pods
        console.print("[yellow]No pod specified. Scanning for broken pods...[/yellow]\n")
        broken = find_broken_pods()
        if not broken:
            console.print("[green]No broken pods found![/green]")
        else:
            console.print(f"Found {len(broken)} broken pods: {', '.join(broken)}")
            for pod in broken[:3]:    # Debug first 3
                debug_pod(pod)
                console.print()

# Debug a specific pod
python pod_debugger.py myapp-xyz-abc production
 
# Auto-discover and debug broken pods
python pod_debugger.py
 
# Example output:
# Root Cause Analysis
# -------------------
# Root Cause: OOMKilled — the container exceeded its 256Mi memory limit.
# Redis is loading a 180MB dataset on startup, which combined with
# the Python process overhead exceeds the configured limit.
#
# Evidence: containerStatuses shows "reason: OOMKilled" and
# previous container logs end with "Loading RDB file" before kill signal.
#
# Fix Commands:
# kubectl patch deployment myapp -p '{"spec":{"template":{"spec":
# {"containers":[{"name":"api","resources":{"limits":{"memory":"512Mi"}}}]}}}}'
#
# Prevention: Set memory requests = limits (currently they differ by 4x).
# Add Vertical Pod Autoscaler to auto-tune resource limits.

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: pod-debugger
rules:
- apiGroups: [""]
  resources: ["pods", "pods/log", "events"]
  verbs: ["get", "list"]
- apiGroups: ["metrics.k8s.io"]
  resources: ["pods"]
  verbs: ["get", "list"]