Skip to content

feat(be): Stop falling back to the legacy job log download #222

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
andyland wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(be): Stop falling back to the legacy job log download #222

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 3 additions & 37 deletions go/job.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,23 +5,17 @@ import (
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"path"
"regexp"
"slices"
"strconv"
"strings"
"time"

sdkapi "github.com/lightning-ai/sdk/go/internal/lightningcloud/openapi/generated/client"
"github.com/lightning-ai/sdk/go/internal/lightningcloud/openapi/generated/client/jobs_service"
"github.com/lightning-ai/sdk/go/internal/lightningcloud/openapi/generated/models"
"github.com/lightning-ai/sdk/go/internal/sdkclient"
)

var datetimePrefix = regexp.MustCompile(`(?m)^\[.*?\] `)

// Job represents a Lightning single-machine job.
type Job struct {
id string
Expand Down Expand Up @@ -623,41 +617,13 @@ func (j *Job) Logs() (string, error) {
}
}
if len(lines) == 0 && !strings.EqualFold(j.status, "running") {
// nothing in the logs API: the job finished before logs moved to the current
// storage format, so fall back to its saved log file
return j.savedLogs(api)
// the job finished before logs moved to the current storage format; reading it
// queues its migration, so its lines appear once that finishes
return "", errors.New("this job's logs are still being migrated from older storage, try again shortly")
}
return strings.Join(lines, "\n"), nil
}

// savedLogs downloads the job's saved log file, for jobs whose lines are not in the logs API.
func (j *Job) savedLogs(api *sdkapi.LightningSdkAPI) (string, error) {
resp, err := api.JobsService.JobsServiceDownloadJobLogs(
jobs_service.NewJobsServiceDownloadJobLogsParamsWithContext(context.Background()).
WithProjectID(j.teamspaceID).
WithID(j.id),
)
if err != nil {
return "", err
}
if resp.Payload == nil || resp.Payload.URL == "" {
return "", fmt.Errorf("download job logs returned empty URL")
}
logResp, err := http.Get(resp.Payload.URL)
if err != nil {
return "", err
}
defer logResp.Body.Close()
if logResp.StatusCode < 200 || logResp.StatusCode >= 300 {
return "", fmt.Errorf("download job logs returned status %d", logResp.StatusCode)
}
body, err := io.ReadAll(logResp.Body)
if err != nil {
return "", err
}
return datetimePrefix.ReplaceAllString(string(body), ""), nil
}

// Dict returns a JSON-friendly public representation of the job.
func (j *Job) Dict() JobDict {
if j == nil {
Expand Down
26 changes: 6 additions & 20 deletions go/job_mmt_lifecycle_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,38 +91,24 @@ func TestJobLifecycleUsesV2Routes(t *testing.T) {
}
}

func TestJobLogsFallBackToSavedFileWhenNotInLogsAPI(t *testing.T) {
var serverURL string
func TestJobLogsErrorWhileMigratingToLogsAPI(t *testing.T) {
var seen []string

server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
seen = append(seen, r.Method+" "+r.URL.RequestURI())
w.Header().Set("Content-Type", "application/json")
switch r.Method + " " + r.URL.Path {
case "GET /v1/projects/project-1/page-logs":
_ = json.NewEncoder(w).Encode(map[string]any{"entries": []map[string]any{}})
case "GET /v1/projects/project-1/jobs/job-1/download-logs":
_ = json.NewEncoder(w).Encode(map[string]any{"url": serverURL + "/logs/job-1"})
case "GET /logs/job-1":
w.Header().Set("Content-Type", "text/plain")
_, _ = w.Write([]byte("[2026-06-16T10:00:00Z] first line\n[2026-06-16T10:00:01Z] second line\n"))
default:
if r.Method+" "+r.URL.Path != "GET /v1/projects/project-1/page-logs" {
assert.Fail(t, fmt.Sprintf("unexpected request: %s %s", r.Method, r.URL.RequestURI()))
}
_ = json.NewEncoder(w).Encode(map[string]any{"entries": []map[string]any{}})
}))
serverURL = server.URL
defer server.Close()
t.Setenv("LIGHTNING_CLOUD_URL", server.URL)

j := mustJob(t, "job-1", "train", "project-1", "studio-1", lit.JobOptions{Status: "completed"})
logs, err := j.Logs()
require.NoErrorf(t, err, "job.Logs returned error")
assert.Equal(t, "first line\nsecond line\n", logs)
assert.Equal(t, []string{
"GET /v1/projects/project-1/page-logs?jobIds=job-1",
"GET /v1/projects/project-1/jobs/job-1/download-logs",
"GET /logs/job-1",
}, seen)
_, err := j.Logs()
require.ErrorContains(t, err, "still being migrated")
assert.Equal(t, []string{"GET /v1/projects/project-1/page-logs?jobIds=job-1"}, seen)
}

func TestJobLogsRunningWithoutHistoryDoesNotFallBack(t *testing.T) {
Expand Down
26 changes: 0 additions & 26 deletions python/lightning_sdk/api/job_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,11 @@
import warnings
from functools import lru_cache
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
from urllib.request import urlopen

from lightning_sdk.api.utils import _get_cloud_url as _cloud_url
from lightning_sdk.api.utils import (
_machine_to_compute_name,
cached_lightning_client,
remove_datetime_prefix,
resolve_path_mappings,
resolve_tags,
)
Expand All @@ -18,7 +16,6 @@
JobsServiceUpdateJobBody,
V1CloudProvider,
V1ClusterAccelerator,
V1DownloadJobLogsResponse,
V1EnvVar,
V1Job,
V1JobSpec,
Expand Down Expand Up @@ -370,29 +367,6 @@ def delete_job(self, job_id: str, teamspace_id: str, cloudspace_id: Optional[str
"""
self._client.jobs_service_delete_job(project_id=teamspace_id, id=job_id, cloudspace_id=cloudspace_id or "")

def get_logs_finished(self, job_id: str, teamspace_id: str, timestamps: bool = False) -> str:
"""Download and return the saved log file of a finished job.

Only used for jobs whose lines are not in the logs API (jobs that finished before
logs moved to the current storage format).

Args:
job_id: The unique identifier of the job whose logs are fetched.
teamspace_id: The ID of the teamspace that owns the job.
timestamps: Keep each line's ``[<RFC3339>] `` prefix instead of stripping it.

Returns:
The decoded log text, with datetime prefixes stripped unless ``timestamps``.
"""
resp: V1DownloadJobLogsResponse = self._client.jobs_service_download_job_logs(
project_id=teamspace_id, id=job_id
)

data = urlopen(resp.url).read().decode("utf-8")
if timestamps:
return str(data)
return remove_datetime_prefix(str(data))

@staticmethod
@lru_cache(maxsize=None)
def _cached_studio_name(project_id: str, cloudspace_id: str) -> Optional[str]:
Expand Down
59 changes: 22 additions & 37 deletions python/lightning_sdk/job.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,10 @@
_RUNNING_LOGS_IDLE_TIMEOUT = 5.0


def _logs_migrating_error() -> RuntimeError:
return RuntimeError("This job's logs are still being migrated from older storage. Try again shortly.")


class _Logs:
"""A logs handle that is both a value and callable.

Expand Down Expand Up @@ -844,11 +848,10 @@ def logs(self) -> _Logs:
``since``, ``until``, ``query`` and ``severity`` are applied by the server, to both the
saved logs and the live stream.

A finished job whose lines are not in the logs API falls back to its saved log file,
which cannot be filtered server-side. ``timestamps`` still works there, but ``query``
and ``severity`` raise :class:`RuntimeError` rather than silently returning nothing —
read unfiltered and filter locally for such a job. A running one with no history in
the logs API has its live stream tailed briefly instead.
A finished job whose lines are not in the logs API yet is still being migrated from
older storage, and raises :class:`RuntimeError`; reading it again shortly after
succeeds. A running one with no history in the logs API has its live stream tailed
briefly instead.
"""
return _Logs(self._compute_logs)

Expand All @@ -861,8 +864,8 @@ def download_logs(

Reads every page of the job's history from the logs API. A multi-machine job
downloads the merged timeline of all its machines, each line prefixed with its
machine name. A job whose lines are not in the logs API falls back to its saved
log file.
machine name. A finished job whose lines are not in the logs API yet raises
:class:`RuntimeError`, as :attr:`logs` does.
"""
if self.is_multi_machine:
text = self._download_multi_machine_logs(timestamps=timestamps)
Expand All @@ -873,16 +876,14 @@ def download_logs(
return str(dest)

def _download_machine_logs(self, *, timestamps: bool) -> str:
"""This machine's complete history, from the logs API or else its saved log file."""
"""This machine's complete history from the logs API."""
job = self._guaranteed_job
lines = list(self._stream_entries(job, follow=False, tail=None, timestamps=timestamps))
if lines:
return "\n".join(lines) + "\n"
if self.status == Status.Running:
return ""
return self._standalone_job_api.get_logs_finished(
job_id=job.id, teamspace_id=self.teamspace.id, timestamps=timestamps
)
raise _logs_migrating_error()

def _download_multi_machine_logs(self, *, timestamps: bool) -> str:
lines = list(
Expand All @@ -892,12 +893,9 @@ def _download_multi_machine_logs(self, *, timestamps: bool) -> str:
)
if lines:
return "\n".join(lines) + "\n"
# nothing in the logs API: stitch each machine's saved log file, prefixed like the merged timeline
out: List[str] = []
for machine in self.machines:
text = machine._download_machine_logs(timestamps=timestamps)
out.extend(f"[{machine.name}] {line}" for line in text.splitlines())
return "\n".join(out) + "\n" if out else ""
if self.status == Status.Running:
return ""
raise _logs_migrating_error()

def _compute_logs(
self,
Expand Down Expand Up @@ -988,27 +986,14 @@ def _read_machine_logs(
)
)
if not lines and status != Status.Running:
if query is not None or severity is not None:
# An empty filtered read is ambiguous: either the filter genuinely matched
# nothing, or this job's lines aren't in the logs API at all. Only the
# latter needs the saved-file fallback, which cannot filter — so probe
# unfiltered before deciding, rather than reporting "no matches" for a
# filter that was never applied.
if not any(self._stream_entries(read_job, follow=False, tail=1, timestamps=False)):
raise RuntimeError(
"This job's logs are only available as a saved file, which cannot be "
"filtered server-side, so `query`/`severity` cannot be applied. Read "
"the logs without them and filter locally."
)
filtered = any(f is not None for f in (since, until, query, severity))
# An empty filtered read is ambiguous: the filter may have matched nothing, or the
# job's lines may not be in the logs API yet. Probe unfiltered to tell them apart.
if filtered and any(self._stream_entries(read_job, follow=False, tail=1, timestamps=False)):
return iter(()) if follow else ""
# Nothing in the logs API for this job (it finished before logs moved to the current
# storage format): fall back to its saved log file.
text = self._standalone_job_api.get_logs_finished(
job_id=read_job.id, teamspace_id=self.teamspace.id, timestamps=timestamps
)
if tail is not None:
text = "\n".join(text.splitlines()[-tail:])
return iter(text.splitlines()) if follow else text
# The job finished before logs moved to the current storage format. Reading it
# queues its migration, so its lines appear once that finishes.
raise _logs_migrating_error()

# `follow` on an already-finished job returns the saved lines as an iterator (and stops),
# keeping the return type consistent with the live-follow path.
Expand Down
65 changes: 17 additions & 48 deletions python/tests/core/test_job.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -242,20 +242,18 @@ def test_download_logs_formats_timestamps(tmp_path, monkeypatch):
assert (tmp_path / "job-my-job.log").read_text() == "2026-07-27T09:00:00+00:00 hello\n"


def test_download_logs_falls_back_to_saved_file(tmp_path, monkeypatch):
def test_download_logs_raises_while_migrating(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
job = _bare_job("my-job")
job._logs_api.stream = mock.MagicMock(return_value=[])
job._standalone_job_api.get_logs_finished = mock.MagicMock(return_value="saved 1\nsaved 2\n")

with mock.patch.object(Job, "status", new_callable=mock.PropertyMock, return_value=Status.Completed):
written = job.download_logs()
with (
mock.patch.object(Job, "status", new_callable=mock.PropertyMock, return_value=Status.Completed),
pytest.raises(RuntimeError, match="still being migrated"),
):
job.download_logs()

assert (tmp_path / "job-my-job.log").read_text() == "saved 1\nsaved 2\n"
assert written == str(tmp_path / "job-my-job.log")
job._standalone_job_api.get_logs_finished.assert_called_once_with(
job_id="job-1", teamspace_id="ts-abc", timestamps=False
)
assert not (tmp_path / "job-my-job.log").exists()


def test_download_logs_multi_machine_writes_merged_timeline(tmp_path, monkeypatch):
Expand Down Expand Up @@ -1440,73 +1438,44 @@ def test_job_logs_timestamps_and_filters(job_api_get_job_by_name_mocker, interna


@mock.patch("lightning_sdk.lightning_cloud.rest_client.Auth", new=mock.MagicMock())
def test_job_logs_finished_falls_back_to_saved_file(job_api_get_job_by_name_mocker, internal_studio_init_mocker):
studio = Studio(name="st-abc", teamspace="ts-abc", org="org-abc")
job = Job("test-job", studio.teamspace)

job._job_api.get_job = mock.MagicMock(return_value=V1Job(id="test-job-id", state="completed"))
# no lines in the logs API (e.g. an older job): the saved log file is used instead
_stub_logs_api(job, [])
logs_mock = mock.MagicMock(return_value="line 1\nline 2\n")
job._standalone_job_api.get_logs_finished = logs_mock

# the proxy behaves like the log text: equality, str(), .splitlines(), iteration
assert job.logs == "line 1\nline 2\n"
assert str(job.logs) == "line 1\nline 2\n"
assert job.logs.splitlines() == ["line 1", "line 2"]
assert list(job.logs) == ["line 1", "line 2"]
logs_mock.assert_called_with(job_id="test-job-id", teamspace_id=job.teamspace.id, timestamps=False)


@mock.patch("lightning_sdk.lightning_cloud.rest_client.Auth", new=mock.MagicMock())
def test_job_logs_finished_saved_file_keeps_timestamps(job_api_get_job_by_name_mocker, internal_studio_init_mocker):
def test_job_logs_finished_raises_while_migrating(job_api_get_job_by_name_mocker, internal_studio_init_mocker):
studio = Studio(name="st-abc", teamspace="ts-abc", org="org-abc")
job = Job("test-job", studio.teamspace)

job._job_api.get_job = mock.MagicMock(return_value=V1Job(id="test-job-id", state="completed"))
# no lines in the logs API: an older job whose logs haven't been migrated yet
_stub_logs_api(job, [])
logs_mock = mock.MagicMock(return_value="[2026-08-13T14:15:03.797142418Z] line 1\n")
job._standalone_job_api.get_logs_finished = logs_mock

assert job.logs(timestamps=True) == "[2026-08-13T14:15:03.797142418Z] line 1\n"
logs_mock.assert_called_with(job_id="test-job-id", teamspace_id=job.teamspace.id, timestamps=True)
with pytest.raises(RuntimeError, match="still being migrated"):
str(job.logs)


@mock.patch("lightning_sdk.lightning_cloud.rest_client.Auth", new=mock.MagicMock())
def test_job_logs_filtered_read_raises_when_only_saved_file_exists(
job_api_get_job_by_name_mocker, internal_studio_init_mocker
):
def test_job_logs_filtered_read_raises_while_migrating(job_api_get_job_by_name_mocker, internal_studio_init_mocker):
"""A filter the server never applied must not be reported as "no matches"."""
studio = Studio(name="st-abc", teamspace="ts-abc", org="org-abc")
job = Job("test-job", studio.teamspace)

job._job_api.get_job = mock.MagicMock(return_value=V1Job(id="test-job-id", state="completed"))
_stub_logs_api(job, []) # nothing in the logs API at all: only the saved file has lines
fallback = mock.MagicMock(return_value="everything\n")
job._standalone_job_api.get_logs_finished = fallback
_stub_logs_api(job, [])

with pytest.raises(RuntimeError, match="cannot be filtered server-side"):
with pytest.raises(RuntimeError, match="still being migrated"):
job.logs(query="nope")
fallback.assert_not_called()


@mock.patch("lightning_sdk.lightning_cloud.rest_client.Auth", new=mock.MagicMock())
def test_job_logs_filtered_read_does_not_fall_back(job_api_get_job_by_name_mocker, internal_studio_init_mocker):
def test_job_logs_filtered_read_matching_nothing_is_empty(job_api_get_job_by_name_mocker, internal_studio_init_mocker):
studio = Studio(name="st-abc", teamspace="ts-abc", org="org-abc")
job = Job("test-job", studio.teamspace)

job._job_api.get_job = mock.MagicMock(return_value=V1Job(id="test-job-id", state="completed"))
# the job IS in the logs API — an unfiltered read has lines, the filter just matched none
job._logs_api.stream = mock.MagicMock(
side_effect=lambda *a, **kw: [] if kw.get("query") or kw.get("severity") else [LogEntry(message="x")]
side_effect=lambda *a, **kw: [] if kw.get("query") or kw.get("since") else [LogEntry(message="x")]
)

fallback = mock.MagicMock(return_value="everything\n")
job._standalone_job_api.get_logs_finished = fallback

# a filtered read matching nothing must stay empty rather than dump the unfiltered file
assert job.logs(query="nope") == ""
fallback.assert_not_called()
assert job.logs(since="2026-01-01T00:00:00Z") == ""


@mock.patch("lightning_sdk.lightning_cloud.rest_client.Auth", new=mock.MagicMock())
Expand Down
Loading