shahar1 commented on code in PR #53009:
URL: https://github.com/apache/airflow/pull/53009#discussion_r4500398457
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -32,15 +32,73 @@
MSGraphTrigger,
ResponseSerializer,
)
+from airflow.providers.microsoft.azure.version_compat import AIRFLOW_V_3_3_PLUS
+from airflow.triggers.base import StartTriggerArgs
if TYPE_CHECKING:
+ from datetime import timedelta
from io import BytesIO
from msgraph_core import APIVersion
+ from airflow.providers.microsoft.azure.sensors.msgraph import MSGraphSensor
from airflow.sdk import Context
+def _is_plain_data(value: Any) -> bool:
+ if isinstance(value, dict):
+ return all(isinstance(key, str) and _is_plain_data(item) for key, item
in value.items())
+ if isinstance(value, (list, tuple)):
+ return all(_is_plain_data(item) for item in value)
+ return value is None or isinstance(value, (str, int, float, bool))
+
+
+def build_start_trigger_args(
+ operator: MSGraphAsyncOperator | MSGraphSensor, timeout: timedelta | None
= None
+) -> StartTriggerArgs | None:
+ """
+ Build the arguments which start the given operator from the triggerer,
when that is possible.
+
+ The triggerer renders the templated fields of the trigger itself, but only
since Airflow 3.3,
+ and it only gets what the serialized Dag holds. It never resolves an
``XComArg``, and a
+ callable or a file-like object cannot be serialized. None is returned in
those cases, so the
+ operator starts on a worker instead.
+
+ :param operator: The operator to build the arguments for.
+ :param timeout: How long the trigger may run before the task times out
(default is None).
+ """
+ if not AIRFLOW_V_3_3_PLUS:
+ return None
+
+ serializer = type(operator.serializer)
+ trigger_kwargs = {
+ "url": operator.url,
+ "response_type": operator.response_type,
+ "path_parameters": operator.path_parameters,
+ "url_template": operator.url_template,
+ "method": operator.method,
+ "query_parameters": operator.query_parameters,
+ "headers": operator.headers,
+ "data": operator.data,
+ "conn_id": operator.conn_id,
+ "timeout": operator.timeout,
Review Comment:
**Minor: the sensor timeout should not reach the trigger any more.**
#75774 removed `timeout=self.timeout` from the trigger the sensor defers, as
the sensor timeout is not an HTTP timeout. This shared helper still passes it,
so the first poll would use it as the HTTP timeout and the later polls none.
`test_start_from_trigger_defers_the_same_trigger_as_execute` for the sensor
should fail on this after the rebase. The fix belongs here, not in that test.
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -32,15 +32,73 @@
MSGraphTrigger,
ResponseSerializer,
)
+from airflow.providers.microsoft.azure.version_compat import AIRFLOW_V_3_3_PLUS
+from airflow.triggers.base import StartTriggerArgs
if TYPE_CHECKING:
+ from datetime import timedelta
from io import BytesIO
from msgraph_core import APIVersion
+ from airflow.providers.microsoft.azure.sensors.msgraph import MSGraphSensor
from airflow.sdk import Context
+def _is_plain_data(value: Any) -> bool:
+ if isinstance(value, dict):
+ return all(isinstance(key, str) and _is_plain_data(item) for key, item
in value.items())
+ if isinstance(value, (list, tuple)):
+ return all(_is_plain_data(item) for item in value)
+ return value is None or isinstance(value, (str, int, float, bool))
+
+
+def build_start_trigger_args(
+ operator: MSGraphAsyncOperator | MSGraphSensor, timeout: timedelta | None
= None
+) -> StartTriggerArgs | None:
+ """
+ Build the arguments which start the given operator from the triggerer,
when that is possible.
+
+ The triggerer renders the templated fields of the trigger itself, but only
since Airflow 3.3,
+ and it only gets what the serialized Dag holds. It never resolves an
``XComArg``, and a
+ callable or a file-like object cannot be serialized. None is returned in
those cases, so the
+ operator starts on a worker instead.
+
+ :param operator: The operator to build the arguments for.
+ :param timeout: How long the trigger may run before the task times out
(default is None).
+ """
+ if not AIRFLOW_V_3_3_PLUS:
Review Comment:
**Major: on Airflow 3.3.x a retry or a clear reads the XComs of the earlier
attempt.**
`ti_run` only asks the worker to clear XComs under `if not ti.next_method:`,
and a task coming from `defer_task` already has `next_method` set on its first
worker run. On 3.3.x the XCom rows are keyed by Dag, task, run and map index,
and nothing else removes them on a retry or a clear. `execute_complete` then
starts from `results = self.pull_xcom(context=context)`, so:
- a retry after a failure on page 2 returns page 1 of the earlier attempt
plus page 1 again;
- clearing a finished task reads its earlier return value. For a dict
response, `list(value)` gives its keys, and the task returns `[*keys,
response]`.
3.4 keys XComs per attempt and rotates the task instance id, so this only
affects 3.3.x, and only the operator: the sensor does not pull XComs.
The smallest fix is to raise this gate to 3.4. If 3.3 support matters,
`execute_complete` needs to know it is handling the first page, for instance
through a marker in the `defer_to_next_link` kwargs, and start from an empty
list then. Either way, please add a test.
##########
providers/microsoft/azure/docs/operators/msgraph.rst:
##########
@@ -80,6 +80,40 @@ Below is an example of using this operator to create an item
schedule in Fabric.
:start-after: [START howto_operator_ms_fabric_create_item_schedule]
:end-before: [END howto_operator_ms_fabric_create_item_schedule]
+.. _howto/operator:MSGraphAsyncOperator:start_from_trigger:
+
+Starting from the triggerer
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The operator does all its requests in the triggerer, so the only thing it does
on a worker at the start
+is to hand the first request over. With ``start_from_trigger=True`` the
scheduler defers the task itself
+and that first worker run is skipped. The worker gets involved once the
response is there, to process it
+and to follow the pagination.
+
+.. code-block:: python
+
+ site_task = MSGraphAsyncOperator(
+ task_id="news_site",
+ conn_id="msgraph_api",
+ url="sites/{{ params.site }}",
+ start_from_trigger=True,
+ )
+
+This needs Airflow 3.3 or later, on older versions the task starts on a worker
as it does without the
+argument. It also starts on a worker when:
+
+* an argument which is passed on to the trigger is an ``XComArg``, a callable
or a file-like object,
+ as the triggerer cannot resolve or receive those;
+* the task is mapped with ``expand``.
+
+Keep the following in mind when you enable it:
Review Comment:
**Minor: the second render pass is only described in the PR description.**
The triggerer takes `template_fields` from `task.start_trigger_args` for
every trigger of the task, so the triggers `execute_complete` and
`retry_execute` defer are rendered again. That covers values already rendered
on the worker, and also the `nextLink` and query parameters which come from a
Graph response. A value holding Jinja markers then changes the request or fails
it.
I would rather fix this than leave it: `MSGraphTrigger` could take a flag
for triggers built from rendered values, with `render_template_fields` doing
nothing when it is set. If it stays, users need to read about it in this list.
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -161,6 +224,8 @@ def __init__(
self.result_processor = result_processor
self.event_handler = event_handler or default_event_handler
self.serializer: ResponseSerializer = serializer()
+ self.start_trigger_args = build_start_trigger_args(self) if
start_from_trigger else None
Review Comment:
**Major: `durable=True` loses its crash recovery here (needs handling in the
rebase).**
Since #74675, `execute()` is where a durable task reads its state: it
returns the kept result of a completed non-GET request, and it resumes from the
cursor. A start-from-trigger retry never calls `execute()`, because
`DagRun.schedule_tis` goes to `TaskInstance.defer_task`, which creates the
first-page trigger again. With `durable=True, start_from_trigger=True`, a retry
after a crash then:
- sends a non-GET request again even though it completed, which is what
`_COMPLETED_KEY` exists to prevent;
- fetches page 1 again, and `push_page` stores it at `cursor["pages"]`,
after the pages already kept, so the final result holds those pages twice.
Suggested fix: a durable task starts on a worker, like the other fallbacks.
Please add it to the docs and cover it with a test.
```python
self.start_trigger_args = (
build_start_trigger_args(self) if start_from_trigger and not durable
else None
)
```
##########
providers/microsoft/azure/docs/sensors/msgraph.rst:
##########
@@ -33,6 +33,24 @@ Below is an example of using this sensor to poll the status
of a PowerBI workspa
:start-after: [START howto_sensor_powerbi_scan_status]
:end-before: [END howto_sensor_powerbi_scan_status]
+Starting from the triggerer
+^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+With ``start_from_trigger=True`` the scheduler defers the sensor itself, so
the first poll is done without
+a worker being involved. The ``timeout`` of the sensor also applies to that
first poll. The requirements
Review Comment:
**Minor: the timeout budget restarts after the first poll.**
#75774 carries `sensor_started_at` through the trigger kwargs so
re-deferring does not reset the timeout. `StartTriggerArgs` has no such value,
because it is built at parse time, so `execute_complete` gets `None` and
`_started_at` uses the current time. The first poll can use the whole `timeout`
under `trigger_timeout`, and the budget then starts over, so the total can
reach about twice `timeout`. Either record when the first poll started, or say
this here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
