Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add asset events to dashboard #44961

Open
wants to merge 5 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions airflow/api_fastapi/core_api/datamodels/assets.py
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,7 @@ class AssetEventResponse(BaseModel):

id: int
asset_id: int
uri: str | None = Field(alias="uri", default=None)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

if we're to add uri here, we'll probably need to add name (and maybe group?)

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I’m not sure we should add uri here. If we need it in the UI, it’d be better to include the entire asset object instead.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is as per the initial UI design for dashboard. I am okay with changing this as per consensus.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The designs were quite preliminary so I am happy with adding everything suggested

extra: dict | None = None
source_task_id: str | None = None
source_dag_id: str | None = None
Expand Down
5 changes: 5 additions & 0 deletions airflow/api_fastapi/core_api/openapi/v1-generated.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -6098,6 +6098,11 @@ components:
asset_id:
type: integer
title: Asset Id
uri:
anyOf:
- type: string
- type: 'null'
title: Uri
extra:
anyOf:
- type: object
Expand Down
11 changes: 11 additions & 0 deletions airflow/ui/openapi-gen/requests/schemas.gen.ts
Original file line number Diff line number Diff line change
Expand Up @@ -180,6 +180,17 @@ export const $AssetEventResponse = {
type: "integer",
title: "Asset Id",
},
uri: {
anyOf: [
{
type: "string",
},
{
type: "null",
},
],
title: "Uri",
},
extra: {
anyOf: [
{
Expand Down
1 change: 1 addition & 0 deletions airflow/ui/openapi-gen/requests/types.gen.ts
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ export type AssetEventCollectionResponse = {
export type AssetEventResponse = {
id: number;
asset_id: number;
uri?: string | null;
extra?: {
[key: string]: unknown;
} | null;
Expand Down
69 changes: 69 additions & 0 deletions airflow/ui/src/pages/Dashboard/HistoricalMetrics/AssetEvent.tsx
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
/*!
* 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";

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 /> <Text> {event.uri ?? ""} </Text>
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Likely better to use the name here, maybe we some rich UI that reveals more asset information on user interaction.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Once we pass name and group we can put more useful information here. And could have a tooltip to show more.

</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>
);
};
111 changes: 111 additions & 0 deletions airflow/ui/src/pages/Dashboard/HistoricalMetrics/AssetEvents.tsx
Original file line number Diff line number Diff line change
@@ -0,0 +1,111 @@
/*!
* 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
collection={assetSortOptions}
data-testid="asset-sort-duration"
defaultValue={["-timestamp"]}
onValueChange={(option) => setAssetSortBy(option.value[0] as string)}
variant="outline" // Flushed seems to be right option to remove border but is not valid as per typescript
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What does this comment mean?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In the UI design the select component has no border in asset events widget. "flushed" is a variant that achieves this effect on select but the type for select only has outline | subtle as valid values though flushed works. This can be kept outline too and would just deviate from the design.

Copy link
Contributor

@bbovenzi bbovenzi Jan 2, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think "flushed" is a valid or at least not a publically-supported variant. So we can remove the comment. And we can add a borderWidth={0} param

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>
);
};
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand All @@ -36,6 +37,7 @@ export const HistoricalMetrics = () => {
now.subtract(Number(defaultHour), "hour").toISOString(),
);
const [endDate, setEndDate] = useState(now.toISOString());
const [assetSortBy, setAssetSortBy] = useState("-timestamp");

const { data, error, isLoading } = useDashboardServiceHistoricalMetrics({
endDate,
Expand Down Expand Up @@ -67,19 +69,31 @@ export const HistoricalMetrics = () => {
setStartDate={setStartDate}
startDate={startDate}
/>
{isLoading ? <MetricSectionSkeleton /> : undefined}
{!isLoading && data !== undefined && (
<Box>
<DagRunMetrics
dagRunStates={data.dag_run_states}
total={dagRunTotal}
<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}
/>
<TaskInstanceMetrics
taskInstanceStates={data.task_instance_states}
total={taskRunTotal}
/>
</Box>
)}
</GridItem>
</SimpleGrid>
</VStack>
</Box>
);
Expand Down
6 changes: 6 additions & 0 deletions tests/api_fastapi/core_api/routes/public/test_assets.py
Original file line number Diff line number Diff line change
Expand Up @@ -542,6 +542,7 @@ def test_should_respond_200(self, test_client, session):
{
"id": 1,
"asset_id": 1,
"uri": "s3://bucket/key/1",
"extra": {"foo": "bar"},
"source_task_id": "source_task_id",
"source_dag_id": "source_dag_id",
Expand All @@ -564,6 +565,7 @@ def test_should_respond_200(self, test_client, session):
{
"id": 2,
"asset_id": 2,
"uri": "s3://bucket/key/2",
"extra": {"foo": "bar"},
"source_task_id": "source_task_id",
"source_dag_id": "source_dag_id",
Expand Down Expand Up @@ -704,6 +706,7 @@ def test_should_mask_sensitive_extra(self, test_client, session):
{
"id": 1,
"asset_id": 1,
"uri": "s3://bucket/key/1",
"extra": {"password": "***"},
"source_task_id": "source_task_id",
"source_dag_id": "source_dag_id",
Expand All @@ -726,6 +729,7 @@ def test_should_mask_sensitive_extra(self, test_client, session):
{
"id": 2,
"asset_id": 2,
"uri": "s3://bucket/key/2",
"extra": {"password": "***"},
"source_task_id": "source_task_id",
"source_dag_id": "source_dag_id",
Expand Down Expand Up @@ -912,6 +916,7 @@ def test_should_respond_200(self, test_client, session):
assert response.json() == {
"id": mock.ANY,
"asset_id": 1,
"uri": "s3://bucket/key/1",
"extra": {"foo": "bar", "from_rest_api": True},
"source_task_id": None,
"source_dag_id": None,
Expand All @@ -938,6 +943,7 @@ def test_should_mask_sensitive_extra(self, test_client, session):
assert response.json() == {
"id": mock.ANY,
"asset_id": 1,
"uri": "s3://bucket/key/1",
"extra": {"password": "***", "from_rest_api": True},
"source_task_id": None,
"source_dag_id": None,
Expand Down
1 change: 1 addition & 0 deletions tests/api_fastapi/core_api/routes/public/test_dag_run.py
Original file line number Diff line number Diff line change
Expand Up @@ -976,6 +976,7 @@ 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,
"source_dag_id": ti.dag_id,
Expand Down