[UI] Query logs using descending #2892

peterschmidt85 · peterschmidt85 · commit 486486312ca9 · 2025-07-18T16:42:56.000+02:00
- [x] Allowed `descending` in `PollLogsRequest`
- [x] Supported `descending` in `FileLogStorage` (implemented an algorithm that reads lines from the end of t he file)
- [x] Updated `CloudWatchLogStorage` to support `MAX_RETRIES` (to skip empty pages)
diff --git a/frontend/src/pages/Runs/Details/Logs/index.tsx b/frontend/src/pages/Runs/Details/Logs/index.tsx
@@ -12,7 +12,7 @@ import { IProps } from './types';
 
 import styles from './styles.module.scss';
 
-const LIMIT_LOG_ROWS = 1000;
+const LIMIT_LOG_ROWS = 100;
 const LOADING_SCROLL_GAP = 300;
 
 export const Logs: React.FC<IProps> = ({ className, projectName, runName, jobSubmissionId }) => {
diff --git a/src/dstack/_internal/server/schemas/logs.py b/src/dstack/_internal/server/schemas/logs.py
@@ -1,7 +1,7 @@
 from datetime import datetime
 from typing import Optional
 
-from pydantic import UUID4, Field, validator
+from pydantic import UUID4, Field
 
 from dstack._internal.core.models.common import CoreModel
 
@@ -15,11 +15,3 @@ class PollLogsRequest(CoreModel):
     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
diff --git a/src/dstack/_internal/server/services/logs/aws.py b/src/dstack/_internal/server/services/logs/aws.py
@@ -55,6 +55,8 @@ class CloudWatchLogStorage(LogStorage):
     PAST_EVENT_MAX_DELTA = int((timedelta(days=14)).total_seconds()) * 1000 - CLOCK_DRIFT
     # "None of the log events in the batch can be more than 2 hours in the future."
     FUTURE_EVENT_MAX_DELTA = int((timedelta(hours=2)).total_seconds()) * 1000 - CLOCK_DRIFT
+    # Maximum number of retries when polling for log events to skip empty pages.
+    MAX_RETRIES = 10
 
     def __init__(self, *, group: str, region: Optional[str] = None) -> None:
         with self._wrap_boto_errors():
@@ -80,7 +82,7 @@ def poll_logs(self, project: ProjectModel, request: PollLogsRequest) -> JobSubmi
         next_token: Optional[str] = None
         with self._wrap_boto_errors():
             try:
-                cw_events, next_token = self._get_log_events(stream, request)
+                cw_events, next_token = self._get_log_events_with_retry(stream, request)
             except botocore.exceptions.ClientError as e:
                 if not self._is_resource_not_found_exception(e):
                     raise
@@ -101,7 +103,45 @@ def poll_logs(self, project: ProjectModel, request: PollLogsRequest) -> JobSubmi
             )
             for cw_event in cw_events
         ]
-        return JobSubmissionLogs(logs=logs, next_token=next_token if len(logs) > 0 else None)
+        return JobSubmissionLogs(logs=logs, next_token=next_token)
+
+    def _get_log_events_with_retry(
+        self, stream: str, request: PollLogsRequest
+    ) -> Tuple[List[_CloudWatchLogEvent], Optional[str]]:
+        current_request = request
+        previous_next_token = request.next_token
+
+        for attempt in range(self.MAX_RETRIES):
+            cw_events, next_token = self._get_log_events(stream, current_request)
+
+            if cw_events:
+                return cw_events, next_token
+
+            if not next_token:
+                return [], None
+
+            if next_token == previous_next_token:
+                return [], None
+
+            previous_next_token = next_token
+            current_request = PollLogsRequest(
+                run_name=request.run_name,
+                job_submission_id=request.job_submission_id,
+                start_time=request.start_time,
+                end_time=request.end_time,
+                descending=request.descending,
+                next_token=next_token,
+                limit=request.limit,
+                diagnose=request.diagnose,
+            )
+
+        if not request.descending:
+            logger.debug(
+                "Stream %s: exhausted %d retries without finding logs, returning empty response",
+                stream,
+                self.MAX_RETRIES,
+            )
+        return [], next_token if request.descending else None
 
     def _get_log_events(
         self, stream: str, request: PollLogsRequest
@@ -115,7 +155,7 @@ def _get_log_events(
         }
 
         if request.start_time:
-            parameters["startTime"] = datetime_to_unix_time_ms(request.start_time) + 1
+            parameters["startTime"] = datetime_to_unix_time_ms(request.start_time)
 
         if request.end_time:
             parameters["endTime"] = datetime_to_unix_time_ms(request.end_time)
diff --git a/src/dstack/_internal/server/services/logs/filelog.py b/src/dstack/_internal/server/services/logs/filelog.py
@@ -1,5 +1,6 @@
+import os
 from pathlib import Path
-from typing import List, Union
+from typing import Generator, List, Optional, Tuple, Union
 from uuid import UUID
 
 from dstack._internal.core.errors import ServerClientError
@@ -37,6 +38,14 @@ def poll_logs(self, project: ProjectModel, request: PollLogsRequest) -> JobSubmi
             producer=log_producer,
         )
 
+        if request.descending:
+            return self._poll_logs_descending(log_file_path, request)
+        else:
+            return self._poll_logs_ascending(log_file_path, request)
+
+    def _poll_logs_ascending(
+        self, log_file_path: Path, request: PollLogsRequest
+    ) -> JobSubmissionLogs:
         start_line = 0
         if request.next_token:
             try:
@@ -94,6 +103,99 @@ def poll_logs(self, project: ProjectModel, request: PollLogsRequest) -> JobSubmi
 
         return JobSubmissionLogs(logs=logs, next_token=next_token)
 
+    def _poll_logs_descending(
+        self, log_file_path: Path, request: PollLogsRequest
+    ) -> JobSubmissionLogs:
+        start_offset = None
+        if request.next_token:
+            try:
+                start_offset = int(request.next_token)
+                if start_offset < 0:
+                    raise ValueError("Offset must be non-negative")
+            except (ValueError, TypeError):
+                raise ServerClientError(
+                    f"Invalid next_token for descending read: {request.next_token}. "
+                    f"Must be a non-negative integer offset."
+                )
+
+        candidate_logs = []
+
+        try:
+            line_generator = self._read_lines_reversed(log_file_path, start_offset)
+
+            for line_bytes, line_start_offset in line_generator:
+                try:
+                    line_str = line_bytes.decode("utf-8")
+                    log_event = LogEvent.__response__.parse_raw(line_str)
+                except Exception:
+                    continue  # Skip malformed lines
+
+                if request.end_time is not None and log_event.timestamp > request.end_time:
+                    continue
+                if request.start_time and log_event.timestamp <= request.start_time:
+                    break
+
+                candidate_logs.append((log_event, line_start_offset))
+
+                if len(candidate_logs) > request.limit:
+                    break
+        except FileNotFoundError:
+            return JobSubmissionLogs(logs=[], next_token=None)
+
+        logs = [log for log, offset in candidate_logs[: request.limit]]
+        next_token = None
+        if len(candidate_logs) > request.limit:
+            # We fetched one more than the limit, so there are more pages.
+            # The next token should point to the start of the last log we are returning.
+            _last_log_event, last_log_offset = candidate_logs[request.limit - 1]
+            next_token = str(last_log_offset)
+
+        return JobSubmissionLogs(logs=logs, next_token=next_token)
+
+    @staticmethod
+    def _read_lines_reversed(
+        filepath: Path, start_offset: Optional[int] = None, chunk_size: int = 8192
+    ) -> Generator[Tuple[bytes, int], None, None]:
+        """
+        A generator that yields lines from a file in reverse order, along with the byte
+        offset of the start of each line. This is memory-efficient for large files.
+        """
+        with open(filepath, "rb") as f:
+            if start_offset is None:
+                f.seek(0, os.SEEK_END)
+                cursor = f.tell()
+            else:
+                f.seek(0, os.SEEK_END)
+                file_size = f.tell()
+                cursor = file_size
+
+            buffer = b""
+
+            while cursor > 0:
+                seek_pos = max(0, cursor - chunk_size)
+                amount_to_read = cursor - seek_pos
+                f.seek(seek_pos)
+                chunk = f.read(amount_to_read)
+                cursor = seek_pos
+
+                buffer = chunk + buffer
+
+                while b"\n" in buffer:
+                    newline_pos = buffer.rfind(b"\n")
+                    line = buffer[newline_pos + 1 :]
+                    line_start_offset = cursor + newline_pos + 1
+
+                    # Skip lines that start at or after the start_offset
+                    if start_offset is None or line_start_offset < start_offset:
+                        yield line, line_start_offset
+
+                    buffer = buffer[:newline_pos]
+
+            # The remaining buffer is the first line of the file.
+            # Only yield it if we're not using start_offset or if it starts before start_offset
+            if buffer and (start_offset is None or 0 < start_offset):
+                yield buffer, 0
+
     def write_logs(
         self,
         project: ProjectModel,
diff --git a/src/tests/_internal/server/services/test_logs.py b/src/tests/_internal/server/services/test_logs.py
