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