apache / apache/airflow

Metrics - `airflow_ti.start.*` and `airflow_ti.finish.*`

Open
#32,159 4 comments 1 reaction 0 assignees View on GitHub
_eol_affected_version:2.6 area:core area:metrics kind:bug needs-triage
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.