kaxil commented on code in PR #74350:
URL: https://github.com/apache/airflow/pull/74350#discussion_r4222311349
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -1496,5 +1585,10 @@ def delete_task_instance(
)
TI.delete_attempts(
- dag_id=dag_id, run_id=dag_run_id, task_id=task_id,
map_index=map_index, session=session
+ dag_id=dag_id,
+ run_id=dag_run_id,
+ task_id=task_id,
+ map_index=map_index,
Review Comment:
This still passes the raw `map_index` query param (default -1), while the
lookup above matched on `scope.region_index`. So `DELETE
.../taskInstances/t?region_id=R®ion_index=2` finds the pass, then
`delete_attempts` deletes `region_index = -1` in R, which matches nothing, and
the endpoint reports success with the pass still there.
`map_index=scope.region_index` fixes it, and `TestDeleteTaskInstance` has no
case with region coordinates that would catch it. The route's `responses` also
still lists only 404 though the dependency raises 400 and 409 (same on
`/{task_id}/dependencies`).
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -398,7 +433,8 @@ def get_task_instance_tries(
TI.dag_id == dag_id,
TI.run_id == dag_run_id,
TI.task_id == task_id,
- TI.map_index == map_index,
+ TI.region_index == scope.region_index,
+ TI.region_id == scope.region_id,
Review Comment:
`/tries` still only lists attempts in `scope.region_id`. For a run from
before regions, a mapped task's old attempts stay in the sentinel region
(archived as superseded) once it re-expands into a region, so they and their
logs drop out of `/tries` unless the client passes
`region_id=00000000-0000-0000-0000-000000000000`, and the docstring doesn't
mention it. Could the sentinel-region attempts for the same index be included,
or at least documented on the route?
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -673,14 +740,18 @@ def get_task_instances(
)
load_legacy_rendered_fields(task_instances, session=session)
return TaskInstanceCollectionResponse(
- task_instances=task_instances,
+ task_instances=[
+ task_coordinate_response(TaskInstanceResponse, ti, resolver)
for ti in task_instances
+ ],
total_entries=total_entries,
total_entries_limit=total_entries_limit,
next_cursor=(
- encode_cursor(task_instances[-1], order_by) if has_next and
task_instances else None
+ encode_cursor(TaskCoordinateView(task_instances[-1],
resolver), order_by)
Review Comment:
The cursor still goes through `TaskCoordinateView`, so it uses
`resolver.public_map_index` (the pinned definition's `get_needs_expansion()`),
while the sort and the `map_index` filter use `public_map_index_expression`
(stored region `node_id == task_id`), and `get_execution` has a third inline
copy at `execution.py:95`. A mapped slot re-pinned to a definition without
`.expand()` sorts as its index in SQL but encodes -1 in the cursor, so the next
page can repeat rows. Selecting `public_map_index_expression` as a labelled
column and using it for both the response and the cursor would make them agree
by construction, and would also skip the pinned-Dag loads `public_map_index`
does per `dag_version` on each page.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/extra_links.py:
##########
@@ -65,19 +69,16 @@ def get_extra_links(
task_id: str,
session: SessionDep,
dag_bag: DagBagDep,
- map_index: int = -1,
+ scope: TaskScopeDep,
Review Comment:
`get_extra_links` still lists `try_number` ahead of `map_index` in the
generated spec, so after a client regen a positional `get_extra_links("d", "r",
"t", 3, 2)` silently swaps slot and try. `get_xcom_entry` (`map_index` now
after `stringify`), `get_log` (after `token`) and `clear_task_state_store`
(`all_map_indices` now last) reorder too. Moving the trailing params into a
dependency declared after `scope` would keep the released order.
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/task_coordinates.py:
##########
@@ -113,3 +110,33 @@ def resolve_task_scope(
map_index=tasks[0].region_index,
region_id=tasks[0].region_id,
)
+
+
+def _coordinate_resolver(dag_bag: DagBagDep, session: SessionDep) ->
TaskCoordinateResolver:
+ return TaskCoordinateResolver(dag_bag, session)
+
+
+CoordinateResolverDep = Annotated[TaskCoordinateResolver,
Depends(_coordinate_resolver)]
+
+
+def _task_scope(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ resolver: CoordinateResolverDep,
+ map_index: int = -1,
Review Comment:
`map_index` lost its `ge=-1` bound when it moved in here, so `minimum: -1`
is gone from seven operations in the spec (get/delete XCom entry and the five
task state store routes), and `map_index=-5` now returns 404 instead of 422.
`get_xcoms` and `_resolve_clear_scope` still declare `Query(ge=-1)`, so the
siblings disagree. `Query()` can't go on this param since it also binds the
`{map_index}` path segment, but an explicit `map_index < -1` check in
`resolve_task_scope` would keep the bound.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -141,52 +152,52 @@
task_instances_prefix = "/dagRuns/{dag_run_id}/taskInstances"
-def _coordinate_resolver(dag_bag: DagBagDep, session: SessionDep) ->
TaskCoordinateResolver:
- return TaskCoordinateResolver(dag_bag, session)
-
-
-CoordinateResolverDep = Annotated[TaskCoordinateResolver,
Depends(_coordinate_resolver)]
-
-
@task_instances_router.get(
task_instances_prefix + "/{task_id}",
- responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+ responses=create_openapi_http_exception_doc(
+ [status.HTTP_400_BAD_REQUEST, status.HTTP_404_NOT_FOUND,
status.HTTP_409_CONFLICT]
+ ),
dependencies=[Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.TASK_INSTANCE))],
)
def get_task_instance(
dag_id: str,
dag_run_id: str,
task_id: str,
session: SessionDep,
+ scope: TaskScopeDep,
Review Comment:
With `TaskScopeDep` here, the unmapped `GET .../taskInstances/{task_id}` now
publishes a `map_index` query param (it's in the spec and in
`GetTaskInstanceData.mapIndex`). For a mapped task, `?map_index=3` resolves
slot 3 and then hits the "Task instance is mapped, add the map_index value to
the URL" 404 below, which is confusing when the client did send one. Should
this route take only `region_id`/`region_index` and leave `map_index` to
`/{task_id}/{map_index}`?
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/execution.py:
##########
@@ -0,0 +1,116 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import Depends, HTTPException, Query, status
+from sqlalchemy import select
+
+from airflow.api_fastapi.common.db.common import SessionDep
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.execution import (
+ ExecutionCollectionResponse,
+ ExecutionRegionResponse,
+ ExecutionTaskResponse,
+)
+from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.security import DagAccessEntity,
requires_access_dag
+from airflow.api_fastapi.core_api.services.public.execution import
get_execution_members
+from airflow.models.dagrun import DagRun
+from airflow.models.dynamic_region import SENTINEL_REGION_ID,
load_region_ancestry
+
+execution_router = AirflowRouter(
+ tags=["DagRun"],
+ prefix="/dags/{dag_id}/dagRuns/{dag_run_id}",
+)
+
+
+@execution_router.get(
+ "/execution",
Review Comment:
This is still on the public v2 router, so `ExecutionRegionResponse` commits
`parent_region_index`, `forked_from_region_id` and `resumes_from_index` to the
stable API, and the Run execution page in the UI stays its only consumer
through the rest of the stack. The new `region_id`/`region_index` params also
have no description, so nothing in the API docs says what a region is or where
a client gets one. Could this live on the private UI router until the region
model settles?
##########
airflow-core/src/airflow/api_fastapi/common/parameters/misc.py:
##########
@@ -245,7 +246,7 @@ def _optional_boolean(value: bool | None) -> bool | None:
FilterParam[int | None],
Depends(
filter_param_factory(
- TaskInstance.region_index,
+ public_map_index_expression(TaskInstance),
Review Comment:
With the HITL `map_index` filter on `public_map_index_expression`, every
loop pass reads as -1, and `get_hitl_details` has no `region_id`/`region_index`
filters, so a client can't list the HITL details for one pass of a loop task
anymore (`?map_index=2` used to select pass 2 through `region_index`). Could
the list route take the same `region_id`/`region_index` filters the TI list
routes got?
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_log.py:
##########
@@ -49,6 +55,81 @@ class TestTaskInstancesLog:
default_time = "2020-06-10T20:00:00+00:00"
+ def _place_tries_in_sibling_regions(self, session):
+ history = session.scalar(
+ select(TaskInstance)
+ .where(
+ TaskInstance.dag_id == self.DAG_ID,
+ TaskInstance.run_id == self.RUN_ID,
+ TaskInstance.task_id == self.TASK_ID,
+ TaskInstance.try_number == 1,
+ )
+ .execution_options(include_all_attempts=True)
+ )
+ current = session.scalar(
+ select(TaskInstance).where(
+ TaskInstance.dag_id == self.DAG_ID,
+ TaskInstance.run_id == self.RUN_ID,
+ TaskInstance.task_id == self.TASK_ID,
+ TaskInstance.working_set.is_(True),
+ )
+ )
+ previous_region, current_region = uuid4(), uuid4()
+ for region_id, source in ((previous_region, None), (current_region,
previous_region)):
+ session.add(
+ DynamicRegion(
+ id=region_id,
+ dag_id=self.DAG_ID,
+ run_id=self.RUN_ID,
+ node_id="loop",
+ forked_from_region_id=source,
+ )
+ )
+ history.region_id, history.region_index = previous_region, 2
+ current.region_id, current.region_index, current.try_number =
current_region, 2, 2
+ session.commit()
+ return history, current, previous_region, current_region
+
+ def test_external_link_uses_the_selected_archived_try(self, session):
+ history, current, previous_region, _ =
self._place_tries_in_sibling_regions(session)
+ with mock.patch(
+ "airflow.api_fastapi.core_api.routes.public.log.TaskLogReader",
autospec=True
+ ) as reader:
+ reader.return_value.supports_external_link = True
+ reader.return_value.log_handler =
mock.create_autospec(ExternalLoggingMixin, instance=True)
+ reader.return_value.log_handler.get_external_log_url.return_value
= "https://logs.example/old";
+ response = self.client.get(
+
f"/dags/{self.DAG_ID}/dagRuns/{self.RUN_ID}/taskInstances/{self.TASK_ID}/externalLogUrl/1",
+ params={"region_id": str(previous_region), "region_index": 2},
Review Comment:
Nothing here pins the `try_number` predicate on `externalLogUrl`: try 1 is
the only row in `previous_region`, so this passes with `TaskInstance.try_number
== try_number` deleted. Putting both tries in the same region and requesting
`/1` and `/2` would bind it.
--
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]
