Add task group detail across dag runs

Author: guan404ming

airflow-core/src/airflow/ui/src/components/Graph/TaskLink.tsx

@@ -26,25 +26,20 @@ type Props = {
 } & TaskNameProps;
 
 export const TaskLink = forwardRef<HTMLAnchorElement, Props>(({ id, isGroup, isMapped, ...rest }, ref) => {
-  const { dagId = "", runId, taskId } = useParams();
+  const { dagId = "", groupId, runId, taskId } = useParams();
   const [searchParams] = useSearchParams();
 
-  if (isGroup && runId === undefined) {
-    return undefined;
-  }
-
-  const pathname = isGroup
-    ? `/dags/${dagId}/runs/${runId}/tasks/group/${id}`
-    : `/dags/${dagId}/${runId === undefined ? "" : `runs/${runId}/`}${taskId === id ? "" : `tasks/${id}`}${isMapped && taskId !== id && runId !== undefined ? "/mapped" : ""}`;
+  const basePath = `/dags/${dagId}${runId === undefined ? "" : `/runs/${runId}`}`;
+  const taskPath = isGroup
+    ? groupId === id
+      ? ""
+      : `/tasks/group/${id}`
+    : taskId === id
+      ? ""
+      : `/tasks/${id}${isMapped && taskId !== id && runId !== undefined ? "/mapped" : ""}`;
 
   return (
-    <RouterLink
-      ref={ref}
-      to={{
-        pathname,
-        search: searchParams.toString(),
-      }}
-    >
+    <RouterLink ref={ref} to={{ pathname: basePath + taskPath, search: searchParams.toString() }}>
       <TaskName isGroup={isGroup} isMapped={isMapped} {...rest} />
     </RouterLink>
   );

airflow-core/src/airflow/ui/src/layouts/Details/DagBreadcrumb.tsx

@@ -32,7 +32,7 @@ import { TogglePause } from "src/components/TogglePause";
 import { isStatePending, useAutoRefresh } from "src/utils";
 
 export const DagBreadcrumb = () => {
-  const { dagId = "", mapIndex = "-1", runId, taskId } = useParams();
+  const { dagId = "", groupId, mapIndex = "-1", runId, taskId } = useParams();
   const refetchInterval = useAutoRefresh({ dagId });
 
   const { data: dag } = useDagServiceGetDagDetails({
@@ -86,6 +86,23 @@ export const DagBreadcrumb = () => {
     });
   }
 
+  // Add group breadcrumb
+  if (groupId !== undefined) {
+    if (runId === undefined) {
+      links.push({
+        label: "All Runs",
+        title: "Dag Run",
+        value: `/dags/${dagId}/runs`,
+      });
+    }
+
+    links.push({
+      label: groupId,
+      title: "Group",
+      value: `/dags/${dagId}/groups/${groupId}`,
+    });
+  }
+
   // Add task breadcrumb
   if (runId !== undefined && taskId !== undefined) {
     if (task?.is_mapped) {

airflow-core/src/airflow/ui/src/layouts/Details/Grid/TaskNames.tsx

@@ -66,16 +66,26 @@ export const TaskNames = ({ nodes }: Props) => {
     >
       {node.isGroup ? (
         <Flex>
-          <TaskName
-            display="inline"
-            fontSize="sm"
-            fontWeight="normal"
-            isGroup={true}
-            isMapped={Boolean(node.is_mapped)}
-            label={node.label}
-            paddingLeft={node.depth * 3 + 2}
-            setupTeardownType={node.setup_teardown_type}
-          />
+          <Link asChild data-testid={node.id}>
+            <RouterLink
+              replace
+              to={{
+                pathname: `/dags/${dagId}/tasks/group/${node.id}`,
+                search: searchParams.toString(),
+              }}
+            >
+              <TaskName
+                display="inline"
+                fontSize="sm"
+                fontWeight="normal"
+                isGroup={true}
+                isMapped={Boolean(node.is_mapped)}
+                label={node.label}
+                paddingLeft={node.depth * 3 + 2}
+                setupTeardownType={node.setup_teardown_type}
+              />
+            </RouterLink>
+          </Link>
           <chakra.button
             aria-label="Toggle group"
             display="inline"

airflow-core/src/airflow/ui/src/pages/Events/Events.tsx

@@ -138,7 +138,7 @@ const eventsColumn = ({ dagId, open, runId, taskId }: EventsColumn): Array<Colum
 ];
 
 export const Events = () => {
-  const { dagId, runId, taskId } = useParams();
+  const { dagId, groupId, runId, taskId } = useParams();
   const { setTableURLState, tableURLState } = useTableURLState();
   const { pagination, sorting } = tableURLState;
   const [sort] = sorting;
@@ -153,7 +153,7 @@ export const Events = () => {
       offset: pagination.pageIndex * pagination.pageSize,
       orderBy,
       runId,
-      taskId,
+      taskId: groupId ?? taskId,
     },
     undefined,
     { enabled: !isNaN(pagination.pageSize) },

airflow-core/src/airflow/ui/src/pages/Task/GroupTaskHeader.tsx

@@ -0,0 +1,29 @@
+/*!
+ * 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 { NodeResponse } from "openapi/requests/types.gen";
+import { TaskIcon } from "src/assets/TaskIcon";
+import { HeaderCard } from "src/components/HeaderCard";
+
+export const GroupTaskHeader = ({ groupTask }: { readonly groupTask: NodeResponse }) => (
+  <HeaderCard
+    icon={<TaskIcon />}
+    stats={[{ label: "Operator", value: groupTask.operator ?? "Null" }]}
+    title={groupTask.label}
+  />
+);

airflow-core/src/airflow/ui/src/pages/Task/Overview/Overview.tsx

@@ -30,7 +30,7 @@ import { isStatePending, useAutoRefresh } from "src/utils";
 const defaultHour = "24";
 
 export const Overview = () => {
-  const { dagId = "", taskId } = useParams();
+  const { dagId = "", groupId, taskId } = useParams();
 
   const now = dayjs();
   const [startDate, setStartDate] = useState(now.subtract(Number(defaultHour), "hour").toISOString());
@@ -46,7 +46,8 @@ export const Overview = () => {
       runAfterGte: startDate,
       runAfterLte: endDate,
       state: ["failed"],
-      taskId,
+      taskDisplayNamePattern: groupId ?? undefined,
+      taskId: Boolean(groupId) ? undefined : taskId,
     });
 
   const { data: taskInstances, isLoading: isLoadingTaskInstances } = useTaskInstanceServiceGetTaskInstances(
@@ -55,7 +56,8 @@ export const Overview = () => {
       dagRunId: "~",
       limit: 14,
       orderBy: "-run_after",
-      taskId,
+      taskDisplayNamePattern: groupId ?? undefined,
+      taskId: Boolean(groupId) ? undefined : taskId,
     },
     undefined,
     {

airflow-core/src/airflow/ui/src/pages/Task/Task.tsx

@@ -21,9 +21,11 @@ import { LuChartColumn } from "react-icons/lu";
 import { MdOutlineEventNote, MdOutlineTask } from "react-icons/md";
 import { useParams } from "react-router-dom";
 
-import { useDagServiceGetDagDetails, useTaskServiceGetTask } from "openapi/queries";
+import { useDagServiceGetDagDetails, useGridServiceGridData, useTaskServiceGetTask } from "openapi/queries";
 import { DetailsLayout } from "src/layouts/Details/DetailsLayout";
+import { getGroupTask } from "src/utils/groupTask";
 
+import { GroupTaskHeader } from "./GroupTaskHeader";
 import { Header } from "./Header";
 
 const tabs = [
@@ -33,9 +35,28 @@ const tabs = [
 ];
 
 export const Task = () => {
-  const { dagId = "", taskId = "" } = useParams();
+  const { dagId = "", groupId, taskId } = useParams();
 
-  const { data: task, error, isLoading } = useTaskServiceGetTask({ dagId, taskId });
+  const {
+    data: task,
+    error,
+    isLoading,
+  } = useTaskServiceGetTask({ dagId, taskId: groupId ?? taskId }, undefined, {
+    enabled: groupId === undefined,
+  });
+
+  const { data: gridData } = useGridServiceGridData(
+    {
+      dagId,
+      includeDownstream: true,
+      includeUpstream: true,
+    },
+    undefined,
+    { enabled: groupId !== undefined },
+  );
+
+  const groupTask =
+    groupId === undefined ? undefined : getGroupTask(gridData?.structure.nodes ?? [], groupId);
 
   const {
     data: dag,
@@ -49,6 +70,7 @@ export const Task = () => {
     <ReactFlowProvider>
       <DetailsLayout dag={dag} error={error ?? dagError} isLoading={isLoading || isDagLoading} tabs={tabs}>
         {task === undefined ? undefined : <Header task={task} />}
+        {groupTask ? <GroupTaskHeader groupTask={groupTask} /> : undefined}
       </DetailsLayout>
     </ReactFlowProvider>
   );

airflow-core/src/airflow/ui/src/router.tsx

@@ -176,6 +176,15 @@ export const routerConfig = [
         element: <GroupTaskInstance />,
         path: "dags/:dagId/runs/:runId/tasks/group/:groupId",
       },
+      {
+        children: [
+          { element: <TaskOverview />, index: true },
+          { element: <TaskInstances />, path: "task_instances" },
+          { element: <Events />, path: "events" },
+        ],
+        element: <Task />,
+        path: "dags/:dagId/tasks/group/:groupId",
+      },
       {
         children: taskInstanceRoutes,
         element: <TaskInstance />,

airflow-core/src/airflow/ui/src/utils/groupTask.ts

@@ -0,0 +1,55 @@
+/*!
+ * 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 { NodeResponse } from "openapi/requests/types.gen";
+
+/**
+ * Finds a task node by its ID in a tree of nodes
+ * @param nodes - Array of root nodes to search through
+ * @param targetId - ID of the node to find
+ * @returns The found node or undefined if not found
+ */
+export const getGroupTask = (nodes: Array<NodeResponse>, targetId: string): NodeResponse | undefined => {
+  if (!nodes.length || !targetId) {
+    return undefined;
+  }
+
+  const queue: Array<NodeResponse> = [...nodes];
+  const [root] = targetId.split(".");
+
+  while (queue.length > 0) {
+    const node = queue.shift();
+
+    if (node) {
+      if (node.id === targetId) {
+        return node;
+      }
+
+      if (node.children && node.children.length > 0) {
+        const nextNode =
+          node.id === root && targetId.includes(".")
+            ? node.children.find((child) => child.id === targetId)
+            : undefined;
+
+        queue.unshift(...(nextNode ? [nextNode] : node.children));
+      }
+    }
+  }
+
+  return undefined;
+};