Add Dag Audit Log to React

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

@@ -50,6 +50,7 @@ import useDags from "./useDags";
 import useDagRuns from "./useDagRuns";
 import useHistoricalMetricsData from "./useHistoricalMetricsData";
 import { useTaskXcomEntry, useTaskXcomCollection } from "./useTaskXcom";
+import useEventLogs from "./useEventLogs";
 
 axios.interceptors.request.use((config) => {
   config.paramsSerializer = {
@@ -96,4 +97,5 @@ export {
   useHistoricalMetricsData,
   useTaskXcomEntry,
   useTaskXcomCollection,
+  useEventLogs,
 };

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

@@ -0,0 +1,59 @@
+/*!
+ * 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 { useQuery } from "react-query";
+
+import { getMetaValue } from "src/utils";
+import type { API } from "src/types";
+import { useAutoRefresh } from "src/context/autorefresh";
+
+export default function useEventLogs({
+  dagId,
+  taskId,
+  limit,
+  offset,
+  orderBy,
+  after,
+  before,
+  owner,
+}: API.GetEventLogsVariables) {
+  const { isRefreshOn } = useAutoRefresh();
+  return useQuery(
+    ["eventLogs", dagId, taskId, limit, offset, orderBy, after, before, owner],
+    () => {
+      const eventsLogUrl = getMetaValue("event_logs_api");
+      const orderParam = orderBy ? { order_by: orderBy } : {};
+      return axios.get<AxiosResponse, API.EventLogCollection>(eventsLogUrl, {
+        params: {
+          offset,
+          limit,
+          ...{ dag_id: dagId },
+          ...{ task_id: taskId },
+          ...orderParam,
+          after,
+          before,
+        },
+      });
+    },
+    {
+      refetchInterval: isRefreshOn && (autoRefreshInterval || 1) * 1000,
+    }
+  );
+}

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

@@ -0,0 +1,170 @@
+/*!
+ * 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 React, { useMemo, useRef, useState } from "react";
+import {
+  Box,
+  Flex,
+  FormControl,
+  FormHelperText,
+  FormLabel,
+  Input,
+  HStack,
+} from "@chakra-ui/react";
+import type { SortingRule } from "react-table";
+import { snakeCase } from "lodash";
+
+import { CodeCell, Table, TimeCell } from "src/components/Table";
+import { useEventLogs } from "src/api";
+import { getMetaValue, useOffsetTop } from "src/utils";
+import type { DagRun } from "src/types";
+import LinkButton from "src/components/LinkButton";
+
+interface Props {
+  taskId?: string;
+  run?: DagRun;
+}
+
+const dagId = getMetaValue("dag_id") || undefined;
+
+const AuditLog = ({ taskId, run }: Props) => {
+  const logRef = useRef<HTMLDivElement>(null);
+  const offsetTop = useOffsetTop(logRef);
+  const limit = 25;
+  const [offset, setOffset] = useState(0);
+  const [sortBy, setSortBy] = useState<SortingRule<object>[]>([
+    { id: "when", desc: true },
+  ]);
+
+  const sort = sortBy[0];
+  const orderBy = sort ? `${sort.desc ? "-" : ""}${snakeCase(sort.id)}` : "";
+
+  const { data, isLoading } = useEventLogs({
+    dagId,
+    taskId,
+    before: run?.lastSchedulingDecision || undefined,
+    after: run?.queuedAt || undefined,
+    orderBy,
+    limit,
+    offset,
+  });
+
+  const columns = useMemo(() => {
+    const when = {
+      Header: "When",
+      accessor: "when",
+      Cell: TimeCell,
+    };
+    const task = {
+      Header: "Task ID",
+      accessor: "taskId",
+    };
+    const rest = [
+      {
+        Header: "Event",
+        accessor: "event",
+      },
+      {
+        Header: "Owner",
+        accessor: "owner",
+      },
+      {
+        Header: "Extra",
+        accessor: "extra",
+        Cell: CodeCell,
+      },
+    ];
+    return !taskId ? [when, task, ...rest] : [when, ...rest];
+  }, [taskId]);
+
+  const memoData = useMemo(() => data?.eventLogs, [data?.eventLogs]);
+  const memoSort = useMemo(() => sortBy, [sortBy]);
+
+  return (
+    <Box
+      height="100%"
+      maxHeight={`calc(100% - ${offsetTop}px)`}
+      ref={logRef}
+      overflowY="auto"
+    >
+      <Flex justifyContent="right">
+        <LinkButton href={getMetaValue("audit_log_url")}>
+          View full cluster Audit Log
+        </LinkButton>
+      </Flex>
+      <HStack spacing={2} alignItems="flex-start">
+        <FormControl>
+          <FormLabel>Show Logs After</FormLabel>
+          <Input
+            type="datetime"
+            // @ts-ignore
+            placeholder={run?.queuedAt ? moment(run?.queuedAt).format() : ""}
+            isDisabled
+          />
+          {!!run && run?.queuedAt && (
+            <FormHelperText>After selected DAG Run Queued At</FormHelperText>
+          )}
+        </FormControl>
+        <FormControl>
+          <FormLabel>Show Logs Before</FormLabel>
+          <Input
+            type="datetime"
+            placeholder={
+              run?.lastSchedulingDecision
+                ? // @ts-ignore
+                  moment(run?.lastSchedulingDecision).format()
+                : ""
+            }
+            isDisabled
+          />
+          {!!run && run?.lastSchedulingDecision && (
+            <FormHelperText>
+              Before selected DAG Run Last Scheduling Decision
+            </FormHelperText>
+          )}
+        </FormControl>
+        <FormControl>
+          <FormLabel>Filter by Task ID</FormLabel>
+          <Input placeholder={taskId} isDisabled />
+          <FormHelperText />
+        </FormControl>
+      </HStack>
+      <Table
+        data={memoData || []}
+        columns={columns}
+        isLoading={isLoading}
+        manualPagination={{
+          offset,
+          setOffset,
+          totalEntries: data?.totalEntries || 0,
+        }}
+        manualSort={{
+          setSortBy,
+          sortBy,
+          initialSortBy: memoSort,
+        }}
+        pageSize={limit}
+      />
+    </Box>
+  );
+};
+
+export default AuditLog;

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

@@ -42,6 +42,7 @@ import {
   MdOutlineViewTimeline,
   MdSyncAlt,
   MdHourglassBottom,
+  MdPlagiarism,
 } from "react-icons/md";
 import { BiBracket } from "react-icons/bi";
 import URLSearchParamsWrapper from "src/utils/URLSearchParamWrapper";
@@ -63,6 +64,7 @@ import ClearInstance from "./taskInstance/taskActions/ClearInstance";
 import MarkInstanceAs from "./taskInstance/taskActions/MarkInstanceAs";
 import XcomCollection from "./taskInstance/Xcom";
 import TaskDetails from "./task";
+import AuditLog from "./AuditLog";
 
 const dagId = getMetaValue("dag_id")!;
 
@@ -82,11 +84,13 @@ const tabToIndex = (tab?: string) => {
       return 2;
     case "code":
       return 3;
+    case "audit_log":
+      return 4;
     case "logs":
     case "mapped_tasks":
-      return 4;
-    case "xcom":
       return 5;
+    case "xcom":
+      return 6;
     case "details":
     default:
       return 0;
@@ -99,20 +103,23 @@ const indexToTab = (
   isMappedTaskSummary: boolean
 ) => {
   switch (index) {
+    case 0:
+      return "details";
     case 1:
       return "graph";
     case 2:
       return "gantt";
     case 3:
       return "code";
     case 4:
+      return "audit_log";
+    case 5:
       if (isMappedTaskSummary) return "mapped_tasks";
       if (isTaskInstance) return "logs";
       return undefined;
-    case 5:
+    case 6:
       if (isTaskInstance) return "xcom";
       return undefined;
-    case 0:
     default:
       return undefined;
   }
@@ -173,13 +180,16 @@ const Details = ({
   );
 
   useEffect(() => {
-    // Default to graph tab when navigating from a task instance to a group/dag/dagrun
-    const tabCount = runId && taskId && !isGroup ? 5 : 4;
-    if (tabCount === 4 && tabIndex > 3) {
-      if (!runId && taskId) onChangeTab(0);
-      else onChangeTab(1);
-    }
-  }, [runId, taskId, tabIndex, isGroup, onChangeTab]);
+    // Change to graph or task duration tab if the tab is no longer defined
+    if (indexToTab(tabIndex, isTaskInstance, isMappedTaskSummary) === undefined)
+      onChangeTab(showTaskDetails ? 0 : 1);
+  }, [
+    tabIndex,
+    isTaskInstance,
+    isMappedTaskSummary,
+    showTaskDetails,
+    onChangeTab,
+  ]);
 
   const run = dagRuns.find((r) => r.runId === runId);
   const { data: mappedTaskInstance } = useTaskInstance({
@@ -275,6 +285,12 @@ const Details = ({
               Code
             </Text>
           </Tab>
+          <Tab>
+            <MdPlagiarism size={16} />
+            <Text as="strong" ml={1}>
+              Audit Log
+            </Text>
+          </Tab>
           {isTaskInstance && (
             <Tab>
               <MdReorder size={16} />
@@ -359,6 +375,12 @@ const Details = ({
           <TabPanel height="100%">
             <DagCode />
           </TabPanel>
+          <TabPanel height="100%">
+            <AuditLog
+              taskId={isGroup || !taskId ? undefined : taskId}
+              run={run}
+            />
+          </TabPanel>
           {isTaskInstance && run && (
             <TabPanel
               pt={mapIndex !== undefined ? "0px" : undefined}

airflow/www/templates/airflow/dag.html

@@ -80,6 +80,8 @@
   <meta name="dag_source_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_dag_source_endpoint_get_dag_source', file_token='_FILE_TOKEN_') }}">
   <meta name="dag_details_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_dag_endpoint_get_dag_details', dag_id=dag.dag_id) }}">
   <meta name="datasets_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_dataset_endpoint_get_datasets') }}">
+  <meta name="event_logs_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_event_log_endpoint_get_event_logs') }}">
+  <meta name="audit_log_url" content="{{ url_for('LogModelView.list') }}">
 
   <!-- End Urls -->
   <meta name="is_paused" content="{{ dag_is_paused }}">