Metrics - `airflow_ti.start.*` and `airflow_ti.finish.*`
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 7h
- Merged PRs (30d)
- 484
Description
### Apache Airflow version
2.6.2
### What happened
There is an issue with the `airflow_ti.start...` and `airflow_ti.finish...` metrics when running Airflow with OpenTelemetry. Both of those get emitted when run with StatsD but are flaky under OTel.
I am submitting this as an Issue since I will be a little distracted for the next bit and figured someone may be able to have a look in the meantime. Please do not assign it to me, I'll get it when I can is nobody else does.
### What you think should happen instead
Behavior should be consistent.
### How to reproduce
To reproduce, you can run Breeze with the statsd or the otel integration (for example `breeze start-airflow --integration otel`) and run one or more of the following DAGs, then open the [OTel](http://localhost:28889/metrics) or [StatsD](http://localhost:29102/metrics) raw data view to verify.
These two DAGs don't generate any `airflow_ti_finish.*` metrics:
```
from airflow import DAG
from airflow.decorators import task
from airflow.utils.timezone import datetime
@task
def task1():
return 'Hello'
@task
def task2():
return 'World!'
@task
def task3(in1, in2):
print(f'{in1} {in2}')
with DAG(
dag_id='taskflow_demo',
start_date=datetime(2021, 1, 1),
schedule=None,
catchup=False
) as dag:
task3(task1(), task2())
```
```
import time
from airflow import DAG
from airflow.decorators import task
from airflow.utils.timezone import datetime
@task
def task1():
time.sleep(10)
with DAG(
dag_id='sleep_10',
start_date=datetime(2021, 1, 1),
schedule=None,
catchup=False
) as dag:
task1()
```
but this one:
```
import time
from datetime import timedelta
from airflow import DAG
from airflow.decorators import task
from airflow.utils.timezone import datetime
def sla_callback(dag, task_list, blocking_task_list, slas, blocking_tis):
print(
"The callback arguments are: ",
{
"dag": dag,
"task_list": task_list,
"blocking_task_list": blocking_task_list,
"slas": slas,
"blocking_tis": blocking_tis,
},
)
@task(sla=timedelta(seconds=10))
def sleep_20():
"""Sleep for 20 seconds"""
time.sleep(20)
@task
def sleep_30():
"""Sleep for 30 seconds"""
time.sleep(30)
with DAG(
dag_id='fail_S_L_A',
start_date=datetime(2021, 1, 1),
schedule="*/2 * * * *",
catchup=False,
sla_miss_callback=sla_callback,
) as dag:
sleep_20() >> sleep_30()
```
triggers all of the following....
```
# HELP airflow_ti_finish_fail_s_l_a_sleep_30_deferred
# TYPE airflow_ti_finish_fail_s_l_a_sleep_30_deferred counter
airflow_ti_finish_fail_s_l_a_sleep_30_deferred{dag_id="fail_S_L_A",job="Airflow",task_id="sleep_30"} 0
# HELP airflow_ti_finish_fail_s_l_a_sleep_30_failed
# TYPE airflow_ti_finish_fail_s_l_a_sleep_30_failed counter
airflow_ti_finish_fail_s_l_a_sleep_30_failed{dag_id="fail_S_L_A",job="Airflow",task_id="sleep_30"} 0
# HELP airflow_ti_finish_fail_s_l_a_sleep_30_none
# TYPE airflow_ti_finish_fail_s_l_a_sleep_30_none counter
airflow_ti_finish_fail_s_l_a_sleep_30_none{dag_id="fail_S_L_A",job="Airflow",task_id="sleep_30"} 0
# HELP airflow_ti_finish_fail_s_l_a_sleep_30_queued
# TYPE airflow_ti_finish_fail_s_l_a_sleep_30_queued counter
airflow_ti_finish_fail_s_l_a_sleep_30_queued{dag_id="fail_S_L_A",job="Airflow",task_id="sleep_30"} 0
# HELP airflow_ti_finish_fail_s_l_a_sleep_30_removed
# TYPE airflow_ti_finish_fail_s_l_a_sleep_30_removed counter
airflow_ti_finish_fail_s_l_a_sleep_30_removed{dag_id="fail_S_L_A",job="Airflow",task_id="sleep_30"} 0
# HELP airflow_ti_finish_fail_s_l_a_sleep_30_restarting
# TYPE airflow_ti_finish_fail_s_l_a_sleep_30_restarting counter
airflow_ti_finish_fail_s_l_a_sleep_30_restarting{dag_id="fail_S_L_A",job="Airflow",task_id="sleep_30"} 0
# HELP airflow_ti_finish_fail_s_l_a_sleep_30_running
# TYPE airflow_ti_finish_fail_s_l_a_sleep_30_running counter
airflow_ti_finish_fail_s_l_a_sleep_30_running{dag_id="fail_S_L_A",job="Airflow",task_id="sleep_30"} 0
# HELP airflow_ti_finish_fail_s_l_a_sleep_30_scheduled
# TYPE airflow_ti_finish_fail_s_l_a_sleep_30_scheduled counter
airflow_ti_finish_fail_s_l_a_sleep_30_scheduled{dag_id="fail_S_L_A",job="Airflow",task_id="sleep_30"} 0
# HELP airflow_ti_finish_fail_s_l_a_sleep_30_shutdown
# TYPE airflow_ti_finish_fail_s_l_a_sleep_30_shutdown counter
airflow_ti_finish_fail_s_l_a_sleep_30_shutdown{dag_id="fail_S_L_A",job="Airflow",task_id="sleep_30"} 0
# HELP airflow_ti_finish_fail_s_l_a_sleep_30_skipped
# TYPE airflow_ti_finish_fail_s_l_a_sleep_30_skipped counter
airflow_ti_finish_fail_s_l_a_sleep_30_skipped{dag_id="fail_S_L_A",job="Airflow",task_id="sleep_30"} 0
# HELP airflow_ti_finish_fail_s_l_a_sleep_30_success
# TYPE airflow_ti_finish_fail_s_l_a_sleep_30_success counter
airflow_ti_finish_fail_s_l_a_sleep_30_success{dag_id="fail_S_L_A",job="Airflow",task_id="sleep_30"} 0
# HELP airflow_ti_finish_fail_s_l_a_sleep_30_up_for_reschedule
# TYPE airflow_ti_finish_fail_s_l_a_sleep_30_up_for_reschedule counter
airflow_ti_finish_fail_s_l_a_sleep_30_up_for_reschedule{dag_id="fail_S_L_A",job="Airflow",task_id="sleep_30"} 0
# HELP airflow_ti_finish_fail_s_l_a_sleep_30_up_for_retry
# TYPE airflow_ti_finish_fail_s_l_a_sleep_30_up_for_retry counter
airflow_ti_finish_fail_s_l_a_sleep_30_up_for_retry{dag_id="fail_S_L_A",job="Airflow",task_id="sleep_30"} 0
# HELP airflow_ti_finish_fail_s_l_a_sleep_30_upstream_failed
# TYPE airflow_ti_finish_fail_s_l_a_sleep_30_upstream_failed counter
airflow_ti_finish_fail_s_l_a_sleep_30_upstream_failed{dag_id="fail_S_L_A",job="Airflow",task_id="sleep_30"} 0
```
Of note: it hit every stage and reported on it, and of note, it's only reporting for that one particular method (sleep_30) so perhaps that's my misunderstanding of when/why it gets triggered.
### Things I have tried and (possibly?) ruled out
- The one which triggers the metric has a `schedule` but scheduling the other two DAGs does not get the metric emitted so it is unlikely to be a scheduled/manual issue.
- Even if thta is the only DAG ever run in a fresh environment it gets the same result so it does not appear to be a name/key collision in the MetricsMap storage object (ie task1 overwriting task2 or something like that).
### Operating System
ubuntu
### Versions of Apache Airflow Providers
_No response_
### Deployment
Docker-Compose
### Deployment details
_No response_
### Anything else
_No response_
### Are you willing to submit PR?
- [X] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Research direction
Start by reproducing the reported behavior with Breeze using the StatsD and OTel integrations, then run the three example DAGs and compare the raw metrics at the listed OTel and StatsD endpoints. Trace how airflow_ti.start.* and airflow_ti.finish.* are emitted for task instances, especially the scheduled fail_S_L_A example. Done means the two integrations report consistent metrics for the relevant task states.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- observability
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100