o-nikolas commented on code in PR #76260:
URL: https://github.com/apache/airflow/pull/76260#discussion_r4448689964
##########
airflow-core/src/airflow/api_fastapi/core_api/security.py:
##########
@@ -1221,6 +1222,52 @@ def depends_permitted_assets_filter(
]
+def _triggers_watching(asset_ids: set[int]) -> Select:
+ # Uncorrelated, as in Trigger.owner_still_refers_expression.
+ return
select(AssetWatcherModel.trigger_id).where(AssetWatcherModel.asset_id.in_(asset_ids))
+
+
+class PermittedTriggerFilter(OrmClause[None]):
+ """
+ Keep the triggers whose owner the caller can see, by one rule for every
kind of trigger.
+
+ A trigger's owner is the Dag recorded on it (the Dag of its task instance
or callback) or the assets its
+ watchers watch. Team access is the auth manager's, through those Dags and
assets.
+ """
+
+ def __init__(self, *, dag_ids: set[str], asset_ids: set[int]) -> None:
+ super().__init__()
+ self.dag_ids = dag_ids
+ self.asset_ids = asset_ids
+
+ def to_orm(self, select: Select) -> Select:
+ return select.where(
+ or_(Trigger.dag_id.in_(self.dag_ids),
Trigger.id.in_(_triggers_watching(self.asset_ids)))
+ )
+
+
+def get_readable_asset_ids(user: BaseUser, auth_manager: BaseAuthManager) ->
set[int]:
+ """
+ Return the ids of the assets the user may read, gated like ``GET /assets``
first.
+
+ ``get_authorized_assets`` may assume that gate has passed: the FAB auth
manager, which has no per-asset
+ permissions, returns every asset there.
+ """
+ if not auth_manager.is_authorized_asset(method="GET",
details=AssetDetails(id=None), user=user):
+ return set()
+ return auth_manager.get_authorized_assets(user=user, method="GET")
+
+
+def readable_triggers_filter(user: GetUserDep, auth_manager: AuthManagerDep)
-> PermittedTriggerFilter:
Review Comment:
This only checks Dag read. A task trigger returns task instance data (state,
try number) and last_error, which elsewhere need Task Instance read: the TI
list route uses requires_access_dag("GET", TASK_INSTANCE) (task_instances.py
L559). Callback data is gated on RUN in routes/ui/deadlines.py L81. So a role
with Dag read but no Task Instance read can now see it through /triggers. Could
you gate task triggers on TASK_INSTANCE and callback triggers on RUN, like
those routes do? That would also stop the permission tables listing these
routes as "Public" (api_permissions_ref.rst L594/L598, FAB access-control.rst
L767/L771).
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/triggers.py:
##########
@@ -0,0 +1,275 @@
+# 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.
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Annotated
+
+from fastapi import Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.orm import contains_eager, defer, selectinload
+
+from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
+from airflow.api_fastapi.common.parameters.base import QueryLimit, QueryOffset
+from airflow.api_fastapi.common.parameters.sort import SortParam
+from airflow.api_fastapi.common.parameters.trigger import (
+ QueryTriggerAssetIdFilter,
+ QueryTriggerDagIdFilter,
+ QueryTriggererHostnameFilter,
+ QueryTriggerKindFilter,
+ QueryTriggerStatusFilter,
+ QueryTriggerTeamsFilter,
+)
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.trigger import (
+ TriggerAssetWatcherReferenceResponse,
+ TriggerCallbackReferenceResponse,
+ TriggerCollectionResponse,
+ TriggerDetailResponse,
+ TriggerTaskInstanceReferenceResponse,
+)
+from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.security import ReadableTriggersFilterDep
+from airflow.models.asset import AssetWatcherModel
+from airflow.models.dagrun import DagRun
+from airflow.models.deadline import Deadline
+from airflow.models.dynamic_region import SENTINEL_REGION_ID
+from airflow.models.task_coordinates import public_map_index_expression
+from airflow.models.taskinstance import TaskInstance
+from airflow.models.trigger import Trigger, TriggerKind
+
+if TYPE_CHECKING:
+ from collections.abc import Iterable
+ from uuid import UUID
+
+ from sqlalchemy.orm import Session
+ from sqlalchemy.sql import ColumnElement, Select
+
+triggers_router = AirflowRouter(tags=["Trigger"], prefix="/triggers")
+
+
+def _sort_null_as_smallest(column: ColumnElement) -> list[ColumnElement]:
+ # Postgres sorts NULL as the largest value, MySQL and SQLite as the
smallest; without the leading key a
+ # descending sort would list the triggers that never failed (or never
fired) first on Postgres only.
+ return [column.is_not(None).label(f"has_{column.key}"), column]
+
+
+def _triggers_select() -> Select:
+ # Only triggers their owner still refers to: the rest are deleted on the
triggerer's next cleanup. The
+ # kwargs are never returned, so they are not even loaded.
+ return (
+ select(Trigger)
+ .outerjoin(Trigger.triggerer_job)
+ .where(Trigger.owner_still_refers_expression())
+ .options(
+ defer(Trigger.encrypted_kwargs),
+ contains_eager(Trigger.triggerer_job),
+
selectinload(Trigger.asset_watchers).joinedload(AssetWatcherModel.asset),
+ selectinload(Trigger.callback),
+ )
+ )
+
+
+def _get_callback_dag_run_ids(triggers: Iterable[Trigger], *, session:
Session) -> dict[UUID, str]:
+ """Return the run id of each callback's Dag run; a callback names its run
only through its deadline."""
+ callback_ids = [trigger.callback.id for trigger in triggers if
trigger.callback is not None]
+ if not callback_ids:
+ return {}
+ rows = session.execute(
+ select(Deadline.callback_id, DagRun.run_id)
+ .join(Deadline.dagrun)
+ .where(Deadline.callback_id.in_(callback_ids))
+ )
+ return {callback_id: run_id for callback_id, run_id in rows}
+
+
+def _get_public_map_indexes(triggers: Iterable[Trigger], *, session: Session)
-> dict[UUID, int]:
+ """Return the public map index of each trigger's task instance; one query
for those in a dynamic region."""
+ task_instances = [trigger.task_instance for trigger in triggers if
trigger.task_instance is not None]
+ map_indexes = {ti.id: ti.region_index for ti in task_instances if
ti.region_id == SENTINEL_REGION_ID}
+ if regional := [ti.id for ti in task_instances if ti.id not in
map_indexes]:
+ rows = session.execute(
+ select(TaskInstance.id, public_map_index_expression(TaskInstance))
+ .where(TaskInstance.id.in_(regional))
+ .execution_options(include_all_attempts=True)
+ )
+ map_indexes.update({ti_id: map_index for ti_id, map_index in rows})
+ return map_indexes
+
+
+def _build_response(
+ trigger: Trigger,
+ *,
+ readable_asset_ids: set[int],
+ callback_dag_run_ids: dict[UUID, str],
+ public_map_indexes: dict[UUID, int],
+) -> TriggerDetailResponse | None:
+ """
+ Describe a trigger and its owner, the same way for every kind; ``None`` if
the owner just let go.
+
+ A trigger can serve watchers on several assets; only those on assets the
caller can read are listed.
+ """
+ if (kind := trigger.kind) is None:
+ return None
+ ti = trigger.task_instance if kind == TriggerKind.TASK else None
+ callback = trigger.callback if kind == TriggerKind.CALLBACK else None
+ job = trigger.triggerer_job
+ return TriggerDetailResponse(
+ id=trigger.id,
+ kind=kind,
+ classpath=trigger.classpath,
+ created_date=trigger.created_date,
+ queue=trigger.queue,
+ team_name=trigger.team_name,
+ status=trigger.status(),
+ triggerer_hostname=job.hostname if job is not None else None,
+ consecutive_failures=trigger.consecutive_failures or 0,
+ first_failed_at=trigger.first_failed_at,
+ last_failed_at=trigger.last_failed_at,
+ last_error=trigger.last_error,
+ last_event_at=trigger.last_event_at,
+ dag_id=trigger.dag_id,
+ task_instance=None
+ if ti is None
+ else TriggerTaskInstanceReferenceResponse(
+ dag_id=ti.dag_id,
+ dag_run_id=ti.run_id,
+ task_id=ti.task_id,
+ map_index=public_map_indexes[ti.id],
Review Comment:
nit: `public_map_indexes[ti.id]` raises `KeyError`, and so a 500, if the
task instance is deleted between the two queries. `.get()` with a fallback
avoids that.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/triggers.py:
##########
@@ -0,0 +1,275 @@
+# 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.
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Annotated
+
+from fastapi import Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.orm import contains_eager, defer, selectinload
+
+from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
+from airflow.api_fastapi.common.parameters.base import QueryLimit, QueryOffset
+from airflow.api_fastapi.common.parameters.sort import SortParam
+from airflow.api_fastapi.common.parameters.trigger import (
+ QueryTriggerAssetIdFilter,
+ QueryTriggerDagIdFilter,
+ QueryTriggererHostnameFilter,
+ QueryTriggerKindFilter,
+ QueryTriggerStatusFilter,
+ QueryTriggerTeamsFilter,
+)
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.trigger import (
+ TriggerAssetWatcherReferenceResponse,
+ TriggerCallbackReferenceResponse,
+ TriggerCollectionResponse,
+ TriggerDetailResponse,
+ TriggerTaskInstanceReferenceResponse,
+)
+from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.security import ReadableTriggersFilterDep
+from airflow.models.asset import AssetWatcherModel
+from airflow.models.dagrun import DagRun
+from airflow.models.deadline import Deadline
+from airflow.models.dynamic_region import SENTINEL_REGION_ID
+from airflow.models.task_coordinates import public_map_index_expression
+from airflow.models.taskinstance import TaskInstance
+from airflow.models.trigger import Trigger, TriggerKind
+
+if TYPE_CHECKING:
+ from collections.abc import Iterable
+ from uuid import UUID
+
+ from sqlalchemy.orm import Session
+ from sqlalchemy.sql import ColumnElement, Select
+
+triggers_router = AirflowRouter(tags=["Trigger"], prefix="/triggers")
+
+
+def _sort_null_as_smallest(column: ColumnElement) -> list[ColumnElement]:
+ # Postgres sorts NULL as the largest value, MySQL and SQLite as the
smallest; without the leading key a
+ # descending sort would list the triggers that never failed (or never
fired) first on Postgres only.
+ return [column.is_not(None).label(f"has_{column.key}"), column]
+
+
+def _triggers_select() -> Select:
Review Comment:
nit: AGENTS.md asks for verb names for functions:
- `_triggers_select` -> `_build_triggers_select`
- `_triggers_watching` -> `_select_triggers_watching`
- `triggerer_alive_clauses` -> `get_triggerer_alive_clauses`
##########
airflow-core/docs/administration-and-deployment/logging-monitoring/triggers.rst:
##########
@@ -0,0 +1,115 @@
+.. 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.
+
+Monitoring triggers
+===================
+
+The triggerer runs three kinds of trigger: the trigger of a :doc:`deferred
task </authoring-and-scheduling/deferring>`,
+a triggerer callback (for example a deadline alert's ``AsyncCallback``), and an
+:doc:`asset watcher </authoring-and-scheduling/event-scheduling>`. ``GET
/api/v2/triggers`` lists all of them the same
+way: what each one runs for, which triggerer runs it, and how it is doing.
Trigger arguments are never returned.
+
+Status
+------
+
+``pending``
+ Not picked up by a triggerer yet.
+
+``running``
+ On a live triggerer and not failed since it last recovered. For an asset
watcher this does not mean the
+ source it watches is healthy, only that the trigger is not raising;
``last_event_at`` says when it last fired.
+
+``failing``
+ On a live triggerer, and it raised or ended without yielding an event
since it last recovered.
+ ``consecutive_failures``, ``first_failed_at`` and ``last_error`` say how
often, since when and why. A trigger
+ that is still wanted, such as an asset watcher, is re-created within
seconds and stays ``failing`` until it
+ recovers: it yields an event, or stays up for two minutes.
+
+``orphaned``
+ Its triggerer stopped sending heartbeats. Another triggerer takes it over
within
+ ``[triggerer] triggerer_health_check_threshold`` seconds.
+
+``triggerer_hostname`` is the host of the triggerer the trigger is assigned
to; each live triggerer's heartbeat is
+on the health endpoint. The failure count is kept on the trigger, so it
carries over when the triggerer restarts
+or another triggerer takes the trigger over. ``last_error`` is kept after a
recovery.
+
+A trigger is listed only while its owner still refers to it. Once a deferred
task resumes or fails, a callback
+finishes, or a watcher is removed, the trigger is deleted on the triggerer's
next cleanup, and its outcome stays
+with its owner: the task instance's state and logs, the callback's state and
logs.
+
+During a rolling upgrade from a version that does not record trigger owners, a
deferred task's or callback's
+trigger created by the older components is listed only once a triggerer of
this version picks it up and records its
+owner, and a trigger still run by an older triggerer is never ``failing``,
since older triggerers do not record
+failures.
+
+Owner
+-----
+
+Each trigger names what it runs for: ``task_instance`` (Dag, run, task, map
index, try and state), ``callback``
+(its id, Dag, run and state), or ``asset_watchers`` (each asset and watcher
name; several assets that declare the
+same trigger share one). ``kind`` is worked out from the owner, and ``dag_id``
is the Dag of the task instance, or
+of the Dag run whose deadline the callback belongs to. In the other direction,
a task instance's ``trigger`` and an
+asset's ``watchers`` (with ``status`` and ``trigger_id``) link to the trigger.
+
+Filtering
+---------
+
+``kind`` (``task``, ``callback`` or ``asset_watcher``), ``status``,
``triggerer_hostname``, ``dag_id``,
+``asset_id`` and ``teams`` can be combined; ``kind``, ``status`` and ``teams``
take several values. For example,
+to find the triggers that need attention:
+
+.. code-block:: bash
+
+ curl -H "Authorization: Bearer $TOKEN" \
+ "$AIRFLOW_URL/api/v2/triggers?status=failing&status=orphaned"
+
+and an asset's watchers, with their status, come with the asset itself:
+
+.. code-block:: bash
+
+ curl -H "Authorization: Bearer $TOKEN" "$AIRFLOW_URL/api/v2/assets/12"
+
+``order_by`` takes ``id`` (the default), ``created_date``,
``consecutive_failures``, ``last_failed_at`` and
+``last_event_at``, ``-`` first for descending. A trigger that never failed or
never fired sorts first ascending and
+last descending, on every database.
+
+``GET /api/v2/triggers/{trigger_id}`` returns one trigger.
+
+Who can see a trigger
+---------------------
+
+A trigger is listed for whoever can see its owner: read access to the Dag of
its task instance or callback, or
+read access to an asset one of its watchers watches. There is no permission of
its own, so the generated
+permission tables show the routes as needing none. With ``[core] multi_team``
on, team access comes with the
+Dag; assets have no team, so an asset's watchers are listed for whoever can
read the asset. A trigger shared
Review Comment:
This only checks Dag read. A task trigger returns task instance data (state,
try number) and last_error, which elsewhere need Task Instance read: the TI
list route uses requires_access_dag("GET", TASK_INSTANCE) (task_instances.py
L559). Callback data is gated on RUN in routes/ui/deadlines.py L81. So a role
with Dag read but no Task Instance read can now see it through /triggers. Could
you gate task triggers on TASK_INSTANCE and callback triggers on RUN, like
those routes do? That would also stop the permission tables listing these
routes as "Public" (api_permissions_ref.rst L594/L598, FAB access-control.rst
L767/L771).
--
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]
