Add option to suppress showing json config on KubernetesPodOperator failure

[iljau]

Description

Currently if container launched via KubernetesPodOperator returns with non-zero exit code, whole pod configuration is printed to logs and this multiple times due to exception re-raising. This makes logs unreadable.

There should be some method to avoid showing pod config json in case of non-zero exit code.

[2021-10-22, 17:48:23 UTC] {{taskinstance.py:1686}} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py", line 369, in execute
    raise AirflowException(f'Pod {self.pod.metadata.name} returned a failure: {remote_pod}')
airflow.exceptions.AirflowException: Pod ... returned a failure: [ .. LONG JSON HERE .. ]

Sample DAG:

import secrets
from datetime import datetime

from airflow import DAG
from airflow.configuration import conf
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import (
    KubernetesPodOperator,
)
from kubernetes.client import models as k8s

default_args = {
    "owner": "airflow",
    "start_date": datetime(2021, 9, 1),
}


with DAG(
    dag_id="test.pod_logs",
    default_args=default_args,
    schedule_interval=None,
    max_active_runs=1,
) as dag:
    namespace = conf.get("kubernetes", "namespace")

    pod_name = "pod-test-pod-logs-%s" % secrets.token_urlsafe(8)

    container_debug = k8s.V1Container(
        name="ubuntu-container",
        image="ubuntu:focal-20211006",

        command=["/bin/bash", "-c"],
        args=["""
        set -x;
        echo 1;
        echo 2;
        echo 3;
        exit 1;
        """],
        resources=k8s.V1ResourceRequirements(
            limits={
                "cpu": "1000m",
                "memory": "1024Mi",
            },
            requests={
                "cpu": "1000m",
                "memory": "1024Mi",
            },
        )
    )

    pod = k8s.V1Pod(
        api_version="v1",
        kind="Pod",
        spec=k8s.V1PodSpec(
            image_pull_secrets=[k8s.V1LocalObjectReference('dockerhub-data')],
            containers=[
                container_debug,
            ],
        ),
        metadata=k8s.V1ObjectMeta(
            name=pod_name,
        ),
    )

    task_run_pod = KubernetesPodOperator(
        task_id="task-pod-run",
        name=pod_name,
        namespace=namespace,
        # is_delete_operator_pod=True,
        is_delete_operator_pod=False,
        in_cluster=True,
        full_pod_spec=pod,
        # image_pull_policy="Always",
        image_pull_policy="IfNotPresent",
    )

Use case/motivation

No response

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template!

[iljau]

Work-around could be:

import textwrap

from airflow.exceptions import AirflowException
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import (
    KubernetesPodOperator,
)


class CustomKubernetesPodOperator(KubernetesPodOperator):
    def execute(self, context):
        try:
            super().execute(context)
        except AirflowException as e:
            msg = str(e)
            short_msg = textwrap.shorten(msg, width=160)
            raise AirflowException(short_msg) from None

[raphaelauv]

A pod failing should be a rare event, and then having all the configuration is very helpful