diff --git a/internal/tui/messages.go b/internal/tui/messages.go index 1999eec..48fa0b6 100644 --- a/internal/tui/messages.go +++ b/internal/tui/messages.go @@ -4,6 +4,7 @@ import ( "io" "tangled.org/samatkins.co/dockscope/internal/domain" + "tangled.org/samatkins.co/dockscope/internal/parser" ) // StatsMsg carries one stats sample (or a stream-closed signal) for a @@ -26,9 +27,21 @@ type LogMsg struct { Err error } +// ParsedLogMsg carries a structured log entry from a background container +// log stream into the ring buffer. +type ParsedLogMsg struct { + Entry parser.LogEntry +} + type logStreamOpenedMsg struct { containerID string reader io.ReadCloser ch <-chan LogMsg err error } + +type logStreamStartedMsg struct { + containerID string + ch <-chan ParsedLogMsg + err error +} diff --git a/internal/tui/model.go b/internal/tui/model.go index d9af1a0..c7cf553 100644 --- a/internal/tui/model.go +++ b/internal/tui/model.go @@ -12,6 +12,7 @@ import ( tea "github.com/charmbracelet/bubbletea" "tangled.org/samatkins.co/dockscope/internal/domain" + "tangled.org/samatkins.co/dockscope/internal/parser" ) // StatsFetcher abstracts the mechanism for streaming container statistics. @@ -49,6 +50,24 @@ const ( paneCount ) +type logStreamStop struct { + ch chan struct{} +} + +func (s *logStreamStop) stop() { + select { + case <-s.ch: + default: + close(s.ch) + } +} + +type streamParams struct { + containerID string + containerName string + stop <-chan struct{} +} + type tickMsg struct{} func tick() tea.Cmd { @@ -79,6 +98,11 @@ type Model struct { logStreamErr error width int height int + ringBuf *RingBuffer[parser.LogEntry] + traceIndex map[string][]int + parser parser.Parser + logStreamStops map[string]*logStreamStop + logStreamChans map[string]<-chan ParsedLogMsg } func (m *Model) visibleContainers() []domain.Container { @@ -114,6 +138,8 @@ func NewModel(ctx context.Context, containers []domain.Container, client Client) lister = client } + const defaultRingCapacity = 50000 + return Model{ containers: containers, fetcher: fetcher, @@ -123,6 +149,11 @@ func NewModel(ctx context.Context, containers []domain.Container, client Client) selected: make(map[string]bool), logClient: logClient, containerLister: lister, + ringBuf: NewRingBuffer[parser.LogEntry](defaultRingCapacity), + traceIndex: make(map[string][]int), + parser: parser.Chain{parser.JSONParser{}, parser.DefaultRegexParser()}, + logStreamStops: make(map[string]*logStreamStop), + logStreamChans: make(map[string]<-chan ParsedLogMsg), } } @@ -142,7 +173,15 @@ func (m *Model) toggleSelected(containerID string) { func (m *Model) Init() tea.Cmd { cmds := []tea.Cmd{tea.EnterAltScreen, tick()} + for _, c := range m.containers { + if c.State == domain.StateRunning { + m.selected[c.ID] = true + } + } + if m.fetcher == nil { + cmds = append(cmds, m.startLogStreamForSelected()) + return tea.Batch(cmds...) } @@ -157,6 +196,8 @@ func (m *Model) Init() tea.Cmd { cmds = append(cmds, listenOnChan(ch, c.ID)) } + cmds = append(cmds, m.startLogStreamForSelected()) + return tea.Batch(cmds...) } @@ -318,6 +359,21 @@ func (m *Model) handleTick() (tea.Model, tea.Cmd) { m.containers = containers m.clampCursor() + for _, c := range containers { + if c.State != domain.StateRunning { + continue + } + + if _, selected := m.selected[c.ID]; !selected { + m.selected[c.ID] = true + slog.Debug("auto-selected new container", fieldContainer, c.ID, "name", c.Name) + + if _, exists := m.logStreamStops[c.ID]; !exists { + cmds = append(cmds, m.startLogStream(c.ID, c.Name)) + } + } + } + slog.Debug("refreshed containers", "count", len(containers)) return m, tea.Batch(cmds...) @@ -336,6 +392,14 @@ func (m *Model) removeStaleContainers(fresh []domain.Container) { delete(m.selected, id) } } + + for id, stop := range m.logStreamStops { + if !active[id] { + stop.stop() + delete(m.logStreamStops, id) + delete(m.logStreamChans, id) + } + } } func (m *Model) startStatsForNewContainers(fresh []domain.Container, cmds []tea.Cmd) []tea.Cmd { @@ -372,3 +436,193 @@ func (m *Model) clampCursor() { m.cursor = n - 1 } } + +func (m *Model) startLogStreamForSelected() tea.Cmd { + var cmds []tea.Cmd + + for _, c := range m.containers { + if m.isSelected(c.ID) && c.State == domain.StateRunning { + if _, exists := m.logStreamStops[c.ID]; !exists { + cmds = append(cmds, m.startLogStream(c.ID, c.Name)) + } + } + } + + if len(cmds) == 0 { + return nil + } + + return tea.Batch(cmds...) +} + +func (m *Model) startLogStream(containerID, containerName string) tea.Cmd { + if m.logClient == nil { + slog.Error("log client not available", fieldContainer, containerID) + + return nil + } + + if _, exists := m.logStreamStops[containerID]; exists { + return nil + } + + stop := &logStreamStop{ch: make(chan struct{})} + m.logStreamStops[containerID] = stop + + client := m.logClient + + return func() tea.Msg { + rc, err := client.StreamLogs(m.ctx, containerID, defaultLogTail) + if err != nil { + stop.stop() + + return logStreamStartedMsg{containerID: containerID, err: err} + } + + if rc == nil { + stop.stop() + + return logStreamStartedMsg{containerID: containerID, err: errNilLogReader} + } + + ch := make(chan ParsedLogMsg, logChanBuf) + + cfg := streamParams{ + containerID: containerID, + containerName: containerName, + stop: stop.ch, + } + + go m.streamParsedLogs(m.ctx, rc, ch, cfg) + + return logStreamStartedMsg{containerID: containerID, ch: ch} + } +} + +var errNilLogReader = errors.New("log client returned nil reader") + +func (m *Model) streamParsedLogs(ctx context.Context, rc io.ReadCloser, ch chan<- ParsedLogMsg, cfg streamParams) { + defer close(ch) + + if rc == nil { + return + } + + defer func() { + if err := rc.Close(); err != nil { + slog.Warn("failed to close log reader", fieldContainer, cfg.containerID, fieldError, err) + } + }() + + br := bufio.NewReader(rc) + + p := m.parser + + for { + line, err := br.ReadString('\n') + + if line != "" { + raw := strings.TrimRight(line, "\r\n") + entry, _ := p.Parse(raw) + entry.ContainerID = cfg.containerID + entry.Container = cfg.containerName + + select { + case ch <- ParsedLogMsg{Entry: entry}: + case <-ctx.Done(): + return + case <-cfg.stop: + return + } + } + + if err != nil { + return + } + } +} + +func (m *Model) handleLogStreamStarted(msg logStreamStartedMsg) (tea.Model, tea.Cmd) { + if msg.err != nil { + slog.Error("failed to start log stream", fieldContainer, msg.containerID, fieldError, msg.err) + delete(m.logStreamStops, msg.containerID) + + return m, nil + } + + m.logStreamChans[msg.containerID] = msg.ch + + slog.Debug("started log stream for container", fieldContainer, msg.containerID) + + return m, listenOnParsedLogCh(msg.ch) +} + +func listenOnParsedLogCh(ch <-chan ParsedLogMsg) tea.Cmd { + return func() tea.Msg { + msg, ok := <-ch + if !ok { + return nil + } + + return msg + } +} + +func (m *Model) ingestLogLine(entry parser.LogEntry) { + res := m.ringBuf.Push(entry) + + if res.DidEvict && res.Evicted.TraceID != "" { + m.removeSlotFromTraceIndex(res.Evicted.TraceID, res.EvictedSlot) + + slog.Debug("ring buffer wrapped, evicted entry from index", + "trace_id", res.Evicted.TraceID, + "slot", res.EvictedSlot, + ) + } + + if entry.TraceID != "" { + m.traceIndex[entry.TraceID] = append(m.traceIndex[entry.TraceID], res.Slot) + } +} + +func (m *Model) removeSlotFromTraceIndex(traceID string, slot int) { + indices := m.traceIndex[traceID] + + for i, s := range indices { + if s == slot { + m.traceIndex[traceID] = append(indices[:i], indices[i+1:]...) + + break + } + } + + if len(m.traceIndex[traceID]) == 0 { + delete(m.traceIndex, traceID) + } +} + +func (m *Model) syncLogStreams() tea.Cmd { + var cmds []tea.Cmd + + for _, c := range m.containers { + if m.isSelected(c.ID) { + if _, exists := m.logStreamStops[c.ID]; !exists { + cmds = append(cmds, m.startLogStream(c.ID, c.Name)) + } + } else { + if stop, exists := m.logStreamStops[c.ID]; exists { + stop.stop() + delete(m.logStreamStops, c.ID) + delete(m.logStreamChans, c.ID) + + slog.Debug("container deselected for log streaming", fieldContainer, c.ID, "name", c.Name) + } + } + } + + if len(cmds) == 0 { + return nil + } + + return tea.Batch(cmds...) +} diff --git a/internal/tui/ringbuffer.go b/internal/tui/ringbuffer.go index 9d01ee3..a26a736 100644 --- a/internal/tui/ringbuffer.go +++ b/internal/tui/ringbuffer.go @@ -1,14 +1,20 @@ package tui type RingBuffer[T any] struct { - buf []T - start int - count int - cap int - nextSlot int - evicted T - evictedSlot int - hasEvicted bool + buf []T + start int + count int + cap int + nextSlot int +} + +// PushResult reports the outcome of a Push: the slot of the inserted value, +// and (if the buffer was full) the slot and value that were evicted. +type PushResult[T any] struct { + Slot int + EvictedSlot int + Evicted T + DidEvict bool } func NewRingBuffer[T any](capacity int) *RingBuffer[T] { @@ -22,25 +28,29 @@ func NewRingBuffer[T any](capacity int) *RingBuffer[T] { } } -func (rb *RingBuffer[T]) Push(val T) int { +func (rb *RingBuffer[T]) Push(val T) PushResult[T] { slot := rb.nextSlot rb.nextSlot++ if rb.count == rb.cap { - rb.evicted = rb.buf[rb.start] - rb.evictedSlot = rb.nextSlot - rb.count - 1 - rb.hasEvicted = true + evicted := rb.buf[rb.start] + evictedSlot := rb.nextSlot - rb.count - 1 rb.buf[rb.start] = val rb.start = (rb.start + 1) % rb.cap - return slot + return PushResult[T]{ + Slot: slot, + EvictedSlot: evictedSlot, + Evicted: evicted, + DidEvict: true, + } } phys := (rb.start + rb.count) % rb.cap rb.buf[phys] = val rb.count++ - return slot + return PushResult[T]{Slot: slot} } func (rb *RingBuffer[T]) At(slot int) (T, bool) { @@ -78,31 +88,6 @@ func (rb *RingBuffer[T]) ForEachTail(n int, fn func(T)) { } } -// LastEvicted returns the value of the most recently evicted entry, or -// (zero, false) if no eviction has ever occurred. The latch is never cleared -// and only ever reflects the most recent eviction: if multiple Pushes have -// evicted entries since the last call, intermediate evictions are lost. -func (rb *RingBuffer[T]) LastEvicted() (T, bool) { - if !rb.hasEvicted { - var zero T - return zero, false - } - - return rb.evicted, true -} - -// EvictedSlot returns the slot id of the most recently evicted entry, or -// (0, false) if no eviction has ever occurred. The latch is never cleared -// and only ever reflects the most recent eviction: if multiple Pushes have -// evicted entries since the last call, intermediate evictions are lost. -func (rb *RingBuffer[T]) EvictedSlot() (int, bool) { - if !rb.hasEvicted { - return 0, false - } - - return rb.evictedSlot, true -} - func (rb *RingBuffer[T]) Len() int { return rb.count } diff --git a/internal/tui/ringbuffer_test.go b/internal/tui/ringbuffer_test.go index c74e03d..feb9364 100644 --- a/internal/tui/ringbuffer_test.go +++ b/internal/tui/ringbuffer_test.go @@ -7,76 +7,62 @@ import ( func TestRingBufferCap3Push4(t *testing.T) { rb := NewRingBuffer[int](3) - s0 := rb.Push(10) - s1 := rb.Push(20) + r0 := rb.Push(10) + r1 := rb.Push(20) _ = rb.Push(30) - s3 := rb.Push(40) + r3 := rb.Push(40) if rb.Len() != 3 { t.Errorf("Len() = %d, want 3", rb.Len()) } - evicted, ok := rb.LastEvicted() - if !ok { - t.Fatal("expected LastEvicted to return true after eviction") + if !r3.DidEvict { + t.Fatal("expected fourth Push to evict") } - if evicted != 10 { - t.Errorf("LastEvicted() = %d, want 10", evicted) + if r3.Evicted != 10 { + t.Errorf("Evicted = %d, want 10", r3.Evicted) } - eslot, ok := rb.EvictedSlot() - if !ok { - t.Fatal("expected EvictedSlot to return true after eviction") - } - if eslot != s0 { - t.Errorf("EvictedSlot() = %d, want %d", eslot, s0) + if r3.EvictedSlot != r0.Slot { + t.Errorf("EvictedSlot = %d, want %d", r3.EvictedSlot, r0.Slot) } - v, ok := rb.At(s0) + v, ok := rb.At(r0.Slot) if ok { - t.Errorf("At(%d) should return false (evicted), got %d", s0, v) + t.Errorf("At(%d) should return false (evicted), got %d", r0.Slot, v) } - v, ok = rb.At(s1) + v, ok = rb.At(r1.Slot) if !ok || v != 20 { - t.Errorf("At(%d) = (%d, %v), want (20, true)", s1, v, ok) + t.Errorf("At(%d) = (%d, %v), want (20, true)", r1.Slot, v, ok) } - v, ok = rb.At(s3) + v, ok = rb.At(r3.Slot) if !ok || v != 40 { - t.Errorf("At(%d) = (%d, %v), want (40, true)", s3, v, ok) + t.Errorf("At(%d) = (%d, %v), want (40, true)", r3.Slot, v, ok) } } func TestRingBufferNoEvictionYet(t *testing.T) { rb := NewRingBuffer[int](5) - rb.Push(10) - - _, ok := rb.LastEvicted() - if ok { - t.Error("LastEvicted() should return false when no eviction has occurred") - } + r := rb.Push(10) - _, ok = rb.EvictedSlot() - if ok { - t.Error("EvictedSlot() should return false when no eviction has occurred") + if r.DidEvict { + t.Error("Push into non-full buffer should not evict") } } func TestRingBufferWrappedOrder(t *testing.T) { rb := NewRingBuffer[int](3) - s0 := rb.Push(10) - s1 := rb.Push(20) - s2 := rb.Push(30) + rb.Push(10) + rb.Push(20) + r2 := rb.Push(30) _ = rb.Push(40) _ = rb.Push(50) - _ = s0 - _ = s1 - - v, ok := rb.At(s2) + v, ok := rb.At(r2.Slot) if !ok || v != 30 { - t.Errorf("At(%d) = (%d, %v), want (30, true)", s2, v, ok) + t.Errorf("At(%d) = (%d, %v), want (30, true)", r2.Slot, v, ok) } var got []int @@ -162,24 +148,22 @@ func TestRingBufferSingleItemEviction(t *testing.T) { rb := NewRingBuffer[int](1) rb.Push(10) - rb.Push(20) + r2 := rb.Push(20) - evicted, ok := rb.LastEvicted() - if !ok { + if !r2.DidEvict { t.Fatal("expected eviction after second push on cap=1") } - if evicted != 10 { - t.Errorf("LastEvicted() = %d, want 10", evicted) + if r2.Evicted != 10 { + t.Errorf("Evicted = %d, want 10", r2.Evicted) } - rb.Push(30) + r3 := rb.Push(30) - evicted, ok = rb.LastEvicted() - if !ok { + if !r3.DidEvict { t.Fatal("expected eviction after third push on cap=1") } - if evicted != 20 { - t.Errorf("LastEvicted() = %d, want 20", evicted) + if r3.Evicted != 20 { + t.Errorf("Evicted = %d, want 20", r3.Evicted) } } @@ -203,10 +187,10 @@ func TestRingBufferLenAndCap(t *testing.T) { func TestRingBufferWrappingMaintainsOrder(t *testing.T) { rb := NewRingBuffer[int](3) - s0 := rb.Push(10) - v, _ := rb.At(s0) + r0 := rb.Push(10) + v, _ := rb.At(r0.Slot) if v != 10 { - t.Fatalf("initial At(%d) = %d, want 10", s0, v) + t.Fatalf("initial At(%d) = %d, want 10", r0.Slot, v) } rb.Push(20) @@ -216,7 +200,7 @@ func TestRingBufferWrappingMaintainsOrder(t *testing.T) { rb.Push(60) rb.Push(70) - _, ok := rb.At(s0) + _, ok := rb.At(r0.Slot) if ok { t.Error("slot 0 should be evicted after many pushes") } @@ -240,14 +224,14 @@ func TestRingBufferWrappingMaintainsOrder(t *testing.T) { func TestRingBufferAtEdgeCases(t *testing.T) { rb := NewRingBuffer[int](3) - slot := rb.Push(10) + r := rb.Push(10) - v, ok := rb.At(slot + 1) + v, ok := rb.At(r.Slot + 1) if ok { t.Errorf("At(slot+1) should return false, got (%d, true)", v) } - v, ok = rb.At(slot - 1) + v, ok = rb.At(r.Slot - 1) if ok { t.Errorf("At(slot-1) should return false, got (%d, true)", v) } diff --git a/internal/tui/trace_ingestion_test.go b/internal/tui/trace_ingestion_test.go new file mode 100644 index 0000000..de8d504 --- /dev/null +++ b/internal/tui/trace_ingestion_test.go @@ -0,0 +1,289 @@ +package tui + +import ( + "testing" + + "tangled.org/samatkins.co/dockscope/internal/domain" + "tangled.org/samatkins.co/dockscope/internal/parser" +) + +func TestInitSelectsRunningContainers(t *testing.T) { + t.Parallel() + + containers := []domain.Container{ + {ID: "a", State: domain.StateRunning}, + {ID: "b", State: domain.StateRunning}, + {ID: "c", State: domain.StateExited}, + } + + m := NewModel(t.Context(), containers, nil) + _ = m.Init() + + if !m.isSelected("a") { + t.Error("running container a should be selected after init") + } + + if !m.isSelected("b") { + t.Error("running container b should be selected after init") + } + + if m.isSelected("c") { + t.Error("exited container c should not be selected after init") + } +} + +func TestInitStartsLogStreamsForAllSelected(t *testing.T) { + t.Parallel() + + containers := []domain.Container{ + {ID: "a", Name: "alpha", State: domain.StateRunning}, + {ID: "b", Name: "beta", State: domain.StateRunning}, + {ID: "c", Name: "gamma", State: domain.StateRunning}, + } + + m := NewModel(t.Context(), containers, &capturingFetcher{}) + _ = m.Init() + + if len(m.logStreamStops) != 3 { + t.Errorf("expected 3 log streams started, got %d", len(m.logStreamStops)) + } + + for _, c := range containers { + if _, exists := m.logStreamStops[c.ID]; !exists { + t.Errorf("log stream should be started for container %s", c.ID) + } + } +} + +func TestIngestLogLinePopulatesTraceIndex(t *testing.T) { + t.Parallel() + + m := NewModel(t.Context(), nil, nil) + _ = m.Init() + + entry := parser.LogEntry{ + TraceID: "abc123", + Raw: "test message", + } + + m.ingestLogLine(entry) + + indices, ok := m.traceIndex["abc123"] + if !ok { + t.Fatal("traceIndex should contain key abc123") + } + + if len(indices) != 1 { + t.Fatalf("expected 1 index, got %d", len(indices)) + } + + v, ok := m.ringBuf.At(indices[0]) + if !ok { + t.Fatal("ring buffer should have entry at trace index") + } + + if v.TraceID != "abc123" { + t.Errorf("entry at index should have TraceID abc123, got %s", v.TraceID) + } +} + +func TestIngestLogLineEvictionRemovesFromTraceIndex(t *testing.T) { + t.Parallel() + + m := NewModel(t.Context(), nil, nil) + _ = m.Init() + + m.ringBuf = NewRingBuffer[parser.LogEntry](2) + + m.ingestLogLine(parser.LogEntry{TraceID: "trace-a", Raw: "a"}) + m.ingestLogLine(parser.LogEntry{TraceID: "trace-b", Raw: "b"}) + m.ingestLogLine(parser.LogEntry{TraceID: "trace-c", Raw: "c"}) + + _, ok := m.traceIndex["trace-a"] + if ok { + t.Error("trace-a should be evicted from traceIndex after buffer wraps") + } + + _, ok = m.traceIndex["trace-b"] + if !ok { + t.Error("trace-b should still be in traceIndex") + } + + _, ok = m.traceIndex["trace-c"] + if !ok { + t.Error("trace-c should be in traceIndex") + } +} + +func TestIngestLogLineEvictionDeletesEmptyKey(t *testing.T) { + t.Parallel() + + m := NewModel(t.Context(), nil, nil) + _ = m.Init() + + m.ringBuf = NewRingBuffer[parser.LogEntry](2) + + m.ingestLogLine(parser.LogEntry{TraceID: "only-entry", Raw: "1"}) + m.ingestLogLine(parser.LogEntry{TraceID: "filler", Raw: "2"}) + m.ingestLogLine(parser.LogEntry{TraceID: "filler", Raw: "3"}) + + _, ok := m.traceIndex["only-entry"] + if ok { + t.Error("only-entry key should be deleted after its sole entry evicted") + } +} + +func TestIngestLogLineTraceIndexReferencesRingBufferSlot(t *testing.T) { + t.Parallel() + + m := NewModel(t.Context(), nil, nil) + _ = m.Init() + + entries := []parser.LogEntry{ + {TraceID: "a", Raw: "1"}, + {TraceID: "b", Raw: "2"}, + {TraceID: "a", Raw: "3"}, + {TraceID: "c", Raw: "4"}, + {TraceID: "a", Raw: "5"}, + {TraceID: "b", Raw: "6"}, + } + + for _, e := range entries { + m.ingestLogLine(e) + } + + indices := m.traceIndex["a"] + if len(indices) != 3 { + t.Fatalf("expected 3 indices for trace a, got %d", len(indices)) + } + + for _, idx := range indices { + entry, ok := m.ringBuf.At(idx) + if !ok { + t.Errorf("slot %d should be valid", idx) + continue + } + + if entry.TraceID != "a" { + t.Errorf("entry at slot %d has TraceID %s, want a", idx, entry.TraceID) + } + } +} + +func TestDeselectCancelsLogStream(t *testing.T) { + t.Parallel() + + m := NewModel(t.Context(), []domain.Container{ + {ID: "a", Name: "alpha", State: domain.StateRunning}, + {ID: "b", Name: "beta", State: domain.StateRunning}, + {ID: "c", Name: "gamma", State: domain.StateRunning}, + }, &capturingFetcher{}) + + m.selected["a"] = true + m.selected["b"] = true + m.selected["c"] = true + + _ = m.syncLogStreams() + + if len(m.logStreamStops) != 3 { + t.Fatalf("expected 3 log streams, got %d", len(m.logStreamStops)) + } + + m.ingestLogLine(parser.LogEntry{ContainerID: "b", Container: "beta", Raw: "log from beta"}) + bufLen := m.ringBuf.Len() + + m.selected["b"] = false + _ = m.syncLogStreams() + + if _, exists := m.logStreamStops["b"]; exists { + t.Error("log stream for b should be cancelled and removed") + } + + if m.ringBuf.Len() != bufLen { + t.Error("entries should remain in ring buffer after deselection") + } +} + +func TestSelectStartsLogStream(t *testing.T) { + t.Parallel() + + m := NewModel(t.Context(), []domain.Container{ + {ID: "a", Name: "alpha", State: domain.StateRunning}, + {ID: "b", Name: "beta", State: domain.StateRunning}, + }, &capturingFetcher{}) + + m.selected["a"] = true + + _ = m.syncLogStreams() + + if len(m.logStreamStops) != 1 { + t.Fatalf("expected 1 log stream for a, got %d", len(m.logStreamStops)) + } + + m.selected["b"] = true + _ = m.syncLogStreams() + + if _, exists := m.logStreamStops["b"]; !exists { + t.Error("log stream for b should be started after selection") + } + + if len(m.logStreamStops) != 2 { + t.Errorf("expected 2 log streams, got %d", len(m.logStreamStops)) + } +} + +func TestParseIntegrationSlogJSONEntry(t *testing.T) { + t.Parallel() + + m := NewModel(t.Context(), nil, nil) + _ = m.Init() + + raw := `{"time":"2026-05-09T14:32:01.088Z","level":"INFO","msg":"GET /orders/4201","trace_id":"abc123def456","span_id":"span-1"}` + entry, _ := m.parser.Parse(raw) + entry.ContainerID = "api-gateway" + entry.Container = "api-gateway" + + m.ingestLogLine(entry) + + indices := m.traceIndex["abc123def456"] + if len(indices) != 1 { + t.Fatalf("expected 1 index for trace abc123def456, got %d", len(indices)) + } + + stored, ok := m.ringBuf.At(indices[0]) + if !ok { + t.Fatal("entry should be in ring buffer") + } + + if stored.Level != "INFO" { + t.Errorf("Level = %q, want INFO", stored.Level) + } + + if stored.ContainerID != "api-gateway" { + t.Errorf("ContainerID = %q, want api-gateway", stored.ContainerID) + } +} + +func TestStartLogStreamSkipsDuplicate(t *testing.T) { + t.Parallel() + + m := NewModel(t.Context(), []domain.Container{ + {ID: "a", Name: "alpha", State: domain.StateRunning}, + }, &capturingFetcher{}) + + cmd1 := m.startLogStream("a", "alpha") + if cmd1 == nil { + t.Fatal("first startLogStream should return a command") + } + + streams1 := len(m.logStreamStops) + + cmd2 := m.startLogStream("a", "alpha") + if cmd2 != nil { + t.Error("second startLogStream should return nil, stream already started") + } + + if len(m.logStreamStops) != streams1 { + t.Errorf("log streams count changed from %d to %d after duplicate call", streams1, len(m.logStreamStops)) + } +} diff --git a/internal/tui/update.go b/internal/tui/update.go index c733a55..c6da457 100644 --- a/internal/tui/update.go +++ b/internal/tui/update.go @@ -30,12 +30,6 @@ func (m *Model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { } return m.updateLive(msg) - case StatsMsg: - return m.handleStats(msg) - case LogMsg: - return m.handleLogMsg(msg) - case logStreamOpenedMsg: - return m.handleLogStreamOpened(msg) case tea.WindowSizeMsg: m.width = msg.Width m.height = msg.Height @@ -43,6 +37,23 @@ func (m *Model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { return m, nil case tickMsg: return m.handleTick() + default: + return m.updateNonKey(msg) + } +} + +func (m *Model) updateNonKey(msg tea.Msg) (tea.Model, tea.Cmd) { + switch msg := msg.(type) { + case StatsMsg: + return m.handleStats(msg) + case LogMsg: + return m.handleLogMsg(msg) + case ParsedLogMsg: + return m.handleParsedLogMsg(msg) + case logStreamOpenedMsg: + return m.handleLogStreamOpened(msg) + case logStreamStartedMsg: + return m.handleLogStreamStarted(msg) default: return m, nil } @@ -298,7 +309,7 @@ func (m *Model) handleSelect() (tea.Model, tea.Cmd) { "selected", m.isSelected(c.ID), ) - return m, nil + return m, m.syncLogStreams() } func (m *Model) toggleFollow() { @@ -310,3 +321,14 @@ func (m *Model) nextPane() { m.activePane = (m.activePane + 1) % paneCount slog.Debug("switched pane", "pane", m.activePane) } + +func (m *Model) handleParsedLogMsg(msg ParsedLogMsg) (tea.Model, tea.Cmd) { + m.ingestLogLine(msg.Entry) + + ch, ok := m.logStreamChans[msg.Entry.ContainerID] + if !ok { + return m, nil + } + + return m, listenOnParsedLogCh(ch) +} diff --git a/justfile b/justfile index ca05fcf..b37e768 100644 --- a/justfile +++ b/justfile @@ -1,5 +1,4 @@ # Show available recipes -[group('dev')] default: @just --list