This is an automated email from the ASF dual-hosted git repository.
hussein-awala pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new c96f5c89453 Add scheduler observability metrics (#68068)
c96f5c89453 is described below
commit c96f5c894537d6d1792db65b385eb0eede816872
Author: safaehar <[email protected]>
AuthorDate: Tue Jul 7 14:34:45 2026 +0200
Add scheduler observability metrics (#68068)
* feat: add scheduler observability metrics
Emit three new metrics from SchedulerJobRunner to improve visibility
into scheduler health:
- scheduler.loop_exceptions{exception_class}: counter incremented when
the scheduler loop exits with an unhandled exception, tagged with the
exception class to aid triaging crash loops.
- scheduler.executor_events.batch_size (gauge) and
scheduler.executor_events.processed (counter): emitted on each
successful call to process_executor_events with the event count.
scheduler.executor_events.failed{reason} is incremented instead when
the call raises, tagged with the exception class.
The original body is extracted into _process_executor_events_core so
process_executor_events can act as a thin metrics wrapper.
- scheduler.zombies.detected{reason}: counter incremented when zombie
task instances are detected, tagged with the detection path:
- heartbeat_timeout: task exceeded the heartbeat threshold
- adopt_failure: orphaned task could not be re-adopted and was reset
These metrics are already running in production via local monkey-patches
at Datadog; this commit contributes them natively upstream so the
patches can eventually be removed.
* Fix test assertions broken by new scheduler.executor_events.processed
metric
Three existing tests used mock_stats.incr.assert_not_called() to verify that
no error/mismatch metrics were emitted in requeued-TI and stale-success
scenarios. The new executor_events.processed counter now fires
unconditionally
for every _process_executor_events call, so those blanket assertions became
too broad. Replace each with an assertion that permits the expected counter
while still verifying that no scheduler.tasks.killed_externally metric
fired.
* Fix static checks: register new metrics in registry and apply ruff format
* Address review: preserve event_buffer return contract, merge to_reset
blocks
* Address review: standardize to exception_class tag, remove adopt_failure
zombie metric
* Update description for failed executor events metric
* Remove unnecessary blank line in test_scheduler_job.py
* Address review: dedupe executor_events batch metric emission
Extract _emit_executor_events_batch_metrics helper to avoid the
batch_size/processed emit drifting across the three return points
in process_executor_events.
* Fix stale-success queued test assertion for new processed-events metric
Rebasing onto main pulled in #68741's queued-after-defer regression test,
which asserted stats.incr was never called. That predates this PR's
executor_events.processed counter, which now fires unconditionally on
every successful process_executor_events call. Align the assertion with
the sibling scheduled-after-defer test fixed earlier in this PR.
---
.../src/airflow/jobs/scheduler_job_runner.py | 34 +++--
airflow-core/tests/unit/jobs/test_scheduler_job.py | 145 ++++++++++++++++++++-
.../observability/metrics/metrics_template.yaml | 33 +++++
3 files changed, 200 insertions(+), 12 deletions(-)
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index 1718987cff6..f4e66fc17c4 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -1254,12 +1254,21 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
return conf.getboolean("traces", "otel_on")
def _process_executor_events(self, executor: BaseExecutor, session:
Session) -> int:
- return SchedulerJobRunner.process_executor_events(
- executor=executor,
- job_id=self.job.id,
- scheduler_dag_bag=self.scheduler_dag_bag,
- session=session,
- )
+ try:
+ return SchedulerJobRunner.process_executor_events(
+ executor=executor,
+ job_id=self.job.id,
+ scheduler_dag_bag=self.scheduler_dag_bag,
+ session=session,
+ )
+ except Exception as exc:
+ stats.incr("scheduler.executor_events.failed",
tags={"exception_class": type(exc).__name__})
+ raise
+
+ @staticmethod
+ def _emit_executor_events_batch_metrics(num_events: int) -> None:
+ stats.gauge("scheduler.executor_events.batch_size", num_events)
+ stats.incr("scheduler.executor_events.processed", num_events)
@classmethod
def process_executor_events(
@@ -1297,6 +1306,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
"""
ti_primary_key_to_try_number_map: dict[tuple[str, str, str, int], int]
= {}
event_buffer = executor.get_event_buffer()
+ num_events = len(event_buffer)
tis_with_right_state: list[TaskInstanceKey] = []
callback_keys_with_events: list[CallbackKey] = []
@@ -1358,11 +1368,13 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
# Return if no finished tasks
if not tis_with_right_state:
+ cls._emit_executor_events_batch_metrics(num_events)
return len(event_buffer)
# Check state of finished tasks
filter_for_tis = TI.filter_for_tis(tis_with_right_state)
if filter_for_tis is None:
+ cls._emit_executor_events_batch_metrics(num_events)
return len(event_buffer)
asset_loader, alias_loader = _eager_load_dag_run_for_validation()
query = (
@@ -1586,6 +1598,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
# Update task state - emails are handled by DAG processor now
ti.handle_failure(error=msg, session=session)
+ cls._emit_executor_events_batch_metrics(num_events)
return len(event_buffer)
def _execute(self) -> int | None:
@@ -1621,7 +1634,8 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
if settings.Session is not None:
settings.Session.remove()
- except Exception:
+ except Exception as exc:
+ stats.incr("scheduler.loop_exceptions", tags={"exception_class":
type(exc).__name__})
self.log.exception("Exception when executing
SchedulerJob._run_scheduler_loop")
raise
finally:
@@ -3347,7 +3361,6 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
stats.incr("scheduler.orphaned_tasks.cleared",
len(to_reset))
stats.incr("scheduler.orphaned_tasks.adopted",
len(tis_to_adopt_or_reset) - len(to_reset))
-
if to_reset:
task_instance_str = "\n\t".join(reset_tis_message)
self.log.info(
@@ -3495,6 +3508,11 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
if task_instances_without_heartbeats :=
self._find_task_instances_without_heartbeats(
session=session
):
+ stats.incr(
+ "scheduler.zombies.detected",
+ len(task_instances_without_heartbeats),
+ tags={"reason": "heartbeat_timeout"},
+ )
self._purge_task_instances_without_heartbeats(
task_instances_without_heartbeats, session=session
)
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index 603fcf4bf66..ef5ea043060 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -863,7 +863,9 @@ class TestSchedulerJob:
ti1.refresh_from_db(session=session)
assert ti1.state == State.QUEUED
self.job_runner.executor.callback_sink.send.assert_not_called()
- mock_stats.incr.assert_not_called()
+ # Only the processed-events counter should have fired across all three
sub-tests;
+ # no killed_externally mismatch metric should appear.
+ assert all(c.args[0] == "scheduler.executor_events.processed" for c in
mock_stats.incr.call_args_list)
@mock.patch("airflow.jobs.scheduler_job_runner.TaskCallbackRequest")
@mock.patch("airflow._shared.observability.metrics.stats._get_backend")
@@ -908,7 +910,9 @@ class TestSchedulerJob:
ti1.refresh_from_db(session=session)
assert ti1.state == State.SCHEDULED
self.job_runner.executor.callback_sink.send.assert_not_called()
- mock_stats.incr.assert_not_called()
+ # Stale success from defer exit must not trigger a mismatch metric —
+ # only the standard processed-events counter should fire.
+
mock_stats.incr.assert_called_once_with("scheduler.executor_events.processed",
count=1)
# Without next_method, scheduled + stale success is still a mismatch
(e.g. external kill).
ti1.next_method = None
@@ -968,7 +972,9 @@ class TestSchedulerJob:
ti1.refresh_from_db(session=session)
assert ti1.state == State.QUEUED
self.job_runner.executor.callback_sink.send.assert_not_called()
- mock_stats.incr.assert_not_called()
+ # Stale success from defer exit must not trigger a mismatch metric —
+ # only the standard processed-events counter should fire.
+
mock_stats.incr.assert_called_once_with("scheduler.executor_events.processed",
count=1)
# Without next_method, queued + stale success is still a mismatch
(e.g. external kill).
ti1.next_method = None
@@ -1020,7 +1026,9 @@ class TestSchedulerJob:
for rec in caplog.records
)
mock_task_callback.assert_not_called()
- mock_stats.incr.assert_not_called()
+ # Only the processed-events counter should fire; duplicate try_number
events
+ # must not trigger any error/mismatch metrics.
+
mock_stats.incr.assert_called_once_with("scheduler.executor_events.processed",
count=2)
@pytest.mark.usefixtures("testing_dag_bundle")
def test_process_executor_events_with_asset_events(self, session,
dag_maker):
@@ -12773,3 +12781,132 @@ def test_resolve_partition_date(mappers,
partition_key, carried_partition_date,
carried_partition_date=carried_partition_date,
)
assert result == expected
+
+
+class TestSchedulerObservabilityMetrics:
+ """Tests for the scheduler observability metrics emitted in
scheduler_job_runner.py."""
+
+ @pytest.fixture(autouse=True)
+ def per_test(self) -> Generator:
+ _clean_db()
+ self.job_runner: SchedulerJobRunner | None = None
+ yield
+ _clean_db()
+
+ # --- scheduler.loop_exceptions ---
+
+ def test_loop_exceptions_incr_on_scheduler_loop_failure(self):
+ """scheduler.loop_exceptions is emitted with exception_class tag when
_run_scheduler_loop raises."""
+ scheduler_job = Job()
+ self.job_runner = SchedulerJobRunner(job=scheduler_job)
+
+ with (
+ mock.patch("airflow.jobs.scheduler_job_runner.stats") as
mock_stats,
+ mock.patch.object(self.job_runner, "register_signals",
return_value=MagicMock()),
+ mock.patch.object(self.job_runner.executor, "start"),
+ mock.patch.object(self.job_runner.executor, "end"),
+ mock.patch.object(self.job_runner, "_run_scheduler_loop",
side_effect=RuntimeError("loop crash")),
+ pytest.raises(RuntimeError, match="loop crash"),
+ ):
+ self.job_runner._execute()
+
+ mock_stats.incr.assert_any_call("scheduler.loop_exceptions",
tags={"exception_class": "RuntimeError"})
+
+ def test_loop_exceptions_not_emitted_on_clean_exit(self):
+ """scheduler.loop_exceptions is NOT emitted when _run_scheduler_loop
returns normally."""
+ scheduler_job = Job()
+ self.job_runner = SchedulerJobRunner(job=scheduler_job)
+
+ with (
+ mock.patch("airflow.jobs.scheduler_job_runner.stats") as
mock_stats,
+ mock.patch.object(self.job_runner, "register_signals",
return_value=MagicMock()),
+ mock.patch.object(self.job_runner.executor, "start"),
+ mock.patch.object(self.job_runner.executor, "end"),
+ mock.patch.object(self.job_runner, "_run_scheduler_loop"),
+ ):
+ self.job_runner._execute()
+
+ emitted_names = [c.args[0] for c in mock_stats.incr.call_args_list]
+ assert "scheduler.loop_exceptions" not in emitted_names
+
+ # --- scheduler.executor_events.{batch_size,processed,failed} ---
+
+ def test_executor_events_batch_metrics_emitted_on_success(self):
+ """batch_size gauge and processed counter are emitted via the
early-return path."""
+ # Empty event buffer → tis_with_right_state is empty → early return
with num_events=0
+ mock_executor = MagicMock()
+ mock_executor.get_event_buffer.return_value = {}
+
+ with mock.patch("airflow.jobs.scheduler_job_runner.stats") as
mock_stats:
+ result = SchedulerJobRunner.process_executor_events(
+ executor=mock_executor, job_id=1,
scheduler_dag_bag=MagicMock(), session=MagicMock()
+ )
+
+ assert result == 0
+
mock_stats.gauge.assert_called_once_with("scheduler.executor_events.batch_size",
0)
+
mock_stats.incr.assert_called_once_with("scheduler.executor_events.processed",
0)
+
+ def test_executor_events_failed_metric_emitted_on_exception(self):
+ """failed counter is emitted in _process_executor_events when
process_executor_events raises."""
+ scheduler_job = Job()
+ self.job_runner = SchedulerJobRunner(job=scheduler_job)
+
+ with (
+ mock.patch("airflow.jobs.scheduler_job_runner.stats") as
mock_stats,
+ mock.patch.object(
+ SchedulerJobRunner, "process_executor_events",
side_effect=ValueError("executor boom")
+ ),
+ pytest.raises(ValueError, match="executor boom"),
+ ):
+ self.job_runner._process_executor_events(executor=MagicMock(),
session=MagicMock())
+
+ mock_stats.incr.assert_called_once_with(
+ "scheduler.executor_events.failed", tags={"exception_class":
"ValueError"}
+ )
+ mock_stats.gauge.assert_not_called()
+
+ # --- scheduler.zombies.detected ---
+
+ def test_zombies_detected_heartbeat_timeout_emitted(self):
+ """scheduler.zombies.detected{reason:heartbeat_timeout} is emitted
when zombies are found."""
+ scheduler_job = Job()
+ self.job_runner = SchedulerJobRunner(job=scheduler_job)
+ fake_tis = [MagicMock(), MagicMock(), MagicMock()]
+
+ mock_session = MagicMock()
+ mock_ctx = MagicMock()
+ mock_ctx.__enter__ = MagicMock(return_value=mock_session)
+ mock_ctx.__exit__ = MagicMock(return_value=False)
+
+ with (
+ mock.patch("airflow.jobs.scheduler_job_runner.stats") as
mock_stats,
+ mock.patch("airflow.jobs.scheduler_job_runner.create_session",
return_value=mock_ctx),
+ mock.patch.object(
+ self.job_runner, "_find_task_instances_without_heartbeats",
return_value=fake_tis
+ ),
+ mock.patch.object(self.job_runner,
"_purge_task_instances_without_heartbeats"),
+ ):
+ self.job_runner._find_and_purge_task_instances_without_heartbeats()
+
+ mock_stats.incr.assert_called_once_with(
+ "scheduler.zombies.detected", 3, tags={"reason":
"heartbeat_timeout"}
+ )
+
+ def test_zombies_detected_not_emitted_when_no_heartbeat_timeout(self):
+ """scheduler.zombies.detected is NOT emitted when no zombie task
instances are found."""
+ scheduler_job = Job()
+ self.job_runner = SchedulerJobRunner(job=scheduler_job)
+
+ mock_session = MagicMock()
+ mock_ctx = MagicMock()
+ mock_ctx.__enter__ = MagicMock(return_value=mock_session)
+ mock_ctx.__exit__ = MagicMock(return_value=False)
+
+ with (
+ mock.patch("airflow.jobs.scheduler_job_runner.stats") as
mock_stats,
+ mock.patch("airflow.jobs.scheduler_job_runner.create_session",
return_value=mock_ctx),
+ mock.patch.object(self.job_runner,
"_find_task_instances_without_heartbeats", return_value=[]),
+ ):
+ self.job_runner._find_and_purge_task_instances_without_heartbeats()
+
+ mock_stats.incr.assert_not_called()
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 6d815cb95d9..1855f737b30 100644
---
a/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
+++
b/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
@@ -186,6 +186,33 @@ metrics:
legacy_name: "-"
name_variables: []
+ - name: "scheduler.loop_exceptions"
+ description: "Number of times the scheduler loop exited with an unhandled
exception.
+ Metric with exception_class tagging."
+ type: "counter"
+ legacy_name: "-"
+ name_variables: []
+
+ - name: "scheduler.executor_events.processed"
+ description: "Number of executor events processed per
``process_executor_events`` call."
+ type: "counter"
+ legacy_name: "-"
+ name_variables: []
+
+ - name: "scheduler.executor_events.failed"
+ description: "Number of times ``process_executor_events`` raised an
exception.
+ Metric with exception_class tagging."
+ type: "counter"
+ legacy_name: "-"
+ name_variables: []
+
+ - name: "scheduler.zombies.detected"
+ description: "Number of zombie task instances detected by the Scheduler.
+ Metric with reason tagging (``heartbeat_timeout``)."
+ type: "counter"
+ legacy_name: "-"
+ name_variables: []
+
- name: "scheduler.critical_section_busy"
description: "Count of times a scheduler process tried to get a lock on
the critical
section (needed to send tasks to the executor) and found it locked by
another process."
@@ -418,6 +445,12 @@ metrics:
legacy_name: "-"
name_variables: []
+ - name: "scheduler.executor_events.batch_size"
+ description: "Number of executor events in the batch processed per
``process_executor_events`` call."
+ type: "gauge"
+ legacy_name: "-"
+ name_variables: []
+
- name: "scheduler.tasks.executable"
description: "Number of tasks that are ready for execution (set to queued)
with respect to pool limits,
Dag concurrency, executor state, and priority."
