skip_on_exit_code parameter in KPO does not take effect

[mohittalele]

Apache Airflow version

Other Airflow 2 version (please specify below)

What happened

I am using following task to simulate the skipping of KPO

@task.kubernetes(image="python:3.8-slim-buster", namespace="dev", skip_on_exit_code=100 )
def print_pattern():
    import sys
    some_condition = True
    if some_condition :
        sys.exit(100)

This task task results in the following logs -

'container_statuses': [{'container_id': 'containerd://0e38f55c0d0b8ac21b2d0d4d4a58a0f',
                'image': 'docker.io/library/python:3.8-slim-buster',
                'image_id': 'docker.io/library/python@sha256:8799b0564103a9f36cfb8a8e1c562e11a9a6f2e3bb214e2adc23982b36a04511',
                'last_state': {'running': None,
                               'terminated': None,
                               'waiting': None},
                'name': 'base',
                'ready': False,
                'restart_count': 0,
                'started': False,
                'state': {'running': None,
                          'terminated': {'container_id': 'containerd://0cd4eddd219dd25b658d240e675c59d0a0e38f55c0d0b8ac21b2d0d4d4a58a0f',
                                         'exit_code': 100,
                                         'finished_at': datetime.datetime(2023, 8, 23, 9, 38, 9, tzinfo=tzlocal()),
                                         'message': None,
                                         'reason': 'Error',
                                         'signal': None,
                                         'started_at': datetime.datetime(2023, 8, 23, 9, 38, 8, tzinfo=tzlocal())},
                          'waiting': None}}],

The state in airflow upon execution of task is failed.

What you think should happen instead

I would expect the task to be skipped based on "skip_on_exit_code" paramter.

How to reproduce

Run the task in airflow installed using latest helm chart version 1.10.0 and airflow 2.6.2

Operating System

"Debian GNU/Linux 11 (bullseye)

Versions of Apache Airflow Providers

apache-airflow-providers-amazon          | 8.1.0
apache-airflow-providers-celery          | 3.2.0
apache-airflow-providers-cncf-kubernetes | 7.0.0
apache-airflow-providers-common-sql      | 1.5.1
apache-airflow-providers-docker          | 3.7.0
apache-airflow-providers-elasticsearch   | 4.5.0
apache-airflow-providers-ftp             | 3.4.1
apache-airflow-providers-google          | 10.1.1
apache-airflow-providers-grpc            | 3.2.0
apache-airflow-providers-hashicorp       | 3.4.0
apache-airflow-providers-http            | 4.4.1
apache-airflow-providers-imap            | 3.2.1
apache-airflow-providers-microsoft-azure | 6.1.1
apache-airflow-providers-microsoft-mssql | 3.2.0
apache-airflow-providers-mysql           | 5.1.0
apache-airflow-providers-odbc            | 3.3.0
apache-airflow-providers-postgres        | 5.5.0
apache-airflow-providers-redis           | 3.2.0
apache-airflow-providers-sendgrid        | 3.2.0
apache-airflow-providers-sftp            | 4.3.0
apache-airflow-providers-slack           | 7.3.0
apache-airflow-providers-snowflake       | 4.1.0
apache-airflow-providers-sqlite          | 3.4.1
apache-airflow-providers-ssh             | 3.7.0

Deployment

Official Apache Airflow Helm Chart

Deployment details

k8s version v1.24.16

Anything else

The airflow code base uses laststate.terminated.exit_code for matching the exit code as described here. However correct code should state.terminated.exit_code.

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

[SamWheating]

I think that this is due to the last_state block being all None in that container status object. Here's the code which is used to determine if we should skip based on the exit code:

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

Lines 723 to 734 in 7141c42

	exit_code = (
	base_container_status.last_state.terminated.exit_code
	if base_container_status
	and base_container_status.last_state
	and base_container_status.last_state.terminated
	else None
	)
	if exit_code in self.skip_on_exit_code:
	raise AirflowSkipException(
	f"Pod {pod and pod.metadata.name} returned exit code "
	f"{self.skip_on_exit_code}. Skipping."
	)

So its relying on the exit code being available under base_container_status.last_state.terminated.exit_code, which it wasn't in your case 🤔 I'm not completely familiar with the expected structure of these objects but I'll do some digging and see if this is something we can fix airflow-side.

Feel free to assign this one to me 🕵️

[potiuk]

Nice catch indeed. @SamWheating -> I suspect that the condition is wrong. From what I understand (just looked it up) last_state is the state from previous try in case the container has been restarted. But in this case (and in general) last_state wil never be set because the container is not set to restart.