Repository navigation
Expand file tree
/
Copy pathconfig.go
More file actions
305 lines (251 loc) · 9.53 KB
/
Copy pathconfig.go
File metadata and controls
305 lines (251 loc) · 9.53 KB
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
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
package analytics
import (
"math"
"math/rand"
"net/http"
"sync"
"time"
"github.com/google/uuid"
)
// Instances of this type carry the different configuration options that may
// be set when instantiating a client.
//
// Each field's zero-value is either meaningful or interpreted as using the
// default value defined by the library.
type Config struct {
// The endpoint to which the client connect and send their messages, set to
// `DefaultEndpoint` by default.
Endpoint string
// The flushing interval of the client. Messages will be sent when they've
// been queued up to the maximum batch size or when the flushing interval
// timer triggers.
Interval time.Duration
// The HTTP transport used by the client, this allows an application to
// redefine how requests are being sent at the HTTP level (for example,
// to change the connection pooling policy).
// If none is specified the client uses `http.DefaultTransport`.
Transport http.RoundTripper
// The logger used by the client to output info or error messages when that
// are generated by background operations.
// If none is specified the client uses a standard logger that outputs to
// `os.Stderr`.
Logger Logger
// The callback object that will be used by the client to notify the
// application when messages sends to the backend API succeeded or failed.
Callback Callback
// The maximum number of messages that will be sent in one API call.
// Messages will be sent when they've been queued up to the maximum batch
// size or when the flushing interval timer triggers.
// Note that the API will still enforce a 500KB limit on each HTTP request
// which is independent from the number of embedded messages.
BatchSize int
// When set to true the client will send more frequent and detailed messages
// to its logger.
Verbose bool
// The default context set on each message sent by the client.
DefaultContext *Context
// The retry policy used by the client to resend requests that have failed.
// The function is called with how many times the operation has been retried
// and is expected to return how long the client should wait before trying
// again.
// If not set the client will fallback to use a default retry policy.
RetryAfter func(int) time.Duration
// Maximum number of counted backoff retries. Zero means use
// DefaultMaxRetries, per the zero-value convention above; there is no way to
// ask for no retries at all. Negative values are rejected.
MaxRetries int
// Wall-clock cap on total time spent in backoff retries. Defaults to DefaultMaxTotalBackoffDuration.
MaxTotalBackoffDuration time.Duration
// Wall-clock cap on total time spent retrying after 429 Retry-After responses. Defaults to DefaultMaxRateLimitDuration.
MaxRateLimitDuration time.Duration
// ShutdownTimeout bounds how long Close waits for in-flight retries before
// dropping their batches. It covers the whole remaining retry schedule,
// including the final request, which is issued with this as its deadline.
// Mirrors analytics-java's NETWORK_TERMINATION_TIMEOUT_S.
ShutdownTimeout time.Duration
// A function called by the client to generate unique message identifiers.
// The client uses a UUID generator if none is provided.
// This field is not exported and only exposed internally to let unit tests
// mock the id generation.
uid func() string
// A function called by the client to get the current time, `time.Now` is
// used by default.
// This field is not exported and only exposed internally to let unit tests
// mock the current time.
now func() time.Time
// The maximum number of goroutines that will be spawned by a client to send
// requests to the backend API.
// This field is not exported and only exposed internally to let unit tests
// mock the current time.
maxConcurrentRequests int
}
// This constant sets the default endpoint to which client instances send
// messages if none was explictly set.
const DefaultEndpoint = "https://api.segment.io"
// This constant sets the default flush interval used by client instances if
// none was explicitly set.
// DefaultShutdownTimeout matches analytics-java's 75s network-executor
// termination timeout.
const DefaultShutdownTimeout = 75 * time.Second
const DefaultInterval = 5 * time.Second
// This constant sets the default batch size used by client instances if none
// was explicitly set.
const DefaultBatchSize = 250
// DefaultMaxRetries is the default number of counted backoff retries.
const DefaultMaxRetries = 10
// DefaultMaxTotalBackoffDuration is the default wall-clock cap on total backoff time.
const DefaultMaxTotalBackoffDuration = 12 * time.Hour
// DefaultMaxRateLimitDuration is the default wall-clock cap on Retry-After retries.
//
// Rate-limited attempts are deliberately uncounted, so this duration is the only
// thing bounding them. It is deliberately several times maxRetryAfterSeconds: when
// the two are equal a single maximal Retry-After consumes the whole budget, leaving
// one attempt and no retry at all.
const DefaultMaxRateLimitDuration = 30 * time.Minute
// maxRetryAfterSeconds is the cap applied to Retry-After header values. A guard
// against an absurd header, not a second budget: waiting less than the server asked
// for does not make the next attempt more likely to succeed, it just sends more
// requests at something already rate-limiting us. How long we keep trying is
// MaxRateLimitDuration's job.
const maxRetryAfterSeconds = int64(300)
// Verifies that fields that don't have zero-values are set to valid values,
// returns an error describing the problem if a field was invalid.
func (c *Config) validate() error {
if c.Interval < 0 {
return ConfigError{
Reason: "negative time intervals are not supported",
Field: "Interval",
Value: c.Interval,
}
}
if c.BatchSize < 0 {
return ConfigError{
Reason: "negative batch sizes are not supported",
Field: "BatchSize",
Value: c.BatchSize,
}
}
// Zero means "use the default" for these, per the zero-value convention above.
// Negatives used to survive into the retry loop, where they dropped every batch
// after its first failure instead of failing here.
if c.MaxRetries < 0 {
return ConfigError{
Reason: "negative retry counts are not supported",
Field: "MaxRetries",
Value: c.MaxRetries,
}
}
if c.MaxTotalBackoffDuration < 0 {
return ConfigError{
Reason: "negative backoff durations are not supported",
Field: "MaxTotalBackoffDuration",
Value: c.MaxTotalBackoffDuration,
}
}
if c.MaxRateLimitDuration < 0 {
return ConfigError{
Reason: "negative rate limit durations are not supported",
Field: "MaxRateLimitDuration",
Value: c.MaxRateLimitDuration,
}
}
if c.ShutdownTimeout < 0 {
return ConfigError{
Reason: "negative shutdown timeouts are not supported",
Field: "ShutdownTimeout",
Value: c.ShutdownTimeout,
}
}
return nil
}
// Given a config object as argument the function will set all zero-values to
// their defaults and return the modified object.
func makeConfig(c Config) Config {
if len(c.Endpoint) == 0 {
c.Endpoint = DefaultEndpoint
}
if c.ShutdownTimeout == 0 {
c.ShutdownTimeout = DefaultShutdownTimeout
}
if c.Interval == 0 {
c.Interval = DefaultInterval
}
if c.Transport == nil {
c.Transport = http.DefaultTransport
}
if c.Logger == nil {
c.Logger = newDefaultLogger()
}
if c.BatchSize == 0 {
c.BatchSize = DefaultBatchSize
}
if c.DefaultContext == nil {
c.DefaultContext = &Context{}
}
if c.RetryAfter == nil {
c.RetryAfter = defaultRetryAfter
}
if c.MaxRetries == 0 {
c.MaxRetries = DefaultMaxRetries
}
if c.MaxTotalBackoffDuration == 0 {
c.MaxTotalBackoffDuration = DefaultMaxTotalBackoffDuration
}
if c.MaxRateLimitDuration == 0 {
c.MaxRateLimitDuration = DefaultMaxRateLimitDuration
}
if c.uid == nil {
c.uid = uid
}
if c.now == nil {
c.now = time.Now
}
if c.maxConcurrentRequests == 0 {
c.maxConcurrentRequests = 1000
}
// We always overwrite the 'library' field of the default context set on the
// client because we want this information to be accurate.
c.DefaultContext.Library = LibraryInfo{
Name: "analytics-go",
Version: Version,
}
return c
}
// This function returns a string representation of a UUID, it's the default
// function used for generating unique IDs.
func uid() string {
return uuid.NewString()
}
// Jitter needs its own source: math/rand's global source is seeded
// deterministically before Go 1.20, so every process would draw the same
// sequence and stay in step with every other process anyway.
var (
jitterMu sync.Mutex
jitterRand = rand.New(rand.NewSource(time.Now().UnixNano()))
)
func jitterFraction() float64 {
jitterMu.Lock()
defer jitterMu.Unlock()
return jitterRand.Float64()
}
// defaultRetryAfter returns how long to wait before counted backoff attempt n:
// 500ms doubling to a 60s ceiling, then reduced by up to 50% at random.
//
// The jitter is applied after the clamp and only ever subtracts, so the ceiling
// holds and clients that started backing off together spread out instead of
// retrying in lockstep. Applying it before the clamp — which is what
// backo-go does, and why it is no longer used here — collapses back to exactly
// the cap once the exponential passes it, putting the whole fleet back in step
// at the point the endpoint can least afford it.
func defaultRetryAfter(attempt int) time.Duration {
const (
base = float64(500 * time.Millisecond)
ceiling = float64(60 * time.Second)
jitter = 0.5
)
delay := base * math.Pow(2, float64(attempt))
if delay > ceiling {
delay = ceiling
}
return time.Duration(delay - jitterFraction()*delay*jitter)
}