Skip to content

Cleanup the DualStatsManager and fix any leftover Stats calls #62705

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
xBis7 wants to merge 6 commits into from
Closed

Cleanup the DualStatsManager and fix any leftover Stats calls #62705

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
5829c24
rename extra_tags to legacy_name_tags
xBis7 Feb 26, 2026
499ca53
use the dual stats manager with leftover stats + cleanup
xBis7 Feb 27, 2026
445333f
fix failing test_jobs
xBis7 Mar 2, 2026
55db426
fix backwards compat issues with test_jobs
xBis7 Mar 2, 2026
9d00dc5
fix test_edge_executor
xBis7 Mar 2, 2026
ac037b1
replace try-except import pattern with AIRFLOW_V_3_2_PLUS checks
xBis7 Mar 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions airflow-core/src/airflow/dag_processing/manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@
from tabulate import tabulate
from uuid6 import uuid7

from airflow._shared.observability.metrics.dual_stats_manager import DualStatsManager
from airflow._shared.observability.metrics.stats import Stats, normalize_name_for_stats
from airflow._shared.timezones import timezone
from airflow.api_fastapi.execution_api.app import InProcessExecutionAPI
Expand Down Expand Up @@ -1288,11 +1289,11 @@ def process_parse_results(
file_name = normalize_name_for_stats(Path(relative_fileloc).stem)
# bundle_name is included to distinguish files with the same name across different bundles
normalized_bundle = normalize_name_for_stats(bundle_name)
Stats.timing(f"dag_processing.last_duration.{normalized_bundle}.{file_name}", stat.last_duration)
Stats.timing(
DualStatsManager.timing(
"dag_processing.last_duration",
stat.last_duration,
tags={"file_name": file_name, "bundle_name": normalized_bundle},
tags={},
legacy_name_tags={"bundle_name": normalized_bundle, "file_name": file_name},

@ashb ashb Mar 2, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My initial assumption here was that the tags would be the other way around -- why don't we want bundle name and filepath included in the modern tags?

@xBis7 xBis7 Mar 2, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

They are.

It goes like this

  • check the legacy_name_tags and cross-reference them with the variables in the YAML file
  • set the legacy_name_tags as variables in the legacy name
  • build the legacy metric name
  • export the legacy metric with the tags param
  • merge the tags and the legacy_name_tags
  • export the new name with the merged dictionary as tags

That was actually a bug because, by just setting the tags, there weren't any available variables for the legacy name.

I've added comments in the code to explain the above.

The legacy_name_tags are going to end up as regular tags for the modern name. And the tags are just for whatever tag is set on both metrics.

@ashb ashb Mar 2, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh. That's slightly confusing. The behabiour makes sense, but, just from the callsite (of which there are going to be lots) it's a bit confusing that legacy_name_tags gets applied to non-legacy metrics.

🤔

@xBis7 xBis7 Mar 2, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I know. The behavior is consistent with before but the DualStatsManager logic is confusing. I renamed the extra_tags param to legacy_name_tags to make it more intuitive and added comments but it's still not that clear.

Do you have any suggestions?

Before the DualStatsManager

  • legacy metrics didn't have any tags and everything was included in the name
  • modern metrics had a simpler name and all the variables from the legacy metric name, as tags

The thing is that I'm trying to cover an edge case where the legacy and the modern metric had some extra tags, apart from the name variables.

@jscheffl jscheffl Mar 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ashb Can you make a final review? Then I think it would be great to merge.

)

if parsing_result is None:
Expand Down
20 changes: 10 additions & 10 deletions airflow-core/src/airflow/jobs/scheduler_job_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -886,7 +886,7 @@ def _executable_task_instances_to_queued(self, max_tis: int, session: Session) -
"pool.starving_tasks",
num_starving_tasks,
tags={},
extra_tags={"pool_name": pool_name},
legacy_name_tags={"pool_name": pool_name},
)

Stats.gauge("scheduler.tasks.starving", num_starving_tasks_total)
Expand Down Expand Up @@ -2145,7 +2145,7 @@ def _update_state(dag: SerializedDAG, dag_run: DagRun):
"dagrun.schedule_delay",
schedule_delay,
tags={},
extra_tags={"dag_id": dag.dag_id},
legacy_name_tags={"dag_id": dag.dag_id},
)

# cache saves time during scheduling of many dag_runs for same dag
Expand Down Expand Up @@ -2289,7 +2289,7 @@ def _schedule_dag_run(
"dagrun.duration.failed",
duration,
tags={},
extra_tags={"dag_id": dag_run.dag_id},
legacy_name_tags={"dag_id": dag_run.dag_id},
)
return callback_to_execute

Expand Down Expand Up @@ -2584,7 +2584,7 @@ def _emit_ti_metrics(self, session: Session = NEW_SESSION) -> None:
f"ti.{state}",
float(count),
tags={},
extra_tags={"queue": queue, "dag_id": dag_id, "task_id": task_id},
legacy_name_tags={"queue": queue, "dag_id": dag_id, "task_id": task_id},
)

for prev_key in self.previous_ti_metrics[state]:
Expand All @@ -2595,7 +2595,7 @@ def _emit_ti_metrics(self, session: Session = NEW_SESSION) -> None:
f"ti.{state}",
0,
tags={},
extra_tags={"queue": queue, "dag_id": dag_id, "task_id": task_id},
legacy_name_tags={"queue": queue, "dag_id": dag_id, "task_id": task_id},
)

self.previous_ti_metrics[state] = ti_metrics
Expand All @@ -2617,31 +2617,31 @@ def _emit_pool_metrics(self, session: Session = NEW_SESSION) -> None:
"pool.open_slots",
slot_stats["open"],
tags={},
extra_tags={"pool_name": normalized_pool_name},
legacy_name_tags={"pool_name": normalized_pool_name},
)
DualStatsManager.gauge(
"pool.queued_slots",
slot_stats["queued"],
tags={},
extra_tags={"pool_name": normalized_pool_name},
legacy_name_tags={"pool_name": normalized_pool_name},
)
DualStatsManager.gauge(
"pool.running_slots",
slot_stats["running"],
tags={},
extra_tags={"pool_name": normalized_pool_name},
legacy_name_tags={"pool_name": normalized_pool_name},
)
DualStatsManager.gauge(
"pool.deferred_slots",
slot_stats["deferred"],
tags={},
extra_tags={"pool_name": normalized_pool_name},
legacy_name_tags={"pool_name": normalized_pool_name},
)
DualStatsManager.gauge(
"pool.scheduled_slots",
slot_stats["scheduled"],
tags={},
extra_tags={"pool_name": normalized_pool_name},
legacy_name_tags={"pool_name": normalized_pool_name},
)

@provide_session
Expand Down
4 changes: 2 additions & 2 deletions airflow-core/src/airflow/jobs/triggerer_job_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -615,15 +615,15 @@ def emit_metrics(self):
"triggers.running",
len(self.running_triggers),
tags={},
extra_tags={"hostname": self.job.hostname},
legacy_name_tags={"hostname": self.job.hostname},
)

capacity_left = self.capacity - len(self.running_triggers)
DualStatsManager.gauge(
"triggerer.capacity_left",
capacity_left,
tags={},
extra_tags={"hostname": self.job.hostname},
legacy_name_tags={"hostname": self.job.hostname},
)

def update_triggers(self, requested_trigger_ids: set[int]):
Expand Down
18 changes: 10 additions & 8 deletions airflow-core/src/airflow/models/dagrun.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1083,7 +1083,7 @@ def recalculate(self) -> _UnfinishedStates:
with DualStatsManager.timer(
"dagrun.dependency-check",
tags={},
extra_tags=self.stats_tags,
legacy_name_tags=self.stats_tags,
):
dag = self.get_dag()
info = self.task_instance_scheduling_decisions(session)
Expand Down Expand Up @@ -1562,10 +1562,12 @@ def _emit_true_scheduling_delay_stats_for_finished_state(self, finished_tis: lis
else:
true_delay = first_start_date - self.run_after
if true_delay.total_seconds() > 0:
Stats.timing(
f"dagrun.{dag.dag_id}.first_task_scheduling_delay", true_delay, tags=self.stats_tags
DualStatsManager.timing(
"dagrun.first_task_scheduling_delay",
true_delay,
tags=self.stats_tags,
legacy_name_tags={"dag_id": dag.dag_id},
)
Stats.timing("dagrun.first_task_scheduling_delay", true_delay, tags=self.stats_tags)
except Exception:
self.log.warning("Failed to record first_task_scheduling_delay metric:", exc_info=True)

Expand All @@ -1584,7 +1586,7 @@ def _emit_duration_stats_for_finished_state(self):
f"dagrun.duration.{self.state}",
dt=duration,
tags=self.stats_tags,
extra_tags={"dag_id": self.dag_id},
legacy_name_tags={"dag_id": self.dag_id},
)

@provide_session
Expand Down Expand Up @@ -1664,7 +1666,7 @@ def _check_for_removed_or_restored_tasks(
DualStatsManager.incr(
"task_restored_to_dag",
tags=self.stats_tags,
extra_tags={"dag_id": dag.dag_id},
legacy_name_tags={"dag_id": dag.dag_id},
)
ti.state = None
except AirflowException:
Expand All @@ -1675,7 +1677,7 @@ def _check_for_removed_or_restored_tasks(
DualStatsManager.incr(
"task_removed_from_dag",
tags=self.stats_tags,
extra_tags={"dag_id": dag.dag_id},
legacy_name_tags={"dag_id": dag.dag_id},
)
ti.state = TaskInstanceState.REMOVED
continue
Expand Down Expand Up @@ -1843,7 +1845,7 @@ def _create_task_instances(
"task_instance_created",
count,
tags=self.stats_tags,
extra_tags={"task_type": task_type},
legacy_name_tags={"task_type": task_type},
)
session.flush()
except IntegrityError:
Expand Down
4 changes: 2 additions & 2 deletions airflow-core/src/airflow/models/taskinstance.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1314,7 +1314,7 @@ def emit_state_change_metric(self, new_state: TaskInstanceState) -> None:
f"task.{metric_name}",
timing,
tags={},
extra_tags={"task_id": self.task_id, "dag_id": self.dag_id, "queue": self.queue},
legacy_name_tags={"task_id": self.task_id, "dag_id": self.dag_id, "queue": self.queue},
)

def clear_next_method_args(self) -> None:
Expand Down Expand Up @@ -1534,7 +1534,7 @@ def fetch_handle_failure_context(
DualStatsManager.incr(
"operator_failures",
tags=ti.stats_tags,
extra_tags={"operator_name": ti.operator},
legacy_name_tags={"operator_name": ti.operator},
)
Stats.incr("ti_failures", tags=ti.stats_tags)

Expand Down
60 changes: 31 additions & 29 deletions airflow-core/tests/unit/core/test_dual_stats_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,79 +119,81 @@ def test_get_dict_with_defined_args(
assert sorted(args_dict) == sorted(expected_args_dict)

@pytest.mark.parametrize(
("args_dict", "tags", "extra_tags", "expected_args_dict"),
("args_dict", "tags", "legacy_name_tags", "expected_args_dict"),
[
pytest.param(
{"count": 1},
{"test": True},
{},
{"count": 1, "tags": {"test": True}},
id="no_extra_tags",
id="no_legacy_name_tags",
),
pytest.param(
{},
{"test": True},
{},
{"tags": {"test": True}},
id="no_args_no_extra_but_tags",
id="no_args_no_legacy_but_tags",
),
pytest.param(
{},
{"test": True},
{"test_extra": True},
{"tags": {"test": True, "test_extra": True}},
id="no_args_but_tags_and_extra",
{"test_legacy": True},
{"tags": {"test": True, "test_legacy": True}},
id="no_args_but_tags_and_legacy",
),
pytest.param(
{"count": 1},
{"test": True},
{},
{"count": 1, "tags": {"test": True}},
id="no_args_no_tags_but_extra_tags",
id="no_args_no_tags_but_legacy_name_tags",
),
pytest.param(
{"count": 1},
{"test": True},
{"test_extra": True},
{"count": 1, "tags": {"test": True, "test_extra": True}},
{"test_legacy": True},
{"count": 1, "tags": {"test": True, "test_legacy": True}},
id="all_params_provided",
),
pytest.param(
{"count": 1, "rate": 3},
{"test1": True, "test2": False},
{"test_extra1": True, "test_extra2": False, "test_extra3": True},
{"test_legacy1": True, "test_legacy2": False, "test_legacy3": True},
{
"count": 1,
"rate": 3,
"tags": {
"test1": True,
"test2": False,
"test_extra1": True,
"test_extra2": False,
"test_extra3": True,
"test_legacy1": True,
"test_legacy2": False,
"test_legacy3": True,
},
},
id="multiple_params",
),
],
)
def test_get_args_dict_with_extra_tags_if_set(
def test_get_args_dict_with_legacy_name_tags_if_set(
self,
args_dict: dict[str, Any] | None,
tags: dict[str, Any] | None,
extra_tags: dict[str, Any] | None,
legacy_name_tags: dict[str, Any] | None,
expected_args_dict: dict[str, Any],
):
dict_full = dual_stats_manager._get_args_dict_with_extra_tags_if_set(args_dict, tags, extra_tags)
dict_full = dual_stats_manager._get_args_dict_with_legacy_name_tags_if_set(
args_dict, tags, legacy_name_tags
)
assert sorted(dict_full) == sorted(expected_args_dict)

@pytest.mark.parametrize(
("tags", "extra_tags", "expected_tags_dict"),
("tags", "legacy_name_tags", "expected_tags_dict"),
[
pytest.param(
{"test": True},
{"test_extra": True},
{"test": True, "test_extra": True},
{"test_legacy": True},
{"test": True, "test_legacy": True},
id="all_params_provided",
),
pytest.param(
Expand All @@ -208,31 +210,31 @@ def test_get_args_dict_with_extra_tags_if_set(
),
pytest.param(
{},
{"test_extra": True},
{"test_extra": True},
id="only_extra",
{"test_legacy": True},
{"test_legacy": True},
id="only_legacy",
),
pytest.param(
{"test1": True, "test2": False},
{"test_extra1": True, "test_extra2": False, "test_extra3": True},
{"test_legacy1": True, "test_legacy2": False, "test_legacy3": True},
{
"test1": True,
"test2": False,
"test_extra1": True,
"test_extra2": False,
"test_extra3": True,
"test_legacy1": True,
"test_legacy2": False,
"test_legacy3": True,
},
id="multiple_params",
),
],
)
def test_get_tags_with_extra(
def test_get_tags_with_legacy(
self,
tags: dict[str, Any] | None,
extra_tags: dict[str, Any] | None,
legacy_name_tags: dict[str, Any] | None,
expected_tags_dict: dict[str, Any],
):
tags_full = dual_stats_manager._get_tags_with_extra(tags, extra_tags)
tags_full = dual_stats_manager._get_tags_with_legacy(tags, legacy_name_tags)
assert sorted(tags_full) == sorted(expected_tags_dict)

@pytest.mark.parametrize(
Expand Down
16 changes: 11 additions & 5 deletions providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
from airflow.executors.base_executor import BaseExecutor
from airflow.models.taskinstance import TaskInstance
from airflow.providers.common.compat.sdk import Stats, timezone
from airflow.providers.common.compat.version_compat import AIRFLOW_V_3_2_PLUS
from airflow.providers.edge3.models.db import EdgeDBManager, check_db_manager_config
from airflow.providers.edge3.models.edge_job import EdgeJobModel
from airflow.providers.edge3.models.edge_logs import EdgeLogsModel
Expand Down Expand Up @@ -192,11 +193,16 @@ def _update_orphaned_jobs(self, session: Session) -> bool:
"queue": job.queue,
"state": str(TaskInstanceState.FAILED),
}
Stats.incr(
f"edge_worker.ti.finish.{job.queue}.{TaskInstanceState.FAILED}.{job.dag_id}.{job.task_id}",
tags=tags,
)
Stats.incr("edge_worker.ti.finish", tags=tags)
if AIRFLOW_V_3_2_PLUS:
from airflow.sdk.observability.stats import DualStatsManager

DualStatsManager.incr("edge_worker.ti.finish", tags={}, legacy_name_tags=tags)
else:
Stats.incr(
f"edge_worker.ti.finish.{job.queue}.{TaskInstanceState.FAILED}.{job.dag_id}.{job.task_id}",
tags=tags,
)
Stats.incr("edge_worker.ti.finish", tags=tags)

return bool(lifeless_jobs)

Expand Down
Loading
Loading