[Bug]: Use a unique token for log pagination instead of a timestamp #2833 (#2845)

Author: peterschmidt85

frontend/src/hooks/useInfiniteScroll.ts

@@ -26,7 +26,7 @@ export const useInfiniteScroll = <DataItem, Args extends InfinityListArgs>({
     const [data, setData] = useState<ListResponse<DataItem>>([]);
     const scrollElement = useRef<HTMLElement>(document.documentElement);
     const isLoadingRef = useRef<boolean>(false);
-    const lastRequestParams = useRef<TRunsRequestParams | undefined>(undefined);
+    const lastRequestParams = useRef<Args | undefined>(undefined);
     const [disabledMore, setDisabledMore] = useState(false);
     const { limit, ...argsProp } = args;
     const lastArgsProps = useRef<Partial<Args>>(null);

frontend/src/pages/Runs/Details/Logs/index.tsx

@@ -8,7 +8,7 @@ import { Terminal } from '@xterm/xterm';
 import { Container, Header, ListEmptyMessage, Loader, TextContent } from 'components';
 
 import { useAppSelector } from 'hooks';
-import { useGetProjectLogsQuery } from 'services/project';
+import { useLazyGetProjectLogsQuery } from 'services/project';
 
 import { selectSystemMode } from 'App/slice';
 
@@ -22,10 +22,50 @@ export const Logs: React.FC<IProps> = ({ className, projectName, runName, jobSub
     const { t } = useTranslation();
     const appliedTheme = useAppSelector(selectSystemMode);
 
-    const terminalInstance = useRef<Terminal>(new Terminal());
-
+    const terminalInstance = useRef<Terminal>(new Terminal({scrollback: 10000000}));
     const fitAddonInstance = useRef<FitAddon>(new FitAddon());
     const [logsData, setLogsData] = useState<ILogItem[]>([]);
+    const [isLoading, setIsLoading] = useState(false);
+
+    const [getProjectLogs] = useLazyGetProjectLogsQuery();
+
+    const writeDataToTerminal = (logs: ILogItem[]) => {
+        logs.forEach((logItem) => {
+            terminalInstance.current.write(logItem.message);
+        });
+
+        fitAddonInstance.current.fit();
+    };
+
+    const getNextLogItems = (nextToken?: string) => {
+        setIsLoading(true);
+
+        if (!jobSubmissionId) {
+            return;
+        }
+
+        getProjectLogs({
+            project_name: projectName,
+            run_name: runName,
+            descending: false,
+            job_submission_id: jobSubmissionId ?? '',
+            next_token: nextToken,
+            limit: LIMIT_LOG_ROWS,
+        })
+            .unwrap()
+            .then((response) => {
+                setLogsData((old) => [...old, ...response.logs]);
+
+                writeDataToTerminal(response.logs);
+
+                if (response.next_token) {
+                    getNextLogItems(response.next_token);
+                } else {
+                    setIsLoading(false);
+                }
+            })
+            .catch(() => setIsLoading(false));
+    };
 
     useEffect(() => {
         if (appliedTheme === Mode.Light) {
@@ -45,6 +85,8 @@ export const Logs: React.FC<IProps> = ({ className, projectName, runName, jobSub
     useEffect(() => {
         terminalInstance.current.loadAddon(fitAddonInstance.current);
 
+        getNextLogItems();
+
         const onResize = () => {
             fitAddonInstance.current.fit();
         };
@@ -56,50 +98,27 @@ export const Logs: React.FC<IProps> = ({ className, projectName, runName, jobSub
         };
     }, []);
 
-    const {
-        data: fetchData,
-        isLoading,
-        isFetching: isFetchingLogs,
-    } = useGetProjectLogsQuery(
-        {
-            project_name: projectName,
-            run_name: runName,
-            descending: true,
-            job_submission_id: jobSubmissionId ?? '',
-            limit: LIMIT_LOG_ROWS,
-        },
-        {
-            skip: !jobSubmissionId,
-        },
-    );
-
-    useEffect(() => {
-        if (fetchData) {
-            const reversed = [...fetchData].reverse();
-            setLogsData((old) => [...reversed, ...old]);
-        }
-    }, [fetchData]);
-
     useEffect(() => {
         const element = document.getElementById('terminal');
 
-        if (logsData.length && terminalInstance.current && element) {
+        if (terminalInstance.current && element) {
             terminalInstance.current.open(element);
-
-            logsData.forEach((logItem) => {
-                terminalInstance.current.write(logItem.message);
-            });
-
-            fitAddonInstance.current.fit();
         }
-    }, [logsData]);
+    }, []);
 
     return (
         <div className={classNames(styles.logs, className)}>
-            <Container header={<Header variant="h2">{t('projects.run.log')}</Header>}>
+            <Container
+                header={
+                    <Header variant="h2">
+                        <div className={styles.headerContainer}>
+                            {t('projects.run.log')}
+                            <Loader show={isLoading} padding={'n'} className={classNames(styles.loader)} loadingText={''} />
+                        </div>
+                    </Header>
+                }
+            >
                 <TextContent>
-                    <Loader padding={'n'} className={classNames(styles.loader, { show: isLoading || isFetchingLogs })} />
-
                     {!isLoading && !logsData.length && (
                         <ListEmptyMessage
                             title={t('projects.run.log_empty_message_title')}

frontend/src/pages/Runs/Details/Logs/styles.module.scss

@@ -1,5 +1,19 @@
 @use '@cloudscape-design/design-tokens/index' as awsui;
 
+.headerContainer {
+    display: flex;
+    gap: 10px;
+    align-items: center;
+}
+
+.loader {
+    position: relative;
+    top: -2px;
+    height: 20px;
+    background-color: rgba(awsui.$color-background-container-content, .8);
+    color: #6e6e6e;
+}
+
 .logs {
     display: flex;
     flex-direction: column;
@@ -39,22 +53,6 @@
                     }
                 }
 
-                .loader {
-                    pointer-events: none;
-                    position: absolute;
-                    left: 0;
-                    right: 0;
-                    top: -20px;
-                    height: 20px;
-                    background-color: rgba(awsui.$color-background-container-content, .8);
-                    transition: transform .3s ease;
-                    color: #6e6e6e;
-
-                    &:global(.show) {
-                        transform: translateY(100%);
-                    }
-                }
-
                 .terminal {
                     flex-grow: 1;
                     min-height: 0;

frontend/src/services/project.ts

@@ -116,7 +116,7 @@ export const projectApi = createApi({
             invalidatesTags: () => ['Projects'],
         }),
 
-        getProjectLogs: builder.query<ILogItem[], TRequestLogsParams>({
+        getProjectLogs: builder.query<TResponseLogsParams, TRequestLogsParams>({
             query: ({ project_name, ...body }) => {
                 return {
                     url: API.PROJECTS.LOGS(project_name),
@@ -127,11 +127,17 @@ export const projectApi = createApi({
 
             keepUnusedDataFor: 0,
             providesTags: () => ['ProjectLogs'],
-            transformResponse: (response: { logs: ILogItem[] }) =>
-                response.logs.map((logItem) => ({
+            transformResponse: (response: { logs: ILogItem[]; next_token: string }) => {
+                const logs = response.logs.map((logItem) => ({
                     ...logItem,
                     message: base64ToArrayBuffer(logItem.message as string),
-                })),
+                }));
+
+                return {
+                    ...response,
+                    logs,
+                };
+            },
         }),
 
         getProjectRepos: builder.query<IRepo[], { project_name: string }>({
@@ -166,6 +172,7 @@ export const {
     useRemoveProjectMemberMutation,
     useDeleteProjectsMutation,
     useGetProjectLogsQuery,
+    useLazyGetProjectLogsQuery,
     useGetProjectReposQuery,
     useUpdateProjectMutation,
 } = projectApi;

frontend/src/types/log.d.ts

@@ -1,16 +1,22 @@
 declare interface ILogItem {
-    log_source: 'stdout' | 'stderr'
-    timestamp: string,
-    message: string | Uint8Array,
+    log_source: 'stdout' | 'stderr';
+    timestamp: string;
+    message: string | Uint8Array;
 }
 
 declare type TRequestLogsParams = {
-    project_name: IProject['project_name'],
-    run_name: IRun['run_name'],
-    job_submission_id: string
-    start_time?: DateTime,
-    end_time?: DateTime,
-    descending?: boolean,
-    limit?: number
-    diagnose?: boolean
-}
+    project_name: IProject['project_name'];
+    run_name: IRun['run_name'];
+    job_submission_id: string;
+    start_time?: DateTime;
+    end_time?: DateTime;
+    descending?: boolean;
+    limit?: number;
+    diagnose?: boolean;
+    next_token?: string;
+};
+
+declare type TResponseLogsParams = {
+    logs: ILogItem[];
+    next_token?: string;
+};

src/dstack/_internal/core/compatibility/logs.py

@@ -0,0 +1,15 @@
+from typing import Dict, Optional
+
+from dstack._internal.server.schemas.logs import PollLogsRequest
+
+
+def get_poll_logs_excludes(request: PollLogsRequest) -> Optional[Dict]:
+    """
+    Returns exclude mapping to exclude certain fields from the request.
+    Use this method to exclude new fields when they are not set to keep
+    clients backward-compatibility with older servers.
+    """
+    excludes = {}
+    if request.next_token is None:
+        excludes["next_token"] = True
+    return excludes if excludes else None

src/dstack/_internal/core/models/logs.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 from enum import Enum
-from typing import List
+from typing import List, Optional
 
 from dstack._internal.core.models.common import CoreModel
 
@@ -23,3 +23,4 @@ class LogEvent(CoreModel):
 
 class JobSubmissionLogs(CoreModel):
     logs: List[LogEvent]
+    next_token: Optional[str]

src/dstack/_internal/server/schemas/logs.py

@@ -1,7 +1,7 @@
 from datetime import datetime
 from typing import Optional
 
-from pydantic import UUID4, Field
+from pydantic import UUID4, Field, validator
 
 from dstack._internal.core.models.common import CoreModel
 
@@ -12,5 +12,14 @@ class PollLogsRequest(CoreModel):
     start_time: Optional[datetime]
     end_time: Optional[datetime]
     descending: bool = False
+    next_token: Optional[str] = None
     limit: int = Field(100, ge=0, le=1000)
     diagnose: bool = False
+
+    @validator("descending")
+    @classmethod
+    def validate_descending(cls, v):
+        # Descending is not supported until we migrate from base64-encoded logs to plain text logs.
+        if v is True:
+            raise ValueError("descending: true is not supported")
+        return v