Grid task logs filtering and local time

airflow/www/jest-setup.js

@@ -23,6 +23,9 @@ import '@testing-library/jest-dom';
 import axios from 'axios';
 import { setLogger } from 'react-query';
 
+// eslint-disable-next-line import/no-extraneous-dependencies
+import moment from 'moment-timezone';
+
 axios.defaults.adapter = require('axios/lib/adapters/http');
 
 axios.interceptors.response.use(
@@ -51,3 +54,5 @@ global.stateColors = {
 };
 
 global.defaultDagRunDisplayNumber = 245;
+
+global.moment = moment;

airflow/www/static/js/grid/details/content/taskInstance/Logs/index.jsx

@@ -17,7 +17,9 @@
  * under the License.
  */
 
-import React, { useRef, useState, useEffect } from 'react';
+import React, {
+  useRef, useState, useEffect, useMemo,
+} from 'react';
 import {
   Text,
   Box,
@@ -26,12 +28,15 @@ import {
   Code,
   Button,
   Checkbox,
+  Select,
 } from '@chakra-ui/react';
 
 import { getMetaValue } from '../../../../../utils';
 import LogLink from './LogLink';
 import useTaskLog from '../../../../api/useTaskLog';
 import LinkButton from '../../../../components/LinkButton';
+import { logLevel, parseLogs } from './utils';
+import { useTimezone } from '../../../../context/timezone';
 
 const showExternalLogRedirect = getMetaValue('show_external_log_redirect') === 'True';
 const externalLogName = getMetaValue('external_log_name');
@@ -65,6 +70,9 @@ const Logs = ({
   const [selectedAttempt, setSelectedAttempt] = useState(1);
   const [shouldRequestFullContent, setShouldRequestFullContent] = useState(false);
   const [wrap, setWrap] = useState(false);
+  const [logLevelFilter, setLogLevelFilter] = useState('');
+  const [fileSourceFilter, setFileSourceFilter] = useState('');
+  const { timezone } = useTimezone();
   const { data, isSuccess } = useTaskLog({
     dagId,
     dagRunId,
@@ -86,55 +94,98 @@ const Logs = ({
     execution_date: executionDate,
   }).toString();
 
+  const { parsedLogs, fileSources = [] } = useMemo(() => parseLogs(
+    data,
+    timezone,
+    logLevelFilter,
+    fileSourceFilter,
+  ),
+  [data, fileSourceFilter, logLevelFilter, timezone]);
+
+  useEffect(() => {
+    // Reset fileSourceFilter and selected attempt when changing to
+    // a task that do not have those filters anymore.
+    if (!internalIndexes.includes(selectedAttempt)) {
+      setSelectedAttempt(internalIndexes[0]);
+    }
+    if (fileSourceFilter && !fileSources.includes(fileSourceFilter)) {
+      setFileSourceFilter('');
+    }
+  }, [data, internalIndexes, fileSourceFilter, fileSources, selectedAttempt]);
+
   return (
     <>
       {tryNumber > 0 && (
       <>
         <Text as="span"> (by attempts)</Text>
-        <Box>
-          <Flex my={1} justifyContent="space-between">
-            <Flex flexWrap="wrap">
-              {internalIndexes.map((index) => (
-                <Button
-                  key={index}
-                  variant="ghost"
-                  colorScheme="blue"
-                  onClick={() => setSelectedAttempt(index)}
-                  data-testid={`log-attempt-select-button-${index}`}
-                >
-                  {index}
-                </Button>
-              ))}
-            </Flex>
-            <Flex>
-              <Checkbox
-                onChange={() => setWrap((previousState) => !previousState)}
-                px={4}
+        <Flex my={1} justifyContent="space-between">
+          <Flex flexWrap="wrap">
+            {internalIndexes.map((index) => (
+              <Button
+                key={index}
+                variant="ghost"
+                colorScheme="blue"
+                onClick={() => setSelectedAttempt(index)}
+                data-testid={`log-attempt-select-button-${index}`}
               >
-                <Text as="strong">Wrap</Text>
-              </Checkbox>
-              <Checkbox
-                onChange={() => setShouldRequestFullContent((previousState) => !previousState)}
-                px={4}
-                data-testid="full-content-checkbox"
+                {index}
+              </Button>
+            ))}
+          </Flex>
+          <Flex alignItems="center">
+            <Box w="90px" mr={2}>
+              <Select
+                size="sm"
+                value={logLevelFilter}
+                onChange={(e) => setLogLevelFilter(e.target.value)}
               >
-                <Text as="strong">Full Logs</Text>
-              </Checkbox>
-              <LogLink
-                index={selectedAttempt}
-                dagId={dagId}
-                taskId={taskId}
-                executionDate={executionDate}
-                isInternal
-              />
-              <LinkButton
-                href={`${logUrl}&${params}`}
+                <option value="" key="all">All Levels</option>
+                {Object.values(logLevel).map((value) => (
+                  <option value={value} key={value}>{value}</option>
+                ))}
+              </Select>
+            </Box>
+            <Box w="110px">
+              <Select
+                size="sm"
+                value={fileSourceFilter}
+                onChange={(e) => setFileSourceFilter(e.target.value)}
               >
-                See More
-              </LinkButton>
-            </Flex>
+                <option value="" key="all">All File Sources</option>
+                {fileSources.map((value) => (
+                  <option value={value} key={value}>{value}</option>
+                ))}
+              </Select>
+            </Box>
           </Flex>
-        </Box>
+          <Flex alignItems="center">
+            <Checkbox
+              onChange={() => setWrap((previousState) => !previousState)}
+              px={4}
+            >
+              <Text as="strong">Wrap</Text>
+            </Checkbox>
+            <Checkbox
+              onChange={() => setShouldRequestFullContent((previousState) => !previousState)}
+              px={4}
+              data-testid="full-content-checkbox"
+            >
+              <Text as="strong">Full Logs</Text>
+            </Checkbox>
+            <LogLink
+              index={selectedAttempt}
+              dagId={dagId}
+              taskId={taskId}
+              executionDate={executionDate}
+              isInternal
+            />
+            <LinkButton
+              href={`${logUrl}&${params}`}
+            >
+              See More
+            </LinkButton>
+          </Flex>
+        </Flex>
         {
           isSuccess && (
             <Code
@@ -148,7 +199,7 @@ const Logs = ({
               borderRadius={3}
               borderColor="blue.500"
             >
-              {data}
+              {parsedLogs}
               <div ref={codeBlockBottomDiv} />
             </Code>
           )

airflow/www/static/js/grid/details/content/taskInstance/Logs/index.test.jsx

@@ -40,7 +40,7 @@ const mockTaskLog = `
 [2022-06-04 00:00:01,921] {standard_task_runner.py:81} INFO - Job 1626: Subtask section_1.get_entry_group
 [2022-06-04 00:00:01,921] {dagbag.py:507} INFO - Filling up the DagBag from /files/dags/test_ui_grid.py
 [2022-06-04 00:00:01,964] {task_command.py:377} INFO - Running <TaskInstance: test_ui_grid.section_1.get_entry_group scheduled__2022-06-03T00:00:00+00:00 [running]> on host 5d28cfda3219
-[2022-06-04 00:00:02,010] {taskinstance.py:1548} INFO - Exporting the following env vars:
+[2022-06-04 00:00:02,010] {taskinstance.py:1548} WARNING - Exporting the following env vars:
 AIRFLOW_CTX_DAG_OWNER=***
 AIRFLOW_CTX_DAG_ID=test_ui_grid
 `;
@@ -67,8 +67,8 @@ describe('Test Logs Component.', () => {
         tryNumber={tryNumber}
       />,
     );
-    expect(getByText('[2022-06-04 00:00:01,906] {taskinstance.py:1330} INFO -', { exact: false })).toBeDefined();
-    expect(getByText('[2022-06-04 00:00:01,921] {standard_task_runner.py:81} INFO - Job 1626: Subtask section_1.get_entry_group',
+    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();
   });

airflow/www/static/js/grid/details/content/taskInstance/Logs/utils.js

@@ -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.
+ */
+
+/* global moment */
+
+import { defaultFormatWithTZ } from '../../../../../datetime_utils';
+
+export const logLevel = {
+  DEBUG: 'DEBUG',
+  INFO: 'INFO',
+  WARNING: 'WARNING',
+  ERROR: 'ERROR',
+  CRITICAL: 'CRITICAL',
+};
+
+export const parseLogs = (data, timezone, logLevelFilter, fileSourceFilter) => {
+  const lines = data.split('\n');
+
+  if (!data) {
+    return {};
+  }
+
+  const parsedLines = [];
+  const fileSources = new Set();
+
+  lines.forEach((line) => {
+    let parsedLine = line;
+
+    // Apply log level filter.
+    if (logLevelFilter && !line.includes(logLevelFilter)) {
+      return;
+    }
+
+    const regExp = /\[(.*?)\] \{(.*?)\}/;
+    const matches = line.match(regExp);
+    let logGroup = '';
+    if (matches) {
+      // Replace UTC with the local timezone.
+      const dateTime = matches[1];
+      [logGroup] = matches[2].split(':');
+      if (dateTime && timezone) {
+        const localDateTime = moment.utc(dateTime).tz(timezone).format(defaultFormatWithTZ);
+        parsedLine = line.replace(dateTime, localDateTime);
+      }
+
+      fileSources.add(logGroup);
+    }
+    if (!fileSourceFilter || fileSourceFilter === logGroup) {
+      parsedLines.push(parsedLine);
+    }
+  });
+
+  return { parsedLogs: parsedLines.join('\n'), fileSources: Array.from(fileSources).sort() };
+};