Repository navigation
Conversation
Waiting for a Dag, task group or task in another Airflow 3 deployment required copying most of ExternalTaskSensor and WorkflowTrigger, because their access to Airflow state was inlined into larger methods. Isolating that access in private methods lets the http provider reuse the existing sensor and trigger logic unchanged and only swap the data source for the remote deployment's REST API. Co-Authored-By: Claude <noreply@anthropic.com>
| if not AIRFLOW_V_3_0_PLUS: | ||
| raise AirflowOptionalProviderFeatureException("Waiting for a remote Airflow deployment needs Airflow 3+.") | ||
|
|
||
| from airflow.providers.standard.triggers.external_task import WorkflowTrigger |
There was a problem hiding this comment.
If someone upgrades the http provider without the standard extra, they keep whatever standard their Airflow install brought (core only pins >=1.9.0), and 1.20.0 has none of these seams: _poke_af3 calls ti.get_dr_count / get_ti_count / get_task_states directly and execute builds WorkflowTrigger inline. HttpExternalTaskSensor still imports fine, so it silently polls the local deployment for external_dag_id. That either waits until timeout or succeeds against a local Dag with the same id, which is likely when two deployments share a Dag repo. A triggerer image with new http and old standard gets the same local polling through the old _get_count_af_3.
Could this module fail fast next to the AF3 guard above, e.g. if not hasattr(WorkflowTrigger, "_get_dr_count"): raise AirflowOptionalProviderFeatureException(...) naming apache-airflow-providers-http[standard]? The sensor module imports this one first, so a single guard covers both the worker and the triggerer. A test that monkeypatches the attribute away and asserts the import raises would pin it.
|
|
||
| # The ``task_group_id`` filter of the task instances endpoint was added in Airflow 3.2.0. | ||
| # Older API servers silently ignore it and would return the task instances of all tasks. | ||
| MIN_TASK_GROUP_FILTER_VERSION = Version("3.2.0") |
There was a problem hiding this comment.
The task_group_id filter on the task instances endpoint was backported to 3.1.6 (#59511); it isn't in 3.1.5 but is in 3.1.6 and later. So a 3.1.6 to 3.1.x remote would answer this correctly and still gets the ValueError. Should the floor be Version("3.1.6"), with the comment above, the operators.rst line and the sensor docstring saying the same? test_get_task_group_states_requires_airflow_3_2 would need a 3.1.5 remote then.
| url = self._hook.url_from_endpoint(f"api/v2/{endpoint}") | ||
| timeout = self._hook.merged_extra.get("timeout") | ||
| response: Response = self._session.request(method, url, timeout=timeout, **kwargs) | ||
| if response.status_code == 401 and self._uses_access_token: |
There was a problem hiding this comment.
Only an expired token gets a second attempt. A connection reset, a read timeout, or a 429/503 from the remote raises straight out of run(), so one blip fails the trigger and the wait restarts on the next try, or fails the sensor outright with retries=0. The local path this stands in for goes through the Execution API client, which retries on its own (API_RETRIES in task-sdk api/client.py). For a sensor that waits across a network for hours, would passing HttpHook(adapter=HTTPAdapter(max_retries=Retry(total=..., backoff_factor=..., status_forcelist=(429, 502, 503, 504), allowed_methods=None))) in __init__ be enough? allowed_methods=None matters because the task count is a POST, which urllib3 doesn't retry by default, and Retry already honours Retry-After on 429/503.
| return sum( | ||
| self._request( | ||
| "POST", | ||
| "dags/~/dagRuns/~/taskInstances/list", |
There was a problem hiding this comment.
Two side effects of the batch endpoint that the Dag and task group paths don't have. First, POST .../taskInstances/list carries action_logging(), so every poke writes one audit Log row per logical date and state set on the remote. At the default 60s poke interval that's a few thousand rows per sensor per day in their Audit Log. Second, it authorizes on dag_id="~" and then filters rows with readable_ti_filter, so a user without read access to the awaited Dag gets total_entries: 0 rather than a 403 and the sensor just waits until timeout. GET dags/{dag_id}/dagRuns/~/taskInstances with task_id, state, logical_date_gte/lte and limit=1 avoids both, at one call per task id. Those filters exist from 3.0.0.
| :class:`~airflow.providers.standard.triggers.external_task.WorkflowTrigger`. | ||
| """ | ||
|
|
||
| def __init__(self, *, http_conn_id: str, **kwargs: Any) -> None: |
There was a problem hiding this comment.
run_ids still reaches WorkflowTrigger.__init__ through **kwargs, and the base run() compares the count against len(self.run_ids or self.logical_dates), but these overrides only count over logical_dates. So constructing this trigger with run_ids counts 0 against len(run_ids) every time and never fires. The sensor never passes it, but the trigger is public. Should __init__ raise if run_ids is set, matching the docstring's "only logical_dates are supported"?
| return _AirflowApiClient(self.http_conn_id) | ||
|
|
||
| async def _get_dr_count(self, states: Collection[str] | None) -> int: | ||
| return await sync_to_async(self._client.get_dr_count)( |
There was a problem hiding this comment.
sync_to_async defaults to thread_sensitive=True, and the triggerer runs under asyncio.run with no AsyncToSync parent, so these three calls land on asgiref's single shared executor thread. Every other trigger in the process that uses sync_to_async runs on that thread too: the local WorkflowTrigger, other providers' triggers, sync secrets-backend lookups. The requests have no timeout unless the connection extra sets one (lines 74 and 86). So a remote API server that accepts the connection and then stalls parks that thread and stalls all of them, and cancelling this trigger waits on the socket as well. HttpAsyncHook with a ClientTimeout would match HttpSensorTrigger in this provider. Short of that, thread_sensitive=False (or asyncio.to_thread) on all three calls plus a default timeout in _request and _create_session would stop one bad remote from blocking the rest.
| count_allowed = self._calculate_count(count, dttm_filter) | ||
| return count_allowed == len(dttm_filter) | ||
|
|
||
| def _get_dr_count( |
There was a problem hiding this comment.
These four methods are now the extension contract HttpExternalTaskSensor overrides from a separately released distribution, but the leading _ marks them private and free to rename, and a rename would bring back the silent local polling with nothing in standard's own tests noticing. Would it be worth making them a documented subclass hook, either dropping the _ prefix or saying in the docstrings that subclasses override them to change where state comes from? Same question for the WorkflowTrigger trio.
| assert trigger.external_task_ids == ["t1"] | ||
| assert trigger.failed_states == ["failed"] | ||
| assert trigger.logical_dates == [LOGICAL_DATE] | ||
| assert trigger.poke_interval == 30 |
There was a problem hiding this comment.
_get_trigger forwards ten kwargs and this checks six. Dropping allowed_states, skipped_states, soft_fail or external_task_group_id from _get_trigger still passes the whole http suite, and a dropped allowed_states is the bad one: the remote queries go out with no state filter and the sensor succeeds while the remote run is still queued. Could this pass non-default values for all of them and compare trigger.serialize() in one assertion?
| event = await anext(aiter(trigger.run())) | ||
|
|
||
| assert event == TriggerEvent({"status": "success"}) | ||
| m.assert_awaited_once() |
There was a problem hiding this comment.
A bare assert_awaited_once() lets the states argument go wrong unnoticed: changing _get_count_af_3 to call self._get_dr_count(None) passes every standard test. The sensor twin of this test checks its arguments; could this one take the expected args from the parametrize ((trigger, ["success"]) for dag and tasks, (trigger,) for the task group) and use assert_awaited_once_with?
This branch has not been deployed
2 participants
Adds
HttpExternalTaskSensorandHttpExternalTaskTriggerto the http provider, so a Dag, task group or task in another Airflow 3 deployment can be awaited via its REST API (v2) with the same parameters asExternalTaskSensor, plushttp_conn_id. To make this possible without copying the sensor,ExternalTaskSensorandWorkflowTriggernow isolate their Airflow state access in private overridable methods (_get_dr_count,_get_ti_count,_get_task_group_states,_get_trigger), andWorkflowTrigger.serializederives its classpath from the actual class; their behaviour is unchanged. The http sensor inherits the existing semantics as-is (includingcheck_existencebeing a no-op on Airflow 3), requires Airflow 3 locally and remotely (3.2+ remotely for task groups), and needs the newstandardextra of the http provider.Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (Opus 5.5) following the guidelines
{pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.🤖 Generated with Claude Code