Optional: Enriching with an LLM
This chapter is optional. Nothing here is required for Week 15's deliverables or the assignment. Build it only when your required pipeline runs green in production and your team wants a Slack message when a scheduled task fails.
Your pipeline runs at 07:00 UTC, before anyone on your team is at a laptop. Airflow's response to a failed task is to colour a square red and wait. If nobody opens the UI, a broken run stays broken. You find out when someone notices the app is showing yesterday's data, usually a teammate asking why the site looks wrong.
This chapter is how you get told without opening Airflow.
By the end of this chapter, you should be able to:
slack_alert callback is attached, and say why it lives in default_argsMonitoring means something reaches a human without that human going to look. Airflow gives you the opposite by default: perfect records, delivered to nobody.
It is worth being precise about what you are asking for, because "add monitoring" is vague enough to build the wrong thing. You want one message, when a task fails, naming what broke and linking to the log. That is it. A dashboard nobody has open at 07:00 does not solve this. Neither does a daily summary email, which tells you at 09:00 about a failure you could have known about at 07:04.
<aside> 🤓 Curious Geek: the pager that started it
On-call rotations come from telephony and hospitals, where a physical pager buzzed the one person responsible. The software version kept the shape: one alert, one owner, one thing to do. The tools got better and the principle did not change.
</aside>
Your repo ships dags/alerts.py, and it is already attached. Open dags/pipeline_dag.py and find these two places:
from alerts import slack_alert
DEFAULT_ARGS = {
"owner": "data-team",
"retries": 3,
"retry_delay": timedelta(minutes=5),
# Inherited by every task, including the one you add at 11pm on a Thursday.
"on_failure_callback": slack_alert,
}
DEFAULT_ARGS then reaches the DAG through the decorator:
@dag(
dag_id="final_project_pipeline",
schedule="0 9 * * *",
default_args=DEFAULT_ARGS,
...
)
def final_project_pipeline():
So there is nothing to wire up. Your job in this chapter is to understand why it sits there, and to prove it works. An inherited callback you have never seen fire is a line of code, not a safety net.
Putting it in default_args rather than on each task is the part that matters. A task you add in week 4 is covered because it inherits the callback, not because you remembered to wire it up. The task you forget is always the one that breaks.
<aside>
📘 Recap from Week 12: default_args applies to every task in the DAG, the same inheritance you used for retries in Sequential Pipeline Steps.
</aside>
The message names the DAG, the task, the attempt number, and the reason. The reason is the last line of the traceback, which is the line that names what actually went wrong. When AIRFLOW_BASE_URL is set, it also links straight to the task log.
Read the bottom of alerts.py and you will find the whole callback wrapped in try / except, printing instead of raising:
except Exception as exc:
print(f"alerts: could not post the failure alert: {type(exc).__name__} {exc}")
That looks like sloppy error handling until you think about when this code runs: only after something has already failed. If the alerting itself raised, one broken task would become two, and the second error would be the one you see first. The real failure would be buried under a Slack problem.
A Slack incoming webhook is a credential: anyone holding the URL can post to your channel. So it lives in Key Vault as fp-slack-webhook (Optional matrix), and the callback reads it through the VM's managed identity at the moment a failure happens.
Two consequences worth stating:
.env, not in airflow_settings.yaml, not in a comment.<aside> 📘 Recap from Week 12: you already fetched your own Airflow password from Key Vault in The Shared Airflow Instance. Same vault, same idea, one difference: there you read a secret as yourself, and here the VM reads it as itself.
</aside>
An alert nobody has seen arrive is a configuration file, not an alert. Wiring that looks right and silently does nothing is the normal outcome, not the unlucky one. The only way to find out is to fail on purpose.
Step 1: Add a task to your DAG that fails immediately:
@task
def fail_on_purpose():
raise RuntimeError("testing the alert, delete me")
Step 2: Trigger the DAG and let the task fail.
Step 3: Confirm the Slack message arrives. Check that it names the DAG, the task, the attempt number, and the reason, and that the reason matches the error you raised.
Step 4: Delete the task and run the DAG green again.
<aside>
💡 The message ends with a link to the task log only when AIRFLOW_BASE_URL is set on the machine running Airflow. Without it the alert still arrives, minus the link. If yours has no link, that is configuration, not a broken callback.
</aside>
Do this once, early, while you still have time to fix it. Discovering in week 5 that alerting was never connected means you spent four weeks believing you had monitoring.
If no message arrives, open the failed task's log first. The callback prints what it did. Either it posted and Slack accepted it, or the secret is not a webhook yet, or the Key Vault read failed. All three say so in plain words.
The other optional chapters sit next to this one. You only need a green scheduled path and a working Slack webhook on the VM.
on_failure_callback in DEFAULT_ARGS and can say why it belongs thereon_failure_callbackNext up: Optional: Enriching with an LLM, or skip ahead to Team Timelines if you are done with optionals.
The HackYourFuture curriculum is licensed under CC BY-NC-SA 4.0 https://hackyourfuture.net/
Built with ❤️ by the HackYourFuture community · Thank you, contributors
Found a mistake or have a suggestion? Let us know in the feedback form.