This is an automated email from the ASF dual-hosted git repository. dheerajturaga pushed a commit to branch aip-125/38-cancel-on-ui-docs in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 7681f32151de02d5fe7246262728b15c632bf0b0 Author: Dheeraj Turaga <[email protected]> AuthorDate: Sat Oct 24 04:43:29 2026 -0500 AIP-125: Document cancelling runs and tasks from webhooks Dags and tasks can declare cancel_on, deliveries record cancel requests, the scheduler applies them, and the UI, the REST API and airflow webhooks check show them, but no page told a Dag author how to declare a condition or what a cancel does, and the Deployment Manager's page still described the outcomes, tables, options, grants, audit entries and metrics of a webhook without them. Several behaviours of cancel_on surprise anyone who has not read the code: only a delivery with a correlation key cancels; a run is about a key only through an event that a delivery to the same webhook recorded, or through the conf key that a condition names; a delivery cancels only the runs whose event, or whose queueing with the key in their conf, came before it; a cancel is a failure, so the Dag's on_failure_callback runs with reason cancelled and the run counts toward max_consecutive_failed_dag_runs when a later run fails; a teardown keeps its run alive and decides how it ends; requests are also applied to runs that the scheduler creates later from earlier events, which is why they are kept until airflow db clean removes them; and the conditions come from the Dag's latest version while each run's own version decides its tasks. A new page for Dag authors states these, with a preview environment example, and the webhooks page gains a section on why a run was or was not cancelled. A delivery that records cancel requests is kept for webhook waits and processed again when redelivered, so the waits, topics and webhooks pages, which said that only early deliveries are kept, and the provider pages that repeat it would now mislead. The security model lists what a leaked secret or a chosen correlation key can cancel, and the public interface, audit log catalog and callback docs name the new outcome, fields, event, metric and callback reason. The release note covers cancel_on because upgrading adds two tables and needs the Task SDK on the Dag processor and the workers upgraded with core. --- .../logging-monitoring/callbacks.rst | 7 + .../administration-and-deployment/webhooks.rst | 332 ++++++++++++---- .../authoring-and-scheduling/asset-scheduling.rst | 3 +- .../docs/authoring-and-scheduling/index.rst | 1 + .../authoring-and-scheduling/webhook-cancel.rst | 442 +++++++++++++++++++++ .../authoring-and-scheduling/webhook-topics.rst | 29 +- .../authoring-and-scheduling/webhook-waits.rst | 64 +-- airflow-core/docs/public-airflow-interface.rst | 33 +- airflow-core/docs/security/audit_logs.rst | 8 +- airflow-core/docs/security/security_model.rst | 20 +- .../newsfragments/aip125-cancel-on.significant.rst | 9 + providers/github/docs/webhooks.rst | 20 +- .../standard/docs/connections/inbound_webhook.rst | 12 +- providers/standard/docs/operators/webhooks.rst | 13 +- 14 files changed, 844 insertions(+), 149 deletions(-) diff --git a/airflow-core/docs/administration-and-deployment/logging-monitoring/callbacks.rst b/airflow-core/docs/administration-and-deployment/logging-monitoring/callbacks.rst index 9b15c8f6c09..b8220e92319 100644 --- a/airflow-core/docs/administration-and-deployment/logging-monitoring/callbacks.rst +++ b/airflow-core/docs/administration-and-deployment/logging-monitoring/callbacks.rst @@ -87,6 +87,13 @@ and the task selected depends on the state of a Dag: #. On Dag run timeout, the latest started but not finished task is passed. #. If tasks are deadlocked, a task that should have run next but couldn't is passed. #. On success, the latest succeeded task is passed. +#. When an :doc:`inbound webhook delivery </authoring-and-scheduling/webhook-cancel>` cancels the run, the + latest started of the tasks that the cancel stopped is passed. If it stopped none, the context has only + ``dag``, ``run_id`` and ``reason``. + +The context of a Dag callback also has ``reason``, which says why the callback runs: ``success``, +``task_failure``, ``timed_out``, ``all_tasks_deadlocked``, or ``cancelled`` when an inbound webhook delivery +cancelled the run through the Dag's ``cancel_on``. It's not recommended to rely on task instance variables in Dag callbacks except for human analysis, as they reflect only partial information about the Dag's state. For example, a timeout may be caused by a number of stalling tasks, but only one will eventually be selected for context. diff --git a/airflow-core/docs/administration-and-deployment/webhooks.rst b/airflow-core/docs/administration-and-deployment/webhooks.rst index 4705dd149d7..662a1bb71f0 100644 --- a/airflow-core/docs/administration-and-deployment/webhooks.rst +++ b/airflow-core/docs/administration-and-deployment/webhooks.rst @@ -41,11 +41,15 @@ records an asset event on each of the delivery's *topic names* that a Dag schedu starts a run of each such Dag. An *asset route* in the webhook's connection can also have a delivery record an event on a data asset, such as a table that the sender has loaded (see `Asset routes`_). A task can also wait in its Dag run for a delivery that matches it, such as one saying that a pull request it -opened was merged: the delivery then resumes the task, through the task's *webhook wait*. +opened was merged: the delivery then resumes the task, through the task's *webhook wait*. And a Dag or a +task can declare, with ``cancel_on``, which deliveries cancel its runs, such as the runs that deploy a +preview of a pull request once the pull request is closed: such a delivery records a *cancel request*, +which the scheduler applies to the runs about the same pull request. This page is for Deployment Managers. Dag authors subscribe to deliveries as described in -:doc:`/authoring-and-scheduling/webhook-topics`, and wait for them as described in -:doc:`/authoring-and-scheduling/webhook-waits`. Provider authors write source types as described in the +:doc:`/authoring-and-scheduling/webhook-topics`, wait for them as described in +:doc:`/authoring-and-scheduling/webhook-waits`, and have them cancel runs as described in +:doc:`/authoring-and-scheduling/webhook-cancel`. Provider authors write source types as described in the Webhook Sources page of the providers documentation. Setting up a webhook @@ -175,7 +179,8 @@ The :ref:`[webhooks] <config:webhooks>` section of the configuration reference d - Largest body a webhook accepts, unless its connection sets its own ``max_body_bytes``. * - ``delivery_retention`` - ``7`` - - Days a delivery's record is kept before ``airflow db clean`` may remove it, at most 365. + - Days a delivery's record, and the cancel requests it recorded, are kept before ``airflow db clean`` + may remove them, at most 365. * - ``rate_limit_per_minute`` - ``0`` - Verified deliveries per minute that each process takes for each webhook; ``0`` is no limit. @@ -187,14 +192,23 @@ The :ref:`[webhooks] <config:webhooks>` section of the configuration reference d - Largest payload copied into a topic event's extra; ``0`` copies every payload. * - ``early_delivery_ttl`` - ``600`` - - Seconds after its receipt that an ``early`` delivery can still go to a webhook wait registered later. + - Seconds after its receipt that a delivery kept for webhook waits, ``early`` or ``cancel_requested``, + can still go to a wait registered later. * - ``early_delivery_max_per_webhook`` - ``1000`` - - Most ``early`` deliveries within ``early_delivery_ttl`` that a webhook keeps; older ones expire. + - Most deliveries kept for webhook waits, ``early`` or ``cancel_requested``, within + ``early_delivery_ttl`` that a webhook keeps; older ones expire. * - ``default_wait_timeout`` - empty - Seconds a webhook wait registered without a timeout lasts, at most 31536000 (365 days). When it is empty, such a wait is refused. + * - ``cancel_requests_per_loop`` + - ``20`` + - Most webhook cancel requests the scheduler applies each time it looks for them, about once a second. + Each request is applied and committed on its own, and the rest wait for the next time. At least 1. + +The scheduler reads ``cancel_requests_per_loop``, at startup, and does not start if it is below 1; the +webhooks process does not read it. The API servers that serve the Execution API read ``[webhooks]`` too, when a task registers a webhook wait: they refuse a wait on a webhook that their ``names`` does not list, or whose connection they cannot @@ -243,11 +257,12 @@ change it in the connection and then in the sender, within ``connection_cache_tt first delivery signed with the new secret then usually passes, and deliveries signed with the old one pass until the cached connection is read again or expires. A failure to read a secrets backend is never cached. -Webhooks have no team in Airflow 3.5. ``[webhooks] names`` refuses a name qualified with a team, such as -``team:name``, the connection is looked up without a team, so a connection that belongs to a team is not -found, and when :ref:`config:core__multi_team` is on, topic events queue only Dags that have no team, an -asset route records nothing for an asset that a Dag of a team schedules on, and a task whose Dag has a -team cannot register a webhook wait. +Webhooks have no team in this version of Airflow. ``[webhooks] names`` refuses a name qualified with a team, +such as ``team:name``, the connection is looked up without a team, so a connection that belongs to a team is +not found, and when :ref:`config:core__multi_team` is on, topic events queue only Dags that have no team, an +asset route records nothing for an asset that a Dag of a team schedules on, a task whose Dag has a team cannot +register a webhook wait, and no delivery records a cancel request for a Dag that has a team (see +:doc:`/authoring-and-scheduling/webhook-cancel`). Source types ------------ @@ -294,7 +309,9 @@ refuses it: 6. The delivery must be within ``[webhooks] rate_limit_per_minute``. 7. The source type parses the delivery. 8. One transaction records the delivery, its topic events, the events of its asset routes and, if it - recorded any event, an audit log entry, and hands the delivery to the + has a correlation key, a cancel request for each Dag whose ``cancel_on`` it matches (see + :doc:`/authoring-and-scheduling/webhook-cancel`), with an audit log entry if it recorded any event or + cancel request, and hands the delivery to the :doc:`webhook waits </authoring-and-scheduling/webhook-waits>` that it matches, with an audit log entry for each (see `Outcomes and repeated deliveries`_). 9. The response is sent once that transaction has committed. @@ -358,13 +375,15 @@ before sending the whole body. Every request under ``/webhooks`` other than ``/w an access log line, refused or not, so the access log grows with every request that reaches the process. The delivery contract follows from this order. A 202 means that the delivery's record, whatever topic -events, asset route events and audit log entries it recorded, and what it did for webhook waits, -committed before the answer was sent; a 200 ``duplicate`` means that an earlier copy's did; a 200 -handshake records nothing. Asset listeners and metrics run after the commit. A 5xx records nothing, so a -sender that retries can deliver again once the cause is fixed. The one exception is an ``early`` delivery, -which looks for webhook waits once more after its record has committed: if that second look fails, the -delivery is answered 5xx but its record is kept, and a retry is processed again in full, as a repeat of -an ``early`` delivery is (see `Outcomes and repeated deliveries`_). Some senders, such as Jira Cloud, retry +events, asset route events, cancel requests and audit log entries it recorded, and what it did for webhook +waits, committed before the answer was sent; a 200 ``duplicate`` means that an earlier copy's did; a 200 +handshake records nothing. Asset listeners and metrics run after the commit, and the scheduler applies the +cancel requests after the answer, within a second or two while it runs. A 5xx records nothing, so a +sender that retries can deliver again once the cause is fixed. The one exception is a delivery kept for +webhook waits, ``early`` or ``cancel_requested``, which looks for waits once more after its record has +committed: if that second look fails, the delivery is answered 5xx but its record and its cancel requests +are kept, and a retry is processed again in full, as a repeat of such a delivery is (see +`Outcomes and repeated deliveries`_). Some senders, such as Jira Cloud, retry on their own, but GitHub does not, so a GitHub delivery refused during an outage has to be sent again from the webhook's **Recent Deliveries** page in GitHub. Delivery is at least once and unordered. @@ -378,39 +397,51 @@ A recorded delivery has the first of these outcomes that applies to it: * ``resumed_wait``: it went to at least one :doc:`webhook wait </authoring-and-scheduling/webhook-waits>`, a task waiting for a delivery that matches it. A wait whose task is parked resumes the task at once, and one whose task is still running resumes it once it parks. Its record lists the waits, and the topic - events and asset route events that it recorded as well, if any. + events, asset route events and cancel requests that it recorded as well, if any. * ``accepted``: it recorded topic events or asset route events, for at least one Dag that is not paused, - and went to no wait. It is not kept for a wait, even if it has a correlation key, so a wait registered - after it never gets it. -* ``early``: it has a correlation key, recorded no event and went to no wait. It is kept, with its - payload, for a wait that matches it and is registered within + and went to no wait. Its record lists the cancel requests that it recorded as well, if any. It is not + kept for a wait, even if it has a correlation key, so a wait registered after it never gets it. +* ``cancel_requested``: it has a correlation key, recorded no event and went to no wait, and recorded a + cancel request for at least one Dag whose ``cancel_on`` it matches, which the scheduler applies to the + runs about its correlation key (see :doc:`/authoring-and-scheduling/webhook-cancel`). It is kept for a + wait as an ``early`` delivery is, and counts toward ``early_delivery_max_per_webhook`` as one does: a + wait that gets it makes it ``resumed_wait``, and its cancel requests stay. +* ``early``: it has a correlation key, recorded no event, requested no cancel and went to no wait. It is + kept, with its payload, for a wait that matches it and is registered within :ref:`[webhooks] early_delivery_ttl <config:webhooks>` seconds of its receipt. Such a wait gets it as soon as it is registered, and resumes its task once the task parks; the delivery's outcome then becomes ``resumed_wait``. Once that time has passed, it keeps its outcome and its payload until its record is removed (see `Retention`_), but no wait gets it. Unpausing a Dag does not replay it, but a repeat of it is processed again in full. -* ``expired``: it was ``early`` until its webhook kept - :ref:`[webhooks] early_delivery_max_per_webhook <config:webhooks>` later early deliveries received +* ``expired``: it was kept for waits, ``early`` or ``cancel_requested``, until its webhook kept + :ref:`[webhooks] early_delivery_max_per_webhook <config:webhooks>` later such deliveries received within ``early_delivery_ttl``. Its payload is dropped, and no wait registered after then gets it unless - the delivery is repeated. + the delivery is repeated. The cancel requests it recorded, if any, stay. * ``dag_paused``: it has no correlation key, and only Dags that are paused or draining schedule on its names or on its routes' assets, so it recorded nothing. Unpausing a Dag does not replay it. * ``no_subscribers``: it has no correlation key, no Dag that an event could start schedules on its names, and none of its asset routes could record an event, so it recorded nothing. -The record of a delivery says why each of its asset routes recorded nothing, whatever its outcome. +The record of a delivery says why each of its asset routes recorded nothing, whatever its outcome. A +delivery requests a cancel whatever else it does, so the cancel requests of a ``resumed_wait`` or +``accepted`` delivery are listed in its record too. ``duplicate`` is only ever answered, never recorded. A delivery repeats a recorded one when it is sent to the same webhook with the same delivery id and the same body; the body counts, because senders send the delivery id in a header that their signature does not cover. A repeat of an ``accepted`` or ``resumed_wait`` delivery is answered 200 ``duplicate`` and records nothing. A repeat of a delivery that -recorded no event and went to no wait, with outcome ``early``, ``expired``, ``dag_paused`` or -``no_subscribers``, is processed again in full, as a first delivery is, without a second record: a Dag -that subscribed or was unpaused since then receives it, a webhook wait registered since then gets it, and -the record then says what the repeat did. The delivery recorded no event before, so no event is recorded -twice. A repeat that is ``early`` again is early for ``early_delivery_ttl`` from when the delivery was -first received. If two copies of a delivery arrive at once, at most one of them acts. A delivery without -a delivery id, or with an empty one, cannot be recognized as a repeat, so each copy of it acts. +recorded no event and went to no wait, with outcome ``early``, ``cancel_requested``, ``expired``, +``dag_paused`` or ``no_subscribers``, is processed again in full, as a first delivery is, without a second +record: a Dag that subscribed or was unpaused since then receives it, a webhook wait registered since then +gets it, a Dag that declared a matching ``cancel_on`` since then gets a cancel request, and the record then +says what the repeat did. The delivery recorded no event before, so no event is recorded twice, and it +never records a second cancel request for a Dag: a repeat of a ``cancel_requested`` delivery keeps the +requests of the first copy, and is answered 202 ``cancel_requested`` again unless it now records an event +or goes to a wait. A request that a repeat records +is dated from when the delivery was first received, as its record is. A repeat that is ``early`` again is +early for ``early_delivery_ttl`` from when the delivery was first received. If two copies of a delivery +arrive at once, at most one of them acts. A delivery without a delivery id, or with an empty one, cannot be +recognized as a repeat, so each copy of it acts. The generic sources read the delivery id from the header or field that the connection's ``delivery_id_header`` or ``delivery_id_field`` names, and other source types from where their sender puts @@ -516,12 +547,12 @@ source type's default if it has one, or else by :ref:`[webhooks] store_payload < text such as commit messages and pull request bodies; * ``none`` keeps nothing. -A delivery that acts keeps a payload: an ``accepted``, ``resumed_wait`` or ``early`` delivery keeps one -in its record, in the extra of each of its topic events and asset route events, and in the event that a -webhook wait resumes its task with. Deliveries that act on nothing, ``no_subscribers`` and -``dag_paused``, keep none, so a sender that sends every event does not fill the database, and an -``expired`` delivery drops its payload. An ``early`` delivery keeps its payload until its record is -removed, also once ``early_delivery_ttl`` has passed, so every delivery with a correlation key that no Dag +A delivery that acts keeps a payload: an ``accepted``, ``resumed_wait``, ``cancel_requested`` or ``early`` +delivery keeps one in its record, in the extra of each of its topic events and asset route events, and in the +event that a webhook wait resumes its task with. Deliveries that act on nothing, ``no_subscribers`` and +``dag_paused``, keep none, so a sender that sends every event does not fill the database, and an ``expired`` +delivery drops its payload. An ``early`` or ``cancel_requested`` delivery keeps its payload until its record +is removed, also once ``early_delivery_ttl`` has passed, so every delivery with a correlation key that no Dag subscribes to, such as each ``github`` delivery about a pull request or an issue, keeps its payload for ``delivery_retention`` days. Each event carries its own copy, and the copy is left out of an event, with ``payload`` null and ``payload_truncated`` true, when it is larger than @@ -545,11 +576,11 @@ removes a record inside its retention, while one in the future does. A changed ` applies to deliveries received after the change. Nothing else removes records, so run ``airflow db clean`` regularly, as for the other tables it cleans; ``--yes`` skips its confirmation prompt, as a scheduled run needs. Unless you pass ``--skip-archive``, it keeps the records it removes, payloads included, in an archive -table: +table. Clean the cancel requests that deliveries record, described below, in the same run: .. code-block:: bash - airflow db clean --tables inbound_delivery --yes \ + airflow db clean --tables inbound_delivery,webhook_cancel_request --yes \ --clean-before-timestamp "$(date -u '+%Y-%m-%dT%H:%M:%S+00:00')" Keep ``delivery_retention`` longer than the longest time over which any sender you configure retries a @@ -561,6 +592,15 @@ acts again. Topic events and audit log entries are removed with their own tables received a delivery whose record was removed before its task parked resumes the task at its deadline, with the timeout event rather than the delivery. +The cancel requests that deliveries record, in the ``webhook_cancel_request`` table, have the +``retain_until`` of their delivery's record, and ``airflow db clean`` removes them by that column, whether +or not the scheduler has applied them. A request is kept once it is applied, because the scheduler also +applies it to a run created later from an earlier delivery, as described in +:doc:`/authoring-and-scheduling/webhook-cancel`. Deleting a Dag removes its requests. The conditions that +Dags declare with ``cancel_on``, in the ``dag_cancel_condition`` table, belong to their Dags: the Dag +processor keeps them in step with each Dag's code, deleting a Dag removes them, and +``airflow db clean`` does not clean them. + .. _inbound-webhooks-deliveries: Viewing deliveries @@ -568,8 +608,9 @@ Viewing deliveries The record of each delivery can be read in the UI and through the REST API: when the delivery was received, its webhook, source type, delivery id, correlation key and attributes, its outcome, the topic -names it offered, the topic events and asset route events it recorded, the asset routes it skipped and -the webhook waits it went to. Neither shows its payload. +names it offered, the topic events and asset route events it recorded, the asset routes it skipped, the +webhook waits it went to, and the cancel requests it recorded, with what the scheduler did with each. +Neither shows its payload. In the UI ~~~~~~~~~ @@ -593,8 +634,11 @@ id, time of receipt, source type and correlation key; its outcome, with a senten attributes; the topic names it offered, each with a button that copies it; the topic events it recorded, each linked to its asset's page; the events its asset routes recorded, each linked to its asset's page with the route's topic and the partition key; the asset routes it skipped, each with -why; and the webhook waits it went to, each linked to the page of the task instance that registered it, -with its Dag id, run id and the wait's id. ``?id=<id>`` in the address opens the panel of the delivery +why; the webhook waits it went to, each linked to the page of the task instance that registered it, +with its Dag id, run id and the wait's id; and the cancel requests it recorded, each with its Dag, its +scope, the tasks it ends, its status, with what it means, and when the scheduler applied it or failed to, +and the runs and task instances it cancelled, each linked to its page, or the runs it left alone, with why. +``?id=<id>`` in the address opens the panel of the delivery with that id, as in ``/webhook_deliveries/demo?id=<id>``. Wherever the UI lists asset events, the source of an event that a delivery recorded reads **Webhook** and links there. @@ -662,13 +706,13 @@ fields: * - ``correlation_key`` - What the delivery is about, if its source type says, or null. * - ``outcome`` - - ``resumed_wait``, ``accepted``, ``early``, ``expired``, ``dag_paused`` or ``no_subscribers`` (see - `Outcomes and repeated deliveries`_). + - ``resumed_wait``, ``accepted``, ``cancel_requested``, ``early``, ``expired``, ``dag_paused`` or + ``no_subscribers`` (see `Outcomes and repeated deliveries`_). * - ``candidates`` - The topic names that the delivery offered. * - ``actions`` - - What the delivery did, or null if it recorded nothing, skipped no asset route and went to no - webhook wait. Each of these keys is present only when its list is not empty: + - What the delivery did, or null if it recorded nothing, skipped no asset route, went to no webhook + wait and recorded no cancel request. Each of these keys is present only when its list is not empty: * ``asset_events``: the topic events that the delivery recorded, sorted by asset name, each with the asset's name, the asset's id and the asset event's id, as in @@ -684,7 +728,40 @@ fields: id, run id, task id and map index of the task instance that registered it, as in ``{"wait_id": "0199...", "ti_id": "0199...", "dag_id": "deploy", "run_id": "manual__2026-10-20", "task_id": "wait_for_merge", "map_index": -1}``, plus its ``region_id`` and ``region_index`` if - the task instance is in a dynamic region, such as a mapped task's expansion or a loop. + the task instance is in a dynamic region, such as a mapped task's expansion or a loop; + * ``cancel_requests``: the cancel requests that the delivery recorded, each with the request's id, + the Dag id, and the scope, ``run`` if a condition of the Dag itself matched or else ``task``, as in + ``{"request_id": "0199...", "dag_id": "deploy_preview", "scope": "run"}``. It summarizes the + ``cancel_requests`` field, which is authoritative. + * - ``cancel_requests`` + - The cancel requests that the delivery recorded, by Dag id, each with what the scheduler did with it + (see :doc:`/authoring-and-scheduling/webhook-cancel`). These records are authoritative: ``actions`` + summarizes them, can miss one that a concurrent repeat of the delivery recorded, and still names a + request whose record was removed. Each has: + + * ``id`` and ``dag_id``; + * ``targets``: what the request cancels in each run about the delivery's correlation key, as in + ``{"run": {"correlation_key_confs": ["pr"]}, "tasks": [{"task_id": "run_e2e_tests", + "cancel_state": "skipped", "correlation_key_confs": []}]}``. ``run`` is null unless a condition of + the Dag itself matched, and fails the whole run; ``tasks`` lists, by task id, each task whose + condition matched, with the state it ends the task in. ``correlation_key_confs`` lists the conf + keys under which a run triggered with a conf holds the correlation key, and is empty when only + the runs that a delivery started are about it; + * ``status``: ``pending`` until the scheduler applies the request, which it does within a second + or two while it runs; then ``applied``, or ``failed`` if applying it raised an error, which the + scheduler logged; + * ``applied_at``: when the scheduler applied the request or failed to, or null; + * ``result``: null until the scheduler acts on the request, then what it did: + ``{"dag_runs": [...], "task_instances": [...], "skipped_runs": [...], "error": null}``. + ``dag_runs`` lists the ``run_id`` and ``state`` of each run that the request cancelled, ``failed``, + or ``running`` or ``queued`` while a teardown still has to run; ``task_instances`` the + ``ti_id``, ``run_id``, ``task_id``, ``map_index`` and new ``state`` of each task instance that it + ended, plus ``region_id`` and ``region_index`` if the task instance is in a dynamic region; + ``skipped_runs`` the ``run_id`` of each run about the key that it left alone, with the + ``reason``, ``dag_not_found`` if the run's Dag version could not be loaded or ``dag_has_team`` + (see :doc:`/authoring-and-scheduling/webhook-cancel`); and ``error`` the class name of the error + that failed it. A run created later from an earlier delivery is added when the scheduler cancels + it as it creates it, which can happen while the request is ``pending``. * - ``team_name`` - Null: webhooks have no team in Airflow 3.5. @@ -766,12 +843,14 @@ the webhook's connection, and never returns the payload of a delivery. Each wait * - ``status`` - ``registered`` until a matching delivery arrives; ``received`` if one arrived before the task parked: while the task was running, or before the wait was registered, when Airflow kept the - delivery ``early``; the task resumes with it once it parks; ``resumed`` once a delivery resumed the - task; ``timed_out`` if none arrived before the deadline; ``ended`` if the task registered another - wait, or if a wait registered on the same events and match found the wait's try no longer running - or parked, or, for a ``registered`` wait, a matching delivery did. Nothing else ends a wait whose - try stopped: it keeps ``registered`` until a matching delivery or such a registration ends it, and - ``received``, since no other delivery is offered to it, until such a registration does. + delivery for waits (``early``, or ``cancel_requested`` if it requested a cancel); the task resumes + with it once it parks; ``resumed`` once a delivery resumed the task; ``timed_out`` if none arrived + before the deadline; ``ended`` if the task registered another wait, if a webhook cancel request + ended its task (see :doc:`/authoring-and-scheduling/webhook-cancel`), or if a wait registered on the + same events and match found the wait's try no longer running or parked, or, for a ``registered`` + wait, a matching delivery did. Nothing else ends a wait whose try stopped: it keeps ``registered`` + until a matching delivery or such a registration ends it, and ``received``, since no other delivery + is offered to it, until such a registration does. * - ``deadline`` - When the wait times out if no matching delivery has arrived. * - ``created_at`` and ``ended_at`` @@ -796,10 +875,10 @@ A webhook's deliveries say why a delivery started no run: an event (see below). Compare those names with the name of the Dag's topic. * ``dag_paused``: only paused or draining Dags schedule on its names. Unpausing a Dag does not replay the delivery, but the sender can deliver it again. -* ``early`` or ``expired``: the delivery has a correlation key and recorded no event: no Dag that an event - could start schedules on its names, or only paused or draining ones do, and no asset route recorded an - event. Unpausing a Dag does not replay the delivery, but the sender can deliver it again, and the repeat - is processed again in full, unless a webhook wait has taken the delivery since. +* ``early``, ``cancel_requested`` or ``expired``: the delivery has a correlation key and recorded no event: no + Dag that an event could start schedules on its names, or only paused or draining ones do, and no asset route + recorded an event. Unpausing a Dag does not replay the delivery, but the sender can deliver it again, and + the repeat is processed again in full, unless a webhook wait has taken the delivery since. * ``accepted``, or ``resumed_wait`` with events: it recorded the topic events and asset route events that its details link to, which queue the Dags scheduled on their assets. A Dag that has no run yet may be waiting for an event on another asset, as with ``a & b``, or, with a partitioned timetable, for the same @@ -821,20 +900,72 @@ route's topic, which its topic names show: ``partition_key_field`` names an asset that no Dag with a partitioned timetable schedules on, or one without names an asset that only such Dags schedule on. +.. _inbound-webhooks-not-cancelled: + +Why a run was or was not cancelled +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +:doc:`/authoring-and-scheduling/webhook-cancel` describes which runs a delivery cancels. When a run that +should have been cancelled was not, the deliveries say why: + +* The delivery has no correlation key, so it requested no cancel: its record shows none. A GitHub ``push`` + or ``workflow_run`` has none, and neither has a delivery to a generic source whose connection sets no + ``correlation_key_field``, or whose field the delivery does not fill. ``airflow webhooks check`` notes a + webhook with ``cancel_on`` conditions whose latest deliveries have no key. +* The delivery lists no cancel request for the Dag: none of the Dag's conditions, or its tasks', matched + its event and attributes, which its record shows, or the Dag declared them only after the delivery + arrived. ``airflow webhooks check`` lists the conditions on each webhook. +* :ref:`config:core__multi_team` is on, and the Dag has a team: no request is recorded for it, and + ``airflow webhooks check`` reports it. +* The request is ``pending``: no scheduler has applied it yet, because none is running, or because many + requests wait before it. +* The request is ``failed``: applying it raised an error, which its ``error`` names and the scheduler + logged. +* The request is ``applied`` but lists no run, or not this one: the run is not about the delivery's + correlation key. It was started by an event that a delivery to another webhook, or with another key, + recorded, or by an event that the REST API, a trigger or a task created; it was triggered without the key in + its ``conf``, under another conf key than the condition's ``correlation_key_conf``, or with a condition that + names none; it was triggered after the delivery was received, or, if only its ``conf`` makes it about the + key, cleared after it, or started from an event newer than the delivery; or it had already finished. +* The request is ``applied`` but lists no run, and the run's only unfinished tasks were teardowns when the + scheduler applied it: a cancel leaves teardowns to run, so it changed nothing, and the run ends as its tasks + make it end. +* The request lists the run under the runs it left alone, with the reason ``dag_not_found``, when the + scheduler could not load the run's Dag version, or ``dag_has_team``. + +A run that was cancelled by mistake can be cleared, which queues it again, as for any failed run. Clearing it +does not cancel it again once the request that cancelled it is ``applied``. A run that the scheduler cancelled +as it created it, while the request was still ``pending``, is cancelled again if it is cleared before the +scheduler applies that request, which takes longer when many requests wait. + +.. _inbound-webhooks-audit-log: + Audit log ~~~~~~~~~ -A delivery that records topic events or asset route events also writes an audit log entry, in the same -transaction, with the event ``webhook_delivery``, the owner ``webhook:<name>``, and in its extra the -delivery id, the topic names it recorded events on as ``names``, its outcome, ``accepted`` or -``resumed_wait``, and, if its asset routes recorded events, the ``topic`` and ``asset`` of each such route -as ``asset_routes``. The entry has no Dag, so only users allowed to read audit log entries that belong to -no Dag see it, such as the **admin** role of the Simple auth manager. +A delivery that records topic events, asset route events or cancel requests also writes an audit log +entry, in the same transaction, with the event ``webhook_delivery``, the owner ``webhook:<name>``, and in +its extra the delivery id, the topic names it recorded events on as ``names``, its outcome, ``accepted``, +``resumed_wait`` or ``cancel_requested``, if its asset routes recorded events, the ``topic`` and +``asset`` of each such route as ``asset_routes``, and, if it recorded cancel requests, the +``request_id``, ``dag_id`` and ``scope`` of each as ``cancel_requests``. A repeat of a delivery writes an +entry only if it records something new, and lists only the cancel requests that it recorded itself. The +entry has no Dag, so only users allowed to read audit log entries that belong to no Dag see it, such as +the **admin** role of the Simple auth manager. A delivery that goes to webhook waits writes one more entry for each wait, with the same event and owner, the task instance that registered the wait, and in its extra the delivery id, the outcome -``resumed_wait`` and the wait's id as ``wait_id``. A wait that gets an ``early`` delivery when it is -registered writes the same entry. Users allowed to read the audit log of the task's Dag see it. +``resumed_wait`` and the wait's id as ``wait_id``. A wait that gets an ``early`` or ``cancel_requested`` +delivery when it is registered writes the same entry. Users allowed to read the audit log of the task's +Dag see it. + +The scheduler writes an entry with the event ``webhook_cancel`` and the owner ``webhook:<name>`` for each +run that a cancel request cancels, with the run's Dag id and run id, and for each task instance that a +task's ``cancel_on`` ends, with the task instance. Its extra has the delivery id, the sender's or else the +id of Airflow's record of the delivery, the request's id as ``request_id``, the ``scope``, ``run`` or +``task``, and for a task instance the ``state`` it ended in. Users allowed to read the audit log of the +Dag see these entries. A task instance that a run's cancel ends gets no entry of its own: the run's entry +covers it. .. _inbound-webhooks-metrics: @@ -842,7 +973,7 @@ Metrics ------- When :doc:`metrics <logging-monitoring/metrics>` are enabled, the webhooks process emits the first three of -these metrics, and the scheduler ``webhooks.open_waits``: +these metrics, and the scheduler ``webhooks.open_waits`` and ``webhooks.cancels``: .. list-table:: :header-rows: 1 @@ -868,6 +999,12 @@ these metrics, and the scheduler ``webhooks.open_waits``: - Open webhook waits whose task is running or parked, tagged ``webhook``, every :ref:`[scheduler] ti_metrics_interval <config:scheduler__ti_metrics_interval>`. A webhook left with none reports 0 once, and nothing more until it has one again. + * - ``webhooks.cancels`` + - counter + - Runs and task instances that webhook cancel requests cancelled, tagged ``webhook`` and ``scope``: + one with the scope ``run`` for each run that a Dag's ``cancel_on`` cancelled, and one with the scope + ``task`` for each task instance that a task's ``cancel_on`` ended. Requests that cancel nothing are + not counted. The tags reach OpenTelemetry, and StatsD only with a tagged format such as InfluxDB or DogStatsD; see :doc:`logging-monitoring/metrics`. Requests under ``/webhooks`` are also counted in the API server's HTTP @@ -879,13 +1016,23 @@ Command line ------------ ``airflow webhooks check [NAME ...]`` - Checks that each webhook, by default each name that ``[webhooks] names`` lists, can be served: that - ``names`` lists it, and that its connection exists without a team, has the ``inbound_webhook`` type and - a password, and configures a source type that an installed provider serves, with options that the - source type accepts. It names the problem otherwise, such as a source type that two packages declare, - or the standard provider release needed for ``hmac_json`` and ``token_header``. It prints the source - type, body limit and ``store_payload`` of each webhook it can serve, and exits with 1 if any webhook has - a problem. + Checks that each webhook, by default each name that ``[webhooks] names`` lists and each webhook that a + Dag's or a task's ``cancel_on`` names, can be served: that ``names`` lists it, and that its connection + exists without a team, has the ``inbound_webhook`` type and a password, and configures a source type + that an installed provider serves, with options that the source type accepts. It names the problem + otherwise, such as a source type that two packages declare, or the standard provider release needed + for ``hmac_json`` and ``token_header``. It prints the source type, body limit and ``store_payload`` of + each webhook it can serve. + + It then lists which Dags and tasks each webhook cancels, one line for each condition of their + ``cancel_on``, as in ``Webhook github cancels the runs of Dag deploy_preview on pull_request deliveries + with action closed, repository my-org/web, and manual runs by conf key pr`` and ``Webhook github sets task + run_e2e_tests of Dag deploy_preview to skipped on pull_request deliveries with action synchronize, + repository my-org/web``. With :ref:`config:core__multi_team` on, a Dag with a team whose ``cancel_on`` + names the webhook is a problem, since no cancel request is recorded for it. For a webhook with conditions, + it notes when none of its latest 100 recorded deliveries has a correlation key, since ``cancel_on`` acts + only on deliveries that have one; the note does not change the exit code. It exits with 1 if any webhook + has a problem. ``airflow webhooks send-test NAME --url URL --data DATA`` Sends one delivery as a sender would, and prints the answer as @@ -895,7 +1042,8 @@ Command line for the content of a file; ``--secret``, ``--signature-header`` and ``--signature-prefix`` replace what the connection ``webhook_<name>`` says; ``--delivery-id`` sends an id in the connection's ``delivery_id_header``; and ``--header 'NAME: VALUE'``, which may be repeated, adds a header. A test - delivery is processed like any other, so it starts the Dags that schedule on its names. + delivery is processed like any other, so it starts the Dags that schedule on its names, and records + cancel requests that cancel the runs about its correlation key. ``airflow providers webhook-sources`` Lists the source types that the installed providers declare, and whether each is served. @@ -930,8 +1078,14 @@ The webhooks process answers anyone who can reach it, so give it only what it ne * for deliveries that go to webhook waits, ``SELECT`` on ``task_instance``, ``dag_run`` and ``dynamic_region``, and ``UPDATE`` on ``inbound_wait`` and on ``task_instance``, which a delivery locks and writes to resume a parked task; + * for deliveries with a correlation key, ``SELECT`` on ``dag_cancel_condition``, which holds the + ``cancel_on`` conditions of Dags, and ``SELECT`` and ``INSERT`` on ``webhook_cancel_request``, which + holds the cancel requests that deliveries record and which a repeat of a delivery reads. The process + writes no run or task instance state for a cancel: the scheduler applies the requests. A request's id + is a UUID, so no sequence is involved; * when :ref:`config:core__multi_team` is on, ``SELECT`` on ``dag_bundle``, ``dag_bundle_team`` and - ``team``, which the team filter reads; + ``team``, which the team filter reads, also with ``dag`` for the ``cancel_on`` conditions of Dags with a + team; * for asset routes with a ``partition_key_field``, ``SELECT`` on ``serialized_dag`` and ``dag_version``, ``SELECT`` and ``INSERT`` on ``asset_partition_dag_run`` and ``partitioned_asset_key_log``, which queue the partition-aware Dags; ``UPDATE`` on ``asset``, which @@ -940,8 +1094,9 @@ The webhooks process answers anyone who can reach it, so give it only what it ne Airflow's tests pin which tables a delivery reads and writes when ``multi_team`` is off, for topic events and for asset routes with a partition key, but not for deliveries that go to webhook waits, and not these - privileges. A delivery that needs a privilege the process lacks is answered 500 or 503, depending on the - database, and the error is logged. + privileges; they also check that a delivery that requests a cancel reads and writes no ``dag_run``, + ``task_instance``, ``serialized_dag`` or ``dag_version``. A delivery that needs a privilege the process + lacks is answered 500 or 503, depending on the database, and the error is logged. * Only the plugins it needs. The first delivery that records an event imports every plugin installed where the process runs, from the plugins folder and from entry points, to find listeners, and the ``on_asset_changed`` and ``on_asset_event_emitted`` hooks of listener plugins then run after each such @@ -960,10 +1115,15 @@ Turning webhooks off from the ``names`` of the API servers that serve the Execution API also refuses new waits on it. * To keep a webhook's deliveries from starting runs while still recording them, pause every Dag that schedules on its topics or on the assets that its asset routes name. Its deliveries are then recorded - as ``dag_paused``, or as ``early`` if they have a correlation key, such as each ``github`` delivery about - a pull request or an issue, and a sender's redelivery of one after unpausing is processed again, so the - unpaused Dags receive it. A delivery that a webhook wait took, at once or while it was ``early``, is - ``resumed_wait`` instead, and a redelivery of it is a duplicate (see :ref:`inbound-webhooks-outcomes`). + as ``dag_paused``, or as ``early``, or ``cancel_requested`` if they request a cancel, if they have a + correlation key, such as each ``github`` delivery about a pull request or an issue, and a sender's + redelivery of one after unpausing is processed again, so the unpaused Dags receive it. A delivery that a + webhook wait took, at once or while it was kept for waits, is ``resumed_wait`` instead, and a redelivery + of it is a duplicate (see :ref:`inbound-webhooks-outcomes`). +* Pausing a Dag does not stop the deliveries that match its ``cancel_on`` from cancelling its runs. To stop + a webhook from cancelling runs, remove the conditions on it from the ``cancel_on`` of the Dags and tasks, + which applies once the Dag processor has parsed them, or stop the webhook as above. The scheduler still + applies the cancel requests that deliveries recorded before. Migrating from airflow-provider-inbound-webhooks ------------------------------------------------ diff --git a/airflow-core/docs/authoring-and-scheduling/asset-scheduling.rst b/airflow-core/docs/authoring-and-scheduling/asset-scheduling.rst index 1794c46c57d..2412ab9de8a 100644 --- a/airflow-core/docs/authoring-and-scheduling/asset-scheduling.rst +++ b/airflow-core/docs/authoring-and-scheduling/asset-scheduling.rst @@ -349,7 +349,8 @@ deliveries record events on data assets, so that the Dags scheduled on them run, :ref:`asset routes <inbound-webhooks-asset-routes>`. For details and examples, see :doc:`webhook-topics`. A task can also wait in a running Dag run for a -delivery, as described in :doc:`webhook-waits`. +delivery, as described in :doc:`webhook-waits`, and a Dag or a task can declare which later deliveries +cancel the runs about the same pull request or other subject, as described in :doc:`webhook-cancel`. Advanced asset scheduling with conditional expressions -------------------------------------------------------- diff --git a/airflow-core/docs/authoring-and-scheduling/index.rst b/airflow-core/docs/authoring-and-scheduling/index.rst index ec55471ae6e..6d2f90c0238 100644 --- a/airflow-core/docs/authoring-and-scheduling/index.rst +++ b/airflow-core/docs/authoring-and-scheduling/index.rst @@ -53,3 +53,4 @@ It's recommended that you first review the pages in :doc:`core concepts </core-c Event-Driven Scheduling <event-scheduling> Scheduling on Inbound Webhooks <webhook-topics> Waiting for Inbound Webhooks <webhook-waits> + Cancelling Runs from Inbound Webhooks <webhook-cancel> diff --git a/airflow-core/docs/authoring-and-scheduling/webhook-cancel.rst b/airflow-core/docs/authoring-and-scheduling/webhook-cancel.rst new file mode 100644 index 00000000000..a25e347edde --- /dev/null +++ b/airflow-core/docs/authoring-and-scheduling/webhook-cancel.rst @@ -0,0 +1,442 @@ + .. 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. + +.. _webhook-cancel: + +Cancelling Runs from Inbound Webhooks +===================================== + +|experimental| + +.. versionadded:: 3.6.0 + +.. warning:: + + Cancelling runs from inbound webhooks is experimental. ``WebhookMatch``, ``cancel_on``, + ``cancel_state``, which runs a delivery cancels and what a cancel does to them may change in a minor + release, so check the release notes when you upgrade. + +A Dag or a task can declare which deliveries to Airflow's inbound webhooks cancel its runs: a Dag that +deploys a preview of a pull request, for example, can have its runs stopped when the pull request is +closed. It declares ``cancel_on``, a list of :class:`~airflow.sdk.WebhookMatch` conditions, each naming a +webhook, the events and the attributes of the deliveries that cancel. When a delivery that has a +*correlation key*, a value that says what the delivery is about, such as a pull request, matches a +condition, Airflow records a *cancel request* for the Dag before it answers the sender. The scheduler then +applies the request, within a second or two, to the Dag's queued and running runs that are about the same +correlation key, through an event that an earlier delivery recorded, including runs the scheduler creates +later from such an event, or through their conf if they were queued before the delivery: + +* a Dag's ``cancel_on`` fails those runs, as marking them failed does; +* a task's ``cancel_on`` ends that task in those runs, in the task's ``cancel_state``, ``failed`` by + default or ``skipped``, and the runs go on. + +A run is about a correlation key when a delivery with that key started it, through a +:doc:`webhook topic <webhook-topics>` or an asset route, or when it was triggered with the key in its +``conf``, under the conf key that the condition names. + +The Deployment Manager sets up the webhook itself, as described in +:doc:`/administration-and-deployment/webhooks`. Cancelling needs Airflow 3.6 or later, in the Airflow core +of the API servers, the scheduler and the Dag processor, and in the Task SDK of the Dag processor and the +workers, which parse the Dag files: an older Task SDK does not know ``cancel_on``, and fails to import a +Dag that uses it. + +Example: a preview environment +------------------------------ + +This Dag deploys a preview of a pull request in ``my-org/web`` each time the label ``deploy-preview`` is +added to it, and serves it while the end-to-end tests run against it. It needs the ``github`` webhook, +connected to the repository with "Pull requests" events as described in the GitHub provider's +:ref:`Inbound Webhooks <apache-airflow-providers-github:howto/webhooks:github>` page. + +.. code-block:: python + + from airflow.sdk import Param, WebhookMatch, WebhookTopic, dag, task + + REPOSITORY = "my-org/web" + + preview_requested = WebhookTopic( + webhook="github", path=REPOSITORY, event="pull_request", action="labeled", detail="deploy-preview" + ) + pr_closed = WebhookMatch( + webhook="github", + events=["pull_request"], + match={"repository": REPOSITORY, "action": "closed"}, + correlation_key_conf="pr", + ) + new_commit = WebhookMatch( + webhook="github", + events=["pull_request"], + match={"repository": REPOSITORY, "action": "synchronize"}, + ) + + + def tear_down(context): + if context["reason"] == "cancelled": + print("The pull request was closed: tearing down the preview of", context["run_id"]) + + + @dag( + schedule=preview_requested, + catchup=False, + cancel_on=[pr_closed], + on_failure_callback=tear_down, + params={"pr": Param(None, type=["null", "string"])}, + ) + def deploy_preview(): + @task + def deploy(*, triggering_asset_events, params) -> str: + events = triggering_asset_events.get(preview_requested, []) + pull_request = events[-1].extra["correlation_key"] if events else params["pr"] + print("Deploying a preview of", pull_request) + return pull_request + + @task + def serve_preview(pull_request: str): ... + + @task(cancel_on=[new_commit], cancel_state="skipped") + def run_e2e_tests(pull_request: str): ... + + pull_request = deploy() + serve_preview(pull_request) + run_e2e_tests(pull_request) + + + deploy_preview() + +The GitHub source type gives every delivery about a pull request or an issue the correlation key +``<repository>#<number>``, whatever its action, and each topic event keeps the key of its delivery in its +``extra``. So with these runs of ``deploy_preview``: + +.. list-table:: + :header-rows: 1 + :widths: 30 25 45 + + * - Run + - Its correlation key + - What cancels it + * - A, started by the label on pull request 42 + - ``my-org/web#42``, from the topic event that started it + - Closing pull request 42 fails run A. A new commit pushed to pull request 42 skips its + ``run_e2e_tests``, and run A goes on. + * - B, started by the label on pull request 57 + - ``my-org/web#57`` + - Only the deliveries about pull request 57. + * - C, triggered with the conf ``{"pr": "my-org/web#42"}`` + - ``my-org/web#42``, through ``pr_closed.correlation_key_conf`` + - Closing pull request 42. ``new_commit`` names no conf key, so it never applies to run C. + * - D, triggered without ``pr`` + - None + - Nothing. + +Closing pull request 42 runs ``tear_down`` for run A and run C, with ``context["reason"]`` set to +``cancelled``. A run started by a label added to pull request 42 after it was closed, or triggered with its +key after it was closed, is not cancelled: a delivery cancels only the runs whose event, or whose queueing +with the key in their ``conf``, came before it. + +Conditions +---------- + +``WebhookMatch(webhook, events=None, match=None, correlation_key_conf=None)`` describes the deliveries to +one webhook that cancel: + +* ``webhook`` is the webhook's name, as ``[webhooks] names`` lists it, such as ``github``. +* ``events`` and ``match`` follow the rules of a webhook wait, described in + :ref:`webhook-waits-match`: ``events`` is None, for any event, or a non-empty list of the events that + match; ``match`` maps the names of the attributes that the webhook's source type declares to a value, or + a list of values of which one must match, and every name in it must match. Values are strings or + integers and compare as strings, and an empty or missing ``match`` accepts every delivery that + ``events`` accepts. Unlike those of ``WaitForWebhookOperator``, these values are not templates: they are + fixed when the Dag file is parsed. +* ``correlation_key_conf`` names a top-level key of a run's ``conf`` whose value is the run's correlation + key, so that a run triggered with a ``conf``, such as a manual run, can be cancelled too. The value is + read as a string from a string or an integer; it is not a dotted path, and no template is rendered. With + None, the default, the condition cancels only runs that a delivery started. + +A Dag declares conditions with ``cancel_on`` in ``DAG(...)`` or ``@dag(...)``, which fails its runs, and a +task with ``cancel_on`` among its arguments, which ends that task, together with ``cancel_state``. A task +takes them as any operator argument: in ``@task(...)``, ``.override()``, ``.partial()`` for a mapped task, +whose every map index they then apply to, and the ``default_args`` of the Dag or of a task group. A +``cancel_on`` in ``default_args`` gives each task the conditions, so a matching delivery ends every +unfinished task of the run in its ``cancel_state`` rather than failing the run as the Dag's ``cancel_on`` +does. ``cancel_on`` takes one condition or a list of them, and conditions that are equal, such as two that +differ only in the order of their values, count once. + +These are refused when the Dag file is parsed, so that the Dag fails to import rather than never +cancelling: + +* a condition that breaks the rules above, which raises ``ValueError``: such as an empty ``events`` list, + a value that is neither a string nor an integer, a webhook name that is not 1 to 64 lowercase letters, + digits, ``-`` and ``_`` starting with a letter or a digit, or a ``correlation_key_conf`` that is blank or + longer than 250 characters; +* a ``cancel_on`` that is not a ``WebhookMatch`` or a list of them, which raises ``TypeError``; +* ``cancel_on`` on a teardown task, which runs however its run ends. If the teardown gets it from + ``default_args``, pass ``cancel_on=None`` to the teardown; +* ``cancel_state`` without ``cancel_on``, unless it comes from ``default_args``, and a ``cancel_state`` + other than ``"failed"`` or ``"skipped"``; +* ``cancel_on`` or ``cancel_state`` in ``.expand()``: pass them to ``.partial()``. + +Airflow does not check that ``[webhooks] names`` lists the webhook, or that its source type declares the +names in ``match``, so a misspelled name makes a condition that no delivery matches. +``airflow webhooks check`` lists the conditions on each webhook and reports a webhook that is not listed +(see :ref:`inbound-webhooks-cli`). + +Which runs a delivery cancels +----------------------------- + +Only a delivery that has a correlation key requests a cancel, and only its source type sets one: + +* The GitHub source type gives every delivery about a pull request or an issue the key + ``<repository>#<number>``, such as ``my-org/web#42``. A ``push``, a ``workflow_run`` or an organization + event has none. +* The generic ``hmac_json`` and ``token_header`` source types read it from the field that the webhook + connection's ``correlation_key_field`` names, and a webhook whose connection does not set it has none. + +When such a delivery matches conditions of a Dag, Airflow records one cancel request for the Dag, in the +transaction that records the delivery. Its *targets* are the whole run if a condition of the Dag itself +matched, and each task whose condition matched, with that task's ``cancel_state``. The scheduler applies +the request to the Dag's runs that are queued or running and about the delivery's correlation key: + +* **Runs that a delivery started.** The run consumed an asset event, a topic event or an asset route + event, that a delivery to the same webhook with the same correlation key recorded before the cancelling + delivery was received. Only events that a delivery recorded count: an event that a task emitted, or that + the REST API or an asset watcher's trigger created, never makes a run about a key, whatever its + ``extra`` says. +* **Runs triggered with a conf.** The run was queued before the delivery was received, and its ``conf`` + holds the correlation key, as a string or an integer, under the ``correlation_key_conf`` of a matched + condition. Clearing a run queues it again, so a run cleared after the delivery was received is no + longer about it. + +For each run, the whole-run target applies if a matched condition of the Dag itself makes the run about +the key: any of them for a run that a delivery started, and for a run triggered with a conf, one whose +``correlation_key_conf`` names the conf key that holds the key. Otherwise the task targets whose matched +conditions make the run about the key apply, in the same way. +In the example, ``new_commit`` names no conf key, so it ends ``run_e2e_tests`` in runs that a delivery +started, never in runs triggered with ``pr``. + +A delivery never cancels: + +* a run that has finished, nor a task instance that has; +* a run that its own events start, or any run started from an event newer than the delivery; +* a run triggered after the delivery was received, or a run that only its ``conf`` makes about the key and + that was cleared after the delivery was received. + +A run that a delivery started is cancelled if it is queued or running when the scheduler applies the request, +even if it was cleared after the cancelling delivery was received. + +A Dag that is paused, draining or stale keeps its conditions, and its runs can still be cancelled, since a +cancel only stops work. Applying a request changes each run at most once, however many ways the run is about +its key. + +What a cancel does +------------------ + +A request changes the unfinished task instances of each run it applies to, by their state: + +.. list-table:: + :header-rows: 1 + :widths: 25 40 35 + + * - Task instance state + - The Dag's ``cancel_on`` + - A task's ``cancel_on`` + * - ``running`` + - ``failed``. The worker's next heartbeat, within + :ref:`[workers] min_heartbeat_interval <config:workers__min_heartbeat_interval>` seconds, finds the + change and stops the task: its ``on_kill`` runs, and the task is killed + :ref:`[core] killed_task_cleanup_time <config:core__killed_task_cleanup_time>` seconds later if it + is still running. + - ``cancel_state``, and the task is stopped the same way. + * - ``deferred`` + - ``failed``. The triggerer stops its trigger, as when a deferred task is marked failed. + - ``cancel_state``, the same way. + * - ``awaiting_input`` + - ``failed``. Its open :doc:`webhook wait <webhook-waits>` ends, with the status ``ended``, and a + Human-in-the-loop request it waits for is left unanswered. + - ``cancel_state``, the same way. + * - ``up_for_reschedule`` + - ``failed``. + - ``cancel_state``. + * - No state yet, ``scheduled``, ``queued``, ``up_for_retry`` or ``restarting`` + - ``skipped``, so it never runs, and a retry does not happen. + - ``cancel_state``, so it never runs. + * - Finished + - Left as it is. + - Left as it is. + * - A teardown task + - Left to run: the run ends when it finishes. + - Cannot have ``cancel_on``. + +The Dag's ``cancel_on`` then fails the run, unless a teardown still has to run. A task's ``cancel_on`` +leaves the run running, and the tasks downstream of the task follow their trigger rules: with the default +``all_success``, they become ``upstream_failed`` downstream of a failed task, and ``skipped`` downstream of +a skipped one. The run ends as any run does, once its tasks have finished. + +A cancel is a failure: + +* **No cancelled state.** A run cancelled by the Dag's ``cancel_on`` is ``failed``, and shows red in the + UI. Airflow has no state for a cancelled run or task. +* **Callbacks.** The Dag's ``on_failure_callback`` runs, in the Dag processor, with ``context["reason"]`` + set to ``cancelled``, as it is set to ``timed_out`` for a run that timed out, so the callback can tell a + cancel from a failure. Its context describes the latest started of the task instances that the cancel + stopped, or, if it stopped none, has only ``dag``, ``run_id`` and ``reason``. A task's ``cancel_on`` + sends no Dag callback of its own. Listeners get ``on_dag_run_failed`` with the message ``cancelled`` for + a run that the cancel failed, and ``on_task_instance_failed`` or ``on_task_instance_skipped`` for each + task instance that it stopped. +* **Auto-pause.** A cancelled run is ``failed``, and counts toward ``max_consecutive_failed_dag_runs`` when + a later run fails. A Dag whose runs are often cancelled can then be paused by a few failures, so give it + a ``max_consecutive_failed_dag_runs`` that allows for its cancels, or none. +* **Teardowns.** A run held by a teardown ends when the teardown finishes, in the state its leaf tasks give it + as for any run: ``failed`` with the reason ``task_failure`` if a leaf task failed, as when the killed + running task is a leaf, or ``success``, with ``on_success_callback``, if every remaining leaf was skipped, + as when the cancel skipped the tasks downstream of the killed task. It gets no ``cancelled`` callback. It + gets the note and the audit log entry described in `Following a cancel`_ only if the cancel ended some of + its tasks: a run whose only unfinished tasks are teardowns is left as it is, with no note, no audit log + entry and no entry in the request's result. +* **Active runs.** A run that the Dag's ``cancel_on`` fails no longer counts toward the Dag's + ``max_active_runs``, so the Dag's next run can start. +* **Same-delivery waits.** A delivery that resumes a webhook wait and also requests the cancel of its run + resumes the wait at once. The scheduler then cancels the run, ending the resumed task as ``failed`` if it + is running, or ``skipped`` if it has not started yet; a task's ``cancel_on`` ends it in its + ``cancel_state`` instead. + +Runs created later +------------------ + +A request can be applied before the run that it is about exists. The scheduler holds an asset-triggered +run back while its Dag has ``max_active_runs`` runs active, and creates the run of a Dag scheduled with +``AssetAndTimeSchedule`` only when its time has come, which is usually well after the event. So when the +scheduler creates an asset-triggered run, partitioned or not, or a run of an ``AssetAndTimeSchedule``, of a +Dag that declares any ``cancel_on``, it applies to it at once the requests about the correlation keys of +the delivery events it consumed that were received after those events, ``pending`` or ``applied``, but +not ``failed`` ones. That is why a request is kept after it is applied, until ``airflow db clean`` removes +it with its delivery's record. The run is then failed, or its tasks are given their ``cancel_state``, +before any of its tasks runs, and, unless another scheduler is applying the request at that moment, the +run is added to the request's result. If applying a request to a new run fails, the scheduler logs the +error and leaves the run as it was created. + +This has some consequences: + +* A pull request closed and then reopened while its label's event waited still cancels the run that the + event starts: deliveries arrive in any order, and the close was received after the label. +* A run that consumed the events of several deliveries, as with + :ref:`config:scheduler__batch_asset_events` or a schedule on several assets, is about each of their keys, + and a request about any of them cancels it. ``cancel_on`` assumes one run per event, the default. +* With more than one scheduler, a request committed after a creating scheduler read the requests, and applied + by another scheduler before the new run commits, is not applied to that run. With one scheduler, a request + that commits while the scheduler creates a run about its key is applied to the run the next time the + scheduler looks for pending requests. Whatever the number of schedulers, a request applied while the API + transaction of a run triggered with the key in its ``conf`` is still open is never applied to that run. + +Latency +------- + +A delivery's cancel requests are committed before it is answered. The scheduler looks for pending +requests about once a second, and applies up to +:ref:`[webhooks] cancel_requests_per_loop <config:webhooks>` of them, oldest first, each in a transaction +of its own, so a request is applied within a second or two of the delivery's answer, then the task is +killed at its next heartbeat. It takes longer when many requests wait, or when the scheduler is busy. +Several schedulers share the pending requests, and never apply one twice. + +While no scheduler runs, nothing is applied: requests stay ``pending``, and the tasks they are about keep +running. A scheduler that starts again applies them to the runs that are still queued or running. A +request whose application raises an error is marked ``failed``, and is not applied again: the scheduler +logs the error, and the request's result names its class. A database error that may pass, such as a lost +connection, leaves the request ``pending`` instead, for the scheduler's next look. + +Dag versions +------------ + +The conditions come from the latest version of the Dag that the Dag processor stored when the delivery +arrives, and the request's targets are fixed then: a change to ``cancel_on`` makes a new Dag version, and +applies to the deliveries received once the Dag processor has stored it. A request already recorded is +applied as it was recorded, even if the Dag no longer declares the condition. + +Each run's own Dag version, or the latest one for a bundle without versions, decides which tasks the run has, +which are teardowns, and whether the Dag's ``on_failure_callback`` runs for the cancel, as for any failure of +the run: a targeted task missing from it is left alone. The task instance page and the Dag details page show +the latest conditions, not those of the run's version. + +Teams +----- + +Webhooks have no team in this version of Airflow. When :ref:`config:core__multi_team` is on, a delivery +records no cancel request for a Dag that has a team, and the scheduler cancels no run of such a Dag: a +request recorded before its Dag got a team lists, with the reason ``dag_has_team``, the runs about its key +that the scheduler finds when it applies the request; a run that the scheduler creates later, from an +earlier event, is left alone without an entry. The Dag gets no warning, but ``airflow webhooks check`` +reports each Dag with a team whose ``cancel_on`` names a webhook, as a problem with that webhook. + +Following a cancel +------------------ + +* **The delivery.** Under **Browse → Webhook Deliveries**, a delivery that recorded cancel requests has the + outcome ``cancel_requested``, unless it also recorded topic or asset route events, which makes it + ``accepted``, or went to a webhook wait, which makes it ``resumed_wait``, or becomes ``expired``, keeping + its cancel requests, if newer deliveries kept for webhook waits pushed it out (see + :ref:`inbound-webhooks-outcomes`). Its details list the requests under **Cancel Requests**: for each, the + Dag; the scope, **Whole run** or **Tasks**, and the tasks it ends, each with the state it ends them in; the + status, **Pending**, **Applied** or **Failed**, with what it means, and when the scheduler applied it or + failed to; the runs and the task instances it cancelled, each linked to its page; the runs about its key + that it left alone, with why; and the error of a failed request. A request whose record ``airflow db clean`` + removed, or that went with its Dag, is listed as gone. The REST API returns the same, as described in + :ref:`inbound-webhooks-deliveries`. +* **The Dag and its tasks.** The Dag's **Details** tab lists its conditions in its **Cancel On** row, and + the **Details** tab of a task instance lists those of its task, with the state that they end it in. + Both show the Dag's latest version. +* **The run.** A run that the Dag's ``cancel_on`` cancelled gets the note + ``Cancelled by webhook <name> delivery <id>``, where ``<id>`` is the sender's id for the delivery, or the + id of Airflow's record of it if the sender gave none. The note is written only if the run has none. +* **The audit log.** The scheduler writes an entry with the event ``webhook_cancel`` and the owner + ``webhook:<name>`` for each run that a request cancels, and for each task instance that a task's + ``cancel_on`` ends, which the users allowed to read the Dag's audit log see (see + :ref:`inbound-webhooks-audit-log`). +* **Metrics.** The scheduler counts the runs and task instances it cancels in ``webhooks.cancels``, tagged + ``webhook`` and ``scope`` (see :ref:`inbound-webhooks-metrics`). +* **The command line.** ``airflow webhooks check`` lists which Dags and tasks each webhook cancels, as in + ``Webhook github cancels the runs of Dag deploy_preview on pull_request deliveries with action closed, + repository my-org/web, and manual runs by conf key pr``. + +When a run was not cancelled, :ref:`inbound-webhooks-not-cancelled` lists the reasons. + +Testing a Dag that cancels +-------------------------- + +Only a running scheduler applies cancel requests, so ``airflow dags test`` does not cancel its run. To try +``deploy_preview`` with a scheduler, trigger it as run C above: + +.. code-block:: bash + + airflow dags trigger deploy_preview --conf '{"pr": "my-org/web#42"}' + +and, while the run is still running, send a closed pull request to the webhooks process, here on port +8081, with ``airflow webhooks send-test``, described in :ref:`inbound-webhooks-cli`: + +.. code-block:: bash + + airflow webhooks send-test github \ + --url http://localhost:8081/webhooks/github \ + --signature-header X-Hub-Signature-256 \ + --signature-prefix sha256= \ + --header 'X-GitHub-Event: pull_request' \ + --header 'X-GitHub-Delivery: test-close-42' \ + --data '{"action": "closed", "number": 42, "repository": {"full_name": "my-org/web"}, + "pull_request": {"number": 42, "merged": false}}' + +It prints ``{"status": 202, "body": {"delivery_id": "test-close-42", "outcome": "cancel_requested"}}``, +unless a topic or a webhook wait also took the delivery, and the run is ``failed`` a second or two later, +with the note ``Cancelled by webhook github delivery test-close-42``. A test delivery is processed like any +other, so it cancels every run about ``my-org/web#42`` of every Dag whose ``cancel_on`` it matches. diff --git a/airflow-core/docs/authoring-and-scheduling/webhook-topics.rst b/airflow-core/docs/authoring-and-scheduling/webhook-topics.rst index 2418747f11d..c63e6c5d4c4 100644 --- a/airflow-core/docs/authoring-and-scheduling/webhook-topics.rst +++ b/airflow-core/docs/authoring-and-scheduling/webhook-topics.rst @@ -36,7 +36,8 @@ verifies each delivery and describes it. For each delivery, Airflow builds a lis records an asset event on each of them that a Dag schedules on. A Dag subscribes to deliveries by scheduling on a topic, a :class:`~airflow.sdk.WebhookTopic`, and each delivery then starts a run of it, with the delivery in ``triggering_asset_events``. To have a task of a running Dag run wait for a delivery -instead, see :doc:`webhook-waits`. +instead, see :doc:`webhook-waits`, and to have a later delivery cancel the runs that a delivery started, +see :doc:`webhook-cancel`. Topic names ----------- @@ -126,17 +127,18 @@ It accepts ``access_control`` like any asset. When a delivery records events ------------------------------ -A delivery records an event on one of its names only if a Dag that is not paused, draining or stale -schedules on that name, directly or through ``Asset.ref``. Otherwise it records no event on that name. A -delivery that records no event, on any of its names or through an -:ref:`asset route <inbound-webhooks-asset-routes>`, and goes to no :doc:`webhook wait <webhook-waits>` has -the outcome ``early`` if it has a correlation key, which keeps it for webhook waits, or else -``dag_paused`` if only paused or draining Dags schedule on its names or on its routes' assets, or -``no_subscribers`` if none do (see :ref:`inbound-webhooks-outcomes`). Unpausing a Dag does not replay the -deliveries received while it was paused. If the sender delivers one of them again, the copy is processed -again in full when the delivery recorded nothing, with outcome ``dag_paused`` or ``early``, so the unpaused -Dag receives it. It is answered as a duplicate when the delivery started another Dag's run or went to a -webhook wait, so in those cases the unpaused Dag never receives it. +A delivery records an event on one of its names only if a Dag that is not paused, draining or stale schedules +on that name, directly or through ``Asset.ref``. Otherwise it records no event on that name. A delivery that +records no event, on any of its names or through an :ref:`asset route <inbound-webhooks-asset-routes>`, and +goes to no :doc:`webhook wait <webhook-waits>` has the outcome ``cancel_requested`` if it has a correlation +key and recorded cancel requests (see :doc:`webhook-cancel`), or ``early`` if it has one and recorded none, +either of which keeps it for webhook waits, or else ``dag_paused`` if only paused or draining Dags schedule on +its names or on its routes' assets, or ``no_subscribers`` if none do (see :ref:`inbound-webhooks-outcomes`). +Unpausing a Dag does not replay the deliveries received while it was paused. If the sender delivers one of +them again, the copy is processed again in full when the delivery recorded no event and went to no wait, with +outcome ``dag_paused``, ``early`` or ``cancel_requested``, so the unpaused Dag receives it. It is answered as +a duplicate when the delivery started another Dag's run or went to a webhook wait, so in those cases the +unpaused Dag never receives it. * ``Asset.ref(name=...)`` and ``Asset.ref(uri=...)`` subscribe only to a topic that some Dag declares, because deliveries never create assets and events are recorded only on active assets. @@ -237,6 +239,9 @@ Runs only through ``Asset.ref`` are not swept in that way. * ``dag_run.conf`` is ``{}``: the delivery is in ``triggering_asset_events``, not in the configuration. A run triggered by hand has no triggering asset events. +* A run is about the correlation key of the delivery that started it, which its topic event keeps, so a + later delivery with the same key, such as the one that closes the pull request whose label started the + run, can cancel it through the ``cancel_on`` of the Dag or of its tasks (see :doc:`webhook-cancel`). * Runs are created in the order the deliveries were recorded, and may finish in any order. A burst of deliveries queues behind ``max_active_runs``; the webhook's rate limit is the only bound on how many can queue. diff --git a/airflow-core/docs/authoring-and-scheduling/webhook-waits.rst b/airflow-core/docs/authoring-and-scheduling/webhook-waits.rst index 96fec451a29..dbb40330a07 100644 --- a/airflow-core/docs/authoring-and-scheduling/webhook-waits.rst +++ b/airflow-core/docs/authoring-and-scheduling/webhook-waits.rst @@ -288,10 +288,11 @@ delivery has a correlation key, went to no other wait and recorded no topic or a connection, records an event on a data asset for the deliveries that offer its topic name, so that the Dags scheduled on the asset run. -A kept delivery's outcome is ``early``, and the first wait that matches it and is registered within -:ref:`[webhooks] early_delivery_ttl <config:webhooks>` seconds of its receipt, 600 by default, takes it: the -task resumes as soon as it parks, without waiting. A wait that several early deliveries match takes the -oldest. +A kept delivery's outcome is ``early``, or ``cancel_requested`` if it also recorded cancel requests for +Dags whose ``cancel_on`` it matches (see :doc:`webhook-cancel`), and the first wait that matches it and is +registered within :ref:`[webhooks] early_delivery_ttl <config:webhooks>` seconds of its receipt, 600 by +default, takes it: the task resumes as soon as it parks, without waiting. A wait that several kept +deliveries match takes the oldest. The correlation key says what a delivery is about, and only its source type sets it: @@ -307,13 +308,17 @@ A wait does not match on the correlation key: it only decides which deliveries a after it never gets it, even one registered while the delivery was being recorded. Do not schedule Dags on the deliveries that a wait may have to get early, and have the Deployment Manager keep asset routes off them, since a route that records an event for such a delivery has the same effect. -* A webhook keeps at most :ref:`[webhooks] early_delivery_max_per_webhook <config:webhooks>` early - deliveries received within ``early_delivery_ttl``, 1000 by default. Keeping one more marks the oldest - ``expired``, and drops its payload, so that no wait gets it. - -The sender can deliver an ``early`` or ``expired`` delivery again, as from the **Recent Deliveries** page -of a GitHub webhook. The repeat is processed again in full, so it goes to a wait registered since, even -after ``early_delivery_ttl``, and a Dag that subscribed or was unpaused since receives it. +* A delivery that recorded cancel requests is kept all the same. A merge gate on a pull request therefore + still gets the delivery that closes the pull request when the same delivery also matches the + ``cancel_on`` of a Dag that deploys a preview of it, and cancels that Dag's runs. +* A webhook keeps at most :ref:`[webhooks] early_delivery_max_per_webhook <config:webhooks>` deliveries + for waits, ``early`` or ``cancel_requested``, received within ``early_delivery_ttl``, 1000 by default. + Keeping one more marks the oldest ``expired``, and drops its payload, so that no wait gets it; the + cancel requests it recorded stay. + +The sender can deliver an ``early``, ``cancel_requested`` or ``expired`` delivery again, as from the **Recent +Deliveries** page of a GitHub webhook. The repeat is processed again in full, so it goes to a wait registered +since, even after ``early_delivery_ttl``, and a Dag that subscribed or was unpaused since receives it. Outcomes -------- @@ -324,15 +329,22 @@ A delivery's outcome, which its record shows, tells what it did, in this order ( * ``resumed_wait``: it went to at least one wait, whether or not it also recorded topic or asset route events. * ``accepted``: it recorded topic or asset route events and went to no wait. It is never kept for a wait. -* ``early``: it has a correlation key, recorded no event and went to no wait, so it is kept for a wait, as - described above. -* ``expired``: it was ``early`` until the webhook kept too many newer early deliveries. +* ``cancel_requested``: it has a correlation key, recorded no event, went to no wait, and recorded cancel + requests for Dags whose ``cancel_on`` it matches (see :doc:`webhook-cancel`), so it is kept for a wait, + as described above. +* ``early``: it has a correlation key, recorded no event, requested no cancel and went to no wait, so it is + kept for a wait, as described above. +* ``expired``: it was kept for waits, ``early`` or ``cancel_requested``, until the webhook kept too many + newer such deliveries. The cancel requests it recorded, if any, stay. * ``dag_paused`` or ``no_subscribers``: it has no correlation key, recorded no event and went to no wait. -A repeat of a ``resumed_wait`` or ``accepted`` delivery is answered as a duplicate and does nothing, while -a repeat of any other is processed again in full. A delivery with a correlation key that only paused Dags -subscribe to is therefore ``early`` rather than ``dag_paused``, and its repeat, after the Dags are -unpaused, starts them unless a wait has taken the delivery meanwhile. +Whatever its outcome, a delivery's record also lists the cancel requests that it recorded. + +A repeat of a ``resumed_wait`` or ``accepted`` delivery is answered as a duplicate and does nothing, while a +repeat of any other is processed again in full, without recording a second cancel request for any Dag. A +delivery with a correlation key that only paused Dags subscribe to is therefore ``early`` rather than +``dag_paused``, and its repeat, after the Dags are unpaused, starts them unless a wait has taken the delivery +meanwhile. Retries, clears and waiting twice --------------------------------- @@ -349,17 +361,23 @@ A wait belongs to one try of its task instance: has passed. Meanwhile the task instance's **Details** tab says that no delivery will resume it, and the ``webhooks.open_waits`` metric leaves such waits out. * A matching delivery ends the ``registered`` wait of a try that no longer runs or waits, and is handled as - if that wait did not exist: if it has a correlation key and recorded no event, it is kept ``early``, so - the new try's wait can take it. + if that wait did not exist: if it has a correlation key and recorded no event, it is kept ``early``, or + ``cancel_requested`` if it recorded cancel requests, so the new try's wait can take it. * A new try that registers a wait on the same key ends the earlier try's wait. If that wait had received a delivery that never resumed its task, as when the try failed between the delivery and its park, the delivery is taken back from it: it stays ``resumed_wait`` if other waits hold it, becomes ``accepted`` if - it recorded topic or asset route events, so that no wait can take it, or becomes ``early`` again if it has - a correlation key, so that the new try can take it within ``early_delivery_ttl`` of when it was first - received. A delivery with none of these keeps the ended wait. + it recorded topic or asset route events, so that no wait can take it, or, if it has a correlation key, + becomes ``cancel_requested`` again if it recorded cancel requests and ``early`` again if not, so that the + new try can take it within ``early_delivery_ttl`` of when it was first received. A delivery with none of + these keeps the ended wait. * A try can wait more than once, such as an operator that waits again in the method that it resumes in. Each registration ends the try's open wait, so a later wait never resumes with a delivery that matched an earlier one. +* A webhook cancel request that ends the task, through the ``cancel_on`` of its Dag or of the task itself, + ends the try's open wait too, with the status ``ended``, and the task is not resumed: it ends ``failed``, + or in its ``cancel_state``, as described in :doc:`webhook-cancel`. A parked task has no process, so + neither its resume method nor its ``on_kill`` runs. A cancel of the whole run calls the Dag's + ``on_failure_callback`` with the reason ``cancelled``, where such cleanup can go. A delivery's record can therefore list a wait that ended before its task resumed, as when a later wait of the same try replaced it, or when a new try took over its key and the delivery had none of the outcomes diff --git a/airflow-core/docs/public-airflow-interface.rst b/airflow-core/docs/public-airflow-interface.rst index 8b946f159c9..8a9c5ef83ac 100644 --- a/airflow-core/docs/public-airflow-interface.rst +++ b/airflow-core/docs/public-airflow-interface.rst @@ -481,21 +481,31 @@ in ``airflow.webhooks.naming``, which are public too. _api/airflow/webhooks/naming/index The rest of the inbound webhooks interface is public and experimental too, and is described in -:doc:`administration-and-deployment/webhooks`, :doc:`authoring-and-scheduling/webhook-topics` and -:doc:`authoring-and-scheduling/webhook-waits`: the routes ``POST /webhooks/<name>`` and -``GET /webhooks/_health`` with their status codes and the body of a 202, the topic name grammar, the keys of -the extra of topic events, which asset routes record events on and the reasons they skip, the -``[webhooks]`` options, ``GET /api/v2/webhooks``, ``GET /api/v2/webhooks/{webhook_name}/deliveries`` with -``~`` for every webhook, the outcomes of a delivery and the shape of its ``actions``, +:doc:`administration-and-deployment/webhooks`, :doc:`authoring-and-scheduling/webhook-topics`, +:doc:`authoring-and-scheduling/webhook-waits` and :doc:`authoring-and-scheduling/webhook-cancel`: the +routes ``POST /webhooks/<name>`` and ``GET /webhooks/_health`` with their status codes and the body of a +202, the topic name grammar, the keys of the extra of topic events, which asset routes record events on +and the reasons they skip, the ``[webhooks]`` options, ``GET /api/v2/webhooks``, +``GET /api/v2/webhooks/{webhook_name}/deliveries`` with ``~`` for every webhook, the outcomes of a delivery, +``cancel_requested`` included, the shape of its ``actions``, and its ``cancel_requests`` with their +targets, statuses and results, ``cancel_on`` in the responses of +``GET /api/v2/dags/{dag_id}/details`` and of the task routes, with ``cancel_state`` in the latter, ``GET /api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/webhookWaits`` and the statuses of -a webhook wait, the match grammar of a wait's ``events`` and ``match``, the ``airflow webhooks`` commands, -the webhook metrics and their tags, and the owner of the audit log entries. +a webhook wait, the match grammar of a wait's ``events`` and ``match``, which runs a cancel request applies +to and what it does to them, the reason ``cancelled`` that a Dag's ``on_failure_callback`` gets for a +cancel, the ``airflow webhooks`` commands, the webhook metrics and their tags, ``webhooks.cancels`` +included, and the owner of the audit log entries and their events, ``webhook_delivery`` and +``webhook_cancel``. :class:`~airflow.sdk.WebhookTopic` is part of the Task SDK, and so are ``BaseOperator.wait_for_webhook``, ``WebhookWaitUnsupported`` in ``airflow.sdk.exceptions``, and the shapes of the events that a waiting task resumes with: ``{"webhook", "event", "delivery_id", "received_at", "attributes", "payload"}`` for a delivery that matches its wait, and ``{"error", "error_type": "timeout"}`` when none does before the -timeout (see :ref:`webhook-waits-resume-event`). The Execution API route that registers a wait is internal. +timeout (see :ref:`webhook-waits-resume-event`). So are :class:`~airflow.sdk.WebhookMatch`, the +``cancel_on`` argument of ``DAG``, and the ``cancel_on`` and ``cancel_state`` arguments of +``BaseOperator`` (see :doc:`authoring-and-scheduling/webhook-cancel`). The Execution API route that +registers a wait is internal, and so are the ``dag_cancel_condition`` and ``webhook_cancel_request`` +tables and the text of the note that a cancel writes on a run. Each source type's name, the ``source_type`` key that selects it in a connection's extra, its own connection extra options, its attributes and the fields its topic names are built from are the public @@ -509,6 +519,11 @@ documentation. .. versionadded:: 3.5.0 +.. versionchanged:: 3.6.0 + Added ``WebhookMatch`` and ``cancel_on``, with the cancel requests of deliveries, the + ``cancel_requested`` outcome, the ``cancelled`` callback reason, the ``webhook_cancel`` audit log event + and the ``webhooks.cancels`` metric. + Secrets Backends ---------------- diff --git a/airflow-core/docs/security/audit_logs.rst b/airflow-core/docs/security/audit_logs.rst index 2cad3a36ec0..70b2355bf5f 100644 --- a/airflow-core/docs/security/audit_logs.rst +++ b/airflow-core/docs/security/audit_logs.rst @@ -213,9 +213,15 @@ Webhook Events ~~~~~~~~~~~~~~ - ``webhook_delivery``: A delivery to an :doc:`inbound webhook </administration-and-deployment/webhooks>` - recorded topic or asset route events, as an entry without a Dag, or went to a task's + recorded topic or asset route events or + :doc:`cancel requests </authoring-and-scheduling/webhook-cancel>`, as an entry without a Dag whose extra + lists the requests as ``cancel_requests``, or went to a task's :doc:`webhook wait </authoring-and-scheduling/webhook-waits>`, as an entry of that task instance. Its owner is ``webhook:<name>``. +- ``webhook_cancel``: The scheduler applied a webhook delivery's cancel request to a Dag run, as an entry + of that run, or to a task instance, as an entry of that task instance. Its owner is ``webhook:<name>``, + and its extra has the ``delivery_id``, the ``request_id``, the ``scope``, ``run`` or ``task``, and for a + task instance the ``state`` it ended in. User Action Events ~~~~~~~~~~~~~~~~~~ diff --git a/airflow-core/docs/security/security_model.rst b/airflow-core/docs/security/security_model.rst index 3fb8e6785a6..763fc04a4dc 100644 --- a/airflow-core/docs/security/security_model.rst +++ b/airflow-core/docs/security/security_model.rst @@ -1208,7 +1208,9 @@ The design leaves the following residual risks, which are not vulnerabilities: * **A leaked shared secret.** Whoever holds a webhook's shared secret can send deliveries that pass as the sender's, for any event and with any attributes, until the secret is rotated by editing the ``webhook_<name>`` connection. Such deliveries start the Dags that subscribe to their topics or that are - scheduled on the assets their asset routes name, and resume the tasks whose webhook waits they match. + scheduled on the assets their asset routes name, resume the tasks whose webhook waits they match, and + fail the active runs, or end tasks, of the Dags whose ``cancel_on`` names the webhook, for any + correlation key they choose. * **A delivery replayed under a new id.** A delivery is recognized as a repeat by its delivery id together with a digest of its body. Some senders, such as GitHub and Jira, sign only the body, others send a token that covers nothing, and the delivery id travels in a header, so a captured delivery can @@ -1237,6 +1239,22 @@ The design leaves the following residual risks, which are not vulnerabilities: asset that the route names, other than a webhook topic, and start the Dags scheduled on it, while ``POST /api/v2/assets/events`` requires permission to create events on the asset. Edit rights on a webhook connection are therefore also rights to record asset events. +* **Cancel by delivery.** A delivery can request cancels only for the Dags whose code declares a + ``cancel_on`` condition on its webhook that it matches, and the scheduler applies a request only to the + queued and running runs of such a Dag that are about the delivery's correlation key, through an event + that an earlier delivery recorded, including runs the scheduler creates later from such an event, or + through their conf if they were queued before the delivery (see + :doc:`/authoring-and-scheduling/webhook-cancel`). Before ``cancel_on``, a leaked secret could only + start runs and resume waiting tasks; with it, it can also fail runs. A generic source's + ``correlation_key_field`` lets the sender choose the key, as the holder of the secret can for any source + type. Whoever triggers a run with a conf chooses its conf key, which only exposes their own run to cancels. + The key of a run started by an event counts only for an event that a delivery recorded: an event that + the REST API, an asset watcher's trigger or a task created never makes a run about a key, whatever its + extra says. A delivery replayed under a new id records new requests, which cancel the runs about its key + that started since the original. The webhooks process writes no run or task instance state for a + cancel, and loads no Dag: it reads the conditions that the Dag processor stored and records requests. The + scheduler reads the requests that the webhooks process records, and runs no Dag code to apply them; the + Dag's ``on_failure_callback`` runs in the Dag processor, as for a run that timed out. * **Credentials held by an exposed process.** The process that serves inbound webhooks holds the metadata database credentials, the credentials of any secrets backend it reads, and the Fernet key unless it is withheld because no ``webhook_<name>`` connection is stored in the metadata database, diff --git a/airflow-core/newsfragments/aip125-cancel-on.significant.rst b/airflow-core/newsfragments/aip125-cancel-on.significant.rst new file mode 100644 index 00000000000..3f3afcdf8b3 --- /dev/null +++ b/airflow-core/newsfragments/aip125-cancel-on.significant.rst @@ -0,0 +1,9 @@ +AIP-125: Cancelling runs and tasks from inbound webhooks (experimental) + +Dags and tasks can declare which deliveries to Airflow's inbound webhooks cancel their runs, with ``cancel_on=[WebhookMatch(...)]``, ``WebhookMatch`` being a new class in ``airflow.sdk`` that names a webhook, events and attributes with the grammar of webhook waits. Only a delivery with a correlation key cancels: it records a cancel request for each Dag whose conditions it matches, and the scheduler applies the request, within a second or two of the delivery's answer, to the Dag's queued [...] + +``airflow db migrate`` adds two tables: ``webhook_cancel_request``, which holds the cancel requests of deliveries and which ``airflow db clean`` cleans by the ``retain_until`` of their delivery, and ``dag_cancel_condition``, which holds the ``cancel_on`` conditions of Dags, which the Dag processor keeps in step with each Dag and which go with their Dag. A Dag that uses ``cancel_on`` needs this release in the Task SDK of the Dag processor and the workers as well as in Airflow core, so upg [...] + +A delivery gets the new outcome ``cancel_requested`` when it recorded cancel requests but no topic or asset route events and went to no webhook wait. Such a delivery is kept for webhook waits as an ``early`` delivery is, and counts toward ``[webhooks] early_delivery_max_per_webhook``. When newer deliveries push it out, it becomes ``expired`` and its cancel requests stay. A redelivery of it is processed again in full, answered 202 ``cancel_requested``, and records no second cancel request [...] + +See :doc:`/authoring-and-scheduling/webhook-cancel` and :doc:`/administration-and-deployment/webhooks`. diff --git a/providers/github/docs/webhooks.rst b/providers/github/docs/webhooks.rst index 40f22799113..0e97fb68560 100644 --- a/providers/github/docs/webhooks.rst +++ b/providers/github/docs/webhooks.rst @@ -213,12 +213,20 @@ and through the REST API, with the ``#`` written as ``%23``: ``GET /api/v2/webhooks/github/deliveries?correlation_key=my-org/web%237``. Webhook waits use the key to keep deliveries that arrive before the wait that they are for is registered. -A delivery with a key that records no topic or asset route event and goes to no wait is kept as ``early`` -for ``[webhooks] early_delivery_ttl`` seconds, so that a task that waits for the pull request or issue it -is about still gets it, as when a merge gate's pull request is merged before the task registers its wait -(see :ref:`apache-airflow:webhook-waits-early`). A wait does not match on the key itself: its ``match`` -names the ``repository`` and ``number`` attributes instead. A delivery without a key, such as a ``push``, -that arrives before its wait is not kept. +A delivery with a key that records no topic or asset route event and goes to no wait is kept as ``early``, +or ``cancel_requested`` if it recorded cancel requests, for ``[webhooks] early_delivery_ttl`` seconds, so +that a task that waits for the pull request or issue it is about still gets it, as when a merge gate's pull +request is merged before the task registers its wait (see :ref:`apache-airflow:webhook-waits-early`). A +wait does not match on the key itself: its ``match`` names the ``repository`` and ``number`` attributes +instead. A delivery without a key, such as a ``push``, that arrives before its wait is not kept. + +A Dag's or a task's ``cancel_on`` uses the key to find the runs that a delivery is about (see +:ref:`apache-airflow:webhook-cancel`). A run that a delivery about pull request 7 of ``my-org/web`` +started, through a topic, is about ``my-org/web#7``, so a later delivery that closes the pull request, and +matches a ``cancel_on`` condition of the run's Dag, fails the run. A run triggered with +``{"pr": "my-org/web#7"}`` in its conf is about the same key when the condition names ``pr`` as its +``correlation_key_conf``. A delivery without a key, such as a ``push`` or a ``workflow_run``, cancels +nothing. Stored payload -------------- diff --git a/providers/standard/docs/connections/inbound_webhook.rst b/providers/standard/docs/connections/inbound_webhook.rst index 0f389bafa77..7fa5d6627e1 100644 --- a/providers/standard/docs/connections/inbound_webhook.rst +++ b/providers/standard/docs/connections/inbound_webhook.rst @@ -127,10 +127,14 @@ Extra (optional) each topic event and asset route event. Deliveries can be filtered by it in the UI, in the **Deliveries** tab of **Browse → Webhook Deliveries**, and through the REST API. Webhook waits use it to keep deliveries that arrive before their wait: a delivery with a correlation key that records no - topic or asset route event and goes to no wait is kept as ``early`` for - ``[webhooks] early_delivery_ttl`` seconds, with its payload, for a wait registered later (see - :ref:`apache-airflow:webhook-waits-early`). A wait matches attributes, not the key, so map the same - field into ``attribute_paths`` too if waits are to match it. + topic or asset route event and goes to no wait is kept as ``early``, or ``cancel_requested`` if it + recorded cancel requests, for ``[webhooks] early_delivery_ttl`` seconds, with its payload, for a wait + registered later (see :ref:`apache-airflow:webhook-waits-early`). A wait matches attributes, not the + key, so map the same field into ``attribute_paths`` too if waits are to match it. + + A Dag's or a task's ``cancel_on`` acts only on deliveries that have a correlation key, and cancels the + runs about that key (see :ref:`apache-airflow:webhook-cancel`), so set ``correlation_key_field`` on the + connection of a webhook whose deliveries are to cancel runs. Without it, they request no cancel. Prefer the fields when the service puts these values in the body. The ``hmac_json`` signature covers the body only, and a token covers nothing, so whoever captures a delivery can send its body diff --git a/providers/standard/docs/operators/webhooks.rst b/providers/standard/docs/operators/webhooks.rst index 9326bcef6ba..671cc10d5b5 100644 --- a/providers/standard/docs/operators/webhooks.rst +++ b/providers/standard/docs/operators/webhooks.rst @@ -66,12 +66,13 @@ Within a webhook, only one running or parked task instance can wait with the sam ``match``. The API server refuses a second such wait with 409, and its task fails. A Dag that needs the same delivery in several tasks can schedule on a webhook topic instead. -A delivery that arrives before its wait is registered is kept for it if the delivery has a correlation -key, such as each GitHub delivery about a pull request or issue, went to no other wait and recorded no -topic or asset route events. Its outcome is then ``early``, and a wait that matches it and is registered -within ``[webhooks] early_delivery_ttl`` of its receipt takes it: the task resumes as soon as it parks. A -delivery that recorded topic or asset route events for a Dag is not kept, so a wait registered after it -never gets it. :ref:`apache-airflow:inbound-webhooks-outcomes` describes the outcomes. +A delivery that arrives before its wait is registered is kept for it if the delivery has a correlation key, +such as each GitHub delivery about a pull request or issue, went to no other wait and recorded no topic or +asset route events. Its outcome is then ``early``, or ``cancel_requested`` if it also recorded cancel +requests, and a wait that matches it and is registered within ``[webhooks] early_delivery_ttl`` of its receipt +takes it: the task resumes as soon as it parks. A delivery that recorded topic or asset route events for a Dag +is not kept, so a wait registered after it never gets it. :ref:`apache-airflow:inbound-webhooks-outcomes` +describes the outcomes. .. _howto/decorator:webhook_wait:
