title: Get task_id of failed task from within an Airflow DAG level failure callback
body: <p>I have an airflow DAG with multiple tasks and I have defined success and failure callbacks so that they send a Microsoft Teams message to a channel with some information about the success or failure.</p>
  <p>In the failure callback I am trying to catch the name of the task_id (or one of the task_ids) that has failed. My Dag is defined as:</p>
  <pre class="lang-py prettyprint-override"><code>with DAG(
      dag_id=&quot;my_dag&quot;,
      start_date=datetime(2022, 2, 9),
      schedule_interval=&quot;0 9-18 * * *&quot;,
      max_active_runs=1,
      dagrun_timeout=timedelta(minutes=60),
      on_failure_callback=fail_notifier,
      catchup=False
  ) as dag:
  
      get_params = PythonOperator(
          task_id=&quot;task_1&quot;,
          python_callable=get_runtime_params
      )
  
      final_task = PythonOperator(
          task_id=&quot;final_task&quot;,
          python_callable=run_final_task,
          trigger_rule=&quot;none_failed_or_skipped&quot;,
          on_success_callback=success_notifier
      )
  
      for _id, name in my_dict.items():
  
          with TaskGroup(group_id=name) as tg:
  
              calculation = PythonOperator(
                  task_id=&quot;calculation_task&quot;,
                  python_callable=&quot;calculation_task&quot;,
                  op_kwargs={'my_id': _id}
              )
  
              calc_perf
  
          chain(get_params, tg, final_task)
  </code></pre>
  <p>so you can see that the <code>on_faiure_callback</code> is defined at the DAG level.</p>
  <p>In my fail_notifier definition (I won't post the full code as it's quite lengthy) I've tried a few different ways to get the failed task instance of the DAG. However, when it runs, the behaviour is not what I am expecting. In all cases it picks a random task_id from within the DAG.</p>
  <p><strong>Attempt 1</strong>: Using the <code>task_instance</code> object
  <code>task_context</code> gets passed to the callback methods so I tried the following:</p>
  <pre class="lang-py prettyprint-override"><code>task_instance = task_context['ti']
  task_id = task_instance.task_id
  </code></pre>
  <p><strong>Attempt 2</strong>: Using the <code>task_instance_key_str</code>
  the <code>task_instance_key_str</code> is a string defined in the docs <a href="https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html#:%7E:text=%7B%7B%20task_instance_key_str%20%7D%7D" rel="nofollow noreferrer">here</a> my idea here was to parse the task_id from the task_instance_key_str using some regex e.g.:</p>
  <pre class="lang-py prettyprint-override"><code>key_str = task_context.get('task_instance_key_str')
  task_id = _parse_task_id_from_key_str(key_str)
  </code></pre>
  <p>Again, this attempt seemed to produce random results for which task_id was returned.</p>
  <p><strong>Attempt 3</strong>: get task instances from the dag run and filter for failed</p>
  <pre class="lang-py prettyprint-override"><code>dag_run = context.get('dag_run')
  failed_tis = dag_run.get_task_instances(state='failed')
  failed_task_ids = [ti.task_id for ti in failed_tis]
  first_task_id = failed_task_ids[:1] or None
  </code></pre>
  <p>In this case the callback doesn't run at all. Coupled with this, nothing appears in the scheduler logs to give any indication why the callback is failing in this case based on <a href="https://stackoverflow.com/questions/57402172/how-to-log-output-from-airflow-dag-for-debugging">this answer</a> it seems any exceptions raise in the dag level callback should appear in the scheduler logs. The code definitely works at task level as I have tested printing the list of tasks to the task log.</p>
  <p>I am using an Astronomer distribution of airflow. Astro version is 2.1.1.post6 and Airflow version is 2.1.1.</p>
  <p>I had considered adding the failure callback at the task level but this is infeasible, since my users would be spammed with multiple messages if more than one task failed (because in the example above multiple tasks run concurrently).</p>
  <p>I have been scratching my head with this problem for a few weeks, and exhausted all useful resources I could find online and on SO. Any help would be greatly appreciated</p>