AI-Powered Capacity Planning Across Multi-Cluster Kubernetes: Where This Is Heading in 2026

def gather_multi_cluster_snapshot(clusters: list[str]) -> dict:
    snapshot = {}
    for cluster in clusters:
        snapshot[cluster] = {
            "cpu_utilization_trend_4w": get_trend(cluster, "cpu", weeks=4),
            "memory_utilization_trend_4w": get_trend(cluster, "memory", weeks=4),
            "node_quota_remaining": get_quota_headroom(cluster),
            "pending_pods_last_7d": get_pending_pod_events(cluster, days=7),
            "cost_per_week": get_cluster_cost(cluster),
            "workload_growth_signals": get_deployment_replica_trend(cluster, weeks=4),
        }
    return snapshot

import anthropic
import json
 
client = anthropic.Anthropic()
 
FORECAST_PROMPT = """Analyze capacity trends across these Kubernetes clusters
and identify which ones need action in the next 30 days.
 
Cluster data (utilization trends, quota headroom, pending pods, cost):
{cluster_data}
 
For each cluster approaching capacity constraints, determine:
1. Estimated weeks until it hits a real constraint (quota, cost ceiling, or
   sustained 90%+ utilization)
2. Whether the fix is: request more quota, move workloads to another cluster
   with headroom, or right-size existing workloads first
3. If moving workloads is recommended, which cluster has both headroom AND
   is a sensible target (same region/compliance zone considerations)
 
Respond with ONLY valid JSON:
{{"clusters_needing_action": [
  {{"cluster": "...", "weeks_until_constraint": N, "recommendation": "...",
    "target_cluster_if_migration": "..." or null}}
]}}"""
 
 
def forecast_capacity(cluster_data: dict) -> dict:
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1500,
        messages=[{"role": "user", "content": FORECAST_PROMPT.format(cluster_data=json.dumps(cluster_data, indent=2))}]
    )
    text = response.content[0].text.strip()
    if text.startswith("```"):
        text = text.split("```")[1].replace("json", "", 1).strip()
    return json.loads(text)

def generate_weekly_capacity_report(clusters: list[str]):
    snapshot = gather_multi_cluster_snapshot(clusters)
    forecast = forecast_capacity(snapshot)
 
    if not forecast["clusters_needing_action"]:
        post_slack_summary("Multi-cluster capacity check: all clusters healthy for the next 30 days")
        return
 
    report = "## Capacity Planning — Action Needed\n\n"
    for item in forecast["clusters_needing_action"]:
        report += f"**{item['cluster']}** — ~{item['weeks_until_constraint']} weeks until constraint\n"
        report += f"{item['recommendation']}\n"
        if item["target_cluster_if_migration"]:
            report += f"Suggested target: {item['target_cluster_if_migration']}\n"
        report += "\n"
 
    post_slack_summary(report)