Memory usage statistics (min, max, mean, 95th percentile) for Kubernetes pods grouped by container name

kubectl_top_pods_container_stats.py

import argparse
import subprocess
import statistics

from collections import defaultdict
from typing import Union


def run_kubectl_top(
    all_namespaces: bool = False,
    label: str | None = None,
    sort_by: Union["cpu", "memory"] = "memory",
) -> str:
    cmd = [
        "kubectl",
        "top",
        "pods",
        f"--sort-by={sort_by}",
        "--no-headers=true",
        "--containers=true",
    ]

    if label:
        cmd.append(f"-l={label}")

    if all_namespaces:
        cmd.append("--all-namespaces")

    result = subprocess.run(cmd, capture_output=True, text=True, check=True)
    return result.stdout


def parse_kubectl_output(all_namespaces: bool, output: str) -> defaultdict:
    containers = defaultdict(list)

    lines = output.strip().split("\n")

    for line in lines:
        parts = line.split()
        if len(parts) < 4:
            continue  # Ignore malformed lines

        if all_namespaces:
            container_name = f"{parts[0]}/{parts[2]}"  # First column is the namespace and third column is the container name
            memory_usage = int(
                parts[4].rstrip("Mi")
            )  # Fourth column is memory usage in MiB
        else:
            container_name = parts[1]  # Second column is the container name
            memory_usage = int(
                parts[3].rstrip("Mi")
            )  # Fourth column is memory usage in MiB

        containers[container_name].append(memory_usage)

    return containers


def compute_statistics(containers: dict) -> dict:
    stats = {}
    for container, values in containers.items():
        values.sort()
        min_val = values[0]
        max_val = values[-1]
        mean_val = round(statistics.mean(values), 2)
        p95_val = (
            round(statistics.quantiles(values, n=100)[94], 2)
            if len(values) >= 20
            else max_val
        )

        stats[container] = {
            "min": min_val,
            "max": max_val,
            "mean": mean_val,
            "p95": p95_val,
            "count": len(values),
            "sum": sum(values),
        }

    return stats


def display_statistics(stats: dict):
    total_memory_usage = 0
    total_containers = 0

    print(
        f"{'Container':<50}{'Min':>10}{'Max':>10}{'Mean':>10}{'P95':>10}{'COUNT':>10}{'SUM':>10}"
    )
    print("-" * 60)
    for container, data in stats.items():
        total_memory_usage += data["sum"]
        total_containers += data["count"]
        print(
            f"{container:<50}{data['min']:>10}{data['max']:>10}{data['mean']:>10}{data['p95']:>10}{data['count']:>10}{data['sum']:>10}"
        )

    print("-" * 60)
    print(
        f"{'Total':<50}{'':>10}{'':>10}{'':>10}{'':>10}{total_containers:>10}{total_memory_usage:>10}"
    )


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Display memory usage statistics for pods"
    )
    parser.add_argument(
        "--all-namespaces",
        action="store_true",
        help="Include pods from all namespaces",
    )
    parser.add_argument("--label", help="Label to filter pods by")
    parser.add_argument(
        "--sort-by",
        choices=["cpu", "memory"],
        default="memory",
        help="Sort by either CPU or memory usage",
    )
    args = parser.parse_args()

    output = run_kubectl_top(
        all_namespaces=args.all_namespaces, label=args.label, sort_by=args.sort_by
    )
    containers = parse_kubectl_output(args.all_namespaces, output)
    stats = compute_statistics(containers)
    display_statistics(stats)