Update pod deletion conditions for local storage used in a daemon set pod (#1095) (#1127)

* Update pod deletion conditions for local storage used in a daemon set pod

Refactor pod deletion logic to handle local storage in daemon set pods correctly.

* Add bugfix for k8s_drain handling local storage pods

Fix logic for handling pods with local storage in k8s_drain.

* Refactor filter_pods function for clarity and efficiency

* Fix sanity and linter errors

---------


(cherry picked from commit b58b2ca70e)

Co-authored-by: Juan Morete <morete@gmail.com>
Co-authored-by: Bianca Henderson <bianca@redhat.com>

changelogs/fragments/20260306-k8s_drain-delete-logic-fixes-for-daemonsets.yaml

@@ -0,0 +1,3 @@
+---
+bugfixes:
+  - k8s_drain - Fix logic for handling pods with local storage to correctly check for empty_dir volumes in replicated pods and pods managed by DaemonSets (https://github.com/ansible-collections/kubernetes.core/pull/1095).

plugins/modules/k8s_drain.py

@@ -214,19 +214,19 @@ def filter_pods(pods, force, ignore_daemonset, delete_emptydir_data):
             to_delete.append((pod.metadata.namespace, pod.metadata.name))
             continue
 
-        # Pod with local storage cannot be deleted
-        if pod.spec.volumes and any(vol.empty_dir for vol in pod.spec.volumes):
-            localStorage.append((pod.metadata.namespace, pod.metadata.name))
-            continue
-
         # Check replicated Pod
-        owner_ref = pod.metadata.owner_references
-        if not owner_ref:
-            unmanaged.append((pod.metadata.namespace, pod.metadata.name))
-        else:
-            for owner in owner_ref:
-                if owner.kind == "DaemonSet":
+        owner_ref = pod.metadata.owner_references or []
+        has_local_storage = bool(
+            pod.spec.volumes and any(vol.empty_dir for vol in pod.spec.volumes)
+        )
+        is_daemonset_managed = any(owner.kind == "DaemonSet" for owner in owner_ref)
+
+        if is_daemonset_managed:
             daemonSet.append((pod.metadata.namespace, pod.metadata.name))
+        elif has_local_storage:
+            localStorage.append((pod.metadata.namespace, pod.metadata.name))
+        elif not owner_ref:
+            unmanaged.append((pod.metadata.namespace, pod.metadata.name))
         else:
             to_delete.append((pod.metadata.namespace, pod.metadata.name))