1// Package logs implements the log access use case.
2package logs
3
4import (
5 "bufio"
6 "bytes"
7 "context"
8 "encoding/binary"
9 "fmt"
10 "io"
11 "os"
12 "strconv"
13 "strings"
14 "time"
15
16 "github.com/bnema/zerowrap"
17
18 "github.com/bnema/gordon/internal/boundaries/out"
19 "github.com/bnema/gordon/internal/domain"
20 "github.com/bnema/gordon/pkg/childenv"
21)
22
23// appStateProvider resolves app services from the durable ACTIVE record.
24type appStateProvider interface {
25 LoadActive(ctx context.Context, app string) (domain.AppActive, bool, error)
26}
27
28// Service implements the LogService interface.
29type Service struct {
30 logFilePath string
31 fileLoggingEnabled bool
32 runtime out.ContainerRuntime
33 log zerowrap.Logger
34 // appState resolves log references to containers recorded in ACTIVE.
35 appState appStateProvider
36}
37
38// execCommandContext starts journalctl without Gordon's GORDON_* variables:
39// the child is dumpable again after exec, so it must not carry secrets.
40var execCommandContext = childenv.CommandContext
41
42// NewService creates a new log service. App state is wired after the store opens.
43func NewService(
44 logFilePath string,
45 fileLoggingEnabled bool,
46 runtime out.ContainerRuntime,
47 log zerowrap.Logger,
48) *Service {
49 return &Service{
50 logFilePath: logFilePath,
51 fileLoggingEnabled: fileLoggingEnabled,
52 runtime: runtime,
53 log: log,
54 }
55}
56
57// WithAppState wires the durable ACTIVE app-state reader used for log resolution.
58func (s *Service) WithAppState(provider appStateProvider) *Service {
59 s.appState = provider
60 return s
61}
62
63// containerForService resolves an app/service reference exclusively through ACTIVE.
64func (s *Service) containerForService(ctx context.Context, ref string) (string, error) {
65 app, service, ok := strings.Cut(ref, "/")
66 if !ok || app == "" || service == "" || strings.Contains(service, "/") || s.appState == nil {
67 return "", fmt.Errorf("invalid app/service log reference: %s", ref)
68 }
69 active, found, err := s.appState.LoadActive(ctx, app)
70 if err != nil {
71 return "", fmt.Errorf("failed to load active app %s: %w", app, err)
72 }
73 if !found {
74 return "", fmt.Errorf("active app/service not found: %s", ref)
75 }
76 target, found := active.Services[service]
77 if !found || target.Container == "" {
78 return "", fmt.Errorf("active app/service not found: %s", ref)
79 }
80 return target.Container, nil
81}
82
83// GetProcessLogs returns the last N lines of Gordon process logs.
84func (s *Service) GetProcessLogs(ctx context.Context, lines int) ([]string, error) {
85 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
86 zerowrap.FieldLayer: "usecase",
87 zerowrap.FieldUseCase: "GetProcessLogs",
88 "lines": lines,
89 })
90 log := zerowrap.FromCtx(ctx)
91
92 if s.useFileProcessLogs() {
93 if s.logFilePath == "" {
94 return nil, fmt.Errorf("log file path not configured")
95 }
96
97 file, err := os.Open(s.logFilePath)
98 if err != nil {
99 if os.IsNotExist(err) {
100 log.Warn().Str("log_path", s.logFilePath).
101 Msg("process log file missing, falling back to journalctl")
102 return s.getProcessLogsFromJournal(ctx, lines)
103 }
104 return nil, log.WrapErr(err, "failed to open log file")
105 }
106 defer file.Close()
107
108 return tailLines(file, lines)
109 }
110
111 return s.getProcessLogsFromJournal(ctx, lines)
112}
113
114// FollowProcessLogs returns a channel that streams Gordon process log lines.
115func (s *Service) FollowProcessLogs(ctx context.Context, initialLines int) (<-chan string, error) {
116 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
117 zerowrap.FieldLayer: "usecase",
118 zerowrap.FieldUseCase: "FollowProcessLogs",
119 "initial_lines": initialLines,
120 })
121 if !s.useFileProcessLogs() {
122 return s.followProcessLogsFromJournal(ctx, initialLines)
123 }
124
125 return s.followProcessLogsFromFileOrJournal(ctx, initialLines)
126}
127
128func (s *Service) useFileProcessLogs() bool {
129 return s.fileLoggingEnabled
130}
131
132func (s *Service) followProcessLogsFromFileOrJournal(
133 ctx context.Context,
134 initialLines int,
135) (<-chan string, error) {
136 log := zerowrap.FromCtx(ctx)
137
138 file, err := s.openProcessLogFile()
139 if err != nil {
140 if os.IsNotExist(err) {
141 log.Warn().Str("log_path", s.logFilePath).
142 Msg("process log file missing, falling back to journalctl follow")
143 return s.followProcessLogsFromJournal(ctx, initialLines)
144 }
145 return nil, log.WrapErr(err, "failed to open log file")
146 }
147
148 return s.followProcessLogsFromFile(ctx, file, initialLines), nil
149}
150
151func (s *Service) openProcessLogFile() (*os.File, error) {
152 if s.logFilePath == "" {
153 return nil, fmt.Errorf("log file path not configured")
154 }
155
156 return os.Open(s.logFilePath)
157}
158
159func (s *Service) followProcessLogsFromFile(
160 ctx context.Context,
161 file *os.File,
162 initialLines int,
163) <-chan string {
164 log := zerowrap.FromCtx(ctx)
165 ch := make(chan string, 100)
166
167 go func() {
168 defer close(ch)
169 defer file.Close()
170
171 if !s.emitInitialFileLogs(ctx, file, initialLines, ch) {
172 return
173 }
174 if err := s.seekToEndForFileFollow(file); err != nil {
175 log.Warn().Err(err).Msg("failed to seek to end")
176 return
177 }
178
179 s.streamFileFollowLogs(ctx, file, ch)
180 }()
181
182 return ch
183}
184
185func (s *Service) emitInitialFileLogs(
186 ctx context.Context,
187 file *os.File,
188 initialLines int,
189 ch chan<- string,
190) bool {
191 if initialLines <= 0 {
192 return true
193 }
194
195 log := zerowrap.FromCtx(ctx)
196 lines, err := tailLines(file, initialLines)
197 if err != nil {
198 log.Warn().Err(err).Msg("failed to read initial lines")
199 return true
200 }
201
202 for _, line := range lines {
203 if !sendLogLineWithChannel(ctx, ch, line) {
204 return false
205 }
206 }
207
208 return true
209}
210
211func (s *Service) seekToEndForFileFollow(file *os.File) error {
212 _, err := file.Seek(0, io.SeekEnd)
213 return err
214}
215
216func (s *Service) streamFileFollowLogs(ctx context.Context, file *os.File, ch chan<- string) {
217 log := zerowrap.FromCtx(ctx)
218 reader := bufio.NewReader(file)
219
220 for {
221 if ctx.Err() != nil {
222 return
223 }
224
225 line, err := reader.ReadString('\n')
226 if err != nil {
227 if err == io.EOF {
228 time.Sleep(100 * time.Millisecond)
229 continue
230 }
231 log.Warn().Err(err).Msg("error reading log file")
232 return
233 }
234
235 line = strings.TrimRight(line, "\n\r")
236 if !sendLogLineWithChannel(ctx, ch, line) {
237 return
238 }
239 }
240}
241
242func sendLogLineWithChannel(ctx context.Context, ch chan<- string, line string) bool {
243 select {
244 case ch <- line:
245 return true
246 case <-ctx.Done():
247 return false
248 }
249}
250
251func (s *Service) getProcessLogsFromJournal(ctx context.Context, lines int) ([]string, error) {
252 userScope, err := resolveJournalctlScope(ctx)
253 if err != nil {
254 return nil, err
255 }
256
257 args := []string{"-u", "gordon", "-n", strconv.Itoa(lines), "--no-pager", "-o", "cat"}
258 if userScope {
259 args = append([]string{"--user"}, args...)
260 }
261
262 out, err := execCommandContext(ctx, "journalctl", args...).Output()
263 if err != nil {
264 return nil, fmt.Errorf("failed to read journalctl logs: %w", err)
265 }
266
267 return outputLines(out), nil
268}
269
270func (s *Service) followProcessLogsFromJournal(ctx context.Context, initialLines int) (<-chan string, error) {
271 userScope, err := resolveJournalctlScope(ctx)
272 if err != nil {
273 return nil, err
274 }
275
276 args := []string{
277 "-u", "gordon",
278 "-n", strconv.Itoa(initialLines),
279 "--no-pager",
280 "-o", "cat",
281 "-f",
282 }
283 if userScope {
284 args = append([]string{"--user"}, args...)
285 }
286
287 cmd := execCommandContext(ctx, "journalctl", args...) // #nosec G204 -- internal args only
288 stdout, err := cmd.StdoutPipe()
289 if err != nil {
290 return nil, fmt.Errorf("failed to open journalctl stdout: %w", err)
291 }
292 cmd.Stderr = io.Discard
293
294 if err := cmd.Start(); err != nil {
295 return nil, fmt.Errorf("failed to start journalctl follow: %w", err)
296 }
297
298 ch := make(chan string, 100)
299
300 go func() {
301 defer close(ch)
302 defer stdout.Close()
303 defer cmd.Wait() //nolint:errcheck // best-effort cleanup
304
305 scanner := bufio.NewScanner(stdout)
306 scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
307
308 for scanner.Scan() {
309 select {
310 case ch <- scanner.Text():
311 case <-ctx.Done():
312 return
313 }
314 }
315 }()
316
317 return ch, nil
318}
319
320func resolveJournalctlScope(ctx context.Context) (bool, error) {
321 userLogs, userErr := hasJournalctlLogs(ctx, true)
322 if userErr == nil && userLogs {
323 return true, nil
324 }
325
326 systemLogs, systemErr := hasJournalctlLogs(ctx, false)
327 if systemErr == nil && systemLogs {
328 return false, nil
329 }
330
331 if userErr != nil && systemErr != nil {
332 return false, fmt.Errorf("journalctl unavailable for user and system services")
333 }
334
335 // Default to user scope when logs are currently empty; this matches rootless setups.
336 return true, nil
337}
338
339func hasJournalctlLogs(ctx context.Context, userService bool) (bool, error) {
340 args := []string{"-u", "gordon", "-n", "1", "--no-pager", "-q", "-o", "cat"}
341 if userService {
342 args = append([]string{"--user"}, args...)
343 }
344
345 out, err := execCommandContext(ctx, "journalctl", args...).Output()
346 if err != nil {
347 return false, err
348 }
349
350 return len(bytes.TrimSpace(out)) > 0, nil
351}
352
353func outputLines(out []byte) []string {
354 trimmed := strings.TrimRight(string(out), "\n\r")
355 if trimmed == "" {
356 return []string{}
357 }
358 return strings.Split(trimmed, "\n")
359}
360
361// validateDomain checks that domain is safe for use in logs and error messages.
362func validateDomain(domain string) error {
363 if domain == "" {
364 return fmt.Errorf("domain cannot be empty")
365 }
366 if len(domain) > 253 {
367 return fmt.Errorf("domain too long")
368 }
369 if strings.Contains(domain, "..") {
370 return fmt.Errorf("invalid domain")
371 }
372 if strings.ContainsAny(domain, "\x00\n\r") {
373 return fmt.Errorf("invalid domain")
374 }
375 return nil
376}
377
378// GetContainerLogs returns the last N lines of container logs for a domain.
379func (s *Service) GetContainerLogs(ctx context.Context, domain string, lines int) ([]string, error) {
380 if err := validateDomain(domain); err != nil {
381 return nil, err
382 }
383
384 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
385 zerowrap.FieldLayer: "usecase",
386 zerowrap.FieldUseCase: "GetContainerLogs",
387 "domain": domain,
388 "lines": lines,
389 })
390 log := zerowrap.FromCtx(ctx)
391
392 // Resolve the app/service to its ACTIVE container.
393 containerID, err := s.containerForService(ctx, domain)
394 if err != nil {
395 return nil, err
396 }
397
398 // Get logs from container runtime (non-follow mode)
399 reader, err := s.runtime.GetContainerLogs(ctx, containerID, false)
400 if err != nil {
401 return nil, log.WrapErr(err, "failed to get container logs")
402 }
403 defer reader.Close()
404
405 // Parse Docker log stream and get lines
406 logLines, err := parseDockerLogs(reader)
407 if err != nil {
408 return nil, log.WrapErr(err, "failed to parse container logs")
409 }
410
411 // Return last N lines
412 if len(logLines) > lines {
413 logLines = logLines[len(logLines)-lines:]
414 }
415
416 return logLines, nil
417}
418
419// FollowContainerLogs returns a channel that streams container log lines.
420func (s *Service) FollowContainerLogs(ctx context.Context, domain string, initialLines int) (<-chan string, error) {
421 if err := validateDomain(domain); err != nil {
422 return nil, err
423 }
424
425 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
426 zerowrap.FieldLayer: "usecase",
427 zerowrap.FieldUseCase: "FollowContainerLogs",
428 "domain": domain,
429 "initial_lines": initialLines,
430 })
431 log := zerowrap.FromCtx(ctx)
432
433 // Resolve the app/service to its ACTIVE container.
434 containerID, err := s.containerForService(ctx, domain)
435 if err != nil {
436 return nil, err
437 }
438
439 // Get logs from container runtime (follow mode)
440 reader, err := s.runtime.GetContainerLogs(ctx, containerID, true)
441 if err != nil {
442 return nil, log.WrapErr(err, "failed to get container logs")
443 }
444
445 ch := make(chan string, 100)
446
447 go func() {
448 defer close(ch)
449 defer reader.Close()
450
451 // Stream Docker logs
452 err := streamDockerLogs(ctx, reader, ch)
453 if err != nil && ctx.Err() == nil {
454 log.Warn().Err(err).Msg("error streaming container logs")
455 }
456 }()
457
458 return ch, nil
459}
460
461// maxTailLines bounds the ring buffer allocated by tailLines.
462const maxTailLines = 10000
463
464// tailLines reads the last N lines from a file using a ring buffer.
465func tailLines(file *os.File, n int) ([]string, error) {
466 if n <= 0 {
467 return []string{}, nil
468 }
469 if n > maxTailLines {
470 n = maxTailLines
471 }
472
473 // Seek to beginning
474 if _, err := file.Seek(0, io.SeekStart); err != nil {
475 return nil, err
476 }
477
478 scanner := bufio.NewScanner(file)
479 buffer := make([]string, n)
480 index := 0
481 total := 0
482
483 for scanner.Scan() {
484 buffer[index] = scanner.Text()
485 index = (index + 1) % n
486 total++
487 }
488
489 if err := scanner.Err(); err != nil {
490 return nil, err
491 }
492
493 if total == 0 {
494 return []string{}, nil
495 }
496
497 // Build result in correct order
498 var result []string
499 if total <= n {
500 result = make([]string, total)
501 for i := 0; i < total; i++ {
502 result[i] = buffer[i]
503 }
504 } else {
505 result = make([]string, n)
506 for i := 0; i < n; i++ {
507 result[i] = buffer[(index+i)%n]
508 }
509 }
510
511 return result, nil
512}
513
514// parseDockerLogs parses Docker multiplexed log stream and returns lines.
515// Docker logs have an 8-byte header: [stream type (1), padding (3), size (4)]
516func parseDockerLogs(reader io.Reader) ([]string, error) {
517 var lines []string
518 header := make([]byte, 8)
519
520 for {
521 // Read header
522 _, err := io.ReadFull(reader, header)
523 if err != nil {
524 if err == io.EOF {
525 break
526 }
527 return lines, nil // Return what we have
528 }
529
530 // Get payload size from header (big endian)
531 size := binary.BigEndian.Uint32(header[4:8])
532 if size == 0 {
533 continue
534 }
535
536 // Read payload
537 payload := make([]byte, size)
538 _, err = io.ReadFull(reader, payload)
539 if err != nil {
540 if err == io.EOF {
541 break
542 }
543 return lines, nil
544 }
545
546 // Split by newlines and add to result
547 content := strings.TrimRight(string(payload), "\n\r")
548 for _, line := range strings.Split(content, "\n") {
549 if line != "" {
550 lines = append(lines, line)
551 }
552 }
553 }
554
555 return lines, nil
556}
557
558// streamDockerLogs streams Docker multiplexed log stream to a channel.
559func streamDockerLogs(ctx context.Context, reader io.Reader, ch chan<- string) error {
560 header := make([]byte, 8)
561
562 for {
563 select {
564 case <-ctx.Done():
565 return ctx.Err()
566 default:
567 }
568
569 // Read header
570 _, err := io.ReadFull(reader, header)
571 if err != nil {
572 if err == io.EOF {
573 return nil
574 }
575 return err
576 }
577
578 // Get payload size
579 size := binary.BigEndian.Uint32(header[4:8])
580 if size == 0 {
581 continue
582 }
583
584 // Read payload
585 payload := make([]byte, size)
586 _, err = io.ReadFull(reader, payload)
587 if err != nil {
588 if err == io.EOF {
589 return nil
590 }
591 return err
592 }
593
594 // Split by newlines and send to channel
595 content := strings.TrimRight(string(payload), "\n\r")
596 for _, line := range strings.Split(content, "\n") {
597 if line != "" {
598 select {
599 case ch <- line:
600 case <-ctx.Done():
601 return ctx.Err()
602 }
603 }
604 }
605 }
606}