#!/usr/bin/env python3 from __future__ import annotations import argparse import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from common import ( DEFAULT_NAMESPACE, human_bytes, human_duration, kubectl_json, now_utc, parse_k8s_timestamp, parse_storage_quantity, print_table, probe_path_usage, ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Show current unrip deployment state, pod uptime, and mounted storage usage." ) parser.add_argument( "--namespace", default=DEFAULT_NAMESPACE, help=f"Kubernetes namespace to inspect (default: {DEFAULT_NAMESPACE})", ) parser.add_argument( "--include-rootfs", action="store_true", help="Also probe '/' for pods that do not have PVC-backed mounts.", ) parser.add_argument( "--include-completed", action="store_true", help="Include completed Job pods in the pod list.", ) parser.add_argument( "--output", choices=("json", "table"), default="json", help="Output format (default: json).", ) return parser.parse_args() def ready_string(pod: dict) -> str: statuses = pod.get("status", {}).get("containerStatuses", []) ready = sum(1 for status in statuses if status.get("ready")) total = len(statuses) return f"{ready}/{total}" def restart_count(pod: dict) -> int: return sum(status.get("restartCount", 0) for status in pod.get("status", {}).get("containerStatuses", [])) def pvc_mounts_for_pod(pod: dict) -> list[dict[str, str]]: volumes = { volume["name"]: volume.get("persistentVolumeClaim", {}).get("claimName") for volume in pod.get("spec", {}).get("volumes", []) } mounts: list[dict[str, str]] = [] for container in pod.get("spec", {}).get("containers", []): for mount in container.get("volumeMounts", []): claim_name = volumes.get(mount["name"]) if claim_name: mounts.append( { "container": container.get("name", "app"), "claim_name": claim_name, "mount_path": mount["mountPath"], } ) return mounts def is_completed_job_pod(pod: dict) -> bool: owners = pod.get("metadata", {}).get("ownerReferences", []) if not any(owner.get("kind") == "Job" for owner in owners): return False return pod.get("status", {}).get("phase") in {"Succeeded", "Failed"} def main() -> int: args = parse_args() namespace = args.namespace deployment_payload = kubectl_json("get", "deploy", "-n", namespace, "-o", "json") pod_payload = kubectl_json("get", "pods", "-n", namespace, "-o", "json") pvc_payload = kubectl_json("get", "pvc", "-n", namespace, "-o", "json") pvc_requests = { item["metadata"]["name"]: item.get("spec", {}) .get("resources", {}) .get("requests", {}) .get("storage", "") for item in pvc_payload.get("items", []) } deployments: list[dict[str, object]] = [] for item in sorted(deployment_payload.get("items", []), key=lambda value: value["metadata"]["name"]): container = item.get("spec", {}).get("template", {}).get("spec", {}).get("containers", [{}])[0] deployments.append( { "name": item["metadata"]["name"], "ready": { "ready_replicas": item.get("status", {}).get("readyReplicas", 0), "desired_replicas": item.get("spec", {}).get("replicas", 0), "display": f"{item.get('status', {}).get('readyReplicas', 0)}/{item.get('spec', {}).get('replicas', 0)}", }, "available_replicas": item.get("status", {}).get("availableReplicas", 0), "image": container.get("image", "-"), } ) pods: list[dict[str, object]] = [] storage: list[dict[str, object]] = [] for pod in sorted(pod_payload.get("items", []), key=lambda value: value["metadata"]["name"]): if not args.include_completed and is_completed_job_pod(pod): continue pod_name = pod["metadata"]["name"] start_time = parse_k8s_timestamp( pod.get("status", {}).get("startTime") or pod.get("status", {}).get("containerStatuses", [{}])[0] .get("state", {}) .get("running", {}) .get("startedAt") ) uptime = human_duration((now_utc() - start_time).total_seconds() if start_time else None) image = pod.get("spec", {}).get("containers", [{}])[0].get("image", "-") pods.append( { "name": pod_name, "phase": pod.get("status", {}).get("phase", "-"), "ready": ready_string(pod), "restarts": restart_count(pod), "uptime": uptime, "node": pod.get("spec", {}).get("nodeName", "-"), "image": image, } ) mounts = pvc_mounts_for_pod(pod) if not mounts and args.include_rootfs: mounts = [{"claim_name": "-", "mount_path": "/", "container": "app"}] if not mounts: storage.append( { "pod": pod_name, "path": None, "claim": None, "requested_bytes": None, "requested": None, "path_bytes": None, "fs_used_bytes": None, "fs_available_bytes": None, "use_percent": None, "note": "no pvc mounts", } ) continue for mount in mounts: usage = probe_path_usage(pod_name, mount["mount_path"], namespace=namespace) requested_raw = pvc_requests.get(mount["claim_name"], "") requested_bytes = parse_storage_quantity(requested_raw) storage.append( { "pod": pod_name, "path": mount["mount_path"], "claim": mount["claim_name"], "requested_bytes": requested_bytes, "requested": requested_raw or None, "path_bytes": usage["path_bytes"], "fs_used_bytes": usage["filesystem_used_bytes"], "fs_available_bytes": usage["filesystem_available_bytes"], "use_percent": usage["filesystem_use_percent"], "note": None, } ) if args.output == "json": print( json.dumps( { "namespace": namespace, "deployments": deployments, "pods": pods, "storage": storage, "notes": [ "storage figures come from container-mounted paths", "pods without pvc-backed mounts do not report dedicated storage", ], }, indent=2, ) ) return 0 deployment_rows = [ [item["name"], item["ready"]["display"], str(item["available_replicas"]), item["image"]] for item in deployments ] pod_rows = [ [item["name"], item["phase"], item["ready"], str(item["restarts"]), item["uptime"], item["node"], item["image"]] for item in pods ] storage_rows = [ [ item["pod"], item["path"] or "-", item["claim"] or "no pvc mounts", human_bytes(item["requested_bytes"]) if item["requested_bytes"] is not None else (item["requested"] or "-"), human_bytes(item["path_bytes"]), human_bytes(item["fs_used_bytes"]), human_bytes(item["fs_available_bytes"]), str(item["use_percent"] or "-"), ] for item in storage ] print(f"Namespace: {namespace}") print() print("Deployments") print_table(["NAME", "READY", "AVAILABLE", "IMAGE"], deployment_rows) print() print("Pods") print_table(["POD", "PHASE", "READY", "RESTARTS", "UPTIME", "NODE", "IMAGE"], pod_rows) print() print("Storage") print_table( ["POD", "PATH", "CLAIM", "REQUESTED", "PATH_BYTES", "FS_USED", "FS_AVAIL", "USE%"], storage_rows, ) print() print("Note: storage figures come from container-mounted paths. Pods without PVC-backed mounts report no dedicated storage.") return 0 if __name__ == "__main__": raise SystemExit(main())