Regression on pid reset to allow task start after heartbeat

[nmehraein]

It's a bug we have found in production in my company.

Here is the bug pattern:

• The task job is started with a previous run of taskinstance (retry) and an old pid
• The job take time to start for some reason (db load, system load ...) and not update the pid in database
• Then the task is killed by heartbeat cause the pid is different from pid in database

It's critical in case of sensors with reschedule mode cause it trigger a fail immediately.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[ephraimbuddy]

Do you have a dag that could be used to reproduce this behaviour?

[nmehraein]

Do you have a dag that could be used to reproduce this behaviour?

@ephraimbuddy IMO it's not a bug that can be reproduced in a simply dag cause it not concern task execution code but task runner.
To reproduce we have to slowdown the task starting process:

• In the task pid update in https://github.com/apache/airflow/blob/main/airflow/models/taskinstance.py#L1180
• Or in the task fork from runner after forking https://github.com/apache/airflow/blob/main/airflow/task/task_runner/standard_task_runner.py#L54
• Or in the raw task command line (executed by scheduler) https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L156

If the starting process take more time than heartbeat (5s default), and it's not the first TI run, the problem occur.

I don't know if you have another idea to reproduce that in a dag ?

In my case, it occur more specifically in production at night, when the scheduler process our data of last day, and lot of task is running in parallel that generate load on our servers and our airflow db.

[nmehraein]

Maybe our first log with that error from airflow can help

[2021-07-18 03:09:34,879] {{logging_mixin.py:104}} INFO - Running <TaskInstance: task_name 2021-07-17T00:00:00+00:00 [running]> on host xxxxxxxxxxx
[2021-07-18 03:09:34,885] {{local_task_job.py:195}} WARNING - Recorded pid 4932 does not match the current pid 14922
[2021-07-18 03:09:34,893] {{process_utils.py:100}} INFO - Sending Signals.SIGTERM to GPID 14922
[2021-07-18 03:09:34,900] {{process_utils.py:66}} INFO - Process psutil.Process(pid=14922, status='terminated', exitcode=<Negsignal.SIGTERM: -15>, started='03:09:29') (14922) terminated with exit code Negsignal.SIGTERM

[jedcunningham]

Should this be up with lines 1043-1044 instead?

[nmehraein]

Yes, i put it back on line where it was in 2.0 , but I'm going to push a commit to put it next to the hostname which I think is a good place too. https://github.com/apache/airflow/blob/2.0.2/airflow/models/taskinstance.py#L1076

[ephraimbuddy]

Quite understandable(makes sense to me). Let's wait for @ashb to also take a look

[ashb]

LGTM @ephraimbuddy

[boring-cyborg]

Awesome work, congrats on your first merged pull request!

[laserpedro]

I incorporated the change https://github.com/apache/airflow/blob/2.0.2/airflow/models/taskinstance.py#L1076 and I am still getting the error ....