This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 87577b644b8 [v3-4-test] Add dag_id and task_id tags to resumable_job
metrics (#75721) (#75732)
87577b644b8 is described below
commit 87577b644b8ea94c7a414105ef26677fc94af60e
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Oct 27 10:08:19 2026 +0000
[v3-4-test] Add dag_id and task_id tags to resumable_job metrics (#75721)
(#75732)
(cherry picked from commit 5243901c8186423b0add017bf3c9e000da0f77e7)
Co-authored-by: Amogh Desai <[email protected]>
---
.../observability/metrics/metrics_template.yaml | 12 +++---
.../src/airflow/sdk/bases/resumablejobmixin.py | 7 +---
.../tests/task_sdk/bases/test_resumablejobmixin.py | 47 +++++++++++++++++-----
3 files changed, 45 insertions(+), 21 deletions(-)
diff --git
a/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
b/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
index b432f40773f..1b9bf737377 100644
---
a/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
+++
b/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
@@ -53,35 +53,37 @@ metrics:
- name: "resumable_job.fresh_submit"
description: "Number of times a ResumableJobMixin operator submitted a
fresh job
- with no prior run ID stored (first run). Metric with operator tagging."
+ with no prior run ID stored (first run). Metric with operator, dag_id and
task_id tagging."
type: "counter"
legacy_name: "-"
name_variables: []
- name: "resumable_job.already_succeeded"
description: "Number of times a ResumableJobMixin operator found a stored
run ID
- whose job had already completed successfully and skipped resubmission.
Metric with operator tagging."
+ whose job had already completed successfully and skipped resubmission.
Metric with operator,
+ dag_id and task_id tagging."
type: "counter"
legacy_name: "-"
name_variables: []
- name: "resumable_job.terminal_resubmit"
description: "Number of times a ResumableJobMixin operator found a stored
run ID
- whose job was in a terminal (failed) state and submitted a fresh job.
Metric with operator tagging."
+ whose job was in a terminal (failed) state and submitted a fresh job.
Metric with operator,
+ dag_id and task_id tagging."
type: "counter"
legacy_name: "-"
name_variables: []
- name: "resumable_job.reconnect_attempt"
description: "Number of times a ResumableJobMixin operator found a stored
run ID
- on retry and attempted to reconnect. Metric with operator tagging."
+ on retry and attempted to reconnect. Metric with operator, dag_id and
task_id tagging."
type: "counter"
legacy_name: "-"
name_variables: []
- name: "resumable_job.reconnect_success"
description: "Number of times a ResumableJobMixin operator successfully
reconnected
- to an active job on retry. Metric with operator tagging."
+ to an active job on retry. Metric with operator, dag_id and task_id
tagging."
type: "counter"
legacy_name: "-"
name_variables: []
diff --git a/task-sdk/src/airflow/sdk/bases/resumablejobmixin.py
b/task-sdk/src/airflow/sdk/bases/resumablejobmixin.py
index ef1629da975..347a700a349 100644
--- a/task-sdk/src/airflow/sdk/bases/resumablejobmixin.py
+++ b/task-sdk/src/airflow/sdk/bases/resumablejobmixin.py
@@ -122,13 +122,8 @@ class ResumableJobMixin(ABC):
self.poll_until_complete(external_id, context)
return self.get_job_result(external_id, context)
- stats_tags = {"operator": type(self).__name__}
- # The task is team-scoped in multi-team deployments; surface team_name
on the
- # resumable_job metrics via the running task instance's stats tags
(omitted when
- # not multi-team or the task has no team).
ti = context.get("ti")
- if ti is not None and (team_name := ti.stats_tags.get("team_name")):
- stats_tags["team_name"] = team_name
+ stats_tags = {**(ti.stats_tags if ti is not None else {}), "operator":
type(self).__name__}
reconnect_to: Any = None
already_succeeded_id: Any = None
diff --git a/task-sdk/tests/task_sdk/bases/test_resumablejobmixin.py
b/task-sdk/tests/task_sdk/bases/test_resumablejobmixin.py
index 1f84baa2d89..4d9487cb3f8 100644
--- a/task-sdk/tests/task_sdk/bases/test_resumablejobmixin.py
+++ b/task-sdk/tests/task_sdk/bases/test_resumablejobmixin.py
@@ -86,14 +86,14 @@ def make_context(task_store: FakeTaskState | None = None)
-> dict:
class FakeTI:
- """Minimal stand-in for RuntimeTaskInstance exposing stats_tags with an
optional team_name."""
+ """Minimal stand-in for RuntimeTaskInstance exposing stats_tags."""
def __init__(self, team_name: str | None = None):
self._team_name = team_name
@property
def stats_tags(self) -> dict[str, str]:
- tags = {"dag_id": "d", "task_id": "t"}
+ tags = {"dag_id": "d", "task_id": "t", "run_type": "scheduled"}
if self._team_name:
tags["team_name"] = self._team_name
return tags
@@ -251,6 +251,8 @@ class TestExternalIdKey:
class TestMetrics:
_PATCH = "airflow.sdk._shared.observability.metrics.stats.incr"
_TAG = {"operator": "ConcreteResumableOperator"}
+ _TI_TAGS = {"dag_id": "d", "task_id": "t", "run_type": "scheduled"}
+ _FULL_TAGS = {**_TI_TAGS, "team_name": "team_alpha", **_TAG}
def test_fresh_submit_fires_only_fresh_submit_counter(self):
op = ConcreteResumableOperator(task_id="test_task")
@@ -297,24 +299,49 @@ class TestMetrics:
assert "resumable_job.fresh_submit" not in called_names
@pytest.mark.parametrize(
- ("team_name", "expected_tag"),
+ ("stored_status", "expected_names"),
[
+ pytest.param(None, ["resumable_job.fresh_submit"],
id="fresh_submit"),
pytest.param(
- "team_alpha",
- {"operator": "ConcreteResumableOperator", "team_name":
"team_alpha"},
- id="with_team",
+ "RUNNING",
+ ["resumable_job.reconnect_attempt",
"resumable_job.reconnect_success"],
+ id="reconnect",
+ ),
+ pytest.param(
+ "SUCCEEDED",
+ ["resumable_job.reconnect_attempt",
"resumable_job.already_succeeded"],
+ id="already_succeeded",
+ ),
+ pytest.param(
+ "FAILED",
+ ["resumable_job.reconnect_attempt",
"resumable_job.terminal_resubmit"],
+ id="terminal_resubmit",
),
- pytest.param(None, {"operator": "ConcreteResumableOperator"},
id="without_team"),
],
)
- def test_team_name_added_to_metric_tags(self, team_name, expected_tag):
+ def test_every_counter_carries_task_instance_stats_tags(self,
stored_status, expected_names):
+ op = ConcreteResumableOperator(task_id="test_task")
+ if stored_status is None:
+ task_state = FakeTaskState()
+ else:
+ op._status_map["job-001"] = stored_status
+ task_state = FakeTaskState({"test_job_id": "job-001"})
+ ctx = make_context(task_state)
+ ctx["ti"] = FakeTI("team_alpha")
+ mock_incr = MagicMock()
+ with patch(self._PATCH, mock_incr):
+ op.execute_resumable(ctx)
+ assert [call.args[0] for call in mock_incr.call_args_list] ==
expected_names
+ assert all(call.kwargs["tags"] == self._FULL_TAGS for call in
mock_incr.call_args_list)
+
+ def test_team_name_omitted_when_task_has_no_team(self):
op = ConcreteResumableOperator(task_id="test_task")
ctx = make_context(FakeTaskState())
- ctx["ti"] = FakeTI(team_name)
+ ctx["ti"] = FakeTI()
mock_incr = MagicMock()
with patch(self._PATCH, mock_incr):
op.execute_resumable(ctx)
- mock_incr.assert_called_once_with("resumable_job.fresh_submit",
tags=expected_tag)
+ mock_incr.assert_called_once_with("resumable_job.fresh_submit",
tags={**self._TI_TAGS, **self._TAG})
class TestTracing:
