diff --git a/acceptance/experimental/air/help/output.txt b/acceptance/experimental/air/help/output.txt index c1cb6bdcda5..e3955892a00 100644 --- a/acceptance/experimental/air/help/output.txt +++ b/acceptance/experimental/air/help/output.txt @@ -61,7 +61,7 @@ Flags: --minutes int Fetch only logs from the last N minutes --node int Fetch logs from this node --retry int View logs from a specific retry attempt; -1 means latest (default -1) - --tail int For completed runs, print the last N log lines (default 10000) + --tail int Print the last N existing lines before following, or from a completed run (default 10000) Global Flags: --debug enable debug logging diff --git a/acceptance/experimental/air/logs/output.txt b/acceptance/experimental/air/logs/output.txt index 61a4202c65a..a070a834a4d 100644 --- a/acceptance/experimental/air/logs/output.txt +++ b/acceptance/experimental/air/logs/output.txt @@ -22,9 +22,11 @@ CUDA out of memory >>> [CLI] experimental air logs 123 --tail 1 CUDA out of memory -=== logs with --tail 0 prints no log lines +=== logs with --tail 0 is rejected >>> [CLI] experimental air logs 123 --tail 0 -No logs available for run 123. Run terminated in state SUCCESS +Error: invalid --tail 0: must be positive + +Exit code: 1 === logs from a specific retry >>> [CLI] experimental air logs 123 --retry 0 diff --git a/acceptance/experimental/air/logs/script b/acceptance/experimental/air/logs/script index 38a0ff2347b..3aceb1892d3 100644 --- a/acceptance/experimental/air/logs/script +++ b/acceptance/experimental/air/logs/script @@ -10,8 +10,8 @@ trace $CLI experimental air logs 123 --minutes 30 title "logs with --tail" trace $CLI experimental air logs 123 --tail 1 -title "logs with --tail 0 prints no log lines" -trace $CLI experimental air logs 123 --tail 0 +title "logs with --tail 0 is rejected" +errcode trace $CLI experimental air logs 123 --tail 0 title "logs from a specific retry" trace $CLI experimental air logs 123 --retry 0 diff --git a/experimental/air/cmd/logbricklens.go b/experimental/air/cmd/logbricklens.go index 48ab7286649..ced20cc7711 100644 --- a/experimental/air/cmd/logbricklens.go +++ b/experimental/air/cmd/logbricklens.go @@ -18,6 +18,7 @@ const bricklensLogsPathFmt = "/api/2.0/ai-training/workflows/by-run-id/%d/logs" type logRecord struct { // TimeUnixNano may arrive as a JSON number or string. TimeUnixNano json.Number `json:"time_unix_nano"` + RecordID string `json:"record_id"` Body string `json:"body"` NodeIndex int `json:"node_index"` } diff --git a/experimental/air/cmd/logs.go b/experimental/air/cmd/logs.go index e7aff495006..cd4571a1d29 100644 --- a/experimental/air/cmd/logs.go +++ b/experimental/air/cmd/logs.go @@ -36,7 +36,7 @@ func newLogsCommand() *cobra.Command { } cmd.Flags().IntVar(&node, "node", 0, "Fetch logs from this node") - cmd.Flags().IntVar(&tail, "tail", 0, "For completed runs, print the last N log lines (default 10000)") + cmd.Flags().IntVar(&tail, "tail", 0, "Print the last N existing lines before following, or from a completed run (default 10000)") cmd.Flags().IntVar(&minutes, "minutes", 0, "Fetch only logs from the last N minutes") cmd.Flags().IntVar(&retry, "retry", -1, "View logs from a specific retry attempt; -1 means latest") cmd.Flags().StringVar(&downloadTo, "download-to", "", "Download all logs to this directory instead of printing") @@ -75,7 +75,7 @@ func newLogsCommand() *cobra.Command { return renderError(ctx, cmd, "INVALID_ARGS", "PERMANENT", false, errors.New("cannot combine --tail with --minutes: --tail selects by line count, --minutes by time window")) } - if tail < 0 { + if cmd.Flags().Changed("tail") && tail <= 0 { return renderError(ctx, cmd, "INVALID_ARGS", "PERMANENT", false, fmt.Errorf("invalid --tail %d: must be positive", tail)) } @@ -98,8 +98,7 @@ func newLogsCommand() *cobra.Command { fmt.Errorf("invalid JOB_RUN_ID %q: must be a positive integer", args[0])) } - // -1 signals "unset" (use the default cap); an explicit --tail 0 stays 0 - // and prints no log lines. + // -1 signals "unset" (use the default cap). tailLines := -1 if cmd.Flags().Changed("tail") { tailLines = tail @@ -116,14 +115,15 @@ func newLogsCommand() *cobra.Command { } err = runLogs(streamCtx, cmd, logRequest{ - runID: runID, - node: node, - nodeSet: cmd.Flags().Changed("node"), - attempt: retry, - windowMinutes: minutes, - tailLines: tailLines, - downloadTo: downloadTo, - jsonOutput: root.OutputType(cmd) == flags.OutputJSON, + runID: runID, + node: node, + nodeSet: cmd.Flags().Changed("node"), + attempt: retry, + windowMinutes: minutes, + tailLines: tailLines, + boundInitialLogs: minutes == 0, + downloadTo: downloadTo, + jsonOutput: root.OutputType(cmd) == flags.OutputJSON, }) if downloadTo != "" || root.OutputType(cmd) == flags.OutputJSON { return err diff --git a/experimental/air/cmd/logs_test.go b/experimental/air/cmd/logs_test.go index c641bfe5e4e..73d7645299b 100644 --- a/experimental/air/cmd/logs_test.go +++ b/experimental/air/cmd/logs_test.go @@ -68,6 +68,12 @@ func TestLogsFlagValidation(t *testing.T) { flags: map[string]string{"tail": "-1"}, wantMsg: "invalid --tail", }, + { + name: "zero tail rejected", + args: []string{"5"}, + flags: map[string]string{"tail": "0"}, + wantMsg: "invalid --tail", + }, { name: "negative minutes rejected", args: []string{"5"}, diff --git a/experimental/air/cmd/logstream.go b/experimental/air/cmd/logstream.go index 98b8b02c1e6..b4845b10c2e 100644 --- a/experimental/air/cmd/logstream.go +++ b/experimental/air/cmd/logstream.go @@ -33,6 +33,10 @@ const ( // bricklensEmptyMLflowProbeEveryNPolls throttles the active-run MLflow // existence probe while Bricklens has returned no records. bricklensEmptyMLflowProbeEveryNPolls = 10 + // Event time and ingestion order can differ, so live polls overlap. + bricklensLogLookback = 30 * time.Second + // A terminal job can have records that are not searchable yet. + bricklensTerminalEmptyPolls = 3 ) // statusMessageType tags a client-facing message packed into @@ -68,6 +72,9 @@ func normalizeStatusMessage(raw string) string { // shrink it. var retryCheckInterval = 3 * time.Second +// bricklensPollInterval is independent of slower retry and discovery backoffs. +var bricklensPollInterval = time.Second + // errBricklensFeatureDisabled signals the caller to fall back to MLflow: Bricklens // is gated off (FEATURE_DISABLED), not deployed (ENDPOINT_NOT_FOUND / 404), // persistently failing, or served every request successfully but never returned a @@ -97,7 +104,9 @@ type logRequest struct { // past retry of an active run: that attempt's logs are immutable, so streaming // would poll forever waiting for the run (not the attempt) to finish. staticView bool - jsonOutput bool + // boundInitialLogs limits existing output before following an active run. + boundInitialLogs bool + jsonOutput bool // onStatusChange, when set, is called on each lifecycle transition while // following the run (current, previous display states). Used by // `air run --watch -o json` to emit STATUS events. @@ -265,10 +274,11 @@ type bricklensStreamer struct { req logRequest status logRunStatus - fromSec int64 - lastNano int64 - firstLogSeen bool - seen *seenSet + fromSec int64 + streamStartSec int64 + lastNano int64 + firstLogSeen bool + seen *seenSet // previousState is the last display state reported to onStatusChange. previousState string // onFirstLog, when set, is called once just before the first log line is @@ -330,6 +340,7 @@ func (st *bricklensStreamer) reportStatusChange() { func (st *bricklensStreamer) run() (bool, error) { now := time.Now() st.fromSec = st.req.fromSeconds(st.status, now) + st.streamStartSec = st.fromSec // A past retry's logs are immutable: render a one-shot tail rather than // following the still-active run, which would poll forever. @@ -354,8 +365,10 @@ func (st *bricklensStreamer) run() (bool, error) { // redundant spinner updates. statusRefreshCounter := 0 emptyStreamPolls := 0 + terminalEmptyPolls := 0 lastSpinnerText := "" for { + pollStarted := time.Now() if !firstIteration { status, err := resolveRunStatus(st.ctx, st.w, st.req.runID) if err != nil { @@ -395,20 +408,38 @@ func (st *bricklensStreamer) run() (bool, error) { statusRefreshCounter++ } - // A run already terminal on the first iteration renders as a tail (most - // recent N lines). An active run streams everything with dedup, so a run - // that terminates while we watch doesn't re-print the boundary second. + // `air logs` starts an active attachment with a bounded tail; subsequent + // polls follow live with dedup so the overlap does not re-print it. + var emitted int var err error - if firstIteration && terminal { - err = st.drainTail(toSec) + if firstIteration && (terminal || st.req.boundInitialLogs) { + err = st.drainTail(toSec, !terminal) + if err == nil && !terminal { + // Baseline omitted history without suppressing newer records. + _, err = st.drainPages(toSec, true) + } } else { - err = st.drainPages(toSec) + emitted, err = st.drainPages(toSec, false) } if err != nil { return false, err } if terminal { + if !firstIteration { + if emitted == 0 { + terminalEmptyPolls++ + } else { + terminalEmptyPolls = 0 + } + if terminalEmptyPolls < bricklensTerminalEmptyPolls { + if err := waitForNextBricklensPoll(st.ctx, pollStarted); err != nil { + return false, err + } + firstIteration = false + continue + } + } if !st.firstLogSeen { // A successful but empty Bricklens stream isn't proof the run has no // logs; they may be in MLflow (as --download-to reads). Fall back @@ -427,12 +458,16 @@ func (st *bricklensStreamer) run() (bool, error) { } firstIteration = false - if err := sleepOrCancel(st.ctx, retryCheckInterval); err != nil { + if err := waitForNextBricklensPoll(st.ctx, pollStarted); err != nil { return false, err } } } +func waitForNextBricklensPoll(ctx context.Context, pollStarted time.Time) error { + return sleepOrCancel(ctx, max(time.Duration(0), bricklensPollInterval-time.Since(pollStarted))) +} + // sleepOrCancel waits for d, or returns early with the context error if the // context is cancelled (e.g. Ctrl-C) so the poll loop exits promptly. func sleepOrCancel(ctx context.Context, d time.Duration) error { @@ -449,7 +484,7 @@ func sleepOrCancel(ctx context.Context, d time.Duration) error { // drainStatic renders a single tail pass without following the run. Success // reflects the run's current result state (empty while active). func (st *bricklensStreamer) drainStatic(toSec int64) (bool, error) { - if err := st.drainTail(toSec); err != nil { + if err := st.drainTail(toSec, false); err != nil { return false, err } if !st.firstLogSeen { @@ -474,7 +509,7 @@ func (req logRequest) tailTarget() int { // drainTail emits the most-recent `target` records oldest-first. Bricklens // returns records newest-first, so it pages until it has `target`, keeps the // newest `target`, and reverses to chronological order. -func (st *bricklensStreamer) drainTail(toSec int64) error { +func (st *bricklensStreamer) drainTail(toSec int64, remember bool) error { target := st.req.tailTarget() if target <= 0 { return nil @@ -500,36 +535,42 @@ func (st *bricklensStreamer) drainTail(toSec int64) error { } for _, c := range slices.Backward(collected) { st.emit(c.Body) + if remember { + st.seen.add(c) + st.lastNano = max(st.lastNano, c.nano()) + } + } + if remember && st.lastNano != 0 { + st.fromSec = max(st.streamStartSec, st.lastNano/1_000_000_000-int64(bricklensLogLookback/time.Second)) } return nil } -// drainPages exhausts all pages from the current from-second in ascending order, -// deduping against the seen-set so a re-queried boundary second is not -// re-printed, then advances fromSec to the newest record's floor-second. -func (st *bricklensStreamer) drainPages(toSec int64) error { +// drainPages exhausts all pages from the current from-second in ascending order. +// Live polls retain a bounded overlap so late records remain visible. The +// initial baseline suppresses omitted history through the tail's watermark. +func (st *bricklensStreamer) drainPages(toSec int64, suppressInitialHistory bool) (int, error) { + emitted := 0 + initialTailNano := st.lastNano + var maximumEvictedNano int64 var pageToken string for { resp, err := st.requestPage(pageToken, toSec, 0, true) if err != nil { - return err + return 0, err } for _, rec := range resp.LogRecords { nano := rec.nano() - if nano != 0 { - // Skip a record older than the last emitted one to keep output - // monotonic (out of order, or a re-queried boundary record). - if st.lastNano != 0 && nano < st.lastNano { - continue - } - if st.seen.has(nano, rec.Body) { - continue - } + if st.seen.has(rec) { + continue + } + if !suppressInitialHistory || (nano != 0 && nano > initialTailNano) { + st.emit(rec.Body) + emitted++ } - st.emit(rec.Body) + maximumEvictedNano = max(maximumEvictedNano, st.seen.add(rec)) if nano != 0 { - st.seen.add(nano, rec.Body) st.lastNano = max(st.lastNano, nano) } } @@ -541,9 +582,13 @@ func (st *bricklensStreamer) drainPages(toSec int64) error { } if st.lastNano != 0 { - st.fromSec = st.lastNano / 1_000_000_000 + st.fromSec = max(st.streamStartSec, st.lastNano/1_000_000_000-int64(bricklensLogLookback/time.Second)) + st.seen.removeBefore(st.fromSec * 1_000_000_000) + if maximumEvictedNano != 0 { + st.fromSec = max(st.fromSec, maximumEvictedNano/1_000_000_000+1) + } } - return nil + return emitted, nil } // requestPage fetches one page, retrying transient failures up to diff --git a/experimental/air/cmd/logstream_support.go b/experimental/air/cmd/logstream_support.go index ac250547e16..bab2e1214fc 100644 --- a/experimental/air/cmd/logstream_support.go +++ b/experimental/air/cmd/logstream_support.go @@ -8,44 +8,72 @@ import ( "time" ) -// seenNano keys the dedup set. Distinct lines can share a nano (each rank stamps -// from its own clock), so the body disambiguates them. -type seenNano struct { +// seenRecord keys the dedup set. Older responses without record IDs fall back +// to timestamp and body, since distinct lines can share a timestamp. +type seenRecord struct { + recordID string + nano int64 + body string +} + +type seenEntry struct { + key seenRecord nano int64 - body string } // seenSet is an insertion-ordered set bounded to a capacity, evicting the // oldest-inserted entry first. type seenSet struct { cap int - items map[seenNano]*list.Element + items map[seenRecord]*list.Element order *list.List } func newSeenSet(capacity int) *seenSet { return &seenSet{ cap: capacity, - items: make(map[seenNano]*list.Element), + items: make(map[seenRecord]*list.Element), order: list.New(), } } -func (s *seenSet) has(nano int64, body string) bool { - _, ok := s.items[seenNano{nano, body}] +func seenRecordKey(record logRecord) seenRecord { + if record.RecordID != "" { + return seenRecord{recordID: record.RecordID} + } + return seenRecord{nano: record.nano(), body: record.Body} +} + +func (s *seenSet) has(record logRecord) bool { + _, ok := s.items[seenRecordKey(record)] return ok } -func (s *seenSet) add(nano int64, body string) { - key := seenNano{nano, body} +func (s *seenSet) add(record logRecord) int64 { + key := seenRecordKey(record) if _, ok := s.items[key]; ok { - return + return 0 } - s.items[key] = s.order.PushBack(key) + s.items[key] = s.order.PushBack(seenEntry{key: key, nano: record.nano()}) if s.order.Len() > s.cap { oldest := s.order.Front() s.order.Remove(oldest) - delete(s.items, oldest.Value.(seenNano)) + evicted := oldest.Value.(seenEntry) + delete(s.items, evicted.key) + return evicted.nano + } + return 0 +} + +func (s *seenSet) removeBefore(nano int64) { + for element := s.order.Front(); element != nil; { + next := element.Next() + entry := element.Value.(seenEntry) + if entry.nano != 0 && entry.nano < nano { + s.order.Remove(element) + delete(s.items, entry.key) + } + element = next } } diff --git a/experimental/air/cmd/logstream_test.go b/experimental/air/cmd/logstream_test.go index 78604fe3f12..e0d691aebdf 100644 --- a/experimental/air/cmd/logstream_test.go +++ b/experimental/air/cmd/logstream_test.go @@ -159,8 +159,7 @@ func TestLogRequestTailTarget(t *testing.T) { func TestDrainPagesDedupAndOrdering(t *testing.T) { // Two pages: page 1 has two ascending records; page 2 repeats the last record - // of page 1 (boundary re-query — must dedup) and includes an older record - // (out of order — must skip), then a genuinely newer one. + // of page 1 (boundary re-query — must dedup), a late record, and a newer one. var page int srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Query().Get("page_token") == "" { @@ -192,15 +191,120 @@ func TestDrainPagesDedupAndOrdering(t *testing.T) { req: logRequest{runID: 1, node: 0, attempt: -1}, seen: newSeenSet(seenRecordsCap), } - require.NoError(t, st.drainPages(0)) + _, err = st.drainPages(0, false) + require.NoError(t, err) require.Equal(t, 2, page) - // "b" prints once (deduped), "stale" is skipped (older than last emitted), and - // fromSec advances to the newest record's floor-second (3000ns -> 0s here). - assert.Equal(t, "a\nb\nc\n", buf.String()) + // "b" prints once while the late record remains visible. + assert.Equal(t, "a\nb\nstale\nc\n", buf.String()) assert.Equal(t, int64(3000), st.lastNano) } +func TestDrainPagesEmitsLateRecordsWithinLookback(t *testing.T) { + var requests int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.HasSuffix(r.URL.Path, "/logs") { + _, _ = w.Write([]byte(`{}`)) + return + } + requests++ + if requests == 1 { + assert.Equal(t, "10", r.URL.Query().Get("from")) + _, _ = w.Write([]byte(`{"log_records":[ + {"record_id":"first","time_unix_nano":50000000000,"body":"same"}, + {"record_id":"second","time_unix_nano":50000000000,"body":"same"}, + {"record_id":"untimed","body":"untimed"} + ]}`)) + return + } + assert.Equal(t, "20", r.URL.Query().Get("from")) + _, _ = w.Write([]byte(`{"log_records":[ + {"record_id":"late","time_unix_nano":30000000000,"body":"late"}, + {"record_id":"first","time_unix_nano":51000000000,"body":"same"}, + {"record_id":"second","time_unix_nano":51000000000,"body":"same"}, + {"record_id":"untimed","body":"untimed"} + ]}`)) + })) + t.Cleanup(srv.Close) + + var buf bytes.Buffer + w := newTestWorkspaceClient(t, srv.URL) + apiClient, err := client.New(w.Config) + require.NoError(t, err) + st := &bricklensStreamer{ + ctx: t.Context(), + w: w, + apiClient: apiClient, + out: &buf, + req: logRequest{runID: 1, node: 0, attempt: -1}, + fromSec: 10, + streamStartSec: 10, + seen: newSeenSet(seenRecordsCap), + } + + _, err = st.drainPages(0, false) + require.NoError(t, err) + _, err = st.drainPages(0, false) + require.NoError(t, err) + assert.Equal(t, "same\nsame\nuntimed\nlate\n", buf.String()) //nolint:dupword +} + +func TestStreamBricklensWaitsForLateTerminalRecords(t *testing.T) { + original := bricklensPollInterval + bricklensPollInterval = time.Millisecond + t.Cleanup(func() { bricklensPollInterval = original }) + + var logRequests int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "/logs"): + logRequests++ + switch logRequests { + case 1: + assert.Equal(t, "2", r.URL.Query().Get("page_size")) + assert.Equal(t, "false", r.URL.Query().Get("ascending")) + _, _ = w.Write([]byte(`{"log_records":[ + {"record_id":"third","time_unix_nano":50000000000,"body":"third"}, + {"record_id":"second","time_unix_nano":40000000000,"body":"second"} + ]}`)) + case 2: + _, _ = w.Write([]byte(`{"log_records":[ + {"record_id":"first","time_unix_nano":30000000000,"body":"first"}, + {"record_id":"second","time_unix_nano":40000000000,"body":"second"}, + {"record_id":"third","time_unix_nano":50000000000,"body":"third"}, + {"record_id":"fourth","time_unix_nano":60000000000,"body":"fourth"} + ]}`)) + case 3: + _, _ = w.Write([]byte(`{"log_records":[ + {"record_id":"late","time_unix_nano":30000000000,"body":"late"}, + {"record_id":"second","time_unix_nano":40000000000,"body":"second"}, + {"record_id":"third","time_unix_nano":50000000000,"body":"third"} + ]}`)) + default: + _, _ = w.Write([]byte(`{"log_records":[]}`)) + } + case r.URL.Path == "/api/2.2/jobs/runs/get": + _, _ = w.Write([]byte(`{"run_id":1,"start_time":10000,"state":{"life_cycle_state":"TERMINATED","result_state":"SUCCESS"}}`)) + default: + _, _ = w.Write([]byte(`{}`)) + } + })) + t.Cleanup(srv.Close) + + var buf bytes.Buffer + ok, err := streamBricklensLogs(t.Context(), newTestWorkspaceClient(t, srv.URL), &buf, + logRequest{runID: 1, node: 0, attempt: -1, tailLines: 2, boundInitialLogs: true, jsonOutput: true}, + logRunStatus{lifeCycleState: "RUNNING", startTimeMs: 10_000}) + require.NoError(t, err) + assert.True(t, ok) + assert.Equal(t, bricklensTerminalEmptyPolls+3, logRequests) + assert.Contains(t, buf.String(), `"line":"second"`) + assert.Contains(t, buf.String(), `"line":"third"`) + assert.Contains(t, buf.String(), `"line":"late"`) + assert.Contains(t, buf.String(), `"line":"fourth"`) + assert.NotContains(t, buf.String(), `"line":"first"`) +} + func TestDisplayState(t *testing.T) { assert.Equal(t, "SUCCESS", logRunStatus{lifeCycleState: "TERMINATED", resultState: "SUCCESS"}.displayState()) assert.Equal(t, "RUNNING", logRunStatus{lifeCycleState: "RUNNING"}.displayState()) @@ -583,21 +687,24 @@ func TestFetchLogsFallsBackToMLflowWhenBricklensEmpty(t *testing.T) { func TestSeenSetEviction(t *testing.T) { s := newSeenSet(2) - s.add(1, "a") - s.add(2, "b") - assert.True(t, s.has(1, "a")) - assert.True(t, s.has(2, "b")) + a := logRecord{TimeUnixNano: "1", Body: "a"} + b := logRecord{TimeUnixNano: "2", Body: "b"} + c := logRecord{TimeUnixNano: "3", Body: "c"} + s.add(a) + s.add(b) + assert.True(t, s.has(a)) + assert.True(t, s.has(b)) // Adding a third evicts the oldest-inserted (1,"a"). - s.add(3, "c") - assert.False(t, s.has(1, "a")) - assert.True(t, s.has(2, "b")) - assert.True(t, s.has(3, "c")) + assert.Equal(t, int64(1), s.add(c)) + assert.False(t, s.has(a)) + assert.True(t, s.has(b)) + assert.True(t, s.has(c)) // Same (nano, body) shares one entry; distinct body under the same nano does not. - s.add(3, "c") - assert.True(t, s.has(3, "c")) - assert.False(t, s.has(3, "d")) + s.add(c) + assert.True(t, s.has(c)) + assert.False(t, s.has(logRecord{TimeUnixNano: "3", Body: "d"})) } func TestSleepOrCancel(t *testing.T) {