diff --git a/frontend/app/src/next/components/runs/run-dag/dag-run-visualizer.tsx b/frontend/app/src/next/components/runs/run-dag/dag-run-visualizer.tsx index 5b84e8e50..025cca5b9 100644 --- a/frontend/app/src/next/components/runs/run-dag/dag-run-visualizer.tsx +++ b/frontend/app/src/next/components/runs/run-dag/dag-run-visualizer.tsx @@ -16,6 +16,7 @@ import { RunDetailProvider, useRunDetail } from '@/next/hooks/use-run-detail'; import { cn } from '@/lib/utils'; import { ChevronDownIcon, ChevronUpIcon } from '@radix-ui/react-icons'; import { Button } from '@/next/components/ui/button'; +import { Fullscreen } from 'lucide-react'; const connectionLineStyleDark = { stroke: '#fff' }; const connectionLineStyleLight = { stroke: '#000' }; @@ -228,6 +229,7 @@ function WorkflowRunVisualizerContent({ if (!reactFlowInstance.current) { return; } + const node = layoutedNodes?.find( (n: Node) => n.data.taskRun?.taskExternalId === selectedTaskId, ); @@ -239,6 +241,8 @@ function WorkflowRunVisualizerContent({ duration: 800, }); lastCenteredTaskId.current = selectedTaskId; + } else { + reactFlowInstance.current.fitView(); } }, 1); }, [selectedTaskId, layoutedNodes]); @@ -311,20 +315,30 @@ function WorkflowRunVisualizerContent({ } snapToGrid={true} /> - +
+ + +
); } diff --git a/frontend/app/src/next/components/runs/run-id.tsx b/frontend/app/src/next/components/runs/run-id.tsx index eb6453bcc..d6546b164 100644 --- a/frontend/app/src/next/components/runs/run-id.tsx +++ b/frontend/app/src/next/components/runs/run-id.tsx @@ -49,10 +49,13 @@ export function RunId({ }, ); + const displayNameIdPrefix = splitTime(displayName); + const friendlyDisplayName = displayNameIdPrefix || displayName; + const name = isTaskRun ? getFriendlyTaskRunId(taskRun) : displayName && id - ? splitTime(displayName) + '-' + id.split('-')[0] + ? friendlyDisplayName + '-' + id.split('-')[0] : getFriendlyWorkflowRunId(wfRun); const handleDoubleClick = () => { diff --git a/frontend/app/src/next/components/waterfall/waterfall.tsx b/frontend/app/src/next/components/waterfall/waterfall.tsx index 3056658f7..0cdc05759 100644 --- a/frontend/app/src/next/components/waterfall/waterfall.tsx +++ b/frontend/app/src/next/components/waterfall/waterfall.tsx @@ -1,4 +1,4 @@ -import { useState, useMemo } from 'react'; +import { useState, useMemo, useCallback } from 'react'; import { Bar, BarChart, @@ -9,7 +9,7 @@ import { ResponsiveContainer, Cell, } from 'recharts'; -import { ChevronDown, ChevronRight } from 'lucide-react'; +import { ChevronDown, ChevronRight, Drill, Loader } from 'lucide-react'; import { ChartContainer, ChartTooltipContent } from '@/components/ui/chart'; import { V1TaskStatus, V1TaskTiming } from '@/lib/api'; @@ -19,15 +19,19 @@ import { ROUTES } from '@/next/lib/routes'; import { useRunDetail } from '@/next/hooks/use-run-detail'; import { Button } from '../ui/button'; import { RunId } from '../runs/run-id'; -import { - BsArrowDownRightCircle, - BsCircle, - BsArrowUpLeftCircle, -} from 'react-icons/bs'; +import { BsArrowUpLeftCircle } from 'react-icons/bs'; import { Skeleton } from '../ui/skeleton'; import { useCurrentTenantId } from '@/next/hooks/use-tenant'; +import { + TooltipProvider, + Tooltip as BaseTooltip, + TooltipContent, + TooltipTrigger, +} from '../ui/tooltip'; + interface ProcessedTaskData { id: string; + taskExternalId: string; workflowRunId?: string; taskDisplayName: string; parentId?: string; @@ -35,8 +39,8 @@ interface ProcessedTaskData { depth: number; isExpanded: boolean; offset: number; - queuedDuration: number; - ranDuration: number; + queuedDuration: number | null; + ranDuration: number | null; status: V1TaskStatus; taskId: number; // Added for tie-breaking attempt: number; @@ -503,9 +507,9 @@ export function Waterfall({ } // Find the global minimum time (queuedAt or startedAt) among visible tasks - let globalMinTime = Number.MAX_SAFE_INTEGER; - visibleTasks.forEach((id) => { + const globalMinTime = [...visibleTasks].reduce((acc, id) => { const task = taskMap.get(id); + if (task) { // Use queuedAt if available, otherwise use startedAt const minTime = task.queuedAt @@ -514,25 +518,32 @@ export function Waterfall({ ? new Date(task.startedAt).getTime() : null; - if (minTime !== null && minTime < globalMinTime) { - globalMinTime = minTime; + if (minTime !== null && minTime < acc) { + return minTime; } } - }); + + return acc; + }, Number.MAX_SAFE_INTEGER); // Create the processed data for rendering - const data = Array.from(visibleTasks) + const data = [...visibleTasks] .map((id) => { const task = taskMap.get(id); - if (!task || !task.startedAt) { + if (!task) { return null; } // Handle missing queuedAt by defaulting to startedAt (no queue time) const queuedAt = task.queuedAt ? new Date(task.queuedAt).getTime() - : new Date(task.startedAt).getTime(); - const startedAt = new Date(task.startedAt).getTime(); + : task.startedAt + ? new Date(task.startedAt).getTime() + : new Date(task.taskInsertedAt).getTime(); + + const startedAt = task.startedAt + ? new Date(task.startedAt).getTime() + : null; // For running tasks, always use current time as finishedAt const now = new Date().getTime(); @@ -545,6 +556,7 @@ export function Waterfall({ return { id: task.metadata.id, + taskExternalId: task.taskExternalId, taskDisplayName: task.taskDisplayName, parentId: task.parentTaskExternalId, hasChildren: taskHasChildrenMap.get(task.metadata.id) || false, @@ -554,8 +566,13 @@ export function Waterfall({ // Chart data offset: (queuedAt - globalMinTime) / 1000, // in seconds // If queuedAt equals startedAt (due to our fallback logic), then queuedDuration will be 0 - queuedDuration: task.queuedAt ? (startedAt - queuedAt) / 1000 : 0, // in seconds - ranDuration: (finishedAt - startedAt) / 1000, // in seconds + queuedDuration: startedAt + ? task.queuedAt + ? (startedAt - queuedAt) / 1000 + : 0 + : null, // in seconds + ranDuration: + startedAt && finishedAt ? (finishedAt - startedAt) / 1000 : null, // in seconds status: task.status, taskId: task.taskId, // Add taskId for tie-breaking in sorting attempt: task.attempt || 1, @@ -583,159 +600,52 @@ export function Waterfall({ return { data, taskPathMap: new Map() }; }, [taskData, expandedTasks, autoExpandedInitially, taskRelationships]); // Only recompute when dependencies change - // Custom tick renderer with expand/collapse buttons - const renderTick = (props: { - x: number; - y: number; - payload: { value: string }; - }) => { - const { x, y, payload } = props; - const task = processedData.data.find( - (t) => t.taskDisplayName === payload.value, - ); - if (!task) { - // Return empty element instead of null - return ; + // Handler for bar click events + const handleBarClick = (data: any) => { + if (data && data.id) { + // Handle task selection for sidebar + if (handleTaskSelect) { + handleTaskSelect(data.id, data.workflowRunId); + } + + // Handle expansion if the task has children + if (data.hasChildren) { + openTask(data.id, data.depth); + } } - - const indentation = task.depth * 12; // 12px indentation per level - - return ( - - -
- {/* Expand/collapse button */} -
- task.hasChildren && - toggleTask(task.id, task.hasChildren, task.depth) - } - > - {task.hasChildren && - (task.isExpanded ? ( - - ) : ( - - ))} -
- - {/* Task label */} -
handleBarClick(task)} - > - handleBarClick(task)} - className={task.id === selectedTaskId ? 'underline' : ''} - attempt={task.attempt} - /> -
- {workflowRunId === task.workflowRunId ? ( - task.parentId ? ( - e.stopPropagation()} - > - - - ) : ( - - ) - ) : ( - - - - )} -
- - - ); }; + const renderTick = useCallback( + (props: { x: number; y: number; payload: { value: string } }) => { + const { x, y, payload } = props; + + return ( + + ); + }, + [workflowRunId, selectedTaskId, handleBarClick, toggleTask, processedData], + ); + // Handle loading or error states if ( - isLoading || - isError || - !processedData.data || - processedData.data.length === 0 + !isLoading && + (isError || !processedData.data || processedData.data.length === 0) ) { - return ( - <> - - - ); + return null; + } + + if (isLoading) { + return ; } // Compute dynamic chart height @@ -756,21 +666,6 @@ export function Waterfall({ }, }; - // Handler for bar click events - const handleBarClick = (data: any) => { - if (data && data.id) { - // Handle task selection for sidebar - if (handleTaskSelect) { - handleTaskSelect(data.id, data.workflowRunId); - } - - // Handle expansion if the task has children - if (data.hasChildren) { - openTask(data.id, data.depth); - } - } - }; - return ( ); } + +const Tick = ({ + x, + y, + payload, + workflowRunId, + selectedTaskId, + handleBarClick, + toggleTask, + processedData, + tenantId, +}: { + x: number; + y: number; + payload: { value: string }; + workflowRunId: string; + selectedTaskId?: string; + handleBarClick: (task: ProcessedTaskData) => void; + toggleTask: (taskId: string, hasChildren: boolean, taskDepth: number) => void; + processedData: ProcessedData; + tenantId: string; +}) => { + const task = processedData.data.find( + (t) => t.taskDisplayName === payload.value, + ); + if (!task) { + // Return empty element instead of null + return ; + } + + const indentation = task.depth * 12; // 12px indentation per level + + return ( + + +
+ {/* Expand/collapse button */} +
+ task.hasChildren && + toggleTask(task.id, task.hasChildren, task.depth) + } + > + {task.hasChildren && + (task.isExpanded ? ( + + ) : ( + + ))} +
+ + {/* Task label */} +
handleBarClick(task)} + > + handleBarClick(task)} + className={task.id === selectedTaskId ? 'underline' : ''} + attempt={task.attempt} + /> +
+ {workflowRunId === task.workflowRunId && + task.taskExternalId === workflowRunId && + task.parentId && ( + e.stopPropagation()} + > + + + )} + {task.hasChildren && ( + + + + )} + {task.queuedDuration === null && ( + + + + + + This task has not started + + + )} +
+ + + ); +}; diff --git a/frontend/app/src/next/pages/authenticated/dashboard/runs/detail-sheet/run-detail-sheet.tsx b/frontend/app/src/next/pages/authenticated/dashboard/runs/detail-sheet/run-detail-sheet.tsx index 27c66b8a7..6a01c7b49 100644 --- a/frontend/app/src/next/pages/authenticated/dashboard/runs/detail-sheet/run-detail-sheet.tsx +++ b/frontend/app/src/next/pages/authenticated/dashboard/runs/detail-sheet/run-detail-sheet.tsx @@ -80,7 +80,7 @@ function RunDetailSheetContent() { return ( <>
-
+
@@ -119,7 +119,7 @@ function RunDetailSheetContent() {
-
+
data?.run, [data]); - const tasks = useMemo(() => data?.tasks, [data]); + const workflow = data?.run; + const tasks = data?.tasks; const { open: openSheet, sheet } = useSideSheet(); @@ -203,38 +206,11 @@ function RunDetailPageContent({ workflowRunId }: RunDetailPageProps) {
); } - const Timing = () => { - const timings: JSX.Element[] = [ - - Created - - , - - Started - - , - - Duration - - - - , - ]; - return ( - - {timings} - - ); - }; return ( - }> + }>
@@ -368,3 +344,31 @@ function RunDetailPageContent({ workflowRunId }: RunDetailPageProps) { ); } + +const Timing = ({ workflow }: { workflow: V1WorkflowRun }) => { + const timings: JSX.Element[] = [ + + Created + + , + + Started + + , + + Duration + + + + , + ]; + return ( + + {timings} + + ); +}; diff --git a/pkg/repository/v1/olap.go b/pkg/repository/v1/olap.go index fbc2f8928..c60039240 100644 --- a/pkg/repository/v1/olap.go +++ b/pkg/repository/v1/olap.go @@ -1392,6 +1392,8 @@ func (r *OLAPRepositoryImpl) GetTaskTimings(ctx context.Context, tenantId string // start out by getting a list of task external ids for the workflow run id rootTaskExternalIds := make([]pgtype.UUID, 0) + sevenDaysAgo := time.Now().Add(-time.Hour * 24 * 7) + minInsertedAt := time.Now() rootTasks, err := r.queries.FlattenTasksByExternalIds(ctx, r.readPool, sqlcv1.FlattenTasksByExternalIdsParams{ Externalids: []pgtype.UUID{workflowRunId}, @@ -1404,12 +1406,24 @@ func (r *OLAPRepositoryImpl) GetTaskTimings(ctx context.Context, tenantId string for _, task := range rootTasks { rootTaskExternalIds = append(rootTaskExternalIds, task.ExternalID) + + if task.InsertedAt.Time.Before(minInsertedAt) { + minInsertedAt = task.InsertedAt.Time + } + } + + // Setting the maximum lookback period to 7 days + // to prevent scanning a zillion partitions on the tasks, + // runs, and dags tables. + if minInsertedAt.Before(sevenDaysAgo) { + minInsertedAt = sevenDaysAgo } runsList, err := r.queries.GetRunsListRecursive(ctx, r.readPool, sqlcv1.GetRunsListRecursiveParams{ Taskexternalids: rootTaskExternalIds, Tenantid: sqlchelpers.UUIDFromStr(tenantId), Depth: depth, + Createdafter: sqlchelpers.TimestamptzFromTime(minInsertedAt), }) if err != nil { diff --git a/pkg/repository/v1/sqlcv1/olap.sql b/pkg/repository/v1/sqlcv1/olap.sql index 1bf782ebb..df370f950 100644 --- a/pkg/repository/v1/sqlcv1/olap.sql +++ b/pkg/repository/v1/sqlcv1/olap.sql @@ -1321,6 +1321,8 @@ WITH RECURSIVE all_runs AS ( WHERE r.tenant_id = @tenantId::uuid AND ar.depth < @depth::int + AND r.inserted_at >= @createdAfter::timestamptz + AND t.inserted_at >= @createdAfter::timestamptz ) SELECT tenant_id, diff --git a/pkg/repository/v1/sqlcv1/olap.sql.go b/pkg/repository/v1/sqlcv1/olap.sql.go index dd5c290bc..fd008d2be 100644 --- a/pkg/repository/v1/sqlcv1/olap.sql.go +++ b/pkg/repository/v1/sqlcv1/olap.sql.go @@ -317,6 +317,8 @@ WITH RECURSIVE all_runs AS ( WHERE r.tenant_id = $1::uuid AND ar.depth < $3::int + AND r.inserted_at >= $4::timestamptz + AND t.inserted_at >= $4::timestamptz ) SELECT tenant_id, @@ -332,9 +334,10 @@ WHERE ` type GetRunsListRecursiveParams struct { - Tenantid pgtype.UUID `json:"tenantid"` - Taskexternalids []pgtype.UUID `json:"taskexternalids"` - Depth int32 `json:"depth"` + Tenantid pgtype.UUID `json:"tenantid"` + Taskexternalids []pgtype.UUID `json:"taskexternalids"` + Depth int32 `json:"depth"` + Createdafter pgtype.Timestamptz `json:"createdafter"` } type GetRunsListRecursiveRow struct { @@ -347,7 +350,12 @@ type GetRunsListRecursiveRow struct { } func (q *Queries) GetRunsListRecursive(ctx context.Context, db DBTX, arg GetRunsListRecursiveParams) ([]*GetRunsListRecursiveRow, error) { - rows, err := db.Query(ctx, getRunsListRecursive, arg.Tenantid, arg.Taskexternalids, arg.Depth) + rows, err := db.Query(ctx, getRunsListRecursive, + arg.Tenantid, + arg.Taskexternalids, + arg.Depth, + arg.Createdafter, + ) if err != nil { return nil, err }