Skip to content

Commit

Permalink
Add autorefresh to dag page (apache#46296)
Browse files Browse the repository at this point in the history 
* Add autorefresh to dag page

* remove areActiveRuns check from task list card

* Fix dag header toggle pause

* Check if dag is paused, move to custom hook

* Address PR feedback
  • Loading branch information
@bbovenzi
bbovenzi authored and pratiksha rajendrabhai badheka committed Jan 31, 2025
1 parent 4cfd529 commit 134e430
Show file tree
Hide file tree
Showing 18 changed files with 336 additions and 284 deletions.
7 changes: 3 additions & 4 deletions airflow/ui/src/components/TaskTrySelect.tsx
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,7 @@ import { Button, createListCollection, HStack, VStack, Heading } from "@chakra-u
import { useTaskInstanceServiceGetMappedTaskInstanceTries } from "openapi/queries";
import type { TaskInstanceHistoryResponse, TaskInstanceResponse } from "openapi/requests/types.gen";
import { StateBadge } from "src/components/StateBadge";
import { useConfig } from "src/queries/useConfig";
import { isStatePending } from "src/utils/refresh";
import { isStatePending, useAutoRefresh } from "src/utils";

import TaskInstanceTooltip from "./TaskInstanceTooltip";
import { Select } from "./ui";
Expand All @@ -43,7 +42,7 @@ export const TaskTrySelect = ({ onSelectTryNumber, selectedTryNumber, taskInstan
try_number: finalTryNumber,
} = taskInstance;

const autoRefreshInterval = useConfig("auto_refresh_interval") as number;
const refetchInterval = useAutoRefresh({ dagId });

const { data: tiHistory } = useTaskInstanceServiceGetMappedTaskInstanceTries(
{
Expand All @@ -59,7 +58,7 @@ export const TaskTrySelect = ({ onSelectTryNumber, selectedTryNumber, taskInstan
// We actually want to use || here
// eslint-disable-next-line @typescript-eslint/prefer-nullish-coalescing
query.state.data?.task_instances.some((ti) => isStatePending(ti.state)) || isStatePending(state)
? autoRefreshInterval * 1000
? refetchInterval
: false,
},
);
Expand Down
4 changes: 2 additions & 2 deletions airflow/ui/src/components/TriggerDag/TriggerDAGForm.tsx
View file
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ const TriggerDAGForm = ({ dagId, onClose, open }: TriggerDAGFormProps) => {
error: errorTrigger,
isPending,
triggerDagRun,
} = useTrigger({ onSuccessConfirm: onClose });
} = useTrigger({ dagId, onSuccessConfirm: onClose });
const { conf, setConf } = useParamStore();

const { control, handleSubmit, reset, watch } = useForm<DagRunTriggerParams>({
Expand Down Expand Up @@ -85,7 +85,7 @@ const TriggerDAGForm = ({ dagId, onClose, open }: TriggerDAGFormProps) => {
const dataIntervalEnd = watch("dataIntervalEnd");

const onSubmit = (data: DagRunTriggerParams) => {
triggerDagRun(dagId, data);
triggerDagRun(data);
};

const validateAndPrettifyJson = (value: string) => {
Expand Down
8 changes: 4 additions & 4 deletions airflow/ui/src/constants/sortParams.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,14 +24,14 @@ export const dagSortOptions = createListCollection({
{ label: "Sort by Display Name (Z-A)", value: "-dag_display_name" },
{ label: "Sort by Next DAG Run (Earliest-Latest)", value: "next_dagrun" },
{ label: "Sort by Next DAG Run (Latest-Earliest)", value: "-next_dagrun" },
{ label: "Sort by Last Run State (A-Z)", value: "last_run_state" },
{ label: "Sort by Last Run State (Z-A)", value: "-last_run_state" },
{ label: "Sort by Latest Run State (A-Z)", value: "last_run_state" },
{ label: "Sort by Latest Run State (Z-A)", value: "-last_run_state" },
{
label: "Sort by Last Run Start Date (Earliest-Latest)",
label: "Sort by Latest Run Start Date (Earliest-Latest)",
value: "last_run_start_date",
},
{
label: "Sort by Last Run Start Date (Latest-Earliest)",
label: "Sort by Latest Run Start Date (Latest-Earliest)",
value: "-last_run_start_date",
},
],
Expand Down
19 changes: 17 additions & 2 deletions airflow/ui/src/pages/Dag/Dag.tsx
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import { useParams } from "react-router-dom";

import { useDagServiceGetDagDetails, useDagsServiceRecentDagRuns } from "openapi/queries";
import { DetailsLayout } from "src/layouts/Details/DetailsLayout";
import { isStatePending, useAutoRefresh } from "src/utils";

import { Header } from "./Header";

Expand All @@ -42,20 +43,34 @@ export const Dag = () => {
dagId,
});

const refetchInterval = useAutoRefresh({ dagId });

// TODO: replace with with a list dag runs by dag id request
const {
data: runsData,
error: runsError,
isLoading: isLoadingRuns,
} = useDagsServiceRecentDagRuns({ dagIds: [dagId] }, undefined, {
enabled: Boolean(dagId),
refetchInterval: (query) =>
query.state.data?.dags
.find((recentDag) => recentDag.dag_id === dagId)
?.latest_dag_runs.some((run) => isStatePending(run.state))
? refetchInterval
: false,
});

const runs = runsData?.dags.find((dagWithRuns) => dagWithRuns.dag_id === dagId)?.latest_dag_runs ?? [];
const dagWithRuns = runsData?.dags.find((recentDag) => recentDag.dag_id === dagId);

return (
<DetailsLayout dag={dag} error={error ?? runsError} isLoading={isLoading || isLoadingRuns} tabs={tabs}>
<Header dag={dag} dagId={dagId} latestRun={runs[0]} />
<Header
dag={dag}
dagWithRuns={dagWithRuns}
isRefreshing={Boolean(
dagWithRuns?.latest_dag_runs.some((dr) => isStatePending(dr.state)) && Boolean(refetchInterval),
)}
/>
</DetailsLayout>
);
};
174 changes: 93 additions & 81 deletions airflow/ui/src/pages/Dag/Header.tsx
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,10 +16,11 @@
* specific language governing permissions and limitations
* under the License.
*/
import { Box, Flex, Heading, HStack, SimpleGrid, Text } from "@chakra-ui/react";
import { Box, Flex, Heading, HStack, SimpleGrid, Spinner, Text } from "@chakra-ui/react";
import { FiBookOpen, FiCalendar } from "react-icons/fi";
import { useParams } from "react-router-dom";

import type { DAGDetailsResponse, DAGRunResponse } from "openapi/requests/types.gen";
import type { DAGDetailsResponse, DAGWithLatestDagRunsResponse } from "openapi/requests/types.gen";
import { DagIcon } from "src/assets/DagIcon";
import DagRunInfo from "src/components/DagRunInfo";
import DisplayMarkdownButton from "src/components/DisplayMarkdownButton";
Expand All @@ -33,87 +34,98 @@ import { DagTags } from "../DagsList/DagTags";

export const Header = ({
dag,
dagId,
latestRun,
dagWithRuns,
isRefreshing,
}: {
readonly dag?: DAGDetailsResponse;
readonly dagId?: string;
readonly latestRun?: DAGRunResponse;
}) => (
<Box borderColor="border" borderRadius={8} borderWidth={1} p={2}>
<Box p={2}>
<Flex alignItems="center" justifyContent="space-between">
<HStack alignItems="center" gap={2}>
<DagIcon height={8} width={8} />
<Heading size="lg">{dag?.dag_display_name ?? dagId}</Heading>
{dag !== undefined && (
<TogglePause dagDisplayName={dag.dag_display_name} dagId={dag.dag_id} isPaused={dag.is_paused} />
)}
</HStack>
<Flex>
{dag ? (
<HStack>
{dag.doc_md === null ? undefined : (
<DisplayMarkdownButton
header="Dag Documentation"
icon={<FiBookOpen />}
mdContent={dag.doc_md}
text="Dag Docs"
/>
)}
<ParseDag dagId={dag.dag_id} fileToken={dag.file_token} />
<TriggerDAGButton dag={dag} />
</HStack>
) : undefined}
readonly dagWithRuns?: DAGWithLatestDagRunsResponse;
readonly isRefreshing?: boolean;
}) => {
// We would still like to show the dagId even if the dag object hasn't loaded yet
const { dagId } = useParams();
const latestRun = dagWithRuns?.latest_dag_runs ? dagWithRuns.latest_dag_runs[0] : undefined;

return (
<Box borderColor="border" borderRadius={8} borderWidth={1} p={2}>
<Box p={2}>
<Flex alignItems="center" justifyContent="space-between">
<HStack alignItems="center" gap={2}>
<DagIcon height={8} width={8} />
<Heading size="lg">{dag?.dag_display_name ?? dagId}</Heading>
{dag !== undefined && (
<TogglePause
dagDisplayName={dag.dag_display_name}
dagId={dag.dag_id}
isPaused={dag.is_paused}
/>
)}
{isRefreshing ? <Spinner /> : <div />}
</HStack>
<Flex>
{dag ? (
<HStack>
{dag.doc_md === null ? undefined : (
<DisplayMarkdownButton
header="Dag Documentation"
icon={<FiBookOpen />}
mdContent={dag.doc_md}
text="Dag Docs"
/>
)}
<ParseDag dagId={dag.dag_id} fileToken={dag.file_token} />
<TriggerDAGButton dag={dag} />
</HStack>
) : undefined}
</Flex>
</Flex>
<SimpleGrid columns={4} gap={4} my={2}>
<Stat label="Schedule">
{Boolean(dag?.timetable_summary) ? (
<Tooltip content={dag?.timetable_description}>
<Text fontSize="sm">
<FiCalendar style={{ display: "inline" }} /> {dag?.timetable_summary}
</Text>
</Tooltip>
) : undefined}
</Stat>
<Stat label="Latest Run">
{Boolean(latestRun) && latestRun !== undefined ? (
<DagRunInfo
dataIntervalEnd={latestRun.data_interval_end}
dataIntervalStart={latestRun.data_interval_start}
endDate={latestRun.end_date}
startDate={latestRun.start_date}
state={latestRun.state}
/>
) : undefined}
</Stat>
<Stat label="Next Run">
{Boolean(dagWithRuns?.next_dagrun) ? (
<DagRunInfo
dataIntervalEnd={dagWithRuns?.next_dagrun_data_interval_end}
dataIntervalStart={dagWithRuns?.next_dagrun_data_interval_start}
nextDagrunCreateAfter={dagWithRuns?.next_dagrun_create_after}
/>
) : undefined}
</Stat>
<div />
<div />
</SimpleGrid>
</Box>
<Flex
alignItems="center"
bg="bg.muted"
borderTopColor="border"
borderTopWidth={1}
color="fg.subtle"
fontSize="sm"
justifyContent="space-between"
px={2}
py={1}
>
<Text>Owner: {dag?.owners.join(", ")}</Text>
<DagTags tags={dag?.tags ?? []} />
</Flex>
<SimpleGrid columns={4} gap={4} my={2}>
<Stat label="Schedule">
{Boolean(dag?.timetable_summary) ? (
<Tooltip content={dag?.timetable_description}>
<Text fontSize="sm">
<FiCalendar style={{ display: "inline" }} /> {dag?.timetable_summary}
</Text>
</Tooltip>
) : undefined}
</Stat>
<Stat label="Last Run">
{Boolean(latestRun) && latestRun !== undefined ? (
<DagRunInfo
dataIntervalEnd={latestRun.data_interval_end}
dataIntervalStart={latestRun.data_interval_start}
endDate={latestRun.end_date}
startDate={latestRun.start_date}
state={latestRun.state}
/>
) : undefined}
</Stat>
<Stat label="Next Run">
{Boolean(dag?.next_dagrun) && dag !== undefined ? (
<DagRunInfo
dataIntervalEnd={dag.next_dagrun_data_interval_end}
dataIntervalStart={dag.next_dagrun_data_interval_start}
nextDagrunCreateAfter={dag.next_dagrun_create_after}
/>
) : undefined}
</Stat>
<div />
<div />
</SimpleGrid>
</Box>
<Flex
alignItems="center"
bg="bg.muted"
borderTopColor="border"
borderTopWidth={1}
color="fg.subtle"
fontSize="sm"
justifyContent="space-between"
px={2}
py={1}
>
<Text>Owner: {dag?.owners.join(", ")}</Text>
<DagTags tags={dag?.tags ?? []} />
</Flex>
</Box>
);
);
};
15 changes: 10 additions & 5 deletions airflow/ui/src/pages/Dag/Runs/Runs.tsx
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ import { RunTypeIcon } from "src/components/RunTypeIcon";
import { StateBadge } from "src/components/StateBadge";
import Time from "src/components/Time";
import { Select } from "src/components/ui";
import { capitalize, getDuration } from "src/utils";
import { capitalize, getDuration, useAutoRefresh, isStatePending } from "src/utils";

const columns: Array<ColumnDef<DAGRunResponse>> = [
{
Expand Down Expand Up @@ -125,11 +125,13 @@ export const Runs = () => {
const { setTableURLState, tableURLState } = useTableURLState();
const { pagination, sorting } = tableURLState;
const [sort] = sorting;
const orderBy = sort ? `${sort.desc ? "-" : ""}${sort.id}` : "-start_date";
const orderBy = sort ? `${sort.desc ? "-" : ""}${sort.id}` : "-logical_date";

const filteredState = searchParams.get(STATE_PARAM);

const { data, error, isFetching, isLoading } = useDagRunServiceGetDagRuns(
const refetchInterval = useAutoRefresh({ dagId });

const { data, error, isLoading } = useDagRunServiceGetDagRuns(
{
dagId: dagId ?? "~",
limit: pagination.pageSize,
Expand All @@ -138,7 +140,11 @@ export const Runs = () => {
state: filteredState === null ? undefined : [filteredState],
},
undefined,
{ enabled: !isNaN(pagination.pageSize) },
{
enabled: !isNaN(pagination.pageSize),
refetchInterval: (query) =>
query.state.data?.dag_runs.some((run) => isStatePending(run.state)) ? refetchInterval : false,
},
);

const handleStateChange = useCallback(
Expand Down Expand Up @@ -197,7 +203,6 @@ export const Runs = () => {
data={data?.dag_runs ?? []}
errorMessage={<ErrorAlert error={error} />}
initialState={tableURLState}
isFetching={isFetching}
isLoading={isLoading}
modelName="Dag Run"
onStateChange={setTableURLState}
Expand Down
Loading

0 comments on commit 134e430

Please sign in to comment.