-
Notifications
You must be signed in to change notification settings - Fork 430
/
bucket.go
401 lines (358 loc) · 12.4 KB
/
bucket.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
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
package leakybucket
import (
"fmt"
"sync"
"sync/atomic"
"time"
"github.com/davecgh/go-spew/spew"
"github.com/mohae/deepcopy"
"github.com/prometheus/client_golang/prometheus"
log "github.com/sirupsen/logrus"
"gopkg.in/tomb.v2"
"github.com/crowdsecurity/go-cs-lib/trace"
"github.com/crowdsecurity/crowdsec/pkg/time/rate"
"github.com/crowdsecurity/crowdsec/pkg/types"
)
// those constants are now defined in types/constants
// const (
// LIVE = iota
// TIMEMACHINE
// )
// Leaky represents one instance of a bucket
type Leaky struct {
Name string
Mode int //LIVE or TIMEMACHINE
//the limiter is what holds the proper "leaky aspect", it determines when/if we can pour objects
Limiter rate.RateLimiter `json:"-"`
SerializedState rate.Lstate
//Queue is used to hold the cache of objects in the bucket, it is used to know 'how many' objects we have in buffer.
Queue *types.Queue
//Leaky buckets are receiving message through a chan
In chan *types.Event `json:"-"`
//Leaky buckets are pushing their overflows through a chan
Out chan *types.Queue `json:"-"`
// shared for all buckets (the idea is to kill this afterward)
AllOut chan types.Event `json:"-"`
//max capacity (for burst)
Capacity int
//CacheRatio is the number of elements that should be kept in memory (compared to capacity)
CacheSize int
//the unique identifier of the bucket (a hash)
Mapkey string
// chan for signaling
Signal chan bool `json:"-"`
Suicide chan bool `json:"-"`
Reprocess bool
Simulated bool
Uuid string
First_ts time.Time
Last_ts time.Time
Ovflw_ts time.Time
Total_count int
Leakspeed time.Duration
BucketConfig *BucketFactory
Duration time.Duration
Pour func(*Leaky, types.Event) `json:"-"`
//Profiling when set to true enables profiling of bucket
Profiling bool
timedOverflow bool
conditionalOverflow bool
logger *log.Entry
scopeType types.ScopeType
hash string
scenarioVersion string
tomb *tomb.Tomb
wgPour *sync.WaitGroup
wgDumpState *sync.WaitGroup
mutex *sync.Mutex //used only for TIMEMACHINE mode to allow garbage collection without races
orderEvent bool
}
var BucketsPour = prometheus.NewCounterVec(
prometheus.CounterOpts{
Name: "cs_bucket_poured_total",
Help: "Total events were poured in bucket.",
},
[]string{"source", "type", "name"},
)
var BucketsOverflow = prometheus.NewCounterVec(
prometheus.CounterOpts{
Name: "cs_bucket_overflowed_total",
Help: "Total buckets overflowed.",
},
[]string{"name"},
)
var BucketsCanceled = prometheus.NewCounterVec(
prometheus.CounterOpts{
Name: "cs_bucket_canceled_total",
Help: "Total buckets canceled.",
},
[]string{"name"},
)
var BucketsUnderflow = prometheus.NewCounterVec(
prometheus.CounterOpts{
Name: "cs_bucket_underflowed_total",
Help: "Total buckets underflowed.",
},
[]string{"name"},
)
var BucketsInstantiation = prometheus.NewCounterVec(
prometheus.CounterOpts{
Name: "cs_bucket_created_total",
Help: "Total buckets were instantiated.",
},
[]string{"name"},
)
var BucketsCurrentCount = prometheus.NewGaugeVec(
prometheus.GaugeOpts{
Name: "cs_buckets",
Help: "Number of buckets that currently exist.",
},
[]string{"name"},
)
var LeakyRoutineCount int64
// Newleaky creates a new leaky bucket from a BucketFactory
// Events created by the bucket (overflow, bucket empty) are sent to a chan defined by BucketFactory
// The leaky bucket implementation is based on rate limiter (see https://godoc.org/golang.org/x/time/rate)
// There's a trick to have an event said when the bucket gets empty to allow its destruction
func NewLeaky(bucketFactory BucketFactory) *Leaky {
bucketFactory.logger.Tracef("Instantiating live bucket %s", bucketFactory.Name)
return FromFactory(bucketFactory)
}
func FromFactory(bucketFactory BucketFactory) *Leaky {
var limiter rate.RateLimiter
//golang rate limiter. It's mainly intended for http rate limiter
Qsize := bucketFactory.Capacity
if bucketFactory.CacheSize > 0 {
//cache is smaller than actual capacity
if bucketFactory.CacheSize <= bucketFactory.Capacity {
Qsize = bucketFactory.CacheSize
//bucket might be counter (infinite size), allow cache limitation
} else if bucketFactory.Capacity == -1 {
Qsize = bucketFactory.CacheSize
}
}
if bucketFactory.Capacity == -1 {
//In this case we allow all events to pass.
//maybe in the future we could avoid using a limiter
limiter = &rate.AlwaysFull{}
} else {
limiter = rate.NewLimiter(rate.Every(bucketFactory.leakspeed), bucketFactory.Capacity)
}
BucketsInstantiation.With(prometheus.Labels{"name": bucketFactory.Name}).Inc()
//create the leaky bucket per se
l := &Leaky{
Name: bucketFactory.Name,
Limiter: limiter,
Uuid: seed.Generate(),
Queue: types.NewQueue(Qsize),
CacheSize: bucketFactory.CacheSize,
Out: make(chan *types.Queue, 1),
Suicide: make(chan bool, 1),
AllOut: bucketFactory.ret,
Capacity: bucketFactory.Capacity,
Leakspeed: bucketFactory.leakspeed,
BucketConfig: &bucketFactory,
Pour: Pour,
Reprocess: bucketFactory.Reprocess,
Profiling: bucketFactory.Profiling,
Mode: types.LIVE,
scopeType: bucketFactory.ScopeType,
scenarioVersion: bucketFactory.ScenarioVersion,
hash: bucketFactory.hash,
Simulated: bucketFactory.Simulated,
tomb: bucketFactory.tomb,
wgPour: bucketFactory.wgPour,
wgDumpState: bucketFactory.wgDumpState,
mutex: &sync.Mutex{},
orderEvent: bucketFactory.orderEvent,
}
if l.BucketConfig.Capacity > 0 && l.BucketConfig.leakspeed != time.Duration(0) {
l.Duration = time.Duration(l.BucketConfig.Capacity+1) * l.BucketConfig.leakspeed
}
if l.BucketConfig.duration != time.Duration(0) {
l.Duration = l.BucketConfig.duration
l.timedOverflow = true
}
if l.BucketConfig.Type == "conditional" {
l.conditionalOverflow = true
l.Duration = l.BucketConfig.leakspeed
}
if l.BucketConfig.Type == "bayesian" {
l.Duration = l.BucketConfig.leakspeed
}
return l
}
/* for now mimic a leak routine */
//LeakRoutine us the life of a bucket. It dies when the bucket underflows or overflows
func LeakRoutine(leaky *Leaky) error {
var (
durationTickerChan = make(<-chan time.Time)
durationTicker *time.Ticker
firstEvent = true
)
defer trace.CatchPanic(fmt.Sprintf("crowdsec/LeakRoutine/%s", leaky.Name))
BucketsCurrentCount.With(prometheus.Labels{"name": leaky.Name}).Inc()
defer BucketsCurrentCount.With(prometheus.Labels{"name": leaky.Name}).Dec()
/*todo : we create a logger at runtime while we want leakroutine to be up asap, might not be a good idea*/
leaky.logger = leaky.BucketConfig.logger.WithFields(log.Fields{"partition": leaky.Mapkey, "bucket_id": leaky.Uuid})
//We copy the processors, as they are coming from the BucketFactory, and thus are shared between buckets
//If we don't copy, processors using local cache (such as Uniq) are subject to race conditions
//This can lead to creating buckets that will discard their first events, preventing the underflow ticker from being initialized
//and preventing them from being destroyed
processors := deepcopy.Copy(leaky.BucketConfig.processors).([]Processor)
leaky.Signal <- true
atomic.AddInt64(&LeakyRoutineCount, 1)
defer atomic.AddInt64(&LeakyRoutineCount, -1)
for _, f := range processors {
err := f.OnBucketInit(leaky.BucketConfig)
if err != nil {
leaky.logger.Errorf("Problem at bucket initializiation. Bail out %T : %v", f, err)
close(leaky.Signal)
return fmt.Errorf("Problem at bucket initializiation. Bail out %T : %v", f, err)
}
}
leaky.logger.Debugf("Leaky routine starting, lifetime : %s", leaky.Duration)
for {
select {
/*receiving an event*/
case msg := <-leaky.In:
/*the msg var use is confusing and is redeclared in a different type :/*/
for _, processor := range processors {
msg = processor.OnBucketPour(leaky.BucketConfig)(*msg, leaky)
// if &msg == nil we stop processing
if msg == nil {
if leaky.orderEvent {
orderEvent[leaky.Mapkey].Done()
}
goto End
}
}
if leaky.logger.Level >= log.TraceLevel {
leaky.logger.Tracef("Pour event: %s", spew.Sdump(msg))
}
BucketsPour.With(prometheus.Labels{"name": leaky.Name, "source": msg.Line.Src, "type": msg.Line.Module}).Inc()
leaky.Pour(leaky, *msg) // glue for now
for _, processor := range processors {
msg = processor.AfterBucketPour(leaky.BucketConfig)(*msg, leaky)
if msg == nil {
if leaky.orderEvent {
orderEvent[leaky.Mapkey].Done()
}
goto End
}
}
//Clear cache on behalf of pour
// if durationTicker isn't initialized, then we're pouring our first event
// reinitialize the durationTicker when it's not a counter bucket
if !leaky.timedOverflow || firstEvent {
if firstEvent {
durationTicker = time.NewTicker(leaky.Duration)
durationTickerChan = durationTicker.C
defer durationTicker.Stop()
} else {
durationTicker.Reset(leaky.Duration)
}
}
firstEvent = false
/*we overflowed*/
if leaky.orderEvent {
orderEvent[leaky.Mapkey].Done()
}
case ofw := <-leaky.Out:
leaky.overflow(ofw)
return nil
/*suiciiiide*/
case <-leaky.Suicide:
close(leaky.Signal)
BucketsCanceled.With(prometheus.Labels{"name": leaky.Name}).Inc()
leaky.logger.Debugf("Suicide triggered")
leaky.AllOut <- types.Event{Type: types.OVFLW, Overflow: types.RuntimeAlert{Mapkey: leaky.Mapkey}}
leaky.logger.Tracef("Returning from leaky routine.")
return nil
/*we underflow or reach bucket deadline (timers)*/
case <-durationTickerChan:
var (
alert types.RuntimeAlert
err error
)
leaky.Ovflw_ts = time.Now().UTC()
close(leaky.Signal)
ofw := leaky.Queue
alert = types.RuntimeAlert{Mapkey: leaky.Mapkey}
if leaky.timedOverflow {
BucketsOverflow.With(prometheus.Labels{"name": leaky.Name}).Inc()
alert, err = NewAlert(leaky, ofw)
if err != nil {
log.Errorf("%s", err)
}
for _, f := range leaky.BucketConfig.processors {
alert, ofw = f.OnBucketOverflow(leaky.BucketConfig)(leaky, alert, ofw)
if ofw == nil {
leaky.logger.Debugf("Overflow has been discarded (%T)", f)
break
}
}
leaky.logger.Infof("Timed Overflow")
} else {
leaky.logger.Debugf("bucket underflow, destroy")
BucketsUnderflow.With(prometheus.Labels{"name": leaky.Name}).Inc()
}
if leaky.logger.Level >= log.TraceLevel {
/*don't sdump if it's not going to be printed, it's expensive*/
leaky.logger.Tracef("Overflow event: %s", spew.Sdump(types.Event{Overflow: alert}))
}
leaky.AllOut <- types.Event{Overflow: alert, Type: types.OVFLW}
leaky.logger.Tracef("Returning from leaky routine.")
return nil
case <-leaky.tomb.Dying():
leaky.logger.Debugf("Bucket externally killed, return")
for len(leaky.Out) > 0 {
ofw := <-leaky.Out
leaky.overflow(ofw)
}
leaky.AllOut <- types.Event{Type: types.OVFLW, Overflow: types.RuntimeAlert{Mapkey: leaky.Mapkey}}
return nil
}
End:
}
}
func Pour(leaky *Leaky, msg types.Event) {
leaky.wgDumpState.Wait()
leaky.wgPour.Add(1)
defer leaky.wgPour.Done()
leaky.Total_count += 1
if leaky.First_ts.IsZero() {
leaky.First_ts = time.Now().UTC()
}
leaky.Last_ts = time.Now().UTC()
if leaky.Limiter.Allow() || leaky.conditionalOverflow {
leaky.Queue.Add(msg)
} else {
leaky.Ovflw_ts = time.Now().UTC()
leaky.logger.Debugf("Last event to be poured, bucket overflow.")
leaky.Queue.Add(msg)
leaky.Out <- leaky.Queue
}
}
func (leaky *Leaky) overflow(ofw *types.Queue) {
close(leaky.Signal)
alert, err := NewAlert(leaky, ofw)
if err != nil {
log.Errorf("%s", err)
}
leaky.logger.Tracef("Overflow hooks time : %v", leaky.BucketConfig.processors)
for _, f := range leaky.BucketConfig.processors {
alert, ofw = f.OnBucketOverflow(leaky.BucketConfig)(leaky, alert, ofw)
if ofw == nil {
leaky.logger.Debugf("Overflow has been discarded (%T)", f)
break
}
}
if leaky.logger.Level >= log.TraceLevel {
leaky.logger.Tracef("Overflow event: %s", spew.Sdump(alert))
}
mt, _ := leaky.Ovflw_ts.MarshalText()
leaky.logger.Tracef("overflow time : %s", mt)
BucketsOverflow.With(prometheus.Labels{"name": leaky.Name}).Inc()
leaky.AllOut <- types.Event{Overflow: alert, Type: types.OVFLW, MarshaledTime: string(mt)}
}