Fix Pod not being removed after istio-sidecar is removed (apache#34500)

Co-authored-by: Joshua Yeung <[email protected]>

airflow/providers/cncf/kubernetes/operators/pod.py

@@ -804,7 +804,6 @@ def kill_istio_sidecar(self, pod: V1Pod) -> None:
             raise Exception("Error while deleting istio-proxy sidecar: %s", output_str)
 
     def process_pod_deletion(self, pod: k8s.V1Pod, *, reraise=True):
-        istio_enabled = self.is_istio_enabled(pod)
         with _optionally_suppress(reraise=reraise):
             if pod is not None:
                 should_delete_pod = (
@@ -818,12 +817,9 @@ def process_pod_deletion(self, pod: k8s.V1Pod, *, reraise=True):
                         and container_is_succeeded(pod, self.base_container_name)
                     )
                 )
-                if should_delete_pod and not istio_enabled:
+                if should_delete_pod:
                     self.log.info("Deleting pod: %s", pod.metadata.name)
                     self.pod_manager.delete_pod(pod)
-                elif should_delete_pod and istio_enabled:
-                    self.log.info("Deleting istio-proxy sidecar inside %s: ", pod.metadata.name)
-                    self.kill_istio_sidecar(pod)
                 else:
                     self.log.info("Skipping deleting pod: %s", pod.metadata.name)
 

tests/providers/cncf/kubernetes/operators/test_pod.py

@@ -624,7 +624,7 @@ def test_termination_message_policy_default_value_correctly_set(self):
             ({"on_finish_action": "delete_succeeded_pod"}, True, False),
         ],
     )
-    @patch(f"{KPO_MODULE}.KubernetesPodOperator.kill_istio_sidecar")
+    @patch(f"{POD_MANAGER_CLASS}.delete_pod")
     @patch(f"{KPO_MODULE}.KubernetesPodOperator.is_istio_enabled")
     @patch(f"{POD_MANAGER_CLASS}.await_pod_completion")
     @patch(f"{KPO_MODULE}.KubernetesPodOperator.find_pod")
@@ -633,7 +633,7 @@ def test_pod_with_istio_delete_after_await_container_error(
         find_pod_mock,
         await_pod_completion_mock,
         is_istio_enabled_mock,
-        kill_istio_sidecar_mock,
+        delete_pod_mock,
         task_kwargs,
         base_container_fail,
         expect_to_delete_pod,
@@ -662,13 +662,13 @@ def test_pod_with_istio_delete_after_await_container_error(
         cont_status_2.state.running = True
         cont_status_2.state.terminated = False
 
-        await_pod_completion_mock.return_value.spec.containers = [sidecar]
+        await_pod_completion_mock.return_value.spec.containers = [sidecar, cont_status_1, cont_status_2]
         await_pod_completion_mock.return_value.status.phase = "Running"
         await_pod_completion_mock.return_value.status.container_statuses = [cont_status_1, cont_status_2]
         await_pod_completion_mock.return_value.metadata.name = "pod-with-istio-sidecar"
         await_pod_completion_mock.return_value.metadata.namespace = "default"
 
-        find_pod_mock.return_value.spec.containers = [sidecar]
+        find_pod_mock.return_value.spec.containers = [sidecar, cont_status_1, cont_status_2]
         find_pod_mock.return_value.status.phase = "Running"
         find_pod_mock.return_value.status.container_statuses = [cont_status_1, cont_status_2]
         find_pod_mock.return_value.metadata.name = "pod-with-istio-sidecar"
@@ -678,22 +678,19 @@ def test_pod_with_istio_delete_after_await_container_error(
 
         context = create_context(k)
         context["ti"].xcom_push = MagicMock()
+
         if base_container_fail:
             self.await_pod_mock.side_effect = AirflowException("fake failure")
             with pytest.raises(AirflowException, match="my-failure"):
                 k.execute(context=context)
         else:
             k.execute(context=context)
 
-        assert is_istio_enabled_mock(find_pod_mock.return_value)
-        if task_kwargs["on_finish_action"] == "delete_pod":
-            kill_istio_sidecar_mock.assert_called_with(await_pod_completion_mock.return_value)
-        elif expect_to_delete_pod and base_container_fail:
-            kill_istio_sidecar_mock.assert_called_with(find_pod_mock.return_value)
-        elif expect_to_delete_pod and not base_container_fail:
-            kill_istio_sidecar_mock.assert_called_with(await_pod_completion_mock.return_value)
+        if expect_to_delete_pod:
+            assert k.is_istio_enabled(find_pod_mock.return_value)
+            delete_pod_mock.assert_called_with(await_pod_completion_mock.return_value)
         else:
-            kill_istio_sidecar_mock.assert_not_called()
+            delete_pod_mock.assert_not_called()
 
     @pytest.mark.parametrize(
         "task_kwargs, should_be_deleted",