How to get context object in sla_miss_callback function

user_11809659 0
Viewed 711

I am able to successfully implement and test on_success_callback and on_failure_callback in Apache Airflow including successfully able to pass parameters to them using context object. However I am not able to successfully implement sla_miss_callback . By going through different online sources I found that arguments that get passed on to this function is

dag, task_list, blocking_task_list, slas, blocking_tis

However the sla_miss_callback unlike success/failure callback doesn't get the context object in its argument list and if I am trying to run Multiple set of operators like Python, Bash Operators they fail and scheduler complains for not passing context to execute function.

SCHEDULER ERROR For SLA MISS CALLBACK function

I tried looking at other online sources and in just one (https://www.rea-group.com/blog/watching-the-watcher/) I found that we can extract context object by using the self object . So I appended self to the additional 5 arguments described above but it didn't work for me. I want to know how is it possible to retrieve or pass context object to sla_miss_callback function not only for running different operators but also retrieving other details about the dag which has missed the SLA

airflow-scheduler airflow python
1 Answers

It seems it is not possible to pass the context dictionary to the SLA callback (see source code for sla_miss_callback) but I've found a reasonable workaround to access some other information about the dag-run such as dag_id, task_id, and execution_date. You can also use any of the build-in macros/parameters which should work fine. While I am using the SlackWebhookOperator for my other callbacks, I am using SlackWebhookHook for the sla_miss_callback. For example:

from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook

def sla_miss_callback(dag, task_list, blocking_task_list, slas, blocking_tis, *args, **kwargs):
dag_id = slas[0].dag_id
task_id = slas[0].task_id
execution_date = slas[0].execution_date.isoformat()
hook = SlackWebhookHook(
    http_conn_id='slack_callbacks',
    webhook_token=BaseHook.get_connection('slack_callbacks').password,
    message=f"""
        :sos: *SLA has been missed*
        *Task:* {task_id}
        *DAG:* {dag_id}
        *Execution Date:* {execution_date}
        """
)
hook.execute()
user_119230250
Related