Kubernetes Cost Calculator

#!/usr/bin/env python3
"""Application pod requests beside live use, per pod. Read-only.

Needs kubectl (read access to pods), Python 3.8+, and metrics-server for the
use columns. Offline: --pods pods.json [--metrics metrics.json], from
  kubectl get pods --all-namespaces -o json
  kubectl get --raw /apis/metrics.k8s.io/v1beta1/pods

Pods are matched on namespace AND name. Requests are the sum over regular
containers plus restartable (sidecar) init containers. Not reproduced, only
flagged: other init containers (the effective request can be higher while
they run), pod overhead, and pod-level resources. DaemonSet and static pods
are totalled apart: the calculator's per-node allowance covers them.

Live use is ONE snapshot. It shows outliers; it does not prove a smaller
request is safe. Take typical and peak use from your metrics history over a
representative period (two weeks or more, busiest days included).
"""
import argparse, json, subprocess, sys
from decimal import Decimal

UNITS = {"Ki": 2**10, "Mi": 2**20, "Gi": 2**30, "Ti": 2**40, "Pi": 2**50, "Ei": 2**60,
         "n": Decimal("1e-9"), "u": Decimal("1e-6"), "m": Decimal("1e-3"),
         "k": 10**3, "M": 10**6, "G": 10**9, "T": 10**12, "P": 10**15, "E": 10**18}
GIB = Decimal(2**30)

def quantity(q):
    """A Kubernetes quantity ("250m", "1.5", "512Mi", "1G", "1e3") in base units."""
    q = str(q).strip()
    for suffix in sorted(UNITS, key=len, reverse=True):
        if q.endswith(suffix) and q[:-len(suffix)]:
            return Decimal(q[:-len(suffix)]) * Decimal(UNITS[suffix])
    return Decimal(q)

def load(path, cmd):
    if path:
        with open(path) as f:
            return json.load(f)
    return json.loads(subprocess.run(cmd, check=True, capture_output=True, text=True).stdout)

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--pods")
    ap.add_argument("--metrics")
    args = ap.parse_args()
    pods = load(args.pods, ["kubectl", "get", "pods", "--all-namespaces", "-o", "json"])["items"]
    try:
        metrics = load(args.metrics, ["kubectl", "get", "--raw", "/apis/metrics.k8s.io/v1beta1/pods"]) \
            if (args.metrics or not args.pods) else {"items": []}
    except Exception as e:
        print(f"# metrics unavailable ({e}); use columns show n/a", file=sys.stderr)
        metrics = {"items": []}
    use = {}
    for m in metrics.get("items", []):
        key = (m["metadata"]["namespace"], m["metadata"]["name"])
        cs = m.get("containers", [])
        use[key] = (sum(quantity(c["usage"].get("cpu", "0")) for c in cs),
                    sum(quantity(c["usage"].get("memory", "0")) for c in cs))

    totals = {"app": [0, Decimal(0), Decimal(0)], "pending": [0, Decimal(0), Decimal(0)],
              "system": [0, Decimal(0), Decimal(0)]}
    no_metrics = 0
    print("NAMESPACE\tPOD\tPHASE\tCPU_REQ\tMEM_REQ_GIB\tCPU_USED\tMEM_USED_GIB\tFLAGS")
    for p in pods:
        meta, spec = p["metadata"], p.get("spec", {})
        phase = p.get("status", {}).get("phase", "Unknown")
        if phase in ("Succeeded", "Failed"):
            continue  # finished pods hold no capacity
        flags = []
        running = list(spec.get("containers", []))
        for c in spec.get("initContainers", []):
            if c.get("restartPolicy") == "Always":
                running.append(c); flags.append("sidecar")
            elif "init" not in flags:
                flags.append("init")
        cpu = mem = Decimal(0)
        for c in running:
            req = c.get("resources", {}).get("requests", {})
            if "cpu" in req: cpu += quantity(req["cpu"])
            elif "no-cpu-request" not in flags: flags.append("no-cpu-request")
            if "memory" in req: mem += quantity(req["memory"])
            elif "no-memory-request" not in flags: flags.append("no-memory-request")
        if spec.get("overhead"): flags.append("overhead")
        if spec.get("resources"): flags.append("pod-level-resources")
        owners = [o.get("kind") for o in meta.get("ownerReferences", [])]
        system = "DaemonSet" in owners or "kubernetes.io/config.mirror" in meta.get("annotations", {})
        if system: flags.append("daemonset-or-static")
        key = (meta["namespace"], meta["name"])
        if key in use:
            cpu_used, mem_used = f"{use[key][0]:.3f}", f"{use[key][1] / GIB:.3f}"
        else:
            cpu_used = mem_used = "n/a"
            if phase == "Running": no_metrics += 1
        bucket = "system" if system else ("pending" if phase == "Pending" else "app")
        totals[bucket][0] += 1; totals[bucket][1] += cpu; totals[bucket][2] += mem
        print(f"{key[0]}\t{key[1]}\t{phase}\t{cpu:.3f}\t{mem / GIB:.3f}\t{cpu_used}\t{mem_used}\t{','.join(flags) or '-'}")

    print()
    for label, k in (("Application pods, enter these on the page", "app"),
                     ("Pending pods, not yet on a node", "pending"),
                     ("DaemonSet and static pods, left out of the page's inputs", "system")):
        n, cpu, mem = totals[k]
        print(f"# {label}: {n} pods, {cpu:.3f} CPU, {mem / GIB:.3f} GiB requested")
    print(f"# Running pods with no metrics (use unknown, not zero): {no_metrics}")
    print("# Live use is one snapshot: not evidence that a smaller request is safe.")

if __name__ == "__main__":
    main()