diff --git a/cli/sync_test.go b/cli/sync_test.go index 974fdd05f2..dcca986301 100644 --- a/cli/sync_test.go +++ b/cli/sync_test.go @@ -14,6 +14,7 @@ import ( "github.com/coder/coder/v2/agent/agentsocket" "github.com/coder/coder/v2/agent/unit" "github.com/coder/coder/v2/cli/clitest" + "github.com/coder/coder/v2/pty/ptytest" "github.com/coder/coder/v2/testutil" "github.com/coder/quartz" ) @@ -485,6 +486,53 @@ func TestSyncCommands_Golden(t *testing.T) { clitest.TestGoldenFile(t, "TestSyncCommands_Golden/timeline_with_units", outBuf.Bytes(), nil) }) + t.Run("timeline_watch", func(t *testing.T) { + t.Parallel() + path, cleanup := setupSocketServer(t) + defer cleanup() + + ctx, cancel := context.WithCancel(testutil.Context(t, testutil.WaitLong)) + defer cancel() + + client, err := agentsocket.NewClient(ctx, agentsocket.WithPath(path)) + require.NoError(t, err) + defer client.Close() + err = client.SyncStart(ctx, "dep-unit") + require.NoError(t, err) + + inv, _ := clitest.New(t, "exp", "sync", "timeline", "--watch", "--socket-path", path) + pty := ptytest.New(t) + inv.Stdout = pty.Output() + inv.Stderr = pty.Output() + + waiter := clitest.StartWithWaiter(t, inv.WithContext(ctx)) + + // Events recorded before the watch started are rendered first. + pty.ExpectMatch(ctx, "registered (pending)") + pty.ExpectMatch(ctx, "pending → started") + + // Events recorded while watching are streamed as new rows. + err = client.SyncComplete(ctx, "dep-unit") + require.NoError(t, err) + pty.ExpectMatch(ctx, "started → completed") + + // Interrupting the watch is a clean exit. + cancel() + waiter.RequireSuccess() + }) + + t.Run("timeline_watch_rejects_json", func(t *testing.T) { + t.Parallel() + path, cleanup := setupSocketServer(t) + defer cleanup() + + ctx := testutil.Context(t, testutil.WaitShort) + + inv, _ := clitest.New(t, "exp", "sync", "timeline", "--watch", "--output", "json", "--socket-path", path) + err := inv.WithContext(ctx).Run() + require.ErrorContains(t, err, "--watch only supports text output") + }) + t.Run("timeline_json_format", func(t *testing.T) { t.Parallel() path, cleanup := setupSocketServer(t) diff --git a/cli/sync_timeline.go b/cli/sync_timeline.go index 704a83e18d..3f6ceb5768 100644 --- a/cli/sync_timeline.go +++ b/cli/sync_timeline.go @@ -1,8 +1,11 @@ package cli import ( + "context" + "errors" "fmt" "strings" + "time" "golang.org/x/xerrors" @@ -13,6 +16,8 @@ import ( ) func (*RootCmd) syncTimeline(socketPath *string) *serpent.Command { + var watch bool + formatter := cliui.NewOutputFormatter( cliui.ChangeFormatterData( cliui.TextFormat(), @@ -29,10 +34,14 @@ func (*RootCmd) syncTimeline(socketPath *string) *serpent.Command { cmd := &serpent.Command{ Use: "timeline", Short: "Show a timeline of unit state changes", - Long: "Show every recorded unit state transition and dependency declaration across all units, in the order they occurred. The default output is an ASCII graph of the dependency DAG in event-time order; use --output json for the raw event list.", + Long: "Show every recorded unit state transition and dependency declaration across all units, in the order they occurred. The default output is an ASCII graph of the dependency DAG in event-time order; use --output json for the raw event list. With --watch, new rows are streamed as events occur until interrupted.", Handler: func(i *serpent.Invocation) error { ctx := i.Context() + if watch && formatter.FormatID() != "text" { + return xerrors.New("--watch only supports text output") + } + opts := []agentsocket.Option{} if *socketPath != "" { opts = append(opts, agentsocket.WithPath(*socketPath)) @@ -53,6 +62,10 @@ func (*RootCmd) syncTimeline(socketPath *string) *serpent.Command { return xerrors.Errorf("get timeline failed: %w", err) } + if watch { + return watchTimeline(ctx, i, client, events) + } + out, err := formatter.Format(ctx, events) if err != nil { return xerrors.Errorf("format timeline: %w", err) @@ -64,10 +77,52 @@ func (*RootCmd) syncTimeline(socketPath *string) *serpent.Command { }, } + cmd.Options = serpent.OptionSet{ + { + Flag: "watch", + FlagShorthand: "w", + Description: "Stream new rows as events occur until interrupted.", + Value: serpent.BoolOf(&watch), + }, + } formatter.AttachOptions(&cmd.Options) return cmd } +// syncWatchPollInterval is how often --watch polls the agent for new +// events. +const syncWatchPollInterval = time.Second + +// watchTimeline renders the events seen so far, then polls the agent and +// streams newly produced graph rows until the context is canceled. +func watchTimeline(ctx context.Context, i *serpent.Invocation, client *agentsocket.Client, events []agentsocket.UnitEvent) error { + renderer := newTimelineRenderer() + _, _ = fmt.Fprint(i.Stdout, renderer.renderEvents(unitEventsFromSocket(events))) + + ticker := time.NewTicker(syncWatchPollInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + // Interrupting a watch is the normal way to end it. + if errors.Is(ctx.Err(), context.Canceled) { + return nil + } + return ctx.Err() + case <-ticker.C: + events, err := client.SyncTimeline(ctx) + if err != nil { + if errors.Is(ctx.Err(), context.Canceled) { + return nil + } + return xerrors.Errorf("get timeline failed: %w", err) + } + _, _ = fmt.Fprint(i.Stdout, renderer.renderEvents(unitEventsFromSocket(events))) + } + } +} + // unitEventsFromSocket converts socket client event wrappers back to // unit.Event values for rendering and derivation. func unitEventsFromSocket(events []agentsocket.UnitEvent) []unit.Event { diff --git a/cli/sync_timeline_internal_test.go b/cli/sync_timeline_internal_test.go index 650d9aeb12..b4217a803c 100644 --- a/cli/sync_timeline_internal_test.go +++ b/cli/sync_timeline_internal_test.go @@ -1,6 +1,7 @@ package cli import ( + "strings" "testing" "time" @@ -61,6 +62,34 @@ func TestRenderTimeline(t *testing.T) { * [+8.0s] dev-server started → completed`, renderTimeline(events)) }) + t.Run("IncrementalMatchesOneShot", func(t *testing.T) { + t.Parallel() + + events := []unit.Event{ + {Seq: 1, Time: at(0), Kind: unit.EventStatusChange, Unit: "db-migrate", From: unit.StatusNotRegistered, To: unit.StatusPending}, + {Seq: 2, Time: at(100 * time.Millisecond), Kind: unit.EventStatusChange, Unit: "db-migrate", From: unit.StatusPending, To: unit.StatusStarted}, + {Seq: 3, Time: at(100 * time.Millisecond), Kind: unit.EventStatusChange, Unit: "dev-server", From: unit.StatusNotRegistered, To: unit.StatusPending}, + {Seq: 4, Time: at(100 * time.Millisecond), Kind: unit.EventDependencyAdded, Unit: "dev-server", DependsOn: "db-migrate", RequiredStatus: unit.StatusComplete}, + {Seq: 5, Time: at(2300 * time.Millisecond), Kind: unit.EventStatusChange, Unit: "db-migrate", From: unit.StatusStarted, To: unit.StatusComplete}, + {Seq: 6, Time: at(2400 * time.Millisecond), Kind: unit.EventStatusChange, Unit: "dev-server", From: unit.StatusPending, To: unit.StatusStarted}, + {Seq: 7, Time: at(8 * time.Second), Kind: unit.EventStatusChange, Unit: "dev-server", From: unit.StatusStarted, To: unit.StatusComplete}, + } + + oneShot := newTimelineRenderer().renderEvents(events) + + // Feeding the log in overlapping chunks, as --watch does when it + // repeatedly fetches the full event list, must produce identical + // output: already-seen events are skipped by sequence number. + incremental := newTimelineRenderer() + var got strings.Builder + got.WriteString(incremental.renderEvents(events[:2])) + got.WriteString(incremental.renderEvents(events[:2])) // no new events + got.WriteString(incremental.renderEvents(events[:5])) // overlap + connector boundary + got.WriteString(incremental.renderEvents(events)) + + require.Equal(t, oneShot, got.String()) + }) + t.Run("ReadyFlipNotDrawnWhenStillBlocked", func(t *testing.T) { t.Parallel() diff --git a/cli/sync_timeline_render.go b/cli/sync_timeline_render.go index 9560bbd4c3..863b54b101 100644 --- a/cli/sync_timeline_render.go +++ b/cli/sync_timeline_render.go @@ -29,129 +29,165 @@ func syncElapsed(base, t time.Time) string { return fmt.Sprintf("+%.1fs", t.Sub(base).Seconds()) } -// renderTimeline renders the unit event log as a git-log-style ASCII +// timelineEdge is one declared dependency of a unit. +type timelineEdge struct { + dependsOn unit.ID + requiredStatus unit.Status +} + +// timelineRenderer renders the unit event log as a git-log-style ASCII // graph. Each unit gets a lane in order of first appearance; every event // prints a "*" row in its unit's lane with an elapsed-time label. When a // status change satisfies another unit's dependencies, a connector row is // drawn from the satisfying lane to the dependent lane, followed by a // derived "ready" row. Lanes disappear once a unit completes. -func renderTimeline(events []unit.Event) string { - if len(events) == 0 { - return "No events found" - } - - type edge struct { - dependsOn unit.ID - requiredStatus unit.Status - } - var ( - lanes []unit.ID - laneOf = make(map[unit.ID]int) - active = make(map[unit.ID]bool) - statuses = make(map[unit.ID]unit.Status) - deps = make(map[unit.ID][]edge) - dependents = make(map[unit.ID][]unit.ID) - ) - - ensureLane := func(u unit.ID) { - if _, ok := laneOf[u]; ok { - return - } - laneOf[u] = len(lanes) - lanes = append(lanes, u) - active[u] = true - } - - // isReady mirrors the Manager's readiness semantics: a pending unit - // is ready when every dependency's current status equals its - // required status. - isReady := func(u unit.ID) bool { - if statuses[u] != unit.StatusPending { - return false - } - for _, e := range deps[u] { - if statuses[e.dependsOn] != e.requiredStatus { - return false - } - } - return true - } +// +// The renderer is incremental: renderEvents may be called repeatedly with +// a growing event log and only renders rows for events it has not seen +// yet, which is how --watch streams new rows as they occur. +type timelineRenderer struct { + lanes []unit.ID + laneOf map[unit.ID]int + active map[unit.ID]bool + statuses map[unit.ID]unit.Status + deps map[unit.ID][]timelineEdge + dependents map[unit.ID][]unit.ID + + // base is the time of the first event seen; elapsed labels are + // relative to it. + base time.Time + started bool + // lastSeq is the highest event sequence number consumed so far. + lastSeq uint64 +} - // glyphRow renders one graph row: mark in the target unit's lane, - // "|" in other active lanes, and spaces elsewhere. - glyphRow := func(target unit.ID, mark string) string { - glyphs := make([]string, len(lanes)) - for i, u := range lanes { - switch { - case u == target: - glyphs[i] = mark - case active[u]: - glyphs[i] = "|" - default: - glyphs[i] = " " - } - } - return strings.TrimRight(strings.Join(glyphs, " "), " ") +func newTimelineRenderer() *timelineRenderer { + return &timelineRenderer{ + laneOf: make(map[unit.ID]int), + active: make(map[unit.ID]bool), + statuses: make(map[unit.ID]unit.Status), + deps: make(map[unit.ID][]timelineEdge), + dependents: make(map[unit.ID][]unit.ID), } +} - // connectorRow draws the dependency-satisfaction edge from lane - // "from" toward lane "to". - connectorRow := func(from, to unit.ID) string { - glyphs := make([]string, len(lanes)) - for i, u := range lanes { - switch { - case u == from && laneOf[from] < laneOf[to]: - glyphs[i] = `\` - case u == from: - glyphs[i] = "/" - case active[u]: - glyphs[i] = "|" - default: - glyphs[i] = " " - } - } - return strings.TrimRight(strings.Join(glyphs, " "), " ") +// renderTimeline renders a complete event log in one shot. +func renderTimeline(events []unit.Event) string { + if len(events) == 0 { + return "No events found" } + return strings.TrimRight(newTimelineRenderer().renderEvents(events), "\n") +} - base := events[0].Time +// renderEvents consumes events with a sequence number greater than any +// previously consumed and returns the newly produced graph rows, +// newline-terminated. Events must be ordered by Seq. +func (r *timelineRenderer) renderEvents(events []unit.Event) string { var sb strings.Builder - writeRow := func(graph string, at time.Time, u unit.ID, description string) { - _, _ = sb.WriteString(fmt.Sprintf("%s [%s] %s %s\n", graph, syncElapsed(base, at), u, description)) - } - for _, ev := range events { - ensureLane(ev.Unit) + if ev.Seq <= r.lastSeq { + continue + } + r.lastSeq = ev.Seq + if !r.started { + r.base = ev.Time + r.started = true + } + r.ensureLane(ev.Unit) switch ev.Kind { case unit.EventStatusChange: // Snapshot dependent readiness before applying the status // change so blocked-to-ready flips can be detected. - wasReady := make(map[unit.ID]bool, len(dependents[ev.Unit])) - for _, dep := range dependents[ev.Unit] { - wasReady[dep] = isReady(dep) + wasReady := make(map[unit.ID]bool, len(r.dependents[ev.Unit])) + for _, dep := range r.dependents[ev.Unit] { + wasReady[dep] = r.isReady(dep) } - statuses[ev.Unit] = ev.To - writeRow(glyphRow(ev.Unit, "*"), ev.Time, ev.Unit, syncEventDescription(ev)) + r.statuses[ev.Unit] = ev.To + r.writeRow(&sb, r.glyphRow(ev.Unit, "*"), ev.Time, ev.Unit, syncEventDescription(ev)) if ev.To == unit.StatusComplete { - active[ev.Unit] = false + r.active[ev.Unit] = false } - for _, dep := range dependents[ev.Unit] { - if !wasReady[dep] && isReady(dep) { - _, _ = sb.WriteString(connectorRow(ev.Unit, dep) + "\n") - writeRow(glyphRow(dep, "*"), ev.Time, dep, "ready (dependency satisfied)") + for _, dep := range r.dependents[ev.Unit] { + if !wasReady[dep] && r.isReady(dep) { + _, _ = sb.WriteString(r.connectorRow(ev.Unit, dep) + "\n") + r.writeRow(&sb, r.glyphRow(dep, "*"), ev.Time, dep, "ready (dependency satisfied)") } } case unit.EventDependencyAdded: - deps[ev.Unit] = append(deps[ev.Unit], edge{ + r.deps[ev.Unit] = append(r.deps[ev.Unit], timelineEdge{ dependsOn: ev.DependsOn, requiredStatus: ev.RequiredStatus, }) - dependents[ev.DependsOn] = append(dependents[ev.DependsOn], ev.Unit) - writeRow(glyphRow(ev.Unit, "*"), ev.Time, ev.Unit, syncEventDescription(ev)) + r.dependents[ev.DependsOn] = append(r.dependents[ev.DependsOn], ev.Unit) + r.writeRow(&sb, r.glyphRow(ev.Unit, "*"), ev.Time, ev.Unit, syncEventDescription(ev)) + } + } + return sb.String() +} + +func (r *timelineRenderer) ensureLane(u unit.ID) { + if _, ok := r.laneOf[u]; ok { + return + } + r.laneOf[u] = len(r.lanes) + r.lanes = append(r.lanes, u) + r.active[u] = true +} + +// isReady mirrors the Manager's readiness semantics: a pending unit is +// ready when every dependency's current status equals its required +// status. +func (r *timelineRenderer) isReady(u unit.ID) bool { + if r.statuses[u] != unit.StatusPending { + return false + } + for _, e := range r.deps[u] { + if r.statuses[e.dependsOn] != e.requiredStatus { + return false + } + } + return true +} + +// glyphRow renders one graph row: mark in the target unit's lane, "|" in +// other active lanes, and spaces elsewhere. +func (r *timelineRenderer) glyphRow(target unit.ID, mark string) string { + glyphs := make([]string, len(r.lanes)) + for i, u := range r.lanes { + switch { + case u == target: + glyphs[i] = mark + case r.active[u]: + glyphs[i] = "|" + default: + glyphs[i] = " " + } + } + return strings.TrimRight(strings.Join(glyphs, " "), " ") +} + +// connectorRow draws the dependency-satisfaction edge from lane "from" +// toward lane "to". +func (r *timelineRenderer) connectorRow(from, to unit.ID) string { + glyphs := make([]string, len(r.lanes)) + for i, u := range r.lanes { + switch { + case u == from && r.laneOf[from] < r.laneOf[to]: + glyphs[i] = `\` + case u == from: + glyphs[i] = "/" + case r.active[u]: + glyphs[i] = "|" + default: + glyphs[i] = " " } } + return strings.TrimRight(strings.Join(glyphs, " "), " ") +} - return strings.TrimRight(sb.String(), "\n") +func (r *timelineRenderer) writeRow(sb *strings.Builder, graph string, at time.Time, u unit.ID, description string) { + _, _ = sb.WriteString(fmt.Sprintf("%s [%s] %s %s\n", graph, syncElapsed(r.base, at), u, description)) } diff --git a/cli/testdata/coder_exp_sync_timeline_--help.golden b/cli/testdata/coder_exp_sync_timeline_--help.golden index de88089932..64a10183eb 100644 --- a/cli/testdata/coder_exp_sync_timeline_--help.golden +++ b/cli/testdata/coder_exp_sync_timeline_--help.golden @@ -8,11 +8,14 @@ USAGE: Show every recorded unit state transition and dependency declaration across all units, in the order they occurred. The default output is an ASCII graph of the dependency DAG in event-time order; use --output json for the raw event - list. + list. With --watch, new rows are streamed as events occur until interrupted. OPTIONS: -o, --output text|json (default: text) Output format. + -w, --watch bool + Stream new rows as events occur until interrupted. + ——— Run `coder --help` for a list of global options.