KubernetesPodOperator xcom push failure

[j-svensmark]

Apache Airflow version

2.5.0

What happened

Kubernetes pod operator failed to push xcom value.

After upgrading airflow from 2.2.4 to 2.5.0 (and apache-airflow-providers-cncf-kubernetes from 3.0.2 to 5.0.0) pushing of xcom values from kubernetes pod operator stopped working.

What you think should happen instead

Example of log before the upgrade

[2023-01-17 06:57:54,357] {pod_launcher.py:313} INFO - Running command... cat /airflow/xcom/return.json

[2023-01-17 06:57:54,398] {pod_launcher.py:313} INFO - Running command... kill -s SIGINT 1

[2023-01-17 06:57:55,012] {pod_launcher.py:186} INFO - ["No non-accuracy metrics changed more than 10.0% between 2023-01-15 and 2023-01-16\n"]

and after the upgrade

[2023-01-18T07:12:32.784+0900] {pod_manager.py:368} INFO - Checking if xcom sidecar container is started.
[2023-01-18T07:12:32.804+0900] {pod_manager.py:372} INFO - The xcom sidecar container is started.
[2023-01-18T07:12:32.845+0900] {pod_manager.py:407} INFO - Running command... if [ -s /airflow/xcom/return.json ]; then cat /airflow/xcom/return.json; else echo __airflow_xcom_result_empty__; fi

[2023-01-18T07:12:32.895+0900] {pod_manager.py:407} INFO - Running command... kill -s SIGINT 1

[2023-01-18T07:12:33.405+0900] {kubernetes_pod.py:431} INFO - Result file is empty.

Looking at other timestamps in the log file, it appears that the xcom sidecar is run before the pod finishes, instead of waiting until the end.

How to reproduce

I used airflow.providers.cncf.kubernetes.operators.kubernetes_pod.KubernetesPodOperator with get_logs=False.

Operating System

Debian GNU/Linux 11 (bullseye) (based on official airflow image)

Versions of Apache Airflow Providers

apache-airflow-providers-cncf-kubernetes 3.0.2 and 5.0.0

Deployment

Official Apache Airflow Helm Chart

Deployment details

Deployed on GKE using the official helm chart

Anything else

No response

Are you willing to submit 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!

[vchiapaikeo]

Fwiw, I'm on 2.5.0 with cncf 5.0.0 and cannot repro this on our installation. @j-svensmark , can you share a minimal example dag that shows this issue?

[j-svensmark]

Sure, I'm seeing the issue with the following DAG

import airflow
import os
import pendulum

from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator


local_tz = pendulum.timezone("Asia/Tokyo")

default_task_args = {
    "start_date": local_tz.convert(airflow.utils.dates.days_ago(1)),
}

dag = DAG(
    "debug_kube_xcom",
    default_args=default_task_args,
)

for get_logs in [False, True]:
    KubernetesPodOperator(
        namespace=os.getenv("KUBE_NAMESPACE"),
        image='alpine',
        cmds=["sh", "-c", "sleep 3; mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json"],
        name=f"write-xcom-get-logs-{get_logs}",
        do_xcom_push=True,
        is_delete_operator_pod=True,
        in_cluster=True,
        task_id=f"write-xcom-get-logs-{get_logs}",
        get_logs=get_logs,
        dag=dag
    )

(The namespace line is specific to my setup, so you probably need to change it if running this)

Here the write-xcom-get-logs-True task gets the xcom value, but the write-xcom-get-logs-False task fails to get it.

[vchiapaikeo]

Ah yes, I am seeing some unusual behavior here. Sorry, I missed the part where get_logs is set to False. My pod ends up staying pending in not yet started state.

Containers show the following:

[hussein-awala]

I have the same problem with some of my pod tasks, I just created this PR, can you check it please?

[amrapali-kumari]

I'm also getting the same issue , whats this command running in if condition can anyone suggest how should I proceed with this :
[2023-01-30, 09:43:45 UTC] {pod_manager.py:385} INFO - Running command... if [ -s /airflow/xcom/return.json ]; then cat /airflow/xcom/return.json; else echo airflow_xcom_result_empty; fi

[2023-01-30, 09:43:45 UTC] {pod_manager.py:385} INFO - Running command... kill -s SIGINT 1
Below is my code
dag = DAG(
dag_id=DAG_NAME,
default_args=DEFAULT_ARGS,
schedule_interval=SCHEDULE_INTERVAL,
max_active_runs=MAX_ACTIVE_RUNS,
catchup=CATCHUP,
)

start = DummyOperator(task_id="start", dag=dag)

print_msg = BashOperator(
task_id="print_msg", bash_command="echo 'I am working fine'", dag=dag
)
task_logger.info("print msg task executed successfully")
t2 = KubernetesPodOperator(
name="docker-testing",
namespace="airflow",
image="test",
cmds=["bash", "-cx"],
arguments=["echo", "Hello world"],
labels={"foo": "bar"},
# in_cluster=in_cluster,
task_id="dry_run_demo",
do_xcom_push=True,
# config_file=config_file,
is_delete_operator_pod=False,
get_logs=True,
dag=dag,
)
task_logger.info("docker class executed successfully")
end = DummyOperator(task_id="end", dag=dag)

start >> print_msg >> t2 >> end

[potiuk]

can anyone suggest how should I proceed with this

Applying the PR from @hussein-awala and verifying that it solves the issue (and confirming it here) might be a sensible thing to do.