Add asset events to dashboard

airflow/api_fastapi/core_api/datamodels/assets.py

@@ -102,6 +102,9 @@ class AssetEventResponse(BaseModel):
 
     id: int
     asset_id: int
+    uri: str | None = Field(alias="uri", default=None)
+    name: str | None = Field(alias="name", default=None)
+    group: str | None = Field(alias="group", default=None)
     extra: dict | None = None
     source_task_id: str | None = None
     source_dag_id: str | None = None

airflow/api_fastapi/core_api/openapi/v1-generated.yaml

@@ -6460,6 +6460,21 @@ components:
         asset_id:
           type: integer
           title: Asset Id
+        uri:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Uri
+        name:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Name
+        group:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Group
         extra:
           anyOf:
           - type: object

airflow/models/asset.py

@@ -714,6 +714,14 @@ class AssetEvent(Base):
     def uri(self):
         return self.asset.uri
 
+    @property
+    def group(self):
+        return self.asset.group
+
+    @property
+    def name(self):
+        return self.asset.name
+
     def __repr__(self) -> str:
         args = []
         for attr in [

airflow/ui/openapi-gen/requests/schemas.gen.ts

@@ -180,6 +180,39 @@ export const $AssetEventResponse = {
       type: "integer",
       title: "Asset Id",
     },
+    uri: {
+      anyOf: [
+        {
+          type: "string",
+        },
+        {
+          type: "null",
+        },
+      ],
+      title: "Uri",
+    },
+    name: {
+      anyOf: [
+        {
+          type: "string",
+        },
+        {
+          type: "null",
+        },
+      ],
+      title: "Name",
+    },
+    group: {
+      anyOf: [
+        {
+          type: "string",
+        },
+        {
+          type: "null",
+        },
+      ],
+      title: "Group",
+    },
     extra: {
       anyOf: [
         {

airflow/ui/openapi-gen/requests/types.gen.ts

@@ -60,6 +60,9 @@ export type AssetEventCollectionResponse = {
 export type AssetEventResponse = {
   id: number;
   asset_id: number;
+  uri?: string | null;
+  name?: string | null;
+  group?: string | null;
   extra?: {
     [key: string]: unknown;
   } | null;

airflow/ui/src/pages/Dashboard/HistoricalMetrics/AssetEvent.tsx

@@ -0,0 +1,75 @@
+/*!
+ * 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 { Box, Text, HStack } from "@chakra-ui/react";
+import { FiDatabase } from "react-icons/fi";
+import { MdOutlineAccountTree } from "react-icons/md";
+import { Link } from "react-router-dom";
+
+import type { AssetEventResponse } from "openapi/requests/types.gen";
+import Time from "src/components/Time";
+import { Tooltip } from "src/components/ui";
+
+export const AssetEvent = ({ event }: { readonly event: AssetEventResponse }) => {
+  const hasDagRuns = event.created_dagruns.length > 0;
+  const source = event.extra?.from_rest_api === true ? "API" : "";
+
+  return (
+    <Box fontSize={13} mt={1} w="full">
+      <Text fontWeight="bold">
+        <Time datetime={event.timestamp} />
+      </Text>
+      <HStack>
+        <FiDatabase />
+        <Tooltip
+          content={
+            <div>
+              <Text> group: {event.group ?? ""} </Text>
+              <Text> uri: {event.uri ?? ""} </Text>
+            </div>
+          }
+          showArrow
+        >
+          <Text> {event.name ?? ""} </Text>
+        </Tooltip>
+      </HStack>
+      <HStack>
+        <MdOutlineAccountTree /> <Text> Source: </Text>
+        {source === "" ? (
+          <Link
+            to={`/dags/${event.source_dag_id}/runs/${event.source_run_id}/tasks/${event.source_task_id}?map_index=${event.source_map_index}`}
+          >
+            <Text color="fg.info"> {event.source_dag_id} </Text>
+          </Link>
+        ) : (
+          source
+        )}
+      </HStack>
+      <HStack>
+        <Text> Triggered Dag Runs: </Text>
+        {hasDagRuns ? (
+          <Link to={`/dags/${event.created_dagruns[0]?.dag_id}/runs/${event.created_dagruns[0]?.run_id}`}>
+            <Text color="fg.info"> {event.created_dagruns[0]?.dag_id} </Text>
+          </Link>
+        ) : (
+          "~"
+        )}
+      </HStack>
+    </Box>
+  );
+};

airflow/ui/src/pages/Dashboard/HistoricalMetrics/AssetEvents.tsx

@@ -0,0 +1,93 @@
+/*!
+ * 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 { Box, Heading, Flex, HStack, VStack, StackSeparator, Skeleton } from "@chakra-ui/react";
+import { createListCollection } from "@chakra-ui/react/collection";
+
+import { useAssetServiceGetAssetEvents } from "openapi/queries";
+import { MetricsBadge } from "src/components/MetricsBadge";
+import { Select } from "src/components/ui";
+
+import { AssetEvent } from "./AssetEvent";
+
+type AssetEventProps = {
+  readonly assetSortBy: string;
+  readonly endDate: string;
+  readonly setAssetSortBy: React.Dispatch<React.SetStateAction<string>>;
+  readonly startDate: string;
+};
+
+export const AssetEvents = ({ assetSortBy, endDate, setAssetSortBy, startDate }: AssetEventProps) => {
+  const { data, isLoading } = useAssetServiceGetAssetEvents({
+    limit: 6,
+    orderBy: assetSortBy,
+    timestampGte: startDate,
+    timestampLte: endDate,
+  });
+
+  const assetSortOptions = createListCollection({
+    items: [
+      { label: "Newest first", value: "-timestamp" },
+      { label: "Oldest first", value: "timestamp" },
+    ],
+  });
+
+  return (
+    <Box borderRadius={5} borderWidth={1} ml={2} pb={2}>
+      <Flex justify="space-between" mr={1} mt={0} pl={3} pt={1}>
+        <HStack>
+          <MetricsBadge backgroundColor="blue.solid" runs={isLoading ? 0 : data?.total_entries} />
+          <Heading marginEnd="auto" size="md">
+            Asset Events
+          </Heading>
+        </HStack>
+        <Select.Root
+          borderWidth={0}
+          collection={assetSortOptions}
+          data-testid="asset-sort-duration"
+          defaultValue={["-timestamp"]}
+          onValueChange={(option) => setAssetSortBy(option.value[0] as string)}
+          width={130}
+        >
+          <Select.Trigger>
+            <Select.ValueText placeholder="Sort by" />
+          </Select.Trigger>
+
+          <Select.Content>
+            {assetSortOptions.items.map((option) => (
+              <Select.Item item={option} key={option.value[0]}>
+                {option.label}
+              </Select.Item>
+            ))}
+          </Select.Content>
+        </Select.Root>
+      </Flex>
+      {isLoading ? (
+        <VStack px={3} separator={<StackSeparator />}>
+          {Array.from({ length: 5 }, (_, index) => index).map((index) => (
+            <Skeleton height={100} key={index} width="full" />
+          ))}
+        </VStack>
+      ) : (
+        <VStack px={3} separator={<StackSeparator />}>
+          {data?.asset_events.map((event) => <AssetEvent event={event} key={event.id} />)}
+        </VStack>
+      )}
+    </Box>
+  );
+};

airflow/ui/src/pages/Dashboard/HistoricalMetrics/HistoricalMetrics.tsx

@@ -16,14 +16,15 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-import { Box, VStack } from "@chakra-ui/react";
+import { Box, VStack, SimpleGrid, GridItem } from "@chakra-ui/react";
 import dayjs from "dayjs";
 import { useState } from "react";
 
 import { useDashboardServiceHistoricalMetrics } from "openapi/queries";
 import { ErrorAlert } from "src/components/ErrorAlert";
 import TimeRangeSelector from "src/components/TimeRangeSelector";
 
+import { AssetEvents } from "./AssetEvents";
 import { DagRunMetrics } from "./DagRunMetrics";
 import { MetricSectionSkeleton } from "./MetricSectionSkeleton";
 import { TaskInstanceMetrics } from "./TaskInstanceMetrics";
@@ -34,6 +35,7 @@ export const HistoricalMetrics = () => {
   const now = dayjs();
   const [startDate, setStartDate] = useState(now.subtract(Number(defaultHour), "hour").toISOString());
   const [endDate, setEndDate] = useState(now.toISOString());
+  const [assetSortBy, setAssetSortBy] = useState("-timestamp");
 
   const { data, error, isLoading } = useDashboardServiceHistoricalMetrics({
     endDate,
@@ -59,13 +61,25 @@ export const HistoricalMetrics = () => {
           setStartDate={setStartDate}
           startDate={startDate}
         />
-        {isLoading ? <MetricSectionSkeleton /> : undefined}
-        {!isLoading && data !== undefined && (
-          <Box>
-            <DagRunMetrics dagRunStates={data.dag_run_states} total={dagRunTotal} />
-            <TaskInstanceMetrics taskInstanceStates={data.task_instance_states} total={taskRunTotal} />
-          </Box>
-        )}
+        <SimpleGrid columns={{ base: 10 }}>
+          <GridItem colSpan={{ base: 7 }}>
+            {isLoading ? <MetricSectionSkeleton /> : undefined}
+            {!isLoading && data !== undefined && (
+              <Box>
+                <DagRunMetrics dagRunStates={data.dag_run_states} total={dagRunTotal} />
+                <TaskInstanceMetrics taskInstanceStates={data.task_instance_states} total={taskRunTotal} />
+              </Box>
+            )}
+          </GridItem>
+          <GridItem colSpan={{ base: 3 }}>
+            <AssetEvents
+              assetSortBy={assetSortBy}
+              endDate={endDate}
+              setAssetSortBy={setAssetSortBy}
+              startDate={startDate}
+            />
+          </GridItem>
+        </SimpleGrid>
       </VStack>
     </Box>
   );

tests/api_fastapi/core_api/routes/public/test_assets.py

@@ -542,7 +542,10 @@ def test_should_respond_200(self, test_client, session):
                 {
                     "id": 1,
                     "asset_id": 1,
+                    "uri": "s3://bucket/key/1",
                     "extra": {"foo": "bar"},
+                    "group": "asset",
+                    "name": "simple1",
                     "source_task_id": "source_task_id",
                     "source_dag_id": "source_dag_id",
                     "source_run_id": "source_run_id_1",
@@ -564,6 +567,9 @@ def test_should_respond_200(self, test_client, session):
                 {
                     "id": 2,
                     "asset_id": 2,
+                    "uri": "s3://bucket/key/2",
+                    "group": "asset",
+                    "name": "simple2",
                     "extra": {"foo": "bar"},
                     "source_task_id": "source_task_id",
                     "source_dag_id": "source_dag_id",
@@ -704,6 +710,9 @@ def test_should_mask_sensitive_extra(self, test_client, session):
                 {
                     "id": 1,
                     "asset_id": 1,
+                    "uri": "s3://bucket/key/1",
+                    "group": "asset",
+                    "name": "sensitive1",
                     "extra": {"password": "***"},
                     "source_task_id": "source_task_id",
                     "source_dag_id": "source_dag_id",
@@ -726,6 +735,9 @@ def test_should_mask_sensitive_extra(self, test_client, session):
                 {
                     "id": 2,
                     "asset_id": 2,
+                    "uri": "s3://bucket/key/2",
+                    "group": "asset",
+                    "name": "sensitive2",
                     "extra": {"password": "***"},
                     "source_task_id": "source_task_id",
                     "source_dag_id": "source_dag_id",
@@ -912,6 +924,9 @@ def test_should_respond_200(self, test_client, session):
         assert response.json() == {
             "id": mock.ANY,
             "asset_id": 1,
+            "uri": "s3://bucket/key/1",
+            "group": "asset",
+            "name": "simple1",
             "extra": {"foo": "bar", "from_rest_api": True},
             "source_task_id": None,
             "source_dag_id": None,
@@ -938,6 +953,9 @@ def test_should_mask_sensitive_extra(self, test_client, session):
         assert response.json() == {
             "id": mock.ANY,
             "asset_id": 1,
+            "uri": "s3://bucket/key/1",
+            "group": "asset",
+            "name": "simple1",
             "extra": {"password": "***", "from_rest_api": True},
             "source_task_id": None,
             "source_dag_id": None,

tests/api_fastapi/core_api/routes/public/test_dag_run.py

@@ -1016,8 +1016,11 @@ def test_should_respond_200(self, test_client, dag_maker, session):
                 {
                     "timestamp": from_datetime_to_zulu(event.timestamp),
                     "asset_id": asset1_id,
+                    "uri": "file:///da1",
                     "extra": {},
                     "id": event.id,
+                    "group": "asset",
+                    "name": "ds1",
                     "source_dag_id": ti.dag_id,
                     "source_map_index": ti.map_index,
                     "source_run_id": ti.run_id,