Dag Run and Overview Tab Display Deadlines

airflow-core/src/airflow/serialization/encoders.py

@@ -162,14 +162,6 @@ def _ensure_serialized(d):
     if isinstance(trigger, dict):
         classpath = trigger["classpath"]
         kwargs = trigger["kwargs"]
-        # unwrap any kwargs that are themselves serialized objects, to avoid double-serialization in the trigger's own serialize() method.
-        unwrapped = {}
-        for k, v in kwargs.items():
-            if isinstance(v, dict) and Encoding.TYPE in v:
-                unwrapped[k] = BaseSerialization.deserialize(v)
-            else:
-                unwrapped[k] = v
-        kwargs = unwrapped
     else:
         classpath, kwargs = trigger.serialize()
     return {

airflow-core/src/airflow/ui/public/i18n/locales/en/dag.json

@@ -7,6 +7,11 @@
   },
   "calendar": {
     "daily": "Daily",
+    "deadlines": {
+      "missed": "Missed",
+      "pending": "Pending",
+      "title": "Deadlines in Period"
+    },
     "hourly": "Hourly",
     "legend": {
       "less": "Less",
@@ -40,6 +45,31 @@
     "parseDuration": "Parse Duration:",
     "parsedAt": "Parsed at:"
   },
+  "deadlineAlerts": {
+    "completionRule": "The run must complete within {{interval}} of {{reference}}",
+    "count_one": "{{count}} deadline",
+    "count_other": "{{count}} deadlines",
+    "referenceType": {
+      "AverageRuntimeDeadline": "average runtime",
+      "DagRunLogicalDateDeadline": "logical date",
+      "DagRunQueuedAtDeadline": "queue time",
+      "FixedDatetimeDeadline": "fixed datetime"
+    },
+    "title": "Deadline Alerts",
+    "unnamed": "Unnamed Alert"
+  },
+  "deadlineStatus": {
+    "completionRule": "Must complete within {{interval}} of {{reference}}",
+    "deadlineIn": "Deadline in {{duration}}",
+    "finishedEarly": "Finished {{duration}} before deadline",
+    "finishedLate": "Finished {{duration}} after deadline",
+    "label": "Deadline",
+    "met": "Met",
+    "missed": "Missed",
+    "stillRunning": "Still running",
+    "upcoming": "Upcoming",
+    "viewAll": "View all {{count}}"
+  },
   "extraLinks": "Extra Links",
   "grid": {
     "buttons": {
@@ -93,6 +123,16 @@
       "assetEvent_one": "Created Asset Event",
       "assetEvent_other": "Created Asset Events"
     },
+    "deadlines": {
+      "completionRule": "Complete within {{interval}} of {{reference}}",
+      "finishedEarly": "Finished {{duration}} before deadline",
+      "finishedLate": "Finished {{duration}} after deadline",
+      "pending": "Pending Deadlines",
+      "recentlyMissed": "Recently Missed Deadlines",
+      "stillRunning": "Still running past deadline",
+      "title": "Deadlines",
+      "viewAll": "View all {{count}}"
+    },
     "failedLogs": {
       "hideLogs": "Hide Logs",
       "showLogs": "Show Logs",

airflow-core/src/airflow/ui/src/pages/Dag/DeadlineAlertsBadge.tsx

@@ -0,0 +1,90 @@
+/*!
+ * 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, Button, Separator, Text, VStack } from "@chakra-ui/react";
+import dayjs from "dayjs";
+import duration from "dayjs/plugin/duration";
+import relativeTime from "dayjs/plugin/relativeTime";
+import { useTranslation } from "react-i18next";
+import { FiClock } from "react-icons/fi";
+
+import { useDeadlinesServiceGetDagDeadlineAlerts } from "openapi/queries";
+import type { DeadlineAlertResponse } from "openapi/requests/types.gen";
+import { Popover } from "src/components/ui";
+
+dayjs.extend(duration);
+dayjs.extend(relativeTime);
+
+const AlertRow = ({ alert }: { readonly alert: DeadlineAlertResponse }) => {
+  const { t: translate } = useTranslation("dag");
+  const reference = translate(`deadlineAlerts.referenceType.${alert.reference_type}`, {
+    defaultValue: alert.reference_type,
+  });
+  const interval = dayjs.duration(alert.interval, "seconds").humanize();
+
+  return (
+    <Box py={2} width="100%">
+      <Text color="fg.muted" fontSize="xs">
+        {translate("deadlineAlerts.completionRule", { interval, reference })}
+        {Boolean(alert.name) && (
+          <Text as="span" color="fg.subtle" fontSize="xs">
+            {" "}
+            ({alert.name})
+          </Text>
+        )}
+      </Text>
+    </Box>
+  );
+};
+
+export const DeadlineAlertsBadge = ({ dagId }: { readonly dagId: string }) => {
+  const { t: translate } = useTranslation("dag");
+
+  const { data } = useDeadlinesServiceGetDagDeadlineAlerts({ dagId });
+
+  const alerts = data?.deadline_alerts ?? [];
+
+  if (alerts.length === 0) {
+    return undefined;
+  }
+
+  return (
+    // eslint-disable-next-line jsx-a11y/no-autofocus
+    <Popover.Root autoFocus={false} lazyMount unmountOnExit>
+      <Popover.Trigger asChild>
+        <Button color="fg.info" size="xs" variant="outline">
+          <FiClock />
+          {translate("deadlineAlerts.count", { count: data?.total_entries ?? alerts.length })}
+        </Button>
+      </Popover.Trigger>
+      <Popover.Content css={{ "--popover-bg": "colors.bg.emphasized" }} maxWidth="360px" width="fit-content">
+        <Popover.Arrow />
+        <Popover.Body>
+          <Text fontWeight="bold" mb={1}>
+            {translate("deadlineAlerts.title")}
+          </Text>
+          <VStack gap={0} separator={<Separator />}>
+            {alerts.map((alert) => (
+              <AlertRow alert={alert} key={alert.id} />
+            ))}
+          </VStack>
+        </Popover.Body>
+      </Popover.Content>
+    </Popover.Root>
+  );
+};

airflow-core/src/airflow/ui/src/pages/Dag/Header.tsx

@@ -35,6 +35,7 @@ import { TogglePause } from "src/components/TogglePause";
 import { DagOwners } from "../DagsList/DagOwners";
 import { DagTags } from "../DagsList/DagTags";
 import { Schedule } from "../DagsList/Schedule";
+import { DeadlineAlertsBadge } from "./DeadlineAlertsBadge";
 
 type LatestRunInfo = {
   dag_id: string;
@@ -123,6 +124,7 @@ export const Header = ({
       actions={
         dag === undefined ? undefined : (
           <>
+            <DeadlineAlertsBadge dagId={dag.dag_id} />
             {dag.doc_md === null ? undefined : (
               <DisplayMarkdownButton
                 header={translate("dagDetails.documentation")}

airflow-core/src/airflow/ui/src/pages/Dag/Overview/AllDeadlinesModal.tsx

@@ -0,0 +1,156 @@
+/*!
+ * 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 { Heading, HStack, Separator, Skeleton, VStack } from "@chakra-ui/react";
+import { useState } from "react";
+
+import { useDagRunServiceGetDagRuns, useDeadlinesServiceGetDeadlines } from "openapi/queries";
+import type { DAGRunResponse, DagRunState, DeadlineAlertResponse } from "openapi/requests/types.gen";
+import { ErrorAlert } from "src/components/ErrorAlert";
+import { Dialog } from "src/components/ui";
+import { Pagination } from "src/components/ui/Pagination";
+
+import { DeadlineRow } from "./DeadlineRow";
+
+const PAGE_LIMIT = 10;
+
+type AllDeadlinesModalProps = {
+  readonly alertMap: Map<string, DeadlineAlertResponse>;
+  readonly dagId: string;
+  readonly endDate: string;
+  readonly missed: boolean;
+  readonly onClose: () => void;
+  readonly open: boolean;
+  readonly refetchInterval: number | false;
+  readonly startDate: string;
+  readonly title: string;
+};
+
+export const AllDeadlinesModal = ({
+  alertMap,
+  dagId,
+  endDate,
+  missed,
+  onClose,
+  open,
+  refetchInterval,
+  startDate,
+  title,
+}: AllDeadlinesModalProps) => {
+  const [page, setPage] = useState(1);
+  const offset = (page - 1) * PAGE_LIMIT;
+
+  const { data, error, isLoading } = useDeadlinesServiceGetDeadlines(
+    missed
+      ? {
+          dagId,
+          dagRunId: "~",
+          lastUpdatedAtGte: startDate,
+          lastUpdatedAtLte: endDate,
+          limit: PAGE_LIMIT,
+          missed: true,
+          offset,
+          orderBy: ["-last_updated_at"],
+        }
+      : {
+          dagId,
+          dagRunId: "~",
+          deadlineTimeGte: endDate,
+          limit: PAGE_LIMIT,
+          missed: false,
+          offset,
+          orderBy: ["deadline_time"],
+        },
+    undefined,
+    { enabled: open, refetchInterval },
+  );
+
+  const { data: runsData } = useDagRunServiceGetDagRuns(
+    { dagId, limit: 100, runAfterGte: startDate, runAfterLte: endDate },
+    undefined,
+    { enabled: open, refetchInterval },
+  );
+
+  const runStateMap = new Map<string, DagRunState>();
+  const runMap = new Map<string, DAGRunResponse>();
+
+  for (const run of runsData?.dag_runs ?? []) {
+    runStateMap.set(run.dag_run_id, run.state);
+    runMap.set(run.dag_run_id, run);
+  }
+
+  const deadlines = data?.deadlines ?? [];
+  const totalEntries = data?.total_entries ?? 0;
+
+  const getAlert = (alertId?: string | null) =>
+    alertId !== undefined && alertId !== null ? alertMap.get(alertId) : undefined;
+
+  const onOpenChange = () => {
+    setPage(1);
+    onClose();
+  };
+
+  return (
+    <Dialog.Root onOpenChange={onOpenChange} open={open} scrollBehavior="inside" size="lg">
+      <Dialog.Content backdrop p={4}>
+        <Dialog.Header>
+          <Heading size="sm">{title}</Heading>
+        </Dialog.Header>
+        <Dialog.CloseTrigger />
+        <Dialog.Body pb={2}>
+          <ErrorAlert error={error} />
+          {isLoading ? (
+            <VStack>
+              {Array.from({ length: PAGE_LIMIT }).map((_, idx) => (
+                // eslint-disable-next-line react/no-array-index-key
+                <Skeleton height="36px" key={idx} width="100%" />
+              ))}
+            </VStack>
+          ) : (
+            <VStack gap={0} separator={<Separator />}>
+              {deadlines.map((dl) => (
+                <DeadlineRow
+                  alert={getAlert(dl.alert_id)}
+                  deadline={dl}
+                  key={dl.id}
+                  run={runMap.get(dl.dag_run_id)}
+                  runState={runStateMap.get(dl.dag_run_id)}
+                />
+              ))}
+            </VStack>
+          )}
+        </Dialog.Body>
+        {totalEntries > PAGE_LIMIT ? (
+          <Pagination.Root
+            count={totalEntries}
+            onPageChange={(event) => setPage(event.page)}
+            p={3}
+            page={page}
+            pageSize={PAGE_LIMIT}
+          >
+            <HStack justify="center">
+              <Pagination.PrevTrigger />
+              <Pagination.Items />
+              <Pagination.NextTrigger />
+            </HStack>
+          </Pagination.Root>
+        ) : undefined}
+      </Dialog.Content>
+    </Dialog.Root>
+  );
+};