diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/common.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/common.py index 2c5832f3246e8..86dd7d74bd23d 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/common.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/common.py @@ -24,6 +24,7 @@ from airflow._shared.timezones import timezone from airflow.api_fastapi.core_api.base import BaseModel +from airflow.api_fastapi.core_api.datamodels.dag_versions import DagVersionResponse from airflow.utils.state import DagRunState from airflow.utils.types import DagRunType @@ -79,6 +80,7 @@ class GridRunsResponse(BaseModel): run_after: datetime state: DagRunState | None run_type: DagRunType + dag_versions: list[DagVersionResponse] = [] has_missed_deadline: bool @computed_field diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/grid.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/grid.py index 46e275a803c3e..70b0c590c3f3c 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/grid.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/grid.py @@ -32,6 +32,7 @@ class LightGridTaskInstanceSummary(BaseModel): child_states: dict[TaskInstanceState | None, int] | None min_start_date: datetime | None max_end_date: datetime | None + dag_version_number: int | None = None class GridTISummaries(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 c3f307afbe73b..e8c97b6cac773 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 @@ -2183,6 +2183,12 @@ components: - type: 'null' run_type: $ref: '#/components/schemas/DagRunType' + dag_versions: + items: + $ref: '#/components/schemas/DagVersionResponse' + type: array + title: Dag Versions + default: [] has_missed_deadline: type: boolean title: Has Missed Deadline @@ -2453,6 +2459,11 @@ components: format: date-time - type: 'null' title: Max End Date + dag_version_number: + anyOf: + - type: integer + - type: 'null' + title: Dag Version Number type: object required: - task_id diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py index 7762656415ecf..ca363c3adab02 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py @@ -23,7 +23,7 @@ import structlog from fastapi import Depends, HTTPException, status from sqlalchemy import exists, select -from sqlalchemy.orm import joinedload +from sqlalchemy.orm import joinedload, load_only, selectinload from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity from airflow.api_fastapi.common.db.common import SessionDep, paginated_select @@ -58,11 +58,13 @@ get_task_group_children_getter, task_group_to_dict_grid, ) +from airflow.models.dag import DagModel from airflow.models.dag_version import DagVersion from airflow.models.dagrun import DagRun from airflow.models.deadline import Deadline from airflow.models.serialized_dag import SerializedDagModel from airflow.models.taskinstance import TaskInstance +from airflow.models.taskinstancehistory import TaskInstanceHistory log = structlog.get_logger(logger_name=__name__) grid_router = AirflowRouter(prefix="/grid", tags=["Grid"]) @@ -282,17 +284,33 @@ def get_grid_runs( .correlate(DagRun) .label("has_missed_deadline") ) - base_query = select( - DagRun.dag_id, - DagRun.run_id, - DagRun.queued_at, - DagRun.start_date, - DagRun.end_date, - DagRun.run_after, - DagRun.state, - DagRun.run_type, - has_missed_deadline, - ).where(DagRun.dag_id == dag_id) + base_query = ( + select(DagRun, has_missed_deadline) + .where(DagRun.dag_id == dag_id) + .options( + load_only( + DagRun.dag_id, + DagRun.run_id, + DagRun.queued_at, + DagRun.start_date, + DagRun.end_date, + DagRun.run_after, + DagRun.state, + DagRun.run_type, + DagRun.bundle_version, + ), + joinedload(DagRun.dag_model).load_only(DagModel._dag_display_property_value), + joinedload(DagRun.created_dag_version).joinedload(DagVersion.bundle), + selectinload(DagRun.task_instances) + .load_only(TaskInstance.dag_version_id) + .joinedload(TaskInstance.dag_version) + .joinedload(DagVersion.bundle), + selectinload(DagRun.task_instances_histories) + .load_only(TaskInstanceHistory.dag_version_id) + .joinedload(TaskInstanceHistory.dag_version) + .joinedload(DagVersion.bundle), + ) + ) # This comparison is to fall back to DAG timetable when no order_by is provided if order_by.value == [order_by.get_primary_key_string()]: @@ -309,8 +327,14 @@ def get_grid_runs( offset=offset, filters=[run_after, run_type, state, triggering_user], limit=limit, + return_total_entries=False, ) - return [GridRunsResponse(**row._mapping) for row in session.execute(dag_runs_select_filter)] + results = session.execute(dag_runs_select_filter).unique().all() + grid_runs = [] + for run, has_missed in results: + run.has_missed_deadline = has_missed + grid_runs.append(GridRunsResponse.model_validate(run, from_attributes=True)) + return grid_runs @grid_router.get( @@ -363,7 +387,9 @@ def get_grid_ti_summaries( TaskInstance.dag_version_id, TaskInstance.start_date, TaskInstance.end_date, + DagVersion.version_number, ) + .outerjoin(DagVersion, TaskInstance.dag_version_id == DagVersion.id) .where(TaskInstance.dag_id == dag_id) .where( TaskInstance.run_id == run_id, @@ -386,6 +412,7 @@ def get_grid_ti_summaries( "state": ti.state, "start_date": ti.start_date, "end_date": ti.end_date, + "dag_version_number": ti.version_number, } ) serdag = _get_serdag( diff --git a/airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py b/airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py index 46ec2c45d98ae..4ee93f4f10e69 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py @@ -72,11 +72,18 @@ def _get_aggs_for_node(detail): max_end_date = max(x["end_date"] for x in detail if x["end_date"]) except ValueError: max_end_date = None + + dag_version_numbers = [ + x.get("dag_version_number") for x in detail if x.get("dag_version_number") is not None + ] + dag_version_number = max(dag_version_numbers) if dag_version_numbers else None + return { "state": agg_state(states), "min_start_date": min_start_date, "max_end_date": max_end_date, "child_states": dict(Counter(states)), + "dag_version_number": dag_version_number, } 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 b9e305f07c983..e54f4579caadf 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 @@ -8143,6 +8143,14 @@ export const $GridRunsResponse = { run_type: { '$ref': '#/components/schemas/DagRunType' }, + dag_versions: { + items: { + '$ref': '#/components/schemas/DagVersionResponse' + }, + type: 'array', + title: 'Dag Versions', + default: [] + }, has_missed_deadline: { type: 'boolean', title: 'Has Missed Deadline' @@ -8258,6 +8266,17 @@ export const $LightGridTaskInstanceSummary = { } ], title: 'Max End Date' + }, + dag_version_number: { + anyOf: [ + { + type: 'integer' + }, + { + type: 'null' + } + ], + title: 'Dag Version Number' } }, type: 'object', 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 aa596d50253fb..79387a6c3f56e 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 @@ -1999,6 +1999,7 @@ export type GridRunsResponse = { run_after: string; state: DagRunState | null; run_type: DagRunType; + dag_versions?: Array; has_missed_deadline: boolean; readonly duration: number; }; @@ -2033,6 +2034,7 @@ export type LightGridTaskInstanceSummary = { } | null; min_start_date: string | null; max_end_date: string | null; + dag_version_number?: number | null; }; /** 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 04380b149a8b2..c38b2a888702d 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 @@ -121,6 +121,15 @@ "graphDirection": { "label": "Graph Direction" }, + "showVersionIndicator": { + "label": "Show Version Indicator", + "options": { + "hideAll": "Hide All", + "showAll": "Show All", + "showBundleVersion": "Show Bundle Version", + "showDagVersion": "Show Dag Version" + } + }, "taskStreamFilter": { "activeFilter": "Active filter", "clearFilter": "Clear Filter", diff --git a/airflow-core/src/airflow/ui/src/constants/localStorage.ts b/airflow-core/src/airflow/ui/src/constants/localStorage.ts index 46482baddd369..72fd47b0909c5 100644 --- a/airflow-core/src/airflow/ui/src/constants/localStorage.ts +++ b/airflow-core/src/airflow/ui/src/constants/localStorage.ts @@ -26,6 +26,7 @@ export const CALENDAR_VIEW_MODE_KEY = "calendar-view-mode"; export const LOG_WRAP_KEY = "log_wrap"; export const LOG_SHOW_TIMESTAMP_KEY = "log_show_timestamp"; export const LOG_SHOW_SOURCE_KEY = "log_show_source"; +export const VERSION_INDICATOR_DISPLAY_MODE_KEY = "version_indicator_display_mode"; // Dag-scoped keys export const dagViewKey = (dagId: string) => `dag_view-${dagId}`; diff --git a/airflow-core/src/airflow/ui/src/constants/showVersionIndicatorOptions.ts b/airflow-core/src/airflow/ui/src/constants/showVersionIndicatorOptions.ts new file mode 100644 index 0000000000000..ceaa3c0d246b5 --- /dev/null +++ b/airflow-core/src/airflow/ui/src/constants/showVersionIndicatorOptions.ts @@ -0,0 +1,46 @@ +/*! + * 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 { createListCollection } from "@chakra-ui/react"; + +export enum VersionIndicatorOptions { + ALL = "all", + BUNDLE_VERSION = "bundle", + DAG_VERSION = "dag", + NONE = "none", +} + +const validOptions = new Set(Object.values(VersionIndicatorOptions)); + +export const isVersionIndicatorOption = (value: unknown): value is VersionIndicatorOptions => + typeof value === "string" && validOptions.has(value); + +export const showVersionIndicatorOptions = createListCollection({ + items: [ + { label: "dag:panel.showVersionIndicator.options.showAll", value: VersionIndicatorOptions.ALL }, + { + label: "dag:panel.showVersionIndicator.options.showBundleVersion", + value: VersionIndicatorOptions.BUNDLE_VERSION, + }, + { + label: "dag:panel.showVersionIndicator.options.showDagVersion", + value: VersionIndicatorOptions.DAG_VERSION, + }, + { label: "dag:panel.showVersionIndicator.options.hideAll", value: VersionIndicatorOptions.NONE }, + ], +}); diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/DetailsLayout.tsx b/airflow-core/src/airflow/ui/src/layouts/Details/DetailsLayout.tsx index 09d116a8370e7..71635ea6dbf1d 100644 --- a/airflow-core/src/airflow/ui/src/layouts/Details/DetailsLayout.tsx +++ b/airflow-core/src/airflow/ui/src/layouts/Details/DetailsLayout.tsx @@ -1,3 +1,5 @@ +/* eslint-disable max-lines */ + /*! * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file @@ -50,6 +52,7 @@ import { showGanttKey, triggeringUserFilterKey, } from "src/constants/localStorage"; +import { VersionIndicatorOptions } from "src/constants/showVersionIndicatorOptions"; import { HoverProvider } from "src/context/hover"; import { OpenGroupsProvider } from "src/context/openGroups"; @@ -88,6 +91,11 @@ export const DetailsLayout = ({ children, error, isLoading, tabs }: Props) => { ); const [showGantt, setShowGantt] = useLocalStorage(showGanttKey(dagId), false); + // Global setting: applies to all Dags (intentionally not scoped to dagId) + const [showVersionIndicatorMode, setShowVersionIndicatorMode] = useLocalStorage( + `version_indicator_display_mode`, + VersionIndicatorOptions.ALL, + ); const { fitView, getZoom } = useReactFlow(); const { data: warningData } = useDagWarningServiceListDagWarnings({ dagId }); const { onClose, onOpen, open } = useDisclosure(); @@ -161,8 +169,10 @@ export const DetailsLayout = ({ children, error, isLoading, tabs }: Props) => { setLimit={setLimit} setRunTypeFilter={setRunTypeFilter} setShowGantt={setShowGantt} + setShowVersionIndicatorMode={setShowVersionIndicatorMode} setTriggeringUserFilter={setTriggeringUserFilter} showGantt={showGantt} + showVersionIndicatorMode={showVersionIndicatorMode} triggeringUserFilter={triggeringUserFilter} /> {dagView === "graph" ? ( @@ -174,6 +184,7 @@ export const DetailsLayout = ({ children, error, isLoading, tabs }: Props) => { limit={limit} runType={runTypeFilter} showGantt={Boolean(runId) && showGantt} + showVersionIndicatorMode={showVersionIndicatorMode} triggeringUser={triggeringUserFilter} /> {showGantt ? ( diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/Grid/Bar.tsx b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/Bar.tsx index 6b52198f51025..c2bb3ed6d04b3 100644 --- a/airflow-core/src/airflow/ui/src/layouts/Details/Grid/Bar.tsx +++ b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/Bar.tsx @@ -19,21 +19,27 @@ import { Flex, Box } from "@chakra-ui/react"; import { useParams, useSearchParams } from "react-router-dom"; -import type { GridRunsResponse } from "openapi/requests"; import { RunTypeIcon } from "src/components/RunTypeIcon"; +import { VersionIndicatorOptions } from "src/constants/showVersionIndicatorOptions"; import { useHover } from "src/context/hover"; import { GridButton } from "./GridButton"; - -const BAR_HEIGHT = 100; +import { BundleVersionIndicator, DagVersionIndicator } from "./VersionIndicator"; +import { BAR_HEIGHT } from "./constants"; +import { + getBundleVersion, + getMaxVersionNumber, + type GridRunWithVersionFlags, +} from "./useGridRunsWithVersionFlags"; type Props = { readonly max: number; readonly onClick?: () => void; - readonly run: GridRunsResponse; + readonly run: GridRunWithVersionFlags; + readonly showVersionIndicatorMode?: VersionIndicatorOptions; }; -export const Bar = ({ max, onClick, run }: Props) => { +export const Bar = ({ max, onClick, run, showVersionIndicatorMode }: Props) => { const { dagId = "", runId } = useParams(); const [searchParams] = useSearchParams(); const { hoveredRunId, setHoveredRunId } = useHover(); @@ -53,6 +59,17 @@ export const Bar = ({ max, onClick, run }: Props) => { position="relative" transition="background-color 0.2s" > + {run.isBundleVersionChange && + (showVersionIndicatorMode === VersionIndicatorOptions.BUNDLE_VERSION || + showVersionIndicatorMode === VersionIndicatorOptions.ALL) ? ( + + ) : undefined} + {run.isDagVersionChange && + (showVersionIndicatorMode === VersionIndicatorOptions.DAG_VERSION || + showVersionIndicatorMode === VersionIndicatorOptions.ALL) ? ( + + ) : undefined} + { +export const Grid = ({ + dagRunState, + limit, + runType, + showGantt, + showVersionIndicatorMode, + triggeringUser, +}: Props) => { const { t: translate } = useTranslation("dag"); const gridRef = useRef(null); const scrollContainerRef = useRef(null); @@ -107,7 +117,13 @@ export const Grid = ({ dagRunState, limit, runType, showGantt, triggeringUser }: .filter((duration: number | null): duration is number => duration !== null), ); - const { flatNodes } = flattenNodes(dagStructure, openGroupIds); + // calculate version change flags + const runsWithVersionFlags = useGridRunsWithVersionFlags({ + gridRuns, + showVersionIndicatorMode, + }); + + const { flatNodes } = useMemo(() => flattenNodes(dagStructure, openGroupIds), [dagStructure, openGroupIds]); const { setMode } = useNavigation({ onToggleGroup: toggleGroupId, @@ -166,8 +182,14 @@ export const Grid = ({ dagRunState, limit, runType, showGantt, triggeringUser }: - {gridRuns?.map((dr: GridRunsResponse) => ( - + {runsWithVersionFlags?.map((dr) => ( + ))} {selectedIsVisible === undefined || !selectedIsVisible ? undefined : ( @@ -202,6 +224,7 @@ export const Grid = ({ dagRunState, limit, runType, showGantt, triggeringUser }: nodes={flatNodes} onCellClick={handleCellClick} run={dr} + showVersionIndicatorMode={showVersionIndicatorMode} virtualItems={virtualItems} /> ))} diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/Grid/TaskInstancesColumn.tsx b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/TaskInstancesColumn.tsx index dcbf7cc212053..7727748fba7bd 100644 --- a/airflow-core/src/airflow/ui/src/layouts/Details/Grid/TaskInstancesColumn.tsx +++ b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/TaskInstancesColumn.tsx @@ -22,22 +22,31 @@ import { useParams } from "react-router-dom"; import type { GridRunsResponse } from "openapi/requests"; import type { LightGridTaskInstanceSummary } from "openapi/requests/types.gen"; +import { VersionIndicatorOptions } from "src/constants/showVersionIndicatorOptions"; import { useHover } from "src/context/hover"; import { useGridTiSummaries } from "src/queries/useGridTISummaries.ts"; import { GridTI } from "./GridTI"; +import { DagVersionIndicator } from "./VersionIndicator"; import type { GridTask } from "./utils"; type Props = { readonly nodes: Array; readonly onCellClick?: () => void; readonly run: GridRunsResponse; + readonly showVersionIndicatorMode?: VersionIndicatorOptions; readonly virtualItems?: Array; }; const ROW_HEIGHT = 20; -export const TaskInstancesColumn = ({ nodes, onCellClick, run, virtualItems }: Props) => { +export const TaskInstancesColumn = ({ + nodes, + onCellClick, + run, + showVersionIndicatorMode, + virtualItems, +}: Props) => { const { dagId = "", runId } = useParams(); const isSelected = runId === run.run_id; const { data: gridTISummaries } = useGridTiSummaries({ @@ -52,12 +61,18 @@ export const TaskInstancesColumn = ({ nodes, onCellClick, run, virtualItems }: P virtualItems ?? nodes.map((_, index) => ({ index, size: ROW_HEIGHT, start: index * ROW_HEIGHT })); const taskInstances = gridTISummaries?.task_instances ?? []; + const taskInstanceMap = new Map(); for (const ti of taskInstances) { taskInstanceMap.set(ti.task_id, ti); } + const versionNumbers = new Set( + taskInstances.map((ti) => ti.dag_version_number).filter((vn) => vn !== null && vn !== undefined), + ); + const hasMixedVersions = versionNumbers.size > 1; + const isHovered = hoveredRunId === run.run_id; const handleMouseEnter = () => setHoveredRunId(run.run_id); @@ -72,7 +87,7 @@ export const TaskInstancesColumn = ({ nodes, onCellClick, run, virtualItems }: P transition="background-color 0.2s" width="18px" > - {itemsToRender.map((virtualItem) => { + {itemsToRender.map((virtualItem, idx) => { const node = nodes[virtualItem.index]; if (!node) { @@ -95,6 +110,23 @@ export const TaskInstancesColumn = ({ nodes, onCellClick, run, virtualItems }: P ); } + let hasVersionChangeFlag = false; + + if ( + hasMixedVersions && + (showVersionIndicatorMode === VersionIndicatorOptions.DAG_VERSION || + showVersionIndicatorMode === VersionIndicatorOptions.ALL) && + idx > 0 + ) { + const prevVirtualItem = itemsToRender[idx - 1]; + const prevNode = prevVirtualItem ? nodes[prevVirtualItem.index] : undefined; + const prevTaskInstance = prevNode ? taskInstanceMap.get(prevNode.id) : undefined; + + hasVersionChangeFlag = Boolean( + prevTaskInstance && prevTaskInstance.dag_version_number !== taskInstance.dag_version_number, + ); + } + return ( + {hasVersionChangeFlag && ( + + )} { + const { t: translate } = useTranslation("components"); + + return ( + + + + + + ); +}; + +const CONTAINER_STYLES = { + horizontal: { + height: 0.5, + left: "50%", + top: 0, + transform: "translate(-50%, -50%)", + width: 4.5, + }, + vertical: { + height: DAG_VERSION_INDICATOR_HEIGHT, + left: -1.25, + top: -1.5, + width: 0.5, + }, +} as const; + +const CIRCLE_STYLES = { + horizontal: { + height: 1.5, + left: "50%", + top: "50%", + transform: "translate(-50%, -50%)", + width: 1.5, + }, + vertical: { + height: 1.5, + left: "50%", + top: -1, + transform: "translateX(-50%)", + width: 1.5, + }, +} as const; + +type DagVersionIndicatorProps = { + readonly dagVersionNumber: number | undefined; + readonly orientation?: "horizontal" | "vertical"; +}; + +export const DagVersionIndicator = ({ + dagVersionNumber, + orientation = "vertical", +}: DagVersionIndicatorProps) => { + const isVertical = orientation === "vertical"; + const currentContainerStyle = CONTAINER_STYLES[orientation]; + const currentCircleStyle = CIRCLE_STYLES[orientation]; + + return ( + + + + + + + + ); +}; diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/Grid/constants.ts b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/constants.ts index 7286818def786..99bd035896de0 100644 --- a/airflow-core/src/airflow/ui/src/layouts/Details/Grid/constants.ts +++ b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/constants.ts @@ -20,7 +20,7 @@ // Grid layout constants - shared between Grid and Gantt for alignment export const ROW_HEIGHT = 20; export const GRID_OUTER_PADDING_PX = 64; // pt={16} = 16 * 4 = 64px -export const GRID_HEADER_PADDING_PX = 8; // pt={2} = 2 * 4 = 8px +export const GRID_HEADER_PADDING_PX = 16; // pt={4} = 4 * 4 = 16px export const GRID_HEADER_HEIGHT_PX = 100; // height="100px" for duration bars // Gantt chart's x-axis height (time labels at top of chart) @@ -30,3 +30,11 @@ export const GANTT_AXIS_HEIGHT_PX = 36; // minus the Gantt axis height since the chart includes its own top axis export const GRID_BODY_OFFSET_PX = GRID_OUTER_PADDING_PX + GRID_HEADER_PADDING_PX + GRID_HEADER_HEIGHT_PX - GANTT_AXIS_HEIGHT_PX; + +// Version indicator constants +export const BAR_HEIGHT = GRID_HEADER_HEIGHT_PX; // Duration bar height matches grid header +export const BUNDLE_VERSION_INDICATOR_TOP = 93; // Position from top for bundle version icon +export const BUNDLE_VERSION_INDICATOR_LEFT = -2; // Position from left for bundle version icon +export const BUNDLE_VERSION_ICON_SIZE = 15; // Size of the git commit icon +export const DAG_VERSION_INDICATOR_HEIGHT = 104; // Height of the vertical line indicator +export const VERSION_INDICATOR_Z_INDEX = 1; // Z-index for version indicators diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/Grid/useGridRunsWithVersionFlags.ts b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/useGridRunsWithVersionFlags.ts new file mode 100644 index 0000000000000..e60012e58ba80 --- /dev/null +++ b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/useGridRunsWithVersionFlags.ts @@ -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 { useMemo } from "react"; + +import type { GridRunsResponse } from "openapi/requests"; +import { VersionIndicatorOptions } from "src/constants/showVersionIndicatorOptions"; + +export type GridRunWithVersionFlags = { + isBundleVersionChange: boolean; + isDagVersionChange: boolean; +} & GridRunsResponse; + +type UseGridRunsWithVersionFlagsParams = { + gridRuns: Array | undefined; + showVersionIndicatorMode?: VersionIndicatorOptions; +}; + +export const getMaxVersionNumber = (run: GridRunsResponse): number | undefined => + run.dag_versions?.at(-1)?.version_number; + +export const getBundleVersion = (run: GridRunsResponse): string | null | undefined => + run.dag_versions?.at(-1)?.bundle_version; + +// Hook to calculate version change flags for grid runs. +export const useGridRunsWithVersionFlags = ({ + gridRuns, + showVersionIndicatorMode, +}: UseGridRunsWithVersionFlagsParams): Array | undefined => { + const isVersionIndicatorEnabled = showVersionIndicatorMode !== VersionIndicatorOptions.NONE; + + return useMemo(() => { + if (!gridRuns) { + return undefined; + } + + if (!isVersionIndicatorEnabled) { + return gridRuns.map((run) => ({ ...run, isBundleVersionChange: false, isDagVersionChange: false })); + } + + return gridRuns.map((run, index) => { + const nextRun = gridRuns[index + 1]; + + const currentBundleVersion = getBundleVersion(run); + const nextBundleVersion = nextRun ? getBundleVersion(nextRun) : undefined; + const isBundleVersionChange = + currentBundleVersion !== undefined && + nextBundleVersion !== undefined && + currentBundleVersion !== nextBundleVersion; + + const currentVersion = getMaxVersionNumber(run); + const nextVersion = nextRun ? getMaxVersionNumber(nextRun) : undefined; + const isDagVersionChange = + currentVersion !== undefined && nextVersion !== undefined && currentVersion !== nextVersion; + + return { ...run, isBundleVersionChange, isDagVersionChange }; + }); + }, [gridRuns, isVersionIndicatorEnabled]); +}; diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/PanelButtons.tsx b/airflow-core/src/airflow/ui/src/layouts/Details/PanelButtons.tsx index cdcaa90f3fc1e..3d6a335f9d326 100644 --- a/airflow-core/src/airflow/ui/src/layouts/Details/PanelButtons.tsx +++ b/airflow-core/src/airflow/ui/src/layouts/Details/PanelButtons.tsx @@ -51,12 +51,14 @@ import { Tooltip } from "src/components/ui"; import { type ButtonGroupOption, ButtonGroupToggle } from "src/components/ui/ButtonGroupToggle"; import { Checkbox } from "src/components/ui/Checkbox"; import { dependenciesKey, directionKey } from "src/constants/localStorage"; +import type { VersionIndicatorOptions } from "src/constants/showVersionIndicatorOptions"; import { dagRunTypeOptions, dagRunStateOptions } from "src/constants/stateOptions"; import { useContainerWidth } from "src/utils/useContainerWidth"; import { DagRunSelect } from "./DagRunSelect"; import { TaskStreamFilter } from "./TaskStreamFilter"; import { ToggleGroups } from "./ToggleGroups"; +import { VersionIndicatorSelect } from "./VersionIndicatorSelect"; type Props = { readonly dagRunStateFilter: DagRunState | undefined; @@ -69,8 +71,10 @@ type Props = { readonly setLimit: React.Dispatch>; readonly setRunTypeFilter: React.Dispatch>; readonly setShowGantt: React.Dispatch>; + readonly setShowVersionIndicatorMode: React.Dispatch>; readonly setTriggeringUserFilter: React.Dispatch>; readonly showGantt: boolean; + readonly showVersionIndicatorMode: VersionIndicatorOptions; readonly triggeringUserFilter: string | undefined; }; @@ -118,8 +122,10 @@ export const PanelButtons = ({ setLimit, setRunTypeFilter, setShowGantt, + setShowVersionIndicatorMode, setTriggeringUserFilter, showGantt, + showVersionIndicatorMode, triggeringUserFilter, }: Props) => { const { t: translate } = useTranslation(["components", "dag"]); @@ -470,6 +476,12 @@ export const PanelButtons = ({ ) : undefined} + + + )} diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/VersionIndicatorSelect.tsx b/airflow-core/src/airflow/ui/src/layouts/Details/VersionIndicatorSelect.tsx new file mode 100644 index 0000000000000..3d4cf9a51c70a --- /dev/null +++ b/airflow-core/src/airflow/ui/src/layouts/Details/VersionIndicatorSelect.tsx @@ -0,0 +1,94 @@ +/*! + * 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 { Circle, Flex, Select, type SelectValueChangeDetails } from "@chakra-ui/react"; +import { useTranslation } from "react-i18next"; +import { FiGitCommit } from "react-icons/fi"; + +import { + isVersionIndicatorOption, + showVersionIndicatorOptions, + VersionIndicatorOptions, +} from "src/constants/showVersionIndicatorOptions"; + +type VersionIndicatorSelectProps = { + readonly onChange: (value: VersionIndicatorOptions) => void; + readonly value: VersionIndicatorOptions; +}; + +export const VersionIndicatorSelect = ({ onChange, value }: VersionIndicatorSelectProps) => { + const { t: translate } = useTranslation(["components", "dag"]); + + const handleChange = (event: SelectValueChangeDetails<{ label: string; value: Array }>) => { + const [selectedDisplayMode] = event.value; + + if (isVersionIndicatorOption(selectedDisplayMode)) { + onChange(selectedDisplayMode); + } + }; + + return ( + + {translate("dag:panel.showVersionIndicator.label")} + + + + + {(value === VersionIndicatorOptions.BUNDLE_VERSION || + value === VersionIndicatorOptions.ALL) && ( + + )} + {(value === VersionIndicatorOptions.DAG_VERSION || value === VersionIndicatorOptions.ALL) && ( + + )} + {translate(showVersionIndicatorOptions.items.find((item) => item.value === value)?.label ?? "")} + + + + + + + + + + {showVersionIndicatorOptions.items.map((option) => ( + + + {(option.value === VersionIndicatorOptions.BUNDLE_VERSION || + option.value === VersionIndicatorOptions.ALL) && ( + + )} + {(option.value === VersionIndicatorOptions.DAG_VERSION || + option.value === VersionIndicatorOptions.ALL) && ( + + )} + {translate(option.label)} + + + ))} + + + + ); +}; diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py index ed4d6fb2e0e43..c9642f7c4922e 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py @@ -61,6 +61,15 @@ GRID_RUN_1 = { "dag_id": "test_dag", + "dag_versions": [ + { + "version_number": 1, + "dag_id": "test_dag", + "bundle_name": "dag_maker", + "created_at": "2024-12-31T00:00:00Z", + "dag_display_name": "test_dag", + } + ], "duration": 283996800.0, "end_date": "2024-12-31T00:00:00Z", "has_missed_deadline": False, @@ -73,6 +82,15 @@ GRID_RUN_2 = { "dag_id": "test_dag", + "dag_versions": [ + { + "version_number": 1, + "dag_id": "test_dag", + "bundle_name": "dag_maker", + "created_at": "2024-12-31T00:00:00Z", + "dag_display_name": "test_dag", + } + ], "duration": 283996800.0, "end_date": "2024-12-31T00:00:00Z", "has_missed_deadline": False, @@ -83,6 +101,18 @@ "state": "failed", } + +def _strip_dag_version_ids(data): + """Strip dynamic `id` fields from dag_versions for deterministic comparison.""" + if isinstance(data, list): + return [_strip_dag_version_ids(item) for item in data] + if isinstance(data, dict) and "dag_versions" in data: + result = dict(data) + result["dag_versions"] = [{k: v for k, v in dv.items() if k != "id"} for dv in result["dag_versions"]] + return result + return data + + GRID_NODES = [ { "children": [{"id": "mapped_task_group.subtask", "is_mapped": True, "label": "subtask"}], @@ -357,10 +387,10 @@ def _freeze_time_for_dagruns(time_machine): @pytest.mark.usefixtures("_freeze_time_for_dagruns") class TestGetGridDataEndpoint: def test_should_response_200(self, test_client): - with assert_queries_count(5): + with assert_queries_count(6): response = test_client.get(f"/grid/runs/{DAG_ID}") assert response.status_code == 200 - assert response.json() == [ + assert _strip_dag_version_ids(response.json()) == [ GRID_RUN_1, GRID_RUN_2, ] @@ -399,10 +429,10 @@ def test_should_response_200(self, test_client): ], ) def test_should_response_200_order_by(self, test_client, order_by, expected): - with assert_queries_count(5): + with assert_queries_count(6): response = test_client.get(f"/grid/runs/{DAG_ID}", params={"order_by": order_by}) assert response.status_code == 200 - assert response.json() == expected + assert _strip_dag_version_ids(response.json()) == expected @pytest.mark.parametrize( ("limit", "expected"), @@ -418,10 +448,10 @@ def test_should_response_200_order_by(self, test_client, order_by, expected): ], ) def test_should_response_200_limit(self, test_client, limit, expected): - with assert_queries_count(5): + with assert_queries_count(6): response = test_client.get(f"/grid/runs/{DAG_ID}", params={"limit": limit}) assert response.status_code == 200 - assert response.json() == expected + assert _strip_dag_version_ids(response.json()) == expected @pytest.mark.parametrize( ("params", "expected"), @@ -443,13 +473,13 @@ def test_should_response_200_limit(self, test_client, limit, expected): ], ) def test_runs_should_response_200_date_filters(self, test_client, params, expected): - with assert_queries_count(5): + with assert_queries_count(6): response = test_client.get( f"/grid/runs/{DAG_ID}", params=params, ) assert response.status_code == 200 - assert response.json() == expected + assert _strip_dag_version_ids(response.json()) == expected @pytest.mark.parametrize( ("params", "expected", "expected_queries_count"), @@ -605,33 +635,10 @@ def test_get_dag_structure(self, session, test_client): def test_get_grid_runs(self, session, test_client): session.commit() - with assert_queries_count(5): + with assert_queries_count(6): response = test_client.get(f"/grid/runs/{DAG_ID}?limit=5") assert response.status_code == 200 - assert response.json() == [ - { - "dag_id": "test_dag", - "duration": 283996800.0, - "end_date": "2024-12-31T00:00:00Z", - "has_missed_deadline": False, - "run_after": "2024-11-30T00:00:00Z", - "run_id": "run_1", - "run_type": "scheduled", - "start_date": "2016-01-01T00:00:00Z", - "state": "success", - }, - { - "dag_id": "test_dag", - "duration": 283996800.0, - "end_date": "2024-12-31T00:00:00Z", - "has_missed_deadline": False, - "run_after": "2024-11-30T00:00:00Z", - "run_id": "run_2", - "run_type": "manual", - "start_date": "2016-01-01T00:00:00Z", - "state": "failed", - }, - ] + assert _strip_dag_version_ids(response.json()) == [GRID_RUN_1, GRID_RUN_2] @pytest.mark.parametrize( ("endpoint", "run_type", "expected"), @@ -646,7 +653,7 @@ def test_filter_by_run_type(self, session, test_client, endpoint, run_type, expe session.commit() response = test_client.get(f"/grid/{endpoint}/{DAG_ID}?run_type={run_type}") assert response.status_code == 200 - assert response.json() == expected + assert _strip_dag_version_ids(response.json()) == expected @pytest.mark.parametrize( ("endpoint", "triggering_user", "expected"), @@ -660,14 +667,14 @@ def test_filter_by_triggering_user(self, session, test_client, endpoint, trigger session.commit() response = test_client.get(f"/grid/{endpoint}/{DAG_ID}?triggering_user={triggering_user}") assert response.status_code == 200 - assert response.json() == expected + assert _strip_dag_version_ids(response.json()) == expected def test_get_grid_runs_filter_by_run_type_and_triggering_user(self, session, test_client): session.commit() - with assert_queries_count(5): + with assert_queries_count(6): response = test_client.get(f"/grid/runs/{DAG_ID}?run_type=manual&triggering_user=user2") assert response.status_code == 200 - assert response.json() == [GRID_RUN_2] + assert _strip_dag_version_ids(response.json()) == [GRID_RUN_2] @pytest.mark.parametrize( ("endpoint", "state", "expected"), @@ -683,7 +690,7 @@ def test_filter_by_state(self, session, test_client, endpoint, state, expected): session.commit() response = test_client.get(f"/grid/{endpoint}/{DAG_ID}?state={state}") assert response.status_code == 200 - assert response.json() == expected + assert _strip_dag_version_ids(response.json()) == expected def test_grid_ti_summaries_group(self, session, test_client): run_id = "run_4-1" @@ -702,6 +709,7 @@ def test_grid_ti_summaries_group(self, session, test_client): "task_id": "t1", "task_display_name": "t1", "child_states": None, + "dag_version_number": 1, "max_end_date": "2025-03-02T00:00:00Z", "min_start_date": "2025-03-01T23:59:58Z", }, @@ -710,6 +718,7 @@ def test_grid_ti_summaries_group(self, session, test_client): "task_id": "t2", "task_display_name": "t2", "child_states": None, + "dag_version_number": 1, "max_end_date": "2025-03-02T00:00:02Z", "min_start_date": "2025-03-02T00:00:00Z", }, @@ -718,11 +727,13 @@ def test_grid_ti_summaries_group(self, session, test_client): "task_id": "t7", "task_display_name": "t7", "child_states": None, + "dag_version_number": 1, "max_end_date": "2025-03-02T00:00:04Z", "min_start_date": "2025-03-02T00:00:02Z", }, { "child_states": {"success": 4}, + "dag_version_number": 1, "max_end_date": "2025-03-02T00:00:12Z", "min_start_date": "2025-03-02T00:00:04Z", "state": "success", @@ -734,11 +745,13 @@ def test_grid_ti_summaries_group(self, session, test_client): "task_id": "task_group-1.t6", "task_display_name": "task_group-1.t6", "child_states": None, + "dag_version_number": 1, "max_end_date": "2025-03-02T00:00:06Z", "min_start_date": "2025-03-02T00:00:04Z", }, { "child_states": {"success": 3}, + "dag_version_number": 1, "max_end_date": "2025-03-02T00:00:12Z", "min_start_date": "2025-03-02T00:00:06Z", "state": "success", @@ -750,6 +763,7 @@ def test_grid_ti_summaries_group(self, session, test_client): "task_id": "task_group-1.task_group-2.t3", "task_display_name": "task_group-1.task_group-2.t3", "child_states": None, + "dag_version_number": 1, "max_end_date": "2025-03-02T00:00:08Z", "min_start_date": "2025-03-02T00:00:06Z", }, @@ -758,6 +772,7 @@ def test_grid_ti_summaries_group(self, session, test_client): "task_id": "task_group-1.task_group-2.t4", "task_display_name": "task_group-1.task_group-2.t4", "child_states": None, + "dag_version_number": 1, "max_end_date": "2025-03-02T00:00:10Z", "min_start_date": "2025-03-02T00:00:08Z", }, @@ -766,6 +781,7 @@ def test_grid_ti_summaries_group(self, session, test_client): "task_id": "task_group-1.task_group-2.t5", "task_display_name": "task_group-1.task_group-2.t5", "child_states": None, + "dag_version_number": 1, "max_end_date": "2025-03-02T00:00:12Z", "min_start_date": "2025-03-02T00:00:10Z", }, @@ -797,6 +813,7 @@ def sort_dict(in_dict): expected = [ { "child_states": {"None": 1}, + "dag_version_number": 1, "task_id": "mapped_task_2", "task_display_name": "mapped_task_2", "max_end_date": None, @@ -805,6 +822,7 @@ def sort_dict(in_dict): }, { "child_states": {"success": 1, "running": 1, "None": 1}, + "dag_version_number": 1, "max_end_date": "2024-12-30T01:02:03Z", "min_start_date": "2024-12-30T01:00:00Z", "state": "running", @@ -816,6 +834,7 @@ def sort_dict(in_dict): "task_id": "mapped_task_group.subtask", "task_display_name": "mapped_task_group.subtask", "child_states": None, + "dag_version_number": 1, "max_end_date": "2024-12-30T01:02:03Z", "min_start_date": "2024-12-30T01:00:00Z", }, @@ -824,11 +843,13 @@ def sort_dict(in_dict): "task_id": "task", "task_display_name": "A Beautiful Task Name \U0001f680", "child_states": None, + "dag_version_number": 1, "max_end_date": None, "min_start_date": None, }, { "child_states": {"None": 6}, + "dag_version_number": 1, "task_id": "task_group", "task_display_name": "task_group", "max_end_date": None, @@ -837,6 +858,7 @@ def sort_dict(in_dict): }, { "child_states": {"None": 2}, + "dag_version_number": 1, "task_id": "task_group.inner_task_group", "task_display_name": "task_group.inner_task_group", "max_end_date": None, @@ -845,6 +867,7 @@ def sort_dict(in_dict): }, { "child_states": {"None": 2}, + "dag_version_number": 1, "task_id": "task_group.inner_task_group.inner_task_group_sub_task", "task_display_name": "Inner Task Group Sub Task Label", "max_end_date": None, @@ -853,6 +876,7 @@ def sort_dict(in_dict): }, { "child_states": {"None": 4}, + "dag_version_number": 1, "task_id": "task_group.mapped_task", "task_display_name": "task_group.mapped_task", "max_end_date": None,