airflow: Tasks intermittently get terminated with SIGTERM on K8s executor

tl;dr: set schedule_after_task_execution to false by either updating your airflow.cfg (recommended) or setting AIRFLOW__SCHEDULER__SCHEDULE_AFTER_TASK_EXECUTION to False.

We had the same issue and with the help of sentry to look through the whole stack trace, I found out why.

The buggy code block is taskinstance.py#L1182-L1201:

        # Recording SUCCESS
        self.end_date = timezone.utcnow()
        self.log.info(
            'Marking task as SUCCESS. '
            'dag_id=%s, task_id=%s, execution_date=%s, start_date=%s, end_date=%s',
            self.dag_id,
            self.task_id,
            self._date_or_empty('execution_date'),
            self._date_or_empty('start_date'),
            self._date_or_empty('end_date'),
        )
        self.set_duration()
        if not test_mode:
            session.add(Log(self.state, self))
            session.merge(self)

        session.commit()

        if not test_mode:
            self._run_mini_scheduler_on_child_tasks(session)

Looking at the code: after marking a task SUCCESS and commit, if it is not test mode, it will call a potentially expensive function _run_mini_scheduler_on_child_tasks. And local_task_job.py#L179-L199 will detect the task SUCCESS very soon and since the task is not running, it will terminate the process which might be still executing _run_mini_scheduler_on_child_tasks:

        if ti.state == State.RUNNING:
            ...
        elif self.task_runner.return_code() is None and hasattr(self.task_runner, 'process'):
            self.log.warning(
                "State of this instance has been externally set to %s. " "Terminating instance.", ti.state
            )
            self.task_runner.terminate()

This is proven by the log @saurabhladhe shared (the line numbers diverge a bit because the log was logged by Airflow 2.0.1):

[2021-03-08 21:46:32,051] {taskinstance.py:1166} INFO - Marking task as SUCCESS. dag_id=canary_dag, task_id=print_the_context, execution_date=20210308T201000, start_date=20210308T214431, end_date=20210308T214632

...

[2021-03-08 21:46:51,152] {local_task_job.py:188} WARNING - State of this instance has been externally set to success. Terminating instance.

[2021-03-08 21:46:51,153] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 22

So the mitigation is to make _run_mini_scheduler_on_child_tasks cheap, which is an optimization controlled by schedule_after_task_execution and can be disabled.

    def _run_mini_scheduler_on_child_tasks(self, session=None) -> None:
        if conf.getboolean('scheduler', 'schedule_after_task_execution', fallback=True):

The proper fix would start from changing

        if not test_mode:
            session.add(Log(self.state, self))
            session.merge(self)

        session.commit()

        if not test_mode:
            self._run_mini_scheduler_on_child_tasks(session)

to

        if not test_mode:
            session.add(Log(self.state, self))
            session.merge(self)
            self._run_mini_scheduler_on_child_tasks(session)

        session.commit()

However, _run_mini_scheduler_on_child_tasks might got an OperationalError and roll back the session completely taskinstance.py#L1248. And we will falsely mark the task failure, though it is only the optimization failure. So I’d leave it to someone who knows the code better to fix it properly. Personally I’d suggest to remove this optimization completely.

@lidalei : awesome, thanks for confirming, do let me know if it does show up again, we will probably try this setting on our next deploy as well

I can confirm it solved our issues.

I am getting this issue in airflow 2.1.3 as well. All retires are getting sigkilled. Error : airflow.exceptions.AirflowException: Task received SIGTERM signal

I believe this is fixed with #16289.

@lidalei : awesome, thanks for confirming, do let me know if it does show up again, we will probably try this setting on our next deploy as well

Hello,

I am facing the same issue: airflow 2.1.3 (tested also with 2.1.2) backend: postgresql executor: LocalExecutor

I have modified the variables killed_task_cleanup_time and schedule_after_task_execution to resp. 100000 and False. I have also installed airflow as non root user and set the default run_as_user to be airflow.

My tasks are getting constantly killed in backfill mode with the traceback:

[2021-09-07 10:21:20,185] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 73011

Honestly, I am a bit discouraged at this point, could you help me please ? tks