1package logexport
2
3import (
4 "context"
5 "errors"
6 "sync"
7 "testing"
8 "time"
9
10 "github.com/stretchr/testify/assert"
11 "github.com/stretchr/testify/mock"
12 "github.com/stretchr/testify/require"
13
14 "github.com/bnema/gordon/internal/boundaries/out/mocks"
15 "github.com/bnema/gordon/internal/domain"
16)
17
18func activeApp(app string, stopped bool, services map[string]domain.AppEffectiveService) domain.AppActive {
19 return domain.AppActive{App: app, StopIntent: stopped, Services: services}
20}
21
22func svc(container string, disabled bool) domain.AppEffectiveService {
23 return domain.AppEffectiveService{Container: container, Spec: domain.AppService{LogExportDisabled: disabled}}
24}
25
26// blockingStreamer blocks each follower until canceled and records calls.
27type blockingStreamer struct {
28 mu sync.Mutex
29 started map[string]int
30 stopped map[string]int
31}
32
33func newBlockingStreamer() *blockingStreamer {
34 return &blockingStreamer{started: map[string]int{}, stopped: map[string]int{}}
35}
36
37func (s *blockingStreamer) StreamContainerLogs(ctx context.Context, containerID string, _ time.Time, _ func(domain.ContainerLogLine)) error {
38 s.mu.Lock()
39 s.started[containerID]++
40 s.mu.Unlock()
41 <-ctx.Done()
42 s.mu.Lock()
43 s.stopped[containerID]++
44 s.mu.Unlock()
45 return ctx.Err()
46}
47
48func (s *blockingStreamer) counts(containerID string) (int, int) {
49 s.mu.Lock()
50 defer s.mu.Unlock()
51 return s.started[containerID], s.stopped[containerID]
52}
53
54func TestCollector_FollowsOnlyExportableContainers(t *testing.T) {
55 state := mocks.NewMockAppStateReader(t)
56 state.EXPECT().ListApps(mock.Anything).Return([]string{"blog", "paused"}, nil)
57 state.EXPECT().LoadActive(mock.Anything, "blog").Return(activeApp("blog", false, map[string]domain.AppEffectiveService{
58 "web": svc("c-web", false),
59 "db": svc("c-db", true), // service opted out
60 "new": svc("", false), // no container yet
61 }), true, nil)
62 state.EXPECT().LoadActive(mock.Anything, "paused").Return(activeApp("paused", true, map[string]domain.AppEffectiveService{
63 "web": svc("c-paused", false),
64 }), true, nil)
65 streamer := newBlockingStreamer()
66 c := NewCollector(state, streamer, mocks.NewMockLogExporter(t))
67
68 ctx, cancel := context.WithCancel(context.Background())
69 require.NoError(t, c.Reconcile(ctx))
70 require.Eventually(t, func() bool { started, _ := streamer.counts("c-web"); return started == 1 }, time.Second, time.Millisecond)
71 cancel()
72 c.stopAll()
73
74 for _, id := range []string{"c-db", "c-paused"} {
75 started, _ := streamer.counts(id)
76 assert.Zero(t, started, id)
77 }
78 _, stopped := streamer.counts("c-web")
79 assert.Equal(t, 1, stopped, "stopAll waits for followers")
80}
81
82func TestCollector_StopsFollowerWhenServiceOptsOut(t *testing.T) {
83 state := mocks.NewMockAppStateReader(t)
84 state.EXPECT().ListApps(mock.Anything).Return([]string{"blog"}, nil)
85 state.EXPECT().LoadActive(mock.Anything, "blog").Return(activeApp("blog", false, map[string]domain.AppEffectiveService{
86 "web": svc("c-web", false),
87 }), true, nil).Once()
88 state.EXPECT().LoadActive(mock.Anything, "blog").Return(activeApp("blog", false, map[string]domain.AppEffectiveService{
89 "web": svc("c-web", true),
90 }), true, nil).Once()
91 streamer := newBlockingStreamer()
92 c := NewCollector(state, streamer, mocks.NewMockLogExporter(t))
93 defer c.stopAll()
94
95 require.NoError(t, c.Reconcile(context.Background()))
96 require.NoError(t, c.Reconcile(context.Background()))
97
98 require.Eventually(t, func() bool { _, stopped := streamer.counts("c-web"); return stopped == 1 }, time.Second, time.Millisecond)
99}
100
101func TestCollector_ExportsLinesWithSourceIdentity(t *testing.T) {
102 state := mocks.NewMockAppStateReader(t)
103 state.EXPECT().ListApps(mock.Anything).Return([]string{"blog"}, nil)
104 state.EXPECT().LoadActive(mock.Anything, "blog").Return(activeApp("blog", false, map[string]domain.AppEffectiveService{
105 "web": svc("c-web", false),
106 }), true, nil)
107
108 ts := time.Date(2026, 9, 10, 12, 0, 0, 0, time.UTC)
109 streamer := mocks.NewMockContainerLogStreamer(t)
110 streamer.EXPECT().StreamContainerLogs(mock.Anything, "c-web", mock.Anything, mock.Anything).
111 RunAndReturn(func(ctx context.Context, _ string, _ time.Time, emit func(domain.ContainerLogLine)) error {
112 emit(domain.ContainerLogLine{Time: ts, Stream: domain.LogStreamStderr, Body: "boom"})
113 <-ctx.Done()
114 return ctx.Err()
115 })
116
117 exported := make(chan domain.LogRecord, 1)
118 exporter := mocks.NewMockLogExporter(t)
119 exporter.EXPECT().Export(mock.Anything, mock.Anything).Run(func(_ context.Context, r domain.LogRecord) {
120 exported <- r
121 }).Return()
122
123 c := NewCollector(state, streamer, exporter)
124 defer c.stopAll()
125 require.NoError(t, c.Reconcile(context.Background()))
126
127 select {
128 case record := <-exported:
129 assert.Equal(t, domain.LogRecord{
130 Time: ts,
131 Source: domain.LogSource{App: "blog", Service: "web"},
132 Type: domain.LogTypeContainer,
133 Stream: domain.LogStreamStderr,
134 Body: "boom",
135 }, record)
136 assert.Equal(t, "blog.web", record.Source.ServiceName())
137 case <-time.After(time.Second):
138 t.Fatal("no record exported")
139 }
140}
141
142func TestCollector_ResumesAfterLastLineOnReconnect(t *testing.T) {
143 state := mocks.NewMockAppStateReader(t)
144 state.EXPECT().ListApps(mock.Anything).Return([]string{"blog"}, nil)
145 state.EXPECT().LoadActive(mock.Anything, "blog").Return(activeApp("blog", false, map[string]domain.AppEffectiveService{
146 "web": svc("c-web", false),
147 }), true, nil)
148
149 start := time.Date(2026, 9, 10, 12, 0, 0, 0, time.UTC)
150 last := start.Add(time.Minute)
151 streamer := mocks.NewMockContainerLogStreamer(t)
152 streamer.EXPECT().StreamContainerLogs(mock.Anything, "c-web", start, mock.Anything).
153 RunAndReturn(func(_ context.Context, _ string, _ time.Time, emit func(domain.ContainerLogLine)) error {
154 emit(domain.ContainerLogLine{Time: last, Body: "line"})
155 return nil // stream ended: container restarting
156 }).Once()
157 resumed := make(chan time.Time, 1)
158 streamer.EXPECT().StreamContainerLogs(mock.Anything, "c-web", mock.Anything, mock.Anything).
159 RunAndReturn(func(ctx context.Context, _ string, since time.Time, _ func(domain.ContainerLogLine)) error {
160 resumed <- since
161 <-ctx.Done()
162 return ctx.Err()
163 }).Once()
164 exporter := mocks.NewMockLogExporter(t)
165 exporter.EXPECT().Export(mock.Anything, mock.Anything).Return()
166
167 c := NewCollector(state, streamer, exporter)
168 c.startedAt = start
169 c.retryDelay = time.Millisecond
170 defer c.stopAll()
171 require.NoError(t, c.Reconcile(context.Background()))
172
173 select {
174 case since := <-resumed:
175 assert.Equal(t, last.Add(time.Nanosecond), since)
176 case <-time.After(time.Second):
177 t.Fatal("follower did not reconnect")
178 }
179}
180
181func TestCollector_ReconcileWrapsStateErrors(t *testing.T) {
182 listErr := assert.AnError
183 state := mocks.NewMockAppStateReader(t)
184 state.EXPECT().ListApps(mock.Anything).Return(nil, listErr)
185 c := NewCollector(state, newBlockingStreamer(), mocks.NewMockLogExporter(t))
186 defer c.stopAll()
187
188 err := c.Reconcile(context.Background())
189 require.Error(t, err)
190 assert.ErrorIs(t, err, listErr)
191
192 loadErr := assert.AnError
193 state2 := mocks.NewMockAppStateReader(t)
194 state2.EXPECT().ListApps(mock.Anything).Return([]string{"blog"}, nil)
195 state2.EXPECT().LoadActive(mock.Anything, "blog").Return(domain.AppActive{}, false, loadErr)
196 c2 := NewCollector(state2, newBlockingStreamer(), mocks.NewMockLogExporter(t))
197 defer c2.stopAll()
198
199 err = c2.Reconcile(context.Background())
200 require.Error(t, err)
201 assert.ErrorIs(t, err, loadErr)
202}
203
204func TestCollector_RecreatedFollowResumesFromSavedCursor(t *testing.T) {
205 state := mocks.NewMockAppStateReader(t)
206 state.EXPECT().ListApps(mock.Anything).Return([]string{"blog"}, nil)
207 state.EXPECT().LoadActive(mock.Anything, "blog").Return(activeApp("blog", false, map[string]domain.AppEffectiveService{
208 "web": svc("c-web", false),
209 }), true, nil)
210
211 start := time.Date(2026, 9, 10, 12, 0, 0, 0, time.UTC)
212 last := start.Add(time.Minute)
213 streamed := make(chan time.Time, 4)
214 streamer := mocks.NewMockContainerLogStreamer(t)
215 first := make(chan struct{})
216 // The first stream emits one line then ends (container restart);
217 // later streams only record their resume point.
218 streamer.EXPECT().StreamContainerLogs(mock.Anything, "c-web", mock.Anything, mock.Anything).
219 RunAndReturn(func(_ context.Context, _ string, since time.Time, emit func(domain.ContainerLogLine)) error {
220 select {
221 case streamed <- since:
222 default:
223 }
224 select {
225 case <-first:
226 return errors.New("boom")
227 default:
228 close(first)
229 emit(domain.ContainerLogLine{Time: last, Body: "line"})
230 return nil
231 }
232 }).Maybe()
233 exporter := mocks.NewMockLogExporter(t)
234 exporter.EXPECT().Export(mock.Anything, mock.Anything).Return().Maybe()
235
236 c := NewCollector(state, streamer, exporter)
237 c.startedAt = start
238 c.retryDelay = 50 * time.Millisecond
239 defer c.stopAll()
240 require.NoError(t, c.Reconcile(context.Background()))
241
242 select {
243 case since := <-streamed:
244 assert.Equal(t, start, since, "first follow starts at collector start")
245 case <-time.After(2 * time.Second):
246 t.Fatal("first follow did not start")
247 }
248
249 // The emitted line must advance the saved cursor past startedAt.
250 require.Eventually(t, func() bool {
251 c.mu.Lock()
252 defer c.mu.Unlock()
253 cursor, ok := c.cursors["c-web"]
254 return ok && cursor.Equal(last.Add(time.Nanosecond))
255 }, 2*time.Second, 10*time.Millisecond, "cursor must advance past the exported line")
256
257 // Simulate the supervision loop recreating the follow (source
258 // reassignment): the new follow must resume from the saved cursor.
259 c.mu.Lock()
260 if f, ok := c.followers["c-web"]; ok {
261 f.cancel()
262 delete(c.followers, "c-web")
263 }
264 c.mu.Unlock()
265 require.NoError(t, c.Reconcile(context.Background()))
266
267 select {
268 case since := <-streamed:
269 assert.Equal(t, last.Add(time.Nanosecond), since, "recreated follow resumes from the saved cursor")
270 case <-time.After(2 * time.Second):
271 t.Fatal("recreated follow did not start")
272 }
273}
274
275func TestCollector_RetiresCursorWhenIDLeavesActive(t *testing.T) {
276 state := mocks.NewMockAppStateReader(t)
277 state.EXPECT().ListApps(mock.Anything).Return([]string{"blog"}, nil).Times(2)
278 state.EXPECT().LoadActive(mock.Anything, "blog").Return(activeApp("blog", false, map[string]domain.AppEffectiveService{
279 "web": svc("c-web", false),
280 }), true, nil).Once()
281 state.EXPECT().LoadActive(mock.Anything, "blog").Return(activeApp("blog", false, map[string]domain.AppEffectiveService{
282 "web": svc("c-new", false),
283 }), true, nil).Once()
284 streamer := newBlockingStreamer()
285 c := NewCollector(state, streamer, mocks.NewMockLogExporter(t))
286 defer c.stopAll()
287
288 require.NoError(t, c.Reconcile(context.Background()))
289 c.mu.Lock()
290 c.cursors["c-web"] = time.Now()
291 c.mu.Unlock()
292 require.NoError(t, c.Reconcile(context.Background()))
293
294 c.mu.Lock()
295 _, ok := c.cursors["c-web"]
296 c.mu.Unlock()
297 assert.False(t, ok, "cursor is dropped once the ID is gone from ACTIVE services")
298}