Build an AI Kubernetes Upgrade Impact Analyzer with Claude API

pip install anthropic pyyaml kubernetes

import yaml
import glob
import anthropic
 
client = anthropic.Anthropic()
 
 
def scan_manifests(manifest_dir: str) -> list[dict]:
    """Collect apiVersion/kind pairs from every manifest, including
    rendered Helm output — this needs to run against what's ACTUALLY applied,
    not just raw chart templates."""
    resources = []
    for filepath in glob.glob(f"{manifest_dir}/**/*.yaml", recursive=True):
        with open(filepath) as f:
            try:
                docs = yaml.safe_load_all(f)
                for doc in docs:
                    if doc and "apiVersion" in doc and "kind" in doc:
                        resources.append({
                            "file": filepath,
                            "apiVersion": doc["apiVersion"],
                            "kind": doc["kind"],
                            "name": doc.get("metadata", {}).get("name", "unknown"),
                        })
            except yaml.YAMLError:
                continue
    return resources
 
 
def scan_live_cluster() -> list[dict]:
    """Also check what's actually running — catches resources created
    outside of version-controlled manifests."""
    from kubernetes import client as k8s_client, config
    config.load_kube_config()
    api = k8s_client.ApiClient()
 
    # Use kubectl under the hood for simplicity across all API groups
    import subprocess
    result = subprocess.run(
        ["kubectl", "api-resources", "--verbs=list", "-o", "name"],
        capture_output=True, text=True
    )
    return result.stdout.splitlines()

ANALYZE_PROMPT = """We are planning a Kubernetes upgrade from {current_version} to {target_version}.
 
Here is a list of API versions and kinds currently in use across our manifests:
{resources}
 
Known deprecations/removals between these versions (from official changelog):
{changelog_notes}
 
For each resource that will break or needs migration:
1. What specifically breaks and at which version
2. The exact replacement API version/kind
3. Whether this is a simple find-replace or needs a structural change to the manifest
 
Respond with ONLY valid JSON:
{{"breaking_changes": [
  {{"file": "...", "resource": "...", "issue": "...", "fix": "...", "complexity": "simple"|"moderate"|"complex"}}
], "safe_to_upgrade": true/false}}"""
 
 
def analyze_upgrade_impact(resources: list[dict], current_version: str, target_version: str, changelog_notes: str) -> dict:
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=3000,
        messages=[{
            "role": "user",
            "content": ANALYZE_PROMPT.format(
                current_version=current_version, target_version=target_version,
                resources=resources, changelog_notes=changelog_notes,
            )
        }]
    )
    text = response.content[0].text.strip()
    if text.startswith("```"):
        text = text.split("```")[1].replace("json", "", 1).strip()
    import json
    return json.loads(text)

def generate_fixes(breaking_changes: list[dict]):
    for change in breaking_changes:
        if change["complexity"] != "simple":
            continue    # Leave moderate/complex changes for manual review
 
        with open(change["file"]) as f:
            content = f.read()
 
        fix_prompt = f"""This manifest needs a fix for a Kubernetes API deprecation:
Issue: {change['issue']}
Required fix: {change['fix']}
 
Current file content:
{content}
 
Return ONLY the corrected YAML, no explanation."""
 
        response = client.messages.create(
            model="claude-sonnet-5", max_tokens=2000,
            messages=[{"role": "user", "content": fix_prompt}]
        )
        fixed_content = response.content[0].text.strip()
        if fixed_content.startswith("```"):
            fixed_content = fixed_content.split("```")[1].replace("yaml", "", 1).strip()
 
        with open(change["file"], "w") as f:
            f.write(fixed_content)
 
        print(f"Fixed: {change['file']} — {change['issue']}")

python upgrade_analyzer.py --manifests ./k8s --current 1.29 --target 1.32
 
# Upgrade Impact Report: 1.29 -> 1.32
#
# BREAKING (3 found):
# k8s/legacy-hpa.yaml — HorizontalPodAutoscaler uses autoscaling/v2beta2,
#   removed in 1.31. Fix: autoscaling/v2. Complexity: simple
# k8s/network-policy.yaml — uses deprecated selector syntax. Complexity: moderate
# k8s/custom-webhook.yaml — AdmissionregistrationV1beta1 removed in 1.30.
#   Fix: admissionregistration.k8s.io/v1. Complexity: simple
#
# Auto-fixing 2 simple cases... done.
# 1 moderate-complexity change needs manual review: k8s/network-policy.yaml

# Complementary tools worth running alongside this
pluto detect-helm --target-versions k8s=v1.32.0
kubent --target-version 1.32.0