В рамках онбординга получил несложную задачу. Есть даг worker_heartbeat, который стартует каждую минуту, и отсылает пинг со своего хоста в InfluxDB. Данные в InfluxDB контролирует tick-скрипт deadman, который поднимает тревогу, если с какого то из хостов airflow в течении 15 минут не было сигнала. Проблема в том, что примерно раз в месяц выскакивает ложный алерт, хотя все воркеры работают без поломок.

Причина на самом деле проста - воркеров три. На каком из них будет выполняться таск определяется случайно. И периодически возникают ситуации когда на какой-то из хостов таск конкретно этого дага не попадает 15-20 раз подряд.

Но собственно пост не про причину, а про то что для решения одной и той же задачи можно использовать совершенно разные способы, со своими плюсами и минусами:

1. В лоб - создать три очереди, каждую из которых привязать к одному из воркеров и далее в даге создавать не один, а три таска - отправляя каждый в свою очередь. Соответственно, каждую минуту на каждом из воркеров будет отрабатывать heartbeat - случайных пропусков в базе InfluxDB больше не будет.

2. Но неужели у airflow нет своих инструментов для этого? Смотрим документацию - есть, правда только для Celery Cluster, зато в двух вариантах - для командной строки и API.

3. Но если подумать - есть вариант проще. Что вообще нужно - убедиться что на хосте запускался какой-нибудь таск за последнюю минуту. Так ведь airflow хранит всю эту информацию у себя во внутренней базе данных!

select
    hostname,
    max(start_date) start_date
from airflow.task_instance
group by hostname

hostname         |max(start_date)           |
-----------------+--------------------------+
airflow3.main.in|2024-09-09 15:22:48.906396|
airflow2.main.in|2024-09-09 15:22:47.011965|
airflow1.main.in|2024-09-09 15:22:26.705087|

Можно напрямую законнектится, если снаружи. вне airflow обращаться. А если из дага - то через airflow.models.taskinstance прочитать.

Понравился кейс - не часто так наглядно совершенно разные пути видно.