Skip to content

Commit 00e0aee

Browse files
air: fall back to MLflow when Bricklens returns no logs for a terminal run
`air logs <run>` printed "No logs available for run <id>. Run terminated in state SUCCESS" and exited 0 for runs whose logs were fully retrievable: `air logs --download-to DIR` on the same run returned the complete log. This happens when Bricklens is enabled for the workspace but never ingested the run — it answers every request successfully with zero records, yet the logs are present in MLflow. The streaming (print) path only fell back to MLflow on errBricklensFeatureDisabled (gated off / not deployed / persistent failure); an empty-but-successful Bricklens response was treated as the final answer. The download path already reads from MLflow, which is why it worked. Treat "Bricklens served every request but never returned a record" the same as feature-disabled: hand off to the MLflow fallback, which owns the real no-logs report and preserves the run-derived exit code. This mirrors the Python CLI fix (databricks-eng/universe#2366012). Applies to both the terminal tail and the static (past-retry) view. Co-authored-by: Isaac
1 parent 7109b44 commit 00e0aee

2 files changed

Lines changed: 118 additions & 12 deletions

File tree

experimental/air/cmd/logstream.go

Lines changed: 14 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -33,8 +33,10 @@ const (
3333
var retryCheckInterval = 3 * time.Second
3434

3535
// errBricklensFeatureDisabled signals the caller to fall back to MLflow: Bricklens
36-
// is gated off (FEATURE_DISABLED), not deployed (ENDPOINT_NOT_FOUND / 404), or
37-
// persistently failing. The flag is evaluated server-side.
36+
// is gated off (FEATURE_DISABLED), not deployed (ENDPOINT_NOT_FOUND / 404),
37+
// persistently failing, or served every request successfully but never returned a
38+
// record for a run whose logs may still be in MLflow. The flag is evaluated
39+
// server-side.
3840
var errBricklensFeatureDisabled = errors.New("bricklens logs unavailable; falling back to mlflow")
3941

4042
// logRequest describes what to fetch, shared by both backends so they honor the
@@ -301,11 +303,12 @@ func (st *bricklensStreamer) run() (bool, error) {
301303

302304
if terminal {
303305
if !st.firstLogSeen {
304-
// Stop the spinner before the no-logs line so frames don't smear.
305-
if st.onFirstLog != nil {
306-
st.onFirstLog()
307-
}
308-
st.emitNoLogs()
306+
// Bricklens served every request but never returned a record. That is
307+
// not proof the run has no logs: Bricklens ingestion can lag or miss a
308+
// run whose logs are fully present in MLflow (the path --download-to
309+
// uses). Defer to the MLflow fallback rather than declaring "no logs"
310+
// here; it owns the real no-logs report, with the same exit code.
311+
return false, errBricklensFeatureDisabled
309312
}
310313
log.Infof(st.ctx, "air logs: run %d finished in state %s", st.req.runID, st.status.displayState())
311314
return st.status.succeeded(), nil
@@ -338,7 +341,10 @@ func (st *bricklensStreamer) drainStatic(toSec int64) (bool, error) {
338341
return false, err
339342
}
340343
if !st.firstLogSeen {
341-
st.emitNoLogs()
344+
// An empty Bricklens tail doesn't mean the attempt has no logs; fall back to
345+
// MLflow, which holds the immutable per-attempt artifacts. See the terminal
346+
// branch in run.
347+
return false, errBricklensFeatureDisabled
342348
}
343349
return st.status.downloadOutcome(), nil
344350
}
@@ -479,10 +485,6 @@ func (st *bricklensStreamer) emit(body string) {
479485
emitLogLine(st.out, st.req, body)
480486
}
481487

482-
func (st *bricklensStreamer) emitNoLogs() {
483-
emitNoLogs(st.out, st.req, st.status)
484-
}
485-
486488
// displayState is the result state, else the lifecycle state, else "UNKNOWN".
487489
func (s logRunStatus) displayState() string {
488490
if s.resultState != "" {

experimental/air/cmd/logstream_test.go

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -373,6 +373,110 @@ func TestRequestPageRetriesThenSucceeds(t *testing.T) {
373373
assert.Equal(t, 3, calls)
374374
}
375375

376+
// emptyLogsServer serves an empty Bricklens log response for any /logs request
377+
// and a stub for everything else (SDK config probes, etc.).
378+
func emptyLogsServer(t *testing.T) *httptest.Server {
379+
t.Helper()
380+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
381+
if strings.HasSuffix(r.URL.Path, "/logs") {
382+
_, _ = w.Write([]byte(`{"log_records": []}`))
383+
return
384+
}
385+
_, _ = w.Write([]byte(`{}`))
386+
}))
387+
t.Cleanup(srv.Close)
388+
return srv
389+
}
390+
391+
func TestStreamBricklensEmptyFallsBackToMLflow(t *testing.T) {
392+
// Bricklens served every request but returned no record. That is not proof the
393+
// run has no logs (they may be in MLflow), so the streamer must hand off via
394+
// errBricklensFeatureDisabled and emit nothing itself, rather than reporting
395+
// "No logs available" (the reported bug: the print path did, --download-to did not).
396+
tests := []struct {
397+
name string
398+
req logRequest
399+
status logRunStatus
400+
}{
401+
{
402+
name: "terminal run",
403+
req: logRequest{runID: 123, node: 0, attempt: -1, tailLines: -1, jsonOutput: true},
404+
status: logRunStatus{lifeCycleState: "TERMINATED", resultState: "SUCCESS", endTimeMs: 1700000012000},
405+
},
406+
{
407+
name: "static view of a past retry",
408+
req: logRequest{runID: 123, node: 0, attempt: 0, tailLines: -1, staticView: true, jsonOutput: true},
409+
status: logRunStatus{lifeCycleState: "RUNNING"},
410+
},
411+
}
412+
for _, tt := range tests {
413+
t.Run(tt.name, func(t *testing.T) {
414+
var buf bytes.Buffer
415+
w := newTestWorkspaceClient(t, emptyLogsServer(t).URL)
416+
_, err := streamBricklensLogs(t.Context(), w, &buf, tt.req, tt.status)
417+
require.ErrorIs(t, err, errBricklensFeatureDisabled)
418+
assert.Empty(t, buf.String(), "nothing should be emitted before the hand-off")
419+
})
420+
}
421+
}
422+
423+
func TestStreamBricklensTerminalWithRecordsDoesNotFallBack(t *testing.T) {
424+
// A terminal run whose Bricklens stream has records prints them and reports the
425+
// run's outcome, without triggering the empty-result fallback.
426+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
427+
if strings.HasSuffix(r.URL.Path, "/logs") {
428+
_, _ = w.Write([]byte(`{"log_records": [{"time_unix_nano": 1700000001000000000, "body": "hello", "node_index": 0}]}`))
429+
return
430+
}
431+
_, _ = w.Write([]byte(`{}`))
432+
}))
433+
t.Cleanup(srv.Close)
434+
435+
var buf bytes.Buffer
436+
w := newTestWorkspaceClient(t, srv.URL)
437+
status := logRunStatus{lifeCycleState: "TERMINATED", resultState: "SUCCESS", endTimeMs: 1700000012000}
438+
ok, err := streamBricklensLogs(t.Context(), w, &buf, logRequest{runID: 123, node: 0, attempt: -1, tailLines: -1, jsonOutput: true}, status)
439+
require.NoError(t, err)
440+
assert.True(t, ok)
441+
assert.Contains(t, buf.String(), `"line":"hello"`)
442+
}
443+
444+
func TestFetchLogsFallsBackToMLflowWhenBricklensEmpty(t *testing.T) {
445+
// End-to-end repro: a terminal SUCCESS run whose Bricklens stream is empty but
446+
// whose logs are in MLflow. The print path must fall back to MLflow and print
447+
// them, exactly as --download-to already does.
448+
var base string
449+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
450+
switch {
451+
case strings.HasSuffix(r.URL.Path, "/logs"):
452+
_, _ = w.Write([]byte(`{"log_records": []}`))
453+
case r.URL.Path == "/api/2.2/jobs/runs/get":
454+
_, _ = w.Write([]byte(`{"run_id": 123, "state": {"life_cycle_state": "TERMINATED", "result_state": "SUCCESS"}, "tasks": [{"run_id": 456}]}`))
455+
case r.URL.Path == "/api/2.2/jobs/runs/get-output":
456+
_, _ = w.Write([]byte(`{"ai_runtime_task_output": {"mlflow_experiment_id": "E1", "mlflow_run_id": "R1"}}`))
457+
case r.URL.Path == "/api/2.0/mlflow/artifacts/list":
458+
_, _ = w.Write([]byte(`{"files": [{"path": "logs/node_0"}, {"path": "logs/node_0/logs-0.chunk.txt"}]}`))
459+
case r.URL.Path == "/api/2.0/mlflow/artifacts/credentials-for-read":
460+
_, _ = w.Write([]byte(`{"credential_infos": [{"signed_uri": "` + base + `/presigned"}]}`))
461+
case r.URL.Path == "/presigned":
462+
_, _ = w.Write([]byte("line 1\nline 2\n"))
463+
default:
464+
_, _ = w.Write([]byte(`{}`))
465+
}
466+
}))
467+
base = srv.URL
468+
t.Cleanup(srv.Close)
469+
470+
var buf bytes.Buffer
471+
w := newTestWorkspaceClient(t, srv.URL)
472+
status := logRunStatus{lifeCycleState: "TERMINATED", resultState: "SUCCESS", endTimeMs: 1700000012000}
473+
ok, err := fetchLogs(t.Context(), w, &buf, logRequest{runID: 123, node: 0, attempt: -1, tailLines: -1, jsonOutput: true}, status)
474+
require.NoError(t, err)
475+
assert.True(t, ok)
476+
assert.Contains(t, buf.String(), `"line":"line 1"`)
477+
assert.Contains(t, buf.String(), `"line":"line 2"`)
478+
}
479+
376480
func TestSeenSetEviction(t *testing.T) {
377481
s := newSeenSet(2)
378482
s.add(1, "a")

0 commit comments

Comments
 (0)