K8s Cost

#!/usr/bin/env python3
"""
Resource Optimization Script
Identifies overprovisioned resources for cost savings.

Usage within Claude Code:
    This script is called by the k8s-cost skill to find
    overprovisioned resources and provide recommendations.
"""

import json
import sys
from typing import Any


def find_overprovisioned(namespace: str = "", context: str = "") -> dict[str, Any]:
    """
    Find overprovisioned resources in cluster.

    Args:
        namespace: Optional namespace filter
        context: Optional kubeconfig context

    Returns:
        Dictionary with findings and recommendations
    """
    analysis = {
        "namespace": namespace or "all",
        "context": context or "current",
        "checks": [],
        "thresholds": {
            "cpu_underutilized": 0.1,      # <10% usage
            "memory_underutilized": 0.3,   # <30% usage
            "cpu_overutilized": 0.8,       # >80% usage
            "memory_overutilized": 0.9     # >90% usage
        }
    }

    # Define checks to run with MCP tools
    analysis["checks"] = [
        {
            "name": "resource_recommendations",
            "tool": "get_resource_recommendations",
            "params": {"namespace": namespace, "context": context},
            "description": "Get VPA-style recommendations"
        },
        {
            "name": "pod_metrics",
            "tool": "get_resource_usage",
            "params": {"namespace": namespace, "context": context},
            "description": "Get current resource usage"
        },
        {
            "name": "unused_pvcs",
            "tool": "find_orphaned_pvcs",
            "params": {"namespace": namespace, "context": context},
            "description": "Find PVCs not mounted by any pod"
        },
        {
            "name": "unused_resources",
            "tool": "find_unused_resources",
            "params": {"namespace": namespace, "context": context},
            "description": "Find unused ConfigMaps/Secrets"
        },
        {
            "name": "namespace_cost",
            "tool": "get_namespace_cost",
            "params": {"namespace": namespace, "context": context},
            "description": "Get namespace cost breakdown"
        }
    ]

    return analysis


def analyze_pod_resources(pod: dict, metrics: dict) -> dict[str, Any] | None:
    """
    Analyze a pod's resource usage vs requests.

    Args:
        pod: Pod spec with resources
        metrics: Pod metrics data

    Returns:
        Finding if optimization possible, None otherwise
    """
    finding = {
        "pod": pod.get("name"),
        "namespace": pod.get("namespace"),
        "containers": []
    }

    has_issues = False

    for container in pod.get("containers", []):
        container_name = container.get("name")
        resources = container.get("resources", {})
        requests = resources.get("requests", {})

        # Get actual usage from metrics
        container_metrics = metrics.get("containers", {}).get(container_name, {})

        cpu_request = parse_cpu(requests.get("cpu", "0"))
        cpu_usage = container_metrics.get("cpu_cores", 0)
        memory_request = parse_memory(requests.get("memory", "0"))
        memory_usage = container_metrics.get("memory_bytes", 0)

        container_finding = {
            "name": container_name,
            "issues": [],
            "recommendations": []
        }

        # Check CPU
        if cpu_request > 0 and cpu_usage > 0:
            cpu_ratio = cpu_usage / cpu_request
            if cpu_ratio < 0.1:
                container_finding["issues"].append(
                    f"CPU severely underutilized: {cpu_ratio*100:.1f}% of request"
                )
                container_finding["recommendations"].append(
                    f"Reduce CPU request from {cpu_request} to {cpu_usage * 2:.3f} cores"
                )
                has_issues = True
            elif cpu_ratio > 0.8:
                container_finding["issues"].append(
                    f"CPU highly utilized: {cpu_ratio*100:.1f}% of request"
                )
                container_finding["recommendations"].append(
                    f"Increase CPU request/limit to prevent throttling"
                )
                has_issues = True

        # Check Memory
        if memory_request > 0 and memory_usage > 0:
            memory_ratio = memory_usage / memory_request
            if memory_ratio < 0.3:
                container_finding["issues"].append(
                    f"Memory underutilized: {memory_ratio*100:.1f}% of request"
                )
                container_finding["recommendations"].append(
                    f"Reduce memory request"
                )
                has_issues = True
            elif memory_ratio > 0.9:
                container_finding["issues"].append(
                    f"Memory near limit: {memory_ratio*100:.1f}% of request"
                )
                container_finding["recommendations"].append(
                    f"Increase memory limit to prevent OOMKill"
                )
                has_issues = True

        if container_finding["issues"]:
            finding["containers"].append(container_finding)

    return finding if has_issues else None


def parse_cpu(cpu_str: str) -> float:
    """Parse CPU string to cores."""
    if not cpu_str:
        return 0
    if cpu_str.endswith("m"):
        return float(cpu_str[:-1]) / 1000
    return float(cpu_str)


def parse_memory(memory_str: str) -> int:
    """Parse memory string to bytes."""
    if not memory_str:
        return 0

    units = {
        "Ki": 1024,
        "Mi": 1024**2,
        "Gi": 1024**3,
        "Ti": 1024**4,
        "K": 1000,
        "M": 1000**2,
        "G": 1000**3,
        "T": 1000**4
    }

    for suffix, multiplier in units.items():
        if memory_str.endswith(suffix):
            return int(float(memory_str[:-len(suffix)]) * multiplier)

    return int(memory_str)


def calculate_savings(findings: list[dict[str, Any]]) -> dict[str, Any]:
    """
    Calculate potential cost savings from findings.

    Args:
        findings: List of optimization findings

    Returns:
        Summary of potential savings
    """
    # Rough cost estimates (cloud provider specific)
    cpu_cost_per_core_month = 30  # USD
    memory_cost_per_gb_month = 5  # USD

    savings = {
        "cpu_cores_reducible": 0,
        "memory_gb_reducible": 0,
        "estimated_monthly_savings_usd": 0,
        "actions": []
    }

    for finding in findings:
        for container in finding.get("containers", []):
            for rec in container.get("recommendations", []):
                if "Reduce CPU" in rec:
                    # Extract cores from recommendation
                    savings["actions"].append({
                        "pod": finding["pod"],
                        "container": container["name"],
                        "action": rec
                    })
                elif "Reduce memory" in rec:
                    savings["actions"].append({
                        "pod": finding["pod"],
                        "container": container["name"],
                        "action": rec
                    })

    return savings


def generate_report(analysis: dict[str, Any], findings: list[dict[str, Any]]) -> str:
    """
    Generate cost optimization report.

    Args:
        analysis: Analysis configuration
        findings: List of findings

    Returns:
        Formatted report
    """
    report = ["# Kubernetes Cost Optimization Report\n"]
    report.append(f"Namespace: {analysis['namespace']}")
    report.append(f"Context: {analysis['context']}\n")

    if not findings:
        report.append("No optimization opportunities found.")
        return "\n".join(report)

    report.append(f"## Found {len(findings)} pods with optimization opportunities\n")

    for finding in findings:
        report.append(f"### Pod: {finding['namespace']}/{finding['pod']}")
        for container in finding.get("containers", []):
            report.append(f"  Container: {container['name']}")
            for issue in container.get("issues", []):
                report.append(f"    - Issue: {issue}")
            for rec in container.get("recommendations", []):
                report.append(f"    - Action: {rec}")
        report.append("")

    return "\n".join(report)


if __name__ == "__main__":
    namespace = sys.argv[1] if len(sys.argv) > 1 else ""
    context = sys.argv[2] if len(sys.argv) > 2 else ""

    result = find_overprovisioned(namespace, context)
    print(json.dumps(result, indent=2))