beenull/crowdsec
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

try this

Browse source
This commit is contained in:
sabban 2023-07-06 17:29:48 +02:00
parent 10ed2b8680
commit 8f1ca97a88
2 changed files with 18 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
pkg/leakybucket/bucket.go
View file

@ -70,6 +70,7 @@ type Leaky struct {
wgPour *sync.WaitGroup wgPour *sync.WaitGroup
wgDumpState *sync.WaitGroup wgDumpState *sync.WaitGroup
mutex *sync.Mutex //used only for TIMEMACHINE mode to allow garbage collection without races mutex *sync.Mutex //used only for TIMEMACHINE mode to allow garbage collection without races
orderEvent bool
} }
var BucketsPour = prometheus.NewCounterVec( var BucketsPour = prometheus.NewCounterVec(

17
pkg/leakybucket/manager_run.go
View file

@ -6,6 +6,7 @@ import (
"fmt" "fmt"
"math" "math"
"os" "os"
"sync"
"time" "time"
"github.com/antonmedv/expr" "github.com/antonmedv/expr"
@ -279,6 +280,8 @@ func LoadOrStoreBucketFromHolder(partitionKey string, buckets *Buckets, holder B
return biface.(*Leaky), nil return biface.(*Leaky), nil
} }
var orderEvent map[string]*sync.WaitGroup
func PourItemToHolders(parsed types.Event, holders []BucketFactory, buckets *Buckets) (bool, error) { func PourItemToHolders(parsed types.Event, holders []BucketFactory, buckets *Buckets) (bool, error) {
var ( var (
ok, condition, poured bool ok, condition, poured bool
@ -344,7 +347,21 @@ func PourItemToHolders(parsed types.Event, holders []BucketFactory, buckets *Buc
return false, fmt.Errorf("failed to load or store bucket: %w", err) return false, fmt.Errorf("failed to load or store bucket: %w", err)
} }
//finally, pour the even into the bucket //finally, pour the even into the bucket
if bucket.orderEvent {
if orderEvent == nil {
orderEvent = make(map[string]*sync.WaitGroup)
}
orderEvent[buckey] = &sync.WaitGroup{}
orderEvent[buckey].Add(1)
}
ok, err := PourItemToBucket(bucket, holders[idx], buckets, &parsed) ok, err := PourItemToBucket(bucket, holders[idx], buckets, &parsed)
if bucket.orderEvent {
orderEvent[buckey].Wait()
}
if err != nil { if err != nil {
return false, fmt.Errorf("failed to pour bucket: %w", err) return false, fmt.Errorf("failed to pour bucket: %w", err)
} }