kaxil commented on code in PR #66006:
URL: https://github.com/apache/airflow/pull/66006#discussion_r3153724533
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -435,11 +436,12 @@ def is_alive(self) -> bool:
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)
+ id = job.id if job is not None else uuid4()
Review Comment:
Minor: `id = job.id if job is not None else uuid4()` shadows the builtin
`id`. Renaming to `proc_id` (or similar) avoids the shadowing and matches what
gets passed to `super().start(id=...)`.
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -703,9 +721,10 @@ def _create_workload(
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.
+ log_id = self.job.id if self.job is not None else self.id
Review Comment:
Worth flagging that the log filename now mixes formats:
`f"...trigger.{job.id}.log"` for the with-job path (`int`) and
`f"...trigger.{uuid}.log"` for the no-job path. The reader side
(`FileTaskHandler.add_triggerer_suffix`) currently uses `ti.triggerer_job.id`
(int), so AIP-92 subclasses without a metadata `Job` will need their own log
retrieval path. Worth calling out in the docstring here or in the PR
description so the next reader sees the constraint.
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -662,20 +668,32 @@ def on_trigger_failure(self, trigger_id: int, exc:
list[str] | None) -> None:
"""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`.
+
+ Override in subclasses that don't have a metadata-DB ``Job`` (e.g.
AIP-92
+ Execution-API-backed triggerers) to provide hostname or other identity
tags
+ from another source.
+ """
+ hostname = self.job.hostname if self.job is not None else None
+ return {"hostname": hostname} if hostname is not None else {}
Review Comment:
This default looks unsafe. When `self.job is None` and a subclass doesn't
override `metric_tags()`, this returns `{}`. `emit_metrics()` then calls
`DualStatsManager.gauge("triggers.running", ..., extra_tags={})`. With
`export_legacy_names=True` (the class default), `gauge()` calls
`get_legacy_stat(..., variables={})`, which checks
`name_variables=["hostname"]` against the empty dict and raises
`ValueError("Missing required variables for metric 'triggers.running':
['hostname']...")` -- `metrics_template.yaml` defines `hostname` as required
for both `triggers.running` and `triggerer.capacity_left`.
That fires every iteration of `run_once()`, so a no-job supervisor that
forgets to override `metric_tags()` spins on exceptions instead of running. The
"override in subclasses" docstring is non-binding -- the default needs to be
safe by itself. Options: skip gauge emission when `extra_tags` is empty, supply
a placeholder hostname, or fail loudly at construction so the breakage isn't
per-tick.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
