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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
|
package main
// Cloudflare does not forward container stdout to observability, so syncd
// ships its own output and git-forge's to the worker. Drops rather than
// blocks, and never logs through log itself.
import (
"bytes"
"fmt"
"net/http"
"os"
"strings"
"sync/atomic"
"time"
)
const (
shipQueue = 256
shipEvery = 500 * time.Millisecond
shipTimeout = 3 * time.Second
// The bound on getting run's attention; the flush itself is waited out,
// since those lines say how the container stopped.
shipWait = 5 * time.Second
)
type shipper struct {
url string
lines chan string
flush chan chan struct{}
dropped atomic.Int64 // never queued, because the queue was full
unsent atomic.Int64 // queued, but the send failed
}
// newShipper tees to stdout and to the worker, or to stdout alone with no
// endpoint.
func newShipper(base string) *shipper {
s := &shipper{
url: base,
lines: make(chan string, shipQueue),
flush: make(chan chan struct{}),
}
if base != "" {
go s.run()
}
return s
}
// close sends what is queued and waits: the last lines say how the container
// shut down.
func (s *shipper) close() {
if s.url == "" {
return
}
ack := make(chan struct{})
select {
case s.flush <- ack:
<-ack
case <-time.After(shipWait):
}
}
// Write always reports success: the caller is a logger.
func (s *shipper) Write(p []byte) (int, error) {
_, _ = os.Stdout.Write(p)
if s.url == "" {
return len(p), nil
}
for _, line := range strings.Split(string(bytes.TrimRight(p, "\n")), "\n") {
if line == "" {
continue
}
select {
case s.lines <- line:
default:
// Dropped rather than stalling a request; counted, so the gap is
// reported.
s.dropped.Add(1)
}
}
return len(p), nil
}
func (s *shipper) run() {
client := &http.Client{Timeout: shipTimeout}
t := time.NewTicker(shipEvery)
defer t.Stop()
var batch []string
send := func() {
lines := int64(len(batch))
dropped, unsent := s.dropped.Swap(0), s.unsent.Swap(0)
if dropped > 0 {
batch = append(batch, fmt.Sprintf("syncd: dropped %d log lines, the queue was full", dropped))
}
if unsent > 0 {
batch = append(batch, fmt.Sprintf("syncd: %d log lines did not reach the worker", unsent))
}
if len(batch) == 0 {
return
}
body := strings.Join(batch, "\n")
batch = batch[:0]
// The status too: a 500 loses the batch as a refused connection does.
landed := false
if resp, err := client.Post(s.url, "text/plain", strings.NewReader(body)); err == nil {
landed = resp.StatusCode/100 == 2
resp.Body.Close()
}
if !landed {
// Put back, so the next send that works reports one total.
s.dropped.Add(dropped)
s.unsent.Add(unsent + lines)
}
}
for {
select {
case line := <-s.lines:
batch = append(batch, line)
if len(batch) >= shipQueue/2 {
send()
}
case <-t.C:
send()
case ack := <-s.flush:
for drained := false; !drained; {
select {
case line := <-s.lines:
batch = append(batch, line)
default:
drained = true
}
}
send()
close(ack)
}
}
}
|