diff --git a/go/job.go b/go/job.go index 7fb9d93c..8382cbf3 100644 --- a/go/job.go +++ b/go/job.go @@ -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 @@ -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 { diff --git a/go/job_mmt_lifecycle_test.go b/go/job_mmt_lifecycle_test.go index 42af5d84..361ff5fd 100644 --- a/go/job_mmt_lifecycle_test.go +++ b/go/job_mmt_lifecycle_test.go @@ -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) { diff --git a/python/lightning_sdk/api/job_api.py b/python/lightning_sdk/api/job_api.py index c6c72890..76ef864c 100644 --- a/python/lightning_sdk/api/job_api.py +++ b/python/lightning_sdk/api/job_api.py @@ -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, ) @@ -18,7 +16,6 @@ JobsServiceUpdateJobBody, V1CloudProvider, V1ClusterAccelerator, - V1DownloadJobLogsResponse, V1EnvVar, V1Job, V1JobSpec, @@ -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 ``[] `` 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]: diff --git a/python/lightning_sdk/job.py b/python/lightning_sdk/job.py index 9581c80d..e07a42af 100644 --- a/python/lightning_sdk/job.py +++ b/python/lightning_sdk/job.py @@ -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. @@ -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) @@ -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) @@ -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( @@ -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, @@ -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. diff --git a/python/tests/core/test_job.py b/python/tests/core/test_job.py index 5ab61dc8..a8256d73 100644 --- a/python/tests/core/test_job.py +++ b/python/tests/core/test_job.py @@ -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): @@ -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())