/
destination.go
284 lines (254 loc) · 8.21 KB
/
destination.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
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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
//go:build !no_logs
package http
import (
"bytes"
"context"
"errors"
"fmt"
"io/ioutil"
"log"
"net/http"
"net/url"
"sync"
"time"
logsconfig "github.com/F1997/categraf/config/logs"
"github.com/F1997/categraf/logs/client"
"github.com/F1997/categraf/pkg/backoff"
httputils "github.com/F1997/categraf/pkg/httpx"
)
// ContentType options,
const (
TextContentType = "text/plain"
JSONContentType = "application/json"
)
// HTTP errors.
var (
errClient = errors.New("client error")
errServer = errors.New("server error")
)
// emptyPayload is an empty payload used to check HTTP connectivity without sending logs.
var emptyPayload []byte
// Destination sends a payload over HTTP.
type Destination struct {
url string
apiKey string
contentType string
host string
contentEncoding ContentEncoding
client *httputils.ResetClient
destinationsContext *client.DestinationsContext
once sync.Once
payloadChan chan []byte
climit chan struct{} // semaphore for limiting concurrent background sends
backoff backoff.Policy
nbErrors int
blockedUntil time.Time
protocol logsconfig.IntakeProtocol
origin logsconfig.IntakeOrigin
}
// NewDestination returns a new Destination.
// If `maxConcurrentBackgroundSends` > 0, then at most that many background payloads will be sent concurrently, else
// there is no concurrency and the background sending pipeline will block while sending each payload.
// TODO: add support for SOCKS5
func NewDestination(endpoint logsconfig.Endpoint, contentType string, destinationsContext *client.DestinationsContext, maxConcurrentBackgroundSends int) *Destination {
return newDestination(endpoint, contentType, destinationsContext, time.Second*10, maxConcurrentBackgroundSends)
}
func newDestination(endpoint logsconfig.Endpoint, contentType string, destinationsContext *client.DestinationsContext, timeout time.Duration, maxConcurrentBackgroundSends int) *Destination {
if maxConcurrentBackgroundSends < 0 {
maxConcurrentBackgroundSends = 0
}
policy := backoff.NewPolicy(
endpoint.BackoffFactor,
endpoint.BackoffBase,
endpoint.BackoffMax,
endpoint.RecoveryInterval,
endpoint.RecoveryReset,
)
return &Destination{
host: endpoint.Host,
url: buildURL(endpoint),
apiKey: endpoint.APIKey,
contentType: contentType,
contentEncoding: buildContentEncoding(endpoint),
client: httputils.NewResetClient(endpoint.ConnectionResetInterval, httpClientFactory(timeout)),
destinationsContext: destinationsContext,
climit: make(chan struct{}, maxConcurrentBackgroundSends),
backoff: policy,
protocol: endpoint.Protocol,
origin: endpoint.Origin,
}
}
func errorToTag(err error) string {
if err == nil {
return "none"
} else if _, ok := err.(*client.RetryableError); ok {
return "retryable"
} else {
return "non-retryable"
}
}
// Send sends a payload over HTTP,
// the error returned can be retryable and it is the responsibility of the callee to retry.
func (d *Destination) Send(payload []byte) error {
if d.blockedUntil.After(time.Now()) {
// log.Printf("%s: sleeping until %v before retrying\n", d.url, d.blockedUntil)
d.waitForBackoff()
}
err := d.unconditionalSend(payload)
if _, ok := err.(*client.RetryableError); ok {
d.nbErrors = d.backoff.IncError(d.nbErrors)
} else {
d.nbErrors = d.backoff.DecError(d.nbErrors)
}
d.blockedUntil = time.Now().Add(d.backoff.GetBackoffDuration(d.nbErrors))
return err
}
func (d *Destination) unconditionalSend(payload []byte) (err error) {
ctx := d.destinationsContext.Context()
encodedPayload, err := d.contentEncoding.encode(payload)
if err != nil {
return err
}
req, err := http.NewRequest("POST", d.url, bytes.NewReader(encodedPayload))
if err != nil {
// the request could not be built,
// this can happen when the method or the url are valid.
return err
}
req.Header.Set("User-Agent", "categraf")
req.Header.Set("CATEGRAF-API-KEY", d.apiKey)
req.Header.Set("Content-Type", d.contentType)
req.Header.Set("Content-Encoding", d.contentEncoding.name())
if d.protocol != "" {
req.Header.Set("CATEGRAF-PROTOCOL", string(d.protocol))
}
if d.origin != "" {
req.Header.Set("CATEGRAF-ORIGIN", string(d.origin))
// TODO agentversion
req.Header.Set("CATEGRAF-ORIGIN-VERSION", "0.0.1")
}
req = req.WithContext(ctx)
resp, err := d.client.Do(req)
if err != nil {
if ctx.Err() == context.Canceled {
return ctx.Err()
}
// most likely a network or a connect error, the callee should retry.
return client.NewRetryableError(err)
}
defer resp.Body.Close()
response, err := ioutil.ReadAll(resp.Body)
if err != nil {
// the read failed because the server closed or terminated the connection
// *after* serving the request.
return err
}
if resp.StatusCode >= 400 {
log.Printf("W! failed to post http payload. code=%d host=%s response=%s\n", resp.StatusCode, d.host, string(response))
}
if resp.StatusCode == 429 || resp.StatusCode >= 500 {
// the server could not serve the request, most likely because of an
// internal error or, (429) because it is overwhelmed
return client.NewRetryableError(errServer)
} else if resp.StatusCode >= 400 {
// the logs-agent is likely to be misconfigured,
// the URL or the API key may be wrong.
return errClient
} else {
return nil
}
}
// SendAsync sends a payload in background.
func (d *Destination) SendAsync(payload []byte) {
d.once.Do(func() {
payloadChan := make(chan []byte, logsconfig.ChanSize)
d.sendInBackground(payloadChan)
d.payloadChan = payloadChan
})
d.payloadChan <- payload
}
// sendInBackground sends all payloads from payloadChan in background.
func (d *Destination) sendInBackground(payloadChan chan []byte) {
ctx := d.destinationsContext.Context()
go func() {
for {
select {
case payload := <-payloadChan:
// if the channel is non-buffered then there is no concurrency and we block on sending each payload
if cap(d.climit) == 0 {
d.unconditionalSend(payload) //nolint:errcheck
break
}
d.climit <- struct{}{}
go func() {
d.unconditionalSend(payload) //nolint:errcheck
<-d.climit
}()
case <-ctx.Done():
return
}
}
}()
}
func httpClientFactory(timeout time.Duration) func() *http.Client {
return func() *http.Client {
return &http.Client{
Timeout: timeout,
// reusing core agent HTTP transport to benefit from proxy settings.
Transport: httputils.CreateHTTPTransport(),
}
}
}
// buildURL buils a url from a logsconfig endpoint.
func buildURL(endpoint logsconfig.Endpoint) string {
var scheme string
if endpoint.UseSSL {
scheme = "https"
} else {
scheme = "http"
}
var address string
if endpoint.Port != 0 {
address = fmt.Sprintf("%v:%v", endpoint.Host, endpoint.Port)
} else {
address = endpoint.Host
}
url := url.URL{
Scheme: scheme,
Host: address,
}
if endpoint.Version == logsconfig.EPIntakeVersion2 && endpoint.TrackType != "" {
url.Path = fmt.Sprintf("/api/v2/%s", endpoint.TrackType)
} else {
url.Path = "/v1/input"
}
return url.String()
}
func buildContentEncoding(endpoint logsconfig.Endpoint) ContentEncoding {
if endpoint.UseCompression {
return NewGzipContentEncoding(endpoint.CompressionLevel)
}
return IdentityContentType
}
// CheckConnectivity check if sending logs through HTTP works
func CheckConnectivity(endpoint logsconfig.Endpoint) logsconfig.HTTPConnectivity {
log.Println("I! Checking HTTP connectivity...")
ctx := client.NewDestinationsContext()
ctx.Start()
defer ctx.Stop()
// Lower the timeout to 5s because HTTP connectivity test is done synchronously during the agent bootstrap sequence
destination := newDestination(endpoint, JSONContentType, ctx, time.Second*5, 0)
log.Println("I! Sending HTTP connectivity request to", destination.url)
err := destination.unconditionalSend(emptyPayload)
if err != nil {
log.Println("E! HTTP connectivity failure:", err)
} else {
log.Println("HTTP connectivity successful")
}
return err == nil
}
func (d *Destination) waitForBackoff() {
ctx, cancel := context.WithDeadline(d.destinationsContext.Context(), d.blockedUntil)
defer cancel()
<-ctx.Done()
}