Build an AI Kubernetes Resource Quota Advisor with Claude API

pip install anthropic kubernetes

import anthropic
from kubernetes import client, config
from datetime import datetime, timedelta
 
client_ai = anthropic.Anthropic()
 
 
def get_namespace_usage_history(namespace: str, days: int = 30) -> dict:
    """Pull actual resource usage over time — not current snapshot,
    the trend and peak-to-average pattern matter more than a point-in-time read."""
    import requests
    prometheus_query = lambda q: requests.get(
        "http://prometheus:9090/api/v1/query_range",
        params={"query": q, "start": (datetime.utcnow() - timedelta(days=days)).timestamp(),
                "end": datetime.utcnow().timestamp(), "step": "1h"}
    ).json()["data"]["result"]
 
    return {
        "cpu_usage_trend": prometheus_query(f'sum(rate(container_cpu_usage_seconds_total{{namespace="{namespace}"}}[5m]))'),
        "memory_usage_trend": prometheus_query(f'sum(container_memory_working_set_bytes{{namespace="{namespace}"}})'),
        "pod_count_trend": prometheus_query(f'count(kube_pod_info{{namespace="{namespace}"}})'),
        "current_quota": get_current_quota(namespace),
    }
 
 
def get_current_quota(namespace: str) -> dict:
    config.load_kube_config()
    v1 = client.CoreV1Api()
    quotas = v1.list_namespaced_resource_quota(namespace)
    if not quotas.items:
        return {"exists": False}
    q = quotas.items[0]
    return {"exists": True, "hard": q.status.hard, "used": q.status.used}

import json
 
RECOMMEND_PROMPT = """Recommend a ResourceQuota for this Kubernetes namespace
based on its actual usage history.
 
Namespace: {namespace}
30-day CPU usage trend (cores over time): {cpu_trend}
30-day memory usage trend (bytes over time): {memory_trend}
30-day pod count trend: {pod_count_trend}
Current quota (if any): {current_quota}
 
Determine:
1. Peak usage observed (not average — quotas need headroom for legitimate spikes)
2. Growth trend — is usage trending up, flat, or down over the 30 days?
   Recommend headroom accordingly (more headroom for growing namespaces)
3. If a current quota exists and usage is consistently near or hitting it,
   flag this as likely causing throttled/blocked deployments
4. If a current quota exists and usage never comes close, flag as
   over-provisioned (wasting cluster capacity reservation)
 
Respond with ONLY valid JSON:
{{"recommended_quota": {{"requests.cpu": "...", "requests.memory": "...",
  "limits.cpu": "...", "limits.memory": "...", "pods": "..."}},
  "reasoning": "...", "current_quota_issue": "too_tight"|"over_provisioned"|"appropriate"|"none_set"}}"""
 
 
def recommend_quota(namespace: str, usage_history: dict) -> dict:
    response = client_ai.messages.create(
        model="claude-sonnet-5",
        max_tokens=800,
        messages=[{
            "role": "user",
            "content": RECOMMEND_PROMPT.format(
                namespace=namespace,
                cpu_trend=summarize_trend(usage_history["cpu_usage_trend"]),
                memory_trend=summarize_trend(usage_history["memory_usage_trend"]),
                pod_count_trend=summarize_trend(usage_history["pod_count_trend"]),
                current_quota=usage_history["current_quota"],
            )
        }]
    )
    text = response.content[0].text.strip()
    if text.startswith("```"):
        text = text.split("```")[1].replace("json", "", 1).strip()
    return json.loads(text)
 
 
def summarize_trend(series: list) -> dict:
    """Reduce a raw time series to peak/average/trend-direction — keeps
    the prompt compact instead of dumping thousands of data points."""
    if not series or not series[0].get("values"):
        return {"peak": 0, "average": 0, "trend": "no_data"}
    values = [float(v[1]) for v in series[0]["values"]]
    first_half_avg = sum(values[:len(values)//2]) / max(len(values)//2, 1)
    second_half_avg = sum(values[len(values)//2:]) / max(len(values) - len(values)//2, 1)
    trend = "increasing" if second_half_avg > first_half_avg * 1.1 else \
            "decreasing" if second_half_avg < first_half_avg * 0.9 else "flat"
    return {"peak": max(values), "average": sum(values) / len(values), "trend": trend}

# Recommended ResourceQuota for namespace: payments
# Current quota: requests.cpu=4, requests.memory=8Gi — usage regularly
# hits 3.8 CPU cores during business hours, causing pending pods during
# peak traffic. Trend: increasing over 30 days (+18%).
 
apiVersion: v1
kind: ResourceQuota
metadata:
  name: payments-quota
  namespace: payments
spec:
  hard:
    requests.cpu: "6"          # 30% headroom over observed peak, accounting for growth trend
    requests.memory: "12Gi"
    limits.cpu: "8"
    limits.memory: "16Gi"
    pods: "40"