54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from ..k8s.client import delete_json, get_json
|
|
from ..utils.logging import get_logger
|
|
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PodCleanerSummary:
|
|
deleted: int
|
|
skipped: int
|
|
failures: int
|
|
|
|
|
|
def _collect_pods(phase: str) -> list[dict[str, Any]]:
|
|
payload = get_json(f"/api/v1/pods?fieldSelector=status.phase={phase}")
|
|
items = payload.get("items") if isinstance(payload.get("items"), list) else []
|
|
return [item for item in items if isinstance(item, dict)]
|
|
|
|
|
|
def _delete_pod(namespace: str, name: str) -> None:
|
|
delete_json(f"/api/v1/namespaces/{namespace}/pods/{name}?gracePeriodSeconds=0")
|
|
|
|
|
|
def clean_finished_pods() -> PodCleanerSummary:
|
|
deleted = 0
|
|
skipped = 0
|
|
failures = 0
|
|
|
|
for phase in ("Succeeded", "Failed"):
|
|
for pod in _collect_pods(phase):
|
|
metadata = pod.get("metadata") if isinstance(pod.get("metadata"), dict) else {}
|
|
namespace = metadata.get("namespace") if isinstance(metadata.get("namespace"), str) else ""
|
|
name = metadata.get("name") if isinstance(metadata.get("name"), str) else ""
|
|
if not namespace or not name:
|
|
skipped += 1
|
|
continue
|
|
try:
|
|
_delete_pod(namespace, name)
|
|
deleted += 1
|
|
except Exception as exc:
|
|
failures += 1
|
|
logger.info(
|
|
"pod delete failed",
|
|
extra={"event": "pod_cleaner", "status": "error", "detail": str(exc)},
|
|
)
|
|
|
|
return PodCleanerSummary(deleted=deleted, skipped=skipped, failures=failures)
|