diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py index 6f9402f23603d..20e549886704e 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py @@ -24,6 +24,7 @@ from pydantic import AliasPath, Field from airflow.api_fastapi.core_api.base import BaseModel +from airflow.utils.state import CallbackState class DeadlineResponse(BaseModel): @@ -37,6 +38,10 @@ class DeadlineResponse(BaseModel): dag_run_id: str = Field(validation_alias=AliasPath("dagrun", "run_id")) alert_id: UUID | None = Field(validation_alias="deadline_alert_id", default=None) alert_name: str | None = Field(validation_alias=AliasPath("deadline_alert", "name"), default=None) + callback_id: UUID | None = Field(validation_alias="callback_id", default=None) + callback_state: CallbackState | None = Field( + validation_alias=AliasPath("callback", "state"), default=None + ) class DeadlineCollectionResponse(BaseModel): diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml index ca8f729998250..461471202a103 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml +++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml @@ -989,6 +989,85 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /ui/dags/{dag_id}/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs: + get: + tags: + - Deadlines + summary: Get Callback Logs + description: 'Get execution logs for a callback associated with a deadline. + + + Returns the logs produced during callback execution. These logs are uploaded + + to remote storage (or written locally) by the callback supervisor after execution.' + operationId: get_callback_logs + security: + - OAuth2PasswordBearer: [] + - HTTPBearer: [] + parameters: + - name: callback_id + in: path + required: true + schema: + type: string + format: uuid + title: Callback Id + - name: dag_id + in: path + required: true + schema: + type: string + title: Dag Id + - name: dag_run_id + in: path + required: true + schema: + type: string + title: Dag Run Id + - name: accept + in: header + required: false + schema: + type: string + enum: + - application/json + - application/x-ndjson + - '*/*' + default: '*/*' + title: Accept + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/TaskInstancesLogResponse' + application/x-ndjson: + schema: + type: string + example: '{"content": "content"} + + {"content": "content"} + + ' + '400': + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPExceptionResponse' + description: Bad Request + '404': + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPExceptionResponse' + description: Not Found + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /ui/structure/structure_data: get: tags: @@ -2236,6 +2315,17 @@ components: - count title: CalendarTimeRangeResponse description: Represents a summary of DAG runs for a specific calendar time range. + CallbackState: + type: string + enum: + - scheduled + - pending + - queued + - running + - success + - failed + title: CallbackState + description: All possible states of callbacks. ConfigResponse: properties: fallback_page_limit: @@ -2924,6 +3014,16 @@ components: - type: string - type: 'null' title: Alert Name + callback_id: + anyOf: + - type: string + format: uuid + - type: 'null' + title: Callback Id + callback_state: + anyOf: + - $ref: '#/components/schemas/CallbackState' + - type: 'null' type: object required: - id @@ -3942,6 +4042,21 @@ components: - nodes title: StructureDataResponse description: Structure Data serializer for responses. + StructuredLogMessage: + properties: + timestamp: + type: string + format: date-time + title: Timestamp + event: + type: string + title: Event + additionalProperties: true + type: object + required: + - event + title: StructuredLogMessage + description: An individual log message. TaskInstanceResponse: properties: id: @@ -4215,6 +4330,28 @@ components: - awaiting_input title: TaskInstanceStateCount description: TaskInstance serializer for responses. + TaskInstancesLogResponse: + properties: + content: + anyOf: + - items: + $ref: '#/components/schemas/StructuredLogMessage' + type: array + - items: + type: string + type: array + title: Content + continuation_token: + anyOf: + - type: string + - type: 'null' + title: Continuation Token + type: object + required: + - content + - continuation_token + title: TaskInstancesLogResponse + description: Log serializer for responses. TeamCollectionResponse: properties: teams: diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py index 06eda42ed8957..668e3998343c5 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py @@ -18,13 +18,16 @@ from __future__ import annotations from typing import Annotated +from uuid import UUID from fastapi import Depends, HTTPException, status +from fastapi.responses import StreamingResponse from sqlalchemy import select -from sqlalchemy.orm import contains_eager, noload +from sqlalchemy.orm import contains_eager, joinedload, noload from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity from airflow.api_fastapi.common.db.common import SessionDep, paginated_select +from airflow.api_fastapi.common.headers import HeaderAcceptJsonOrNdjson from airflow.api_fastapi.common.parameters import ( FilterParam, QueryLimit, @@ -35,16 +38,23 @@ filter_param_factory, ) from airflow.api_fastapi.common.router import AirflowRouter +from airflow.api_fastapi.common.types import Mimetype +from airflow.api_fastapi.core_api.datamodels.log import TaskInstancesLogResponse from airflow.api_fastapi.core_api.datamodels.ui.deadline import ( DeadlineAlertCollectionResponse, DeadlineCollectionResponse, ) from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc +from airflow.api_fastapi.core_api.routes.public.log import ( + _buffered_ndjson_stream, + ndjson_example_response_for_get_log, +) from airflow.api_fastapi.core_api.security import ReadableDagRunsFilterDep, requires_access_dag from airflow.models.dagrun import DagRun from airflow.models.deadline import Deadline from airflow.models.deadline_alert import DeadlineAlert from airflow.models.serialized_dag import SerializedDagModel +from airflow.utils.log.callback_log_reader import read_callback_log, validate_log_path_component deadlines_router = AirflowRouter(prefix="/dags/{dag_id}", tags=["Deadlines"]) @@ -106,7 +116,7 @@ def get_deadlines( .options( contains_eager(Deadline.dagrun).options(noload(DagRun.deadlines)), contains_eager(Deadline.deadline_alert), - noload(Deadline.callback), + joinedload(Deadline.callback), ) ) @@ -201,3 +211,75 @@ def get_dag_deadline_alerts( alerts = session.scalars(alerts_select) return DeadlineAlertCollectionResponse(deadline_alerts=alerts, total_entries=total_entries) + + +def _validated_log_path_params(dag_id: str, dag_run_id: str) -> tuple[str, str]: + """Reject dag_id/dag_run_id values that are unsafe as log path components (path traversal).""" + for param_name, param_value in (("dag_id", dag_id), ("dag_run_id", dag_run_id)): + try: + validate_log_path_component(param_value) + except ValueError: + raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid characters in {param_name}") + return dag_id, dag_run_id + + +@deadlines_router.get( + "/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs", + responses={ + **create_openapi_http_exception_doc([status.HTTP_400_BAD_REQUEST, status.HTTP_404_NOT_FOUND]), + status.HTTP_200_OK: { + "description": "Successful Response", + "content": ndjson_example_response_for_get_log, + }, + }, + dependencies=[ + Depends( + requires_access_dag( + method="GET", + access_entity=DagAccessEntity.TASK_LOGS, + ) + ), + ], + response_model=TaskInstancesLogResponse, + response_model_exclude_unset=True, +) +def get_callback_logs( + path_params: Annotated[tuple[str, str], Depends(_validated_log_path_params)], + callback_id: UUID, + accept: HeaderAcceptJsonOrNdjson, + session: SessionDep, +): + """ + Get execution logs for a callback associated with a deadline. + + Returns the logs produced during callback execution. These logs are uploaded + to remote storage (or written locally) by the callback supervisor after execution. + """ + dag_id, dag_run_id = path_params + + # A single exists-only check that the callback belongs to this dag run via its Deadline. + deadline_exists = session.scalar( + select(Deadline.id) + .join(Deadline.dagrun) + .where( + Deadline.callback_id == callback_id, + DagRun.dag_id == dag_id, + DagRun.run_id == dag_run_id, + ) + .limit(1) + ) + if deadline_exists is None: + raise HTTPException( + status.HTTP_404_NOT_FOUND, + f"Callback `{callback_id}` with a deadline for DagRun `{dag_run_id}` of Dag `{dag_id}` was not found", + ) + + log_stream = read_callback_log(dag_id=dag_id, run_id=dag_run_id, callback_id=str(callback_id)) + + if accept == Mimetype.NDJSON: + return StreamingResponse( + media_type="application/x-ndjson", + content=_buffered_ndjson_stream(f"{log.model_dump_json()}\n" for log in log_stream), + ) + + return TaskInstancesLogResponse.model_construct(content=list(log_stream), continuation_token=None) diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py b/airflow-core/src/airflow/jobs/triggerer_job_runner.py index f851e491a69dd..51cab4c63859b 100644 --- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py +++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py @@ -421,7 +421,8 @@ def from_api_response(cls, response: HITLDetailResponse) -> HITLDetailResponseRe class TriggerLoggingFactory: log_path: str - ti: RuntimeTI = attrs.field(repr=False) + # Callback triggers have no task instance; ``upload_to_remote`` accepts ``ti=None``. + ti: RuntimeTI | None = attrs.field(default=None, repr=False) bound_logger: WrappedLogger = attrs.field(init=False, repr=False) @@ -846,6 +847,19 @@ def _create_workload( if trigger.assets: watched_assets = {a.name: a.uri for a in trigger.assets} + if callback := getattr(trigger, "callback", None): + # Callback triggers get dedicated logging so their output is captured to a + # file the UI callback log endpoint can read. dag_id is stored on the callback + # data; run_id comes from the deadline context injected at miss time. + callback_data = callback.data or {} + context = (callback_data.get("kwargs") or {}).get("context") or {} + dag_run_data = context.get("dag_run") or {} + dag_id = callback_data.get("dag_id") or dag_run_data.get("dag_id") or "unknown" + run_id = dag_run_data.get("dag_run_id") or "unknown" + self.logger_cache[trigger.id] = TriggerLoggingFactory( + log_path=f"triggerer_callbacks/{dag_id}/{run_id}/{callback.id}", + ) + return workloads.RunTrigger( id=trigger.id, classpath=trigger.classpath, diff --git a/airflow-core/src/airflow/utils/log/callback_log_reader.py b/airflow-core/src/airflow/utils/log/callback_log_reader.py new file mode 100644 index 0000000000000..6bd765e8efc50 --- /dev/null +++ b/airflow-core/src/airflow/utils/log/callback_log_reader.py @@ -0,0 +1,156 @@ +# 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. +"""Reader for callback execution logs stored in remote or local storage.""" + +from __future__ import annotations + +import os +import re +from collections.abc import Generator +from contextlib import suppress +from pathlib import Path +from typing import TYPE_CHECKING + +from airflow.configuration import conf +from airflow.utils.log.file_task_handler import ( + StructuredLogMessage, + _get_compatible_log_stream, + _interleave_logs, + _stream_lines_by_chunk, +) + +if TYPE_CHECKING: + from airflow._shared.logging.remote import LogSourceInfo, RawLogStream, StreamingLogResponse + +_SAFE_PATH_COMPONENT = re.compile(r"[A-Za-z0-9._:+\-~@]+") + + +def validate_log_path_component(component: str) -> str: + """Validate a single log path component, raising ValueError if it could escape the log folder.""" + if component in (".", "..") or not _SAFE_PATH_COMPONENT.fullmatch(component): + raise ValueError(f"Invalid log path component: {component!r}") + return component + + +def _get_callback_log_relative_paths(dag_id: str, run_id: str, callback_id: str) -> list[str]: + """ + Construct the relative log paths for a callback execution. + + The executor path matches the format used in ExecuteCallback.make(): + executor_callbacks/{dag_id}/{run_id}/{callback_id} + The triggerer path matches what TriggerLoggingFactory writes for callback triggers: + triggerer_callbacks/{dag_id}/{run_id}/{callback_id} + """ + for component in (dag_id, run_id, callback_id): + validate_log_path_component(component) + return [ + f"executor_callbacks/{dag_id}/{run_id}/{callback_id}", + f"triggerer_callbacks/{dag_id}/{run_id}/{callback_id}", + ] + + +def read_callback_log( + dag_id: str, + run_id: str, + callback_id: str, +) -> Generator[StructuredLogMessage, None, None]: + """ + Stream callback logs from remote and/or local storage. + + Tries both executor_callbacks and triggerer_callbacks paths. For each path, tries + remote storage first (if configured), then falls back to the local filesystem. + + :param dag_id: The Dag ID associated with the callback. + :param run_id: The Dag run ID associated with the callback. + :param callback_id: The unique callback identifier. + :return: Generator of StructuredLogMessage objects. + """ + relative_paths = _get_callback_log_relative_paths(dag_id, run_id, callback_id) + + sources: LogSourceInfo = [] + log_streams: list[RawLogStream] = [] + + for relative_path in relative_paths: + with suppress(Exception): + remote_sources, remote_log_streams = _read_callback_remote_logs(relative_path) + sources.extend(remote_sources) + log_streams.extend(remote_log_streams) + + if not log_streams: + local_sources, local_log_streams = _read_callback_local_logs(relative_path) + sources.extend(local_sources) + log_streams.extend(local_log_streams) + + # If we found logs at this path, no need to check the next path + if log_streams: + break + + if not log_streams: + yield StructuredLogMessage(event="No callback logs found.") + return + + yield StructuredLogMessage(event="::group::Log message source details", sources=sources) # type: ignore[call-arg] + yield StructuredLogMessage(event="::endgroup::") + yield from _interleave_logs(*log_streams) + + +def _read_callback_remote_logs(relative_path: str) -> StreamingLogResponse: + """Read callback logs from the configured remote log storage.""" + from airflow.logging_config import get_remote_task_log + + remote_io = get_remote_task_log() + if remote_io is None: + return [], [] + + # Callbacks have no TaskInstance, so pass ti=None; remote handlers only use it + # for optional metadata (e.g. CloudWatch end_date) and read by relative path. + if stream_method := getattr(remote_io, "stream", None): + sources, logs = stream_method(relative_path, None) + return sources, logs or [] + + sources, logs = remote_io.read(relative_path, None) # type: ignore[arg-type] + if not logs: + return sources, [] + + return sources, [_get_compatible_log_stream(logs)] + + +def _read_callback_local_logs(relative_path: str) -> StreamingLogResponse: + """Read callback logs from the local filesystem.""" + base_log_folder = os.path.realpath(conf.get("logging", "base_log_folder")) + log_path = Path(base_log_folder, *(validate_log_path_component(p) for p in relative_path.split("/"))) + + sources: list[str] = [] + log_streams: list[RawLogStream] = [] + + for path in sorted(log_path.parent.glob(log_path.name + "*")): + # Containment check (defense in depth, e.g. against symlinks escaping the log folder). + resolved_path = os.path.realpath(path) + try: + if os.path.commonpath([base_log_folder, resolved_path]) != base_log_folder: + continue + except ValueError: + continue + + try: + log_stream = _stream_lines_by_chunk(open(resolved_path, encoding="utf-8")) + except OSError: + continue + sources.append(os.fspath(path)) + log_streams.append(log_stream) + + return sources, log_streams diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py index acadab3a6b16f..ceb713df5aa1f 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py @@ -17,10 +17,14 @@ from __future__ import annotations +import json +import uuid + import pytest from sqlalchemy import select from airflow._shared.timezones import timezone +from airflow.models.dagrun import DagRun from airflow.models.deadline import Deadline from airflow.models.deadline_alert import DeadlineAlert from airflow.models.serialized_dag import SerializedDagModel @@ -31,6 +35,7 @@ from airflow.utils.types import DagRunTriggeredByType, DagRunType from tests_common.test_utils.asserts import assert_queries_count +from tests_common.test_utils.config import conf_vars from tests_common.test_utils.db import ( clear_db_dags, clear_db_deadline, @@ -514,3 +519,74 @@ def test_should_response_401(self, unauthenticated_test_client): def test_should_response_403(self, unauthorized_test_client): response = unauthorized_test_client.get(f"/dags/{DAG_ID}/deadlineAlerts") assert response.status_code == 403 + + +class TestGetCallbackLogs: + """Tests for GET /dags/{dag_id}/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs.""" + + @pytest.fixture + def missed_callback_id(self, session): + deadline = session.scalar(select(Deadline).join(Deadline.dagrun).where(DagRun.run_id == RUN_MISSED)) + return str(deadline.callback_id) + + @pytest.fixture + def log_folder(self, tmp_path): + with conf_vars({("logging", "base_log_folder"): str(tmp_path)}): + yield tmp_path + + @staticmethod + def _write_local_log(log_folder, callback_id, content="callback ran\n", prefix="executor_callbacks"): + log_dir = log_folder / prefix / DAG_ID / RUN_MISSED + log_dir.mkdir(parents=True) + (log_dir / callback_id).write_text(content) + + def test_returns_logs_from_local_storage(self, test_client, missed_callback_id, log_folder): + self._write_local_log(log_folder, missed_callback_id) + response = test_client.get(f"/dags/{DAG_ID}/dagRuns/{RUN_MISSED}/callbacks/{missed_callback_id}/logs") + assert response.status_code == 200 + events = [entry["event"] for entry in response.json()["content"]] + assert "callback ran" in events + + def test_ndjson_streaming_response(self, test_client, missed_callback_id, log_folder): + self._write_local_log(log_folder, missed_callback_id) + response = test_client.get( + f"/dags/{DAG_ID}/dagRuns/{RUN_MISSED}/callbacks/{missed_callback_id}/logs", + headers={"Accept": "application/x-ndjson"}, + ) + assert response.status_code == 200 + assert response.headers["content-type"].startswith("application/x-ndjson") + lines = [json.loads(line) for line in response.text.splitlines() if line] + assert any(line["event"] == "callback ran" for line in lines) + + def test_no_logs_found_message(self, test_client, missed_callback_id, log_folder): + response = test_client.get(f"/dags/{DAG_ID}/dagRuns/{RUN_MISSED}/callbacks/{missed_callback_id}/logs") + assert response.status_code == 200 + assert response.json()["content"][0]["event"] == "No callback logs found." + + def test_unknown_callback_returns_404(self, test_client): + response = test_client.get(f"/dags/{DAG_ID}/dagRuns/{RUN_MISSED}/callbacks/{uuid.uuid4()}/logs") + assert response.status_code == 404 + + def test_callback_of_other_run_returns_404(self, test_client, missed_callback_id): + """A callback that exists but belongs to a different dag run is rejected.""" + response = test_client.get(f"/dags/{DAG_ID}/dagRuns/{RUN_SINGLE}/callbacks/{missed_callback_id}/logs") + assert response.status_code == 404 + + # Note: a literal ".." segment is normalized away by HTTP clients before reaching the + # server, so only encoded/otherwise-unsafe variants exercise the endpoint validation. + @pytest.mark.parametrize("bad_run_id", ["%2e%2e", "..%5c..%5cetc", "run%20id"]) + def test_path_traversal_in_dag_run_id_returns_400(self, test_client, missed_callback_id, bad_run_id): + response = test_client.get(f"/dags/{DAG_ID}/dagRuns/{bad_run_id}/callbacks/{missed_callback_id}/logs") + assert response.status_code == 400 + + def test_should_response_401(self, unauthenticated_test_client): + response = unauthenticated_test_client.get( + f"/dags/{DAG_ID}/dagRuns/{RUN_MISSED}/callbacks/{uuid.uuid4()}/logs" + ) + assert response.status_code == 401 + + def test_should_response_403(self, unauthorized_test_client): + response = unauthorized_test_client.get( + f"/dags/{DAG_ID}/dagRuns/{RUN_MISSED}/callbacks/{uuid.uuid4()}/logs" + ) + assert response.status_code == 403 diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py b/airflow-core/tests/unit/jobs/test_triggerer_job.py index 31a77b50b6592..496cf71ce0743 100644 --- a/airflow-core/tests/unit/jobs/test_triggerer_job.py +++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py @@ -609,6 +609,58 @@ def test_create_workload_uses_supervisor_id_without_job(jobless_supervisor, mock assert factory.log_path == f"/logs/ti.trigger.{jobless_supervisor.id}.log" +def test_create_workload_sets_up_logging_for_callback_trigger(jobless_supervisor, mocker): + """_create_workload() populates logger_cache for callback triggers so their logs are captured.""" + callback_id = uuid.uuid4() + callback = mocker.Mock() + callback.id = callback_id + callback.data = { + "dag_id": "test_dag", + "kwargs": {"context": {"dag_run": {"dag_id": "test_dag", "dag_run_id": "manual__2024-01-01"}}}, + } + + trigger = mocker.Mock() + trigger.id = 8 + trigger.classpath = "airflow.triggers.callback.CallbackTrigger" + trigger.encrypted_kwargs = "" + trigger.task_instance = None + trigger.assets = None + trigger.callback = callback + + workload = jobless_supervisor._create_workload( + trigger=trigger, + dag_bag=mocker.Mock(), + render_log_fname=mocker.Mock(), + session=mocker.Mock(), + ) + + assert workload is not None + factory = jobless_supervisor.logger_cache[trigger.id] + assert factory.log_path == f"triggerer_callbacks/test_dag/manual__2024-01-01/{callback_id}" + assert factory.ti is None + + +def test_create_workload_no_logging_for_non_callback_trigger(jobless_supervisor, mocker): + """_create_workload() leaves logger_cache alone for non-callback triggers without a TI.""" + trigger = mocker.Mock() + trigger.id = 9 + trigger.classpath = "airflow.triggers.temporal.DateTimeTrigger" + trigger.encrypted_kwargs = "" + trigger.task_instance = None + trigger.assets = None + trigger.callback = None + + workload = jobless_supervisor._create_workload( + trigger=trigger, + dag_bag=mocker.Mock(), + render_log_fname=mocker.Mock(), + session=mocker.Mock(), + ) + + assert workload is not None + assert trigger.id not in jobless_supervisor.logger_cache + + def test_create_workload_sets_watched_assets_for_asset_only_trigger(jobless_supervisor, mocker): """_create_workload() should populate watched_assets when trigger.task_instance is None and assets exist.""" asset1 = mocker.Mock(spec=Asset) diff --git a/airflow-core/tests/unit/utils/log/test_callback_log_reader.py b/airflow-core/tests/unit/utils/log/test_callback_log_reader.py new file mode 100644 index 0000000000000..2400854b54e43 --- /dev/null +++ b/airflow-core/tests/unit/utils/log/test_callback_log_reader.py @@ -0,0 +1,100 @@ +# 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 unittest.mock import patch + +import pytest + +from airflow.utils.log.callback_log_reader import read_callback_log, validate_log_path_component + +from tests_common.test_utils.config import conf_vars + + +class TestValidateLogPathComponent: + @pytest.mark.parametrize("component", ["my_dag", "manual__2024-01-01T00:00:00+00:00", "abc.123~x@y"]) + def test_valid_components_pass(self, component): + assert validate_log_path_component(component) == component + + @pytest.mark.parametrize("component", ["..", ".", "a/b", "a\\b", "", "a b", "../etc"]) + def test_unsafe_components_raise(self, component): + with pytest.raises(ValueError, match="Invalid log path component"): + validate_log_path_component(component) + + +class TestReadCallbackLog: + def test_no_logs_found_yields_message(self, tmp_path): + with conf_vars({("logging", "base_log_folder"): str(tmp_path)}): + msgs = list(read_callback_log("dag1", "run1", "cb1")) + assert [m.event for m in msgs] == ["No callback logs found."] + + @pytest.mark.parametrize("prefix", ["executor_callbacks", "triggerer_callbacks"]) + def test_reads_local_logs(self, tmp_path, prefix): + log_dir = tmp_path / prefix / "dag1" / "run1" + log_dir.mkdir(parents=True) + (log_dir / "cb1").write_text(f"{prefix} line\n") + + with conf_vars({("logging", "base_log_folder"): str(tmp_path)}): + msgs = list(read_callback_log("dag1", "run1", "cb1")) + + assert any(m.event == f"{prefix} line" for m in msgs) + + def test_executor_path_preferred_over_triggerer(self, tmp_path): + for prefix in ("executor_callbacks", "triggerer_callbacks"): + log_dir = tmp_path / prefix / "dag1" / "run1" + log_dir.mkdir(parents=True) + (log_dir / "cb1").write_text(f"{prefix} line\n") + + with conf_vars({("logging", "base_log_folder"): str(tmp_path)}): + events = [m.event for m in read_callback_log("dag1", "run1", "cb1")] + + assert "executor_callbacks line" in events + assert "triggerer_callbacks line" not in events + + def test_symlink_escaping_log_folder_is_skipped(self, tmp_path): + outside = tmp_path / "outside" + outside.mkdir() + secret = outside / "secret" + secret.write_text("secret data\n") + + log_folder = tmp_path / "logs" + log_dir = log_folder / "executor_callbacks" / "dag1" / "run1" + log_dir.mkdir(parents=True) + (log_dir / "cb1").symlink_to(secret) + + with conf_vars({("logging", "base_log_folder"): str(log_folder)}): + msgs = list(read_callback_log("dag1", "run1", "cb1")) + + assert [m.event for m in msgs] == ["No callback logs found."] + + def test_remote_logs_used_when_available(self, tmp_path): + def one_stream(): + yield '{"event": "remote line"}\n' + + with conf_vars({("logging", "base_log_folder"): str(tmp_path)}): + with patch( + "airflow.utils.log.callback_log_reader._read_callback_remote_logs", + return_value=(["s3://bucket/log"], [one_stream()]), + ): + msgs = list(read_callback_log("dag1", "run1", "cb1")) + + assert any(m.event == "remote line" for m in msgs) + + def test_path_traversal_components_rejected(self, tmp_path): + with conf_vars({("logging", "base_log_folder"): str(tmp_path)}): + with pytest.raises(ValueError, match="Invalid log path component"): + list(read_callback_log("../etc", "run1", "cb1"))