From 18ea2f6dbe18ea1473755b183ff8ff661e427211 Mon Sep 17 00:00:00 2001 From: Gustavo Bertoi Date: Wed, 1 Jul 2026 00:39:57 -0300 Subject: [PATCH] feat(logs): log streaming + Bubble Tea dashboard cockpit (spec 16) Implements spec 16's read-only observation surface: - `logs [service...]`: multiplexes container logs across the workspace's project + shared stacks via the read-only Engine SDK, filtered on the tool-owned label + an optional per-service filter (shared alias, bare engine, service, or project name). Flags --follow/-f, --tail, --since, --timestamps, --no-color, and a --json {ts,service,project,stream, container,line} line contract. TTY gets a color-keyed gutter; a pipe gets plain lines. Non-follow exits at EOF; -f is signal-aware for clean teardown. - docker.Client gains ContainerLogStream: a demuxed (stdcopy for non-TTY, raw for TTY), optionally-following/timestamped line channel with bounded fan-in backpressure; TTY-ness learned via inspect; unreadable logging drivers surface a one-liner. Mirrored in MockClient (Streams seam). - `dashboard`: a Bubble Tea v2 cockpit (bubbles/v2 table + lipgloss/v2) showing shared + project services with state/health + refs + URL and a log tail pane, keybindings (q/esc/ctrl+c quit, r refresh), 2s safety poll. Non-TTY / --json / --quiet refuses the TUI and prints a one-shot snapshot reusing the shared-status + per-project projections. Removes the logs + dashboard stubs. Tests: log stream demux/emitter/ctx, target label-filter resolution, --json line shape, multiplex over the mock, dashboard model Update/View transitions, and the non-TTY snapshot path. No real daemon required. Co-Authored-By: Claude Opus 4.8 (1M context) --- go.mod | 4 +- internal/cli/dashboard.go | 158 ++++++++++++++ internal/cli/dashboard_model.go | 258 ++++++++++++++++++++++ internal/cli/dashboard_model_test.go | 130 +++++++++++ internal/cli/dashboard_test.go | 130 +++++++++++ internal/cli/logs.go | 311 +++++++++++++++++++++++++++ internal/cli/logs_test.go | 168 +++++++++++++++ internal/cli/root.go | 2 + internal/cli/stubs.go | 8 +- internal/docker/docker.go | 26 +++ internal/docker/inspect.go | 1 + internal/docker/logstream.go | 141 ++++++++++++ internal/docker/logstream_test.go | 89 ++++++++ internal/docker/mock.go | 27 +++ 14 files changed, 1446 insertions(+), 7 deletions(-) create mode 100644 internal/cli/dashboard.go create mode 100644 internal/cli/dashboard_model.go create mode 100644 internal/cli/dashboard_model_test.go create mode 100644 internal/cli/dashboard_test.go create mode 100644 internal/cli/logs.go create mode 100644 internal/cli/logs_test.go create mode 100644 internal/docker/logstream.go create mode 100644 internal/docker/logstream_test.go diff --git a/go.mod b/go.mod index 7f61ab1..97907d2 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,8 @@ module github.com/open-source-cloud/devstack go 1.25.8 require ( + charm.land/bubbles/v2 v2.0.0 + charm.land/bubbletea/v2 v2.0.2 charm.land/fang/v2 v2.0.1 charm.land/huh/v2 v2.0.3 charm.land/lipgloss/v2 v2.0.1 @@ -35,8 +37,6 @@ require ( ) require ( - charm.land/bubbles/v2 v2.0.0 // indirect - charm.land/bubbletea/v2 v2.0.2 // indirect filippo.io/hpke v0.4.0 // indirect github.com/Microsoft/go-winio v0.6.2 // indirect github.com/atotto/clipboard v0.1.4 // indirect diff --git a/internal/cli/dashboard.go b/internal/cli/dashboard.go new file mode 100644 index 0000000..df4a86f --- /dev/null +++ b/internal/cli/dashboard.go @@ -0,0 +1,158 @@ +package cli + +import ( + "context" + "fmt" + "os" + "sort" + "time" + + tea "charm.land/bubbletea/v2" + "github.com/spf13/cobra" + + "github.com/open-source-cloud/devstack/internal/docker" + "github.com/open-source-cloud/devstack/internal/workspace" +) + +// dashboardPoll is the default safety-poll cadence (spec 16: event-driven + a 2s +// fallback; the model uses the fallback poll for its live refresh). +const dashboardPoll = 2 * time.Second + +// newDashboardCmd wires `devstack dashboard` (spec 16): a Bubble Tea cockpit over +// the read-only Engine SDK + the ledger. On a non-TTY (or --json/--quiet) it +// refuses to launch the TUI and prints a one-shot status snapshot instead — the +// scriptable, non-interactive equivalent. +func newDashboardCmd(g *GlobalOpts) *cobra.Command { + var noStats bool + cmd := &cobra.Command{ + Use: "dashboard", + Short: "Live TUI cockpit: shared + project services, health, and a log tail", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + mgr, closeFn, err := buildManager(cmd) + if err != nil { + return err + } + defer closeFn() + + // Best-effort self-heal so ref counts are truthful (lock-free reads, + // this reconcile takes the lock only if it prunes — same as status). + _, _ = mgr.Reconcile(cmd.Context()) + + if !dashboardInteractive(cmd, g) { + return printDashboardSnapshot(cmd, g, mgr) + } + + ctx := cmd.Context() + fetch := func(c context.Context) dashboardData { return collectDashboardData(c, mgr) } + model := newDashboardModel(ctx, fetch, dashboardPoll) + _, err = tea.NewProgram(model, tea.WithContext(ctx)).Run() + return err + }, + } + cmd.Flags().BoolVar(&noStats, "no-stats", false, "reserved: disable the CPU/mem stats stream (stats are opt-in in this build)") + return cmd +} + +// dashboardInteractive reports whether the TUI may launch: a real stdout TTY and +// neither --json nor --quiet requested. +func dashboardInteractive(cmd *cobra.Command, g *GlobalOpts) bool { + if g.JSON || g.Quiet { + return false + } + f, ok := cmd.OutOrStdout().(*os.File) + return ok && isTerminal(f) +} + +// printDashboardSnapshot is the non-TTY fallback: a one-shot projection reusing +// the same shared-status + per-project views as `status`, plus a redirect to the +// scriptable commands. +func printDashboardSnapshot(cmd *cobra.Command, g *GlobalOpts, mgr *workspace.Manager) error { + ctx := cmd.Context() + projects := collectProjectStatus(ctx, mgr) + shared, err := mgr.Status() + if err != nil { + return err + } + if g.JSON { + return writeJSON(cmd, map[string]any{"projects": projects, "shared": shared}) + } + if g.Quiet { + return nil + } + w := cmd.OutOrStdout() + fmt.Fprintln(w, "dashboard needs an interactive terminal; showing a one-shot snapshot.") + fmt.Fprintln(w, "use `devstack logs` (follow) or `devstack status --json` for non-interactive output.") + fmt.Fprintln(w) + renderStatus(cmd, projects, shared) + return nil +} + +// collectDashboardData is the read-only collector: it fans in shared-service rows +// (ledger), per-project service rows (live containers + health), and a bounded +// tail of recent log lines into one snapshot. Lock-free. +func collectDashboardData(ctx context.Context, mgr *workspace.Manager) dashboardData { + var data dashboardData + + shared, err := mgr.Status() + if err != nil { + data.Err = err.Error() + } + for _, s := range shared { + data.Rows = append(data.Rows, dashRow{ + Name: s.Alias, + Kind: "shared", + State: s.Status, + Refs: s.RefCount, + Projects: s.Projects, + Engine: dashEngine(s), + }) + } + + for _, p := range collectProjectStatus(ctx, mgr) { + for _, svc := range p.Services { + data.Rows = append(data.Rows, dashRow{ + Name: p.Project + "/" + svc.Name, + Kind: "project", + State: svc.State, + Health: svc.Health, + URL: fmt.Sprintf("https://%s.%s.localhost", svc.Name, p.Project), + }) + } + } + + data.Logs = collectRecentLogs(ctx, mgr.Docker, 8) + return data +} + +// dashEngine renders a shared row's "engine version" detail. +func dashEngine(s workspace.SharedStatus) string { + if s.Major == "" || s.Major == "default" { + return s.Engine + } + return s.Engine + " " + s.Major +} + +// collectRecentLogs pulls up to `tail` trailing lines from each managed container +// (non-follow) into a service-tagged, bounded slice for the log pane. Best-effort: +// an unreadable service is skipped, not fatal. +func collectRecentLogs(ctx context.Context, client docker.Client, tail int) []dashLog { + targets, err := resolveLogTargets(ctx, client, nil) + if err != nil { + return nil + } + var out []dashLog + for _, t := range targets { + ch, err := client.ContainerLogStream(ctx, t.ID, docker.LogOptions{Tail: tail}) + if err != nil { + continue + } + for ll := range ch { + out = append(out, dashLog{Service: t.Service, Line: ll.Text}) + } + } + // Deterministic-ish ordering: group by service (arrival order within a + // service is preserved by the stream). + sort.SliceStable(out, func(i, j int) bool { return out[i].Service < out[j].Service }) + return clampLogs(out) +} diff --git a/internal/cli/dashboard_model.go b/internal/cli/dashboard_model.go new file mode 100644 index 0000000..cfe688f --- /dev/null +++ b/internal/cli/dashboard_model.go @@ -0,0 +1,258 @@ +package cli + +import ( + "context" + "fmt" + "strings" + "time" + + table "charm.land/bubbles/v2/table" + tea "charm.land/bubbletea/v2" + "charm.land/lipgloss/v2" +) + +// dashRow is one service row in the cockpit's top-left table (spec 16 layout). +type dashRow struct { + Name string // shared-postgres | api/php + Kind string // "shared" | "project" + State string // running | exited | ... + Health string // starting|healthy|unhealthy|"" (no healthcheck) + Refs int // shared: referencing-project count + Projects []string // shared: the referencing projects + Engine string // shared: engine + version + URL string // project: https://..localhost +} + +// dashLog is one recent, service-tagged log line for the bottom pane. +type dashLog struct { + Service string + Line string +} + +// dashboardData is the read-only snapshot the collector fans out to the model: +// service rows + a bounded tail of recent log lines. It is produced lock-free. +type dashboardData struct { + Rows []dashRow + Logs []dashLog + Err string // a collector-level error to surface in the footer +} + +// dashDataMsg delivers a fresh snapshot into the model's Update loop. +type dashDataMsg dashboardData + +// dashTickMsg is the safety-poll cadence (event streams are out of scope for the +// unit-testable model; a steady poll keeps the cockpit live). +type dashTickMsg struct{} + +const dashLogCap = 500 // bounded ring for the log pane + +// dashboardModel is the Bubble Tea cockpit. It is fully unit-testable without a +// TTY: construct it with a canned fetch, drive Update with messages, and inspect +// the table rows / View output. +type dashboardModel struct { + ctx context.Context + fetch func(context.Context) dashboardData + poll time.Duration + theme dashTheme + table table.Model + rows []dashRow + logs []dashLog + width int + height int + err string + quit bool +} + +// dashTheme carries the few styles the cockpit needs so it tracks internal/prompt. +type dashTheme struct { + Title lipgloss.Style + Detail lipgloss.Style + Footer lipgloss.Style + Border lipgloss.Style +} + +func defaultDashTheme() dashTheme { + return dashTheme{ + Title: lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color("63")), + Detail: lipgloss.NewStyle().Foreground(lipgloss.Color("252")), + Footer: lipgloss.NewStyle().Faint(true), + Border: lipgloss.NewStyle().Border(lipgloss.RoundedBorder()).BorderForeground(lipgloss.Color("63")), + } +} + +// newDashboardModel builds the cockpit around an injected fetch (the read-only +// collector). poll<=0 disables the safety poll (used in tests). +func newDashboardModel(ctx context.Context, fetch func(context.Context) dashboardData, poll time.Duration) dashboardModel { + t := table.New( + table.WithColumns([]table.Column{ + {Title: "SERVICE", Width: 22}, + {Title: "STATE", Width: 10}, + {Title: "HEALTH", Width: 10}, + {Title: "REFS", Width: 18}, + }), + table.WithFocused(true), + table.WithHeight(10), + ) + return dashboardModel{ctx: ctx, fetch: fetch, poll: poll, theme: defaultDashTheme(), table: t} +} + +func (m dashboardModel) Init() tea.Cmd { + return tea.Batch(m.fetchCmd(), m.tickCmd()) +} + +// fetchCmd runs the read-only collector off the render loop and delivers a snapshot. +func (m dashboardModel) fetchCmd() tea.Cmd { + return func() tea.Msg { + if m.fetch == nil { + return dashDataMsg{} + } + return dashDataMsg(m.fetch(m.ctx)) + } +} + +func (m dashboardModel) tickCmd() tea.Cmd { + if m.poll <= 0 { + return nil + } + return tea.Tick(m.poll, func(time.Time) tea.Msg { return dashTickMsg{} }) +} + +func (m dashboardModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) { + switch msg := msg.(type) { + case dashDataMsg: + m.rows = msg.Rows + m.logs = clampLogs(msg.Logs) + m.err = msg.Err + m.table.SetRows(dashTableRows(m.rows)) + return m, nil + case dashTickMsg: + return m, tea.Batch(m.fetchCmd(), m.tickCmd()) + case tea.WindowSizeMsg: + m.width, m.height = msg.Width, msg.Height + m.resize() + return m, nil + case tea.KeyPressMsg: + switch msg.String() { + case "q", "ctrl+c", "esc": + m.quit = true + return m, tea.Quit + case "r": + return m, m.fetchCmd() + } + } + var cmd tea.Cmd + m.table, cmd = m.table.Update(msg) + return m, cmd +} + +// resize distributes the terminal size between the table and the log pane. +func (m *dashboardModel) resize() { + if m.width > 0 { + m.table.SetWidth(m.width) + } + // Reserve the top table (~half, min 4 rows) and leave the rest for logs. + th := m.height/2 - 2 + if th < 4 { + th = 4 + } + m.table.SetHeight(th) +} + +func (m dashboardModel) View() tea.View { + var b strings.Builder + b.WriteString(m.theme.Title.Render("devstack dashboard")) + b.WriteString(" ") + b.WriteString(m.theme.Footer.Render("read-only · no daemon · command-scoped")) + b.WriteString("\n\n") + b.WriteString(m.table.View()) + b.WriteString("\n\n") + + // Detail for the selected row. + if d := m.selectedDetail(); d != "" { + b.WriteString(m.theme.Detail.Render(d)) + b.WriteString("\n\n") + } + + // Log pane. + b.WriteString(m.theme.Title.Render("logs")) + b.WriteString("\n") + if len(m.logs) == 0 { + b.WriteString(m.theme.Footer.Render(" (no recent log lines)")) + b.WriteString("\n") + } + for _, l := range m.logsTail() { + fmt.Fprintf(&b, "%-18s │ %s\n", l.Service, l.Line) + } + if m.err != "" { + b.WriteString("\n") + b.WriteString(m.theme.Footer.Render("! " + m.err)) + b.WriteString("\n") + } + b.WriteString("\n") + b.WriteString(m.theme.Footer.Render("[r]efresh [j/k]scroll [q]uit")) + + v := tea.NewView(b.String()) + v.AltScreen = true + return v +} + +// selectedDetail renders the right-pane detail for the highlighted service. +func (m dashboardModel) selectedDetail() string { + i := m.table.Cursor() + if i < 0 || i >= len(m.rows) { + return "" + } + r := m.rows[i] + var parts []string + parts = append(parts, "▸ "+r.Name) + if r.Engine != "" { + parts = append(parts, "engine: "+r.Engine) + } + if r.Kind == "shared" { + parts = append(parts, fmt.Sprintf("refs: %d %v", r.Refs, r.Projects)) + } + if r.URL != "" { + parts = append(parts, "url: "+r.URL) + } + state := r.State + if r.Health != "" { + state += " (" + r.Health + ")" + } + parts = append(parts, "state: "+state) + return strings.Join(parts, " ") +} + +// logsTail returns the last visible slice of the log ring for the pane height. +func (m dashboardModel) logsTail() []dashLog { + n := m.height/2 - 4 + if n < 3 { + n = 8 + } + if len(m.logs) <= n { + return m.logs + } + return m.logs[len(m.logs)-n:] +} + +// dashTableRows projects rows into the bubbles table shape. +func dashTableRows(rows []dashRow) []table.Row { + out := make([]table.Row, 0, len(rows)) + for _, r := range rows { + refs := "" + switch { + case r.Kind == "shared": + refs = fmt.Sprintf("refs=%d", r.Refs) + case r.URL != "": + refs = r.URL + } + out = append(out, table.Row{r.Name, r.State, r.Health, refs}) + } + return out +} + +func clampLogs(logs []dashLog) []dashLog { + if len(logs) <= dashLogCap { + return logs + } + return logs[len(logs)-dashLogCap:] +} diff --git a/internal/cli/dashboard_model_test.go b/internal/cli/dashboard_model_test.go new file mode 100644 index 0000000..10a3e63 --- /dev/null +++ b/internal/cli/dashboard_model_test.go @@ -0,0 +1,130 @@ +package cli + +import ( + "context" + "strings" + "testing" + + tea "charm.land/bubbletea/v2" +) + +func sampleData() dashboardData { + return dashboardData{ + Rows: []dashRow{ + {Name: "shared-postgres", Kind: "shared", State: "running", Refs: 2, Projects: []string{"api", "web"}, Engine: "postgres 18"}, + {Name: "shop/api", Kind: "project", State: "running", Health: "healthy", URL: "https://api.shop.localhost"}, + }, + Logs: []dashLog{{Service: "shop/api", Line: "GET /healthz 200"}}, + } +} + +func newTestModel() dashboardModel { + return newDashboardModel(context.Background(), func(context.Context) dashboardData { return sampleData() }, 0) +} + +// TestDashboardDataMsg feeds a snapshot and asserts the table rows are populated. +func TestDashboardDataMsg(t *testing.T) { + m := newTestModel() + updated, _ := m.Update(dashDataMsg(sampleData())) + dm := updated.(dashboardModel) + if len(dm.rows) != 2 { + t.Fatalf("rows = %d, want 2", len(dm.rows)) + } + if got := len(dm.table.Rows()); got != 2 { + t.Fatalf("table rows = %d, want 2", got) + } + if len(dm.logs) != 1 { + t.Fatalf("logs = %d, want 1", len(dm.logs)) + } +} + +// TestDashboardQuitKey verifies q sets quitting and returns tea.Quit. +func TestDashboardQuitKey(t *testing.T) { + for _, key := range []string{"q", "ctrl+c", "esc"} { + m := newTestModel() + updated, cmd := m.Update(keyPress(key)) + dm := updated.(dashboardModel) + if !dm.quit { + t.Fatalf("key %q did not set quit", key) + } + if cmd == nil { + t.Fatalf("key %q returned nil cmd (want tea.Quit)", key) + } + if _, ok := cmd().(tea.QuitMsg); !ok { + t.Fatalf("key %q cmd is not tea.Quit", key) + } + } +} + +// TestDashboardRefreshKey verifies r yields a fetch command producing a dashDataMsg. +func TestDashboardRefreshKey(t *testing.T) { + m := newTestModel() + _, cmd := m.Update(keyPress("r")) + if cmd == nil { + t.Fatal("r returned nil cmd") + } + if _, ok := cmd().(dashDataMsg); !ok { + t.Fatalf("r cmd did not produce dashDataMsg, got %T", cmd()) + } +} + +// TestDashboardResize verifies WindowSizeMsg records dimensions. +func TestDashboardResize(t *testing.T) { + m := newTestModel() + updated, _ := m.Update(tea.WindowSizeMsg{Width: 120, Height: 40}) + dm := updated.(dashboardModel) + if dm.width != 120 || dm.height != 40 { + t.Fatalf("dims = %dx%d, want 120x40", dm.width, dm.height) + } +} + +// TestDashboardView asserts the rendered view is non-empty and shows key content. +func TestDashboardView(t *testing.T) { + m := newTestModel() + updated, _ := m.Update(dashDataMsg(sampleData())) + dm := updated.(dashboardModel) + dm2, _ := dm.Update(tea.WindowSizeMsg{Width: 100, Height: 30}) + view := dm2.(dashboardModel).View() + if strings.TrimSpace(view.Content) == "" { + t.Fatal("view is empty") + } + for _, want := range []string{"dashboard", "shared-postgres", "logs", "[q]uit"} { + if !strings.Contains(view.Content, want) { + t.Errorf("view missing %q", want) + } + } + if !view.AltScreen { + t.Error("dashboard view should request AltScreen") + } +} + +// TestDashboardSelectedDetail checks the right-pane detail for the highlighted row. +func TestDashboardSelectedDetail(t *testing.T) { + m := newTestModel() + updated, _ := m.Update(dashDataMsg(sampleData())) + dm := updated.(dashboardModel) + detail := dm.selectedDetail() + if !strings.Contains(detail, "shared-postgres") || !strings.Contains(detail, "refs: 2") { + t.Fatalf("detail = %q", detail) + } +} + +// TestDashboardInitBatch verifies Init issues a fetch (poll=0 → no tick). +func TestDashboardInitBatch(t *testing.T) { + m := newTestModel() + if cmd := m.Init(); cmd == nil { + t.Fatal("Init returned nil cmd") + } +} + +func keyPress(s string) tea.KeyPressMsg { + switch s { + case "ctrl+c": + return tea.KeyPressMsg{Mod: tea.ModCtrl, Code: 'c'} + case "esc": + return tea.KeyPressMsg{Code: tea.KeyEscape} + default: + r := []rune(s)[0] + return tea.KeyPressMsg{Code: r, Text: s} + } +} diff --git a/internal/cli/dashboard_test.go b/internal/cli/dashboard_test.go new file mode 100644 index 0000000..08a3557 --- /dev/null +++ b/internal/cli/dashboard_test.go @@ -0,0 +1,130 @@ +package cli + +import ( + "bytes" + "encoding/json" + "path/filepath" + "strings" + "testing" +) + +func TestLogsDashboardRegistered(t *testing.T) { + for _, name := range []string{"logs", "dashboard"} { + if !findCmd(t, name) { + t.Errorf("command %q is not registered as a real RunE command", name) + } + } +} + +// TestDashboardInteractiveDecision covers the non-TTY / --json / --quiet gate. +func TestDashboardInteractiveDecision(t *testing.T) { + root := NewRootCmd(Options{}) + root.SetOut(&bytes.Buffer{}) // a buffer is never a TTY + tests := []struct { + name string + g *GlobalOpts + want bool + }{ + {"buffer-out", &GlobalOpts{}, false}, + {"json", &GlobalOpts{JSON: true}, false}, + {"quiet", &GlobalOpts{Quiet: true}, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := dashboardInteractive(root, tt.g); got != tt.want { + t.Fatalf("dashboardInteractive = %v, want %v", got, tt.want) + } + }) + } +} + +// seedDashWorkspace seeds a minimal workspace + isolated ledger under a temp dir, +// chdir'ing into it so config discovery + the ledger both resolve locally. +func seedDashWorkspace(t *testing.T) string { + t.Helper() + dir := t.TempDir() + mustWrite(t, filepath.Join(dir, "workspace.yaml"), + "apiVersion: devstack/v1\nkind: Workspace\nname: demo\nshared:\n postgres: { template: postgres }\nprojects:\n - { name: app, path: app }\n") + mustWrite(t, filepath.Join(dir, "app", "devstack.yaml"), + "apiVersion: devstack/v1\nkind: Project\nname: app\nservices:\n web: { template: node.vite, uses: [workspace.shared.postgres] }\n") + t.Chdir(dir) + t.Setenv("XDG_DATA_HOME", filepath.Join(dir, "data")) + t.Setenv("XDG_RUNTIME_DIR", filepath.Join(dir, "run")) + return dir +} + +// TestDashboardSnapshotNonTTY drives `dashboard` with a buffered (non-TTY) stdout; +// it must print the one-shot snapshot + the redirect note, never the TUI. +func TestDashboardSnapshotNonTTY(t *testing.T) { + seedDashWorkspace(t) + var out bytes.Buffer + root := NewRootCmd(Options{}) + root.SetArgs([]string{"dashboard"}) + root.SetOut(&out) + root.SetErr(&out) + if err := root.Execute(); err != nil { + t.Fatalf("dashboard: %v\n%s", err, out.String()) + } + s := out.String() + if !strings.Contains(s, "interactive terminal") || !strings.Contains(s, "devstack logs") { + t.Fatalf("snapshot missing redirect note:\n%s", s) + } + if !strings.Contains(s, "PROJECTS") { + t.Fatalf("snapshot missing status projection:\n%s", s) + } +} + +// TestDashboardJSONSnapshot asserts --json emits a machine-readable snapshot. +func TestDashboardJSONSnapshot(t *testing.T) { + seedDashWorkspace(t) + var out bytes.Buffer + root := NewRootCmd(Options{}) + root.SetArgs([]string{"dashboard", "--json"}) + root.SetOut(&out) + root.SetErr(&out) + if err := root.Execute(); err != nil { + t.Fatalf("dashboard --json: %v\n%s", err, out.String()) + } + var got map[string]any + if err := json.Unmarshal(out.Bytes(), &got); err != nil { + t.Fatalf("--json output is not valid JSON: %v\n%s", err, out.String()) + } + if _, ok := got["projects"]; !ok { + t.Errorf("json snapshot missing 'projects' key: %v", got) + } + if _, ok := got["shared"]; !ok { + t.Errorf("json snapshot missing 'shared' key: %v", got) + } +} + +// TestLogsNoTargetsNotice covers the zero-state message contract directly (the +// end-to-end path is daemon-dependent, so the message is unit-tested instead). +func TestLogsNoTargetsNotice(t *testing.T) { + root := NewRootCmd(Options{}) + var out bytes.Buffer + root.SetOut(&out) + + if err := logsNoTargets(root, &GlobalOpts{}, nil); err != nil { + t.Fatal(err) + } + if !strings.Contains(out.String(), "no managed services") { + t.Fatalf("expected zero-state notice, got %q", out.String()) + } + + out.Reset() + if err := logsNoTargets(root, &GlobalOpts{}, []string{"api"}); err != nil { + t.Fatal(err) + } + if !strings.Contains(out.String(), "no running services match") { + t.Fatalf("expected filtered zero-state notice, got %q", out.String()) + } + + // --json and --quiet stay silent (empty stream == zero objects). + out.Reset() + if err := logsNoTargets(root, &GlobalOpts{JSON: true}, nil); err != nil { + t.Fatal(err) + } + if out.Len() != 0 { + t.Fatalf("--json zero-state should be silent, got %q", out.String()) + } +} diff --git a/internal/cli/logs.go b/internal/cli/logs.go new file mode 100644 index 0000000..ab8866d --- /dev/null +++ b/internal/cli/logs.go @@ -0,0 +1,311 @@ +package cli + +import ( + "context" + "encoding/json" + "fmt" + "hash/fnv" + "io" + "os" + "os/signal" + "sort" + "sync" + "syscall" + + "charm.land/lipgloss/v2" + "github.com/spf13/cobra" + + "github.com/open-source-cloud/devstack/internal/docker" + "github.com/open-source-cloud/devstack/internal/generate" +) + +// newLogsCmd wires `devstack logs [service...]` (spec 16) — the non-interactive +// sibling of `dashboard`. It multiplexes container logs across the workspace's +// project + shared stacks, streaming from the read-only Engine SDK filtered on +// devstack's own label (+ an optional per-service filter). A TTY gets a +// color-keyed `service │` gutter; a pipe gets plain prefixed lines; `--json` +// emits one structured object per line. Non-follow mode exits at EOF. +func newLogsCmd(g *GlobalOpts) *cobra.Command { + var ( + follow bool + tail int + since string + timestamps bool + noColor bool + ) + cmd := &cobra.Command{ + Use: "logs [service...]", + Short: "Stream service logs across the workspace's project + shared stacks", + Long: "logs multiplexes container logs across every project + shared service in\n" + + "this workspace (or the named services), color-keyed by service. It streams\n" + + "from the read-only Engine SDK; with no daemon it reports nothing to stream.\n" + + "`--json` emits one {ts,service,project,stream,container,line} object per line.", + ValidArgsFunction: logsServiceCompletion, + RunE: func(cmd *cobra.Command, args []string) error { + client, closeFn, err := newLogsClient(cmd) + if err != nil { + return err + } + defer closeFn() + + targets, err := resolveLogTargets(cmd.Context(), client, args) + if err != nil { + return err + } + if len(targets) == 0 { + return logsNoTargets(cmd, g, args) + } + + // Follow mode runs until interrupted; wire a signal-aware context so + // Ctrl-C tears down every SDK stream cleanly (the root ctx is not + // signal-aware). Non-follow returns at EOF on its own. + ctx := cmd.Context() + if follow { + var stop context.CancelFunc + ctx, stop = signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM) + defer stop() + } + + opts := docker.LogOptions{Follow: follow, Tail: tail, Since: since, Timestamps: timestamps} + w := &logRenderer{ + out: cmd.OutOrStdout(), + json: g.JSON, + color: logsUseColor(cmd, g, noColor), + } + return streamLogs(ctx, client, targets, opts, w) + }, + } + f := cmd.Flags() + f.BoolVarP(&follow, "follow", "f", false, "follow log output (stream until interrupted)") + f.IntVar(&tail, "tail", 200, "number of trailing lines to show per service (<=0 = all)") + f.StringVar(&since, "since", "", "show logs since a duration (e.g. 10m) or timestamp") + f.BoolVar(×tamps, "timestamps", false, "prefix each line with the container timestamp") + f.BoolVar(&noColor, "no-color", false, "disable the color-keyed gutter") + return cmd +} + +// logTarget is one resolved container to stream, with the metadata each line is +// tagged with. +type logTarget struct { + ID string + Container string // container name + Project string // "" for shared services + Service string // display + filter name: bare service, or shared- +} + +// resolveLogTargets enumerates the workspace's managed containers (tool-owned +// label, All=true, one-offs excluded — the same rules that keep ref-counting +// honest) and narrows them to the requested services. A shared service is named +// by its DNS alias (shared-); a project service by its service label. +// An empty filter selects every managed container. Sorted for deterministic +// ordering. +func resolveLogTargets(ctx context.Context, client docker.Client, services []string) ([]logTarget, error) { + cs, err := client.ListManaged(ctx, map[string]string{generate.LabelManaged: "true"}) + if err != nil { + return nil, err + } + want := map[string]bool{} + for _, s := range services { + want[s] = true + } + var out []logTarget + for _, c := range cs { + t := logTarget{ID: c.ID, Container: c.Name, Project: c.Labels[generate.LabelProject]} + switch { + case c.Labels[generate.LabelShared] != "": + t.Service = generate.SharedAlias(c.Labels[generate.LabelShared]) + case c.Labels[generate.LabelService] != "": + t.Service = c.Labels[generate.LabelService] + default: + t.Service = c.Name + } + if len(want) > 0 && !logTargetWanted(t, c, want) { + continue + } + out = append(out, t) + } + sort.Slice(out, func(i, j int) bool { + if out[i].Service != out[j].Service { + return out[i].Service < out[j].Service + } + return out[i].Container < out[j].Container + }) + return out, nil +} + +// logTargetWanted reports whether a target matches the requested filter set. A +// user may name the display service (shared-postgres / api), the bare shared +// engine (postgres), or the owning project. +func logTargetWanted(t logTarget, c docker.Container, want map[string]bool) bool { + if want[t.Service] { + return true + } + if s := c.Labels[generate.LabelShared]; s != "" && want[s] { + return true + } + if t.Project != "" && want[t.Project] { + return true + } + return false +} + +// logJSONLine is the scriptable per-line contract (spec 16): one object per line. +type logJSONLine struct { + TS string `json:"ts,omitempty"` + Service string `json:"service"` + Project string `json:"project,omitempty"` + Stream string `json:"stream"` + Container string `json:"container"` + Line string `json:"line"` +} + +// logRenderer writes tagged log lines in one of the two output contracts. +type logRenderer struct { + out io.Writer + json bool + color bool + mu sync.Mutex // serialize concurrent stream writers onto one stdout + enc *json.Encoder +} + +// write renders one line from one target. It is safe for concurrent callers. +func (r *logRenderer) write(t logTarget, ll docker.LogLine) { + r.mu.Lock() + defer r.mu.Unlock() + if r.json { + if r.enc == nil { + r.enc = json.NewEncoder(r.out) + } + _ = r.enc.Encode(logJSONLine{ + TS: ll.TS, Service: t.Service, Project: t.Project, + Stream: ll.Stream, Container: t.Container, Line: ll.Text, + }) + return + } + gutter := t.Service + if r.color { + gutter = logColor(t.Service).Render(gutter) + } + ts := "" + if ll.TS != "" { + ts = ll.TS + " " + } + fmt.Fprintf(r.out, "%s │ %s%s\n", gutter, ts, ll.Text) +} + +// streamLogs opens a stream per target, fans them into the renderer, and returns +// when every stream ends (non-follow EOF) or ctx is cancelled (follow Ctrl-C). +func streamLogs(ctx context.Context, client docker.Client, targets []logTarget, opts docker.LogOptions, r *logRenderer) error { + var wg sync.WaitGroup + var firstErr error + var errMu sync.Mutex + for _, t := range targets { + ch, err := client.ContainerLogStream(ctx, t.ID, opts) + if err != nil { + // A single unreadable service (e.g. `none` logging driver) is a + // one-line notice, not a fatal — the rest still stream (spec 16). + fmt.Fprintf(r.out, "%s: %v\n", t.Service, err) + errMu.Lock() + if firstErr == nil { + firstErr = err + } + errMu.Unlock() + continue + } + wg.Add(1) + go func(t logTarget, ch <-chan docker.LogLine) { + defer wg.Done() + for ll := range ch { + r.write(t, ll) + } + }(t, ch) + } + wg.Wait() + // In follow mode a clean Ctrl-C teardown is success, not an error. + if ctx.Err() != nil { + return nil + } + _ = firstErr // non-fatal; surfaced inline above + return nil +} + +// logsUseColor decides whether to emit the color-keyed gutter: never for --json +// or --quiet or --no-color or NO_COLOR/$TERM=dumb, and only when stdout is a TTY +// (a piped stdout gets plain lines even if stderr is a terminal — spec 16). +func logsUseColor(cmd *cobra.Command, g *GlobalOpts, noColor bool) bool { + if g.JSON || g.Quiet || noColor { + return false + } + if os.Getenv("NO_COLOR") != "" || os.Getenv("TERM") == "dumb" { + return false + } + if f, ok := cmd.OutOrStdout().(*os.File); ok { + return isTerminal(f) + } + return false +} + +var logPalette = []string{"39", "208", "76", "170", "220", "51", "199", "141", "45", "214", "84", "205"} + +// logColor maps a service name to a stable color from the palette (hash → index). +func logColor(name string) lipgloss.Style { + h := fnv.New32a() + _, _ = h.Write([]byte(name)) + c := logPalette[int(h.Sum32())%len(logPalette)] + return lipgloss.NewStyle().Foreground(lipgloss.Color(c)).Bold(true) +} + +// logsNoTargets prints the right zero-state for the plain and --json contracts. +// --json stays silent (an empty stream is zero objects) to keep the line contract +// clean for consumers. +func logsNoTargets(cmd *cobra.Command, g *GlobalOpts, services []string) error { + if g.JSON || g.Quiet { + return nil + } + w := cmd.OutOrStdout() + if len(services) > 0 { + fmt.Fprintf(w, "no running services match %v (try `devstack status`)\n", services) + return nil + } + fmt.Fprintln(w, "no managed services are running in this workspace (run `devstack up`)") + return nil +} + +// logsServiceCompletion completes service names from live managed containers. +func logsServiceCompletion(cmd *cobra.Command, _ []string, _ string) ([]string, cobra.ShellCompDirective) { + client, closeFn, err := newLogsClient(cmd) + if err != nil { + return nil, cobra.ShellCompDirectiveNoFileComp + } + defer closeFn() + targets, err := resolveLogTargets(cmd.Context(), client, nil) + if err != nil { + return nil, cobra.ShellCompDirectiveNoFileComp + } + set := map[string]bool{} + for _, t := range targets { + set[t.Service] = true + } + out := make([]string, 0, len(set)) + for s := range set { + out = append(out, s) + } + sort.Strings(out) + return out, cobra.ShellCompDirectiveNoFileComp +} + +// newLogsClient builds the read-only Engine client for log streaming. An +// unreachable daemon (construction OR ping failure) degrades to an empty mock (no +// targets) rather than an error, so `logs` in a dead-daemon workspace prints a +// clean zero-state instead of a stack trace. +func newLogsClient(cmd *cobra.Command) (docker.Client, func(), error) { + c, err := docker.NewClient(cmd.Context()) + if err != nil { + return &docker.MockClient{}, func() {}, nil + } + if err := c.Ping(cmd.Context()); err != nil { + _ = c.Close() + return &docker.MockClient{}, func() {}, nil + } + return c, func() { _ = c.Close() }, nil +} diff --git a/internal/cli/logs_test.go b/internal/cli/logs_test.go new file mode 100644 index 0000000..1ee587d --- /dev/null +++ b/internal/cli/logs_test.go @@ -0,0 +1,168 @@ +package cli + +import ( + "bytes" + "context" + "encoding/json" + "strings" + "testing" + + "github.com/open-source-cloud/devstack/internal/docker" + "github.com/open-source-cloud/devstack/internal/generate" +) + +// managedContainers is a small fixture: two project services + two shared engines. +func managedContainers() []docker.Container { + return []docker.Container{ + {ID: "api1", Name: "devstack-shop-api-1", State: "running", Labels: map[string]string{ + generate.LabelManaged: "true", generate.LabelProject: "shop", generate.LabelService: "api"}}, + {ID: "web1", Name: "devstack-shop-web-1", State: "running", Labels: map[string]string{ + generate.LabelManaged: "true", generate.LabelProject: "shop", generate.LabelService: "web"}}, + {ID: "pg1", Name: "devstack-shared-postgres-1", State: "running", Labels: map[string]string{ + generate.LabelManaged: "true", generate.LabelShared: "postgres"}}, + {ID: "rd1", Name: "devstack-shared-redis-1", State: "running", Labels: map[string]string{ + generate.LabelManaged: "true", generate.LabelShared: "redis"}}, + } +} + +func TestResolveLogTargets(t *testing.T) { + client := &docker.MockClient{Containers: managedContainers()} + tests := []struct { + name string + filter []string + wantSvcs []string + }{ + {"all", nil, []string{"api", "shared-postgres", "shared-redis", "web"}}, + {"by-service", []string{"api"}, []string{"api"}}, + {"by-shared-alias", []string{"shared-postgres"}, []string{"shared-postgres"}}, + {"by-bare-engine", []string{"redis"}, []string{"shared-redis"}}, + {"by-project", []string{"shop"}, []string{"api", "web"}}, + {"no-match", []string{"nope"}, nil}, + {"mixed", []string{"api", "shared-redis"}, []string{"api", "shared-redis"}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + targets, err := resolveLogTargets(context.Background(), client, tt.filter) + if err != nil { + t.Fatal(err) + } + var got []string + for _, tg := range targets { + got = append(got, tg.Service) + } + if strings.Join(got, ",") != strings.Join(tt.wantSvcs, ",") { + t.Fatalf("targets = %v, want %v", got, tt.wantSvcs) + } + }) + } +} + +// TestResolveLogTargetsSorted asserts deterministic ordering (sorted by service). +func TestResolveLogTargetsSorted(t *testing.T) { + client := &docker.MockClient{Containers: managedContainers()} + targets, err := resolveLogTargets(context.Background(), client, nil) + if err != nil { + t.Fatal(err) + } + for i := 1; i < len(targets); i++ { + if targets[i-1].Service > targets[i].Service { + t.Fatalf("targets not sorted: %q before %q", targets[i-1].Service, targets[i].Service) + } + } +} + +func TestLogRendererJSONShape(t *testing.T) { + var buf bytes.Buffer + r := &logRenderer{out: &buf, json: true} + r.write( + logTarget{Service: "api", Project: "shop", Container: "devstack-shop-api-1"}, + docker.LogLine{Stream: "stdout", TS: "2026-06-14T12:03:01.412Z", Text: "GET /healthz 200"}, + ) + var got logJSONLine + if err := json.Unmarshal(buf.Bytes(), &got); err != nil { + t.Fatalf("output is not one JSON object per line: %v\n%s", err, buf.String()) + } + want := logJSONLine{ + TS: "2026-06-14T12:03:01.412Z", Service: "api", Project: "shop", + Stream: "stdout", Container: "devstack-shop-api-1", Line: "GET /healthz 200", + } + if got != want { + t.Fatalf("json line = %+v, want %+v", got, want) + } + // The raw output must carry every contract key. + for _, k := range []string{`"ts"`, `"service"`, `"project"`, `"stream"`, `"container"`, `"line"`} { + if !strings.Contains(buf.String(), k) { + t.Errorf("json output missing key %s: %s", k, buf.String()) + } + } +} + +// TestLogRendererPlainGutter checks the non-JSON, no-color gutter format. +func TestLogRendererPlainGutter(t *testing.T) { + var buf bytes.Buffer + r := &logRenderer{out: &buf, json: false, color: false} + r.write(logTarget{Service: "api"}, docker.LogLine{Stream: "stdout", Text: "hello"}) + if got := buf.String(); got != "api │ hello\n" { + t.Fatalf("plain gutter = %q", got) + } +} + +// TestStreamLogsJSON drives the full multiplex path over the mock: two services, +// each with seeded lines, into the JSON contract. Non-follow exits at EOF. +func TestStreamLogsJSON(t *testing.T) { + client := &docker.MockClient{ + Containers: managedContainers(), + Streams: map[string][]docker.LogLine{ + "api1": {{Stream: "stdout", Text: "api-a"}, {Stream: "stderr", Text: "api-b"}}, + "pg1": {{Stream: "stdout", Text: "pg-a"}}, + }, + } + targets, err := resolveLogTargets(context.Background(), client, []string{"api", "shared-postgres"}) + if err != nil { + t.Fatal(err) + } + var buf bytes.Buffer + r := &logRenderer{out: &buf, json: true} + if err := streamLogs(context.Background(), client, targets, docker.LogOptions{}, r); err != nil { + t.Fatal(err) + } + lines := strings.Split(strings.TrimSpace(buf.String()), "\n") + if len(lines) != 3 { + t.Fatalf("want 3 json lines, got %d:\n%s", len(lines), buf.String()) + } + svcCount := map[string]int{} + for _, l := range lines { + var obj logJSONLine + if err := json.Unmarshal([]byte(l), &obj); err != nil { + t.Fatalf("line not valid json: %q", l) + } + svcCount[obj.Service]++ + } + if svcCount["api"] != 2 || svcCount["shared-postgres"] != 1 { + t.Fatalf("unexpected per-service counts: %v", svcCount) + } +} + +// TestStreamLogsUnreadableService verifies a per-service stream error is a +// one-line notice, not fatal, and the rest still stream. +func TestStreamLogsUnreadableService(t *testing.T) { + client := &docker.MockClient{ + Containers: managedContainers(), + StreamErr: errNoneDriver, + } + targets, _ := resolveLogTargets(context.Background(), client, []string{"api"}) + var buf bytes.Buffer + r := &logRenderer{out: &buf} + if err := streamLogs(context.Background(), client, targets, docker.LogOptions{}, r); err != nil { + t.Fatalf("streamLogs should not fail fatally: %v", err) + } + if !strings.Contains(buf.String(), "api:") { + t.Fatalf("expected a per-service notice, got %q", buf.String()) + } +} + +var errNoneDriver = errStub("configured logging driver does not support reading") + +type errStub string + +func (e errStub) Error() string { return string(e) } diff --git a/internal/cli/root.go b/internal/cli/root.go index 0c404e0..76d364a 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -81,6 +81,8 @@ func NewRootCmd(opts Options) *cobra.Command { newDownCmd(g), newShellCmd(g), newStatusCmd(g), + newLogsCmd(g), + newDashboardCmd(g), newDnsCmd(g), newTrustCmd(g), newTunnelCmd(g), diff --git a/internal/cli/stubs.go b/internal/cli/stubs.go index b52f38e..1e91489 100644 --- a/internal/cli/stubs.go +++ b/internal/cli/stubs.go @@ -28,13 +28,11 @@ func rootName(c *cobra.Command) string { return c.Root().Name() } // addStubCommands reserves the post-1.0 command surface from spec 07 as // milestone-tagged placeholders so `--help`/completions stay consistent (exit 0, -// clear notice). `shell` has GRADUATED to a real command (spec 26); `logs` stays a -// stub, re-tagged to v2 (its full read-only-SDK design is owned by spec 16). `db` -// has GRADUATED to a real command group (spec 29, see db.go). +// clear notice). `shell` has GRADUATED to a real command (spec 26); `logs` and +// `dashboard` have GRADUATED to real commands (spec 16, see logs.go/dashboard.go). +// `db` has GRADUATED to a real command group (spec 29, see db.go). func addStubCommands(root *cobra.Command, _ *GlobalOpts) { root.AddCommand( - stub("logs", "Stream service logs", "v2 (spec 16)"), - stub("dashboard", "Live TUI cockpit", "v2 (spec 16)"), stub("ide", "Generate devcontainer/.code-workspace/launch configs", "v2 (spec 17)"), stub("telemetry", "Opt-in usage telemetry (default OFF)", "a later release (spec 20)"), ) diff --git a/internal/docker/docker.go b/internal/docker/docker.go index 35057b4..0a827ed 100644 --- a/internal/docker/docker.go +++ b/internal/docker/docker.go @@ -55,10 +55,35 @@ type Client interface { // stdout+stderr (tail<=0 means all) — the fail-fast diagnostic inlined when a // dependency goes unhealthy or exits during `up` (spec 10). Read-only. ContainerLogs(ctx context.Context, id string, tail int) (string, error) + // ContainerLogStream opens a demuxed, optionally-following log stream for one + // container (spec 16). It returns a channel of per-line records tagged with + // their stream (stdout/stderr) and, when opts.Timestamps is set, the container + // timestamp. The channel is closed when the stream ends: EOF in non-follow + // mode, ctx cancellation, or the container stopping while following. Non-TTY + // containers are demuxed through stdcopy; TTY containers stream raw. Read-only. + ContainerLogStream(ctx context.Context, id string, opts LogOptions) (<-chan LogLine, error) // Close releases the underlying connection. Close() error } +// LogOptions configures a ContainerLogStream. Tail<=0 means "all"; Since accepts +// either a duration (e.g. "10m") or a timestamp, passed straight to the Engine. +type LogOptions struct { + Follow bool + Tail int + Since string + Timestamps bool +} + +// LogLine is one demuxed log line from a container. TS is the RFC3339Nano +// container timestamp when the stream was opened with Timestamps (empty +// otherwise); Text never carries a trailing newline. +type LogLine struct { + Stream string // "stdout" | "stderr" + TS string // container timestamp, "" unless Timestamps requested + Text string +} + // HealthStatus mirrors Docker's .State.Health.Status. The empty string means the // container declares no healthcheck at all (inspect reports a nil Health block); // `none` can also appear via the list API. Both mean "no health signal" — use @@ -84,6 +109,7 @@ type ContainerDetails struct { Running bool // .State.Running ExitCode int // .State.ExitCode (meaningful once exited) Health HealthStatus // "" when the container has no Health block + TTY bool // .Config.Tty — a TTY container's logs are NOT stdcopy-framed } // HasHealthcheck reports whether the container declares a healthcheck (so its diff --git a/internal/docker/inspect.go b/internal/docker/inspect.go index d9c1813..3cb835f 100644 --- a/internal/docker/inspect.go +++ b/internal/docker/inspect.go @@ -117,6 +117,7 @@ func (m *mobyClient) ContainerInspect(ctx context.Context, id string) (Container } if c.Config != nil { d.Labels = c.Config.Labels + d.TTY = c.Config.Tty } if c.State != nil { d.State = string(c.State.Status) diff --git a/internal/docker/logstream.go b/internal/docker/logstream.go new file mode 100644 index 0000000..d0f04f3 --- /dev/null +++ b/internal/docker/logstream.go @@ -0,0 +1,141 @@ +package docker + +import ( + "bytes" + "context" + "fmt" + "io" + "strconv" + "strings" + + "github.com/moby/moby/api/pkg/stdcopy" + moby "github.com/moby/moby/client" +) + +// logChanBuffer bounds the per-container fan-in channel so a chatty container +// applies natural backpressure (a slow consumer blocks the demux goroutine) +// rather than growing an unbounded queue (spec 16 §backpressure). +const logChanBuffer = 256 + +// ContainerLogStream opens a demuxed, optionally-following log stream for one +// container. It first inspects the container to learn whether it has a TTY (a +// TTY stream is NOT stdcopy-framed and must be read raw — reading a non-TTY +// stream raw corrupts it with the 8-byte frame header, spec 16 §demux). +func (m *mobyClient) ContainerLogStream(ctx context.Context, id string, opts LogOptions) (<-chan LogLine, error) { + // Learn the TTY-ness up front; a missing/erroring inspect defaults to the + // safe non-TTY (demuxed) path. + tty := false + if det, err := m.ContainerInspect(ctx, id); err == nil { + tty = det.TTY + } + + lopts := moby.ContainerLogsOptions{ + ShowStdout: true, + ShowStderr: true, + Follow: opts.Follow, + Since: opts.Since, + Timestamps: opts.Timestamps, + } + if opts.Tail > 0 { + lopts.Tail = strconv.Itoa(opts.Tail) + } + rc, err := m.cli.ContainerLogs(ctx, id, lopts) + if err != nil { + // The Engine returns a distinct error for services on a logging driver + // that can't be read (syslog/none); surface a one-liner (spec 16 gotcha). + if strings.Contains(strings.ToLower(err.Error()), "logging driver") { + return nil, fmt.Errorf("logs for %q unavailable: its logging driver does not support reading (use json-file/local): %w", id, err) + } + return nil, fmt.Errorf("logs for container %q: %w", id, err) + } + + out := make(chan LogLine, logChanBuffer) + go func() { + defer close(out) + defer func() { _ = rc.Close() }() + if tty { + // Raw single stream: everything is stdout as far as the TTY is concerned. + e := &lineEmitter{ctx: ctx, out: out, stream: "stdout", ts: opts.Timestamps} + _, _ = io.Copy(e, rc) + e.flush() + return + } + outE := &lineEmitter{ctx: ctx, out: out, stream: "stdout", ts: opts.Timestamps} + errE := &lineEmitter{ctx: ctx, out: out, stream: "stderr", ts: opts.Timestamps} + _, _ = stdcopy.StdCopy(outE, errE, rc) + outE.flush() + errE.flush() + }() + return out, nil +} + +// lineEmitter is an io.Writer that splits its input into lines and forwards each +// as a LogLine on out, parsing the Engine's leading RFC3339 timestamp when ts is +// set. Partial lines are buffered across Writes; flush emits any final unterminated +// line at EOF. +type lineEmitter struct { + ctx context.Context + out chan<- LogLine + stream string + ts bool + buf []byte +} + +func (e *lineEmitter) Write(p []byte) (int, error) { + e.buf = append(e.buf, p...) + for { + i := bytes.IndexByte(e.buf, '\n') + if i < 0 { + break + } + line := string(e.buf[:i]) + e.buf = e.buf[i+1:] + if !e.send(line) { + return len(p), context.Canceled + } + } + return len(p), nil +} + +func (e *lineEmitter) flush() { + if len(e.buf) == 0 { + return + } + line := string(e.buf) + e.buf = nil + e.send(line) +} + +// send parses the optional timestamp prefix and delivers the line, honoring ctx +// cancellation so a torn-down stream never blocks forever. Returns false if ctx +// is done. +func (e *lineEmitter) send(line string) bool { + line = strings.TrimRight(line, "\r") + ll := LogLine{Stream: e.stream, Text: line} + if e.ts { + ll.TS, ll.Text = splitTimestamp(line) + } + select { + case e.out <- ll: + return true + case <-e.ctx.Done(): + return false + } +} + +// splitTimestamp separates the Engine's leading RFC3339Nano timestamp (added +// when Timestamps=true) from the log text. A line without a leading timestamp is +// returned unchanged with an empty TS. +func splitTimestamp(line string) (ts, text string) { + sp := strings.IndexByte(line, ' ') + if sp <= 0 { + return "", line + } + head := line[:sp] + // A valid Engine timestamp contains a 'T' and is date-shaped; a cheap guard + // avoids mis-splitting an untimestamped line whose first token has no 'T'. + if !strings.ContainsRune(head, 'T') { + return "", line + } + return head, line[sp+1:] +} diff --git a/internal/docker/logstream_test.go b/internal/docker/logstream_test.go new file mode 100644 index 0000000..e2989a0 --- /dev/null +++ b/internal/docker/logstream_test.go @@ -0,0 +1,89 @@ +package docker + +import ( + "context" + "testing" +) + +func TestSplitTimestamp(t *testing.T) { + tests := []struct { + name string + line string + wantTS string + wantText string + }{ + {"rfc3339", "2026-06-14T12:03:01.412Z GET /healthz 200", "2026-06-14T12:03:01.412Z", "GET /healthz 200"}, + {"no-timestamp", "plain log line", "", "plain log line"}, + {"first-token-no-T", "hello world here", "", "hello world here"}, + {"empty", "", "", ""}, + {"only-ts", "2026-06-14T12:03:01Z", "", "2026-06-14T12:03:01Z"}, // no space → whole line is text + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ts, text := splitTimestamp(tt.line) + if ts != tt.wantTS || text != tt.wantText { + t.Fatalf("splitTimestamp(%q) = (%q, %q), want (%q, %q)", tt.line, ts, text, tt.wantTS, tt.wantText) + } + }) + } +} + +// TestLineEmitter verifies the writer splits input into lines, tags the stream, +// buffers partial lines across writes, and flushes the final unterminated line. +func TestLineEmitter(t *testing.T) { + out := make(chan LogLine, 8) + e := &lineEmitter{ctx: context.Background(), out: out, stream: "stderr"} + + // Two writes; the second completes a line split across the boundary. + _, _ = e.Write([]byte("first\nseco")) + _, _ = e.Write([]byte("nd\nthird-no-newline")) + e.flush() + + var got []LogLine + close(out) + for ll := range out { + got = append(got, ll) + } + want := []string{"first", "second", "third-no-newline"} + if len(got) != len(want) { + t.Fatalf("got %d lines, want %d: %+v", len(got), len(want), got) + } + for i := range want { + if got[i].Text != want[i] { + t.Errorf("line %d = %q, want %q", i, got[i].Text, want[i]) + } + if got[i].Stream != "stderr" { + t.Errorf("line %d stream = %q, want stderr", i, got[i].Stream) + } + } +} + +// TestLineEmitterCtxCancel ensures a cancelled context stops the emitter instead +// of blocking on a full channel (follow-mode teardown). +func TestLineEmitterCtxCancel(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + out := make(chan LogLine) // unbuffered: send would block without ctx guard + e := &lineEmitter{ctx: ctx, out: out, stream: "stdout"} + if ok := e.send("blocked"); ok { + t.Fatal("send returned true on a cancelled context") + } +} + +// TestMockContainerLogStream verifies the mock emits seeded lines then closes. +func TestMockContainerLogStream(t *testing.T) { + m := &MockClient{Streams: map[string][]LogLine{ + "c1": {{Stream: "stdout", Text: "one"}, {Stream: "stderr", Text: "two"}}, + }} + ch, err := m.ContainerLogStream(context.Background(), "c1", LogOptions{}) + if err != nil { + t.Fatal(err) + } + var got []LogLine + for ll := range ch { + got = append(got, ll) + } + if len(got) != 2 || got[0].Text != "one" || got[1].Text != "two" { + t.Fatalf("unexpected stream: %+v", got) + } +} diff --git a/internal/docker/mock.go b/internal/docker/mock.go index 9c6421d..fbc6892 100644 --- a/internal/docker/mock.go +++ b/internal/docker/mock.go @@ -23,11 +23,17 @@ type MockClient struct { Details map[string]ContainerDetails // LogLines maps a container ID or name to its canned log text. LogLines map[string]string + // Streams maps a container ID or name to the demuxed lines ContainerLogStream + // emits (in order) before closing the channel — the seam for logs/dashboard + // tests without a daemon. + Streams map[string][]LogLine // NetworkErr / ListErr / InspectErr / LogsErr force the op to fail. NetworkErr error ListErr error InspectErr error LogsErr error + // StreamErr forces ContainerLogStream to fail (e.g. the unreadable-driver case). + StreamErr error } var _ Client = (*MockClient)(nil) @@ -119,6 +125,27 @@ func (m *MockClient) ContainerLogs(_ context.Context, id string, tail int) (stri return out, nil } +// ContainerLogStream emits the seeded lines for id then closes the channel, +// honoring ctx cancellation so follow-mode teardown is testable without a daemon. +func (m *MockClient) ContainerLogStream(ctx context.Context, id string, _ LogOptions) (<-chan LogLine, error) { + if m.StreamErr != nil { + return nil, m.StreamErr + } + lines := m.Streams[id] + out := make(chan LogLine, len(lines)+1) + go func() { + defer close(out) + for _, ll := range lines { + select { + case out <- ll: + case <-ctx.Done(): + return + } + } + }() + return out, nil +} + // lastLines returns the final n lines of s, preserving a trailing newline. func lastLines(s string, n int) string { if s == "" || n <= 0 {