diff --git a/changelogs/fragments/20260306-k8s_drain-delete-logic-fixes-for-daemonsets.yaml b/changelogs/fragments/20260306-k8s_drain-delete-logic-fixes-for-daemonsets.yaml
new file mode 100644
index 00000000..cae29ded
--- /dev/null
+++ b/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).
diff --git a/plugins/modules/k8s_drain.py b/plugins/modules/k8s_drain.py
index d4fd1356..84ebfe67 100644
--- a/plugins/modules/k8s_drain.py
+++ b/plugins/modules/k8s_drain.py
@@ -214,21 +214,21 @@ 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:
+        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:
-            for owner in owner_ref:
-                if owner.kind == "DaemonSet":
-                    daemonSet.append((pod.metadata.namespace, pod.metadata.name))
-                else:
-                    to_delete.append((pod.metadata.namespace, pod.metadata.name))
+            to_delete.append((pod.metadata.namespace, pod.metadata.name))
 
     warnings, errors = [], []
     if unmanaged: