This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 71e74bc9554 Add deadline callback log visibility to the Dag Run view
in UI (#73580)
71e74bc9554 is described below
commit 71e74bc9554bd8783050c662af2881e70267ad22
Author: Sean Ghaeli <[email protected]>
AuthorDate: Sun Oct 18 00:49:53 2026 -0700
Add deadline callback log visibility to the Dag Run view in UI (#73580)
* Show deadline callback logs in a Callbacks tab on the Dag run view
Add a Callbacks tab to the Dag run view, shown only when the run has
deadline callbacks. It lists them (path, type, state, deadline alert,
deadline time, created at) from the existing deadlines endpoint, which
now also returns each deadline's callback id, type, state and path, and
links each one to its logs, rendered with the task instance log viewer.
The new GET
/ui/dags/{dag_id}/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs
reads a callback's logs from remote storage or the local log folder. The
triggerer now writes async callback output to triggerer_callbacks/... so
those logs can be read too.
* Show deadline and alert details above callback logs
Above a callback's logs, show the deadline alert and its rule, the
deadline time, how late the run was, and the callback type, state and
path, all from the existing deadlines and deadline alerts endpoints.
In the Callbacks tab, replace the Created At column with Missed by and
show callback types as Sync/Async.
* Show the callback's row from the Callbacks table above its logs
* Even out spacing above the callback log header
* Trim callback log UI and tests
* Explain why the callback log reader prefers stream() over read()
* Make the callback log reader generic
Replace read_callback_log with read_logs_at_paths, which reads logs not
tied to a TaskInstance from the given paths. The callback logs route now
builds the executor/triggerer callback paths itself.
* Type callback columns for react-table v9
* Give each triggerer callback its own log directory
With [logging] delete_local_logs=True, the S3, GCS, OSS, CloudWatch and
Elasticsearch handlers delete the uploaded log file's parent directory.
Triggerer callback logs sat directly in the per-Dag-run directory, so the
first callback to finish deleted the local logs of every other callback
still running in the same Dag run, and those were never uploaded.
* Expand callback logs in place in the Callbacks tab
Clicking Logs on a callback now opens its logs in a row under it,
instead of going to a separate page that repeated the callback's row.
That page also looked the callback up in the first page of the list,
so it lost the summary for callbacks past that page; it is removed.
DataTable gains optional getRowId and renderSubComponent props, using
TanStack's row expanding feature.
* Drop the isFetching prop from the Callbacks table
main removed isFetching from DataTable in #74426 ("Drop isFetching from
tables"), which this branch's new Callbacks tab still passed, so tsc
failed after the merge with TS2322 on Callbacks.tsx:154.
* Read callback logs only at the path for the callback's type
CloudWatch returns a lazy stream for any log stream name and reports a
missing
stream as a log line, so trying executor_callbacks/ first always won and
async
(triggerer) callback logs showed "No log stream found" instead of their log.
* Let remote log handlers read logs without a task instance
Widen RemoteLogIO.read and RemoteLogStreamIO.stream to accept ti=None,
matching
upload, so callback logs no longer need a type: ignore. Elasticsearch,
OpenSearch
and Cloud Logging key logs by task instance, so they now return no logs for
ti=None instead of raising AttributeError.
* Fix callback log tests on SQLite by passing a UUID to the callback id
filter
The Uuid column's SQLite bind processor rejects a str, so set_callback_type
failed with "'str' object has no attribute 'hex'".
---
.../api_fastapi/core_api/datamodels/ui/deadline.py | 7 +
.../api_fastapi/core_api/openapi/_private_ui.yaml | 146 +++++++++++++++-
.../api_fastapi/core_api/routes/ui/deadlines.py | 75 +++++++-
.../src/airflow/jobs/triggerer_job_runner.py | 16 +-
.../src/airflow/ui/openapi-gen/queries/common.ts | 9 +
.../ui/openapi-gen/queries/ensureQueryData.ts | 19 ++-
.../src/airflow/ui/openapi-gen/queries/prefetch.ts | 19 ++-
.../src/airflow/ui/openapi-gen/queries/queries.ts | 19 ++-
.../src/airflow/ui/openapi-gen/queries/suspense.ts | 19 ++-
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 39 ++++-
.../ui/openapi-gen/requests/services.gen.ts | 35 +++-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 43 ++++-
.../airflow/ui/public/i18n/locales/en/common.json | 1 +
.../src/airflow/ui/public/i18n/locales/en/dag.json | 16 ++
.../DataTable/DataTable.expanding.test.tsx | 74 ++++++++
.../ui/src/components/DataTable/DataTable.tsx | 15 +-
.../ui/src/components/DataTable/TableList.tsx | 32 +++-
.../ui/src/components/DataTable/features.ts | 5 +-
.../ui/src/components/renderStructuredLog.test.tsx | 18 +-
.../ui/src/components/renderStructuredLog.tsx | 3 +-
.../airflow/ui/src/pages/Run/CallbackLogs.test.tsx | 85 +++++++++
.../src/airflow/ui/src/pages/Run/CallbackLogs.tsx | 189 +++++++++++++++++++++
.../airflow/ui/src/pages/Run/Callbacks.test.tsx | 139 +++++++++++++++
.../src/airflow/ui/src/pages/Run/Callbacks.tsx | 161 ++++++++++++++++++
.../src/airflow/ui/src/pages/Run/Run.test.tsx | 81 +++++++++
airflow-core/src/airflow/ui/src/pages/Run/Run.tsx | 35 ++--
.../src/airflow/ui/src/queries/useLogs.tsx | 45 ++++-
airflow-core/src/airflow/ui/src/router.tsx | 2 +
.../src/airflow/utils/log/path_log_reader.py | 98 +++++++++++
.../core_api/routes/ui/test_deadlines.py | 125 +++++++++++++-
.../tests/unit/api_fastapi/core_api/test_app.py | 1 +
airflow-core/tests/unit/jobs/test_triggerer_job.py | 31 ++++
.../tests/unit/utils/log/test_path_log_reader.py | 76 +++++++++
.../alibaba/cloud/log/oss_task_handler.py | 2 +-
.../amazon/aws/log/cloudwatch_task_handler.py | 6 +-
.../providers/amazon/aws/log/s3_task_handler.py | 2 +-
.../providers/apache/hdfs/log/hdfs_task_handler.py | 2 +-
.../providers/elasticsearch/log/es_task_handler.py | 5 +-
.../unit/elasticsearch/log/test_es_task_handler.py | 3 +
.../google/cloud/log/cloud_logging_task_handler.py | 5 +-
.../providers/google/cloud/log/gcs_task_handler.py | 4 +-
.../cloud/log/test_cloud_logging_task_handler.py | 5 +
.../microsoft/azure/log/wasb_task_handler.py | 2 +-
.../providers/opensearch/log/os_task_handler.py | 5 +-
.../unit/opensearch/log/test_os_task_handler.py | 3 +
.../logging/src/airflow_shared/logging/remote.py | 16 +-
46 files changed, 1675 insertions(+), 63 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py
index 53b583d1f40..0ea484817ac 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py
@@ -25,6 +25,7 @@ from uuid import UUID
from pydantic import AliasPath, Field, field_validator
from airflow.api_fastapi.core_api.base import BaseModel
+from airflow.utils.state import CallbackState
class DeadlineResponse(BaseModel):
@@ -39,6 +40,12 @@ class DeadlineResponse(BaseModel):
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)
team_name: str | None = Field(validation_alias=AliasPath("dagrun",
"team_name"), default=None)
+ callback_id: UUID
+ callback_type: str = Field(validation_alias=AliasPath("callback", "type"))
+ callback_state: CallbackState | None = Field(
+ validation_alias=AliasPath("callback", "state"), default=None
+ )
+ callback_path: str | None = Field(validation_alias=AliasPath("callback",
"data", "path"), default=None)
class DeadlineCollectionResponse(BaseModel):
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
index 150e27d3beb..f939caedacd 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
+++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
@@ -1329,13 +1329,15 @@ paths:
type: string
description: 'Attributes to order by, multi criteria sort is
supported.
Prefix with `-` for descending order. Supported attributes: `id,
deadline_time,
- created_at, last_updated_at, missed, dag_id, dag_run_id,
alert_name`'
+ created_at, last_updated_at, missed, dag_id, dag_run_id,
alert_name, callback_state,
+ callback_type`'
default:
- deadline_time
title: Order By
description: 'Attributes to order by, multi criteria sort is
supported. Prefix
with `-` for descending order. Supported attributes: `id,
deadline_time,
- created_at, last_updated_at, missed, dag_id, dag_run_id, alert_name`'
+ created_at, last_updated_at, missed, dag_id, dag_run_id, alert_name,
callback_state,
+ callback_type`'
- name: missed
in: query
required: false
@@ -1524,6 +1526,80 @@ paths:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
+ /ui/dags/{dag_id}/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs:
+ get:
+ tags:
+ - Deadlines
+ summary: Get Callback Logs
+ description: Get the execution logs of a deadline callback.
+ operationId: get_callback_logs
+ security:
+ - OAuth2PasswordBearer: []
+ - HTTPBearer: []
+ parameters:
+ - name: dag_id
+ in: path
+ required: true
+ schema:
+ type: string
+ title: Dag Id
+ - name: dag_run_id
+ in: path
+ required: true
+ schema:
+ type: string
+ title: Dag Run Id
+ - name: callback_id
+ in: path
+ required: true
+ schema:
+ type: string
+ format: uuid
+ title: Callback Id
+ - name: accept
+ in: header
+ required: false
+ schema:
+ type: string
+ enum:
+ - application/json
+ - application/x-ndjson
+ - '*/*'
+ default: '*/*'
+ title: Accept
+ responses:
+ '200':
+ description: Successful Response
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/TaskInstancesLogResponse'
+ application/x-ndjson:
+ schema:
+ type: string
+ example: '{"content": "content"}
+
+ {"content": "content"}
+
+ '
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Bad Request
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Not Found
+ '422':
+ description: Validation Error
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPValidationError'
/ui/structure/structure_data:
get:
tags:
@@ -3274,6 +3350,17 @@ components:
- count
title: CalendarTimeRangeResponse
description: Represents a summary of DAG runs for a specific calendar
time range.
+ CallbackState:
+ type: string
+ enum:
+ - scheduled
+ - pending
+ - queued
+ - running
+ - success
+ - failed
+ title: CallbackState
+ description: All possible states of callbacks.
ConfigResponse:
properties:
fallback_page_limit:
@@ -4102,6 +4189,22 @@ components:
- type: string
- type: 'null'
title: Team Name
+ callback_id:
+ type: string
+ format: uuid
+ title: Callback Id
+ callback_type:
+ type: string
+ title: Callback Type
+ callback_state:
+ anyOf:
+ - $ref: '#/components/schemas/CallbackState'
+ - type: 'null'
+ callback_path:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Callback Path
type: object
required:
- id
@@ -4110,6 +4213,8 @@ components:
- created_at
- dag_id
- dag_run_id
+ - callback_id
+ - callback_type
title: DeadlineResponse
description: Deadline serializer for responses.
DurationStats:
@@ -5598,6 +5703,21 @@ components:
- nodes
title: StructureDataResponse
description: Structure Data serializer for responses.
+ StructuredLogMessage:
+ properties:
+ timestamp:
+ type: string
+ format: date-time
+ title: Timestamp
+ event:
+ type: string
+ title: Event
+ additionalProperties: true
+ type: object
+ required:
+ - event
+ title: StructuredLogMessage
+ description: An individual log message.
TaskInletAssetReference:
properties:
dag_id:
@@ -5920,6 +6040,28 @@ components:
- awaiting_input
title: TaskInstanceStateCount
description: TaskInstance serializer for responses.
+ TaskInstancesLogResponse:
+ properties:
+ content:
+ anyOf:
+ - items:
+ $ref: '#/components/schemas/StructuredLogMessage'
+ type: array
+ - items:
+ type: string
+ type: array
+ title: Content
+ continuation_token:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Continuation Token
+ type: object
+ required:
+ - content
+ - continuation_token
+ title: TaskInstancesLogResponse
+ description: Log serializer for responses.
TaskOutletAssetReference:
properties:
dag_id:
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py
index 6a239540a75..c6298e2062e 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py
@@ -18,14 +18,17 @@
from __future__ import annotations
from typing import Annotated
+from uuid import UUID
from fastapi import Depends, HTTPException, status
+from fastapi.responses import StreamingResponse
from sqlalchemy import select
from sqlalchemy.orm import contains_eager, noload
from airflow.api_fastapi.auth.managers.models.resource_details import
DagAccessEntity
from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
from airflow.api_fastapi.common.db.dags import eager_load_teams
+from airflow.api_fastapi.common.headers import HeaderAcceptJsonOrNdjson
from airflow.api_fastapi.common.parameters import (
FilterParam,
QueryLimit,
@@ -38,17 +41,25 @@ from airflow.api_fastapi.common.parameters import (
teams_filter_factory,
)
from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.common.types import Mimetype
+from airflow.api_fastapi.core_api.datamodels.log import
TaskInstancesLogResponse
from airflow.api_fastapi.core_api.datamodels.ui.deadline import (
DeadlineAlertCollectionResponse,
DeadlineCollectionResponse,
)
from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.routes.public.log import (
+ _buffered_ndjson_stream,
+ ndjson_example_response_for_get_log,
+)
from airflow.api_fastapi.core_api.security import ReadableDagRunsFilterDep,
requires_access_dag
+from airflow.models.callback import Callback, CallbackType
from airflow.models.dag_version import DagVersion
from airflow.models.dagrun import DagRun
from airflow.models.deadline import Deadline
from airflow.models.deadline_alert import DeadlineAlert
from airflow.models.serialized_dag import SerializedDagModel
+from airflow.utils.log.path_log_reader import read_logs_at_path,
validate_log_path_component
deadlines_router = AirflowRouter(prefix="/dags/{dag_id}", tags=["Deadlines"])
@@ -87,6 +98,8 @@ def get_deadlines(
"dag_id": DagRun.dag_id,
"dag_run_id": DagRun.run_id,
"alert_name": DeadlineAlert.name,
+ "callback_state": Callback.state,
+ "callback_type": Callback.type,
},
).dynamic_depends(default="deadline_time")
),
@@ -107,13 +120,14 @@ def get_deadlines(
query = (
select(Deadline)
.join(Deadline.dagrun)
+ .join(Deadline.callback)
.outerjoin(Deadline.deadline_alert)
.options(
contains_eager(Deadline.dagrun).options(
noload(DagRun.deadlines), *eager_load_teams(DagRun.dag_model)
),
contains_eager(Deadline.deadline_alert),
- noload(Deadline.callback),
+ contains_eager(Deadline.callback),
)
)
@@ -217,3 +231,62 @@ 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_400_BAD_REQUEST,
status.HTTP_404_NOT_FOUND]),
+ status.HTTP_200_OK: {
+ "description": "Successful Response",
+ "content": ndjson_example_response_for_get_log,
+ },
+ },
+ dependencies=[Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.TASK_LOGS))],
+ response_model=TaskInstancesLogResponse,
+ response_model_exclude_unset=True,
+)
+def get_callback_logs(
+ dag_id: str,
+ dag_run_id: str,
+ callback_id: UUID,
+ accept: HeaderAcceptJsonOrNdjson,
+ session: SessionDep,
+):
+ """Get the execution logs of a deadline callback."""
+ # Both are used as log path components, so reject anything that could
escape the log folder.
+ for param_name, param_value in (("dag_id", dag_id), ("dag_run_id",
dag_run_id)):
+ try:
+ validate_log_path_component(param_value)
+ except ValueError:
+ raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid
characters in {param_name}")
+
+ callback_type = session.scalar(
+ select(Callback.type)
+ .join(Deadline, Deadline.callback_id == Callback.id)
+ .join(Deadline.dagrun)
+ .where(Deadline.callback_id == callback_id, DagRun.dag_id == dag_id,
DagRun.run_id == dag_run_id)
+ .limit(1)
+ )
+ if callback_type is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"Callback `{callback_id}` with a deadline for DagRun
`{dag_run_id}` of Dag `{dag_id}` was not found",
+ )
+
+ # Executor callbacks log to ``executor_callbacks/...`` (see
``ExecuteCallback.make()``) and
+ # triggerer callbacks to ``triggerer_callbacks/...`` (see
``TriggerLoggingFactory``). Read only the
+ # path for this callback's type: CloudWatch reports a missing log stream
as a log line, not as no logs.
+ callback_path = f"{dag_id}/{dag_run_id}/{callback_id}"
+ if callback_type == CallbackType.TRIGGERER:
+ log_stream =
read_logs_at_path(f"triggerer_callbacks/{callback_path}/callback.log")
+ else:
+ log_stream = read_logs_at_path(f"executor_callbacks/{callback_path}")
+
+ if accept == Mimetype.NDJSON:
+ return StreamingResponse(
+ media_type="application/x-ndjson",
+ content=_buffered_ndjson_stream(f"{log.model_dump_json()}\n" for
log in log_stream),
+ )
+
+ return TaskInstancesLogResponse.model_construct(content=list(log_stream),
continuation_token=None)
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index 24a1432a2d8..b29cf66a42e 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -454,7 +454,8 @@ The types of messages that the async Trigger Runner can
send back up to the supe
class TriggerLoggingFactory:
log_path: str
- ti: RuntimeTI = attrs.field(repr=False)
+ # Callback triggers have no task instance; ``upload_to_remote`` accepts
``ti=None``.
+ ti: RuntimeTI | None = attrs.field(default=None, repr=False)
bound_logger: WrappedLogger = attrs.field(init=False, repr=False)
@@ -891,6 +892,19 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
if trigger.assets:
watched_assets = {a.name: a.uri for a in trigger.assets}
+ if callback := getattr(trigger, "callback", None):
+ # Log callback triggers to a file the UI callback logs
endpoint can read.
+ callback_data = callback.data or {}
+ context = (callback_data.get("kwargs") or {}).get("context")
or {}
+ dag_run_data = context.get("dag_run") or {}
+ dag_id = callback_data.get("dag_id") or
dag_run_data.get("dag_id") or "unknown"
+ run_id = dag_run_data.get("dag_run_id") or "unknown"
+ # One directory per callback: with delete_local_logs, remote
handlers delete the
+ # log file's parent directory after upload, which must not
hold other callbacks' logs.
+ self.logger_cache[trigger.id] = TriggerLoggingFactory(
+
log_path=f"triggerer_callbacks/{dag_id}/{run_id}/{callback.id}/callback.log",
+ )
+
return workloads.RunTrigger(
id=trigger.id,
classpath=trigger.classpath,
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
index 02ed3b0f2cd..22ca2c78ed2 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
@@ -1160,6 +1160,15 @@ export const
UseDeadlinesServiceGetDagDeadlineAlertsKeyFn = ({ dagId, limit, off
orderBy?: string[];
versionNumber?: number;
}, queryKey?: Array<unknown>) => [useDeadlinesServiceGetDagDeadlineAlertsKey,
...(queryKey ?? [{ dagId, limit, offset, orderBy, versionNumber }])];
+export type DeadlinesServiceGetCallbackLogsDefaultResponse =
Awaited<ReturnType<typeof DeadlinesService.getCallbackLogs>>;
+export type DeadlinesServiceGetCallbackLogsQueryResult<TData =
DeadlinesServiceGetCallbackLogsDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
+export const useDeadlinesServiceGetCallbackLogsKey =
"DeadlinesServiceGetCallbackLogs";
+export const UseDeadlinesServiceGetCallbackLogsKeyFn = ({ accept, callbackId,
dagId, dagRunId }: {
+ accept?: "application/json" | "*/*" | "application/x-ndjson";
+ callbackId: string;
+ dagId: string;
+ dagRunId: string;
+}, queryKey?: Array<unknown>) => [useDeadlinesServiceGetCallbackLogsKey,
...(queryKey ?? [{ accept, callbackId, dagId, dagRunId }])];
export type StructureServiceStructureDataDefaultResponse =
Awaited<ReturnType<typeof StructureService.structureData>>;
export type StructureServiceStructureDataQueryResult<TData =
StructureServiceStructureDataDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
export const useStructureServiceStructureDataKey =
"StructureServiceStructureData";
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
index 7624712dac2..426338f18ec 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
@@ -2247,7 +2247,7 @@ export const ensureUseDashboardServiceDagStatsData =
(queryClient: QueryClient)
* @param data.dagRunId
* @param data.limit
* @param data.offset
-* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
deadline_time, created_at, last_updated_at, missed, dag_id, dag_run_id,
alert_name`
+* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
deadline_time, created_at, last_updated_at, missed, dag_id, dag_run_id,
alert_name, callback_state, callback_type`
* @param data.missed
* @param data.deadlineTimeGte
* @param data.deadlineTimeGt
@@ -2298,6 +2298,23 @@ export const
ensureUseDeadlinesServiceGetDagDeadlineAlertsData = (queryClient: Q
versionNumber?: number;
}) => queryClient.ensureQueryData({ queryKey:
Common.UseDeadlinesServiceGetDagDeadlineAlertsKeyFn({ dagId, limit, offset,
orderBy, versionNumber }), queryFn: () =>
DeadlinesService.getDagDeadlineAlerts({ dagId, limit, offset, orderBy,
versionNumber }) });
/**
+* Get Callback Logs
+* Get the execution logs of a deadline callback.
+* @param data The data for the request.
+* @param data.dagId
+* @param data.dagRunId
+* @param data.callbackId
+* @param data.accept
+* @returns TaskInstancesLogResponse Successful Response
+* @throws ApiError
+*/
+export const ensureUseDeadlinesServiceGetCallbackLogsData = (queryClient:
QueryClient, { accept, callbackId, dagId, dagRunId }: {
+ accept?: "application/json" | "*/*" | "application/x-ndjson";
+ callbackId: string;
+ dagId: string;
+ dagRunId: string;
+}) => queryClient.ensureQueryData({ queryKey:
Common.UseDeadlinesServiceGetCallbackLogsKeyFn({ accept, callbackId, dagId,
dagRunId }), queryFn: () => DeadlinesService.getCallbackLogs({ accept,
callbackId, dagId, dagRunId }) });
+/**
* Structure Data
* Get Structure Data.
* @param data The data for the request.
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
index a4741bc8602..2a3c519a9c8 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -2247,7 +2247,7 @@ export const prefetchUseDashboardServiceDagStats =
(queryClient: QueryClient) =>
* @param data.dagRunId
* @param data.limit
* @param data.offset
-* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
deadline_time, created_at, last_updated_at, missed, dag_id, dag_run_id,
alert_name`
+* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
deadline_time, created_at, last_updated_at, missed, dag_id, dag_run_id,
alert_name, callback_state, callback_type`
* @param data.missed
* @param data.deadlineTimeGte
* @param data.deadlineTimeGt
@@ -2298,6 +2298,23 @@ export const
prefetchUseDeadlinesServiceGetDagDeadlineAlerts = (queryClient: Que
versionNumber?: number;
}) => queryClient.prefetchQuery({ queryKey:
Common.UseDeadlinesServiceGetDagDeadlineAlertsKeyFn({ dagId, limit, offset,
orderBy, versionNumber }), queryFn: () =>
DeadlinesService.getDagDeadlineAlerts({ dagId, limit, offset, orderBy,
versionNumber }) });
/**
+* Get Callback Logs
+* Get the execution logs of a deadline callback.
+* @param data The data for the request.
+* @param data.dagId
+* @param data.dagRunId
+* @param data.callbackId
+* @param data.accept
+* @returns TaskInstancesLogResponse Successful Response
+* @throws ApiError
+*/
+export const prefetchUseDeadlinesServiceGetCallbackLogs = (queryClient:
QueryClient, { accept, callbackId, dagId, dagRunId }: {
+ accept?: "application/json" | "*/*" | "application/x-ndjson";
+ callbackId: string;
+ dagId: string;
+ dagRunId: string;
+}) => queryClient.prefetchQuery({ queryKey:
Common.UseDeadlinesServiceGetCallbackLogsKeyFn({ accept, callbackId, dagId,
dagRunId }), queryFn: () => DeadlinesService.getCallbackLogs({ accept,
callbackId, dagId, dagRunId }) });
+/**
* Structure Data
* Get Structure Data.
* @param data The data for the request.
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
index d3039078814..03a2a716e0f 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
@@ -2247,7 +2247,7 @@ export const useDashboardServiceDagStats = <TData =
Common.DashboardServiceDagSt
* @param data.dagRunId
* @param data.limit
* @param data.offset
-* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
deadline_time, created_at, last_updated_at, missed, dag_id, dag_run_id,
alert_name`
+* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
deadline_time, created_at, last_updated_at, missed, dag_id, dag_run_id,
alert_name, callback_state, callback_type`
* @param data.missed
* @param data.deadlineTimeGte
* @param data.deadlineTimeGt
@@ -2298,6 +2298,23 @@ export const useDeadlinesServiceGetDagDeadlineAlerts =
<TData = Common.Deadlines
versionNumber?: number;
}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseDeadlinesServiceGetDagDeadlineAlertsKeyFn({ dagId, limit, offset,
orderBy, versionNumber }, queryKey), queryFn: () =>
DeadlinesService.getDagDeadlineAlerts({ dagId, limit, offset, orderBy,
versionNumber }) as TData, ...options });
/**
+* Get Callback Logs
+* Get the execution logs of a deadline callback.
+* @param data The data for the request.
+* @param data.dagId
+* @param data.dagRunId
+* @param data.callbackId
+* @param data.accept
+* @returns TaskInstancesLogResponse Successful Response
+* @throws ApiError
+*/
+export const useDeadlinesServiceGetCallbackLogs = <TData =
Common.DeadlinesServiceGetCallbackLogsDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ accept, callbackId, dagId,
dagRunId }: {
+ accept?: "application/json" | "*/*" | "application/x-ndjson";
+ callbackId: string;
+ dagId: string;
+ dagRunId: string;
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseDeadlinesServiceGetCallbackLogsKeyFn({ accept, callbackId, dagId,
dagRunId }, queryKey), queryFn: () => DeadlinesService.getCallbackLogs({
accept, callbackId, dagId, dagRunId }) as TData, ...options });
+/**
* Structure Data
* Get Structure Data.
* @param data The data for the request.
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
index d9e4d58598b..299870edff3 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
@@ -2247,7 +2247,7 @@ export const useDashboardServiceDagStatsSuspense = <TData
= Common.DashboardServ
* @param data.dagRunId
* @param data.limit
* @param data.offset
-* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
deadline_time, created_at, last_updated_at, missed, dag_id, dag_run_id,
alert_name`
+* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
deadline_time, created_at, last_updated_at, missed, dag_id, dag_run_id,
alert_name, callback_state, callback_type`
* @param data.missed
* @param data.deadlineTimeGte
* @param data.deadlineTimeGt
@@ -2298,6 +2298,23 @@ export const
useDeadlinesServiceGetDagDeadlineAlertsSuspense = <TData = Common.D
versionNumber?: number;
}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseDeadlinesServiceGetDagDeadlineAlertsKeyFn({ dagId, limit, offset,
orderBy, versionNumber }, queryKey), queryFn: () =>
DeadlinesService.getDagDeadlineAlerts({ dagId, limit, offset, orderBy,
versionNumber }) as TData, ...options });
/**
+* Get Callback Logs
+* Get the execution logs of a deadline callback.
+* @param data The data for the request.
+* @param data.dagId
+* @param data.dagRunId
+* @param data.callbackId
+* @param data.accept
+* @returns TaskInstancesLogResponse Successful Response
+* @throws ApiError
+*/
+export const useDeadlinesServiceGetCallbackLogsSuspense = <TData =
Common.DeadlinesServiceGetCallbackLogsDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ accept, callbackId, dagId,
dagRunId }: {
+ accept?: "application/json" | "*/*" | "application/x-ndjson";
+ callbackId: string;
+ dagId: string;
+ dagRunId: string;
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseDeadlinesServiceGetCallbackLogsKeyFn({ accept, callbackId, dagId,
dagRunId }, queryKey), queryFn: () => DeadlinesService.getCallbackLogs({
accept, callbackId, dagId, dagRunId }) as TData, ...options });
+/**
* Structure Data
* Get Structure Data.
* @param data The data for the request.
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index b02c1d69136..b1fb2e97730 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -10308,6 +10308,13 @@ export const $CalendarTimeRangeResponse = {
description: 'Represents a summary of DAG runs for a specific calendar
time range.'
} as const;
+export const $CallbackState = {
+ type: 'string',
+ enum: ['scheduled', 'pending', 'queued', 'running', 'success', 'failed'],
+ title: 'CallbackState',
+ description: 'All possible states of callbacks.'
+} as const;
+
export const $ConfigResponse = {
properties: {
fallback_page_limit: {
@@ -11290,10 +11297,40 @@ export const $DeadlineResponse = {
}
],
title: 'Team Name'
+ },
+ callback_id: {
+ type: 'string',
+ format: 'uuid',
+ title: 'Callback Id'
+ },
+ callback_type: {
+ type: 'string',
+ title: 'Callback Type'
+ },
+ callback_state: {
+ anyOf: [
+ {
+ '$ref': '#/components/schemas/CallbackState'
+ },
+ {
+ type: 'null'
+ }
+ ]
+ },
+ callback_path: {
+ anyOf: [
+ {
+ type: 'string'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Callback Path'
}
},
type: 'object',
- required: ['id', 'deadline_time', 'missed', 'created_at', 'dag_id',
'dag_run_id'],
+ required: ['id', 'deadline_time', 'missed', 'created_at', 'dag_id',
'dag_run_id', 'callback_id', 'callback_type'],
title: 'DeadlineResponse',
description: 'Deadline serializer for responses.'
} as const;
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
index b9d60978b19..2aedd73b91e 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
@@ -3,7 +3,7 @@
import type { CancelablePromise } from './core/CancelablePromise';
import { OpenAPI } from './core/OpenAPI';
import { request as __request } from './core/request';
-import type { GetAssetsData, GetAssetsResponse, GetAssetAliasesData,
GetAssetAliasesResponse, GetAssetAliasData, GetAssetAliasResponse,
GetAssetEventsData, GetAssetEventsResponse, CreateAssetEventData,
CreateAssetEventResponse, MaterializeAssetData, MaterializeAssetResponse,
GetAssetQueuedEventsData, GetAssetQueuedEventsResponse,
DeleteAssetQueuedEventsData, DeleteAssetQueuedEventsResponse, GetAssetData,
GetAssetResponse, GetDagAssetQueuedEventsData, GetDagAssetQueuedEventsResponse,
Dele [...]
+import type { GetAssetsData, GetAssetsResponse, GetAssetAliasesData,
GetAssetAliasesResponse, GetAssetAliasData, GetAssetAliasResponse,
GetAssetEventsData, GetAssetEventsResponse, CreateAssetEventData,
CreateAssetEventResponse, MaterializeAssetData, MaterializeAssetResponse,
GetAssetQueuedEventsData, GetAssetQueuedEventsResponse,
DeleteAssetQueuedEventsData, DeleteAssetQueuedEventsResponse, GetAssetData,
GetAssetResponse, GetDagAssetQueuedEventsData, GetDagAssetQueuedEventsResponse,
Dele [...]
export class AssetService {
/**
@@ -5486,7 +5486,7 @@ export class DeadlinesService {
* @param data.dagRunId
* @param data.limit
* @param data.offset
- * @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
deadline_time, created_at, last_updated_at, missed, dag_id, dag_run_id,
alert_name`
+ * @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
deadline_time, created_at, last_updated_at, missed, dag_id, dag_run_id,
alert_name, callback_state, callback_type`
* @param data.missed
* @param data.deadlineTimeGte
* @param data.deadlineTimeGt
@@ -5563,6 +5563,37 @@ export class DeadlinesService {
});
}
+ /**
+ * Get Callback Logs
+ * Get the execution logs of a deadline callback.
+ * @param data The data for the request.
+ * @param data.dagId
+ * @param data.dagRunId
+ * @param data.callbackId
+ * @param data.accept
+ * @returns TaskInstancesLogResponse Successful Response
+ * @throws ApiError
+ */
+ public static getCallbackLogs(data: GetCallbackLogsData):
CancelablePromise<GetCallbackLogsResponse> {
+ return __request(OpenAPI, {
+ method: 'GET',
+ url:
'/ui/dags/{dag_id}/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs',
+ path: {
+ dag_id: data.dagId,
+ dag_run_id: data.dagRunId,
+ callback_id: data.callbackId
+ },
+ headers: {
+ accept: data.accept
+ },
+ errors: {
+ 400: 'Bad Request',
+ 404: 'Not Found',
+ 422: 'Validation Error'
+ }
+ });
+ }
+
}
export class StructureService {
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index 21b8d6eea05..d55498a3c66 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -2681,6 +2681,11 @@ export type CalendarTimeRangeResponse = {
export type state = 'queued' | 'running' | 'success' | 'failed' | 'planned';
+/**
+ * All possible states of callbacks.
+ */
+export type CallbackState = 'scheduled' | 'pending' | 'queued' | 'running' |
'success' | 'failed';
+
/**
* configuration serializer.
*/
@@ -2942,6 +2947,10 @@ export type DeadlineResponse = {
alert_id?: string | null;
alert_name?: string | null;
team_name?: string | null;
+ callback_id: string;
+ callback_type: string;
+ callback_state?: CallbackState | null;
+ callback_path?: string | null;
};
/**
@@ -5462,7 +5471,7 @@ export type GetDeadlinesData = {
missed?: boolean | null;
offset?: number;
/**
- * Attributes to order by, multi criteria sort is supported. Prefix with
`-` for descending order. Supported attributes: `id, deadline_time, created_at,
last_updated_at, missed, dag_id, dag_run_id, alert_name`
+ * Attributes to order by, multi criteria sort is supported. Prefix with
`-` for descending order. Supported attributes: `id, deadline_time, created_at,
last_updated_at, missed, dag_id, dag_run_id, alert_name, callback_state,
callback_type`
*/
orderBy?: Array<(string)>;
teams?: Array<(string)>;
@@ -5483,6 +5492,15 @@ export type GetDagDeadlineAlertsData = {
export type GetDagDeadlineAlertsResponse = DeadlineAlertCollectionResponse;
+export type GetCallbackLogsData = {
+ accept?: 'application/json' | 'application/x-ndjson' | '*/*';
+ callbackId: string;
+ dagId: string;
+ dagRunId: string;
+};
+
+export type GetCallbackLogsResponse = TaskInstancesLogResponse;
+
export type StructureDataData = {
dagId: string;
depth?: number | null;
@@ -9851,6 +9869,29 @@ export type $OpenApiTs = {
};
};
};
+ '/ui/dags/{dag_id}/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs': {
+ get: {
+ req: GetCallbackLogsData;
+ res: {
+ /**
+ * Successful Response
+ */
+ 200: TaskInstancesLogResponse;
+ /**
+ * Bad Request
+ */
+ 400: HTTPExceptionResponse;
+ /**
+ * Not Found
+ */
+ 404: HTTPExceptionResponse;
+ /**
+ * Validation Error
+ */
+ 422: HTTPValidationError;
+ };
+ };
+ };
'/ui/structure/structure_data': {
get: {
req: StructureDataData;
diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
index e1e03529b32..6d1cdb99622 100644
--- a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
+++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
@@ -414,6 +414,7 @@
"no_status": "No Status",
"none": "No Status",
"open": "Open",
+ "pending": "Pending",
"planned": "Planned",
"queued": "Queued",
"removed": "Removed",
diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/dag.json
b/airflow-core/src/airflow/ui/public/i18n/locales/en/dag.json
index db2aec35b56..b098205824e 100644
--- a/airflow-core/src/airflow/ui/public/i18n/locales/en/dag.json
+++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/dag.json
@@ -36,6 +36,21 @@
"wednesday": "Wed"
}
},
+ "callbacks": {
+ "callback_one": "Callback",
+ "callback_other": "Callbacks",
+ "columns": {
+ "alertName": "Deadline Alert",
+ "callback": "Callback",
+ "deadlineTime": "Deadline Time",
+ "missedBy": "Missed by",
+ "type": "Type"
+ },
+ "types": {
+ "executor": "Sync",
+ "triggerer": "Async"
+ }
+ },
"code": {
"bundleUrl": "Bundle Url",
"noCode": "No Code Found",
@@ -244,6 +259,7 @@
"auditLog": "Audit Log",
"backfills": "Backfills",
"calendar": "Calendar",
+ "callbacks": "Callbacks",
"code": "Code",
"details": "Details",
"logs": "Logs",
diff --git
a/airflow-core/src/airflow/ui/src/components/DataTable/DataTable.expanding.test.tsx
b/airflow-core/src/airflow/ui/src/components/DataTable/DataTable.expanding.test.tsx
new file mode 100644
index 00000000000..7140a0bc3f2
--- /dev/null
+++
b/airflow-core/src/airflow/ui/src/components/DataTable/DataTable.expanding.test.tsx
@@ -0,0 +1,74 @@
+/*!
+ * 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 { Text } from "@chakra-ui/react";
+import type { ColumnDef } from "@tanstack/react-table";
+import "@testing-library/jest-dom/vitest";
+import { act, fireEvent, render, screen } from "@testing-library/react";
+import { describe, expect, it } from "vitest";
+
+import { ChakraWrapper } from "src/utils/ChakraWrapper.tsx";
+
+import { DataTable } from "./DataTable.tsx";
+import type { DataTableFeatures } from "./features.ts";
+
+const columns: Array<ColumnDef<DataTableFeatures, { name: string }>> = [
+ {
+ accessorKey: "name",
+ cell: (info) => info.getValue(),
+ header: "Name",
+ },
+ {
+ cell: ({ row }) => (
+ <button onClick={row.getToggleExpandedHandler()} type="button">
+ {`expand ${row.original.name}`}
+ </button>
+ ),
+ header: "Expand",
+ id: "expand",
+ },
+];
+
+const data = [{ name: "John Doe" }, { name: "Jane Doe" }];
+
+const table = (rows: Array<{ name: string }>) => (
+ <DataTable
+ columns={columns}
+ data={rows}
+ getRowId={(row) => row.name}
+ modelName="task"
+ renderSubComponent={(row) => <Text>{`details of ${row.name}`}</Text>}
+ />
+);
+
+describe("DataTable row expanding", () => {
+ it("keeps an expanded row open, under the same record, when the data
refetches in another order", async () => {
+ const { rerender } = render(table(data), { wrapper: ChakraWrapper });
+
+ expect(screen.queryByTestId("table-sub-row")).not.toBeInTheDocument();
+ fireEvent.click(screen.getByRole("button", { name: "expand Jane Doe" }));
+ expect(screen.getByTestId("table-sub-row")).toHaveTextContent("details of
Jane Doe");
+
+ rerender(table([...data].reverse()));
+ // TanStack schedules its auto-reset of expanded rows in a microtask after
the new data is read.
+ await act(() => Promise.resolve());
+
+ expect(screen.getAllByTestId("table-sub-row")).toHaveLength(1);
+ expect(screen.getByTestId("table-sub-row")).toHaveTextContent("details of
Jane Doe");
+ });
+});
diff --git a/airflow-core/src/airflow/ui/src/components/DataTable/DataTable.tsx
b/airflow-core/src/airflow/ui/src/components/DataTable/DataTable.tsx
index e7783f11a3f..67e59fc0b4d 100644
--- a/airflow-core/src/airflow/ui/src/components/DataTable/DataTable.tsx
+++ b/airflow-core/src/airflow/ui/src/components/DataTable/DataTable.tsx
@@ -60,6 +60,8 @@ type DataTableProps<TData extends RowData> = {
* Anything that changes how already-returned rows are drawn belongs in
`presentationActions`.
*/
readonly filterActions?: ReactNode;
+ /** Stable row ids, so an expanded row stays open across refetches and
re-sorts. Defaults to the row index. */
+ readonly getRowId?: (row: TData) => string;
/** Rendered next to the row count heading, and on its own when the heading
is hidden. */
readonly headingExtra?: ReactNode;
readonly hideRowCountHeading?: boolean;
@@ -100,6 +102,11 @@ type DataTableProps<TData extends RowData> = {
* actions over a selection belong in the `ActionBar` below the table.
*/
readonly primaryActions?: ReactNode;
+ /**
+ * Content shown in a full-width row under each expanded row. Rows open and
close through
+ * `row.getToggleExpandedHandler()` in a cell. See
https://tanstack.com/table/latest/docs/framework/react/examples/sub-components
+ */
+ readonly renderSubComponent?: (row: TData) => ReactNode;
/** Defaults to showing the column visibility menu only when there are many
columns. */
readonly showColumnsMenu?: boolean;
readonly showDisplayToggle?: boolean;
@@ -131,6 +138,7 @@ export const DataTable = <TData extends RowData>({
enableMultiSort = false,
errorMessage,
filterActions,
+ getRowId,
headingExtra,
hideRowCountHeading,
initialState,
@@ -143,6 +151,7 @@ export const DataTable = <TData extends RowData>({
presentationActions,
previousCursor,
primaryActions,
+ renderSubComponent,
showColumnsMenu,
showDisplayToggle,
skeletonCount = 10,
@@ -216,11 +225,15 @@ export const DataTable = <TData extends RowData>({
const rest = Boolean(isLoading) ? createSkeletonMock(display, skeletonCount,
columns) : {};
const table = useTable({
+ // Refetches replace `data`, which would otherwise close every expanded
row.
+ autoResetExpanded: false,
columns,
data,
enableHiding: true,
enableMultiSort,
features: dataTableFeatures,
+ getRowCanExpand: () => renderSubComponent !== undefined,
+ getRowId,
manualPagination: true,
manualSorting: true,
onColumnVisibilityChange: setColumnVisibility,
@@ -329,7 +342,7 @@ export const DataTable = <TData extends RowData>({
{display === "card" && cardDef !== undefined ? (
<CardList cardDef={cardDef} isLoading={isLoading}
noRowsMessage={noRowsNode} rows={rows} />
) : (
- <TableList noRowsMessage={noRowsNode} table={table} />
+ <TableList noRowsMessage={noRowsNode}
renderSubComponent={renderSubComponent} table={table} />
)}
</Box>
{hasOffsetPagination ? (
diff --git a/airflow-core/src/airflow/ui/src/components/DataTable/TableList.tsx
b/airflow-core/src/airflow/ui/src/components/DataTable/TableList.tsx
index f6bda4bd402..9e1a9229e6e 100644
--- a/airflow-core/src/airflow/ui/src/components/DataTable/TableList.tsx
+++ b/airflow-core/src/airflow/ui/src/components/DataTable/TableList.tsx
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-import type { ReactNode } from "react";
+import { Fragment, type ReactNode } from "react";
import { Button, Icon, Table, Text } from "@chakra-ui/react";
import { flexRender, type RowData, type Table as TanStackTable } from
"@tanstack/react-table";
@@ -29,10 +29,15 @@ import type { DataTableFeatures } from "./features";
type TableListProps<TData extends RowData> = {
readonly noRowsMessage?: ReactNode;
+ readonly renderSubComponent?: (row: TData) => ReactNode;
readonly table: TanStackTable<DataTableFeatures, TData>;
};
-export const TableList = <TData extends RowData>({ noRowsMessage, table }:
TableListProps<TData>) => {
+export const TableList = <TData extends RowData>({
+ noRowsMessage,
+ renderSubComponent,
+ table,
+}: TableListProps<TData>) => {
"use no memo"; // remove if https://github.com/TanStack/table/issues/5567 is
resolved
const { t: translate } = useTranslation("components");
const { rows } = table.getRowModel();
@@ -115,13 +120,22 @@ export const TableList = <TData extends RowData>({
noRowsMessage, table }: Table
</Table.Row>
) : (
rows.map((row) => (
- <Table.Row key={row.id}>
- {row.getVisibleCells().map((cell) => (
- <Table.Cell data-testid={`table-cell-${cell.column.id}`}
key={cell.id}>
- {flexRender(cell.column.columnDef.cell, cell.getContext())}
- </Table.Cell>
- ))}
- </Table.Row>
+ <Fragment key={row.id}>
+ <Table.Row>
+ {row.getVisibleCells().map((cell) => (
+ <Table.Cell data-testid={`table-cell-${cell.column.id}`}
key={cell.id}>
+ {flexRender(cell.column.columnDef.cell, cell.getContext())}
+ </Table.Cell>
+ ))}
+ </Table.Row>
+ {renderSubComponent !== undefined && row.getIsExpanded() ? (
+ <Table.Row data-testid="table-sub-row">
+ <Table.Cell colSpan={row.getVisibleCells().length}>
+ {renderSubComponent(row.original)}
+ </Table.Cell>
+ </Table.Row>
+ ) : undefined}
+ </Fragment>
))
)}
</Table.Body>
diff --git a/airflow-core/src/airflow/ui/src/components/DataTable/features.ts
b/airflow-core/src/airflow/ui/src/components/DataTable/features.ts
index b8ba223f340..690effeea53 100644
--- a/airflow-core/src/airflow/ui/src/components/DataTable/features.ts
+++ b/airflow-core/src/airflow/ui/src/components/DataTable/features.ts
@@ -18,15 +18,18 @@
*/
import {
columnVisibilityFeature,
+ rowExpandingFeature,
rowPaginationFeature,
rowSortingFeature,
tableFeatures,
} from "@tanstack/react-table";
// Pagination and sorting are server-side
(`manualPagination`/`manualSorting`), so no row models
-// are registered: rows render in the order and slice the API returns them.
+// are registered: rows render in the order and slice the API returns them.
Expanding only tracks
+// which rows are open, for tables that pass `renderSubComponent`.
export const dataTableFeatures = tableFeatures({
columnVisibilityFeature,
+ rowExpandingFeature,
rowPaginationFeature,
rowSortingFeature,
});
diff --git
a/airflow-core/src/airflow/ui/src/components/renderStructuredLog.test.tsx
b/airflow-core/src/airflow/ui/src/components/renderStructuredLog.test.tsx
index 98ee85cdd15..321b9476394 100644
--- a/airflow-core/src/airflow/ui/src/components/renderStructuredLog.test.tsx
+++ b/airflow-core/src/airflow/ui/src/components/renderStructuredLog.test.tsx
@@ -22,7 +22,12 @@ import { describe, it, expect } from "vitest";
import { Wrapper } from "src/utils/Wrapper";
-import { renderStructuredLog, renderTIContextPreamble, tiContextFields } from
"./renderStructuredLog";
+import {
+ extractTIContext,
+ renderStructuredLog,
+ renderTIContextPreamble,
+ tiContextFields,
+} from "./renderStructuredLog";
const translate = (key: string) => key;
@@ -160,6 +165,17 @@ describe("renderStructuredLog — TI context field
stripping", () => {
});
});
+describe("extractTIContext", () => {
+ it("ignores TI fields logged as null, e.g. by callback triggers", () => {
+ const nullContext = { dag_id: null, event: "trigger starting", task_id:
null, ti_id: null };
+
+ expect(extractTIContext([nullContext as never])).toBeUndefined();
+ expect(extractTIContext([nullContext as never, { dag_id: "my_dag", event:
"x" }])).toEqual({
+ dag_id: "my_dag",
+ });
+ });
+});
+
describe("renderTIContextPreamble", () => {
it("text mode: returns key=value pairs joined by spaces, prefixed with
label", () => {
const result = renderTIContextPreamble(
diff --git a/airflow-core/src/airflow/ui/src/components/renderStructuredLog.tsx
b/airflow-core/src/airflow/ui/src/components/renderStructuredLog.tsx
index 37fdb5d1c15..8744cccee62 100644
--- a/airflow-core/src/airflow/ui/src/components/renderStructuredLog.tsx
+++ b/airflow-core/src/airflow/ui/src/components/renderStructuredLog.tsx
@@ -154,7 +154,8 @@ const extractFromStructuredDatum = (
const ctx: Record<string, unknown> = {};
for (const field of tiContextFields) {
- if (Object.hasOwn(line, field) && line[field] !== undefined) {
+ // Callback triggers log these fields as null since they have no task
instance.
+ if (Object.hasOwn(line, field) && line[field] !== undefined && line[field]
!== null) {
ctx[field] = line[field];
}
}
diff --git a/airflow-core/src/airflow/ui/src/pages/Run/CallbackLogs.test.tsx
b/airflow-core/src/airflow/ui/src/pages/Run/CallbackLogs.test.tsx
new file mode 100644
index 00000000000..ef1c44603c0
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/pages/Run/CallbackLogs.test.tsx
@@ -0,0 +1,85 @@
+/*!
+ * 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 "@testing-library/jest-dom";
+import { render, screen, waitFor } from "@testing-library/react";
+import { MemoryRouter, Route, Routes } from "react-router-dom";
+import { beforeAll, describe, expect, it, vi } from "vitest";
+
+import type * as OpenapiQueries from "openapi/queries";
+
+import { TimezoneProvider } from "src/context/timezone";
+import { BaseWrapper } from "src/utils/Wrapper";
+
+import { CallbackLogs } from "./CallbackLogs";
+
+vi.mock("openapi/queries", async (importOriginal) => {
+ const actual = await importOriginal<typeof OpenapiQueries>();
+
+ return { ...actual, useDeadlinesServiceGetCallbackLogs: vi.fn() };
+});
+
+vi.mock("src/queries/useConfig", () => ({ useConfig: () => undefined }));
+
+const { useDeadlinesServiceGetCallbackLogs } = await import("openapi/queries");
+
+const CALLBACK_ID = "0199a8e0-0000-7000-8000-000000000001";
+
+beforeAll(() => {
+ Object.defineProperty(HTMLElement.prototype, "offsetHeight", { value: 20 });
+ Object.defineProperty(HTMLElement.prototype, "offsetWidth", { value: 800 });
+});
+
+describe("CallbackLogs", () => {
+ it("renders the callback's logs with the task log viewer", async () => {
+ vi.mocked(useDeadlinesServiceGetCallbackLogs).mockReturnValue({
+ data: {
+ content: [
+ { event: "Deadline miss callback started", level: "info", timestamp:
"2026-10-01T00:00:06Z" },
+ { event: "Deadline miss callback finished", level: "info",
timestamp: "2026-10-01T00:00:07Z" },
+ ],
+ },
+ error: null,
+ isLoading: false,
+ } as unknown as ReturnType<typeof useDeadlinesServiceGetCallbackLogs>);
+
+ render(
+ <BaseWrapper>
+ <MemoryRouter initialEntries={["/dags/my_dag/runs/run_1/callbacks"]}>
+ <TimezoneProvider>
+ <Routes>
+ <Route
+ element={<CallbackLogs callbackId={CALLBACK_ID} />}
+ path="/dags/:dagId/runs/:runId/callbacks"
+ />
+ </Routes>
+ </TimezoneProvider>
+ </MemoryRouter>
+ </BaseWrapper>,
+ );
+
+
expect(vi.mocked(useDeadlinesServiceGetCallbackLogs).mock.lastCall?.[0]).toEqual({
+ accept: "application/x-ndjson",
+ callbackId: CALLBACK_ID,
+ dagId: "my_dag",
+ dagRunId: "run_1",
+ });
+ await waitFor(() => expect(screen.getByText(/Deadline miss callback
started/u)).toBeInTheDocument());
+ expect(screen.getByText(/Deadline miss callback
finished/u)).toBeInTheDocument();
+ });
+});
diff --git a/airflow-core/src/airflow/ui/src/pages/Run/CallbackLogs.tsx
b/airflow-core/src/airflow/ui/src/pages/Run/CallbackLogs.tsx
new file mode 100644
index 00000000000..55df7d2dfa1
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/pages/Run/CallbackLogs.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 { useState } from "react";
+
+import { Box, Heading } from "@chakra-ui/react";
+import { useTranslation } from "react-i18next";
+import { useParams, useSearchParams } from "react-router-dom";
+import { useLocalStorage } from "usehooks-ts";
+
+import { Modal } from "src/system-components";
+
+import { TaskLogContent, type TaskLogContentProps } from
"src/pages/TaskInstance/Logs/TaskLogContent";
+import { TaskLogHeader, type TaskLogHeaderProps } from
"src/pages/TaskInstance/Logs/TaskLogHeader";
+import { getDownloadText } from "src/pages/TaskInstance/Logs/utils";
+
+import {
+ LOG_SHOW_LOG_LEVEL_KEY,
+ LOG_SHOW_SOURCE_KEY,
+ LOG_SHOW_TIMESTAMP_KEY,
+ LOG_WRAP_KEY,
+} from "src/constants/localStorage";
+import { SearchParamsKeys } from "src/constants/searchParams";
+import { SHORTCUTS } from "src/context/keyboardShortcuts";
+import { useShortcut } from "src/hooks/useShortcut";
+import { useConfig } from "src/queries/useConfig";
+import { useCallbackLogs } from "src/queries/useLogs";
+
+// Shown under a callback's row in the Callbacks tab.
+export const CallbackLogs = ({ callbackId }: { readonly callbackId: string })
=> {
+ const { dagId = "", runId = "" } = useParams();
+ const [searchParams] = useSearchParams();
+ const { t: translate } = useTranslation("dag");
+
+ const logLevelFilters = searchParams.getAll(SearchParamsKeys.LOG_LEVEL);
+ const sourceFilters = searchParams.getAll(SearchParamsKeys.SOURCE);
+
+ const defaultWrap = Boolean(useConfig("default_wrap"));
+
+ const [wrap, setWrap] = useLocalStorage<boolean>(LOG_WRAP_KEY, defaultWrap);
+ const [showTimestamp, setShowTimestamp] =
useLocalStorage<boolean>(LOG_SHOW_TIMESTAMP_KEY, true);
+ const [showSource, setShowSource] =
useLocalStorage<boolean>(LOG_SHOW_SOURCE_KEY, false);
+ const [showLogLevel, setShowLogLevel] =
useLocalStorage<boolean>(LOG_SHOW_LOG_LEVEL_KEY, true);
+ const [fullscreen, setFullscreen] = useState(false);
+ const [expanded, setExpanded] = useState(false);
+ const [searchQuery, setSearchQuery] = useState("");
+ const [activeSearchIndex, setActiveSearchIndex] = useState(0);
+
+ const { error, fetchedData, isLoading, parsedData } = useCallbackLogs({
+ callbackId,
+ dagId,
+ dagRunId: runId,
+ logLevelFilters,
+ showLogLevel,
+ showSource,
+ showTimestamp,
+ sourceFilters,
+ });
+
+ const getLogString = () =>
+ getDownloadText({
+ fetchedData,
+ logLevelFilters,
+ showLogLevel,
+ showSource,
+ showTimestamp,
+ sourceFilters,
+ translate,
+ })
+ .filter((line) => line !== "")
+ .join("\n");
+
+ const searchMatchIndices = searchQuery
+ ? parsedData.searchableText.flatMap((line, index) =>
+ line.toLowerCase().includes(searchQuery.toLowerCase()) ? [index] : [],
+ )
+ : [];
+
+ const downloadLogs = () => {
+ const element = document.createElement("a");
+
+ element.href = URL.createObjectURL(new Blob([getLogString()], { type:
"text/plain" }));
+ element.download = `logs_${dagId}_${runId}_callback_${callbackId}.txt`;
+ document.body.append(element);
+ element.click();
+ element.remove();
+ };
+
+ const toggleWrap = () => setWrap(!wrap);
+ const toggleTimestamp = () => setShowTimestamp(!showTimestamp);
+ const toggleLogLevel = () => setShowLogLevel(!showLogLevel);
+ const toggleSource = () => setShowSource(!showSource);
+ const toggleFullscreen = () => setFullscreen(!fullscreen);
+ const toggleExpanded = () => setExpanded((act) => !act);
+
+ useShortcut({ ...SHORTCUTS.logs.toggleWrap, callback: toggleWrap });
+ useShortcut({ ...SHORTCUTS.logs.toggleFullscreen, callback: toggleFullscreen
});
+ useShortcut({ ...SHORTCUTS.logs.toggleExpand, callback: toggleExpanded });
+ useShortcut({ ...SHORTCUTS.logs.toggleTimestamp, callback: toggleTimestamp
});
+ useShortcut({ ...SHORTCUTS.logs.toggleLogLevel, callback: toggleLogLevel });
+ useShortcut({ ...SHORTCUTS.logs.toggleSource, callback: toggleSource });
+ useShortcut({ ...SHORTCUTS.logs.downloadLogs, callback: downloadLogs });
+
+ const logHeaderProps: TaskLogHeaderProps = {
+ downloadLogs,
+ expanded,
+ getLogString,
+ onSelectTryNumber: () => undefined,
+ search: {
+ currentMatchIndex: activeSearchIndex,
+ onSearchChange: (query: string) => {
+ setSearchQuery(query);
+ setActiveSearchIndex(0);
+ },
+ onSearchNext: () => setActiveSearchIndex((prev) => (prev + 1) %
Math.max(searchMatchIndices.length, 1)),
+ onSearchPrevious: () =>
+ setActiveSearchIndex(
+ (prev) => (prev - 1 + searchMatchIndices.length) %
Math.max(searchMatchIndices.length, 1),
+ ),
+ searchQuery,
+ totalMatches: searchMatchIndices.length,
+ },
+ showLogLevel,
+ showSource,
+ showTimestamp,
+ sourceOptions: parsedData.sources,
+ toggleExpanded,
+ toggleFullscreen,
+ toggleLogLevel,
+ toggleSource,
+ toggleTimestamp,
+ toggleWrap,
+ wrap,
+ };
+
+ const logContentProps: TaskLogContentProps = {
+ currentMatchLineIndex: searchMatchIndices[activeSearchIndex],
+ error,
+ expanded,
+ isLoading,
+ logError: error,
+ parsedLogs: parsedData.parsedLogs ?? [],
+ searchMatchIndices: searchQuery ? new Set(searchMatchIndices) : undefined,
+ searchQuery: searchQuery || undefined,
+ wrap,
+ };
+
+ return (
+ <Box display="flex" flexDirection="column" maxH="60vh" p={2}>
+ <TaskLogHeader {...logHeaderProps} />
+ <TaskLogContent {...logContentProps} />
+ <Modal
+ bodyProps={{ display: "flex", flexDirection: "column" }}
+ headerProps={{
+ children: (
+ <Box display="flex" flexDirection="column" width="100%">
+ <Heading mb={2} size="xl">
+ {callbackId}
+ </Heading>
+ <TaskLogHeader {...logHeaderProps} isFullscreen />
+ </Box>
+ ),
+ width: "100%",
+ }}
+ onOpenChange={() => setFullscreen(false)}
+ open={fullscreen}
+ scrollBehavior="inside"
+ size="full"
+ >
+ <TaskLogContent {...logContentProps} />
+ </Modal>
+ </Box>
+ );
+};
diff --git a/airflow-core/src/airflow/ui/src/pages/Run/Callbacks.test.tsx
b/airflow-core/src/airflow/ui/src/pages/Run/Callbacks.test.tsx
new file mode 100644
index 00000000000..21db4f5ccb7
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/pages/Run/Callbacks.test.tsx
@@ -0,0 +1,139 @@
+/*!
+ * 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 "@testing-library/jest-dom";
+import { fireEvent, render, screen, waitFor } from "@testing-library/react";
+import { MemoryRouter, Route, Routes } from "react-router-dom";
+import { beforeAll, describe, expect, it, vi } from "vitest";
+
+import type * as OpenapiQueries from "openapi/queries";
+import type { DeadlineResponse } from "openapi/requests/types.gen";
+
+import { TimezoneProvider } from "src/context/timezone";
+import i18n from "src/i18n/config";
+import { BaseWrapper } from "src/utils/Wrapper";
+
+import { Callbacks } from "./Callbacks";
+
+vi.mock("openapi/queries", async (importOriginal) => {
+ const actual = await importOriginal<typeof OpenapiQueries>();
+
+ return {
+ ...actual,
+ useDeadlinesServiceGetCallbackLogs: vi.fn(),
+ useDeadlinesServiceGetDeadlines: vi.fn(),
+ };
+});
+
+vi.mock("src/queries/useConfig", () => ({ useConfig: () => undefined }));
+
+const { useDeadlinesServiceGetCallbackLogs, useDeadlinesServiceGetDeadlines } =
+ await import("openapi/queries");
+
+beforeAll(() => {
+ Object.defineProperty(HTMLElement.prototype, "offsetHeight", { value: 20 });
+ Object.defineProperty(HTMLElement.prototype, "offsetWidth", { value: 800 });
+});
+
+const deadlines: Array<DeadlineResponse> = [
+ {
+ alert_name: "sla_alert",
+ callback_id: "0199a8e0-0000-7000-8000-000000000001",
+ callback_path: "dags.notify.on_miss",
+ callback_state: "success",
+ callback_type: "executor",
+ created_at: "2026-10-01T00:00:00Z",
+ dag_id: "my_dag",
+ dag_run_id: "run_1",
+ deadline_time: "2026-10-01T00:00:05Z",
+ id: "0199a8e0-0000-7000-8000-00000000000a",
+ missed: true,
+ },
+ {
+ alert_name: null,
+ callback_id: "0199a8e0-0000-7000-8000-000000000002",
+ callback_path: "dags.notify.async_on_miss",
+ callback_state: "pending",
+ callback_type: "triggerer",
+ created_at: "2026-10-01T00:00:00Z",
+ dag_id: "my_dag",
+ dag_run_id: "run_1",
+ deadline_time: "2026-10-01T00:00:10Z",
+ id: "0199a8e0-0000-7000-8000-00000000000b",
+ missed: true,
+ },
+];
+
+describe("Callbacks", () => {
+ it("lists the run's callbacks and expands a callback's logs in place", async
() => {
+ vi.mocked(useDeadlinesServiceGetCallbackLogs).mockReturnValue({
+ data: {
+ content: [
+ { event: "Deadline miss callback started", level: "info", timestamp:
"2026-10-01T00:00:06Z" },
+ ],
+ },
+ error: null,
+ isLoading: false,
+ } as unknown as ReturnType<typeof useDeadlinesServiceGetCallbackLogs>);
+ vi.mocked(useDeadlinesServiceGetDeadlines).mockReturnValue({
+ data: { deadlines, total_entries: deadlines.length },
+ error: null,
+ isFetching: false,
+ isLoading: false,
+ } as unknown as ReturnType<typeof useDeadlinesServiceGetDeadlines>);
+
+ render(
+ <BaseWrapper>
+ <MemoryRouter initialEntries={["/dags/my_dag/runs/run_1/callbacks"]}>
+ <TimezoneProvider>
+ <Routes>
+ <Route element={<Callbacks />}
path="/dags/:dagId/runs/:runId/callbacks" />
+ </Routes>
+ </TimezoneProvider>
+ </MemoryRouter>
+ </BaseWrapper>,
+ );
+
+
expect(vi.mocked(useDeadlinesServiceGetDeadlines).mock.lastCall?.[0]).toMatchObject({
+ dagId: "my_dag",
+ dagRunId: "run_1",
+ });
+ expect(screen.getByText("dags.notify.on_miss")).toBeInTheDocument();
+ expect(screen.getByText("sla_alert")).toBeInTheDocument();
+
expect(screen.getByText(i18n.t("common:states.success"))).toBeInTheDocument();
+
expect(screen.getByText(i18n.t("common:states.pending"))).toBeInTheDocument();
+
+ const [firstLogsButton] = screen.getAllByRole("button", { name:
i18n.t("dag:tabs.logs") });
+
+ expect(screen.queryByTestId("table-sub-row")).not.toBeInTheDocument();
+ expect(firstLogsButton).toHaveAttribute("aria-expanded", "false");
+
+ fireEvent.click(firstLogsButton as HTMLElement);
+
+ expect(firstLogsButton).toHaveAttribute("aria-expanded", "true");
+ expect(screen.getAllByTestId("table-sub-row")).toHaveLength(1);
+
expect(vi.mocked(useDeadlinesServiceGetCallbackLogs).mock.lastCall?.[0]).toMatchObject({
+ callbackId: deadlines[0]?.callback_id,
+ });
+ await waitFor(() => expect(screen.getByText(/Deadline miss callback
started/u)).toBeInTheDocument());
+
+ fireEvent.click(firstLogsButton as HTMLElement);
+
+ expect(screen.queryByTestId("table-sub-row")).not.toBeInTheDocument();
+ });
+});
diff --git a/airflow-core/src/airflow/ui/src/pages/Run/Callbacks.tsx
b/airflow-core/src/airflow/ui/src/pages/Run/Callbacks.tsx
new file mode 100644
index 00000000000..340b314d9b7
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/pages/Run/Callbacks.tsx
@@ -0,0 +1,161 @@
+/*!
+ * 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 { Button } from "@chakra-ui/react";
+import type { ColumnDef } from "@tanstack/react-table";
+import dayjs from "dayjs";
+import type { TFunction } from "i18next";
+import { useTranslation } from "react-i18next";
+import { FiChevronDown, FiChevronRight } from "react-icons/fi";
+import { useParams } from "react-router-dom";
+
+import { useDagRunServiceGetDagRun, useDeadlinesServiceGetDeadlines } from
"openapi/queries";
+import type { DeadlineResponse } from "openapi/requests/types.gen";
+
+import { DataTable } from "src/components/DataTable";
+import type { DataTableFeatures } from "src/components/DataTable/features";
+import { useTableURLState } from "src/components/DataTable/useTableUrlState";
+import { ErrorAlert } from "src/components/ErrorAlert";
+import { StateBadge } from "src/components/StateBadge";
+import Time from "src/components/Time";
+import { TruncatedText } from "src/components/TruncatedText";
+
+import { type DurationFormat, useAutoRefresh, useDurationFormat } from
"src/utils";
+
+import { CallbackLogs } from "./CallbackLogs";
+
+const ACTIVE_STATES = new Set(["pending", "queued", "running"]);
+
+const getCallbackColumns = (
+ translate: TFunction,
+ runEndDate: string | null | undefined,
+ renderDuration: DurationFormat["renderDuration"],
+): Array<ColumnDef<DataTableFeatures, DeadlineResponse>> => [
+ {
+ accessorKey: "callback_path",
+ cell: ({ row: { original } }) => <TruncatedText
text={original.callback_path ?? original.callback_id} />,
+ enableSorting: false,
+ header: translate("callbacks.columns.callback"),
+ },
+ {
+ accessorKey: "callback_type",
+ cell: ({ row: { original } }) =>
+ translate(`callbacks.types.${original.callback_type}`, { defaultValue:
original.callback_type }),
+ header: translate("callbacks.columns.type"),
+ },
+ {
+ accessorKey: "callback_state",
+ cell: ({ row: { original } }) => (
+ // "pending" is callback-only, so it borrows the "scheduled" badge style.
+ <StateBadge state={original.callback_state === "pending" ? "scheduled" :
original.callback_state}>
+ {translate(`common:states.${original.callback_state ?? "no_status"}`)}
+ </StateBadge>
+ ),
+ header: translate("common:state"),
+ },
+ {
+ accessorKey: "alert_name",
+ cell: ({ row: { original } }) => original.alert_name ?? "",
+ enableSorting: false,
+ header: translate("callbacks.columns.alertName"),
+ },
+ {
+ accessorKey: "deadline_time",
+ cell: ({ row: { original } }) => <Time datetime={original.deadline_time}
/>,
+ header: translate("callbacks.columns.deadlineTime"),
+ },
+ {
+ accessorKey: "missed_by",
+ // How late the run finished relative to the deadline, like the run
header's deadline badge.
+ cell: ({ row: { original } }) => {
+ if (!original.missed) {
+ return "-";
+ }
+ if (runEndDate === null || runEndDate === undefined) {
+ return translate("deadlineStatus.stillRunning");
+ }
+ const diff = dayjs(runEndDate).diff(dayjs(original.deadline_time));
+
+ return diff < 0 ? "-" : (renderDuration(diff / 1000) ?? "-");
+ },
+ enableSorting: false,
+ header: translate("callbacks.columns.missedBy"),
+ },
+ {
+ accessorKey: "logs",
+ cell: ({ row }) => (
+ <Button
+ aria-expanded={row.getIsExpanded()}
+ color="fg.info"
+ onClick={row.getToggleExpandedHandler()}
+ size="xs"
+ variant="ghost"
+ >
+ {row.getIsExpanded() ? <FiChevronDown /> : <FiChevronRight />}
+ {translate("tabs.logs")}
+ </Button>
+ ),
+ enableSorting: false,
+ header: translate("tabs.logs"),
+ },
+];
+
+export const Callbacks = () => {
+ const { t: translate } = useTranslation(["dag", "common"]);
+ const { dagId = "", runId = "" } = useParams();
+ const { setTableURLState, tableURLState } = useTableURLState();
+ const refetchInterval = useAutoRefresh({ dagId });
+ const { renderDuration } = useDurationFormat();
+ // Already loaded (and refreshed while running) by the Run page.
+ const { data: dagRun } = useDagRunServiceGetDagRun({ dagId, dagRunId: runId
});
+
+ const { pagination, sorting } = tableURLState;
+ const [sort] = sorting;
+ const orderBy = sort ? [`${sort.desc ? "-" : ""}${sort.id}`] : undefined;
+
+ const { data, error, isLoading } = useDeadlinesServiceGetDeadlines(
+ {
+ dagId,
+ dagRunId: runId,
+ limit: pagination.pageSize,
+ offset: pagination.pageIndex * pagination.pageSize,
+ orderBy,
+ },
+ undefined,
+ {
+ refetchInterval: (query) =>
+ query.state.data?.deadlines.some(({ callback_state: state }) =>
ACTIVE_STATES.has(state ?? "")) &&
+ refetchInterval,
+ },
+ );
+
+ return (
+ <DataTable
+ columns={getCallbackColumns(translate, dagRun?.end_date, renderDuration)}
+ data={data?.deadlines ?? []}
+ errorMessage={<ErrorAlert error={error} />}
+ getRowId={(deadline) => deadline.callback_id}
+ initialState={tableURLState}
+ isLoading={isLoading}
+ modelName="dag:callbacks.callback"
+ onStateChange={setTableURLState}
+ renderSubComponent={(deadline) => <CallbackLogs
callbackId={deadline.callback_id} />}
+ total={data?.total_entries}
+ />
+ );
+};
diff --git a/airflow-core/src/airflow/ui/src/pages/Run/Run.test.tsx
b/airflow-core/src/airflow/ui/src/pages/Run/Run.test.tsx
new file mode 100644
index 00000000000..1774e84e378
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/pages/Run/Run.test.tsx
@@ -0,0 +1,81 @@
+/*!
+ * 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 type { PropsWithChildren } from "react";
+
+import "@testing-library/jest-dom";
+import { render, screen } from "@testing-library/react";
+import { MemoryRouter, Route, Routes } from "react-router-dom";
+import { describe, expect, it, vi } from "vitest";
+
+import type * as OpenapiQueries from "openapi/queries";
+
+import { NavTabs, type NavTab } from "src/layouts/Details/NavTabs";
+
+import i18n from "src/i18n/config";
+import { BaseWrapper } from "src/utils/Wrapper";
+
+import { Run } from "./Run";
+
+vi.mock("openapi/queries", async (importOriginal) => {
+ const actual = await importOriginal<typeof OpenapiQueries>();
+
+ return {
+ ...actual,
+ useDagRunServiceGetDagRun: vi.fn(() => ({ data: undefined, error: null,
isLoading: false })),
+ useDeadlinesServiceGetDeadlines: vi.fn(),
+ };
+});
+vi.mock("src/hooks/usePluginTabs", () => ({ usePluginTabs: vi.fn(() => []) }));
+vi.mock("src/layouts/Details/DetailsLayout", () => ({
+ DetailsLayout: ({ tabs }: PropsWithChildren<{ readonly tabs: Array<NavTab>
}>) => <NavTabs tabs={tabs} />,
+}));
+vi.mock("src/utils", async () => {
+ const actual = await vi.importActual("src/utils");
+
+ return { ...actual, useAutoRefresh: vi.fn(() => false), useDocumentTitle:
vi.fn() };
+});
+
+const { useDeadlinesServiceGetDeadlines } = await import("openapi/queries");
+
+describe("Run", () => {
+ it.each([
+ { tabCount: 0, totalEntries: 0 },
+ { tabCount: 1, totalEntries: 1 },
+ ])(
+ "shows the Callbacks tab only when the run has callbacks ($totalEntries)",
+ ({ tabCount, totalEntries }) => {
+ vi.mocked(useDeadlinesServiceGetDeadlines).mockReturnValue({
+ data: { deadlines: [], total_entries: totalEntries },
+ } as unknown as ReturnType<typeof useDeadlinesServiceGetDeadlines>);
+
+ render(
+ <BaseWrapper>
+ <MemoryRouter initialEntries={["/dags/my_dag/runs/run_1"]}>
+ <Routes>
+ <Route element={<Run />} path="/dags/:dagId/runs/:runId" />
+ </Routes>
+ </MemoryRouter>
+ </BaseWrapper>,
+ );
+
+
expect(screen.queryAllByTitle(i18n.t("dag:tabs.callbacks"))).toHaveLength(tabCount);
+
expect(screen.getByTitle(i18n.t("dag:tabs.assetEvents"))).toBeInTheDocument();
+ },
+ );
+});
diff --git a/airflow-core/src/airflow/ui/src/pages/Run/Run.tsx
b/airflow-core/src/airflow/ui/src/pages/Run/Run.tsx
index 62966082512..e462975715a 100644
--- a/airflow-core/src/airflow/ui/src/pages/Run/Run.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/Run/Run.tsx
@@ -18,11 +18,11 @@
*/
import { ReactFlowProvider } from "@xyflow/react";
import { useTranslation } from "react-i18next";
-import { FiCode, FiDatabase } from "react-icons/fi";
+import { FiCode, FiDatabase, FiPhoneCall } from "react-icons/fi";
import { MdDetails, MdOutlineEventNote, MdOutlineTask } from "react-icons/md";
import { useParams } from "react-router-dom";
-import { useDagRunServiceGetDagRun } from "openapi/queries";
+import { useDagRunServiceGetDagRun, useDeadlinesServiceGetDeadlines } from
"openapi/queries";
import { DetailsLayout } from "src/layouts/Details/DetailsLayout";
@@ -40,15 +40,6 @@ export const Run = () => {
// Get external views with dag_run destination
const externalTabs = usePluginTabs("dag_run");
- const tabs = [
- { icon: <MdOutlineTask />, label: translate("tabs.taskInstances"), value:
"" },
- { icon: <FiDatabase />, label: translate("tabs.assetEvents"), value:
"asset_events" },
- { icon: <MdOutlineEventNote />, label: translate("tabs.auditLog"), value:
"events" },
- { icon: <FiCode />, label: translate("tabs.code"), value: "code" },
- { icon: <MdDetails />, label: translate("tabs.details"), value: "details"
},
- ...externalTabs,
- ];
-
const refetchInterval = useAutoRefresh({ dagId });
const {
@@ -66,6 +57,28 @@ export const Run = () => {
},
);
+ // Like Required Actions, the Callbacks tab only shows when the run has
callbacks (one per deadline).
+ const { data: deadlines } = useDeadlinesServiceGetDeadlines(
+ { dagId, dagRunId: runId, limit: 1 },
+ undefined,
+ {
+ refetchInterval: isStatePending(dagRun?.state) && refetchInterval,
+ },
+ );
+ const hasCallbacks = (deadlines?.total_entries ?? 0) > 0;
+
+ const tabs = [
+ { icon: <MdOutlineTask />, label: translate("tabs.taskInstances"), value:
"" },
+ { icon: <FiDatabase />, label: translate("tabs.assetEvents"), value:
"asset_events" },
+ ...(hasCallbacks
+ ? [{ icon: <FiPhoneCall />, label: translate("tabs.callbacks"), value:
"callbacks" }]
+ : []),
+ { icon: <MdOutlineEventNote />, label: translate("tabs.auditLog"), value:
"events" },
+ { icon: <FiCode />, label: translate("tabs.code"), value: "code" },
+ { icon: <MdDetails />, label: translate("tabs.details"), value: "details"
},
+ ...externalTabs,
+ ];
+
return (
<ReactFlowProvider>
<DetailsLayout error={error} isLoading={isLoading} tabs={tabs}>
diff --git a/airflow-core/src/airflow/ui/src/queries/useLogs.tsx
b/airflow-core/src/airflow/ui/src/queries/useLogs.tsx
index 5233610093d..759fa1a636b 100644
--- a/airflow-core/src/airflow/ui/src/queries/useLogs.tsx
+++ b/airflow-core/src/airflow/ui/src/queries/useLogs.tsx
@@ -25,7 +25,7 @@ import type { TFunction } from "i18next";
import { useTranslation } from "react-i18next";
import innerText from "react-innertext";
-import { useTaskInstanceServiceGetLog } from "openapi/queries";
+import { useDeadlinesServiceGetCallbackLogs, useTaskInstanceServiceGetLog }
from "openapi/queries";
import type {
StructuredLogMessage,
TaskInstanceResponse,
@@ -313,6 +313,17 @@ const truncateData = (data: TaskInstancesLogResponse |
undefined, limit?: number
};
};
+// Build a 1:1 searchable text array from parsedLogs so search indices align
+// with the rendered output. Each entry maps to exactly one line.
+const getSearchableText = (parsedLogs: Array<ParsedLogEntry> | undefined):
Array<string> =>
+ (parsedLogs ?? []).map((entry) => {
+ if (typeof entry.element === "string") {
+ return entry.element;
+ }
+
+ return entry.element ? innerText(entry.element) : "";
+ });
+
export const useLogs = (
{
accept = "application/x-ndjson",
@@ -367,15 +378,33 @@ export const useLogs = (
tryNumber,
});
- // Build a 1:1 searchable text array from parsedLogs so search indices align
- // with the rendered output. Each entry maps to exactly one line.
- const searchableText: Array<string> = (parsedData.parsedLogs ??
[]).map((entry) => {
- if (typeof entry.element === "string") {
- return entry.element;
- }
+ const searchableText = getSearchableText(parsedData.parsedLogs);
- return entry.element ? innerText(entry.element) : "";
+ return { parsedData: { ...parsedData, searchableText }, ...rest,
fetchedData: data };
+};
+
+type CallbackLogsProps = { callbackId: string; dagRunId: string } & Omit<
+ Props,
+ "accept" | "limit" | "taskInstance" | "tryNumber"
+>;
+
+export const useCallbackLogs = ({ callbackId, dagId, dagRunId,
...displayOptions }: CallbackLogsProps) => {
+ const { t: translate } = useTranslation();
+
+ const { data, ...rest } = useDeadlinesServiceGetCallbackLogs({
+ accept: "application/x-ndjson",
+ callbackId,
+ dagId,
+ dagRunId,
+ });
+
+ const parsedData = parseLogs({
+ data: parseStreamingLogContent(data),
+ ...displayOptions,
+ translate,
+ tryNumber: 1,
});
+ const searchableText = getSearchableText(parsedData.parsedLogs);
return { parsedData: { ...parsedData, searchableText }, ...rest,
fetchedData: data };
};
diff --git a/airflow-core/src/airflow/ui/src/router.tsx
b/airflow-core/src/airflow/ui/src/router.tsx
index 2b15dcd00aa..13998d22cb8 100644
--- a/airflow-core/src/airflow/ui/src/router.tsx
+++ b/airflow-core/src/airflow/ui/src/router.tsx
@@ -59,6 +59,7 @@ import { Pools } from "src/pages/Pools";
import { Providers } from "src/pages/Providers";
import { Run } from "src/pages/Run";
import { AssetEvents as DagRunAssetEvents } from "src/pages/Run/AssetEvents";
+import { Callbacks as DagRunCallbacks } from "src/pages/Run/Callbacks";
import { Details as DagRunDetails } from "src/pages/Run/Details";
import { Security } from "src/pages/Security";
import { Settings } from "src/pages/Settings";
@@ -250,6 +251,7 @@ export const routerConfig = [
{ element: <Code />, path: "code" },
{ element: <DagRunDetails />, path: "details" },
{ element: <DagRunAssetEvents />, path: "asset_events" },
+ { element: <DagRunCallbacks />, path: "callbacks" },
pluginRoute,
],
element: <Run />,
diff --git a/airflow-core/src/airflow/utils/log/path_log_reader.py
b/airflow-core/src/airflow/utils/log/path_log_reader.py
new file mode 100644
index 00000000000..acaf19f1aa0
--- /dev/null
+++ b/airflow-core/src/airflow/utils/log/path_log_reader.py
@@ -0,0 +1,98 @@
+# 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 logs not tied to a TaskInstance, looked up by their path
relative to the log folder."""
+
+from __future__ import annotations
+
+import os
+import re
+from collections.abc import Generator
+from contextlib import suppress
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from airflow.configuration import conf
+from airflow.utils.log.file_task_handler import (
+ FileTaskHandler,
+ StructuredLogMessage,
+ _get_compatible_log_stream,
+ _interleave_logs,
+)
+
+if TYPE_CHECKING:
+ from airflow._shared.logging.remote import LogSourceInfo, RawLogStream,
StreamingLogResponse
+
+_SAFE_PATH_COMPONENT = re.compile(r"[A-Za-z0-9._:+\-~@]+")
+
+
+def validate_log_path_component(component: str) -> str:
+ """Validate a single log path component, raising ValueError if it could
escape the log folder."""
+ if component in (".", "..") or not
_SAFE_PATH_COMPONENT.fullmatch(component):
+ raise ValueError(f"Invalid log path component: {component!r}")
+ return component
+
+
+def read_logs_at_path(relative_path: str) -> Generator[StructuredLogMessage,
None, None]:
+ """
+ Stream logs not tied to a TaskInstance, trying remote storage first and
then the local filesystem.
+
+ The path is relative to the log folder. Callers build it and must validate
any user-supplied
+ parts with ``validate_log_path_component``. Deadline callbacks are the
current user.
+ """
+ sources: LogSourceInfo = []
+ log_streams: list[RawLogStream] = []
+
+ with suppress(Exception):
+ sources, log_streams = _read_remote_logs(relative_path)
+
+ if not log_streams:
+ local_sources, log_streams = _read_local_logs(relative_path)
+ sources = [*sources, *local_sources]
+
+ if not log_streams:
+ yield StructuredLogMessage(event="No logs found.")
+ return
+
+ yield StructuredLogMessage(event="::group::Log message source details",
sources=sources) # type: ignore[call-arg]
+ yield StructuredLogMessage(event="::endgroup::")
+ yield from _interleave_logs(*log_streams)
+
+
+def _read_remote_logs(relative_path: str) -> StreamingLogResponse:
+ from airflow.logging_config import get_remote_task_log
+
+ remote_io = get_remote_task_log()
+ if remote_io is None:
+ return [], []
+
+ # Prefer .stream(), which yields the remote log lazily, over .read(),
which loads it all into memory.
+ # These logs have no TaskInstance, so ti is None.
+ 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, []
+
+ return sources, [_get_compatible_log_stream(logs)]
+
+
+def _read_local_logs(relative_path: str) -> StreamingLogResponse:
+ """Read with the task handler's symlink-safe local reader."""
+ base_log_folder = os.path.realpath(conf.get("logging", "base_log_folder"))
+ return
FileTaskHandler(base_log_folder)._read_from_local(Path(base_log_folder,
relative_path))
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py
index dd10f3ded99..3943b9cc9b4 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py
@@ -17,16 +17,22 @@
from __future__ import annotations
+import json
+import uuid
from datetime import timedelta
+from unittest import mock
from unittest.mock import Mock
import pytest
from fastapi import HTTPException
-from sqlalchemy import select
+from sqlalchemy import select, update
from airflow._shared.timezones import timezone
from airflow.api_fastapi.core_api.routes.ui.deadlines import
get_dag_deadline_alerts
+from airflow.logging.remote import RemoteLogStreamIO
+from airflow.models.callback import Callback, CallbackType
from airflow.models.dag_version import DagVersion
+from airflow.models.dagrun import DagRun
from airflow.models.deadline import Deadline
from airflow.models.deadline_alert import DeadlineAlert
from airflow.models.serialized_dag import SerializedDagModel
@@ -68,6 +74,12 @@ ALERT_NAME = "SLA Breach Alert"
_CALLBACK_PATH =
"tests.unit.api_fastapi.core_api.routes.ui.test_deadlines._noop_callback"
+_CALLBACK_LOG_FILES = [
+ (CallbackType.EXECUTOR, "executor_callbacks/{dag_id}/{run_id}/{cb_id}"),
+ (CallbackType.TRIGGERER,
"triggerer_callbacks/{dag_id}/{run_id}/{cb_id}/callback.log"),
+]
+
+
async def _noop_callback(**kwargs):
"""No-op callback used to satisfy Deadline creation in tests."""
@@ -261,6 +273,10 @@ class TestGetDagRunDeadlines:
assert deadline1["alert_id"] is None
assert deadline1["dag_id"] == DAG_ID
assert deadline1["dag_run_id"] == RUN_SINGLE
+ assert deadline1["callback_type"] == "triggerer"
+ assert deadline1["callback_state"] == "scheduled"
+ assert deadline1["callback_path"] == _CALLBACK_PATH
+ assert "callback_id" in deadline1
assert "id" in deadline1
assert "created_at" in deadline1
@@ -293,8 +309,8 @@ class TestGetDagRunDeadlines:
@pytest.mark.parametrize(
"order_by",
- ["deadline_time", "id", "created_at", "alert_name"],
- ids=["deadline_time", "id", "created_at", "alert_name"],
+ ["deadline_time", "id", "created_at", "alert_name", "callback_state"],
+ ids=["deadline_time", "id", "created_at", "alert_name",
"callback_state"],
)
def test_should_response_200_order_by(self, test_client, order_by):
url = f"/dags/{DAG_ID}/dagRuns/{RUN_MULTI}/deadlines"
@@ -711,3 +727,106 @@ class TestDeadlineAlertsIntervalSerialization:
assert response.status_code == 200
alerts = {a["name"]: a for a in response.json()["deadline_alerts"]}
assert alerts["dynamic_interval_alert"]["interval"] is None
+
+
+class TestGetCallbackLogs:
+ """Tests for GET
/dags/{dag_id}/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs."""
+
+ @pytest.fixture
+ def missed_callback_id(self, session):
+ deadline =
session.scalar(select(Deadline).join(Deadline.dagrun).where(DagRun.run_id ==
RUN_MISSED))
+ return str(deadline.callback_id)
+
+ @pytest.fixture
+ def set_callback_type(self, session, missed_callback_id):
+ def _set(callback_type: CallbackType) -> None:
+ session.execute(
+ update(Callback)
+ .where(Callback.id == uuid.UUID(missed_callback_id))
+ .values(type=callback_type)
+ )
+ session.commit()
+
+ return _set
+
+ @pytest.mark.parametrize("accept", ["application/json",
"application/x-ndjson"])
+ @pytest.mark.parametrize(
+ ("callback_type", "log_file"), _CALLBACK_LOG_FILES, ids=["executor",
"triggerer"]
+ )
+ def test_returns_logs_from_local_storage(
+ self, test_client, missed_callback_id, set_callback_type, tmp_path,
accept, callback_type, log_file
+ ):
+ set_callback_type(callback_type)
+ log_path = tmp_path / log_file.format(dag_id=DAG_ID,
run_id=RUN_MISSED, cb_id=missed_callback_id)
+ log_path.parent.mkdir(parents=True)
+ log_path.write_text("callback ran\n")
+
+ with conf_vars({("logging", "base_log_folder"): str(tmp_path)}):
+ response = test_client.get(
+
f"/dags/{DAG_ID}/dagRuns/{RUN_MISSED}/callbacks/{missed_callback_id}/logs",
+ headers={"Accept": accept},
+ )
+ assert response.status_code == 200
+ assert response.headers["content-type"].startswith(accept)
+ if accept == "application/x-ndjson":
+ entries = [json.loads(line) for line in response.text.splitlines()
if line]
+ else:
+ entries = response.json()["content"]
+ assert "callback ran" in [entry["event"] for entry in entries]
+
+ @pytest.mark.parametrize(
+ ("callback_type", "log_file"), _CALLBACK_LOG_FILES, ids=["executor",
"triggerer"]
+ )
+ @mock.patch("airflow.logging_config.get_remote_task_log")
+ def test_reads_remote_log_at_the_path_for_the_callback_type(
+ self,
+ mock_get_remote_task_log,
+ test_client,
+ missed_callback_id,
+ set_callback_type,
+ callback_type,
+ log_file,
+ ):
+ # Like CloudWatch, the remote IO returns a lazy stream for any path,
even one that has no log.
+ def lines(relative_path):
+ yield json.dumps({"event": f"remote log at {relative_path}"}) +
"\n"
+
+ remote_io = mock.create_autospec(RemoteLogStreamIO, instance=True)
+ remote_io.stream.side_effect = lambda relative_path, ti:
([relative_path], [lines(relative_path)])
+ mock_get_remote_task_log.return_value = remote_io
+ set_callback_type(callback_type)
+
+ response =
test_client.get(f"/dags/{DAG_ID}/dagRuns/{RUN_MISSED}/callbacks/{missed_callback_id}/logs")
+
+ assert response.status_code == 200
+ expected_path = log_file.format(dag_id=DAG_ID, run_id=RUN_MISSED,
cb_id=missed_callback_id)
+ remote_io.stream.assert_called_once_with(expected_path, None)
+ assert f"remote log at {expected_path}" in [entry["event"] for entry
in response.json()["content"]]
+
+ @pytest.mark.parametrize(
+ ("run_id", "known_callback"),
+ [(RUN_MISSED, False), (RUN_SINGLE, True)],
+ ids=["unknown_callback", "callback_of_other_run"],
+ )
+ def test_should_response_404(self, test_client, missed_callback_id,
run_id, known_callback):
+ callback_id = missed_callback_id if known_callback else uuid.uuid4()
+ response =
test_client.get(f"/dags/{DAG_ID}/dagRuns/{run_id}/callbacks/{callback_id}/logs")
+ assert response.status_code == 404
+
+ # HTTP clients normalize away a literal "..", so only encoded or otherwise
unsafe values reach the check.
+ @pytest.mark.parametrize("bad_run_id", ["%2e%2e", "..%5c..%5cetc",
"run%20id"])
+ def test_path_traversal_in_dag_run_id_returns_400(self, test_client,
missed_callback_id, bad_run_id):
+ response =
test_client.get(f"/dags/{DAG_ID}/dagRuns/{bad_run_id}/callbacks/{missed_callback_id}/logs")
+ assert response.status_code == 400
+
+ def test_should_response_401(self, unauthenticated_test_client):
+ response = unauthenticated_test_client.get(
+
f"/dags/{DAG_ID}/dagRuns/{RUN_MISSED}/callbacks/{uuid.uuid4()}/logs"
+ )
+ assert response.status_code == 401
+
+ def test_should_response_403(self, unauthorized_test_client):
+ response = unauthorized_test_client.get(
+
f"/dags/{DAG_ID}/dagRuns/{RUN_MISSED}/callbacks/{uuid.uuid4()}/logs"
+ )
+ assert response.status_code == 403
diff --git a/airflow-core/tests/unit/api_fastapi/core_api/test_app.py
b/airflow-core/tests/unit/api_fastapi/core_api/test_app.py
index 85f09f3675d..43f28bcba5f 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/test_app.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/test_app.py
@@ -71,6 +71,7 @@ class TestStreamingEndpointSessionScope:
allowed = {
"airflow.api_fastapi.core_api.routes.public.log.get_log",
"airflow.api_fastapi.core_api.routes.public.dag_run.wait_dag_run_until_finished",
+
"airflow.api_fastapi.core_api.routes.ui.deadlines.get_callback_logs",
}
app = create_app()
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py
b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index 212a40aa529..ab6a9e6f4f1 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -876,6 +876,37 @@ def
test_load_triggers_survives_task_missing_from_resolved_dag_version(superviso
assert "Task not found in resolved Dag version; building plain workload"
in caplog
+def
test_create_workload_sets_up_logging_for_callback_trigger(jobless_supervisor,
mocker):
+ """_create_workload() populates logger_cache for callback triggers so
their logs are captured."""
+ callback_id = uuid.uuid4()
+ callback = mocker.Mock()
+ callback.id = callback_id
+ callback.data = {
+ "dag_id": "test_dag",
+ "kwargs": {"context": {"dag_run": {"dag_id": "test_dag", "dag_run_id":
"manual__2024-01-01"}}},
+ }
+
+ trigger = mocker.Mock()
+ trigger.id = 8
+ trigger.classpath = "airflow.triggers.callback.CallbackTrigger"
+ trigger.encrypted_kwargs = ""
+ trigger.task_instance = None
+ trigger.assets = None
+ trigger.callback = callback
+
+ workload = jobless_supervisor._create_workload(
+ trigger=trigger,
+ dag_bag=mocker.Mock(),
+ render_log_fname=mocker.Mock(),
+ session=mocker.Mock(),
+ )
+
+ assert workload is not None
+ factory = jobless_supervisor.logger_cache[trigger.id]
+ assert factory.log_path ==
f"triggerer_callbacks/test_dag/manual__2024-01-01/{callback_id}/callback.log"
+ assert factory.ti is None
+
+
def
test_create_workload_sets_watched_assets_for_asset_only_trigger(jobless_supervisor,
mocker):
"""_create_workload() should populate watched_assets when
trigger.task_instance is None and assets exist."""
asset1 = mocker.Mock(spec=Asset)
diff --git a/airflow-core/tests/unit/utils/log/test_path_log_reader.py
b/airflow-core/tests/unit/utils/log/test_path_log_reader.py
new file mode 100644
index 00000000000..92b169c2b6b
--- /dev/null
+++ b/airflow-core/tests/unit/utils/log/test_path_log_reader.py
@@ -0,0 +1,76 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+
+from airflow.utils.log.path_log_reader import read_logs_at_path,
validate_log_path_component
+
+from tests_common.test_utils.config import conf_vars
+
+PATH = "executor_callbacks/dag1/run1/cb1"
+
+
+class TestValidateLogPathComponent:
+ @pytest.mark.parametrize("component", ["my_dag",
"manual__2024-01-01T00:00:00+00:00", "abc.123~x@y"])
+ def test_valid_components_pass(self, component):
+ assert validate_log_path_component(component) == component
+
+ @pytest.mark.parametrize("component", ["..", ".", "a/b", "a\\b", "", "a
b", "../etc"])
+ def test_unsafe_components_raise(self, component):
+ with pytest.raises(ValueError, match="Invalid log path component"):
+ validate_log_path_component(component)
+
+
+class TestReadLogsAtPath:
+ @pytest.fixture(autouse=True)
+ def log_folder(self, tmp_path):
+ with conf_vars({("logging", "base_log_folder"): str(tmp_path /
"logs")}):
+ yield tmp_path / "logs"
+
+ def test_no_logs_found_yields_message(self):
+ assert [m.event for m in read_logs_at_path(PATH)] == ["No logs found."]
+
+ def test_reads_local_logs(self, log_folder):
+ log_dir = log_folder / "executor_callbacks" / "dag1" / "run1"
+ log_dir.mkdir(parents=True)
+ (log_dir / "cb1").write_text("local line\n")
+
+ assert "local line" in [m.event for m in read_logs_at_path(PATH)]
+
+ def test_symlink_escaping_log_folder_is_skipped(self, log_folder,
tmp_path):
+ secret = tmp_path / "secret"
+ secret.write_text("secret data\n")
+ log_dir = log_folder / "executor_callbacks" / "dag1" / "run1"
+ log_dir.mkdir(parents=True)
+ (log_dir / "cb1").symlink_to(secret)
+
+ assert [m.event for m in read_logs_at_path(PATH)] == ["No logs found."]
+
+ def test_remote_logs_used_when_available(self):
+ def one_stream():
+ yield '{"event": "remote line"}\n'
+
+ with patch(
+ "airflow.utils.log.path_log_reader._read_remote_logs",
+ return_value=(["s3://bucket/log"], [one_stream()]),
+ ):
+ msgs = list(read_logs_at_path(PATH))
+
+ assert any(m.event == "remote line" for m in msgs)
diff --git
a/providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py
b/providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py
index d679f5cf251..fd16d6bb970 100644
---
a/providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py
+++
b/providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py
@@ -112,7 +112,7 @@ class OSSRemoteLogIO(LoggingMixin): # noqa: D101
remote_conn_id,
)
- def read(self, relative_path, ti: RuntimeTI) -> tuple[LogSourceInfo,
LogMessages | None]:
+ def read(self, relative_path, ti: RuntimeTI | None) ->
tuple[LogSourceInfo, LogMessages | None]:
logs: list[str] = []
messages = [relative_path]
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py
b/providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py
index d9d0b3f52f4..192d8d10566 100644
---
a/providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py
+++
b/providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py
@@ -251,13 +251,13 @@ class CloudWatchRemoteLogIO(LoggingMixin): # noqa: D101
if parent.exists():
self.log.warning("Failed to delete local log dir: %s",
parent)
- def read(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
+ def read(self, relative_path: str, ti: RuntimeTI | None) -> LogResponse:
messages, logs = self.stream(relative_path, ti)
str_logs: list[str] = [f"{msg}\n" for group in logs for msg in group]
return messages, str_logs
- def stream(self, relative_path: str, ti: RuntimeTI) ->
StreamingLogResponse:
+ def stream(self, relative_path: str, ti: RuntimeTI | None) ->
StreamingLogResponse:
logs: list[RawLogStream] = []
messages = [
f"Reading remote log from Cloudwatch log_group: {self.log_group}
log_stream: {relative_path}"
@@ -274,7 +274,7 @@ class CloudWatchRemoteLogIO(LoggingMixin): # noqa: D101
return messages, logs
def get_cloudwatch_logs(
- self, stream_name: str, task_instance: RuntimeTI
+ self, stream_name: str, task_instance: RuntimeTI | None
) -> Generator[CloudWatchLogEvent, None, None]:
"""
Return all logs from the given log stream.
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/log/s3_task_handler.py
b/providers/amazon/src/airflow/providers/amazon/aws/log/s3_task_handler.py
index 705b8d53931..32b0b1ee4dd 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/log/s3_task_handler.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/log/s3_task_handler.py
@@ -198,7 +198,7 @@ class S3RemoteLogIO(LoggingMixin): # noqa: D101
return False
return True
- def read(self, relative_path: str, ti: RuntimeTI) -> tuple[LogSourceInfo,
LogMessages | None]:
+ def read(self, relative_path: str, ti: RuntimeTI | None) ->
tuple[LogSourceInfo, LogMessages | None]:
logs: list[str] = []
messages = []
bucket, prefix =
self.hook.parse_s3_url(s3url=os.path.join(self.remote_base, relative_path))
diff --git
a/providers/apache/hdfs/src/airflow/providers/apache/hdfs/log/hdfs_task_handler.py
b/providers/apache/hdfs/src/airflow/providers/apache/hdfs/log/hdfs_task_handler.py
index 5694c62602a..c6ac2060974 100644
---
a/providers/apache/hdfs/src/airflow/providers/apache/hdfs/log/hdfs_task_handler.py
+++
b/providers/apache/hdfs/src/airflow/providers/apache/hdfs/log/hdfs_task_handler.py
@@ -92,7 +92,7 @@ class HdfsRemoteLogIO(LoggingMixin): # noqa: D101
"""Returns WebHDFSHook."""
return WebHDFSHook(webhdfs_conn_id=conf.get("logging",
"REMOTE_LOG_CONN_ID"))
- def read(self, relative_path: str, ti: RuntimeTI) -> tuple[LogSourceInfo,
LogMessages]:
+ def read(self, relative_path: str, ti: RuntimeTI | None) ->
tuple[LogSourceInfo, LogMessages]:
logs = []
messages = []
file_path = os.path.join(self.remote_base, relative_path)
diff --git
a/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py
b/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py
index 24320cece00..2e9e84e569e 100644
---
a/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py
+++
b/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py
@@ -863,7 +863,10 @@ class ElasticsearchRemoteLogIO(LoggingMixin): # noqa: D101
self.log.exception("Unable to insert logs into Elasticsearch.
Reason: %s", str(e))
return False
- def read(self, _relative_path: str, ti: RuntimeTI) -> tuple[LogSourceInfo,
LogMessages]:
+ def read(self, _relative_path: str, ti: RuntimeTI | None) ->
tuple[LogSourceInfo, LogMessages]:
+ if ti is None:
+ # Logs are keyed by task instance, and ``upload`` skips logs
without one.
+ return [], []
log_id = _render_log_id(self.log_id_template, ti, ti.try_number) #
type: ignore[arg-type]
self.log.info("Reading log %s from Elasticsearch", log_id)
offset = 0
diff --git
a/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py
b/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py
index b0b4e39d997..a54af43babb 100644
---
a/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py
+++
b/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py
@@ -930,6 +930,9 @@ class TestElasticsearchRemoteLogIO:
def test_upload_returns_early_when_ti_is_none(self, tmp_json_file):
self.elasticsearch_io.upload(tmp_json_file, ti=None)
+ def test_read_returns_no_logs_when_ti_is_none(self):
+ assert self.elasticsearch_io.read("executor_callbacks/dag/run/cb",
None) == ([], [])
+
class TestFormatErrorDetail:
def test_returns_none_for_empty(self):
diff --git
a/providers/google/src/airflow/providers/google/cloud/log/cloud_logging_task_handler.py
b/providers/google/src/airflow/providers/google/cloud/log/cloud_logging_task_handler.py
index 1a9a2b296c7..6c6e5cf9e24 100644
---
a/providers/google/src/airflow/providers/google/cloud/log/cloud_logging_task_handler.py
+++
b/providers/google/src/airflow/providers/google/cloud/log/cloud_logging_task_handler.py
@@ -248,8 +248,11 @@ class CloudLoggingRemoteLogIO(LoggingMixin):
if parent.exists():
shutil.rmtree(parent, ignore_errors=True)
- def read(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
+ def read(self, relative_path: str, ti: RuntimeTI | None) -> LogResponse:
"""Read logs from Cloud Logging using task instance labels."""
+ if ti is None:
+ # Entries are found by task instance labels, so logs without a
task instance can't be read.
+ return [], None
ti_labels = _task_instance_to_labels(ti)
log_filter = self.prepare_log_filter(ti_labels)
messages, end_of_log, _ = self.read_logs(log_filter,
next_page_token=None, all_pages=True)
diff --git
a/providers/google/src/airflow/providers/google/cloud/log/gcs_task_handler.py
b/providers/google/src/airflow/providers/google/cloud/log/gcs_task_handler.py
index 2a7e2282957..8042621eb02 100644
---
a/providers/google/src/airflow/providers/google/cloud/log/gcs_task_handler.py
+++
b/providers/google/src/airflow/providers/google/cloud/log/gcs_task_handler.py
@@ -209,7 +209,7 @@ class GCSRemoteLogIO(LoggingMixin): # noqa: D101
exc, "resp", {}
).get("status") == "404"
- def read(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
+ def read(self, relative_path: str, ti: RuntimeTI | None) -> LogResponse:
messages, log_streams = self.stream(relative_path, ti)
if not log_streams:
return messages, None
@@ -224,7 +224,7 @@ class GCSRemoteLogIO(LoggingMixin): # noqa: D101
return messages, logs
- def stream(self, relative_path: str, ti: RuntimeTI) ->
StreamingLogResponse:
+ def stream(self, relative_path: str, ti: RuntimeTI | None) ->
StreamingLogResponse:
messages: list[str] = []
log_streams: list[RawLogStream] = []
remote_loc = os.path.join(self.remote_base, relative_path)
diff --git
a/providers/google/tests/unit/google/cloud/log/test_cloud_logging_task_handler.py
b/providers/google/tests/unit/google/cloud/log/test_cloud_logging_task_handler.py
index bdbb9889f77..33750727c1c 100644
---
a/providers/google/tests/unit/google/cloud/log/test_cloud_logging_task_handler.py
+++
b/providers/google/tests/unit/google/cloud/log/test_cloud_logging_task_handler.py
@@ -152,6 +152,11 @@ class TestCloudLoggingRemoteLogIO:
delete_local_copy=True,
)
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.LoggingServiceV2Client")
+ def test_read_returns_no_logs_when_ti_is_none(self, mock_client):
+ assert self.io.read("executor_callbacks/dag/run/cb", None) == ([],
None)
+ mock_client.assert_not_called()
+
@mock.patch(
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
)
diff --git
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/log/wasb_task_handler.py
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/log/wasb_task_handler.py
index 284517cf349..d00c9143bb3 100644
---
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/log/wasb_task_handler.py
+++
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/log/wasb_task_handler.py
@@ -153,7 +153,7 @@ class WasbRemoteLogIO(LoggingMixin): # noqa: D101
return self._build_legacy_url(blob_name)
- def read(self, relative_path, ti: RuntimeTI) -> tuple[LogSourceInfo,
LogMessages | None]:
+ def read(self, relative_path, ti: RuntimeTI | None) ->
tuple[LogSourceInfo, LogMessages | None]:
messages = []
logs = []
# TODO: fix this - "relative path" i.e currently
REMOTE_BASE_LOG_FOLDER should start with "wasb"
diff --git
a/providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py
b/providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py
index 96c27c1d955..e8b1859585b 100644
---
a/providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py
+++
b/providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py
@@ -1032,7 +1032,10 @@ class OpensearchRemoteLogIO(LoggingMixin): # noqa: D101
self.log.exception("Unable to insert logs into OpenSearch. Reason:
%s", str(e))
return False
- def read(self, _relative_path: str, ti: RuntimeTI) -> tuple[LogSourceInfo,
LogMessages]:
+ def read(self, _relative_path: str, ti: RuntimeTI | None) ->
tuple[LogSourceInfo, LogMessages]:
+ if ti is None:
+ # Logs are keyed by task instance, and ``upload`` skips logs
without one.
+ return [], []
log_id = _render_log_id(self.log_id_template, ti, ti.try_number) #
type: ignore[arg-type]
self.log.info("Reading log %s from Opensearch", log_id)
responses = []
diff --git
a/providers/opensearch/tests/unit/opensearch/log/test_os_task_handler.py
b/providers/opensearch/tests/unit/opensearch/log/test_os_task_handler.py
index 71bbecd1580..54c3442e418 100644
--- a/providers/opensearch/tests/unit/opensearch/log/test_os_task_handler.py
+++ b/providers/opensearch/tests/unit/opensearch/log/test_os_task_handler.py
@@ -865,6 +865,9 @@ class TestOpensearchRemoteLogIO:
assert response is None
mock_exception.assert_called_once()
+ def test_read_returns_no_logs_when_ti_is_none(self):
+ assert self.opensearch_io.read("executor_callbacks/dag/run/cb", None)
== ([], [])
+
def test_read_returns_missing_log_message_when_os_read_returns_none(self,
ti):
with patch.object(self.opensearch_io, "_os_read", return_value=None):
log_source_info, log_messages = self.opensearch_io.read("", ti)
diff --git a/shared/logging/src/airflow_shared/logging/remote.py
b/shared/logging/src/airflow_shared/logging/remote.py
index b80a3ceb9cd..1d0250487a0 100644
--- a/shared/logging/src/airflow_shared/logging/remote.py
+++ b/shared/logging/src/airflow_shared/logging/remote.py
@@ -60,8 +60,12 @@ class RemoteLogIO(Protocol):
"""Upload the given log path to the remote storage."""
...
- def read(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
- """Read logs from the given remote log path."""
+ def read(self, relative_path: str, ti: RuntimeTI | None) -> LogResponse:
+ """
+ Read logs from the given remote log path.
+
+ ``ti`` is None for logs not tied to a task instance, such as deadline
callbacks.
+ """
...
@@ -69,8 +73,12 @@ class RemoteLogIO(Protocol):
class RemoteLogStreamIO(RemoteLogIO, Protocol):
"""Interface for remote task loggers with stream-based read support."""
- def stream(self, relative_path: str, ti: RuntimeTI) ->
StreamingLogResponse:
- """Stream-based read interface for reading logs from the given remote
log path."""
+ def stream(self, relative_path: str, ti: RuntimeTI | None) ->
StreamingLogResponse:
+ """
+ Stream-based read interface for reading logs from the given remote log
path.
+
+ ``ti`` is None for logs not tied to a task instance, such as deadline
callbacks.
+ """
...
