You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "seanghaeli (via GitHub)" <gi...@apache.org> on 2026/09/21 23:56:36 UTC
[PR] Add API endpoint and UI for deadline callback log visibility [airflow]
seanghaeli opened a new pull request, #66610:
URL: https://github.com/apache/airflow/pull/66610
## Summary
Make deadline callback logs visible in the Airflow UI. When a deadline is missed and a callback fires, users can now view the callback's execution logs directly from the web interface.
## Changes
**Backend:**
- New `CallbackLogReader` utility (`airflow-core/src/airflow/utils/log/callback_log_reader.py`) — reads callback logs from remote or local storage using the same `RemoteLogIO` infrastructure as task logs
- New API endpoint `GET /ui/dags/{dag_id}/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs` on the deadlines UI router
- Updated `DeadlineResponse` datamodel — added `callback_id` and `callback_state` fields
**Frontend:**
- `CallbackLogViewer.tsx` component — fetches and displays callback execution logs in a modal dialog
- Simplified deadline status display: inline "Missed" badge + "Callback Logs" button + expected/actual times (no success badge, no "Finished late" text, no alert name)
## Motivation
Kaxil requested this: "Some way to know if a deadline didn't fire and it failed: something on UI would be great, like Task log. If deadlines are visible, it would be even better experience than previous SLAs."
## Dependencies
- Builds on #66379 (remote log upload for callbacks) which uploads the logs to storage
- The log path format is `executor_callbacks/{dag_id}/{run_id}/{callback_id}`
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (Opus 4.6)
Generated-by: Claude Code (Opus 4.6) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
## End-to-End Verification
Verified in breeze with full Airflow stack (scheduler + dag-processor + triggerer + API server + LocalExecutor). A test DAG with a `SyncCallback` deadline was parsed, unpaused, scheduled, and the deadline missed naturally. The callback executed through `supervise_callback()` and produced real structured logs.
### Dag Run Page — Deadline Status with Callback
### Callback Logs Modal
**Verified:**
- 6/6 API unit tests pass
- 12/12 log reader unit tests pass
- Path traversal protection via regex validation + `os.path.commonpath` containment check
- Stress tested with 10K-100K line logs: 10K lines renders in 0.23s, acceptable for realistic callback output
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3237872891
##########
airflow-core/src/airflow/utils/log/callback_log_reader.py:
##########
@@ -0,0 +1,150 @@
+# 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 logging
+import os
+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,
+ _interleave_logs,
+ _stream_lines_by_chunk,
+)
+
+if TYPE_CHECKING:
+ from airflow._shared.logging.remote import LogSourceInfo, RawLogStream
+
+logger = logging.getLogger(__name__)
+
+
+def _get_callback_log_relative_path(dag_id: str, run_id: str, callback_id: str) -> str:
+ """
+ Construct the relative log path for a callback execution.
+
+ This must match the path format used in ExecuteCallback.make():
+ executor_callbacks/{dag_id}/{run_id}/{callback_id}
+ """
+ return f"executor_callbacks/{dag_id}/{run_id}/{callback_id}"
+
+
+def read_callback_log(
+ dag_id: str,
+ run_id: str,
+ callback_id: str,
+) -> Generator[StructuredLogMessage, None, None]:
+ """
+ Read callback logs from remote and/or local storage.
+
+ Tries remote storage first (if configured), then falls back to local filesystem.
+ Returns a generator of StructuredLogMessage objects suitable for the API response.
+
+ :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_path = _get_callback_log_relative_path(dag_id, run_id, callback_id)
+
+ sources: LogSourceInfo = []
+ remote_logs: list[RawLogStream] = []
+ local_logs: list[RawLogStream] = []
+
+ # Try remote storage first
+ with suppress(Exception):
+ remote_sources, remote_log_streams = _read_callback_remote_logs(relative_path)
+ if remote_log_streams:
+ sources.extend(remote_sources)
+ remote_logs.extend(remote_log_streams)
+
+ # Try local filesystem
+ if not remote_logs:
+ local_sources, local_log_streams = _read_callback_local_logs(relative_path)
+ if local_log_streams:
+ sources.extend(local_sources)
+ local_logs.extend(local_log_streams)
+
+ if not remote_logs and not local_logs:
+ yield StructuredLogMessage(event="No callback logs found.", timestamp=None)
+ return
+
+ # Emit source information header
+ yield StructuredLogMessage(event="::group::Log message source details", sources=sources) # type: ignore[call-arg]
+ yield StructuredLogMessage(event="::endgroup::")
+
+ # Interleave and yield all log streams
+ log_stream = _interleave_logs(*remote_logs, *local_logs)
+ yield from log_stream
+
+
+def _read_callback_remote_logs(
+ relative_path: str,
+) -> tuple[list[str], list[RawLogStream]]:
+ """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 [], []
+
+ # RemoteLogIO.read() takes (relative_path, ti) -- for S3 the ti is not used,
+ # for CloudWatch it uses ti.end_date (with getattr fallback to None).
+ # We pass None since callbacks don't have a TaskInstance.
+ 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, []
+
+ # Convert legacy string logs to stream format
+ from airflow.utils.log.file_task_handler import _get_compatible_log_stream
+
+ return sources, [_get_compatible_log_stream(logs)]
+
+
+def _read_callback_local_logs(
+ relative_path: str,
+) -> tuple[list[str], list[RawLogStream]]:
+ """Read callback logs from the local filesystem."""
+ base_log_folder = conf.get("logging", "base_log_folder")
+ log_path = Path(base_log_folder) / relative_path
+
+ # Prevent path traversal attacks
+ if not log_path.resolve().is_relative_to(Path(base_log_folder).resolve()):
Review Comment:
## CodeQL / Uncontrolled data used in path expression
This path depends on a [user-provided value](1).
This path depends on a [user-provided value](2).
This path depends on a [user-provided value](3).
[Show more details](https://github.com/apache/airflow/security/code-scanning/600)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4711209497
> Thanks @SameerMesiah97, that's good feedback. I'll leave it out of this PR so we're making less assumptions about what the user wants to see, but will keep this in mind for a follow-up
Yes. Personally, it’s up to you (and the AWS team). Those were just my thoughts when I saw your dev list post.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4524246270
## End-to-End Verification Screenshots
The UI callback logs feature has been verified end-to-end in breeze. The following screenshots show:
### Dag Run Page — Deadline Status with Callback State
The run header shows the **Missed** deadline badge with the "SLA Check" alert name, expected time, and a green **success** badge for the callback. A **"View Logs"** button is available.
### Callback Logs Modal
Clicking "View Logs" opens a dialog showing the structured callback log content fetched from the API endpoint (`/ui/dags/{dagId}/dagRuns/{runId}/callbacks/{callbackId}/logs`).
**Verified:**
- Deadlines API returns 200 with correct deadline data (`callback_id`, `callback_state`, `alert_name`)
- Callback Logs API returns 200 with structured log entries
- UI renders DeadlineStatus component in run header
- "View Logs" button opens CallbackLogViewer modal with log content
- Path traversal guard in `callback_log_reader.py` prevents directory escape
---
Drafted-by: Claude Opus 4.6 (1M context) (no human review before posting)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3211741013
##########
airflow-core/src/airflow/utils/log/callback_log_reader.py:
##########
@@ -0,0 +1,146 @@
+# 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 logging
+import os
+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,
+ _interleave_logs,
+ _stream_lines_by_chunk,
+)
+
+if TYPE_CHECKING:
+ from airflow._shared.logging.remote import LogSourceInfo, RawLogStream
+
+logger = logging.getLogger(__name__)
+
+
+def _get_callback_log_relative_path(dag_id: str, run_id: str, callback_id: str) -> str:
+ """
+ Construct the relative log path for a callback execution.
+
+ This must match the path format used in ExecuteCallback.make():
+ executor_callbacks/{dag_id}/{run_id}/{callback_id}
+ """
+ return f"executor_callbacks/{dag_id}/{run_id}/{callback_id}"
+
+
+def read_callback_log(
+ dag_id: str,
+ run_id: str,
+ callback_id: str,
+) -> Generator[StructuredLogMessage, None, None]:
+ """
+ Read callback logs from remote and/or local storage.
+
+ Tries remote storage first (if configured), then falls back to local filesystem.
+ Returns a generator of StructuredLogMessage objects suitable for the API response.
+
+ :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_path = _get_callback_log_relative_path(dag_id, run_id, callback_id)
+
+ sources: LogSourceInfo = []
+ remote_logs: list[RawLogStream] = []
+ local_logs: list[RawLogStream] = []
+
+ # Try remote storage first
+ with suppress(Exception):
+ remote_sources, remote_log_streams = _read_callback_remote_logs(relative_path)
+ if remote_log_streams:
+ sources.extend(remote_sources)
+ remote_logs.extend(remote_log_streams)
+
+ # Try local filesystem
+ if not remote_logs:
+ local_sources, local_log_streams = _read_callback_local_logs(relative_path)
+ if local_log_streams:
+ sources.extend(local_sources)
+ local_logs.extend(local_log_streams)
+
+ if not remote_logs and not local_logs:
+ yield StructuredLogMessage(event="No callback logs found.", timestamp=None)
+ return
+
+ # Emit source information header
+ yield StructuredLogMessage(event="::group::Log message source details", sources=sources) # type: ignore[call-arg]
+ yield StructuredLogMessage(event="::endgroup::")
+
+ # Interleave and yield all log streams
+ log_stream = _interleave_logs(*remote_logs, *local_logs)
+ yield from log_stream
+
+
+def _read_callback_remote_logs(
+ relative_path: str,
+) -> tuple[list[str], list[RawLogStream]]:
+ """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 [], []
+
+ # RemoteLogIO.read() takes (relative_path, ti) -- for S3 the ti is not used,
+ # for CloudWatch it uses ti.end_date (with getattr fallback to None).
+ # We pass None since callbacks don't have a TaskInstance.
+ 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)
+ if not logs:
+ return sources, []
+
+ # Convert legacy string logs to stream format
+ from airflow.utils.log.file_task_handler import _get_compatible_log_stream
+
+ return sources, [_get_compatible_log_stream(logs)]
+
+
+def _read_callback_local_logs(
+ relative_path: str,
+) -> tuple[list[str], list[RawLogStream]]:
+ """Read callback logs from the local filesystem."""
+ base_log_folder = conf.get("logging", "base_log_folder")
+ log_path = Path(base_log_folder) / relative_path
+
+ sources: list[str] = []
+ log_streams: list[RawLogStream] = []
+
+ # Look for log files matching the path pattern (similar to FileTaskHandler._read_from_local)
+ paths = sorted(log_path.parent.glob(log_path.name + "*"))
Review Comment:
## CodeQL / Uncontrolled data used in path expression
This path depends on a [user-provided value](1).
This path depends on a [user-provided value](2).
This path depends on a [user-provided value](3).
[Show more details](https://github.com/apache/airflow/security/code-scanning/599)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3448184594
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py:
##########
@@ -201,3 +206,80 @@ def get_dag_deadline_alerts(
alerts = session.scalars(alerts_select)
return DeadlineAlertCollectionResponse(deadline_alerts=alerts, total_entries=total_entries)
+
+
+@deadlines_router.get(
+ "/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs",
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ ]
+ ),
+ dependencies=[
+ Depends(
+ requires_access_dag(
+ method="GET",
+ access_entity=DagAccessEntity.TASK_LOGS,
+ )
+ ),
+ ],
+ response_model=TaskInstancesLogResponse,
+ response_model_exclude_unset=True,
+)
+def get_callback_logs(
+ dag_id: str,
+ dag_run_id: str,
+ callback_id: UUID,
+ session: SessionDep,
+) -> TaskInstancesLogResponse:
+ """
+ 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.
+ """
+ # Sanitize path components to prevent path traversal via URL parameters.
+ for param_name, param_value in [("dag_id", dag_id), ("dag_run_id", dag_run_id)]:
+ if os.sep in param_value or "\\" in param_value or ".." in param_value:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"Invalid characters in {param_name}",
+ )
+
+ # Verify the callback exists
+ callback = session.scalar(select(Callback).where(Callback.id == callback_id))
+ if callback is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Callback with id `{callback_id}` was not found",
+ )
+
+ # Verify the dag_run exists with matching dag_id
+ dag_run = session.scalar(select(DagRun).where(DagRun.dag_id == dag_id, DagRun.run_id == dag_run_id))
+ if dag_run is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"DagRun with dag_id: `{dag_id}` and run_id: `{dag_run_id}` was not found",
+ )
+
+ # Verify the callback actually belongs to this dag_run (via the Deadline relationship)
+ deadline = session.scalar(
+ select(Deadline).where(Deadline.callback_id == callback_id, Deadline.dagrun_id == dag_run.id)
+ )
+ if deadline is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Callback `{callback_id}` is not associated with DagRun `{dag_run_id}` of Dag `{dag_id}`",
+ )
+
+ try:
+ log_stream = read_callback_log(
+ dag_id=dag_id,
+ run_id=dag_run_id,
+ callback_id=str(callback_id),
+ )
+ except ValueError:
+ raise HTTPException(status.HTTP_400_BAD_REQUEST, "Invalid callback log path")
+
+ content = list(log_stream)
+ return TaskInstancesLogResponse.model_construct(content=content, continuation_token=None)
Review Comment:
We should support JSONL "content type" header type that support streaming response like the Get TaskInstance logs route.
The `list` copy and the pydantic model serialization will cause the OOM for API server when the log size increases.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py:
##########
@@ -201,3 +206,80 @@ def get_dag_deadline_alerts(
alerts = session.scalars(alerts_select)
return DeadlineAlertCollectionResponse(deadline_alerts=alerts, total_entries=total_entries)
+
+
+@deadlines_router.get(
+ "/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs",
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ ]
+ ),
+ dependencies=[
+ Depends(
+ requires_access_dag(
+ method="GET",
+ access_entity=DagAccessEntity.TASK_LOGS,
+ )
+ ),
+ ],
+ response_model=TaskInstancesLogResponse,
+ response_model_exclude_unset=True,
+)
+def get_callback_logs(
+ dag_id: str,
+ dag_run_id: str,
+ callback_id: UUID,
+ session: SessionDep,
+) -> TaskInstancesLogResponse:
+ """
+ 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.
+ """
+ # Sanitize path components to prevent path traversal via URL parameters.
+ for param_name, param_value in [("dag_id", dag_id), ("dag_run_id", dag_run_id)]:
+ if os.sep in param_value or "\\" in param_value or ".." in param_value:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"Invalid characters in {param_name}",
+ )
Review Comment:
Would it be better to move this to depends level so that all the injected parameters at route level are already valid.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py:
##########
@@ -201,3 +206,80 @@ def get_dag_deadline_alerts(
alerts = session.scalars(alerts_select)
return DeadlineAlertCollectionResponse(deadline_alerts=alerts, total_entries=total_entries)
+
+
+@deadlines_router.get(
+ "/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs",
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ ]
+ ),
+ dependencies=[
+ Depends(
+ requires_access_dag(
+ method="GET",
+ access_entity=DagAccessEntity.TASK_LOGS,
+ )
+ ),
+ ],
+ response_model=TaskInstancesLogResponse,
+ response_model_exclude_unset=True,
+)
+def get_callback_logs(
+ dag_id: str,
+ dag_run_id: str,
+ callback_id: UUID,
+ session: SessionDep,
+) -> TaskInstancesLogResponse:
+ """
+ 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.
+ """
+ # Sanitize path components to prevent path traversal via URL parameters.
+ for param_name, param_value in [("dag_id", dag_id), ("dag_run_id", dag_run_id)]:
+ if os.sep in param_value or "\\" in param_value or ".." in param_value:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"Invalid characters in {param_name}",
+ )
+
+ # Verify the callback exists
+ callback = session.scalar(select(Callback).where(Callback.id == callback_id))
+ if callback is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Callback with id `{callback_id}` was not found",
+ )
+
Review Comment:
The PR body claims "6/6 API unit tests pass (test_callback_logs.py)", but no such file is in the diff.
##########
airflow-core/src/airflow/utils/log/callback_log_reader.py:
##########
@@ -0,0 +1,182 @@
+# 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 logging
+import os
+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,
+ _interleave_logs,
+ _stream_lines_by_chunk,
+)
+
+if TYPE_CHECKING:
+ from airflow._shared.logging.remote import LogSourceInfo, RawLogStream
+
+logger = logging.getLogger(__name__)
+
+
+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.
+
+ Returns paths for both executor callbacks (sync) and triggerer callbacks (async).
+ 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}
+ """
+ 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]:
+ """
+ Read 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 local filesystem.
+ Returns a generator of StructuredLogMessage objects suitable for the API response.
+
+ :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 = []
+ remote_logs: list[RawLogStream] = []
+ local_logs: list[RawLogStream] = []
+
+ for relative_path in relative_paths:
+ # Try remote storage first
+ with suppress(Exception):
+ remote_sources, remote_log_streams = _read_callback_remote_logs(relative_path)
+ if remote_log_streams:
+ sources.extend(remote_sources)
+ remote_logs.extend(remote_log_streams)
+
+ # Try local filesystem
+ if not remote_logs:
+ local_sources, local_log_streams = _read_callback_local_logs(relative_path)
+ if local_log_streams:
+ sources.extend(local_sources)
+ local_logs.extend(local_log_streams)
+
+ # If we found logs at this path, no need to check the next path
+ if remote_logs or local_logs:
+ break
+
+ if not remote_logs and not local_logs:
+ yield StructuredLogMessage(event="No callback logs found.", timestamp=None)
+ return
+
+ # Emit source information header
+ yield StructuredLogMessage(event="::group::Log message source details", sources=sources) # type: ignore[call-arg]
+ yield StructuredLogMessage(event="::endgroup::")
+
+ # Interleave and yield all log streams
+ log_stream = _interleave_logs(*remote_logs, *local_logs)
+ yield from log_stream
+
+
+def _read_callback_remote_logs(
+ relative_path: str,
+) -> tuple[list[str], list[RawLogStream]]:
Review Comment:
There's an existing `StreamingLogResponse` type annotation alias for `tuple[list[str], list[RawLogStream]]`.
```suggestion
def _read_callback_remote_logs(
relative_path: str,
) -> StreamingLogResponse:
```
https://github.com/apache/airflow/blob/db9fa4289fe0d01e764b377c5c7f13d247af8218/airflow-core/src/airflow/utils/log/file_task_handler.py#L967-L968
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -428,8 +428,35 @@ def upload_to_remote(self):
# Never actually called, nothing to do
return
+ if self.ti is None:
+ # Callback triggers have no task instance — upload using the path directly.
+ self._upload_callback_log_to_remote()
+ return
+
upload_to_remote(self.bound_logger, self.ti)
+ def _upload_callback_log_to_remote(self):
+ """Upload callback trigger logs to remote storage without a task instance."""
+ from airflow.sdk.log import load_remote_log_handler, relative_path_from_logger
Review Comment:
`sdk/log.upload_to_remote(logger, ti=None)` already defaults and accepts `ti=None`, and `RemoteLogIO.upload(self, path, ti=None)` (`_shared/logging/remote.py:59`) accepts it too.
`_upload_callback_log_to_remote`'s body is a near-verbatim copy of `upload_to_remote` minus a `try/except` warning — and the caller at `triggerer_job_runner.py:555-559` *already* wraps `factory.upload_to_remote()` in a try/except that logs. So both the `if self.ti is None:` dispatch and the whole helper can go; just call `upload_to_remote(self.bound_logger, self.ti)` unconditionally. That removes ~20 lines and the `# type: ignore[arg-type]`.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py:
##########
@@ -201,3 +206,80 @@ def get_dag_deadline_alerts(
alerts = session.scalars(alerts_select)
return DeadlineAlertCollectionResponse(deadline_alerts=alerts, total_entries=total_entries)
+
+
+@deadlines_router.get(
+ "/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs",
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ ]
+ ),
+ dependencies=[
+ Depends(
+ requires_access_dag(
+ method="GET",
+ access_entity=DagAccessEntity.TASK_LOGS,
+ )
+ ),
+ ],
+ response_model=TaskInstancesLogResponse,
+ response_model_exclude_unset=True,
+)
+def get_callback_logs(
+ dag_id: str,
+ dag_run_id: str,
+ callback_id: UUID,
+ session: SessionDep,
+) -> TaskInstancesLogResponse:
+ """
+ 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.
+ """
+ # Sanitize path components to prevent path traversal via URL parameters.
+ for param_name, param_value in [("dag_id", dag_id), ("dag_run_id", dag_run_id)]:
+ if os.sep in param_value or "\\" in param_value or ".." in param_value:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"Invalid characters in {param_name}",
+ )
+
+ # Verify the callback exists
+ callback = session.scalar(select(Callback).where(Callback.id == callback_id))
Review Comment:
Additionally, IIUC, for these select statement we just need to make sure these entities (DagRun, Deadline as well) are "existed". No need to select all the columns.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py:
##########
@@ -201,3 +206,80 @@ def get_dag_deadline_alerts(
alerts = session.scalars(alerts_select)
return DeadlineAlertCollectionResponse(deadline_alerts=alerts, total_entries=total_entries)
+
+
+@deadlines_router.get(
+ "/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs",
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ ]
+ ),
+ dependencies=[
+ Depends(
+ requires_access_dag(
+ method="GET",
+ access_entity=DagAccessEntity.TASK_LOGS,
+ )
+ ),
+ ],
+ response_model=TaskInstancesLogResponse,
+ response_model_exclude_unset=True,
+)
+def get_callback_logs(
+ dag_id: str,
+ dag_run_id: str,
+ callback_id: UUID,
+ session: SessionDep,
+) -> TaskInstancesLogResponse:
+ """
+ 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.
+ """
+ # Sanitize path components to prevent path traversal via URL parameters.
+ for param_name, param_value in [("dag_id", dag_id), ("dag_run_id", dag_run_id)]:
+ if os.sep in param_value or "\\" in param_value or ".." in param_value:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"Invalid characters in {param_name}",
+ )
+
+ # Verify the callback exists
+ callback = session.scalar(select(Callback).where(Callback.id == callback_id))
+ if callback is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Callback with id `{callback_id}` was not found",
+ )
+
+ # Verify the dag_run exists with matching dag_id
+ dag_run = session.scalar(select(DagRun).where(DagRun.dag_id == dag_id, DagRun.run_id == dag_run_id))
+ if dag_run is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"DagRun with dag_id: `{dag_id}` and run_id: `{dag_run_id}` was not found",
+ )
+
+ # Verify the callback actually belongs to this dag_run (via the Deadline relationship)
+ deadline = session.scalar(
+ select(Deadline).where(Deadline.callback_id == callback_id, Deadline.dagrun_id == dag_run.id)
+ )
+ if deadline is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Callback `{callback_id}` is not associated with DagRun `{dag_run_id}` of Dag `{dag_id}`",
+ )
+
+ try:
+ log_stream = read_callback_log(
+ dag_id=dag_id,
+ run_id=dag_run_id,
+ callback_id=str(callback_id),
+ )
+ except ValueError:
+ raise HTTPException(status.HTTP_400_BAD_REQUEST, "Invalid callback log path")
Review Comment:
The `read_callback_log` is a generator, so the `ValueError` exception here is a dead code that will never fired.
##########
airflow-core/src/airflow/utils/log/callback_log_reader.py:
##########
@@ -0,0 +1,182 @@
+# 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 logging
+import os
+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,
+ _interleave_logs,
+ _stream_lines_by_chunk,
+)
+
+if TYPE_CHECKING:
+ from airflow._shared.logging.remote import LogSourceInfo, RawLogStream
+
+logger = logging.getLogger(__name__)
+
+
+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.
+
+ Returns paths for both executor callbacks (sync) and triggerer callbacks (async).
+ 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}
+ """
+ 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]:
+ """
+ Read 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 local filesystem.
+ Returns a generator of StructuredLogMessage objects suitable for the API response.
+
+ :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 = []
+ remote_logs: list[RawLogStream] = []
+ local_logs: list[RawLogStream] = []
+
+ for relative_path in relative_paths:
+ # Try remote storage first
+ with suppress(Exception):
+ remote_sources, remote_log_streams = _read_callback_remote_logs(relative_path)
+ if remote_log_streams:
+ sources.extend(remote_sources)
+ remote_logs.extend(remote_log_streams)
+
+ # Try local filesystem
+ if not remote_logs:
+ local_sources, local_log_streams = _read_callback_local_logs(relative_path)
+ if local_log_streams:
+ sources.extend(local_sources)
+ local_logs.extend(local_log_streams)
+
+ # If we found logs at this path, no need to check the next path
+ if remote_logs or local_logs:
+ break
+
+ if not remote_logs and not local_logs:
+ yield StructuredLogMessage(event="No callback logs found.", timestamp=None)
+ return
+
+ # Emit source information header
+ yield StructuredLogMessage(event="::group::Log message source details", sources=sources) # type: ignore[call-arg]
+ yield StructuredLogMessage(event="::endgroup::")
+
+ # Interleave and yield all log streams
+ log_stream = _interleave_logs(*remote_logs, *local_logs)
+ yield from log_stream
+
+
+def _read_callback_remote_logs(
+ relative_path: str,
+) -> tuple[list[str], list[RawLogStream]]:
+ """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 [], []
+
+ # RemoteLogIO.read() takes (relative_path, ti) -- for S3 the ti is not used,
+ # for CloudWatch it uses ti.end_date (with getattr fallback to None).
+ # We pass None since callbacks don't have a TaskInstance.
+ 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, []
+
+ # Convert legacy string logs to stream format
+ from airflow.utils.log.file_task_handler import _get_compatible_log_stream
+
+ return sources, [_get_compatible_log_stream(logs)]
+
+
+def _validate_path_component(component: str) -> str:
+ """Validate and return a path component, raising ValueError if unsafe."""
+ import re
+
+ if component in (".", "..") or not re.fullmatch(r"[A-Za-z0-9._:+\-~@]+", component):
+ raise ValueError(f"Invalid path component: {component!r}")
+ return component
+
+
+def _read_callback_local_logs(
+ relative_path: str,
+) -> tuple[list[str], list[RawLogStream]]:
Review Comment:
Ditto.
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py:
##########
@@ -37,6 +37,8 @@ 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: str | None = Field(validation_alias=AliasPath("callback", "state"), default=None)
Review Comment:
Should we annotate as `CallbackState` instead of `str`?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3450010906
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py:
##########
@@ -201,3 +206,80 @@ def get_dag_deadline_alerts(
alerts = session.scalars(alerts_select)
return DeadlineAlertCollectionResponse(deadline_alerts=alerts, total_entries=total_entries)
+
+
+@deadlines_router.get(
+ "/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs",
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ ]
+ ),
+ dependencies=[
+ Depends(
+ requires_access_dag(
+ method="GET",
+ access_entity=DagAccessEntity.TASK_LOGS,
+ )
+ ),
+ ],
+ response_model=TaskInstancesLogResponse,
+ response_model_exclude_unset=True,
+)
+def get_callback_logs(
+ dag_id: str,
+ dag_run_id: str,
+ callback_id: UUID,
+ session: SessionDep,
+) -> TaskInstancesLogResponse:
+ """
+ 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.
+ """
+ # Sanitize path components to prevent path traversal via URL parameters.
+ for param_name, param_value in [("dag_id", dag_id), ("dag_run_id", dag_run_id)]:
+ if os.sep in param_value or "\\" in param_value or ".." in param_value:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"Invalid characters in {param_name}",
+ )
Review Comment:
I agree it would be better, but I'm going to leave it since it's a nice-to-have and refactor + re-verify overhead won't be worth 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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4734591400
Can you update the screenshots to reflect the last changes?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3461031217
##########
airflow-core/src/airflow/ui/src/pages/Run/CallbackLogViewer.tsx:
##########
@@ -0,0 +1,184 @@
+/*!
+ * 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.
+ */
+import { Box, Button, Heading, HStack, Text } from "@chakra-ui/react";
+import type { TFunction } from "i18next";
+import { useMemo, useState } from "react";
+import { useTranslation } from "react-i18next";
+import { FiFileText } from "react-icons/fi";
+import innerText from "react-innertext";
+
+import { useDeadlinesServiceGetCallbackLogs } from "openapi/queries";
+import type { TaskInstanceState, TaskInstancesLogResponse } from "openapi/requests/types.gen";
+import { StateBadge } from "src/components/StateBadge";
+import { renderStructuredLog } from "src/components/renderStructuredLog";
+import { Dialog } from "src/components/ui";
+import { TaskLogContent } from "src/pages/TaskInstance/Logs/TaskLogContent";
+import type { ParsedLogEntry } from "src/queries/useLogs";
+import { parseStreamingLogContent } from "src/utils/logs";
+
+type CallbackLogViewerProps = {
+ readonly callbackId: string;
+ readonly callbackState?: TaskInstanceState | null;
+ readonly dagId: string;
+ readonly dagRunId: string;
+};
+
+/**
+ * Parse callback log data using the same structured log rendering pipeline
+ * as the task instance logs, providing consistent formatting, grouping, and display.
+ */
+const parseCallbackLogs = (
+ data: TaskInstancesLogResponse["content"],
+ translate: TFunction,
+): Array<ParsedLogEntry> => {
+ let lineNumber = 0;
+ const lineNumbers = data.map((datum) => {
+ const text = typeof datum === "string" ? datum : datum.event;
+
+ if (text.includes("::group::") || text.includes("::endgroup::")) {
+ return undefined;
+ }
+ const current = lineNumber;
+
Review Comment:
Looks duplicated from the current parseLog function we can probably re-use some pieces instead.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3461044748
##########
airflow-core/src/airflow/ui/src/pages/Run/DeadlineStatus.tsx:
##########
@@ -160,42 +153,28 @@ export const DeadlineStatus = ({ dagId, dagRunId, endDate }: DeadlineStatusProps
);
}
- // Single deadline — show inline with Expected / Actual dates and precise duration.
+ // Single deadline — show inline with Expected / Actual times.
const [dl] = deadlines;
Review Comment:
can't an entity have multiple missed dealines?
I don't think the way we display this information should have 3 branches based on items count. It's making things confusing.
But that's not related to this PR, can be a follow up.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3461044748
##########
airflow-core/src/airflow/ui/src/pages/Run/DeadlineStatus.tsx:
##########
@@ -160,42 +153,28 @@ export const DeadlineStatus = ({ dagId, dagRunId, endDate }: DeadlineStatusProps
);
}
- // Single deadline — show inline with Expected / Actual dates and precise duration.
+ // Single deadline — show inline with Expected / Actual times.
const [dl] = deadlines;
Review Comment:
can't an entity have multiple missed dealines?
I don't think the way we display this information should have 3 branches based on items count. It's making things confusing.
But that's not related to this PR, can be a follow up.
Can you add screenshot for the log button when there are multiple callbacks/deadlines to show.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3450017330
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py:
##########
@@ -201,3 +206,80 @@ def get_dag_deadline_alerts(
alerts = session.scalars(alerts_select)
return DeadlineAlertCollectionResponse(deadline_alerts=alerts, total_entries=total_entries)
+
+
+@deadlines_router.get(
+ "/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs",
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ ]
+ ),
+ dependencies=[
+ Depends(
+ requires_access_dag(
+ method="GET",
+ access_entity=DagAccessEntity.TASK_LOGS,
+ )
+ ),
+ ],
+ response_model=TaskInstancesLogResponse,
+ response_model_exclude_unset=True,
+)
+def get_callback_logs(
+ dag_id: str,
+ dag_run_id: str,
+ callback_id: UUID,
+ session: SessionDep,
+) -> TaskInstancesLogResponse:
+ """
+ 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.
+ """
+ # Sanitize path components to prevent path traversal via URL parameters.
+ for param_name, param_value in [("dag_id", dag_id), ("dag_run_id", dag_run_id)]:
+ if os.sep in param_value or "\\" in param_value or ".." in param_value:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"Invalid characters in {param_name}",
+ )
+
+ # Verify the callback exists
+ callback = session.scalar(select(Callback).where(Callback.id == callback_id))
Review Comment:
You're right, worth noting that we can collapse this into a single query, but would rather not rework it in this PR to avoid the overhead.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4538571549
@seanghaeli Converting to **draft** — this PR doesn't yet meet our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
- :x: **Merge conflicts**. See [docs](https://github.com/apache/airflow/blob/main/contributing-docs/10_working_with_git.rst).
- :x: **Unresolved review comments**: 1 thread(s). See [docs](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
See the linked criteria for how to fix each item, then mark the PR "Ready for review". This is **not** a rejection — just an invitation to bring the PR up to standard. No rush.
---
_Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this [two-stage triage process](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md#why-the-first-pass-is-automated) so that our maintainers' limited time is spent where it matters most: the conversation with you._
---
Drafted-by: Claude Code (Opus 4.7); reviewed by @potiuk before posting
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3432628566
##########
airflow-core/src/airflow/ui/src/pages/Run/CallbackLogViewer.tsx:
##########
@@ -0,0 +1,189 @@
+/*!
+ * 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.
+ */
+import { Badge, Box, Button, Heading, HStack, Text } from "@chakra-ui/react";
+import type { TFunction } from "i18next";
+import { useMemo, useState } from "react";
+import { useTranslation } from "react-i18next";
+import { FiFileText } from "react-icons/fi";
+import innerText from "react-innertext";
+
+import { useDeadlinesServiceGetCallbackLogs } from "openapi/queries";
+import type { TaskInstancesLogResponse } from "openapi/requests/types.gen";
+import { renderStructuredLog } from "src/components/renderStructuredLog";
+import { Dialog } from "src/components/ui";
+import { TaskLogContent } from "src/pages/TaskInstance/Logs/TaskLogContent";
+import type { ParsedLogEntry } from "src/queries/useLogs";
+import { parseStreamingLogContent } from "src/utils/logs";
+
+type CallbackLogViewerProps = {
+ readonly callbackId: string;
+ readonly callbackState?: string | null;
Review Comment:
Nit: callbackState is typed string; if a generated state enum (TaskInstanceState) fits, prefer it over a bare string
##########
airflow-core/src/airflow/ui/src/pages/Run/CallbackLogViewer.tsx:
##########
@@ -0,0 +1,189 @@
+/*!
+ * 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.
+ */
+import { Badge, Box, Button, Heading, HStack, Text } from "@chakra-ui/react";
+import type { TFunction } from "i18next";
+import { useMemo, useState } from "react";
+import { useTranslation } from "react-i18next";
+import { FiFileText } from "react-icons/fi";
+import innerText from "react-innertext";
+
+import { useDeadlinesServiceGetCallbackLogs } from "openapi/queries";
+import type { TaskInstancesLogResponse } from "openapi/requests/types.gen";
+import { renderStructuredLog } from "src/components/renderStructuredLog";
+import { Dialog } from "src/components/ui";
+import { TaskLogContent } from "src/pages/TaskInstance/Logs/TaskLogContent";
+import type { ParsedLogEntry } from "src/queries/useLogs";
+import { parseStreamingLogContent } from "src/utils/logs";
+
+type CallbackLogViewerProps = {
+ readonly callbackId: string;
+ readonly callbackState?: string | null;
+ readonly dagId: string;
+ readonly dagRunId: string;
+};
+
+const stateColorMap: Record<string, string> = {
Review Comment:
Reuse `StateBadge (state={callbackState})` instead of the hand-rolled `stateColorMap`; it already maps state→colorPalette and adds the state icon, used right next door in Run/Details.tsx:60.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli closed pull request #66610: Add API endpoint and UI for deadline callback log visibility
URL: https://github.com/apache/airflow/pull/66610
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4672167565
@ashb updated to use `TaskLogContent`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3450025081
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py:
##########
@@ -201,3 +206,80 @@ def get_dag_deadline_alerts(
alerts = session.scalars(alerts_select)
return DeadlineAlertCollectionResponse(deadline_alerts=alerts, total_entries=total_entries)
+
+
+@deadlines_router.get(
+ "/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs",
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ ]
+ ),
+ dependencies=[
+ Depends(
+ requires_access_dag(
+ method="GET",
+ access_entity=DagAccessEntity.TASK_LOGS,
+ )
+ ),
+ ],
+ response_model=TaskInstancesLogResponse,
+ response_model_exclude_unset=True,
+)
+def get_callback_logs(
+ dag_id: str,
+ dag_run_id: str,
+ callback_id: UUID,
+ session: SessionDep,
+) -> TaskInstancesLogResponse:
+ """
+ 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.
+ """
+ # Sanitize path components to prevent path traversal via URL parameters.
+ for param_name, param_value in [("dag_id", dag_id), ("dag_run_id", dag_run_id)]:
+ if os.sep in param_value or "\\" in param_value or ".." in param_value:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"Invalid characters in {param_name}",
+ )
+
+ # Verify the callback exists
+ callback = session.scalar(select(Callback).where(Callback.id == callback_id))
+ if callback is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Callback with id `{callback_id}` was not found",
+ )
+
+ # Verify the dag_run exists with matching dag_id
+ dag_run = session.scalar(select(DagRun).where(DagRun.dag_id == dag_id, DagRun.run_id == dag_run_id))
+ if dag_run is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"DagRun with dag_id: `{dag_id}` and run_id: `{dag_run_id}` was not found",
+ )
+
+ # Verify the callback actually belongs to this dag_run (via the Deadline relationship)
+ deadline = session.scalar(
+ select(Deadline).where(Deadline.callback_id == callback_id, Deadline.dagrun_id == dag_run.id)
+ )
+ if deadline is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Callback `{callback_id}` is not associated with DagRun `{dag_run_id}` of Dag `{dag_id}`",
+ )
+
+ try:
+ log_stream = read_callback_log(
+ dag_id=dag_id,
+ run_id=dag_run_id,
+ callback_id=str(callback_id),
+ )
+ except ValueError:
+ raise HTTPException(status.HTTP_400_BAD_REQUEST, "Invalid callback log path")
+
+ content = list(log_stream)
+ return TaskInstancesLogResponse.model_construct(content=content, continuation_token=None)
Review Comment:
Good callout, I'd rather include this fix in a separate PR since 1) it's an edge case that usually doesn't occur 2) would require some thought about how to handle the UI consumer
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3450036538
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py:
##########
@@ -201,3 +206,80 @@ def get_dag_deadline_alerts(
alerts = session.scalars(alerts_select)
return DeadlineAlertCollectionResponse(deadline_alerts=alerts, total_entries=total_entries)
+
+
+@deadlines_router.get(
+ "/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs",
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ ]
+ ),
+ dependencies=[
+ Depends(
+ requires_access_dag(
+ method="GET",
+ access_entity=DagAccessEntity.TASK_LOGS,
+ )
+ ),
+ ],
+ response_model=TaskInstancesLogResponse,
+ response_model_exclude_unset=True,
+)
+def get_callback_logs(
+ dag_id: str,
+ dag_run_id: str,
+ callback_id: UUID,
+ session: SessionDep,
+) -> TaskInstancesLogResponse:
+ """
+ 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.
+ """
+ # Sanitize path components to prevent path traversal via URL parameters.
+ for param_name, param_value in [("dag_id", dag_id), ("dag_run_id", dag_run_id)]:
+ if os.sep in param_value or "\\" in param_value or ".." in param_value:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"Invalid characters in {param_name}",
+ )
+
+ # Verify the callback exists
+ callback = session.scalar(select(Callback).where(Callback.id == callback_id))
+ if callback is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Callback with id `{callback_id}` was not found",
+ )
+
Review Comment:
Relates to a local test file, shouldn't have been mentioned in the PR desc.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3449916381
##########
airflow-core/src/airflow/utils/log/callback_log_reader.py:
##########
@@ -0,0 +1,182 @@
+# 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 logging
+import os
+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,
+ _interleave_logs,
+ _stream_lines_by_chunk,
+)
+
+if TYPE_CHECKING:
+ from airflow._shared.logging.remote import LogSourceInfo, RawLogStream
+
+logger = logging.getLogger(__name__)
+
+
+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.
+
+ Returns paths for both executor callbacks (sync) and triggerer callbacks (async).
+ 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}
+ """
+ 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]:
+ """
+ Read 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 local filesystem.
+ Returns a generator of StructuredLogMessage objects suitable for the API response.
+
+ :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 = []
+ remote_logs: list[RawLogStream] = []
+ local_logs: list[RawLogStream] = []
+
+ for relative_path in relative_paths:
+ # Try remote storage first
+ with suppress(Exception):
+ remote_sources, remote_log_streams = _read_callback_remote_logs(relative_path)
+ if remote_log_streams:
+ sources.extend(remote_sources)
+ remote_logs.extend(remote_log_streams)
+
+ # Try local filesystem
+ if not remote_logs:
+ local_sources, local_log_streams = _read_callback_local_logs(relative_path)
+ if local_log_streams:
+ sources.extend(local_sources)
+ local_logs.extend(local_log_streams)
+
+ # If we found logs at this path, no need to check the next path
+ if remote_logs or local_logs:
+ break
+
+ if not remote_logs and not local_logs:
+ yield StructuredLogMessage(event="No callback logs found.", timestamp=None)
+ return
+
+ # Emit source information header
+ yield StructuredLogMessage(event="::group::Log message source details", sources=sources) # type: ignore[call-arg]
+ yield StructuredLogMessage(event="::endgroup::")
+
+ # Interleave and yield all log streams
+ log_stream = _interleave_logs(*remote_logs, *local_logs)
+ yield from log_stream
+
+
+def _read_callback_remote_logs(
Review Comment:
Additionally, all the `_read_...*` functions are mostly duplicated from the existing `airflow-core/src/airflow/utils/log/log_reader.py`. It's fine to keep this as-is, but it would be nice to introduce common functions between two log reading path as follow-up. Thanks.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4535935819
I like the interface but I would like to have @bbovenzi opinion on this one
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4670897093
Why is the callback logs a modal, rather than a tab?
Additionally we should likely use the existing log component so that it gets the same filtering, formatting, and grouping, not just displayed as plain text.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4671006523
@ashb I thought about making the logs appear in a new tab next to Details where the rest of the logs are, but this page only appears when a deadline is missed, so grouping it up with the “Deadline Missed” section makes more spatial sense in my opinion.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3450035073
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -428,8 +428,35 @@ def upload_to_remote(self):
# Never actually called, nothing to do
return
+ if self.ti is None:
+ # Callback triggers have no task instance — upload using the path directly.
+ self._upload_callback_log_to_remote()
+ return
+
upload_to_remote(self.bound_logger, self.ti)
+ def _upload_callback_log_to_remote(self):
+ """Upload callback trigger logs to remote storage without a task instance."""
+ from airflow.sdk.log import load_remote_log_handler, relative_path_from_logger
Review Comment:
It's a good callout, agreed it's a near duplicate. Since this is related to code cleanup and it touches the triggerer upload path, I'd rather not rework that path
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4736445226
@vincbeck Screenshots are up to date now
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli closed pull request #66610: Add API endpoint and UI for deadline callback log visibility
URL: https://github.com/apache/airflow/pull/66610
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli closed pull request #66610: Add API endpoint and UI for deadline callback log visibility
URL: https://github.com/apache/airflow/pull/66610
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3236434672
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py:
##########
@@ -201,3 +205,59 @@ def get_dag_deadline_alerts(
alerts = session.scalars(alerts_select)
return DeadlineAlertCollectionResponse(deadline_alerts=alerts, total_entries=total_entries)
+
+
+@deadlines_router.get(
+ "/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs",
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ ]
+ ),
+ dependencies=[
+ Depends(
+ requires_access_dag(
+ method="GET",
+ access_entity=DagAccessEntity.TASK_LOGS,
+ )
+ ),
+ ],
+ response_model=TaskInstancesLogResponse,
+ response_model_exclude_unset=True,
+)
+def get_callback_logs(
+ dag_id: str,
+ dag_run_id: str,
+ callback_id: UUID,
+ session: SessionDep,
+) -> TaskInstancesLogResponse:
+ """
+ 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.
+ """
+ # Verify the callback exists
+ callback = session.scalar(select(Callback).where(Callback.id == callback_id))
+ if callback is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Callback with id `{callback_id}` was not found",
+ )
+
+ # Verify the dag_run exists with matching dag_id
+ dag_run = session.scalar(select(DagRun).where(DagRun.dag_id == dag_id, DagRun.run_id == dag_run_id))
+ if dag_run is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"DagRun with dag_id: `{dag_id}` and run_id: `{dag_run_id}` was not found",
+ )
+
Review Comment:
I know this is a WIP, but a small point. You check if the callback exists, and you check if the dagrun exists, but maybe you should verify that the callback is tied to that specific run instead of trusting the input? Or maybe that's overthinking 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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3448560264
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -428,8 +428,35 @@ def upload_to_remote(self):
# Never actually called, nothing to do
return
+ if self.ti is None:
+ # Callback triggers have no task instance — upload using the path directly.
+ self._upload_callback_log_to_remote()
+ return
+
upload_to_remote(self.bound_logger, self.ti)
+ def _upload_callback_log_to_remote(self):
+ """Upload callback trigger logs to remote storage without a task instance."""
+ from airflow.sdk.log import load_remote_log_handler, relative_path_from_logger
Review Comment:
The `_upload_callback_log_to_remote` looks almost exactly same as the existing `upload_to_remote`.
The context that I double check with Claude:
---
`sdk/log.upload_to_remote(logger, ti=None)` already defaults and accepts `ti=None`, and `RemoteLogIO.upload(self, path, ti=None)` (`_shared/logging/remote.py:59`) accepts it too.
`_upload_callback_log_to_remote`'s body is a near-verbatim copy of `upload_to_remote` minus a `try/except` warning — and the caller at `triggerer_job_runner.py:555-559` *already* wraps `factory.upload_to_remote()` in a try/except that logs. So both the `if self.ti is None:` dispatch and the whole helper can go; just call `upload_to_remote(self.bound_logger, self.ti)` unconditionally. That removes ~20 lines and the `# type: ignore[arg-type]`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4765307748
Thanks for the review @jason810496 Fixed what I think is mission critical / low-risk, left off some others for follow on PRs. Let me know if you disagree with any of the points I've chosen to defer
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3461031217
##########
airflow-core/src/airflow/ui/src/pages/Run/CallbackLogViewer.tsx:
##########
@@ -0,0 +1,184 @@
+/*!
+ * 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.
+ */
+import { Box, Button, Heading, HStack, Text } from "@chakra-ui/react";
+import type { TFunction } from "i18next";
+import { useMemo, useState } from "react";
+import { useTranslation } from "react-i18next";
+import { FiFileText } from "react-icons/fi";
+import innerText from "react-innertext";
+
+import { useDeadlinesServiceGetCallbackLogs } from "openapi/queries";
+import type { TaskInstanceState, TaskInstancesLogResponse } from "openapi/requests/types.gen";
+import { StateBadge } from "src/components/StateBadge";
+import { renderStructuredLog } from "src/components/renderStructuredLog";
+import { Dialog } from "src/components/ui";
+import { TaskLogContent } from "src/pages/TaskInstance/Logs/TaskLogContent";
+import type { ParsedLogEntry } from "src/queries/useLogs";
+import { parseStreamingLogContent } from "src/utils/logs";
+
+type CallbackLogViewerProps = {
+ readonly callbackId: string;
+ readonly callbackState?: TaskInstanceState | null;
+ readonly dagId: string;
+ readonly dagRunId: string;
+};
+
+/**
+ * Parse callback log data using the same structured log rendering pipeline
+ * as the task instance logs, providing consistent formatting, grouping, and display.
+ */
+const parseCallbackLogs = (
+ data: TaskInstancesLogResponse["content"],
+ translate: TFunction,
+): Array<ParsedLogEntry> => {
+ let lineNumber = 0;
+ const lineNumbers = data.map((datum) => {
+ const text = typeof datum === "string" ? datum : datum.event;
+
+ if (text.includes("::group::") || text.includes("::endgroup::")) {
+ return undefined;
+ }
+ const current = lineNumber;
+
Review Comment:
This looks really duplicated from the current parseLog function
##########
airflow-core/src/airflow/ui/src/pages/Run/DeadlineStatus.tsx:
##########
@@ -160,42 +153,28 @@ export const DeadlineStatus = ({ dagId, dagRunId, endDate }: DeadlineStatusProps
);
}
- // Single deadline — show inline with Expected / Actual dates and precise duration.
+ // Single deadline — show inline with Expected / Actual times.
const [dl] = deadlines;
Review Comment:
can't an entity have multiple missed dealines? Why is it possible to only see logs of one?
I don't think the way we display this information should have 3 branches based on item count. It's making things confusing.
But that's not related to this PR, can be a follow up.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4672307313
@ashb updated to use `TaskLogContent`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4710840720
Thanks @SameerMesiah97, that's good feedback. I'll leave it out of this PR so we're making less assumptions about what the user wants to see, but will keep this in mind for a follow-up
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3293384320
##########
airflow-core/src/airflow/utils/log/callback_log_reader.py:
##########
@@ -0,0 +1,154 @@
+# 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 logging
+import os
+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,
+ _interleave_logs,
+ _stream_lines_by_chunk,
+)
+
+if TYPE_CHECKING:
+ from airflow._shared.logging.remote import LogSourceInfo, RawLogStream
+
+logger = logging.getLogger(__name__)
+
+
+def _get_callback_log_relative_path(dag_id: str, run_id: str, callback_id: str) -> str:
+ """
+ Construct the relative log path for a callback execution.
+
+ This must match the path format used in ExecuteCallback.make():
+ executor_callbacks/{dag_id}/{run_id}/{callback_id}
+ """
+ return f"executor_callbacks/{dag_id}/{run_id}/{callback_id}"
+
+
+def read_callback_log(
+ dag_id: str,
+ run_id: str,
+ callback_id: str,
+) -> Generator[StructuredLogMessage, None, None]:
+ """
+ Read callback logs from remote and/or local storage.
+
+ Tries remote storage first (if configured), then falls back to local filesystem.
+ Returns a generator of StructuredLogMessage objects suitable for the API response.
+
+ :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_path = _get_callback_log_relative_path(dag_id, run_id, callback_id)
+
+ sources: LogSourceInfo = []
+ remote_logs: list[RawLogStream] = []
+ local_logs: list[RawLogStream] = []
+
+ # Try remote storage first
+ with suppress(Exception):
+ remote_sources, remote_log_streams = _read_callback_remote_logs(relative_path)
+ if remote_log_streams:
+ sources.extend(remote_sources)
+ remote_logs.extend(remote_log_streams)
+
+ # Try local filesystem
+ if not remote_logs:
+ local_sources, local_log_streams = _read_callback_local_logs(relative_path)
+ if local_log_streams:
+ sources.extend(local_sources)
+ local_logs.extend(local_log_streams)
+
+ if not remote_logs and not local_logs:
+ yield StructuredLogMessage(event="No callback logs found.", timestamp=None)
+ return
+
+ # Emit source information header
+ yield StructuredLogMessage(event="::group::Log message source details", sources=sources) # type: ignore[call-arg]
+ yield StructuredLogMessage(event="::endgroup::")
+
+ # Interleave and yield all log streams
+ log_stream = _interleave_logs(*remote_logs, *local_logs)
+ yield from log_stream
+
+
+def _read_callback_remote_logs(
+ relative_path: str,
+) -> tuple[list[str], list[RawLogStream]]:
+ """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 [], []
+
+ # RemoteLogIO.read() takes (relative_path, ti) -- for S3 the ti is not used,
+ # for CloudWatch it uses ti.end_date (with getattr fallback to None).
+ # We pass None since callbacks don't have a TaskInstance.
+ 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, []
+
+ # Convert legacy string logs to stream format
+ from airflow.utils.log.file_task_handler import _get_compatible_log_stream
+
+ return sources, [_get_compatible_log_stream(logs)]
+
+
+def _read_callback_local_logs(
+ relative_path: str,
+) -> tuple[list[str], list[RawLogStream]]:
+ """Read callback logs from the local filesystem."""
+ base_log_folder = conf.get("logging", "base_log_folder")
+ resolved_base = os.path.realpath(base_log_folder)
+ resolved_path = os.path.realpath(os.path.join(base_log_folder, relative_path))
+
+ # Prevent path traversal attacks — reject any path that escapes the base folder.
+ if not resolved_path.startswith(resolved_base + os.sep) and resolved_path != resolved_base:
+ raise ValueError("Path traversal detected: resolved path is outside base log folder")
+
+ # After validation, construct the safe path from the resolved (canonicalized) value.
+ safe_path = Path(resolved_base) / Path(resolved_path).relative_to(resolved_base)
+
+ sources: list[str] = []
+ log_streams: list[RawLogStream] = []
+
+ # Look for log files matching the path pattern (similar to FileTaskHandler._read_from_local)
+ paths = sorted(safe_path.parent.glob(safe_path.name + "*"))
Review Comment:
## CodeQL / Uncontrolled data used in path expression
This path depends on a [user-provided value](1).
This path depends on a [user-provided value](2).
This path depends on a [user-provided value](3).
[Show more details](https://github.com/apache/airflow/security/code-scanning/605)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4524606997
## UI Screenshots
### Dag Run Page — Deadline Status with Callback
The run header shows the **Missed** deadline badge with "(SLA Check)" label, expected/actual times, green **success** callback badge, and **View Logs** button.
### Callback Logs Modal
Clicking "View Logs" opens a dialog showing structured callback execution logs fetched via the new API endpoint.
---
Drafted-by: Claude Opus 4.6 (1M context) (no human review before posting)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4736446391
@vincbeck Screenshots in the PR description are up to date now
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4782617317
Closing this for now, it has a data-layer dependency on the work in #66608, which has been fully reverted (#68909). Will circle back when the dependencies are merged
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #66610:
URL: https://github.com/apache/airflow/pull/66610#issuecomment-4607524452
Looks like my comments have ll been addressed. let me know when 66608 gets merged and I can re-review this one and approve 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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add API endpoint and UI for deadline callback log visibility [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on code in PR #66610:
URL: https://github.com/apache/airflow/pull/66610#discussion_r3319279021
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py:
##########
@@ -201,3 +205,59 @@ def get_dag_deadline_alerts(
alerts = session.scalars(alerts_select)
return DeadlineAlertCollectionResponse(deadline_alerts=alerts, total_entries=total_entries)
+
+
+@deadlines_router.get(
+ "/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs",
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ ]
+ ),
+ dependencies=[
+ Depends(
+ requires_access_dag(
+ method="GET",
+ access_entity=DagAccessEntity.TASK_LOGS,
+ )
+ ),
+ ],
+ response_model=TaskInstancesLogResponse,
+ response_model_exclude_unset=True,
+)
+def get_callback_logs(
+ dag_id: str,
+ dag_run_id: str,
+ callback_id: UUID,
+ session: SessionDep,
+) -> TaskInstancesLogResponse:
+ """
+ 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.
+ """
+ # Verify the callback exists
+ callback = session.scalar(select(Callback).where(Callback.id == callback_id))
+ if callback is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Callback with id `{callback_id}` was not found",
+ )
+
+ # Verify the dag_run exists with matching dag_id
+ dag_run = session.scalar(select(DagRun).where(DagRun.dag_id == dag_id, DagRun.run_id == dag_run_id))
+ if dag_run is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"DagRun with dag_id: `{dag_id}` and run_id: `{dag_run_id}` was not found",
+ )
+
Review Comment:
Good point, added — the endpoint now queries the Deadline to verify the callback is associated with the requested DagRun before returning logs. If the association doesn't exist, it returns 404.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org