[ui] Fix endless log loading (#26446)

## Summary & Motivation

We received a bug report from an org with some very, very long log output on some of their runs. Specifically, hundreds of thousands of large logs for individual runs. I noticed two issues during testing:

- When a chunk of logs loads and indicates that there are more older logs left to stream, we consider that a "loading" state and simply don't show any logs in the logs table at all.
  - This is unnecessary, since we're streaming logs in. Instead, just show the loading state if there are no logs available yet at all.
- First, we're using the spread-to-concat antipattern in the subscription handler reducer, which means iterating over the entire existing list of nodes before adding the newly received logs.
  - To resolve this, I'm changing the state array to track *chunks* of logs instead of flattened logs. This means we can more efficiently push chunks into state without mutating state or taking the time to copy entire chunks, then flatten them as needed for filtering/rendering.
  - I used `[].concat(...chunks)` as the flattening approach, based on benchmarking. it seems to be faster than `flat()`.

## How I Tested These Changes

View run with neverending log loading. Verify that logs appear and stream in.

Use Chrome profiler to determine that the chunking and flattening behavior appears to be performant at scale. The `filterLogs` function remains expensive, but I think in this case it's because the logs themselves are enormous and require a ton of JSON stringification to support text searching.

## Changelog

[ui] Fix run log streaming for runs with a large volume of logs.

js_modules/dagster-ui/packages/ui-core/src/runs/LogsProvider.tsx

@@ -39,7 +39,7 @@ export interface LogFilter {
 }
 
 export interface LogsProviderLogs {
-  allNodes: LogNode[];
+  allNodeChunks: LogNode[][];
   counts: LogLevelCounts;
   loading: boolean;
 }
@@ -72,7 +72,7 @@ const BATCH_INTERVAL = 100;
 const QUERY_LOG_LIMIT = 1000;
 
 type State = {
-  nodes: LogNode[];
+  nodeChunks: LogNode[][];
   cursor: string | null;
   counts: LogLevelCounts;
   loading: boolean;
@@ -99,25 +99,29 @@ const reducer = (state: State, action: Action) => {
         ...node,
         clientsideKey: `csk${node.timestamp}-${ii}`,
       }));
-      const nodes = [...state.nodes, ...queuedNodes];
+
+      const copy = state.nodeChunks.slice();
+      copy.push(queuedNodes);
+
       const counts = {...state.counts};
       queuedNodes.forEach((node) => {
         const level = logNodeLevel(node);
         counts[level]++;
       });
-      return {nodes, counts, loading: action.hasMore, cursor: action.cursor};
+
+      return {nodeChunks: copy, counts, loading: action.hasMore, cursor: action.cursor};
     }
     case 'set-cursor':
       return {...state, cursor: action.cursor};
     case 'reset':
-      return {nodes: [], counts: emptyCounts, cursor: null, loading: true};
+      return {nodeChunks: [], counts: emptyCounts, cursor: null, loading: true};
     default:
       return state;
   }
 };
 
 const initialState: State = {
-  nodes: [],
+  nodeChunks: [] as LogNode[][],
   counts: emptyCounts,
   cursor: null,
   loading: true,
@@ -184,7 +188,7 @@ const useLogsProviderWithSubscription = (runId: string) => {
     }, BATCH_INTERVAL);
   }, [syncPipelineStatusToApolloCache]);
 
-  const {nodes, counts, cursor, loading} = state;
+  const {nodeChunks, counts, cursor, loading} = state;
 
   const {availability, disabled, status} = React.useContext(WebSocketContext);
   const lostWebsocket = !disabled && availability === 'available' && status === WebSocket.CLOSED;
@@ -227,10 +231,10 @@ const useLogsProviderWithSubscription = (runId: string) => {
 
   return React.useMemo(
     () =>
-      nodes !== null
-        ? {allNodes: nodes, counts, loading, subscriptionComponent}
-        : {allNodes: [], counts, loading, subscriptionComponent},
-    [counts, loading, nodes, subscriptionComponent],
+      nodeChunks !== null
+        ? {allNodeChunks: nodeChunks, counts, loading, subscriptionComponent}
+        : {allNodeChunks: [], counts, loading, subscriptionComponent},
+    [counts, loading, nodeChunks, subscriptionComponent],
   );
 };
 
@@ -285,7 +289,7 @@ const POLL_INTERVAL = 5000;
 const LogsProviderWithQuery = (props: LogsProviderWithQueryProps) => {
   const {children, runId} = props;
   const [state, dispatch] = React.useReducer(reducer, initialState);
-  const {counts, cursor, nodes} = state;
+  const {counts, cursor, nodeChunks} = state;
 
   const dependency = useTraceDependency('RunLogsQuery');
 
@@ -332,9 +336,9 @@ const LogsProviderWithQuery = (props: LogsProviderWithQueryProps) => {
   return (
     <>
       {children(
-        nodes !== null && nodes.length > 0
-          ? {allNodes: nodes, counts, loading: false}
-          : {allNodes: [], counts, loading: true},
+        nodeChunks !== null && nodeChunks.length > 0
+          ? {allNodeChunks: nodeChunks, counts, loading: false}
+          : {allNodeChunks: [], counts, loading: true},
       )}
     </>
   );
@@ -350,7 +354,7 @@ export const LogsProvider = (props: LogsProviderProps) => {
   }
 
   if (availability === 'attempting-to-connect') {
-    return <>{children({allNodes: [], counts: emptyCounts, loading: true})}</>;
+    return <>{children({allNodeChunks: [], counts: emptyCounts, loading: true})}</>;
   }
 
   return <LogsProviderWithSubscription runId={runId}>{children}</LogsProviderWithSubscription>;

js_modules/dagster-ui/packages/ui-core/src/runs/LogsScrollingTable.tsx

@@ -1,4 +1,4 @@
-import {Box, Colors, NonIdealState, Row} from '@dagster-io/ui-components';
+import {Box, Colors, NonIdealState, Row, SpinnerWithText} from '@dagster-io/ui-components';
 import {useVirtualizer} from '@tanstack/react-virtual';
 import {useEffect, useRef} from 'react';
 import styled from 'styled-components';
@@ -81,12 +81,10 @@ export const LogsScrollingTable = (props: Props) => {
   }, [totalHeight, virtualizer]);
 
   const content = () => {
-    if (logs.loading) {
+    if (logs.allNodeChunks.length === 0 && logs.loading) {
       return (
-        <Box margin={{top: 32}}>
-          <ListEmptyState>
-            <NonIdealState icon="spinner" title="Fetching logs..." />
-          </ListEmptyState>
+        <Box margin={{top: 32}} flex={{direction: 'column', alignItems: 'center'}}>
+          <SpinnerWithText label="Loading run logs…" />
         </Box>
       );
     }

js_modules/dagster-ui/packages/ui-core/src/runs/RunMetadataProvider.tsx

@@ -3,6 +3,7 @@ import * as React from 'react';
 import {LogsProviderLogs} from './LogsProvider';
 import {RunContext} from './RunContext';
 import {gql} from '../apollo-client';
+import {flattenOneLevel} from '../util/flattenOneLevel';
 import {RunFragment} from './types/RunFragments.types';
 import {RunMetadataProviderMessageFragment} from './types/RunMetadataProvider.types';
 import {StepEventStatus} from '../graphql/types';
@@ -371,7 +372,8 @@ export const RunMetadataProvider = ({logs, children}: IRunMetadataProviderProps)
   const run = React.useContext(RunContext);
   const runMetadata = React.useMemo(() => extractMetadataFromRun(run), [run]);
   const metadata = React.useMemo(
-    () => (logs.loading ? runMetadata : extractMetadataFromLogs(logs.allNodes)),
+    () =>
+      logs.loading ? runMetadata : extractMetadataFromLogs(flattenOneLevel(logs.allNodeChunks)),
     [logs, runMetadata],
   );
   return <>{children(metadata)}</>;

js_modules/dagster-ui/packages/ui-core/src/runs/StepLogsDialog.tsx

@@ -8,7 +8,7 @@ import {
   Mono,
   Spinner,
 } from '@dagster-io/ui-components';
-import {useState} from 'react';
+import {useMemo, useState} from 'react';
 import * as React from 'react';
 import {Link} from 'react-router-dom';
 import styled from 'styled-components';
@@ -22,6 +22,7 @@ import {IRunMetadataDict, RunMetadataProvider} from './RunMetadataProvider';
 import {titleForRun} from './RunUtils';
 import {useComputeLogFileKeyForSelection} from './useComputeLogFileKeyForSelection';
 import {DagsterEventType} from '../graphql/types';
+import {flattenOneLevel} from '../util/flattenOneLevel';
 
 export function useStepLogs({runId, stepKeys}: {runId?: string; stepKeys?: string[]}) {
   const [showingLogs, setShowingLogs] = React.useState<{runId: string; stepKeys: string[]} | null>(
@@ -113,9 +114,12 @@ export const StepLogsModalContent = ({
   const [logType, setComputeLogType] = useState<LogType>(LogType.structured);
   const [computeLogUrl, setComputeLogUrl] = React.useState<string | null>(null);
 
-  const firstLogForStep = logs.allNodes.find(
+  const flatLogs = useMemo(() => flattenOneLevel(logs.allNodeChunks), [logs]);
+
+  const firstLogForStep = flatLogs.find(
     (l) => l.eventType === DagsterEventType.STEP_START && l.stepKey && stepKeys.includes(l.stepKey),
   );
+
   const firstLogForStepTime = firstLogForStep ? Number(firstLogForStep.timestamp) : 0;
 
   const [filter, setFilter] = useState<LogFilter>({

js_modules/dagster-ui/packages/ui-core/src/runs/filterLogs.tsx

@@ -3,9 +3,10 @@ import {eventTypeToDisplayType} from './getRunFilterProviders';
 import {logNodeLevel} from './logNodeLevel';
 import {LogNode} from './types';
 import {weakmapMemoize} from '../app/Util';
+import {flattenOneLevel} from '../util/flattenOneLevel';
 
 export function filterLogs(logs: LogsProviderLogs, filter: LogFilter, filterStepKeys: string[]) {
-  const filteredNodes = logs.allNodes.filter((node) => {
+  const filteredNodes = flattenOneLevel(logs.allNodeChunks).filter((node) => {
     // These events are used to determine which assets a run will materialize and are not intended
     // to be displayed in the Dagster UI. Pagination is offset based, so we remove these logs client-side.
     if (

js_modules/dagster-ui/packages/ui-core/src/util/flattenOneLevel.tsx

@@ -0,0 +1,10 @@
+/**
+ * Flattens a two-dimensional array into a one-dimensional array.
+ *
+ * @param nodeChunks - The two-dimensional array to flatten.
+ * @returns The flattened one-dimensional array.
+ */
+// https://jsbench.me/o8kqzo8olz/1
+export function flattenOneLevel<T>(arrays: T[][]) {
+  return ([] as T[]).concat(...arrays);
+}