airflow-failure-alerts 0.1.0

Global Apache Airflow task-failure alerting to MAX and Telegram (cluster policy via entry point)

airflow-failure-alerts

Global Apache Airflow task-failure alerting to the MAX and Telegram messengers. Installed as a pip package, it uses a cluster policy (the airflow.policy entry point) to attach an on_failure_callback to every task in every DAG - the DAGs themselves are not modified.

It works in managed environments where there is no access to $AIRFLOW_HOME/config or the plugins folder: the package is loaded from site-packages.

How it works

pyproject.toml registers the entry point:

[project.entry-points."airflow.policy"]
_ = "airflow_failure_alerts.policy"

On startup (settings.initialize -> policy plugin loading) Airflow imports airflow_failure_alerts.policy and registers its @hookimpl task_policy. During DAG parsing that policy sets on_failure_callback = notify_failure on every task (existing callbacks are preserved). On terminal task failure (after all retries are exhausted) a message is sent to MAX and Telegram.

Installation

Requires Apache Airflow 2.6+ (the airflow.policy entry point).

pip install airflow-failure-alerts

In a managed Airflow service, add airflow-failure-alerts to the cluster pip dependencies and apply the change (this restarts the components and loads the policy).

Airflow Variables

Set these in the Airflow UI -> Admin -> Variables. Each channel is optional; if a channel is not configured, alerts go only to the configured one.

Variable	Description
TELEGRAM_BOT_TOKEN	Telegram bot token (from @BotFather)
TELEGRAM_CHAT_ID	chat/channel id (channels look like -100...)
MAX_BOT_TOKEN	MAX bot token (from @MasterBot)
MAX_CHAT_ID	MAX chat id
ALERTS_ENV	(optional) environment label added to the message, e.g. prod

Getting a chat_id

• Telegram: add the bot to the chat/channel, post a message, open https://api.telegram.org/bot<TOKEN>/getUpdates and read chat.id. For channels the bot must be an administrator.
• MAX: create a bot via @MasterBot, add it to the chat; chat_id arrives in webhook events or after the first interaction in the chat.

Verification

Add a DAG with a deliberately failing task (retries=0, raise) and trigger it manually. Both messengers receive a message of the form:

Airflow task failed [prod]
DAG: my_pipeline
Task: my_task
Run: manual__2026-01-01T...
Execution date: 2026-01-01T00:00:00+00:00
Try: 1
Error: <exception text>
Log: https://airflow.../log?...

If no alert arrives, inspect the task log (the "Post task execution logs" section): notify_failure records the reason there (Variables not set, HTTP error) and never fails the task.

MAX Bot API

POST https://platform-api.max.ru/messages?chat_id=<id>, header Authorization: Bearer <token>, body {"text": "..."}. The token query parameter is not supported - header only. Text limit is 4000 characters.