diff --git a/pvc_inspector.py b/pvc_inspector.py index dc1bd6e..a16517d 100755 --- a/pvc_inspector.py +++ b/pvc_inspector.py @@ -15,6 +15,7 @@ import time import typer from typing import Any, Optional import yaml +import os logging.basicConfig(level=logging.WARNING) @@ -458,6 +459,157 @@ sleep 3600 return inspector_job +PROTECTED_NAMESPACES = { + "kube-system", + "kube-public", + "kube-node-lease", + "argo-events", + "argo-rollouts", + "argo-workflows", + "argocd", + "cert-manager", + "ingress-nginx", + "monitoring", + "prometheus", + "istio-system", + "linkerd", +} + +@app.command() +def exec_inspector( + namespace: Optional[str] = typer.Option( + None, + help="Kubernetes namespace. If not specified, will search for inspectors across all namespaces.", + ), +): + """ + Execute a shell in a running inspector pod. If multiple inspectors are found, + presents a fuzzy finder to select one. + """ + config.load_kube_config() + batch_api = client.BatchV1Api() + + try: + if namespace: + logging.debug(f"Listing jobs in namespace {namespace}") + jobs = batch_api.list_namespaced_job( + namespace=namespace, label_selector="app=pvc-inspector" + ) + else: + logging.debug("Listing jobs in all namespaces") + jobs = batch_api.list_job_for_all_namespaces( + label_selector="app=pvc-inspector" + ) + + running_inspectors = [] + for job in jobs.items: + # Get the pod for this job + v1 = client.CoreV1Api() + pods = v1.list_namespaced_pod( + namespace=job.metadata.namespace, + label_selector=f"job-name={job.metadata.name}" + ) + for pod in pods.items: + if pod.status.phase == "Running": + running_inspectors.append((pod.metadata.name, pod.metadata.namespace)) + + if not running_inspectors: + typer.echo("No running inspector pods found.") + raise typer.Exit(1) + + if len(running_inspectors) == 1: + pod_name, pod_namespace = running_inspectors[0] + else: + pod_name, pod_namespace = fuzzy_select(running_inspectors) + if not pod_name: + typer.echo("No inspector selected.") + raise typer.Exit(1) + + # Execute the shell + typer.echo(f"Connecting to inspector {pod_namespace}/{pod_name}...") + os.execvp("kubectl", ["kubectl", "exec", "-it", "-n", pod_namespace, pod_name, "--", "sh", "-l"]) + + except client.exceptions.ApiException as e: + logging.error(f"Failed to list jobs: {e}") + typer.echo(f"Failed to list jobs: {e}", err=True) + raise typer.Exit(1) + + +@app.command() +def cleanup_inspectors( + namespace: Optional[str] = typer.Option( + None, + help="Kubernetes namespace. If not specified, will cleanup in all namespaces.", + ), + yes: bool = typer.Option( + False, + "--yes", "-y", + help="Skip confirmation prompt.", + ), +): + """ + Delete all PVC inspector jobs in the specified namespace or all namespaces + """ + config.load_kube_config() + batch_api = client.BatchV1Api() + + try: + if namespace: + if namespace in PROTECTED_NAMESPACES: + typer.echo(f"Error: Cannot cleanup in protected namespace {namespace}") + raise typer.Exit(1) + logging.debug(f"Listing jobs in namespace {namespace}") + jobs = batch_api.list_namespaced_job( + namespace=namespace, label_selector="app=pvc-inspector" + ) + else: + logging.debug("Listing jobs in all namespaces") + jobs = batch_api.list_job_for_all_namespaces( + label_selector="app=pvc-inspector" + ) + + # Filter out jobs in protected namespaces + jobs.items = [job for job in jobs.items if job.metadata.namespace not in PROTECTED_NAMESPACES] + + if not jobs.items: + typer.echo("No PVC inspector jobs found.") + return + + # Show confirmation prompt + if not yes: + job_list = "\n".join(f" {job.metadata.namespace}/{job.metadata.name}" for job in jobs.items) + typer.echo(f"The following inspector jobs will be deleted:\n{job_list}") + if not typer.confirm("Are you sure you want to continue?"): + typer.echo("Operation cancelled.") + return + + # Delete each job + for job in jobs.items: + try: + logging.debug( + f"Deleting job {job.metadata.namespace}/{job.metadata.name}" + ) + batch_api.delete_namespaced_job( + name=job.metadata.name, + namespace=job.metadata.namespace, + body=client.V1DeleteOptions(propagation_policy="Background"), + ) + typer.echo(f"Deleted job: {job.metadata.namespace}/{job.metadata.name}") + except client.exceptions.ApiException as e: + logging.error( + f"Failed to delete job {job.metadata.namespace}/{job.metadata.name}: {e}" + ) + typer.echo( + f"Failed to delete job {job.metadata.namespace}/{job.metadata.name}: {e}", + err=True, + ) + + except client.exceptions.ApiException as e: + logging.error(f"Failed to list jobs: {e}") + typer.echo(f"Failed to list jobs: {e}", err=True) + raise typer.Exit(1) + + @app.command() def create_inspector( namespace: Optional[str] = typer.Option( @@ -490,63 +642,5 @@ def create_inspector( typer.echo(yaml.dump(clean_dict(inspector_job), sort_keys=False)) -@app.command() -def cleanup_inspectors( - namespace: Optional[str] = typer.Option( - None, - help="Kubernetes namespace. If not specified, will cleanup in all namespaces.", - ), -): - """ - Delete all PVC inspector jobs in the specified namespace or all namespaces - """ - config.load_kube_config() - batch_api = client.BatchV1Api() - - try: - if namespace: - logging.debug(f"Listing jobs in namespace {namespace}") - jobs = batch_api.list_namespaced_job( - namespace=namespace, label_selector="app=pvc-inspector" - ) - else: - logging.debug("Listing jobs in all namespaces") - jobs = batch_api.list_job_for_all_namespaces( - label_selector="app=pvc-inspector" - ) - - if not jobs.items: - typer.echo("No PVC inspector jobs found.") - return - - # Delete each job - for job in jobs.items: - try: - logging.debug( - f"Deleting job {job.metadata.namespace}/{job.metadata.name}" - ) - batch_api.delete_namespaced_job( - name=job.metadata.name, - namespace=job.metadata.namespace, - body=client.V1DeleteOptions(propagation_policy="Background"), - ) - typer.echo(f"Deleted job: {job.metadata.namespace}/{job.metadata.name}") - except client.exceptions.ApiException as e: - logging.error( - f"Failed to delete job {job.metadata.namespace}/{job.metadata.name}: {e}" - ) - typer.echo( - f"Failed to delete job {job.metadata.namespace}/{job.metadata.name}: {e}", - err=True, - ) - - except client.exceptions.ApiException as e: - logging.error(f"Failed to list jobs: {e}") - typer.echo(f"Failed to list jobs: {e}", err=True) - raise typer.Exit(1) - - typer.echo("Cleanup complete.") - - if __name__ == "__main__": app()