Grid logs for mapped instances (#25610)

* Add logs for mapped tasks

* Mapped Task for download link, see more and paginated Table.

* Handle refresh on select mapped task instance + other fix

* Add useSelection tests for mapIndex

* Stop propagation on mapped instance icon clicked

* Add nav link for mapped task instances

* Add Back to Dynamic Task Summary button

* Move MappedInstances to its own tab

* Update following code review

* Simplifies the logic by using get_task_instance endpoint

* Pull instance either from useGridData or useTaskInstance

* Update test

airflow/www/static/js/api/index.ts

@@ -36,6 +36,7 @@ import useDatasets from './useDatasets';
 import useDataset from './useDataset';
 import useDatasetEvents from './useDatasetEvents';
 import useUpstreamDatasetEvents from './useUpstreamDatasetEvents';
+import useTaskInstance from './useTaskInstance';
 
 axios.interceptors.response.use(
   (res: AxiosResponse) => (res.data ? camelcaseKeys(res.data, { deep: true }) : res),
@@ -45,19 +46,20 @@ axios.defaults.headers.common.Accept = 'application/json';
 
 export {
   useClearRun,
-  useQueueRun,
-  useMarkFailedRun,
-  useMarkSuccessRun,
-  useRunTask,
   useClearTask,
-  useMarkFailedTask,
-  useMarkSuccessTask,
-  useExtraLinks,
   useConfirmMarkTask,
-  useGridData,
-  useMappedInstances,
-  useDatasets,
   useDataset,
   useDatasetEvents,
+  useDatasets,
+  useExtraLinks,
+  useGridData,
+  useMappedInstances,
+  useMarkFailedRun,
+  useMarkFailedTask,
+  useMarkSuccessRun,
+  useMarkSuccessTask,
+  useQueueRun,
+  useRunTask,
+  useTaskInstance,
   useUpstreamDatasetEvents,
 };

airflow/www/static/js/api/useTaskInstance.tsx

@@ -0,0 +1,70 @@
+/*!
+ * 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 axios, { AxiosResponse } from 'axios';
+import type { API, TaskInstance } from 'src/types';
+import { useQuery } from 'react-query';
+import { useAutoRefresh } from 'src/context/autorefresh';
+
+import { getMetaValue } from 'src/utils';
+
+/* GridData.TaskInstance and API.TaskInstance are not compatible at the moment.
+ * Remove this function when changing the api response for grid_data_url to comply
+ * with API.TaskInstance.
+ */
+const convertTaskInstance = (
+  ti:
+  API.TaskInstance,
+) => ({ ...ti, runId: ti.dagRunId }) as TaskInstance;
+
+const taskInstanceApi = getMetaValue('task_instance_api');
+
+const useTaskInstance = ({
+  dagId, dagRunId, taskId, mapIndex, enabled,
+}: {
+  dagId: string,
+  dagRunId: string,
+  taskId: string | null,
+  mapIndex?: number,
+  enabled: boolean
+}) => {
+  let url: string = '';
+  if (taskInstanceApi) {
+    url = taskInstanceApi.replace('_DAG_RUN_ID_', dagRunId).replace('_TASK_ID_', taskId || '');
+  }
+
+  if (mapIndex !== undefined && mapIndex >= 0) {
+    url += `/${mapIndex.toString()}`;
+  }
+
+  const { isRefreshOn } = useAutoRefresh();
+
+  return useQuery(
+    ['taskIntance', dagId, dagRunId, taskId, mapIndex],
+    () => axios.get<AxiosResponse, API.TaskInstance>(url, { headers: { Accept: 'text/plain' } }),
+    {
+      placeholderData: {},
+      refetchInterval: isRefreshOn && (autoRefreshInterval || 1) * 1000,
+      enabled,
+      select: convertTaskInstance,
+    },
+  );
+};
+
+export default useTaskInstance;

airflow/www/static/js/api/useTaskLog.tsx

@@ -26,12 +26,13 @@ import { getMetaValue } from 'src/utils';
 const taskLogApi = getMetaValue('task_log_api');
 
 const useTaskLog = ({
-  dagId, dagRunId, taskId, taskTryNumber, fullContent,
+  dagId, dagRunId, taskId, taskTryNumber, mapIndex, fullContent,
 }: {
   dagId: string,
   dagRunId: string,
   taskId: string,
   taskTryNumber: number,
+  mapIndex?: number,
   fullContent: boolean,
 }) => {
   let url: string = '';
@@ -42,8 +43,8 @@ const useTaskLog = ({
   const { isRefreshOn } = useAutoRefresh();
 
   return useQuery(
-    ['taskLogs', dagId, dagRunId, taskId, taskTryNumber, fullContent],
-    () => axios.get<AxiosResponse, string>(url, { headers: { Accept: 'text/plain' }, params: { full_content: fullContent } }),
+    ['taskLogs', dagId, dagRunId, taskId, mapIndex, taskTryNumber, fullContent],
+    () => axios.get<AxiosResponse, string>(url, { headers: { Accept: 'text/plain' }, params: { map_index: mapIndex, full_content: fullContent } }),
     {
       placeholderData: '',
       refetchInterval: isRefreshOn && (autoRefreshInterval || 1) * 1000,

airflow/www/static/js/dag/details/BreadcrumbText.tsx

@@ -30,7 +30,16 @@ interface Props {
 
 const BreadcrumbText = ({ label, value }: Props) => (
   <Box position="relative">
-    <Heading as="h5" size="sm" color="gray.300" position="absolute" top="-12px">{label}</Heading>
+    <Heading
+      as="h5"
+      size="sm"
+      color="gray.300"
+      position="absolute"
+      top="-12px"
+      whiteSpace="nowrap"
+    >
+      {label}
+    </Heading>
     <Heading as="h3" size="md">{value}</Heading>
   </Box>
 );

airflow/www/static/js/dag/details/Header.tsx

@@ -38,7 +38,7 @@ const dagId = getMetaValue('dag_id');
 const Header = () => {
   const { data: { dagRuns, groups } } = useGridData();
 
-  const { selected: { taskId, runId }, onSelect, clearSelection } = useSelection();
+  const { selected: { taskId, runId, mapIndex }, onSelect, clearSelection } = useSelection();
   const dagRun = dagRuns.find((r) => r.runId === runId);
 
   // clearSelection if the current selected dagRun is
@@ -75,7 +75,8 @@ const Header = () => {
 
   const isDagDetails = !runId && !taskId;
   const isRunDetails = !!(runId && !taskId);
-  const isTaskDetails = runId && taskId;
+  const isTaskDetails = runId && taskId && mapIndex === null;
+  const isMappedTaskDetails = runId && taskId && mapIndex !== null;
 
   return (
     <Breadcrumb separator={<Text color="gray.300">/</Text>}>
@@ -93,11 +94,18 @@ const Header = () => {
       )}
       {taskId && (
         <BreadcrumbItem isCurrentPage mt={4}>
-          <BreadcrumbLink _hover={isTaskDetails ? { cursor: 'default' } : undefined}>
+          <BreadcrumbLink onClick={() => onSelect({ runId, taskId })} _hover={isTaskDetails ? { cursor: 'default' } : undefined}>
             <BreadcrumbText label="Task" value={`${taskName}${group?.isMapped ? ' []' : ''}`} />
           </BreadcrumbLink>
         </BreadcrumbItem>
       )}
+      {mapIndex !== null && (
+        <BreadcrumbItem isCurrentPage mt={4}>
+          <BreadcrumbLink _hover={isMappedTaskDetails ? { cursor: 'default' } : undefined}>
+            <BreadcrumbText label="Map Index" value={mapIndex} />
+          </BreadcrumbLink>
+        </BreadcrumbItem>
+      )}
     </Breadcrumb>
   );
 };

airflow/www/static/js/dag/details/index.tsx

@@ -32,7 +32,7 @@ import DagRunContent from './dagRun';
 import DagContent from './Dag';
 
 const Details = () => {
-  const { selected: { runId, taskId } } = useSelection();
+  const { selected: { runId, taskId, mapIndex }, onSelect } = useSelection();
   return (
     <Flex flexDirection="column" pl={3} mr={3} flexGrow={1} maxWidth="750px">
       <Header />
@@ -46,6 +46,8 @@ const Details = () => {
         <TaskInstanceContent
           runId={runId}
           taskId={taskId}
+          mapIndex={mapIndex === null ? undefined : mapIndex}
+          onSelect={onSelect}
         />
         )}
       </Box>

airflow/www/static/js/dag/details/taskInstance/BackToTaskSummary.tsx

@@ -0,0 +1,45 @@
+/*!
+ * 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 React from 'react';
+import { Button, Flex } from '@chakra-ui/react';
+
+interface Props {
+  isMapIndexDefined: boolean;
+  onClick: () => void;
+}
+
+const BackToTaskSummary = ({ isMapIndexDefined, onClick }: Props) => {
+  if (!isMapIndexDefined) return null;
+
+  return (
+    <Flex justifyContent="right">
+      <Button
+        variant="ghost"
+        colorScheme="blue"
+        onClick={onClick}
+        size="lg"
+      >
+        Back to Dynamic Task Summary
+      </Button>
+    </Flex>
+  );
+};
+
+export default BackToTaskSummary;

airflow/www/static/js/dag/details/taskInstance/Details.tsx

@@ -52,6 +52,7 @@ const Details = ({ instance, group }: Props) => {
     endDate,
     state,
     mappedStates,
+    mapIndex,
   } = instance;
 
   const {
@@ -142,6 +143,12 @@ const Details = ({ instance, group }: Props) => {
             <Td>Run ID</Td>
             <Td><Text whiteSpace="nowrap"><ClipboardText value={runId} /></Text></Td>
           </Tr>
+          {mapIndex !== undefined && (
+            <Tr>
+              <Td>Map Index</Td>
+              <Td>{mapIndex}</Td>
+            </Tr>
+          )}
           {operator && (
             <Tr>
               <Td>Operator</Td>

airflow/www/static/js/dag/details/taskInstance/Logs/LogLink.test.tsx

@@ -41,7 +41,7 @@ describe('Test LogLink Component.', () => {
     expect(linkElement).toBeDefined();
     expect(linkElement).not.toHaveAttribute('target');
     expect(linkElement?.href.includes(
-      `?dag_id=dummyDagId&task_id=dummyTaskId&execution_date=2020%3A01%3A01T01%3A00%2B00%3A00&format=file&try_number=${tryNumber}`,
+      `?dag_id=dummyDagId&task_id=dummyTaskId&execution_date=2020%3A01%3A01T01%3A00%2B00%3A00&map_index=-1&format=file&try_number=${tryNumber}`,
     )).toBeTruthy();
   });
 
@@ -61,7 +61,7 @@ describe('Test LogLink Component.', () => {
     expect(linkElement).toBeDefined();
     expect(linkElement).toHaveAttribute('target', '_blank');
     expect(linkElement?.href.includes(
-      `?dag_id=dummyDagId&task_id=dummyTaskId&execution_date=2020%3A01%3A01T01%3A00%2B00%3A00&try_number=${tryNumber}`,
+      `?dag_id=dummyDagId&task_id=dummyTaskId&execution_date=2020%3A01%3A01T01%3A00%2B00%3A00&map_index=-1&try_number=${tryNumber}`,
     )).toBeTruthy();
   });
 });

airflow/www/static/js/dag/details/taskInstance/Logs/LogLink.tsx

@@ -32,15 +32,17 @@ interface Props {
   executionDate: DagRun['executionDate'];
   isInternal?: boolean;
   tryNumber: TaskInstance['tryNumber'];
+  mapIndex?: TaskInstance['mapIndex'];
 }
 
 const LogLink = ({
-  dagId, taskId, executionDate, isInternal, tryNumber,
+  dagId, taskId, executionDate, isInternal, tryNumber, mapIndex,
 }: Props) => {
   let fullMetadataUrl = `${isInternal ? logsWithMetadataUrl : externalLogUrl
   }?dag_id=${encodeURIComponent(dagId)
   }&task_id=${encodeURIComponent(taskId)
   }&execution_date=${encodeURIComponent(executionDate)
+  }&map_index=${encodeURIComponent(mapIndex?.toString() ?? '-1')
   }`;
 
   if (isInternal && tryNumber) {

airflow/www/static/js/dag/details/taskInstance/Logs/index.test.tsx

@@ -78,6 +78,43 @@ describe('Test Logs Component.', () => {
       { exact: false },
     )).toBeDefined();
     expect(getByText('AIRFLOW_CTX_DAG_ID=test_ui_grid', { exact: false })).toBeDefined();
+
+    expect(useTaskLogMock).toHaveBeenLastCalledWith({
+      dagId: 'dummyDagId',
+      dagRunId: 'dummyDagRunId',
+      fullContent: false,
+      taskId: 'dummyTaskId',
+      taskTryNumber: 1,
+    });
+  });
+
+  test('Test Logs Content Mapped Task', () => {
+    const tryNumber = 2;
+    const { getByText } = render(
+      <Logs
+        dagId="dummyDagId"
+        dagRunId="dummyDagRunId"
+        taskId="dummyTaskId"
+        executionDate="2020:01:01T01:00+00:00"
+        mapIndex={1}
+        tryNumber={tryNumber}
+      />,
+    );
+    expect(getByText('[2022-06-04, 00:00:01 UTC] {taskinstance.py:1329} INFO -', { exact: false })).toBeDefined();
+    expect(getByText(
+      '[2022-06-04, 00:00:01 UTC] {standard_task_runner.py:81} INFO - Job 1626: Subtask section_1.get_entry_group',
+      { exact: false },
+    )).toBeDefined();
+    expect(getByText('AIRFLOW_CTX_DAG_ID=test_ui_grid', { exact: false })).toBeDefined();
+
+    expect(useTaskLogMock).toHaveBeenLastCalledWith({
+      dagId: 'dummyDagId',
+      dagRunId: 'dummyDagRunId',
+      fullContent: false,
+      mapIndex: 1,
+      taskId: 'dummyTaskId',
+      taskTryNumber: 1,
+    });
   });
 
   test('Test Logs Attempt Select Button', () => {