1package logging
2
3import (
4 "bufio"
5 "context"
6 "fmt"
7 "sync"
8 "time"
9
10 "gordon/internal/config"
11 "gordon/pkg/runtime"
12
13 "github.com/rs/zerolog/log"
14 "gopkg.in/natefinch/lumberjack.v2"
15)
16
17// LogCollector manages log collection for a single container
18type LogCollector struct {
19 containerID string
20 containerName string
21 runtime runtime.Runtime
22 logWriter *lumberjack.Logger
23 ctx context.Context
24 cancel context.CancelFunc
25 wg sync.WaitGroup
26}
27
28// LogManager manages all container log collectors
29type LogManager struct {
30 collectors map[string]*LogCollector
31 config *config.Config
32 runtime runtime.Runtime
33 mu sync.RWMutex
34}
35
36// NewLogManager creates a new log manager
37func NewLogManager(cfg *config.Config, rt runtime.Runtime) *LogManager {
38 return &LogManager{
39 collectors: make(map[string]*LogCollector),
40 config: cfg,
41 runtime: rt,
42 }
43}
44
45// StartCollection starts log collection for a container
46func (lm *LogManager) StartCollection(containerID, containerName string) error {
47 if !lm.config.Logging.Enabled {
48 return nil // Logging disabled, nothing to do
49 }
50
51 lm.mu.Lock()
52 defer lm.mu.Unlock()
53
54 // Stop existing collector if it exists
55 if existing, exists := lm.collectors[containerID]; exists {
56 log.Debug().Str("container_id", containerID).Msg("Stopping existing log collector")
57 existing.Stop()
58 delete(lm.collectors, containerID)
59 }
60
61 // Create log writer for this container
62 logWriter, err := GetContainerLogWriter(lm.config, containerID, containerName)
63 if err != nil {
64 return fmt.Errorf("failed to create log writer for container %s: %w", containerID, err)
65 }
66
67 // Create collector context
68 ctx, cancel := context.WithCancel(context.Background())
69
70 collector := &LogCollector{
71 containerID: containerID,
72 containerName: containerName,
73 runtime: lm.runtime,
74 logWriter: logWriter,
75 ctx: ctx,
76 cancel: cancel,
77 }
78
79 // Start collection goroutine
80 collector.wg.Add(1)
81 go collector.collectLogs()
82
83 // Store collector
84 lm.collectors[containerID] = collector
85
86 log.Info().
87 Str("container_id", containerID).
88 Str("container_name", containerName).
89 Msg("Started container log collection")
90
91 return nil
92}
93
94// StopCollection stops log collection for a container
95func (lm *LogManager) StopCollection(containerID string) {
96 lm.mu.Lock()
97 defer lm.mu.Unlock()
98
99 if collector, exists := lm.collectors[containerID]; exists {
100 log.Debug().Str("container_id", containerID).Msg("Stopping log collection")
101 collector.Stop()
102 delete(lm.collectors, containerID)
103 }
104}
105
106// StopAll stops all log collection
107func (lm *LogManager) StopAll() {
108 lm.mu.Lock()
109 defer lm.mu.Unlock()
110
111 for containerID, collector := range lm.collectors {
112 log.Debug().Str("container_id", containerID).Msg("Stopping log collection")
113 collector.Stop()
114 }
115 lm.collectors = make(map[string]*LogCollector)
116}
117
118// GetActiveCollectors returns the number of active collectors
119func (lm *LogManager) GetActiveCollectors() int {
120 lm.mu.RLock()
121 defer lm.mu.RUnlock()
122 return len(lm.collectors)
123}
124
125// collectLogs runs the log collection loop for a single container
126func (lc *LogCollector) collectLogs() {
127 defer lc.wg.Done()
128
129 log.Debug().
130 Str("container_id", lc.containerID).
131 Str("container_name", lc.containerName).
132 Msg("Starting log collection goroutine")
133
134 // Retry loop for log collection
135 retryDelay := time.Second
136 maxRetryDelay := 30 * time.Second
137
138 for {
139 select {
140 case <-lc.ctx.Done():
141 log.Debug().Str("container_id", lc.containerID).Msg("Log collection context cancelled")
142 return
143 default:
144 // Attempt to collect logs
145 if err := lc.collectLogsOnce(); err != nil {
146 log.Warn().
147 Err(err).
148 Str("container_id", lc.containerID).
149 Dur("retry_delay", retryDelay).
150 Msg("Log collection failed, retrying")
151
152 // Wait before retrying, with exponential backoff
153 select {
154 case <-lc.ctx.Done():
155 return
156 case <-time.After(retryDelay):
157 // Increase retry delay, up to maximum
158 retryDelay *= 2
159 if retryDelay > maxRetryDelay {
160 retryDelay = maxRetryDelay
161 }
162 }
163 continue
164 }
165
166 // Reset retry delay on success
167 retryDelay = time.Second
168 }
169 }
170}
171
172// collectLogsOnce attempts to collect logs once
173func (lc *LogCollector) collectLogsOnce() error {
174 // Get log stream from runtime
175 logStream, err := lc.runtime.GetContainerLogs(lc.ctx, lc.containerID, true)
176 if err != nil {
177 return fmt.Errorf("failed to get log stream: %w", err)
178 }
179 defer logStream.Close()
180
181 log.Debug().Str("container_id", lc.containerID).Msg("Started reading container logs")
182
183 // Read logs line by line and write to file
184 scanner := bufio.NewScanner(logStream)
185 for scanner.Scan() {
186 select {
187 case <-lc.ctx.Done():
188 return nil
189 default:
190 // Get log line
191 line := scanner.Text()
192
193 // Add timestamp and write to log file
194 timestampedLine := fmt.Sprintf("%s %s\n",
195 time.Now().Format(time.RFC3339),
196 line)
197
198 if _, err := lc.logWriter.Write([]byte(timestampedLine)); err != nil {
199 log.Warn().
200 Err(err).
201 Str("container_id", lc.containerID).
202 Msg("Failed to write container log line")
203 }
204 }
205 }
206
207 // Check for scanner errors
208 if err := scanner.Err(); err != nil {
209 return fmt.Errorf("error reading log stream: %w", err)
210 }
211
212 return nil
213}
214
215// Stop gracefully stops the log collector
216func (lc *LogCollector) Stop() {
217 log.Debug().Str("container_id", lc.containerID).Msg("Stopping log collector")
218
219 // Cancel context to stop goroutines
220 lc.cancel()
221
222 // Wait for collection goroutine to finish
223 lc.wg.Wait()
224
225 // Close log writer
226 if lc.logWriter != nil {
227 if err := lc.logWriter.Close(); err != nil {
228 log.Warn().
229 Err(err).
230 Str("container_id", lc.containerID).
231 Msg("Failed to close log writer")
232 }
233 }
234
235 log.Debug().Str("container_id", lc.containerID).Msg("Log collector stopped")
236}