seanghaeli opened a new pull request, #76101: URL: https://github.com/apache/airflow/pull/76101
A trigger that keeps failing leaves nothing in the database, so no API, UI or alert can tell, and the row doesn't say who the trigger runs for. `triggers.failed` can't tell a watcher from a task's trigger either. Part of the triggerer observability proposal on dev@ ([thread](https://lists.apache.org/thread/klcsojt3740xf710qfmshfsz22qkq7yr)). The follow-up PR for `GET /api/v2/triggers` is built on these columns. - Each failure writes `consecutive_failures`, `first_failed_at`, `last_failed_at` and `last_error` to the trigger's row. `last_error` is the exception type and message only, masked like task logs and cut to 1,000 characters. An event sets `last_event_at`. - The count resets when the trigger fires an event or stays up for two minutes (`Trigger recovered` is logged). A triggerer that takes it over keeps counting. - Task and callback triggers record `dag_id` and `owner_ref` when created; the migration adds nullable columns and fills the owner in for existing rows, and the triggerer fills it in for rows older components create during a rolling upgrade. - `triggers.failed` and `triggers.succeeded` get a `trigger_kind` tag, and a new `triggers.failing` gauge counts failing triggers per triggerer and kind. - Restarts are unchanged (backoff is a separate PR): one row update per failure. The migration is 0152 on `38c8df3fcb9c`. #75228, #74790 and #75673 also add a 0152, so the number moves if one of them merges first. How to review: `TriggerFailureTracker` in `jobs/triggerer_job_runner.py`, then the row writes, `Trigger.submit_failure` and `clear_recorded_failures` in `models/trigger.py`. Breeze on Postgres, a watcher with expired credentials next to healthy watchers, deferred tasks and a deadline callback: ```text # before: main ddb89d5e00 === A) [09:32:41] creds watcher failing, re-created as on main --- row: (id, consecutive_failures, first_failed_at, last_failed_at, last_event_at, last_error) (trigger table has no state columns) creds instance starts: 12 (bad=True 12); gaps s: [2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0] triggers.failing (latest per kind): asset_watcher=<none> callback=<none> task=<none> triggers.failed lines by tag: 33 triggers.failed; === deferred task trigger that raised: wait_fail ti state=failed, trigger instances started=1, last poll=TO-TASKWAIT wait_fail poll 20, task triggers left on rows: 0 triggers.failing (latest per kind): asset_watcher=<none> callback=<none> task=<none> triggers.failed lines by tag: 37 triggers.failed; === B) [09:32:49] credentials fixed and the watcher fires: recovery by an event --- row: (trigger table has no state columns) creds instance starts: 1 (bad=True 0); gaps s: [] triggers.failing (latest per kind): asset_watcher=<none> callback=<none> task=<none> triggers.failed lines by tag: 38 triggers.failed; === C) [09:33:53] fails again, then credentials fixed with no event: recovery by staying up # after: this PR 9b015af7ae === A) [09:39:21] creds watcher failing, re-created as on main --- row: (id, consecutive_failures, first_failed_at, last_failed_at, last_event_at, last_error) (1, 36, '09:38:16', '09:39:27', None, 'ConnectionError: TO creds: the security token included in the request is expired') creds instance starts: 16 (bad=True 16); gaps s: [2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0] ... triggers.failing (latest per kind): asset_watcher=1 callback=0 task=0 triggers.failed lines by tag: 37 triggers.failed,trigger_kind=asset_watcher; 1 triggers.failed,trigger_kind=task; === deferred task trigger that raised: wait_fail ti state=failed, trigger instances started=1, last poll=TO-TASKWAIT wait_fail poll 20, task triggers left on rows: 0 triggers.failing (latest per kind): asset_watcher=1 callback=0 task=0 triggers.failed lines by tag: 40 triggers.failed,trigger_kind=asset_watcher; 1 triggers.failed,trigger_kind=task; === B) [09:39:37] credentials fixed and the watcher fires: recovery by an event --- row: (1, None, None, '09:39:37', '09:39:39', 'ConnectionError: TO creds: the security token included in the request is expired') creds instance starts: 1 (bad=True 0); gaps s: [] triggers.failing (latest per kind): asset_watcher=0 callback=0 task=0 triggers.failed lines by tag: 41 triggers.failed,trigger_kind=asset_watcher; 1 triggers.failed,trigger_kind=task; === C) [09:39:51] fails again, then credentials fixed with no event: recovery by staying up --- [09:40:03] failing again: row: failures=7 gauge(asset_watcher)=1 --- row: (1, 8, '09:39:54', '09:40:08', '09:39:39', 'ConnectionError: TO creds: the security token included in the request is expired') [09:40:10] credentials fixed, no event [09:40:12] re-created with good credentials, up from about 2026-10-29T09:40:12.039683 --- [09:41:12] up ~60 s, no event yet: row: failures=9 gauge(asset_watcher)=1 --- [09:42:15] up ~120 s: row: failures=None gauge(asset_watcher)=0 --- row: (1, None, None, '09:40:10', '09:39:39', 'ConnectionError: TO creds: the security token included in the request is expired') --- triggerer output: recovery lines 2026-10-29T09:39:39.465335Z [info ] Trigger recovered [airflow.jobs.triggerer_job_runner] after_failures=41 loc=triggerer_job_runner.py:961 trigger_id=1 trigger_kind=asset_watcher 2026-10-29T09:42:11.787707Z [info ] Trigger recovered [airflow.jobs.triggerer_job_runner] after_failures=9 loc=triggerer_job_runner.py:961 trigger_id=1 trigger_kind=asset_watcher # rows created without an owner during a rolling upgrade (owner emptied as an older worker/scheduler writes it), with GET /triggers on top: before 273adf05a5 --- [09:21:24] rows as an older component writes them (owner emptied), no triggerer yet (1, None, False) (2, None, False) (3, None, False) --- [09:21:51] a triggerer picked them up: (id, dag_id, has owner_ref) (1, None, False) (2, None, False) (3, None, False) GET /api/v2/triggers?kind=task -> HTTP 200 ids=[] GET /api/v2/triggers?kind=callback -> HTTP 200 ids=[] GET /api/v2/triggers?dag_id=to_kinds_tasks -> HTTP 200 ids=[] GET /api/v2/triggers?dag_id=to_kinds_deadline -> HTTP 200 ids=[] GET /api/v2/triggers/2 (deferred task) -> HTTP 404 # after: this PR, with GET /triggers on top 014ac45834 --- [09:23:46] rows as an older component writes them (owner emptied), no triggerer yet (1, None, False) (2, None, False) (3, None, False) --- [09:24:13] a triggerer picked them up: (id, dag_id, has owner_ref) (1, None, False) (2, 'to_kinds_tasks', True) (3, 'to_kinds_deadline', True) GET /api/v2/triggers?kind=task -> HTTP 200 ids=[2] GET /api/v2/triggers?kind=callback -> HTTP 200 ids=[3] GET /api/v2/triggers?dag_id=to_kinds_tasks -> HTTP 200 ids=[2] GET /api/v2/triggers?dag_id=to_kinds_deadline -> HTTP 200 ids=[3] GET /api/v2/triggers/2 (deferred task) -> HTTP 200 ``` --- ##### Was generative AI tooling used to co-author this PR? <!-- If generative AI tooling has been used in the process of authoring this PR, please change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by". --> - [X] Yes (please specify the tool below) Generated-by: Claude Code (Opus 5.5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) -- 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]
