Kubernetes Executor: non-existent upstream failed

[bertrand-buffat]

Apache Airflow version

2.1.3

Operating System

Debian GNU/Linux

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==1.4.0
apache-airflow-providers-cncf-kubernetes==2.0.2

Deployment

Other 3rd-party Helm chart

Deployment details

No response

What happened

Some tasks are randomly set to Upstream Failed while upstream task actually succeeded

What you expected to happen

I expect downstream tasks to be scheduled

How to reproduce

Failure happens randomly during peak hours of scheduling

We use Kubernetes Executor

Config for scheduler:

[scheduler]
job_heartbeat_sec = 5
clean_tis_without_dagrun_interval = 15.0
scheduler_heartbeat_sec = 5
num_runs = -1
processor_poll_interval = 1
min_file_process_interval = 300
dag_dir_list_interval = 300
print_stats_interval = 30
pool_metrics_interval = 5.0
scheduler_health_check_threshold = 30
orphaned_tasks_check_interval = 300.0
child_process_log_directory = /home/airflow/logs/scheduler
scheduler_zombie_task_threshold = 300
catchup_by_default = True
max_tis_per_query = 512
use_row_level_locking = True
max_dagruns_to_create_per_loop = 500
max_dagruns_per_loop_to_schedule = 500
schedule_after_task_execution = True
parsing_processes = 8
file_parsing_sort_mode = alphabetical
use_job_schedule = True
allow_trigger_in_future = False
dependency_detector = airflow.serialization.serialized_objects.DependencyDetector
run_duration = -1
statsd_port = 8125
statsd_prefix = airflow
max_threads = 5
authenticate = False

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!

[sweco]

Hey there, we're observing the same issue on Airflow 2.1.4 in a big DAG having hundreds of tasks. By looking in the scheduler logs from the moment when the downstream task was marked as upstream_failed it seems that it happened when a pod with a task could not be started.

However, the question is, why is the task put in a failed state instead of retrying? And if this is normal, why does not the downstream wait for all the retries to be finished?

Pod "pod_name" has been pending for longer than 300 seconds. It will be deleted and set to failed.
Event: pod_name had an event of type MODIFIED
Event: pod_name Pending
Event: pod_name had an event of type DELETED
Event: Failed to start pod pod_name
Attempting to finish pod; pod_id: pod_name; state: failed; annotations: {..., 'try_number': 1}
Changing state of task_instance, <TaskInstanceState.FAILED: 'failed'>, 'pod_name', 'airflow', '...') to failed
Executor reports execution of task_id execution_date=execution_date exited with status failed for try_number 1
Executor reports task instance <TaskInstance: task_id execution_date [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally?
Setting task instance <TaskInstance: task_name execution_date [queued]> state to failed as reported by executor

# Later on

1 tasks up for execution:
<task_id execution_date [scheduled]>

Also, looking at the successful task, it indeed has two tries, from which the first one really finished at the incriminated timestamp.

[sweco]

Steps to reproduce

1. Prepare a relatively small cluster, e.g. 1 node with 8 GB of RAM.

2. Deploy Airflow using Kubernetes executor there.

3. Prepare a DAG according to the following specification

   • It tries to run 5 tasks concurrently
   • Each task specifies 2 GB of RAM in their pod requests - i.e. the sum of requested RAM is greater than the total available RAM (5 * 2 GB = 10 GB > 8 GB)
   • Each task is taking more than 300 seconds (the scheduler does not crash immediately after 300 seconds, so be sure to put some reserve, e.g. 600 seconds)

   t1 = []
   for i in range(5):
       t = BashOperator(
           task_id=f't1_{i}',
           bash_command='sleep 600 && echo Hello World!',
           executor_config={
               'pod_override': k8s.V1Pod(
                   spec=k8s.V1PodSpec(
                       containers=[
                           k8s.V1Container(
                               name="base",
                               resources=k8s.V1ResourceRequirements(
                                   requests={
                                       'memory': '2000Mi',
                                       'cpu': '1000m'
                                   }
                               )
                           )
                       ]
                   )
               )
           }
       )
       t1.append(t)
   
   t2 = BashOperator(
       task_id='t2',
       bash_command='sleep 20 && date',
   )
   
   t1 >> t2

4. Run the DAG, you should see that at least one of the tasks' pods will not be able to be run and will thus be in Pending state.

   

5. After ~6 minutes the pods will be deleted by the scheduler, marked as up_for_retry and at the same time t2 will be marked as upstream_failed. The scheduler reports this error log for each task:

   ERROR - Pod "internaltestt1..." has been pending for longer than 300 seconds.It will be deleted and set to failed.
   

   In the next screenshot we can see the upstream task, whose pod was deleted, marked as up_for_retry and t2 marked as upstream_failed even though the upstream task can still succeed running on the second try.

   

[romainfage]

Hello all!

When looking at the issue and logs, We may have understood why the task is set to upstream_failed whereas the task is retrying. (we are using Airflow 2.1.3)

We are getting this logs inside our scheduler:

{kubernetes_executor.py:368} INFO - Attempting to finish pod; pod_id: POD_ID_1; state: failed; annotations: {'dag_id': 'DAG1', 'task_id': 'TASK_ID', 'execution_date': '2021-10-29T13:05:00+00:00', 'try_number': '1'}
{kubernetes_executor.py:546} INFO - Changing state of (TaskInstanceKey(dag_id='DAG1', task_id='TASK_ID', execution_date=datetime.datetime(2021, 10, 29, 13, 5, tzinfo=tzlocal()), try_number=1), <TaskInstanceState.FAILED: 'failed'>, 'POD_ID_1', 'airflow', '164023816') to failed
{scheduler_job.py:611} INFO - Executor reports execution of DAG1.TASK_ID execution_date=2021-10-29 13:05:00+00:00 exited with status failed for try_number 1
{scheduler_job.py:647} ERROR - Executor reports task instance <TaskInstance: DAG1.TASK_ID 2021-10-29 13:05:00+00:00 [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally?
{scheduler_job.py:654} INFO - Setting task instance <TaskInstance: DAG1.TASK_ID 2021-10-29 13:05:00+00:00 [queued]> state to failed as reported by executor

Our understanding is (as demonstrate @sweco), that K8S executors try to finished the pod that is not launched yet (maybe a timeout of any kind, I cannot currently pin point where and why the executors do that). That is why we see this log:

{kubernetes_executor.py:368} INFO - Attempting to finish pod; pod_id: POD_ID_1; state: failed; annotations: {'dag_id': 'DAG1', 'task_id': 'TASK_ID', 'execution_date': '2021-10-29T13:05:00+00:00', 'try_number': '1'}
{kubernetes_executor.py:546} INFO - Changing state of (TaskInstanceKey(dag_id='DAG1', task_id='TASK_ID', execution_date=datetime.datetime(2021, 10, 29, 13, 5, tzinfo=tzlocal()), try_number=1), <TaskInstanceState.FAILED: 'failed'>, 'POD_ID_1', 'airflow', '164023816') to failed

Afterward, the scheduler received an event from the scheduler that change the state of the dag to failed. We can see the reception of the event here:

{scheduler_job.py:611} INFO - Executor reports execution of DAG1.TASK_ID execution_date=2021-10-29 13:05:00+00:00 exited with status failed for try_number 1

Afterward, the scheduler try to handle the event, to do so it look at the current state inside the metadata db of the task. However, since the pod did not start yet, the state of the task is still queued. The scheduler think because the old state is queued and the new state is failed, it is not a classic behavior so someone external should have done something. That is why we see the log:

{scheduler_job.py:647} ERROR - Executor reports task instance <TaskInstance: DAG1.TASK_ID 2021-10-29 13:05:00+00:00 [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally?

We can notice the question from the scheduler Was the task killed externally?, meaning that it is did not sure to handle well the case.

When looking at the code, we can pinpoint here. Preview:

    @provide_session
    def _process_executor_events(self, session: Session = None) -> int:
            [...]
            if ti.try_number == buffer_key.try_number and ti.state == State.QUEUED:
                Stats.incr('scheduler.tasks.killed_externally')
                msg = (
                    "Executor reports task instance %s finished (%s) although the "
                    "task says its %s. (Info: %s) Was the task killed externally?"
                )
                self.log.error(msg, ti, state, ti.state, info)

                request = TaskCallbackRequest(
                    full_filepath=ti.dag_model.fileloc,
                    simple_task_instance=SimpleTaskInstance(ti),
                    msg=msg % (ti, state, ti.state, info),
                )
                self.log.info('Setting task instance %s state to %s as reported by executor', ti, state)
                ti.set_state(state)
                self.processor_agent.send_callback_to_execute(request)

By consequence, as the scheduler think something external marks the task as failed and the scheduler marks downstream tasks as upstream_failed. Then, afterward the task is retried, finished in success and is green in Airflow but no downstream tasks are launched since they are in state upstream_failed.

I don't have currently enough insight on how the Airflow scheduler code really work, so I cannot provide a proper fix to this problem, but to fix the problem, may we do:

• Provide another failure state to make difference between external task kill and airflow task kill?
• Provide more information in the event received by the scheduler from the executor on the nature of the kill?
• When the task retries, it clear the down stream task state, therefore they will not be in upstream_failed state?

Thank you for reading me

[eladkal]

cc @jedcunningham

[kehao95]

We are experiencing the same issue at v2.1.4. task is showing upstream_failed
It may be related to resources issue. We have 2 env dev&prod. The Dev has very limited resources which this happened frequently. And Production has much larger CPU/Mem quotas where the issue happened less frequently.

---

Update: experiencing the same issue at v2.1.2 after downgrading.

[aprettyloner]

Hi @jedcunningham @eladkal has this bug been fixed in 2.2.1+?

version: 2.1.4
gitSync: enabled

We are seeing this intermittently every day with both complex and simple dags.

More complex dags

Simple linear dags

[aprettyloner]

Hi @jedcunningham @eladkal has this bug been fixed in 2.2.1+?

version: 2.1.4 gitSync: enabled

We are seeing this intermittently every day with both complex and simple dags.

More complex dags

Simple linear dags

cc @ephraimbuddy

[eladkal]

@aprettyloner does the issue still happen in latest airflow version + latest provider?

[bertrand-buffat]

In our case, we implemented a dag that monitored the problem every 10 minutes and cleared the blocked tasks if any. Then the error ceased to appear after an upgrade from aws.
I can close the issue if you want.