This is an automated email from the ASF dual-hosted git repository.
ephraimbuddy 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 67014e1056e Make TriggerRunnerSupervisor.job optional (#66006)
67014e1056e is described below
commit 67014e1056ec3da98d5550552dc96137c85c2b89
Author: Ephraim Anierobi <[email protected]>
AuthorDate: Thu Apr 30 10:27:01 2026 +0100
Make TriggerRunnerSupervisor.job optional (#66006)
* Make TriggerRunnerSupervisor.job optional
Allow subclasses targeting AIP-92 (Execution-API-backed triggerers) to
construct without a metadata-DB-backed Job. heartbeat() and load_triggers()
no-op when job is None; _create_workload() falls back to the supervisor's
UUID for the log filename; emit_metrics() routes through a new metric_tags()
hook that subclasses can override to supply hostname or other identity tags
from any source.
* Fail loudly when TriggerRunnerSupervisor methods are called without a Job
heartbeat(), load_triggers(), and metric_tags() now raise RuntimeError
when self.job is None instead of silently no-opping or returning empty
tags. The empty-tags path was unsafe — DualStatsManager would hit the
legacy-name branch on every tick and raise ValueError because hostname
is a required name variable for triggers.running and
triggerer.capacity_left. The silent no-ops in heartbeat/load_triggers
also turned a missing subclass override into a zombie supervisor that
appeared alive but did nothing.
Also rename a local that shadowed the builtin id, document the int-vs-UUID
log-suffix divergence at the _create_workload site, and update the class
docstring to spell out the override contract for AIP-92 subclasses.
---
.../src/airflow/jobs/triggerer_job_runner.py | 57 +++++++--
airflow-core/tests/unit/jobs/test_triggerer_job.py | 133 +++++++++++++++++++++
2 files changed, 183 insertions(+), 7 deletions(-)
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index 122adb4974b..cef5c4a259d 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -31,6 +31,7 @@ from datetime import datetime
from socket import socket
from traceback import format_exception
from typing import TYPE_CHECKING, Annotated, Any, BinaryIO, ClassVar, Literal,
TextIO, TypedDict
+from uuid import uuid4
import anyio
import attrs
@@ -395,9 +396,16 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
This class (which runs in the main/sync process) is responsible for
querying the DB, sending RunTrigger
workload messages to the subprocess, and collecting results and updating
them in the DB.
+
+ ``job`` is optional so AIP-92 Execution-API-backed subclasses can run
without a metadata-DB
+ ``Job``. Such subclasses **must** override the methods that read
``self.job`` —
+ :meth:`heartbeat`, :meth:`load_triggers`, and :meth:`metric_tags` — to
source their data
+ from the Execution API or another backend. The base implementations of
those methods raise
+ :class:`RuntimeError` when called with ``job=None`` so a missing override
fails immediately
+ rather than silently zombieing the supervisor.
"""
- job: Job
+ job: Job | None = None
capacity: int
queues: set[str] | None = None
@@ -435,11 +443,12 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
def start( # type: ignore[override]
cls,
*,
- job: Job,
+ job: Job | None = None,
logger=None,
**kwargs,
):
- proc = super().start(id=job.id, job=job, target=cls.run_in_process,
logger=logger, **kwargs)
+ proc_id = job.id if job is not None else uuid4()
+ proc = super().start(id=proc_id, job=job, target=cls.run_in_process,
logger=logger, **kwargs)
msg = messages.StartTriggerer()
proc.send_msg(msg, request_id=0)
@@ -623,6 +632,11 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
self.emit_metrics()
def heartbeat(self):
+ if self.job is None:
+ raise RuntimeError(
+ "TriggerRunnerSupervisor.heartbeat() requires a Job; "
+ "subclasses without a metadata-DB Job must override this
method."
+ )
perform_heartbeat(self.job,
heartbeat_callback=self.heartbeat_callback, only_if_necessary=True)
def heartbeat_callback(self, session: Session | None = None) -> None:
@@ -630,6 +644,11 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
def load_triggers(self) -> None:
"""Assign triggers to this triggerer and update the runner with the
IDs it should run."""
+ if self.job is None:
+ raise RuntimeError(
+ "TriggerRunnerSupervisor.load_triggers() requires a Job; "
+ "subclasses without a metadata-DB Job must override this
method."
+ )
Trigger.assign_unassigned(
self.job.id,
self.capacity,
@@ -674,12 +693,31 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
"""Record that a trigger failed."""
Trigger.submit_failure(trigger_id=trigger_id, exc=exc)
+ def metric_tags(self) -> dict[str, str]:
+ """
+ Return extra tags applied to gauges emitted by :meth:`emit_metrics`.
+
+ Subclasses that don't have a metadata-DB ``Job`` (e.g. AIP-92
Execution-API-backed
+ triggerers) **must** override this to supply ``hostname`` (and any
other identity
+ tags) from another source. ``triggers.running`` and
``triggerer.capacity_left``
+ declare ``hostname`` as a required name variable in
``metrics_template.yaml``, so
+ an empty dict here would crash :meth:`emit_metrics` on every tick.
+ """
+ hostname = self.job.hostname if self.job is not None else None
+ if hostname is None:
+ raise RuntimeError(
+ "TriggerRunnerSupervisor.metric_tags() requires a Job with a
hostname; "
+ "subclasses without a metadata-DB Job must override this
method."
+ )
+ return {"hostname": hostname}
+
def emit_metrics(self):
+ extra_tags = self.metric_tags()
DualStatsManager.gauge(
"triggers.running",
len(self.running_triggers),
tags={},
- extra_tags={"hostname": self.job.hostname},
+ extra_tags=extra_tags,
)
capacity_left = self.capacity - len(self.running_triggers)
@@ -687,7 +725,7 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
"triggerer.capacity_left",
capacity_left,
tags={},
- extra_tags={"hostname": self.job.hostname},
+ extra_tags=extra_tags,
)
def _create_workload(
@@ -715,9 +753,14 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
log_path = render_log_fname(ti=trigger.task_instance)
ser_ti = TaskInstanceDTO.model_validate(trigger.task_instance,
from_attributes=True)
- # When producing logs from TIs, include the job id producing the logs
to disambiguate it.
+ # When producing logs from TIs, include the supervisor id producing
the logs to disambiguate it.
+ # Note: with a Job this is an int (Job.id); without a Job it's a UUID.
The reader side
+ # (FileTaskHandler.add_triggerer_suffix) currently formats the suffix
from
+ # ``ti.triggerer_job.id``, which only resolves the int form — AIP-92
subclasses without a
+ # metadata Job need their own log retrieval path.
+ log_id = self.job.id if self.job is not None else self.id
self.logger_cache[trigger.id] = TriggerLoggingFactory(
- log_path=f"{log_path}.trigger.{self.job.id}.log",
+ log_path=f"{log_path}.trigger.{log_id}.log",
ti=ser_ti, # type: ignore
)
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py
b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index 7f553b4b81e..a2db553c6fa 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -304,6 +304,139 @@ def
test_run_context_exits_when_subprocess_dies(supervisor_builder, mocker):
assert events == ["enter", "tick", "exit"]
[email protected]
+def jobless_supervisor(mocker):
+ """Build a TriggerRunnerSupervisor with ``job=None`` for testing the
optional-job path."""
+ import psutil
+
+ process = mocker.Mock(spec=psutil.Process, pid=42)
+ mock_stdin = mocker.Mock(spec=socket)
+ mock_stdin.write = mocker.Mock()
+ mock_stdin.sendall = mocker.Mock()
+
+ supervisor = TriggerRunnerSupervisor(
+ process_log=mocker.Mock(spec=FilteringBoundLogger),
+ id=uuid.uuid4(),
+ job=None,
+ pid=process.pid,
+ stdin=mock_stdin,
+ process=process,
+ capacity=10,
+ )
+ mock_selector = mocker.Mock(spec=selectors.DefaultSelector)
+ mock_selector.select.return_value = []
+ supervisor.selector = mock_selector
+ return supervisor
+
+
+def test_start_without_job_generates_uuid_id(mocker):
+ """start() called without a Job should generate a UUID for the supervisor
id."""
+ from airflow.sdk.execution_time.supervisor import WatchedSubprocess
+
+ fake_proc = mocker.Mock()
+ captured: dict = {}
+
+ @classmethod
+ def fake_super_start(cls, **kwargs):
+ captured.update(kwargs)
+ return fake_proc
+
+ mocker.patch.object(WatchedSubprocess, "start", fake_super_start)
+
+ TriggerRunnerSupervisor.start(capacity=10)
+
+ assert isinstance(captured["id"], uuid.UUID)
+ assert captured["job"] is None
+ fake_proc.send_msg.assert_called_once()
+
+
+def test_heartbeat_raises_without_job(jobless_supervisor, mocker):
+ """heartbeat() must fail loudly when job is None so missing subclass
overrides surface."""
+ perform_heartbeat =
mocker.patch("airflow.jobs.triggerer_job_runner.perform_heartbeat")
+
+ with pytest.raises(RuntimeError, match="must override this method"):
+ jobless_supervisor.heartbeat()
+
+ perform_heartbeat.assert_not_called()
+
+
+def test_metric_tags_default_uses_job_hostname(supervisor_builder):
+ """metric_tags() defaults to {"hostname": job.hostname} when a job is
present."""
+ supervisor = supervisor_builder()
+
+ assert supervisor.metric_tags() == {"hostname": supervisor.job.hostname}
+
+
+def test_metric_tags_raises_without_job(jobless_supervisor):
+ """metric_tags() must fail loudly when job is None — empty tags would
crash emit_metrics()."""
+ with pytest.raises(RuntimeError, match="must override this method"):
+ jobless_supervisor.metric_tags()
+
+
+def test_emit_metrics_uses_metric_tags_override(jobless_supervisor, mocker):
+ """Subclasses can supply tags by overriding metric_tags() instead of
threading args."""
+ gauge =
mocker.patch("airflow.jobs.triggerer_job_runner.DualStatsManager.gauge")
+ mocker.patch.object(
+ TriggerRunnerSupervisor,
+ "metric_tags",
+ return_value={"hostname": "astro-host", "deployment": "demo"},
+ )
+
+ jobless_supervisor.emit_metrics()
+
+ assert gauge.call_count == 2
+ for call in gauge.call_args_list:
+ assert call.kwargs["extra_tags"] == {"hostname": "astro-host",
"deployment": "demo"}
+
+
+def test_load_triggers_raises_without_job(jobless_supervisor, mocker):
+ """load_triggers() must fail loudly when job is None so missing subclass
overrides surface."""
+ assign_unassigned =
mocker.patch("airflow.jobs.triggerer_job_runner.Trigger.assign_unassigned")
+ ids_for_triggerer =
mocker.patch("airflow.jobs.triggerer_job_runner.Trigger.ids_for_triggerer")
+ update_triggers = mocker.patch.object(TriggerRunnerSupervisor,
"update_triggers")
+
+ with pytest.raises(RuntimeError, match="must override this method"):
+ jobless_supervisor.load_triggers()
+
+ assign_unassigned.assert_not_called()
+ ids_for_triggerer.assert_not_called()
+ update_triggers.assert_not_called()
+
+
+def test_create_workload_uses_supervisor_id_without_job(jobless_supervisor,
mocker):
+ """_create_workload() should fall back to self.id for the log filename
when job is None."""
+ trigger = mocker.Mock()
+ trigger.id = 7
+ trigger.classpath = "some.path.Trigger"
+ trigger.encrypted_kwargs = ""
+ trigger.task_instance.dag_version_id = uuid.uuid4()
+ trigger.task_instance.task_id = "t"
+ trigger.task_instance.trigger_timeout = None
+
+ mocker.patch(
+ "airflow.jobs.triggerer_job_runner.TaskInstanceDTO.model_validate",
+ return_value=mocker.Mock(spec=TaskInstanceDTO),
+ )
+
+ dag_bag = mocker.Mock()
+ serialized_dag_model = mocker.Mock()
+ task = mocker.Mock(start_from_trigger=False)
+ serialized_dag_model.dag.get_task.return_value = task
+ dag_bag.get_serialized_dag_model.return_value = serialized_dag_model
+
+ render_log_fname = mocker.Mock(return_value="/logs/ti")
+
+ jobless_supervisor._create_workload(
+ trigger=trigger,
+ dag_bag=dag_bag,
+ render_log_fname=render_log_fname,
+ session=mocker.Mock(),
+ )
+
+ factory = jobless_supervisor.logger_cache[trigger.id]
+ assert factory.log_path == f"/logs/ti.trigger.{jobless_supervisor.id}.log"
+
+
def test_trigger_lifecycle(spy_agency: SpyAgency, session, testing_dag_bundle):
"""
Checks that the triggerer will correctly see a new Trigger in the database
