1// Package logwriter provides container log collection with file rotation.
2package logwriter
3
4import (
5 "context"
6 "io"
7 "os"
8 "path/filepath"
9 "strings"
10 "sync"
11 "sync/atomic"
12
13 "github.com/bnema/zerowrap"
14 "github.com/docker/docker/pkg/stdcopy"
15 "gopkg.in/natefinch/lumberjack.v2"
16)
17
18// Config holds the configuration for the log writer.
19type Config struct {
20 // Dir is the directory where container logs are stored.
21 Dir string
22 // MaxSize is the maximum size in megabytes before rotation.
23 MaxSize int
24 // MaxBackups is the number of old log files to retain.
25 MaxBackups int
26 // MaxAge is the maximum number of days to retain old log files.
27 MaxAge int
28}
29
30// LogWriter implements the ContainerLogWriter interface.
31type LogWriter struct {
32 config Config
33 streams map[string]*streamInfo
34 mu sync.RWMutex
35}
36
37// streamInfo tracks an active log stream for a container.
38type streamInfo struct {
39 containerID string
40 domain string
41 logStream io.ReadCloser
42 logger *lumberjack.Logger
43 stopped atomic.Bool
44 done chan struct{}
45}
46
47// New creates a new LogWriter.
48func New(config Config) (*LogWriter, error) {
49 // Ensure the log directory exists
50 if err := os.MkdirAll(config.Dir, 0700); err != nil {
51 return nil, err
52 }
53
54 return &LogWriter{
55 config: config,
56 streams: make(map[string]*streamInfo),
57 }, nil
58}
59
60// StartLogging begins collecting logs for a container.
61func (w *LogWriter) StartLogging(ctx context.Context, containerID string, domain string, logStream io.ReadCloser) error {
62 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
63 zerowrap.FieldLayer: "adapter",
64 zerowrap.FieldAdapter: "logwriter",
65 zerowrap.FieldAction: "StartLogging",
66 zerowrap.FieldEntityID: containerID,
67 "domain": domain,
68 })
69 log := zerowrap.FromCtx(ctx)
70
71 w.mu.Lock()
72 defer w.mu.Unlock()
73
74 // Stop existing logging for this container if any
75 if existing, ok := w.streams[containerID]; ok {
76 log.Debug().Msg("stopping existing log stream before starting new one")
77 w.stopStreamLocked(existing)
78 }
79
80 // Sanitize domain for filename
81 filename := sanitizeDomain(domain) + ".log"
82 logPath := filepath.Join(w.config.Dir, filename)
83
84 // Create lumberjack logger for rotation
85 logger := &lumberjack.Logger{
86 Filename: logPath,
87 MaxSize: w.config.MaxSize,
88 MaxBackups: w.config.MaxBackups,
89 MaxAge: w.config.MaxAge,
90 Compress: true,
91 }
92
93 done := make(chan struct{})
94
95 info := &streamInfo{
96 containerID: containerID,
97 domain: domain,
98 logStream: logStream,
99 logger: logger,
100 done: done,
101 }
102
103 w.streams[containerID] = info
104
105 // Start streaming goroutine
106 go w.streamLogs(info, log)
107
108 log.Info().Str("path", logPath).Msg("started container log collection")
109 return nil
110}
111
112// StopLogging stops log collection for a container.
113func (w *LogWriter) StopLogging(containerID string) error {
114 w.mu.Lock()
115 defer w.mu.Unlock()
116
117 if info, ok := w.streams[containerID]; ok {
118 w.stopStreamLocked(info)
119 delete(w.streams, containerID)
120 }
121 return nil
122}
123
124// Close stops all logging and releases resources.
125func (w *LogWriter) Close() error {
126 w.mu.Lock()
127 defer w.mu.Unlock()
128
129 for containerID, info := range w.streams {
130 w.stopStreamLocked(info)
131 delete(w.streams, containerID)
132 }
133 return nil
134}
135
136// stopStreamLocked stops a stream (must be called with lock held).
137func (w *LogWriter) stopStreamLocked(info *streamInfo) {
138 // Signal the goroutine to stop
139 info.stopped.Store(true)
140
141 // Close the log stream to unblock reads
142 info.logStream.Close()
143
144 // Wait for goroutine to finish
145 <-info.done
146
147 // Close the lumberjack logger
148 info.logger.Close()
149}
150
151// streamLogs copies from the Docker log stream to the file.
152func (w *LogWriter) streamLogs(info *streamInfo, log zerowrap.Logger) {
153 defer close(info.done)
154
155 // Docker container logs are multiplexed with 8-byte headers.
156 // stdcopy.StdCopy demultiplexes them to separate stdout/stderr writers.
157 // We combine both to a single file.
158 _, err := stdcopy.StdCopy(info.logger, info.logger, info.logStream)
159
160 if err != nil && !info.stopped.Load() {
161 // Only log error if we weren't intentionally stopped
162 log.Warn().Err(err).
163 Str("container_id", info.containerID).
164 Str("domain", info.domain).
165 Msg("container log streaming ended with error")
166 }
167}
168
169// sanitizeDomain converts a domain to a safe filename.
170func sanitizeDomain(domain string) string {
171 // Replace dots with underscores, remove other unsafe characters
172 safe := strings.ReplaceAll(domain, ".", "_")
173 safe = strings.ReplaceAll(safe, "/", "_")
174 safe = strings.ReplaceAll(safe, ":", "_")
175 safe = strings.ReplaceAll(safe, " ", "_")
176 return safe
177}