1package docker
2
3import (
4 "bufio"
5 "context"
6 "fmt"
7 "io"
8 "strings"
9 "time"
10
11 cerrdefs "github.com/containerd/errdefs"
12 "github.com/moby/moby/api/pkg/stdcopy"
13 "github.com/moby/moby/client"
14
15 "github.com/bnema/gordon/internal/boundaries/out"
16 "github.com/bnema/gordon/internal/domain"
17)
18
19// maxLogLineBytes bounds one exported line; longer lines are truncated
20// so one runaway writer cannot exhaust memory.
21const maxLogLineBytes = 64 * 1024
22
23var _ out.ContainerLogStreamer = (*Runtime)(nil)
24
25// StreamContainerLogs implements out.ContainerLogStreamer. It follows
26// stdout and stderr with runtime timestamps, demultiplexes them, and
27// emits one line at a time until the stream ends or ctx is canceled.
28func (r *Runtime) StreamContainerLogs(ctx context.Context, containerID string, since time.Time, emit func(domain.ContainerLogLine)) error {
29 options := client.ContainerLogsOptions{
30 ShowStdout: true,
31 ShowStderr: true,
32 Follow: true,
33 Timestamps: true,
34 }
35 if !since.IsZero() {
36 options.Since = since.UTC().Format(time.RFC3339Nano)
37 }
38 logs, err := r.client.ContainerLogs(ctx, containerID, options)
39 if err != nil {
40 if cerrdefs.IsNotFound(err) {
41 return fmt.Errorf("%w: %s", domain.ErrContainerNotFound, containerID)
42 }
43 return fmt.Errorf("follow container logs: %w", err)
44 }
45 defer logs.Close()
46 // Closing the stream unblocks the demux copy on cancellation.
47 stop := context.AfterFunc(ctx, func() { _ = logs.Close() })
48 defer stop()
49
50 stdoutR, stdoutW := io.Pipe()
51 stderrR, stderrW := io.Pipe()
52 done := make(chan struct{}, 2)
53 lines := make(chan domain.ContainerLogLine)
54 go scanLogLines(stdoutR, domain.LogStreamStdout, lines, done)
55 go scanLogLines(stderrR, domain.LogStreamStderr, lines, done)
56
57 copyErr := make(chan error, 1)
58 go func() {
59 _, err := stdcopy.StdCopy(stdoutW, stderrW, logs)
60 _ = stdoutW.Close()
61 _ = stderrW.Close()
62 copyErr <- err
63 }()
64
65 for pending := 2; pending > 0; {
66 select {
67 case line := <-lines:
68 emit(line)
69 case <-done:
70 pending--
71 }
72 }
73 err = <-copyErr
74 if ctx.Err() != nil {
75 return ctx.Err()
76 }
77 if err != nil {
78 return fmt.Errorf("demultiplex container logs: %w", err)
79 }
80 return nil
81}
82
83// scanLogLines splits one demultiplexed stream into timestamped lines.
84// The reader is always drained so the demux copy never blocks.
85func scanLogLines(r *io.PipeReader, stream string, lines chan<- domain.ContainerLogLine, done chan<- struct{}) {
86 defer func() { done <- struct{}{} }()
87 reader := bufio.NewReaderSize(r, 4096)
88 for {
89 raw, err := readBoundedLine(reader)
90 if raw != "" {
91 lines <- parseTimestampedLine(raw, stream)
92 }
93 if err != nil {
94 _ = r.CloseWithError(err)
95 return
96 }
97 }
98}
99
100// readBoundedLine returns one line without its newline, truncated to
101// maxLogLineBytes; the rest of an oversized line is discarded.
102func readBoundedLine(reader *bufio.Reader) (string, error) {
103 var b strings.Builder
104 for {
105 chunk, err := reader.ReadSlice('\n')
106 if room := maxLogLineBytes - b.Len(); room > 0 {
107 if len(chunk) > room {
108 chunk = chunk[:room]
109 }
110 b.Write(chunk)
111 }
112 if err == bufio.ErrBufferFull {
113 continue
114 }
115 return strings.TrimRight(b.String(), "\r\n"), err
116 }
117}
118
119// parseTimestampedLine splits the RFC3339Nano prefix Docker adds when
120// Timestamps is set. Lines without a valid prefix keep the receive time.
121func parseTimestampedLine(raw, stream string) domain.ContainerLogLine {
122 raw = strings.ToValidUTF8(raw, "\uFFFD")
123 line := domain.ContainerLogLine{Time: time.Now().UTC(), Stream: stream, Body: raw}
124 prefix, body, ok := strings.Cut(raw, " ")
125 if !ok {
126 return line
127 }
128 if ts, err := time.Parse(time.RFC3339Nano, prefix); err == nil {
129 line.Time = ts
130 line.Body = body
131 }
132 return line
133}