[airflow] branch main updated: Add k8s container's error message in airflow exception (#22871)

[po...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new d81703c577 Add k8s container's error message in airflow exception (#22871)
d81703c577 is described below

commit d81703c5778e13470fcd267578697158776b8318
Author: karunpoudel <62...@users.noreply.github.com>
AuthorDate: Wed Apr 13 19:16:03 2022 -0500

    Add k8s container's error message in airflow exception (#22871)
---
 .../providers/cncf/kubernetes/operators/kubernetes_pod.py   | 13 +++++++++++--
 airflow/providers/cncf/kubernetes/utils/pod_manager.py      |  9 +++++++++
 .../cncf/kubernetes/operators/test_kubernetes_pod.py        |  2 +-
 3 files changed, 21 insertions(+), 3 deletions(-)

diff --git a/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py b/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py
index 587ecd4fdf..056d7fe3f3 100644
--- a/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py
+++ b/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py
@@ -43,7 +43,12 @@ from airflow.providers.cncf.kubernetes.backcompat.backwards_compat_converters im
     convert_volume_mount,
 )
 from airflow.providers.cncf.kubernetes.utils import xcom_sidecar  # type: ignore[attr-defined]
-from airflow.providers.cncf.kubernetes.utils.pod_manager import PodLaunchFailedException, PodManager, PodPhase
+from airflow.providers.cncf.kubernetes.utils.pod_manager import (
+    PodLaunchFailedException,
+    PodManager,
+    PodPhase,
+    get_container_termination_message,
+)
 from airflow.settings import pod_mutation_hook
 from airflow.utils import yaml
 from airflow.utils.helpers import prune_dict, validate_key
@@ -409,7 +414,11 @@ class KubernetesPodOperator(BaseOperator):
                     self.patch_already_checked(pod)
             with _suppress(Exception):
                 self.process_pod_deletion(pod)
-            raise AirflowException(f'Pod {pod and pod.metadata.name} returned a failure: {remote_pod}')
+            error_message = get_container_termination_message(remote_pod, self.BASE_CONTAINER_NAME)
+            error_message = "\n" + error_message if error_message else ""
+            raise AirflowException(
+                f'Pod {pod and pod.metadata.name} returned a failure:{error_message}\n{remote_pod}'
+            )
         else:
             with _suppress(Exception):
                 self.process_pod_deletion(pod)
diff --git a/airflow/providers/cncf/kubernetes/utils/pod_manager.py b/airflow/providers/cncf/kubernetes/utils/pod_manager.py
index f84a044d68..993ba12e31 100644
--- a/airflow/providers/cncf/kubernetes/utils/pod_manager.py
+++ b/airflow/providers/cncf/kubernetes/utils/pod_manager.py
@@ -82,6 +82,15 @@ def container_is_running(pod: V1Pod, container_name: str) -> bool:
     return container_status.state.running is not None
 
 
+def get_container_termination_message(pod: V1Pod, container_name: str):
+    try:
+        container_statuses = pod.status.container_statuses
+        container_status = next(iter([x for x in container_statuses if x.name == container_name]), None)
+        return container_status.state.terminated.message if container_status else None
+    except AttributeError:
+        return None
+
+
 @dataclass
 class PodLoggingStatus:
     """Used for returning the status of the pod and last log time when exiting from `fetch_container_logs`"""
diff --git a/tests/providers/cncf/kubernetes/operators/test_kubernetes_pod.py b/tests/providers/cncf/kubernetes/operators/test_kubernetes_pod.py
index ebe6030f76..0faa4ef8e1 100644
--- a/tests/providers/cncf/kubernetes/operators/test_kubernetes_pod.py
+++ b/tests/providers/cncf/kubernetes/operators/test_kubernetes_pod.py
@@ -550,7 +550,7 @@ class TestKubernetesPodOperator:
         remote_pod_mock.status.phase = 'Failed'
         self.await_pod_mock.return_value = remote_pod_mock
 
-        with pytest.raises(AirflowException, match=f"Pod {name_base}.[a-z0-9]+ returned a failure: .*"):
+        with pytest.raises(AirflowException, match=f"Pod {name_base}.[a-z0-9]+ returned a failure:.*"):
             context = create_context(k)
             k.execute(context=context)