/
task_logs.go
77 lines (66 loc) · 1.67 KB
/
task_logs.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
package producers
import (
"context"
"fmt"
"io"
"time"
"github.com/replicatedcom/support-bundle/pkg/collect/plugins/docker/util"
"github.com/replicatedcom/support-bundle/pkg/collect/types"
dockertypes "github.com/docker/docker/api/types"
)
const (
LogsReaderIdleTimeout = 20 * time.Second
)
func (d *Docker) TaskLogs(taskID string, opts *dockertypes.ContainerLogsOptions) types.StreamsProducer {
return func(ctx context.Context) (map[string]io.Reader, error) {
if opts == nil {
opts = &dockertypes.ContainerLogsOptions{}
}
if !opts.ShowStdout && !opts.ShowStderr {
opts.ShowStdout = true
opts.ShowStderr = true
}
opts.Timestamps = true
opts.Follow = false
reader, err := d.client.TaskLogs(ctx, taskID, *opts)
if err != nil {
return nil, err
}
readerWithTimeout := logsReaderWithTimeout(reader, LogsReaderIdleTimeout)
return util.DemuxLogs(ctx, readerWithTimeout, taskID)
}
}
func logsReaderWithTimeout(r io.ReadCloser, timeout time.Duration) io.Reader {
pr, pw := io.Pipe()
go func() {
defer r.Close()
size := 32 * 1024
buf := make([]byte, size)
for {
var n int
var err error
done := make(chan struct{})
go func() {
n, err = r.Read(buf)
close(done)
}()
select {
case <-done:
if err != nil {
pw.CloseWithError(err) // this will handle the EOF
return
}
if _, err := pw.Write(buf[:n]); err != nil {
pw.CloseWithError(err)
return
}
case <-time.After(timeout):
// Fix issue https://github.com/moby/moby/issues/38640
// Service and task logs hang for large logs
pw.CloseWithError(fmt.Errorf("reader timeout after %s", timeout))
return
}
}
}()
return pr
}