This is an automated email from the ASF dual-hosted git repository.
jasonliu pushed a commit to branch v3-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-0-test by this push:
new 79be815a5ce Revert "fix: resolve 404 log error for non-latest task
tries in multi-host wo…" (#51145)
79be815a5ce is described below
commit 79be815a5cec6252116067bb5388ac5a86e7d4ab
Author: Pierre Jeambrun <[email protected]>
AuthorDate: Wed May 28 12:21:59 2025 +0200
Revert "fix: resolve 404 log error for non-latest task tries in multi-host
wo…" (#51145)
---
.../airflow/api_fastapi/common/db/task_instance.py | 57 ----------------
.../api_fastapi/core_api/routes/public/log.py | 32 ++++++---
.../src/airflow/utils/log/file_task_handler.py | 4 +-
.../tests/unit/api_fastapi/common/db/__init__.py | 16 -----
.../api_fastapi/common/db/test_task_instance.py | 79 ----------------------
5 files changed, 25 insertions(+), 163 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/db/task_instance.py
b/airflow-core/src/airflow/api_fastapi/common/db/task_instance.py
deleted file mode 100644
index b6748bdd61a..00000000000
--- a/airflow-core/src/airflow/api_fastapi/common/db/task_instance.py
+++ /dev/null
@@ -1,57 +0,0 @@
-# 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 pydantic import PositiveInt
-from sqlalchemy.orm import joinedload
-from sqlalchemy.sql import select
-
-from airflow.api_fastapi.common.db.common import SessionDep
-from airflow.models import TaskInstance, Trigger
-from airflow.models.taskinstancehistory import TaskInstanceHistory
-
-
-def get_task_instance_or_history_for_try_number(
- dag_id: str,
- dag_run_id: str,
- task_id: str,
- try_number: PositiveInt,
- session: SessionDep,
- map_index: int,
-) -> TaskInstance | TaskInstanceHistory:
- query = (
- select(TaskInstance)
- .where(
- TaskInstance.task_id == task_id,
- TaskInstance.dag_id == dag_id,
- TaskInstance.run_id == dag_run_id,
- TaskInstance.map_index == map_index,
- )
- .join(TaskInstance.dag_run)
-
.options(joinedload(TaskInstance.trigger).joinedload(Trigger.triggerer_job))
- )
- ti = session.scalar(query)
- if ti is None or ti.try_number != try_number:
- query = select(TaskInstanceHistory).where(
- TaskInstanceHistory.task_id == task_id,
- TaskInstanceHistory.dag_id == dag_id,
- TaskInstanceHistory.run_id == dag_run_id,
- TaskInstanceHistory.map_index == map_index,
- TaskInstanceHistory.try_number == try_number,
- )
- ti = session.scalar(query)
- return ti
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/log.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/log.py
index da459dd95cf..3873cadf69d 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/log.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/log.py
@@ -23,10 +23,11 @@ import textwrap
from fastapi import Depends, HTTPException, Request, Response, status
from itsdangerous import BadSignature, URLSafeSerializer
from pydantic import PositiveInt
+from sqlalchemy.orm import joinedload
+from sqlalchemy.sql import select
from airflow.api_fastapi.common.dagbag import DagBagDep
from airflow.api_fastapi.common.db.common import SessionDep
-from airflow.api_fastapi.common.db.task_instance import
get_task_instance_or_history_for_try_number
from airflow.api_fastapi.common.headers import HeaderAcceptJsonOrText
from airflow.api_fastapi.common.router import AirflowRouter
from airflow.api_fastapi.common.types import Mimetype
@@ -34,6 +35,8 @@ from airflow.api_fastapi.core_api.datamodels.log import
TaskInstancesLogResponse
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.exceptions import TaskNotFound
+from airflow.models import TaskInstance, Trigger
+from airflow.models.taskinstancehistory import TaskInstanceHistory
from airflow.utils.log.log_reader import TaskLogReader
task_instances_log_router = AirflowRouter(
@@ -101,14 +104,27 @@ def get_log(
if not task_log_reader.supports_read:
raise HTTPException(status.HTTP_400_BAD_REQUEST, "Task log handler
does not support read logs.")
- ti = get_task_instance_or_history_for_try_number(
- dag_id=dag_id,
- dag_run_id=dag_run_id,
- task_id=task_id,
- try_number=try_number,
- session=session,
- map_index=map_index,
+ query = (
+ select(TaskInstance)
+ .where(
+ TaskInstance.task_id == task_id,
+ TaskInstance.dag_id == dag_id,
+ TaskInstance.run_id == dag_run_id,
+ TaskInstance.map_index == map_index,
+ )
+ .join(TaskInstance.dag_run)
+
.options(joinedload(TaskInstance.trigger).joinedload(Trigger.triggerer_job))
)
+ ti = session.scalar(query)
+ if ti is None:
+ query = select(TaskInstanceHistory).where(
+ TaskInstanceHistory.task_id == task_id,
+ TaskInstanceHistory.dag_id == dag_id,
+ TaskInstanceHistory.run_id == dag_run_id,
+ TaskInstanceHistory.map_index == map_index,
+ TaskInstanceHistory.try_number == try_number,
+ )
+ ti = session.scalar(query)
if ti is None:
metadata["end_of_log"] = True
diff --git a/airflow-core/src/airflow/utils/log/file_task_handler.py
b/airflow-core/src/airflow/utils/log/file_task_handler.py
index 1585d7ede20..2e8f71d0399 100644
--- a/airflow-core/src/airflow/utils/log/file_task_handler.py
+++ b/airflow-core/src/airflow/utils/log/file_task_handler.py
@@ -586,9 +586,7 @@ class FileTaskHandler(logging.Handler):
sources = []
logs = []
try:
- log_type = (
- LogType.TRIGGER if hasattr(ti, "triggerer_job") and
ti.triggerer_job else LogType.WORKER
- )
+ log_type = LogType.TRIGGER if ti.triggerer_job else LogType.WORKER
url, rel_path = self._get_log_retrieval_url(ti,
worker_log_rel_path, log_type=log_type)
response = _fetch_logs_from_service(url, rel_path)
if response.status_code == 403:
diff --git a/airflow-core/tests/unit/api_fastapi/common/db/__init__.py
b/airflow-core/tests/unit/api_fastapi/common/db/__init__.py
deleted file mode 100644
index 13a83393a91..00000000000
--- a/airflow-core/tests/unit/api_fastapi/common/db/__init__.py
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
diff --git
a/airflow-core/tests/unit/api_fastapi/common/db/test_task_instance.py
b/airflow-core/tests/unit/api_fastapi/common/db/test_task_instance.py
deleted file mode 100644
index e5ed829dc8f..00000000000
--- a/airflow-core/tests/unit/api_fastapi/common/db/test_task_instance.py
+++ /dev/null
@@ -1,79 +0,0 @@
-# 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
-
-import pytest
-
-from airflow.api_fastapi.common.db.task_instance import
get_task_instance_or_history_for_try_number
-from airflow.models.taskinstance import TaskInstance
-from airflow.models.taskinstancehistory import TaskInstanceHistory
-from airflow.providers.standard.operators.empty import EmptyOperator
-from airflow.utils import timezone
-from airflow.utils.types import DagRunType
-
-from tests_common.test_utils.db import clear_db_runs
-
-pytestmark = pytest.mark.db_test
-
-
-class TestDBTaskInstance:
- DAG_ID = "dag_for_testing_db_task_instance"
- RUN_ID = "dag_run_id_for_testing_db_task_instance"
- TASK_ID = "task_for_testing_db_task_instance"
- TRY_NUMBER = 1
-
- default_time = "2020-06-10T20:00:00+00:00"
-
- @pytest.fixture(autouse=True)
- def setup_attrs(self, dag_maker, session) -> None:
- with dag_maker(self.DAG_ID,
start_date=timezone.parse(self.default_time), session=session) as dag:
- EmptyOperator(task_id=self.TASK_ID)
-
- dr = dag_maker.create_dagrun(
- run_id=self.RUN_ID,
- run_type=DagRunType.SCHEDULED,
- logical_date=timezone.parse(self.default_time),
- start_date=timezone.parse(self.default_time),
- )
-
- for ti in dr.task_instances:
- ti.try_number = 1
- ti.hostname = "localhost"
- session.merge(ti)
- dag.clear()
- for ti in dr.task_instances:
- ti.try_number = 2
- ti.hostname = "localhost"
- session.merge(ti)
- session.commit()
-
- def teardown_method(self):
- clear_db_runs()
-
- @pytest.mark.parametrize("try_number", [1, 2])
- def test_get_task_instance_or_history_for_try_number(self, try_number,
session):
- ti = get_task_instance_or_history_for_try_number(
- self.DAG_ID,
- self.RUN_ID,
- self.TASK_ID,
- try_number,
- session=session,
- map_index=-1,
- )
-
- assert isinstance(ti, TaskInstanceHistory) if try_number == 1 else
TaskInstance
