This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v3-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 76e9e78dadd2f393bf32d292e434ec93f981be3d
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Jul 8 10:29:46 2025 -0400
[v3-0-test] Fix log for skipped task (#53024) (#53028)
* [v3-0-test] Fix log for skipped taks (#53024)
(cherry picked from commit 0d6e4172e6e4eccabd5c8b12606c10cfd23ee58b)
Co-authored-by: Pierre Jeambrun <[email protected]>
* Fix CI
---------
Co-authored-by: Pierre Jeambrun <[email protected]>
---
.../core_api/openapi/v2-rest-api-generated.yaml | 2 +-
.../api_fastapi/core_api/routes/public/log.py | 4 +--
.../ui/src/pages/TaskInstance/Logs/Logs.tsx | 2 +-
.../src/airflow/utils/log/file_task_handler.py | 9 ++++++
airflow-core/tests/unit/utils/test_log_handlers.py | 37 ++++++++++++++++++++++
5 files changed, 50 insertions(+), 4 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index 74784659110..330861ec1d2 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -6323,7 +6323,7 @@ paths:
required: true
schema:
type: integer
- exclusiveMinimum: 0
+ minimum: 0
title: Try Number
- name: full_content
in: query
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 de7b327b6a4..1a7242c9936 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
@@ -22,7 +22,7 @@ import textwrap
from fastapi import Depends, HTTPException, Request, Response, status
from itsdangerous import BadSignature, URLSafeSerializer
-from pydantic import PositiveInt
+from pydantic import NonNegativeInt
from sqlalchemy.orm import joinedload
from sqlalchemy.sql import select
@@ -75,7 +75,7 @@ def get_log(
dag_id: str,
dag_run_id: str,
task_id: str,
- try_number: PositiveInt,
+ try_number: NonNegativeInt,
accept: HeaderAcceptJsonOrNdjson,
request: Request,
dag_bag: DagBagDep,
diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Logs/Logs.tsx
b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Logs/Logs.tsx
index 3b9864a6db3..5b84591a605 100644
--- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Logs/Logs.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Logs/Logs.tsx
@@ -84,7 +84,7 @@ export const Logs = () => {
logLevelFilters,
sourceFilters,
taskInstance,
- tryNumber: tryNumber === 0 ? 1 : tryNumber,
+ tryNumber,
});
return (
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 db18e47f13f..89bdf385ee8 100644
--- a/airflow-core/src/airflow/utils/log/file_task_handler.py
+++ b/airflow-core/src/airflow/utils/log/file_task_handler.py
@@ -482,6 +482,15 @@ class FileTaskHandler(logging.Handler):
"""
if try_number is None:
try_number = task_instance.try_number
+
+ if task_instance.state == TaskInstanceState.SKIPPED:
+ logs = [
+ StructuredLogMessage( # type: ignore[call-arg]
+ event="Task was skipped, no logs available."
+ )
+ ]
+ return logs, {"end_of_log": True}
+
if try_number is None or try_number < 1:
logs = [
StructuredLogMessage( # type: ignore[call-arg]
diff --git a/airflow-core/tests/unit/utils/test_log_handlers.py
b/airflow-core/tests/unit/utils/test_log_handlers.py
index 5813c22819e..286eae5c842 100644
--- a/airflow-core/tests/unit/utils/test_log_handlers.py
+++ b/airflow-core/tests/unit/utils/test_log_handlers.py
@@ -151,6 +151,43 @@ class TestFileTaskLogHandler:
# Remove the generated tmp log file.
os.remove(log_filename)
+ def test_file_task_handler_when_ti_is_skipped(self, dag_maker):
+ def task_callable(ti):
+ ti.log.info("test")
+
+ with dag_maker("dag_for_testing_file_task_handler", schedule=None):
+ task = PythonOperator(
+ task_id="task_for_testing_file_log_handler",
+ python_callable=task_callable,
+ )
+ dagrun = dag_maker.create_dagrun()
+ ti = TaskInstance(task=task, run_id=dagrun.run_id)
+
+ ti.try_number = 0
+ ti.state = State.SKIPPED
+
+ logger = ti.log
+ ti.log.disabled = False
+
+ file_handler = next(
+ (handler for handler in logger.handlers if handler.name ==
FILE_TASK_HANDLER), None
+ )
+ assert file_handler is not None
+
+ set_context(logger, ti)
+ assert file_handler.handler is not None
+ # We expect set_context generates a file locally.
+ log_filename = file_handler.handler.baseFilename
+ assert os.path.isfile(log_filename)
+ assert log_filename.endswith("0.log"), log_filename
+
+ # Return value of read must be a tuple of list and list.
+ logs, metadata = file_handler.read(ti)
+ assert logs[0].event == "Task was skipped, no logs available."
+
+ # Remove the generated tmp log file.
+ os.remove(log_filename)
+
@pytest.mark.xfail(reason="TODO: Needs to be ported over to the new
structlog based logging")
def test_file_task_handler(self, dag_maker, session):
def task_callable(ti):
