aa8a868c14
All checks were successful
deploy / deploy (push) Successful in 41s
Proof: Successful trades only dashboard is empty Assumptions: The downtime resulted in a massive gap we can attribute Still fake: Some outcomes may be heuristic
250 lines
8.6 KiB
Python
Executable file
250 lines
8.6 KiB
Python
Executable file
#!/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())
|