ariadne: split workspace cleanup flow for pedantic lint

ariadne/services/jenkins_workspace_cleanup.py

@@ -141,22 +141,12 @@ def _workspace_pvc_candidates(active_claims: set[str]) -> list[dict[str, Any]]:
     return candidates
 
 
-def cleanup_jenkins_workspace_storage() -> JenkinsWorkspaceCleanupSummary:
+def _delete_stale_pvcs(stale_pvcs: list[dict[str, Any]], namespace: str) -> tuple[int, int, int]:
-    """Delete stale Jenkins workspace PVC/PV artifacts and orphan Longhorn volumes."""
+    """Delete stale workspace PVCs and return deleted/skipped/failure counters."""
 
-    namespace = settings.jenkins_workspace_namespace
+    deleted = 0
-    prefix = settings.jenkins_workspace_pvc_prefix
-    pvs_deleted = 0
-    pvcs_deleted = 0
-    volumes_deleted = 0
     skipped = 0
     failures = 0
-
-    active_claims = _active_workspace_claims()
-    stale_pvs, all_pv_names = _workspace_pv_candidates(active_claims)
-    stale_pvcs = _workspace_pvc_candidates(active_claims)
-    removed_pv_names: set[str] = set()
-
     for pvc in stale_pvcs:
         metadata = pvc.get("metadata") if isinstance(pvc.get("metadata"), dict) else {}
         claim_name = metadata.get("name")
@@ -165,14 +155,23 @@ def cleanup_jenkins_workspace_storage() -> JenkinsWorkspaceCleanupSummary:
             continue
         try:
             delete_json(f"/api/v1/namespaces/{namespace}/persistentvolumeclaims/{claim_name}")
-            pvcs_deleted += 1
+            deleted += 1
         except Exception as exc:
             failures += 1
             logger.info(
                 "jenkins workspace pvc delete failed",
                 extra={"event": "jenkins_workspace_cleanup", "claim": claim_name, "detail": str(exc)},
             )
+    return deleted, skipped, failures
 
+
+def _delete_stale_pvs(stale_pvs: list[dict[str, Any]]) -> tuple[int, int, int, set[str]]:
+    """Delete stale workspace PVs and return counters plus removed PV names."""
+
+    deleted = 0
+    skipped = 0
+    failures = 0
+    removed_pv_names: set[str] = set()
     for pv in stale_pvs:
         metadata = pv.get("metadata") if isinstance(pv.get("metadata"), dict) else {}
         pv_name = metadata.get("name")
@@ -182,14 +181,44 @@ def cleanup_jenkins_workspace_storage() -> JenkinsWorkspaceCleanupSummary:
         try:
             delete_json(f"/api/v1/persistentvolumes/{pv_name}")
             removed_pv_names.add(pv_name)
-            pvs_deleted += 1
+            deleted += 1
         except Exception as exc:
             failures += 1
             logger.info(
                 "jenkins workspace pv delete failed",
                 extra={"event": "jenkins_workspace_cleanup", "pv": pv_name, "detail": str(exc)},
             )
+    return deleted, skipped, failures, removed_pv_names
 
+
+def _volume_should_delete(
+    metadata: dict[str, Any],
+    removed_pv_names: set[str],
+    all_pv_names: set[str],
+    prefix: str,
+) -> bool:
+    """Return true when a Longhorn volume belongs to stale Jenkins workspace storage."""
+
+    name = metadata.get("name")
+    if not isinstance(name, str) or not name:
+        return False
+    if name in removed_pv_names:
+        return True
+    labels = metadata.get("labels") if isinstance(metadata.get("labels"), dict) else {}
+    pvc_name = labels.get("kubernetes.io/created-for/pvc/name")
+    return isinstance(pvc_name, str) and pvc_name.startswith(prefix) and name not in all_pv_names
+
+
+def _delete_stale_longhorn_volumes(
+    removed_pv_names: set[str],
+    all_pv_names: set[str],
+    prefix: str,
+) -> tuple[int, int, int]:
+    """Delete stale Longhorn volumes tied to Jenkins workspace PVCs."""
+
+    deleted = 0
+    skipped = 0
+    failures = 0
     payload = get_json("/apis/longhorn.io/v1beta2/namespaces/longhorn-system/volumes")
     items = payload.get("items") if isinstance(payload.get("items"), list) else []
     for volume in items:
@@ -200,28 +229,37 @@ def cleanup_jenkins_workspace_storage() -> JenkinsWorkspaceCleanupSummary:
         if not isinstance(name, str) or not name:
             skipped += 1
             continue
-        should_delete = name in removed_pv_names
+        if not _volume_should_delete(metadata, removed_pv_names, all_pv_names, prefix):
-        if not should_delete:
-            labels = metadata.get("labels") if isinstance(metadata.get("labels"), dict) else {}
-            pvc_name = labels.get("kubernetes.io/created-for/pvc/name")
-            should_delete = (
-                isinstance(pvc_name, str)
-                and pvc_name.startswith(prefix)
-                and name not in all_pv_names
-            )
-        if not should_delete:
             continue
         if not _is_old_enough(metadata):
             continue
         try:
             delete_json(f"/apis/longhorn.io/v1beta2/namespaces/longhorn-system/volumes/{name}")
-            volumes_deleted += 1
+            deleted += 1
         except Exception as exc:
             failures += 1
             logger.info(
                 "jenkins workspace longhorn volume delete failed",
                 extra={"event": "jenkins_workspace_cleanup", "volume": name, "detail": str(exc)},
             )
+    return deleted, skipped, failures
+
+
+def cleanup_jenkins_workspace_storage() -> JenkinsWorkspaceCleanupSummary:
+    """Delete stale Jenkins workspace PVC/PV artifacts and orphan Longhorn volumes."""
+
+    namespace = settings.jenkins_workspace_namespace
+    prefix = settings.jenkins_workspace_pvc_prefix
+    active_claims = _active_workspace_claims()
+    stale_pvs, all_pv_names = _workspace_pv_candidates(active_claims)
+    stale_pvcs = _workspace_pvc_candidates(active_claims)
+    pvcs_deleted, pvc_skipped, pvc_failures = _delete_stale_pvcs(stale_pvcs, namespace)
+    pvs_deleted, pv_skipped, pv_failures, removed_pv_names = _delete_stale_pvs(stale_pvs)
+    volumes_deleted, volume_skipped, volume_failures = _delete_stale_longhorn_volumes(
+        removed_pv_names, all_pv_names, prefix
+    )
+    skipped = pvc_skipped + pv_skipped + volume_skipped
+    failures = pvc_failures + pv_failures + volume_failures
 
     return JenkinsWorkspaceCleanupSummary(
         pvs_deleted=pvs_deleted,