Add dupvc

k8sutils/dupvc.py

+import kr8s
+from typing import Annotated
+import sys
+from io import BytesIO
+import typer
+
+app = typer.Typer()
+
+@app.command()
+def main(
+  namespace: str = typer.Option(default=kr8s.ALL),
+  human_readable: Annotated[bool, typer.Option("--human-readable", "-h", help="Human-readable sizes")] = False,
+):
+    """Run `du -s` in all PVCs."""
+    pods = kr8s.get("pod", namespace=namespace)
+
+    h_arg = ["-h"] if human_readable else []
+
+    for pod in pods:
+        pvc_volumes = {}
+    
+        for volume in pod.spec.volumes:
+            if "persistentVolumeClaim" in volume:
+                pvc_volumes[volume.name] = volume.persistentVolumeClaim.claimName
+    
+        for container in pod.spec.containers:
+            for mount in container.volumeMounts:
+                if mount.name in pvc_volumes:
+                    stdout = BytesIO()
+    
+                    pod.exec(["du", "-s"] + h_arg + [mount.mountPath], stdout=stdout, stderr=stdout, check=False)
+                    stdout.seek(0)
+                    print(pod.namespace, pvc_volumes[mount.name], stdout.read().decode(), end="")

pyproject.toml

@@ -19,3 +19,4 @@ build-backend = "poetry.core.masonry.api"
 [tool.poetry.scripts]
 s3auth = 'k8sutils.s3auth:app'
 softdrain = 'k8sutils.softdrain:app'
+dupvc = 'k8sutils.dupvc:app'