This is an automated email from the ASF dual-hosted git repository.
potiuk 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 8b178f59feb Record trigger failures and owner on its row; tag metrics
by kind (#76101)
8b178f59feb is described below
commit 8b178f59feb68e66233715de5c0da0af8ba24da2
Author: Sean Ghaeli <[email protected]>
AuthorDate: Sun Nov 1 13:40:13 2026 -0700
Record trigger failures and owner on its row; tag metrics by kind (#76101)
* Record trigger failures and owner on its row; tag metrics by kind
A trigger that kept failing left no trace in the database, so no API, UI or
alert could say it was failing, how often or why. Nothing on the row said
who
the trigger ran for either, so a view of every trigger would need a
different
lookup for each kind.
Each failure now records on the row how many failures in a row, since when,
and the error (type and message only, masked with the task SDK masker as the
task runner does, cut to 1,000 characters), and a triggerer that picks the
trigger up continues the count. The count resets once the trigger fires an
event or stays up for two minutes; the last error is kept. Failing triggers
are still re-created every couple of seconds as before, so this is one row
update per failure. A deferred task's or callback's trigger also records its
Dag and owner when it is created, and the migration fills them in for rows
that already exist. Rows created without one afterwards, by components still
on the previous version during a rolling upgrade, get it from the triggerer
when it picks them up.
triggers.failed and triggers.succeeded also could not tell an asset watcher
from a deferred task's or a callback's trigger, so there was no metric to
alert on for failing watchers alone, and nothing said how many triggers are
failing right now. They are now tagged with trigger_kind, and a new
triggers.failing gauge counts the triggers of each kind that failed since
they last recovered. Per-trigger tags would make cardinality grow with the
number of triggers, so the metrics stay per kind; which trigger is failing
is on its row.
* Store trigger errors the database can accept
Postgres refuses a NUL character in text, so a trigger error containing one
made writing the failure raise and stop the triggerer.
---
airflow-core/docs/migrations-ref.rst | 4 +-
.../execution_api/routes/task_instances.py | 2 +
.../src/airflow/jobs/triggerer_job_runner.py | 188 ++++++++-
.../0152_3_5_0_add_state_and_owner_to_trigger.py | 134 +++++++
airflow-core/src/airflow/models/callback.py | 6 +
airflow-core/src/airflow/models/taskinstance.py | 2 +
airflow-core/src/airflow/models/trigger.py | 99 ++++-
airflow-core/src/airflow/utils/db.py | 2 +-
.../versions/head/test_task_instances.py | 1 +
airflow-core/tests/unit/jobs/test_triggerer_job.py | 443 ++++++++++++++++++++-
.../test_0152_add_state_and_owner_to_trigger.py | 207 ++++++++++
airflow-core/tests/unit/models/test_callback.py | 17 +
.../tests/unit/models/test_taskinstance.py | 1 +
airflow-core/tests/unit/models/test_trigger.py | 134 ++++++-
.../observability/metrics/metrics_template.yaml | 13 +-
15 files changed, 1234 insertions(+), 19 deletions(-)
diff --git a/airflow-core/docs/migrations-ref.rst
b/airflow-core/docs/migrations-ref.rst
index 608170aa039..413d4b3bd64 100644
--- a/airflow-core/docs/migrations-ref.rst
+++ b/airflow-core/docs/migrations-ref.rst
@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are
executed via when you ru
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
| Revision ID | Revises ID | Airflow Version | Description
|
+=========================+==================+===================+==============================================================+
-| ``38c8df3fcb9c`` (head) | ``c71065a1b14f`` | ``3.5.0`` | Add index
on dag_run (dag_id, run_after). |
+| ``5fbfa42a2d01`` (head) | ``38c8df3fcb9c`` | ``3.5.0`` | Add
recorded state and owner to trigger. |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``38c8df3fcb9c`` | ``c71065a1b14f`` | ``3.5.0`` | Add index
on dag_run (dag_id, run_after). |
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
| ``c71065a1b14f`` | ``b4f1a7d2c903`` | ``3.5.0`` | Add
session_id and registration_id to Job. |
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
index 8bc084493ce..9473accb162 100644
---
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
+++
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
@@ -968,6 +968,8 @@ def _create_ti_state_update_query_and_update_state(
kwargs={},
queue=ti_patch_payload.queue,
team_name=get_team_name_for_ti(task_instance_id, session),
+ dag_id=dag_id,
+ owner_ref={"ti_id": str(task_instance_id)},
)
trigger_row.encrypted_kwargs = trigger_kwargs
session.add(trigger_row)
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index 2680759dbfb..ea862dc9025 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -28,7 +28,7 @@ import signal
import sys
import threading
import time
-from collections import deque
+from collections import Counter, deque
from collections.abc import Callable, Generator, Hashable, Iterable, Iterator
from contextlib import contextmanager, suppress
from datetime import datetime
@@ -61,8 +61,9 @@ from airflow.jobs.base_job_runner import BaseJobRunner
from airflow.jobs.job import perform_heartbeat
from airflow.models.dagbag import DBDagBag
from airflow.models.task_coordinates import TaskCoordinateResolver
-from airflow.models.trigger import Trigger
+from airflow.models.trigger import RecordedFailure, Trigger, TriggerKind
from airflow.observability.metrics import stats_utils
+from airflow.sdk._shared.secrets_masker import redact # noqa: SDK001
from airflow.sdk.api.datamodels._generated import HITLDetailResponse
from airflow.sdk.bases.xcom import _current_trigger_ti
from airflow.sdk.definitions.asset import Asset
@@ -325,6 +326,91 @@ class TriggererJobRunner(BaseJobRunner, LoggingMixin):
log: FilteringBoundLogger = structlog.get_logger(logger_name=__name__)
+# Stored on the trigger row and shown by the API to everyone who can see the
trigger's owner; the traceback
+# stays in the logs.
+TRIGGER_ERROR_MAX_LENGTH = 1000
+
+# A trigger that stays up this long after it was last (re-)created has
recovered, so one that crashes after
+# running a while is not counted as failing ever since its first crash.
+TRIGGER_RECOVERY_UPTIME = 120.0
+
+
+def format_trigger_error(exc: list[str] | None) -> str:
+ """
+ Return the exception type and message of a trigger failure, redacted and
truncated.
+
+ The runner formats the traceback before anything masks it, so it is
redacted here with the task SDK
+ masker, which learns the connections and variables the runner fetched, as
the task runner does.
+ """
+ if not exc:
+ return "The trigger exited without yielding an event"
+ error = str(redact(exc[-1].strip()))[:TRIGGER_ERROR_MAX_LENGTH]
+ # Escape what a text column refuses (NUL on Postgres, a lone surrogate in
UTF-8), as repr() would, so
+ # writing the failure cannot raise out of the triggerer's loop.
+ return error.replace("\x00", "\\x00").encode("utf-8",
"backslashreplace").decode("utf-8")
+
+
[email protected]
+class _TriggerFailures:
+ consecutive: int = 0
+ # When the trigger was last (re-)created, on the monotonic clock; None
until it is running again.
+ started_at: float | None = None
+
+
[email protected]
+class TriggerFailureTracker:
+ """
+ Count the failures of each trigger since it last recovered, for the
triggers assigned to this triggerer.
+
+ A trigger recovers when it fires an event, or once it has stayed up for
``recovery_uptime`` seconds
+ since it was last (re-)created. Only triggers that failed since they last
recovered are tracked.
+ """
+
+ recovery_uptime: float = TRIGGER_RECOVERY_UPTIME
+ failures: dict[int, _TriggerFailures] = attrs.field(factory=dict)
+
+ def consecutive(self, trigger_id: int) -> int:
+ return state.consecutive if (state := self.failures.get(trigger_id))
is not None else 0
+
+ def continue_from(self, trigger_id: int, consecutive_failures: int | None)
-> None:
+ """Continue the count recorded on the trigger row, e.g. by the
triggerer that ran it before."""
+ if consecutive_failures and trigger_id not in self.failures:
+ self.failures[trigger_id] =
_TriggerFailures(consecutive=consecutive_failures)
+
+ def record_failure(self, trigger_id: int, now: float) -> int:
+ """Count a failure and return the trigger's consecutive failures."""
+ state = self.failures.setdefault(trigger_id, _TriggerFailures())
+ if state.started_at is not None and now - state.started_at >=
self.recovery_uptime:
+ state.consecutive = 0
+ state.consecutive += 1
+ state.started_at = None
+ return state.consecutive
+
+ def record_success(self, trigger_id: int) -> int:
+ """Forget the failures of a trigger that fired an event; return how
many it recovered from."""
+ state = self.failures.pop(trigger_id, None)
+ return state.consecutive if state is not None else 0
+
+ def record_started(self, trigger_ids: Iterable[int], now: float) -> None:
+ for trigger_id in trigger_ids:
+ if (state := self.failures.get(trigger_id)) is not None:
+ state.started_at = now
+
+ def pop_recovered(self, now: float) -> dict[int, int]:
+ """Forget the triggers that stayed up long enough; return the failures
each recovered from."""
+ recovered = {
+ trigger_id: state.consecutive
+ for trigger_id, state in self.failures.items()
+ if state.started_at is not None and now - state.started_at >=
self.recovery_uptime
+ }
+ for trigger_id in recovered:
+ del self.failures[trigger_id]
+ return recovered
+
+ def forget(self, trigger_ids: Iterable[int]) -> None:
+ for trigger_id in trigger_ids:
+ self.failures.pop(trigger_id, None)
+
class TriggerEventEntry(NamedTuple):
"""A fired trigger event queued for dispatch to the Trigger model."""
@@ -554,6 +640,13 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
# Outbound queue of failed triggers
failed_triggers: deque[tuple[int, list[str] | None]] =
attrs.field(factory=deque, init=False)
+ # Failures of the triggers assigned here since each last recovered,
recorded on the trigger row.
+ failure_tracker: TriggerFailureTracker =
attrs.field(factory=TriggerFailureTracker, init=False)
+ # Triggers that recovered since the last loop; their failure state on the
trigger row is cleared.
+ recovered_triggers: set[int] = attrs.field(factory=set, init=False)
+ # TriggerKind value of each trigger assigned here, for the trigger_kind
metric tag
+ trigger_kinds: dict[int, str] = attrs.field(factory=dict, init=False)
+
def is_alive(self) -> bool:
# Set by `_service_subprocess` in the loop
return self._exit_code is None
@@ -630,10 +723,17 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
self._last_runner_comms = time.monotonic()
if isinstance(msg, messages.TriggerStateChanges):
+ # Events before failures: a trigger that fired and then failed
starts a new count.
if msg.events:
self.events.extend(msg.events)
+ for entry in msg.events:
+ self.record_trigger_success(entry.trigger_id)
if msg.failures:
self.failed_triggers.extend(msg.failures)
+ now = time.monotonic()
+ for trigger_id, _ in msg.failures:
+ self.failure_tracker.record_failure(trigger_id, now)
+ self.recovered_triggers.discard(trigger_id)
for id in msg.finished or ():
self.running_triggers.discard(id)
self.cancelling_triggers.discard(id)
@@ -665,6 +765,7 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
workload = self.creating_triggers.popleft()
response.to_create.append(workload)
self.running_triggers.update(m.id for m in response.to_create)
+ self.failure_tracker.record_started((m.id for m in
response.to_create), time.monotonic())
resp = response
elif isinstance(msg, GetConnection):
@@ -760,6 +861,7 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
self._service_subprocess(1)
self.handle_events()
+ self.handle_recovered_triggers()
self.handle_failed_triggers()
self.clean_unused()
self.heartbeat()
@@ -819,7 +921,7 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
if entry.persist_seq is not None:
self.persisted_event_seqs.append(entry.persist_seq)
# Emit stat event
- stats.incr("triggers.succeeded", tags=prune_dict({"team_name":
self.team_name}))
+ stats.incr("triggers.succeeded",
tags=self._trigger_kind_tags(entry.trigger_id))
def on_trigger_event(self, trigger_id: int, event: TriggerEvent) -> None:
"""Record that a trigger fired an event."""
@@ -842,12 +944,50 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
trigger_id, exc = self.failed_triggers.popleft()
self.on_trigger_failure(trigger_id=trigger_id, exc=exc)
# Emit stat event
- stats.incr("triggers.failed", tags=prune_dict({"team_name":
self.team_name}))
+ stats.incr("triggers.failed",
tags=self._trigger_kind_tags(trigger_id))
+
+ def _trigger_kind_tags(self, trigger_id: int) -> dict[str, str]:
+ return prune_dict({"team_name": self.team_name, "trigger_kind":
self.trigger_kinds.get(trigger_id)})
def on_trigger_failure(self, trigger_id: int, exc: list[str] | None) ->
None:
- """Record that a trigger failed."""
+ """Record that a trigger failed, and on its row why, whatever kind of
trigger it is."""
+ # A deferred task's trigger fails its task and is not re-created, so
it fails once.
+ failure = RecordedFailure(
+ consecutive_failures=self.failure_tracker.consecutive(trigger_id)
or 1,
+ error=format_trigger_error(exc),
+ )
Trigger.submit_failure(
- trigger_id=trigger_id, exc=exc, triggerer_id=self.job.id if
self.job is not None else None
+ trigger_id=trigger_id,
+ exc=exc,
+ triggerer_id=self.job.id if self.job is not None else None,
+ failure=failure,
+ )
+
+ def record_trigger_success(self, trigger_id: int) -> None:
+ if after_failures := self.failure_tracker.record_success(trigger_id):
+ self._mark_recovered(trigger_id, after_failures)
+
+ def _mark_recovered(self, trigger_id: int, after_failures: int) -> None:
+ log.info(
+ "Trigger recovered",
+ trigger_id=trigger_id,
+ trigger_kind=self.trigger_kinds.get(trigger_id),
+ after_failures=after_failures,
+ )
+ self.recovered_triggers.add(trigger_id)
+
+ def handle_recovered_triggers(self) -> None:
+ """Clear the recorded failures of triggers that fired an event or have
stayed up long enough."""
+ for trigger_id, after_failures in
self.failure_tracker.pop_recovered(time.monotonic()).items():
+ self._mark_recovered(trigger_id, after_failures)
+ if self.recovered_triggers:
+ trigger_ids, self.recovered_triggers = self.recovered_triggers,
set()
+ self.on_triggers_recovered(trigger_ids)
+
+ def on_triggers_recovered(self, trigger_ids: set[int]) -> None:
+ """Record that triggers recovered from their failures."""
+ Trigger.clear_recorded_failures(
+ trigger_ids, triggerer_id=self.job.id if self.job is not None else
None
)
def metric_tags(self) -> dict[str, str]:
@@ -882,6 +1022,27 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
capacity_left,
tags=tags,
)
+ failing = Counter(self.trigger_kinds.get(trigger_id) for trigger_id in
self.failure_tracker.failures)
+ for kind in TriggerKind:
+ stats.gauge("triggers.failing", failing[kind.value], tags={**tags,
"trigger_kind": kind.value})
+
+ @staticmethod
+ def _record_missing_owner(trigger: Trigger) -> None:
+ """
+ Record the owner of a deferred task's or callback's trigger whose row
has none.
+
+ During a rolling upgrade, components still on the previous version
keep creating these triggers after
+ ``db migrate`` filled in the owners of the rows that existed then.
Written with the session that builds
+ the workload.
+ """
+ if getattr(trigger, "owner_ref", None) is not None:
+ return
+ if (ti := trigger.task_instance) is not None:
+ trigger.dag_id, trigger.owner_ref = ti.dag_id, {"ti_id":
str(ti.id)}
+ elif (callback := getattr(trigger, "callback", None)) is not None:
+ dag_run = ((callback.data.get("kwargs") or {}).get("context") or
{}).get("dag_run") or {}
+ trigger.dag_id = callback.get_dag_id() or dag_run.get("dag_id")
+ trigger.owner_ref = {"callback_id": str(callback.id)}
def _create_workload(
self,
@@ -891,6 +1052,15 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
session: Session,
log_context: TaskLogContext | None = None,
) -> workloads.RunTrigger | None:
+ self._record_missing_owner(trigger)
+ self.trigger_kinds[trigger.id] = (
+ TriggerKind.TASK
+ if trigger.task_instance is not None
+ else TriggerKind.ASSET_WATCHER
+ if trigger.assets
+ else TriggerKind.CALLBACK
+ ).value
+ self.failure_tracker.continue_from(trigger.id,
trigger.consecutive_failures)
if trigger.task_instance is None:
watched_assets: dict[str, str] | None = None
@@ -1089,6 +1259,12 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
self.running_triggers - requested_trigger_ids -
self.cancelling_triggers - self.releasing_triggers
)
+ for trigger_id in self.trigger_kinds.keys() - requested_trigger_ids -
known_trigger_ids:
+ del self.trigger_kinds[trigger_id]
+ self.failure_tracker.forget(
+ self.failure_tracker.failures.keys() - requested_trigger_ids -
known_trigger_ids
+ )
+
if new_trigger_ids:
workloads_to_create = self.build_trigger_workloads(new_trigger_ids)
diff --git
a/airflow-core/src/airflow/migrations/versions/0152_3_5_0_add_state_and_owner_to_trigger.py
b/airflow-core/src/airflow/migrations/versions/0152_3_5_0_add_state_and_owner_to_trigger.py
new file mode 100644
index 00000000000..1b8909ea025
--- /dev/null
+++
b/airflow-core/src/airflow/migrations/versions/0152_3_5_0_add_state_and_owner_to_trigger.py
@@ -0,0 +1,134 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Add recorded state and owner to trigger.
+
+Revision ID: 5fbfa42a2d01
+Revises: 38c8df3fcb9c
+Create Date: 2026-10-28 12:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import json
+
+import sqlalchemy as sa
+from alembic import context, op
+
+from airflow.migrations.utils import disable_sqlite_fkeys
+from airflow.models.base import StringID
+from airflow.utils.sqlalchemy import UtcDateTime
+
+revision = "5fbfa42a2d01"
+down_revision = "38c8df3fcb9c"
+branch_labels = None
+depends_on = None
+airflow_version = "3.5.0"
+
+_trigger = sa.table(
+ "trigger",
+ sa.column("id", sa.Integer),
+ sa.column("dag_id", sa.String),
+ sa.column("owner_ref", sa.JSON),
+)
+_task_instance = sa.table(
+ "task_instance",
+ sa.column("id", sa.Uuid),
+ sa.column("trigger_id", sa.Integer),
+ sa.column("dag_id", sa.String),
+)
+_callback = sa.table(
+ "callback", sa.column("id", sa.Uuid), sa.column("trigger_id", sa.Integer),
sa.column("data", sa.Text)
+)
+
+
+def _decode(value):
+ """Undo the dict encoding of ``callback.data`` (``{"__type": "dict",
"__var": {...}}`` at every level)."""
+ if isinstance(value, dict):
+ if value.get("__type") == "dict" and isinstance(value.get("__var"),
dict):
+ value = value["__var"]
+ return {key: _decode(item) for key, item in value.items()}
+ return value
+
+
+def _callback_dag_id(raw_data) -> str | None:
+ data = _decode(json.loads(raw_data) if isinstance(raw_data, str) else
raw_data) or {}
+ context = (data.get("kwargs") or {}).get("context") or {}
+ return data.get("dag_id") or (context.get("dag_run") or {}).get("dag_id")
+
+
+def upgrade():
+ """Add nullable recorded-state and owner columns to trigger, and record
the owner of existing rows."""
+ with disable_sqlite_fkeys(op):
+ with op.batch_alter_table("trigger", schema=None) as batch_op:
+ batch_op.add_column(sa.Column("consecutive_failures",
sa.Integer(), nullable=True))
+ batch_op.add_column(sa.Column("first_failed_at", UtcDateTime(),
nullable=True))
+ batch_op.add_column(sa.Column("last_failed_at", UtcDateTime(),
nullable=True))
+ batch_op.add_column(sa.Column("last_error", sa.Text(),
nullable=True))
+ batch_op.add_column(sa.Column("last_event_at", UtcDateTime(),
nullable=True))
+ batch_op.add_column(sa.Column("dag_id", StringID(), nullable=True))
+ batch_op.add_column(sa.Column("owner_ref", sa.JSON(),
nullable=True))
+
+ if context.is_offline_mode():
+ # Existing rows keep no owner, and a deferred task's or callback's
trigger is then not shown by the
+ # triggers API until the trigger is re-created.
+ return
+ # Trigger rows only live while something waits on them, so there are few
to fill in.
+ conn = op.get_bind()
+ owners = [
+ {"tid": trigger_id, "b_dag_id": dag_id, "b_owner_ref": {"ti_id":
str(ti_id)}}
+ for ti_id, trigger_id, dag_id in conn.execute(
+ sa.select(_task_instance.c.id, _task_instance.c.trigger_id,
_task_instance.c.dag_id).where(
+ _task_instance.c.trigger_id.is_not(None)
+ )
+ )
+ ]
+ owners += [
+ {
+ "tid": trigger_id,
+ "b_dag_id": _callback_dag_id(raw_data),
+ "b_owner_ref": {"callback_id": str(callback_id)},
+ }
+ for callback_id, trigger_id, raw_data in conn.execute(
+ sa.select(_callback.c.id, _callback.c.trigger_id,
_callback.c.data).where(
+ _callback.c.trigger_id.is_not(None)
+ )
+ )
+ ]
+ if owners:
+ conn.execute(
+ _trigger.update()
+ .where(_trigger.c.id == sa.bindparam("tid"))
+ .values(dag_id=sa.bindparam("b_dag_id"),
owner_ref=sa.bindparam("b_owner_ref", type_=sa.JSON)),
+ owners,
+ )
+
+
+def downgrade():
+ """Remove the recorded-state and owner columns from trigger."""
+ with disable_sqlite_fkeys(op):
+ with op.batch_alter_table("trigger", schema=None) as batch_op:
+ batch_op.drop_column("owner_ref")
+ batch_op.drop_column("dag_id")
+ batch_op.drop_column("last_event_at")
+ batch_op.drop_column("last_error")
+ batch_op.drop_column("last_failed_at")
+ batch_op.drop_column("first_failed_at")
+ batch_op.drop_column("consecutive_failures")
diff --git a/airflow-core/src/airflow/models/callback.py
b/airflow-core/src/airflow/models/callback.py
index 360f7b8b7af..490a167d7cc 100644
--- a/airflow-core/src/airflow/models/callback.py
+++ b/airflow-core/src/airflow/models/callback.py
@@ -266,6 +266,10 @@ class TriggererCallback(Callback):
if self.bundle_name and conf.getboolean("core", "multi_team"):
team_name = DagBundleModel.get_team_name(self.bundle_name,
session=session)
+ if self.id is None:
+ # Not flushed yet; the trigger records which callback it runs for.
+ self.id = uuid6.uuid7()
+ dag_run = ((self.data.get("kwargs") or {}).get("context") or
{}).get("dag_run") or {}
# The kwargs stay encoded: the triggerer decodes them when it creates
the CallbackTrigger.
self.trigger = Trigger.from_serialized_kwargs(
classpath=qualname(CallbackTrigger),
@@ -274,6 +278,8 @@ class TriggererCallback(Callback):
"callback_kwargs":
encode_queued_callback_kwargs(self.data["kwargs"] or {}),
},
queue=self.data.get("queue"),
+ dag_id=self.get_dag_id() or dag_run.get("dag_id"),
+ owner_ref={"callback_id": str(self.id)},
)
self.trigger.team_name = team_name
super().queue(session=session)
diff --git a/airflow-core/src/airflow/models/taskinstance.py
b/airflow-core/src/airflow/models/taskinstance.py
index 9501224dcc0..3de44152302 100644
--- a/airflow-core/src/airflow/models/taskinstance.py
+++ b/airflow-core/src/airflow/models/taskinstance.py
@@ -2729,6 +2729,8 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
classpath=start_trigger_args.trigger_cls,
kwargs=trigger_kwargs,
team_name=team_name,
+ dag_id=self.dag_id,
+ owner_ref={"ti_id": str(self.id)},
)
# First, make the trigger entry
diff --git a/airflow-core/src/airflow/models/trigger.py
b/airflow-core/src/airflow/models/trigger.py
index 656a2a474a1..46a320c8061 100644
--- a/airflow-core/src/airflow/models/trigger.py
+++ b/airflow-core/src/airflow/models/trigger.py
@@ -22,9 +22,9 @@ from collections.abc import Iterable
from enum import Enum
from functools import singledispatch
from traceback import format_exception
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, NamedTuple
-from sqlalchemy import ForeignKey, Index, Integer, String, Text, delete, func,
or_, select, update
+from sqlalchemy import JSON, ForeignKey, Index, Integer, String, Text, delete,
func, or_, select, update
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.orm import Mapped, Session, mapped_column, relationship,
selectinload
from sqlalchemy.sql.functions import coalesce
@@ -33,7 +33,7 @@ from airflow._shared.timezones import timezone
from airflow.assets.manager import AssetManager
from airflow.configuration import conf
from airflow.models.asset import AssetWatcherModel
-from airflow.models.base import Base
+from airflow.models.base import Base, StringID
from airflow.models.taskinstance import TaskInstance
from airflow.serialization.enums import stringify_encoding_keys
from airflow.triggers.base import BaseTaskEndEvent
@@ -58,6 +58,26 @@ Internal use only.
log = logging.getLogger(__name__)
+class RecordedFailure(NamedTuple):
+ """
+ A trigger failure, as recorded on its trigger row.
+
+ :meta private:
+ """
+
+ consecutive_failures: int
+ # Exception type and message only, already redacted and truncated
+ error: str
+
+
+class TriggerKind(str, Enum):
+ """What a trigger runs for, worked out from its owner rather than
stored."""
+
+ TASK = "task"
+ CALLBACK = "callback"
+ ASSET_WATCHER = "asset_watcher"
+
+
class TriggerFailureReason(str, Enum):
"""
Reasons for trigger failures.
@@ -100,6 +120,20 @@ class Trigger(Base):
triggerer_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
queue: Mapped[str | None] = mapped_column(String(256), nullable=True)
+ # What the triggerer records for every trigger, whatever its kind, and
only when it changes: a failure,
+ # a recovery (clears the count and its start; last_failed_at and
last_error survive it) or an event.
+ consecutive_failures: Mapped[int | None] = mapped_column(Integer,
nullable=True)
+ first_failed_at: Mapped[datetime.datetime | None] =
mapped_column(UtcDateTime, nullable=True)
+ last_failed_at: Mapped[datetime.datetime | None] =
mapped_column(UtcDateTime, nullable=True)
+ last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
+ last_event_at: Mapped[datetime.datetime | None] =
mapped_column(UtcDateTime, nullable=True)
+
+ # Who the trigger runs for, written once by whoever creates it: ``dag_id``
is the owner's Dag and
+ # ``owner_ref`` the owner within it, ``{"ti_id": ...}`` or
``{"callback_id": ...}``. Both stay empty for
+ # asset watchers, whose owners are the asset_watcher rows, shared and
changing as Dags are parsed.
+ dag_id: Mapped[str | None] = mapped_column(StringID(), nullable=True)
+ owner_ref: Mapped[dict[str, Any] | None] =
mapped_column(JSON(none_as_null=True), nullable=True)
+
# Denormalized from dag_bundle_team to keep the triggerer's ~1s polling
queries join-free,
# especially since it's eventually consistent and trigger rows are
ephemeral.
# Without this, filtering by team requires 2-3 joins depending on trigger
type.
@@ -132,6 +166,9 @@ class Trigger(Base):
created_date: datetime.datetime | None = None,
queue: str | None = None,
team_name: str | None = None,
+ *,
+ dag_id: str | None = None,
+ owner_ref: dict[str, Any] | None = None,
) -> None:
super().__init__()
self.classpath = classpath
@@ -139,6 +176,8 @@ class Trigger(Base):
self.created_date = created_date or timezone.utcnow()
self.queue = queue
self.team_name = team_name
+ self.dag_id = dag_id
+ self.owner_ref = owner_ref
@property
def kwargs(self) -> dict[str, Any]:
@@ -208,7 +247,13 @@ class Trigger(Base):
@classmethod
def from_serialized_kwargs(
- cls, classpath: str, serialized_kwargs: dict[str, Any], queue: str |
None = None
+ cls,
+ classpath: str,
+ serialized_kwargs: dict[str, Any],
+ queue: str | None = None,
+ *,
+ dag_id: str | None = None,
+ owner_ref: dict[str, Any] | None = None,
) -> Trigger:
"""
Create a trigger row from kwargs that are already in their
serde-encoded form.
@@ -216,7 +261,7 @@ class Trigger(Base):
The kwargs are stored exactly as given and decoded by the triggerer
when it creates
the trigger, the same as kwargs stored by :meth:`from_object`.
"""
- trigger = cls(classpath=classpath, kwargs={}, queue=queue)
+ trigger = cls(classpath=classpath, kwargs={}, queue=queue,
dag_id=dag_id, owner_ref=owner_ref)
trigger.encrypted_kwargs =
cls._encrypt_serialized_kwargs(serialized_kwargs)
return trigger
@@ -322,11 +367,19 @@ class Trigger(Base):
)
if trigger.callback:
trigger.callback.handle_event(event, session)
+ # Recorded for every kind of trigger; written with the rest of this
transaction.
+ trigger.last_event_at = timezone.utcnow()
@classmethod
@provide_session
def submit_failure(
- cls, trigger_id, exc=None, *, triggerer_id: int | None = None,
session: Session = NEW_SESSION
+ cls,
+ trigger_id,
+ exc=None,
+ *,
+ triggerer_id: int | None = None,
+ failure: RecordedFailure | None = None,
+ session: Session = NEW_SESSION,
) -> None:
"""
When a trigger has failed unexpectedly, mark everything that depended
on it as failed.
@@ -334,6 +387,9 @@ class Trigger(Base):
With ``triggerer_id``, only act while that triggerer still owns the
trigger: a reassigned
trigger is being run elsewhere, so a failure from its previous owner
must not fail the task.
+ ``failure`` is recorded on the trigger row for every kind of trigger,
so the failure shows the same
+ way whether a task depended on the trigger or not (an asset watcher
has no task to fail).
+
Notably, we have to actually run the failure code from a worker as it
may
have linked callbacks, so hilariously we have to re-schedule the task
instances to a worker just so they can then fail.
@@ -364,6 +420,37 @@ class Trigger(Base):
task_instance.state = TaskInstanceState.SCHEDULED
task_instance.scheduled_dttm = timezone.utcnow()
+ if failure is not None:
+ now = timezone.utcnow()
+ stmt = update(cls).where(cls.id == trigger_id)
+ if triggerer_id is not None:
+ stmt = stmt.where(cls.triggerer_id == triggerer_id)
+ session.execute(
+ stmt.values(
+ consecutive_failures=failure.consecutive_failures,
+ first_failed_at=now
+ if failure.consecutive_failures == 1
+ else coalesce(cls.first_failed_at, now),
+ last_failed_at=now,
+ last_error=failure.error,
+ ).execution_options(synchronize_session=False)
+ )
+
+ @classmethod
+ @provide_session
+ def clear_recorded_failures(
+ cls, trigger_ids: Iterable[int], *, triggerer_id: int | None = None,
session: Session = NEW_SESSION
+ ) -> None:
+ """Record that these triggers recovered; their last error is kept."""
+ stmt = update(cls).where(cls.id.in_(trigger_ids),
cls.consecutive_failures.is_not(None))
+ if triggerer_id is not None:
+ stmt = stmt.where(cls.triggerer_id == triggerer_id)
+ session.execute(
+ stmt.values(consecutive_failures=None,
first_failed_at=None).execution_options(
+ synchronize_session=False
+ )
+ )
+
@classmethod
@provide_session
def ids_for_triggerer(
diff --git a/airflow-core/src/airflow/utils/db.py
b/airflow-core/src/airflow/utils/db.py
index 061f5042b8b..a0956e49b1d 100644
--- a/airflow-core/src/airflow/utils/db.py
+++ b/airflow-core/src/airflow/utils/db.py
@@ -112,7 +112,7 @@ _REVISION_HEADS_MAP: dict[str, str] = {
"3.2.0": "1d6611b6ab7c",
"3.3.0": "d2f4e1b3c5a7",
"3.4.0": "b4f1a7d2c903",
- "3.5.0": "38c8df3fcb9c",
+ "3.5.0": "5fbfa42a2d01",
}
# Prefix used to identify tables holding data moved during migration.
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
index 02a605effdd..2c8dc7e95a9 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
@@ -3973,6 +3973,7 @@ class TestTIUpdateState:
assert t[0].queue == "default"
else:
assert t[0].queue is None
+ assert (t[0].dag_id, t[0].owner_ref) == (ti.dag_id, {"ti_id":
str(ti.id)})
@staticmethod
def _defer_ti_in_team_bundle(client, session, create_task_instance):
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py
b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index 18b0cccda82..ff0e03e8099 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -61,24 +61,29 @@ from airflow.jobs.job import Job
from airflow.jobs.triggerer_job_runner import (
_REASSIGNED_CANCEL_MSG,
_USER_ACTION_CANCEL_MSG,
+ TRIGGER_RECOVERY_UPTIME,
ToTriggerRunner,
ToTriggerSupervisor,
TriggerCommsDecoder,
TriggererJobRunner,
TriggerEventEntry,
+ TriggerFailureTracker,
TriggerLoggingFactory,
TriggerRunner,
TriggerRunnerSupervisor,
_make_trigger_span,
+ format_trigger_error,
messages,
)
from airflow.models import Connection, DagModel, DagRun, Trigger, Variable
+from airflow.models.asset import AssetModel
from airflow.models.dag_version import DagVersion
from airflow.models.dagbag import DBDagBag
from airflow.models.dagbundle import DagBundleModel
from airflow.models.dynamic_region import DynamicRegion
from airflow.models.serialized_dag import SerializedDagModel
from airflow.models.taskinstance import TaskInstance
+from airflow.models.trigger import RecordedFailure
from airflow.models.xcom import XComModel, XComModelV2
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.python import PythonOperator
@@ -101,6 +106,7 @@ from airflow.sdk.execution_time.comms import (
GetAssetStateStoreByName,
GetAssetStateStoreByUri,
GetXCom,
+ MaskSecret,
OKResponse,
SetAssetStateStoreByName,
SetAssetStateStoreByUri,
@@ -800,9 +806,11 @@ def
test_emit_metrics_uses_metric_tags_override(jobless_supervisor, mocker):
jobless_supervisor.emit_metrics()
- assert gauge.call_count == 2
+ # Two triggerer gauges, and the failing-triggers gauge once per trigger
kind
+ assert gauge.call_count == 2 + 3
for call in gauge.call_args_list:
- assert call.kwargs["tags"] == {"hostname": "astro-host", "deployment":
"demo"}
+ tags = {k: v for k, v in call.kwargs["tags"].items() if k !=
"trigger_kind"}
+ assert tags == {"hostname": "astro-host", "deployment": "demo"}
def test_load_triggers_raises_without_job(jobless_supervisor, mocker):
@@ -848,6 +856,7 @@ def
test_create_workload_uses_supervisor_id_without_job(jobless_supervisor, mock
trigger.id = 7
trigger.classpath = "some.path.Trigger"
trigger.encrypted_kwargs = ""
+ trigger.consecutive_failures = None # nothing recorded on the row
trigger.task_instance.dag_version_id = uuid.uuid4()
trigger.task_instance.task_id = "t"
trigger.task_instance.trigger_timeout = None
@@ -877,6 +886,7 @@ def
test_create_workload_uses_supervisor_id_without_job(jobless_supervisor, mock
factory = jobless_supervisor.logger_cache[trigger.id]
assert factory.log_path == f"/logs/ti.trigger.{jobless_supervisor.id}.log"
+ assert jobless_supervisor.trigger_kinds[trigger.id] == "task"
@pytest.mark.parametrize(
@@ -892,6 +902,7 @@ def
test_create_workload_resolves_serialized_dag_from_run(jobless_supervisor, mo
trigger.id = 8
trigger.classpath = "some.path.Trigger"
trigger.encrypted_kwargs = ""
+ trigger.consecutive_failures = None # nothing recorded on the row
trigger.task_instance.dag_version_id = bumped_ti_version
trigger.task_instance.task_id = "t"
trigger.task_instance.trigger_timeout = None
@@ -981,6 +992,7 @@ def
test_create_workload_sets_up_logging_for_callback_trigger(jobless_supervisor
trigger.id = 8
trigger.classpath = "airflow.triggers.callback.CallbackTrigger"
trigger.encrypted_kwargs = ""
+ trigger.consecutive_failures = None # nothing recorded on the row
trigger.task_instance = None
trigger.assets = None
trigger.callback = callback
@@ -1014,6 +1026,7 @@ def
test_create_workload_sets_watched_assets_for_asset_only_trigger(jobless_supe
trigger.encrypted_kwargs = "encrypted"
trigger.task_instance = None # Not tied to a Task (similar to a
BaseEventTrigger)
trigger.assets = [asset1, asset2]
+ trigger.consecutive_failures = None # nothing recorded on the row
workload = jobless_supervisor._create_workload(
trigger=trigger,
@@ -1032,6 +1045,7 @@ def
test_create_workload_watched_assets_none_when_no_assets(jobless_supervisor,
trigger.id = 43
trigger.classpath = "some.path.Trigger"
trigger.encrypted_kwargs = "encrypted"
+ trigger.consecutive_failures = None # nothing recorded on the row
trigger.task_instance = None
trigger.assets = [] # No Assets are attached to the trigger
@@ -4039,3 +4053,428 @@ async def
test_trigger_event_payload_available_at_debug(cap_structlog):
log.get("event") == "Trigger fired event payload" and
log.get("result") == TriggerEvent(payload)
for log in debug_logs
), "Full event payload must be logged at DEBUG level"
+
+
+class TestTriggerFailureRecording:
+ @pytest.mark.parametrize(
+ ("exc", "expected"),
+ [
+ pytest.param(
+ [
+ "Traceback (most recent call last):\n",
+ ' File "x.py", line 1\n',
+ "ValueError: expired token\n",
+ ],
+ "ValueError: expired token",
+ id="type-and-message-only",
+ ),
+ pytest.param(None, "The trigger exited without yielding an event",
id="no-exception"),
+ pytest.param([f"ValueError: {'x' * 5000}\n"], f"ValueError: {'x' *
988}", id="truncated"),
+ pytest.param(["ValueError: a\x00b\n"], "ValueError: a\\x00b",
id="nul-escaped"),
+ pytest.param(["ValueError: a\udcffb\n"], "ValueError: a\\udcffb",
id="lone-surrogate-escaped"),
+ ],
+ )
+ def test_format_trigger_error(self, exc, expected):
+ assert format_trigger_error(exc) == expected
+
+ @pytest.fixture
+ def clock(self, mocker):
+ return
mocker.patch("airflow.jobs.triggerer_job_runner.time.monotonic",
return_value=1000.0)
+
+ @pytest.fixture
+ def supervisor(self, jobless_supervisor, mocker, clock):
+ mocker.patch.object(TriggerRunnerSupervisor, "send_msg", autospec=True)
+ return jobless_supervisor
+
+ @staticmethod
+ def _fail(supervisor, trigger_id, error="ValueError: expired\n", req_id=1):
+ supervisor._handle_request(
+ messages.TriggerStateChanges(failures=[(trigger_id, [error])],
finished=[trigger_id]),
+ log=MagicMock(),
+ req_id=req_id,
+ )
+
+ @staticmethod
+ def _start(supervisor, trigger_id, req_id=1):
+ supervisor.creating_triggers.append(
+ RunTrigger(id=trigger_id, classpath="x.Y", encrypted_kwargs="",
ti=None)
+ )
+ supervisor._handle_request(messages.TriggerStateChanges(),
log=MagicMock(), req_id=req_id)
+
+ @pytest.mark.enable_redact
+ def
test_failure_is_recorded_redacted_with_the_secrets_the_runner_masked(self,
supervisor, mocker):
+ submit = mocker.patch.object(Trigger, "submit_failure", autospec=True)
+ password = f"EnvPw-{uuid.uuid4().hex[:8]}"
+ # What the runner sends after it resolves a connection.
+ supervisor._handle_request(MaskSecret(value=password,
name="password"), log=MagicMock(), req_id=1)
+ traceback = [f"PermissionError: auth failed for
user:{password}@host\n"]
+ self._fail(supervisor, 1, error=traceback[0], req_id=2)
+
+ supervisor.handle_failed_triggers()
+
+ submit.assert_called_once_with(
+ trigger_id=1,
+ exc=traceback,
+ triggerer_id=None,
+ failure=RecordedFailure(
+ consecutive_failures=1, error="PermissionError: auth failed
for user:***@host"
+ ),
+ )
+
+ def test_every_failure_is_recorded_with_its_count(self, supervisor,
mocker):
+ submit = mocker.patch.object(Trigger, "submit_failure", autospec=True)
+
+ for req_id in (1, 2, 3):
+ self._start(supervisor, 1, req_id=10 + req_id)
+ self._fail(supervisor, 1, req_id=req_id)
+ supervisor.handle_failed_triggers()
+
+ assert [c.kwargs["failure"] for c in submit.call_args_list] == [
+ RecordedFailure(consecutive_failures=n, error="ValueError:
expired") for n in (1, 2, 3)
+ ]
+
+ def test_deferred_task_trigger_failure_is_recorded(self, supervisor,
mocker):
+ submit = mocker.patch.object(Trigger, "submit_failure", autospec=True)
+ supervisor.trigger_kinds[2] = "task"
+ self._fail(supervisor, 2, error="TimeoutError: boom\n")
+
+ supervisor.handle_failed_triggers()
+
+ assert submit.call_args.kwargs["failure"] == RecordedFailure(
+ consecutive_failures=1, error="TimeoutError: boom"
+ )
+
+ @pytest.mark.parametrize(
+ ("failures_after_event", "cleared"),
+ [pytest.param(None, {1}, id="recovered"), pytest.param([(1, None)],
None, id="failed-again")],
+ )
+ def test_an_event_clears_the_row_unless_the_trigger_failed_again(
+ self, supervisor, mocker, cap_structlog, failures_after_event, cleared
+ ):
+ clear = mocker.patch.object(Trigger, "clear_recorded_failures",
autospec=True)
+ supervisor.failure_tracker.continue_from(1, 3)
+ event = TriggerEventEntry(trigger_id=1, event=TriggerEvent(True),
persist_seq=None)
+
+ supervisor._handle_request(
+ messages.TriggerStateChanges(events=[event],
failures=failures_after_event),
+ log=MagicMock(),
+ req_id=1,
+ )
+ supervisor.handle_recovered_triggers()
+ supervisor.handle_recovered_triggers()
+
+ if cleared:
+ clear.assert_called_once_with(cleared, triggerer_id=None)
+ assert {"event": "Trigger recovered", "trigger_id": 1,
"after_failures": 3} in [
+ {k: e.get(k) for k in ("event", "trigger_id",
"after_failures")} for e in cap_structlog
+ ]
+ else:
+ # The event ended the old run of failures; the failure after it
starts a new one.
+ clear.assert_not_called()
+ assert supervisor.failure_tracker.consecutive(1) == 1
+
+ def test_a_trigger_that_stays_up_recovers(self, supervisor, mocker, clock,
cap_structlog):
+ mocker.patch.object(Trigger, "submit_failure", autospec=True)
+ clear = mocker.patch.object(Trigger, "clear_recorded_failures",
autospec=True)
+ self._fail(supervisor, 1)
+ supervisor.handle_failed_triggers()
+ self._start(supervisor, 1, req_id=2)
+
+ clock.return_value = 1000.0 + TRIGGER_RECOVERY_UPTIME - 1
+ supervisor.handle_recovered_triggers()
+ clear.assert_not_called()
+ assert supervisor.failure_tracker.consecutive(1) == 1
+
+ clock.return_value = 1000.0 + TRIGGER_RECOVERY_UPTIME
+ supervisor.handle_recovered_triggers()
+ clear.assert_called_once_with({1}, triggerer_id=None)
+ assert supervisor.failure_tracker.consecutive(1) == 0
+ assert "Trigger recovered" in cap_structlog
+
+ def test_a_trigger_that_has_not_been_restarted_yet_does_not_recover(self,
supervisor, clock):
+ self._fail(supervisor, 1)
+ clock.return_value = 1000.0 + 10 * TRIGGER_RECOVERY_UPTIME
+
+ assert supervisor.failure_tracker.pop_recovered(clock.return_value) ==
{}
+
+ def test_a_crash_after_staying_up_starts_a_new_count(self, supervisor,
mocker, clock):
+ submit = mocker.patch.object(Trigger, "submit_failure", autospec=True)
+ supervisor.failure_tracker.continue_from(1, 5)
+ self._start(supervisor, 1)
+
+ # Fails before the loop got to notice it had recovered.
+ clock.return_value = 1000.0 + TRIGGER_RECOVERY_UPTIME + 1
+ self._fail(supervisor, 1, req_id=2)
+ supervisor.handle_failed_triggers()
+
+ assert submit.call_args.kwargs["failure"].consecutive_failures == 1
+
+ @pytest.mark.parametrize("kind", ["asset_watcher", "callback"])
+ def test_new_owner_continues_the_recorded_count(self, supervisor, mocker,
kind):
+ callback = mocker.Mock(data={}, id=uuid.uuid4()) if kind == "callback"
else None
+ trigger = mocker.Mock(
+ spec=Trigger,
+ id=1,
+ callback=callback,
+ task_instance=None,
+ classpath="x.Y",
+ encrypted_kwargs="",
+ assets=[mocker.Mock(uri="s3://b/k")] if kind == "asset_watcher"
else [],
+ asset_watchers=[],
+ consecutive_failures=4,
+ )
+ for asset in trigger.assets:
+ asset.name = "a"
+ mocker.patch.object(
+ TriggerRunnerSupervisor,
+ "build_trigger_workloads",
+ autospec=True,
+ side_effect=lambda sup, ids: [
+ sup._create_workload(trigger, dag_bag=None,
render_log_fname=None, session=None) for _ in ids
+ ],
+ )
+
+ supervisor.update_triggers({1})
+
+ assert supervisor.failure_tracker.consecutive(1) == 4
+ # Without a backoff it is re-created right away, as before.
+ assert [w.id for w in supervisor.creating_triggers] == [1]
+
+ def test_a_trigger_no_longer_assigned_here_is_forgotten(self, supervisor):
+ self._fail(supervisor, 1)
+ supervisor.failed_triggers.clear() # handled
+
+ supervisor.update_triggers(set())
+
+ assert supervisor.failure_tracker.failures == {}
+
+ def test_failure_state_survives_a_move_to_another_triggerer(self, session,
supervisor_builder, mocker):
+ mocker.patch.object(TriggerRunnerSupervisor, "send_msg", autospec=True)
+ trigger = Trigger.from_object(SuccessTrigger())
+ session.add(trigger)
+ session.flush()
+ asset = AssetModel(name="moved", uri="s3://b/moved")
+ asset.add_trigger(trigger, "moved_watcher")
+ session.add(asset)
+ first = supervisor_builder()
+ trigger.triggerer_id = first.job.id
+ session.commit()
+
+ first.update_triggers({trigger.id})
+ first._handle_request(messages.TriggerStateChanges(), log=MagicMock(),
req_id=1)
+ self._fail(first, trigger.id, error="ValueError: boom\n", req_id=2)
+ first.handle_failed_triggers()
+ session.expire_all()
+ row = session.get(Trigger, trigger.id)
+ assert (row.consecutive_failures, row.last_error) == (1, "ValueError:
boom")
+ first_failed_at = row.first_failed_at
+
+ second = supervisor_builder()
+ row.triggerer_id = second.job.id
+ session.commit()
+ second.update_triggers({trigger.id})
+ second._handle_request(messages.TriggerStateChanges(),
log=MagicMock(), req_id=1)
+ self._fail(second, trigger.id, error="ValueError: again\n", req_id=2)
+ second.handle_failed_triggers()
+
+ session.expire_all()
+ row = session.get(Trigger, trigger.id)
+ assert (row.consecutive_failures, row.first_failed_at, row.last_error)
== (
+ 2,
+ first_failed_at,
+ "ValueError: again",
+ )
+
+
+class TestTriggerFailureTracker:
+ def test_only_triggers_that_failed_are_tracked(self):
+ tracker = TriggerFailureTracker(recovery_uptime=10)
+
+ tracker.record_started([1, 2], now=0)
+ tracker.continue_from(3, None)
+
+ assert tracker.failures == {}
+
+ def test_counts_until_an_event(self):
+ tracker = TriggerFailureTracker(recovery_uptime=10)
+
+ assert [tracker.record_failure(1, now=n) for n in range(3)] == [1, 2,
3]
+ assert tracker.record_success(1) == 3
+ assert tracker.record_success(1) == 0
+ assert tracker.consecutive(1) == 0
+
+ def test_counts_until_it_stays_up(self):
+ tracker = TriggerFailureTracker(recovery_uptime=10)
+ tracker.record_failure(1, now=0)
+ tracker.record_failure(2, now=0)
+ tracker.record_started([1, 2], now=1)
+ tracker.record_failure(2, now=5)
+
+ assert tracker.pop_recovered(now=10) == {}
+ assert tracker.pop_recovered(now=11) == {1: 1}
+ assert tracker.consecutive(2) == 2
+
+
+class TestTriggerMetrics:
+ @pytest.fixture
+ def supervisor(self, jobless_supervisor):
+ jobless_supervisor.trigger_kinds.update({1: "asset_watcher", 2:
"callback", 3: "task"})
+ return jobless_supervisor
+
+ @mock.patch.object(TriggerRunnerSupervisor, "on_trigger_failure",
autospec=True)
+ @mock.patch("airflow.jobs.triggerer_job_runner.stats.incr")
+ def test_failures_are_tagged_by_kind(self, incr, _, supervisor):
+ supervisor.failed_triggers.extend([(1, None), (2, None), (3, None)])
+
+ supervisor.handle_failed_triggers()
+
+ assert incr.call_args_list == [
+ mock.call("triggers.failed", tags={"trigger_kind":
"asset_watcher"}),
+ mock.call("triggers.failed", tags={"trigger_kind": "callback"}),
+ mock.call("triggers.failed", tags={"trigger_kind": "task"}),
+ ]
+
+ @mock.patch.object(TriggerRunnerSupervisor, "on_trigger_failure",
autospec=True)
+ @mock.patch("airflow.jobs.triggerer_job_runner.stats.incr")
+ def test_trigger_failures_carry_the_team(self, incr, _, supervisor):
+ supervisor.team_name = "team_a"
+ supervisor.failed_triggers.append((1, None))
+
+ supervisor.handle_failed_triggers()
+
+ incr.assert_called_once_with(
+ "triggers.failed", tags={"trigger_kind": "asset_watcher",
"team_name": "team_a"}
+ )
+
+ @mock.patch.object(TriggerRunnerSupervisor, "on_trigger_event",
autospec=True)
+ @mock.patch("airflow.jobs.triggerer_job_runner.stats.incr")
+ def test_kind_is_kept_until_a_removed_trigger_is_handled(self, incr, _,
supervisor):
+ # The trigger left the requested set (its Dag was paused, say) with
its event still queued.
+ supervisor.events.append(TriggerEventEntry(1, TriggerEvent(True),
None))
+
+ supervisor.update_triggers(set())
+ supervisor.handle_events()
+
+ incr.assert_called_once_with("triggers.succeeded",
tags={"trigger_kind": "asset_watcher"})
+
+ @mock.patch.object(TriggerRunnerSupervisor, "on_trigger_event",
autospec=True)
+ @mock.patch("airflow.jobs.triggerer_job_runner.stats.incr")
+ def test_events_are_tagged_by_kind(self, incr, _, supervisor):
+ supervisor.events.append(TriggerEventEntry(3, TriggerEvent(True),
None))
+
+ supervisor.handle_events()
+
+ incr.assert_called_once_with("triggers.succeeded",
tags={"trigger_kind": "task"})
+
+ def
test_failing_triggers_gauge_counts_triggers_with_failures_per_kind(self,
supervisor_builder, mocker):
+ supervisor = supervisor_builder()
+ supervisor.trigger_kinds.update(
+ {1: "asset_watcher", 2: "asset_watcher", 3: "callback", 4:
"asset_watcher"}
+ )
+ supervisor.failure_tracker.continue_from(1, 2)
+ supervisor.failure_tracker.continue_from(3, 1)
+ # Re-created after its failures: still failing until it recovers.
+ supervisor.failure_tracker.continue_from(4, 3)
+ supervisor.failure_tracker.record_started([4], now=0)
+ gauge = mocker.patch("airflow.jobs.triggerer_job_runner.stats.gauge")
+
+ supervisor.emit_metrics()
+
+ hostname = supervisor.job.hostname
+ assert [c for c in gauge.call_args_list if c.args[0] ==
"triggers.failing"] == [
+ mock.call("triggers.failing", 0, tags={"hostname": hostname,
"trigger_kind": "task"}),
+ mock.call("triggers.failing", 1, tags={"hostname": hostname,
"trigger_kind": "callback"}),
+ mock.call("triggers.failing", 2, tags={"hostname": hostname,
"trigger_kind": "asset_watcher"}),
+ ]
+
+ @pytest.mark.parametrize(
+ ("kwargs", "expected"),
+ [
+ pytest.param(
+ {"assets": [mock.Mock(uri="s3://b/k")], "callback": None},
"asset_watcher", id="watcher"
+ ),
+ pytest.param({"assets": [], "callback": mock.Mock(data={})},
"callback", id="callback"),
+ ],
+ )
+ def
test_kind_is_recorded_when_the_workload_is_built_and_dropped_with_the_trigger(
+ self, jobless_supervisor, mocker, kwargs, expected
+ ):
+ for asset in kwargs["assets"]:
+ asset.name = "a"
+ trigger = mocker.Mock(
+ spec=Trigger,
+ id=4,
+ task_instance=None,
+ classpath="x.Y",
+ encrypted_kwargs="",
+ consecutive_failures=None,
+ asset_watchers=[],
+ **kwargs,
+ )
+ jobless_supervisor._create_workload(
+ trigger=trigger, dag_bag=mocker.Mock(),
render_log_fname=mocker.Mock(), session=mocker.Mock()
+ )
+ assert jobless_supervisor.trigger_kinds[4] == expected
+
+ jobless_supervisor.update_triggers(set())
+
+ assert 4 not in jobless_supervisor.trigger_kinds
+
+
+class TestOwnerOfTriggersCreatedWithoutOne:
+ """Rows created by components still on the previous version during a
rolling upgrade have no owner."""
+
+ @pytest.mark.usefixtures("testing_dag_bundle")
+ def test_the_owner_of_a_deferred_tasks_trigger_is_recorded(self, session,
jobless_supervisor):
+ trigger = Trigger.from_object(SuccessTrigger())
+ session.add(trigger)
+ session.flush()
+ with DAG(dag_id="owner_dag") as dag:
+ task = PythonOperator(task_id="owner_task", python_callable=print)
+ session.add(DagModel(dag_id="owner_dag", bundle_name="testing"))
+ SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag),
bundle_name="testing")
+ dag_run = DagRun(dag.dag_id, run_id="owner_run", run_type="manual",
run_after=timezone.utcnow())
+ ti = create_task_instance(
+ task,
+ run_id=dag_run.run_id,
+ state=TaskInstanceState.DEFERRED,
+ dag_version_id=DagVersion.get_latest_version(dag.dag_id).id,
+ )
+ ti.trigger_id = trigger.id
+ session.add_all([dag_run, ti])
+ session.commit()
+
+ jobless_supervisor.build_trigger_workloads({trigger.id})
+
+ session.expire_all()
+ row = session.get(Trigger, trigger.id)
+ assert (row.dag_id, row.owner_ref) == ("owner_dag", {"ti_id":
str(ti.id)})
+
+ @pytest.mark.parametrize(
+ ("data", "expected_dag_id"),
+ [
+ pytest.param({"dag_id": "data_dag"}, "data_dag",
id="dag-id-in-data"),
+ pytest.param(
+ {"kwargs": {"context": {"dag_run": {"dag_id":
"context_dag"}}}},
+ "context_dag",
+ id="in-context",
+ ),
+ ],
+ )
+ def test_the_owner_of_a_callbacks_trigger_is_recorded(self, mocker, data,
expected_dag_id):
+ callback = mocker.Mock(data=data, id=uuid.uuid4())
+ callback.get_dag_id.return_value = data.get("dag_id")
+ trigger = mocker.Mock(
+ spec=Trigger, owner_ref=None, dag_id=None, task_instance=None,
callback=callback
+ )
+
+ TriggerRunnerSupervisor._record_missing_owner(trigger)
+
+ assert (trigger.dag_id, trigger.owner_ref) == (expected_dag_id,
{"callback_id": str(callback.id)})
+
+ def test_a_recorded_owner_is_kept(self, mocker):
+ trigger = mocker.Mock(spec=Trigger, owner_ref={"ti_id": "x"},
dag_id="d", task_instance=mocker.Mock())
+
+ TriggerRunnerSupervisor._record_missing_owner(trigger)
+
+ assert (trigger.dag_id, trigger.owner_ref) == ("d", {"ti_id": "x"})
diff --git
a/airflow-core/tests/unit/migrations/test_0152_add_state_and_owner_to_trigger.py
b/airflow-core/tests/unit/migrations/test_0152_add_state_and_owner_to_trigger.py
new file mode 100644
index 00000000000..9dcb0bea4dd
--- /dev/null
+++
b/airflow-core/tests/unit/migrations/test_0152_add_state_and_owner_to_trigger.py
@@ -0,0 +1,207 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Tests for the trigger state and owner migration, written against the schema as
it stood at that revision.
+
+The tables are declared here instead of taken from ``airflow.models``, and
alembic moves the database
+to the revision, so later migrations and model changes cannot change what
these tests exercise.
+"""
+
+from __future__ import annotations
+
+import json
+from importlib import import_module
+from uuid import UUID, uuid4
+
+import pytest
+import sqlalchemy as sa
+from sqlalchemy import inspect
+
+from airflow import settings
+from airflow._shared.timezones import timezone
+from airflow.utils.db import downgrade, upgradedb
+from airflow.utils.sqlalchemy import CompactUUID, UtcDateTime
+
+pytestmark = pytest.mark.db_test
+
+_migration =
import_module("airflow.migrations.versions.0152_3_5_0_add_state_and_owner_to_trigger")
+REVISION = _migration.revision
+PREVIOUS_REVISION = _migration.down_revision
+DAG_ID = "migration_0152"
+CLASSPATH = "migration_0152.Trigger"
+NEW_COLUMNS = {
+ "consecutive_failures",
+ "first_failed_at",
+ "last_failed_at",
+ "last_error",
+ "last_event_at",
+ "dag_id",
+ "owner_ref",
+}
+
+metadata = sa.MetaData()
+dag_run = sa.Table(
+ "dag_run",
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ sa.Column("dag_id", sa.String(250)),
+ sa.Column("run_id", sa.String(250)),
+ sa.Column("run_type", sa.String(50)),
+ sa.Column("run_after", UtcDateTime),
+)
+trigger = sa.Table(
+ "trigger",
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ sa.Column("classpath", sa.String(1000)),
+ sa.Column("kwargs", sa.Text),
+ sa.Column("created_date", UtcDateTime),
+)
+task_instance = sa.Table(
+ "task_instance",
+ metadata,
+ sa.Column("id", sa.Uuid, primary_key=True),
+ sa.Column("dag_id", sa.String(250)),
+ sa.Column("task_id", sa.String(250)),
+ sa.Column("run_id", sa.String(250)),
+ sa.Column("region_index", sa.Integer),
+ sa.Column("region_id", CompactUUID),
+ sa.Column("working_set", sa.Boolean),
+ sa.Column("try_number", sa.Integer),
+ sa.Column("pool", sa.String(256)),
+ sa.Column("pool_slots", sa.Integer),
+ sa.Column("trigger_id", sa.Integer),
+)
+callback = sa.Table(
+ "callback",
+ metadata,
+ sa.Column("id", sa.Uuid, primary_key=True),
+ sa.Column("type", sa.String(20)),
+ sa.Column("fetch_method", sa.String(20)),
+ # JSON on every database: jsonb on Postgres, so a string would not bind
there.
+ sa.Column("data", sa.JSON),
+ sa.Column("priority_weight", sa.Integer),
+ sa.Column("created_at", UtcDateTime),
+ sa.Column("trigger_id", sa.Integer),
+)
+
+
+def _delete_rows(connection):
+ trigger_ids = sa.select(trigger.c.id).where(trigger.c.classpath ==
CLASSPATH)
+
connection.execute(callback.delete().where(callback.c.trigger_id.in_(trigger_ids)))
+ connection.execute(task_instance.delete().where(task_instance.c.dag_id ==
DAG_ID))
+ connection.execute(trigger.delete().where(trigger.c.classpath ==
CLASSPATH))
+ connection.execute(dag_run.delete().where(dag_run.c.dag_id == DAG_ID))
+
+
[email protected]
+def database_before_revision():
+ """Move the database to the revision before the migration for one test,
then back to the current head."""
+ original_async_conn = settings.SQL_ALCHEMY_CONN_ASYNC
+ settings.SQL_ALCHEMY_CONN_ASYNC = ""
+ settings.dispose_orm(do_log=False)
+ settings.configure_orm(disable_connection_pool=True)
+ try:
+ downgrade(to_revision=PREVIOUS_REVISION)
+ with settings.engine.begin() as connection:
+ _delete_rows(connection)
+ yield settings.engine
+ finally:
+ try:
+ with settings.engine.begin() as connection:
+ _delete_rows(connection)
+ upgradedb()
+ finally:
+ settings.SQL_ALCHEMY_CONN_ASYNC = original_async_conn
+ settings.reconfigure_orm()
+
+
+def _add_trigger(connection) -> int:
+ return connection.execute(
+ trigger.insert().values(classpath=CLASSPATH, kwargs="{}",
created_date=timezone.utcnow())
+ ).inserted_primary_key[0]
+
+
+def _owners(engine) -> dict[int, tuple[str | None, dict | None]]:
+ table = sa.Table("trigger", sa.MetaData(), autoload_with=engine)
+ with engine.connect() as connection:
+ rows = connection.execute(
+ sa.select(table.c.id, table.c.dag_id,
table.c.owner_ref).where(table.c.classpath == CLASSPATH)
+ )
+ return {
+ trigger_id: (dag_id, json.loads(owner_ref) if
isinstance(owner_ref, str) else owner_ref)
+ for trigger_id, dag_id, owner_ref in rows
+ }
+
+
+def
test_upgrade_records_the_owner_of_existing_task_and_callback_triggers(database_before_revision):
+ ti_id, callback_id = uuid4(), uuid4()
+ with database_before_revision.begin() as connection:
+ task_trigger, callback_trigger, watcher_trigger =
(_add_trigger(connection) for _ in range(3))
+ connection.execute(
+ dag_run.insert().values(
+ dag_id=DAG_ID, run_id="run", run_type="manual",
run_after=timezone.utcnow()
+ )
+ )
+ connection.execute(
+ task_instance.insert().values(
+ id=ti_id,
+ dag_id=DAG_ID,
+ task_id="wait",
+ run_id="run",
+ region_index=-1,
+ region_id=UUID(int=0),
+ working_set=True,
+ try_number=1,
+ pool="default_pool",
+ pool_slots=1,
+ trigger_id=task_trigger,
+ )
+ )
+ # callback.data as ExtendedJSON stores it: every dict wrapped as
{"__type": "dict", "__var": ...}.
+ context = {"__type": "dict", "__var": {"dag_run": {"__type": "dict",
"__var": {"dag_id": DAG_ID}}}}
+ connection.execute(
+ callback.insert().values(
+ id=callback_id,
+ type="triggerer",
+ fetch_method="import_path",
+ data={
+ "__type": "dict",
+ "__var": {"kwargs": {"__type": "dict", "__var":
{"context": context}}},
+ },
+ priority_weight=1,
+ created_at=timezone.utcnow(),
+ trigger_id=callback_trigger,
+ )
+ )
+
+ upgradedb(to_revision=REVISION)
+
+ assert _owners(database_before_revision) == {
+ task_trigger: (DAG_ID, {"ti_id": str(ti_id)}),
+ callback_trigger: (DAG_ID, {"callback_id": str(callback_id)}),
+ watcher_trigger: (None, None),
+ }
+
+
+def test_downgrade_drops_the_new_columns(database_before_revision):
+ upgradedb(to_revision=REVISION)
+ assert {c["name"] for c in
inspect(database_before_revision).get_columns("trigger")} >= NEW_COLUMNS
+
+ downgrade(to_revision=PREVIOUS_REVISION)
+
+ assert not NEW_COLUMNS & {c["name"] for c in
inspect(database_before_revision).get_columns("trigger")}
diff --git a/airflow-core/tests/unit/models/test_callback.py
b/airflow-core/tests/unit/models/test_callback.py
index 4622616d04b..f1e5f884add 100644
--- a/airflow-core/tests/unit/models/test_callback.py
+++ b/airflow-core/tests/unit/models/test_callback.py
@@ -320,6 +320,23 @@ class TestTriggererCallback:
assert callback.trigger.kwargs["callback_kwargs"] ==
TEST_ASYNC_CALLBACK.kwargs
assert callback.trigger.queue is None
assert callback.state == CallbackState.QUEUED
+ assert callback.trigger.owner_ref == {"callback_id": str(callback.id)}
+
+ @pytest.mark.parametrize(
+ ("callback_kwargs", "data_dag_id"),
+ [
+ pytest.param({}, TEST_DAG_ID, id="dag-id-in-data"),
+ pytest.param({"context": {"dag_run": {"dag_id": TEST_DAG_ID}}},
None, id="dag-id-in-context"),
+ ],
+ )
+ def test_queue_records_the_owner_dag_on_the_trigger(self, session,
callback_kwargs, data_dag_id):
+ callback = TriggererCallback(TEST_ASYNC_CALLBACK)
+ callback.data["kwargs"] = callback_kwargs
+ callback.data["dag_id"] = data_dag_id
+
+ callback.queue(session=session)
+
+ assert callback.trigger.dag_id == TEST_DAG_ID
@pytest.mark.usefixtures("allow_tracked_kwarg")
@pytest.mark.parametrize("encoding", ["current", "legacy"])
diff --git a/airflow-core/tests/unit/models/test_taskinstance.py
b/airflow-core/tests/unit/models/test_taskinstance.py
index 6e8afa70ba1..04cbc99da90 100644
--- a/airflow-core/tests/unit/models/test_taskinstance.py
+++ b/airflow-core/tests/unit/models/test_taskinstance.py
@@ -3678,6 +3678,7 @@ def test_defer_task(create_task_instance):
assert isinstance(trigger_row, Trigger)
assert trigger_row.classpath == "trigger_cls"
assert trigger_row.kwargs == {"key": "value"}
+ assert (trigger_row.dag_id, trigger_row.owner_ref) == ("test_defer_task",
{"ti_id": str(ti.id)})
# Check that session.flush was called
session.flush.assert_called_once()
diff --git a/airflow-core/tests/unit/models/test_trigger.py
b/airflow-core/tests/unit/models/test_trigger.py
index ff12adf0f37..66a9fd6565f 100644
--- a/airflow-core/tests/unit/models/test_trigger.py
+++ b/airflow-core/tests/unit/models/test_trigger.py
@@ -26,6 +26,7 @@ import pendulum
import pytest
import pytz
import sqlalchemy as sa
+import time_machine
from cryptography.fernet import Fernet
from sqlalchemy import delete, event, func, insert, select, update
from sqlalchemy.exc import OperationalError
@@ -37,7 +38,7 @@ from airflow.models import TaskInstance, Trigger
from airflow.models.asset import AssetEvent, AssetModel, AssetWatcherModel
from airflow.models.callback import Callback, TriggererCallback
from airflow.models.dag_version import DagVersion
-from airflow.models.trigger import handle_event_submit
+from airflow.models.trigger import RecordedFailure, handle_event_submit
from airflow.models.xcom import XComModel
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.sdk.definitions.callback import AsyncCallback
@@ -440,6 +441,33 @@ def test_submit_event(mock_callback_handle_event, session,
create_task_instance)
mock_callback_handle_event.assert_called_once_with(event, session)
[email protected]("owner", ["task", "asset_watcher", "callback",
"none"])
+def test_submit_event_records_the_event_time_for_every_kind(session,
create_task_instance, owner):
+ t0 = timezone.datetime(2026, 10, 21, 12, 0, 0)
+ trigger = Trigger(classpath="airflow.triggers.testing.SuccessTrigger",
kwargs={})
+ session.add(trigger)
+ session.flush()
+ if owner == "task":
+ task_instance = create_task_instance(session=session, logical_date=t0,
state=State.DEFERRED)
+ task_instance.trigger_id = trigger.id
+ elif owner == "asset_watcher":
+ asset = AssetModel("event_time")
+ asset.add_trigger(trigger, "w")
+ session.add(asset)
+ elif owner == "callback":
+ callback =
TriggererCallback(callback_def=AsyncCallback("classpath.callback"))
+ callback.trigger = trigger
+ session.add(callback)
+ session.commit()
+
+ with time_machine.travel(t0, tick=False), patch.object(TriggererCallback,
"handle_event"):
+ Trigger.submit_event(trigger.id, TriggerEvent("payload"),
session=session)
+ session.commit()
+
+ session.expire_all()
+ assert session.get(Trigger, trigger.id).last_event_at == t0
+
+
@pytest.mark.parametrize(("asset_count", "expected_query_count"), [(1, 6), (5,
6)])
@patch("airflow.models.trigger.AssetManager.register_asset_change")
def test_submit_event_no_n_plus_one_for_assets(_, session, asset_count,
expected_query_count):
@@ -626,6 +654,110 @@ def test_submit_failure(
assert updated_task_instance.next_method == expected_next_method
+def _watcher_trigger(session, triggerer_id=1, **columns):
+ trigger = Trigger(classpath="airflow.triggers.testing.SuccessTrigger",
kwargs={})
+ trigger.triggerer_id = triggerer_id
+ for name, value in columns.items():
+ setattr(trigger, name, value)
+ session.add(trigger)
+ session.flush()
+ asset = AssetModel("watched")
+ asset.add_trigger(trigger, "watcher")
+ session.add(asset)
+ session.commit()
+ return trigger
+
+
+def _failure_state(session, trigger_id):
+ session.expire_all()
+ trigger = session.get(Trigger, trigger_id)
+ return (
+ trigger.consecutive_failures,
+ trigger.first_failed_at,
+ trigger.last_failed_at,
+ trigger.last_error,
+ )
+
+
[email protected](
+ ("caller_triggerer_id", "recorded"),
+ [
+ pytest.param(None, True, id="unguarded"),
+ pytest.param(1, True, id="owning-triggerer"),
+ pytest.param(2, False, id="previous-owner"),
+ ],
+)
+def test_submit_failure_records_the_failure_on_the_row(session,
caller_triggerer_id, recorded):
+ t0 = timezone.datetime(2026, 10, 21, 12, 0, 0)
+ trigger = _watcher_trigger(session)
+
+ with time_machine.travel(t0, tick=False):
+ Trigger.submit_failure(
+ trigger.id,
+ triggerer_id=caller_triggerer_id,
+ failure=RecordedFailure(1, "ConnectionError: expired"),
+ session=session,
+ )
+ with time_machine.travel(t0 + datetime.timedelta(seconds=6), tick=False):
+ Trigger.submit_failure(
+ trigger.id,
+ triggerer_id=caller_triggerer_id,
+ failure=RecordedFailure(2, "ConnectionError: still"),
+ session=session,
+ )
+ session.commit()
+
+ expected = (
+ 2,
+ t0, # the start of the streak is kept
+ t0 + datetime.timedelta(seconds=6),
+ "ConnectionError: still",
+ )
+ assert _failure_state(session, trigger.id) == (expected if recorded else
(None,) * 4)
+
+
+def test_a_trigger_without_an_owner_stores_sql_null(session):
+ # An asset watcher's trigger: its owners are the asset_watcher rows, and
rows the migration did not fill
+ # in are NULL too.
+ session.add(Trigger(classpath="airflow.triggers.testing.SuccessTrigger",
kwargs={}))
+ session.commit()
+
+ assert
session.scalar(select(func.count()).select_from(Trigger).where(Trigger.owner_ref.is_(None)))
== 1
+
+
+def test_submit_failure_starts_a_new_streak_at_the_first_failure(session):
+ t0 = timezone.datetime(2026, 10, 21, 12, 0, 0)
+ trigger = _watcher_trigger(session, first_failed_at=t0)
+
+ with time_machine.travel(t0 + datetime.timedelta(hours=1), tick=False):
+ Trigger.submit_failure(trigger.id, failure=RecordedFailure(1, "E: x"),
session=session)
+ session.commit()
+
+ assert _failure_state(session, trigger.id)[1] == t0 +
datetime.timedelta(hours=1)
+
+
[email protected](
+ ("caller_triggerer_id", "cleared"),
+ [pytest.param(1, True, id="owning-triggerer"), pytest.param(2, False,
id="previous-owner")],
+)
+def test_clear_recorded_failures_keeps_the_last_error(session,
caller_triggerer_id, cleared):
+ t0 = timezone.datetime(2026, 10, 21, 12, 0, 0)
+ failing = {
+ "consecutive_failures": 3,
+ "first_failed_at": t0,
+ "last_failed_at": t0,
+ "last_error": "E: x",
+ }
+ trigger = _watcher_trigger(session, **failing)
+
+ Trigger.clear_recorded_failures({trigger.id},
triggerer_id=caller_triggerer_id, session=session)
+ session.commit()
+
+ assert _failure_state(session, trigger.id) == (
+ (None, None, t0, "E: x") if cleared else (3, t0, t0, "E: x")
+ )
+
+
@pytest.mark.parametrize(
("event_cls", "expected"),
[
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 5758848964f..1f698ad2daa 100644
---
a/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
+++
b/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
@@ -372,13 +372,15 @@ metrics:
name_variables: []
- name: "triggers.failed"
- description: "Number of triggers that errored before they could fire an
event"
+ description: "Number of triggers that errored before they could fire an
event.
+ Tagged with trigger_kind (asset_watcher, callback or task)."
type: "counter"
legacy_name: "-"
name_variables: []
- name: "triggers.succeeded"
- description: "Number of triggers that have fired at least one event"
+ description: "Number of triggers that have fired at least one event.
+ Tagged with trigger_kind (asset_watcher, callback or task)."
type: "counter"
legacy_name: "-"
name_variables: []
@@ -679,6 +681,13 @@ metrics:
legacy_name: "triggerer.capacity_left.{hostname}"
name_variables: ["hostname"]
+ - name: "triggers.failing"
+ description: "Triggers currently failing on a triggerer, that is failed
since they last recovered.
+ One value per trigger_kind (asset_watcher, callback or task)."
+ type: "gauge"
+ legacy_name: "-"
+ name_variables: []
+
- name: "ti.scheduled"
description: "Number of scheduled tasks in a given Dag."
type: "gauge"
