1// Package logexport follows app container output and exports it as
2// log records, one source identity per app service.
3package logexport
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "sync"
10 "time"
11
12 "github.com/bnema/zerowrap"
13
14 "github.com/bnema/gordon/internal/boundaries/out"
15 "github.com/bnema/gordon/internal/domain"
16)
17
18const (
19 defaultReconcileInterval = 10 * time.Second
20 defaultRetryDelay = 2 * time.Second
21)
22
23// Collector keeps one follower per exportable ACTIVE container.
24//
25// A container is exportable when its app is not stopped and its service
26// did not opt out (domain.AppService.LogExportDisabled). Reconciliation
27// polls ACTIVE state so deploys, stops, and opt-out changes converge
28// without coupling the deployment engine to log export.
29type Collector struct {
30 state out.AppStateReader
31 streamer out.ContainerLogStreamer
32 exporter out.LogExporter
33
34 reconcileInterval time.Duration
35 retryDelay time.Duration
36 // startedAt bounds replay: a follower exports output emitted since
37 // the collector was created, so a container deployed later is
38 // exported from its first line, while restarting Gordon never
39 // re-exports older output.
40 startedAt time.Time
41
42 mu sync.Mutex
43 followers map[string]*follower
44 cursors map[string]time.Time
45 wg sync.WaitGroup
46}
47
48type follower struct {
49 source domain.LogSource
50 cancel context.CancelFunc
51}
52
53// NewCollector creates a collector. Call Run to start it.
54func NewCollector(state out.AppStateReader, streamer out.ContainerLogStreamer, exporter out.LogExporter) *Collector {
55 return &Collector{
56 state: state,
57 streamer: streamer,
58 exporter: exporter,
59 reconcileInterval: defaultReconcileInterval,
60 retryDelay: defaultRetryDelay,
61 startedAt: time.Now(),
62 followers: map[string]*follower{},
63 cursors: map[string]time.Time{},
64 }
65}
66
67// Run reconciles until ctx is canceled, then stops every follower and
68// waits for them to exit.
69func (c *Collector) Run(ctx context.Context) {
70 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
71 zerowrap.FieldLayer: "usecase",
72 zerowrap.FieldUseCase: "LogExport",
73 })
74 defer c.stopAll()
75 log := zerowrap.FromCtx(ctx)
76
77 ticker := time.NewTicker(c.reconcileInterval)
78 defer ticker.Stop()
79 for {
80 if err := c.Reconcile(ctx); err != nil && ctx.Err() == nil {
81 log.Warn().Err(err).Msg("log export reconciliation failed")
82 }
83 select {
84 case <-ctx.Done():
85 return
86 case <-ticker.C:
87 }
88 }
89}
90
91// Reconcile starts followers for new exportable containers and stops
92// followers whose container is gone, stopped, or opted out.
93func (c *Collector) Reconcile(ctx context.Context) error {
94 desired, activeIDs, err := c.desiredContainers(ctx)
95 if err != nil {
96 return err
97 }
98
99 c.mu.Lock()
100 defer c.mu.Unlock()
101 for containerID, f := range c.followers {
102 if source, ok := desired[containerID]; !ok || source != f.source {
103 f.cancel()
104 delete(c.followers, containerID)
105 }
106 }
107 for containerID, source := range desired {
108 if _, ok := c.followers[containerID]; ok {
109 continue
110 }
111 c.startLocked(ctx, containerID, source)
112 }
113 for containerID := range c.cursors {
114 if _, ok := activeIDs[containerID]; !ok {
115 delete(c.cursors, containerID)
116 }
117 }
118 return nil
119}
120
121// desiredContainers maps exportable ACTIVE container IDs to their source,
122// plus the set of every container ID referenced by an ACTIVE service
123// (exportable or not) so callers can retire per-container state only when
124// the ID is gone from all ACTIVE services.
125func (c *Collector) desiredContainers(ctx context.Context) (map[string]domain.LogSource, map[string]struct{}, error) {
126 apps, err := c.state.ListApps(ctx)
127 if err != nil {
128 return nil, nil, fmt.Errorf("logexport: list apps: %w", err)
129 }
130 desired := map[string]domain.LogSource{}
131 activeIDs := map[string]struct{}{}
132 for _, app := range apps {
133 active, found, err := c.state.LoadActive(ctx, app)
134 if err != nil {
135 return nil, nil, fmt.Errorf("logexport: load active state for app %q: %w", app, err)
136 }
137 if !found || active.StopIntent {
138 continue
139 }
140 for name, svc := range active.Services {
141 if svc.Container == "" {
142 continue
143 }
144 activeIDs[svc.Container] = struct{}{}
145 if svc.Spec.LogExportDisabled {
146 continue
147 }
148 desired[svc.Container] = domain.LogSource{App: app, Service: name}
149 }
150 }
151 return desired, activeIDs, nil
152}
153
154func (c *Collector) startLocked(parent context.Context, containerID string, source domain.LogSource) {
155 ctx, cancel := context.WithCancel(parent)
156 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
157 zerowrap.FieldEntityID: containerID,
158 "app": source.App,
159 "service": source.Service,
160 })
161 c.followers[containerID] = &follower{source: source, cancel: cancel}
162 since := c.startedAt
163 if cursor, ok := c.cursors[containerID]; ok && cursor.After(since) {
164 since = cursor
165 }
166 c.wg.Add(1)
167 go func() {
168 defer c.wg.Done()
169 c.follow(ctx, containerID, source, since)
170 }()
171}
172
173// follow streams one container until canceled, starting at since.
174// A stream that ends (container restarting) resumes after the last
175// exported line, so a reconnect neither duplicates nor replays output.
176func (c *Collector) follow(ctx context.Context, containerID string, source domain.LogSource, since time.Time) {
177 log := zerowrap.FromCtx(ctx)
178 for {
179 err := c.streamer.StreamContainerLogs(ctx, containerID, since, func(line domain.ContainerLogLine) {
180 c.exporter.Export(ctx, domain.LogRecord{
181 Time: line.Time,
182 Source: source,
183 Type: domain.LogTypeContainer,
184 Stream: line.Stream,
185 Body: line.Body,
186 })
187 if line.Time.After(since) {
188 since = line.Time.Add(time.Nanosecond)
189 c.mu.Lock()
190 if cursor, ok := c.cursors[containerID]; !ok || since.After(cursor) {
191 c.cursors[containerID] = since
192 }
193 c.mu.Unlock()
194 }
195 })
196 if ctx.Err() != nil {
197 return
198 }
199 if err != nil && !errors.Is(err, domain.ErrContainerNotFound) {
200 log.Debug().Err(err).Msg("container log stream interrupted, retrying")
201 }
202 select {
203 case <-ctx.Done():
204 return
205 case <-time.After(c.retryDelay):
206 }
207 }
208}
209
210func (c *Collector) stopAll() {
211 c.mu.Lock()
212 for containerID, f := range c.followers {
213 f.cancel()
214 delete(c.followers, containerID)
215 }
216 c.mu.Unlock()
217 c.wg.Wait()
218}