refact pkg/leakybucket: trim down redundant Leaky struct fields (#4290)

* rename Leaky.BucketConfig -> Leaky.Factory

* remove Leaky.Name

* remove Leaky.CacheSize

* remove Leaky.Capacity

* remove Leaky.Leakspeed

* remove Leaky.Reprocess

* remove Leaky.scopeType

* remove Leaky.scenarioVersion

* remove Leaky.orderEvent

* rename BucketFactory.hash -> BucketFactory.scenarioHash

* remove Leaky.hash

* comments

Author: mmetc

pkg/leakybucket/bayesian.go

@@ -100,7 +100,7 @@ func (e *BayesianEvent) bayesianUpdate(p *BayesianProcessor, msg pipeline.Event,
 	}
 
 	l.logger.Debugf("running condition expression: %s", e.rawCondition.ConditionalFilterName)
-	ret, err := exprhelpers.Run(e.conditionalFilterRuntime, map[string]any{"evt": &msg, "queue": l.Queue, "leaky": l}, l.logger, l.BucketConfig.Spec.Debug)
+	ret, err := exprhelpers.Run(e.conditionalFilterRuntime, map[string]any{"evt": &msg, "queue": l.Queue, "leaky": l}, l.logger, l.Factory.Spec.Debug)
 	if err != nil {
 		return fmt.Errorf("unable to run conditional filter: %w", err)
 	}

pkg/leakybucket/bucket.go

@@ -24,7 +24,6 @@ type pourGate interface {
 
 // 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:"-"`
@@ -37,35 +36,25 @@ type Leaky struct {
 	Out chan *pipeline.Queue `json:"-"`
 	// shared for all buckets (the idea is to kill this afterward)
 	AllOut chan pipeline.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
 	ready        chan struct{} // closed when LeakRoutine is ready
 	readyOnce    sync.Once     // use to prevent double close
 	done         chan struct{} // closed when LeakRoutine has stopped processing
 	doneOnce     sync.Once     // use to prevent double close
 	Suicide      chan bool `json:"-"`
-	Reprocess    bool
 	Uuid         string
 	First_ts     time.Time
 	Last_ts      time.Time
 	Ovflw_ts     time.Time
 	Total_count  int
-	Leakspeed    time.Duration
-	BucketConfig *BucketFactory
+	Factory      *BucketFactory
 	Duration     time.Duration
 	Pour         func(*Leaky, pourGate, pipeline.Event) `json:"-"`
 	timedOverflow       bool
 	conditionalOverflow bool
 	logger              *log.Entry
-	scopeType           ScopeType
-	hash                string
-	scenarioVersion     string
 	mutex               *sync.Mutex // used only for TIMEMACHINE mode to allow garbage collection without races
-	orderEvent          bool
 	cancel              context.CancelFunc
 }
 
@@ -99,25 +88,16 @@ func NewLeakyFromFactory(f *BucketFactory) *Leaky {
 
 	// create the leaky bucket per se
 	l := &Leaky{
-		Name:            f.Spec.Name,
 		Limiter:         limiter,
 		Uuid:            seed.Generate(),
 		Queue:           pipeline.NewQueue(Qsize),
-		CacheSize:       f.Spec.CacheSize,
 		Out:             make(chan *pipeline.Queue, 1),
 		Suicide:         make(chan bool, 1),
 		AllOut:          f.ret,
-		Capacity:        f.Spec.Capacity,
-		Leakspeed:       f.leakspeed,
-		BucketConfig:    f,
+		Factory:         f,
 		Pour:            Pour,
-		Reprocess:       f.Spec.Reprocess,
 		Mode:            pipeline.LIVE,
-		scopeType:       f.Spec.ScopeType,
-		scenarioVersion: f.Spec.ScenarioVersion,
-		hash:            f.hash,
 		mutex:           &sync.Mutex{},
-		orderEvent:      f.orderEvent,
 	}
 	if f.Spec.Capacity > 0 && f.leakspeed != time.Duration(0) {
 		l.Duration = time.Duration(f.Spec.Capacity+1) * f.leakspeed
@@ -155,24 +135,24 @@ func (l *Leaky) LeakRoutine(ctx context.Context, gate pourGate) {
 		}
 	}()
 
-	defer trace.CatchPanic(fmt.Sprintf("crowdsec/LeakRoutine/%s", l.Name))
+	defer trace.CatchPanic(fmt.Sprintf("crowdsec/LeakRoutine/%s", l.Factory.Spec.Name))
 
-	metrics.BucketsCurrentCount.With(prometheus.Labels{"name": l.Name}).Inc()
-	defer metrics.BucketsCurrentCount.With(prometheus.Labels{"name": l.Name}).Dec()
+	metrics.BucketsCurrentCount.With(prometheus.Labels{"name": l.Factory.Spec.Name}).Inc()
+	defer metrics.BucketsCurrentCount.With(prometheus.Labels{"name": l.Factory.Spec.Name}).Dec()
 
 	// TODO: we create a logger at runtime while we want leakroutine to be up asap, might not be a good idea
-	l.logger = l.BucketConfig.logger.WithFields(log.Fields{"partition": l.Mapkey, "bucket_id": l.Uuid})
+	l.logger = l.Factory.logger.WithFields(log.Fields{"partition": l.Mapkey, "bucket_id": l.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(l.BucketConfig.processors).([]Processor)
+	processors := deepcopy.Copy(l.Factory.processors).([]Processor)
 
 	l.markReady()
 
 	for _, f := range processors {
-		err := f.OnBucketInit(l.BucketConfig)
+		err := f.OnBucketInit(l.Factory)
 		if err != nil {
 			l.logger.Errorf("Problem at bucket initializiation. Bail out %T : %v", f, err)
 			return
@@ -186,10 +166,10 @@ func (l *Leaky) LeakRoutine(ctx context.Context, gate pourGate) {
 		case msg := <-l.In:
 			// the msg var use is confusing and is redeclared in a different type :/
 			for _, processor := range processors {
-				msg = processor.OnBucketPour(l.BucketConfig, *msg, l)
+				msg = processor.OnBucketPour(l.Factory, *msg, l)
 				// if &msg == nil we stop processing
 				if msg == nil {
-					if l.orderEvent {
+					if l.Factory.orderEvent {
 						orderEvent[l.Mapkey].Done()
 					}
 					goto End
@@ -198,14 +178,14 @@ func (l *Leaky) LeakRoutine(ctx context.Context, gate pourGate) {
 			if l.logger.Level >= log.TraceLevel {
 				l.logger.Tracef("Pour event: %s", spew.Sdump(msg))
 			}
-			metrics.BucketsPour.With(prometheus.Labels{"name": l.Name, "source": msg.Line.Src, "type": msg.Line.Module}).Inc()
+			metrics.BucketsPour.With(prometheus.Labels{"name": l.Factory.Spec.Name, "source": msg.Line.Src, "type": msg.Line.Module}).Inc()
 
 			l.Pour(l, gate, *msg) // glue for now
 
 			for _, processor := range processors {
-				msg = processor.AfterBucketPour(l.BucketConfig, *msg, l)
+				msg = processor.AfterBucketPour(l.Factory, *msg, l)
 				if msg == nil {
-					if l.orderEvent {
+					if l.Factory.orderEvent {
 						orderEvent[l.Mapkey].Done()
 					}
 					goto End
@@ -227,7 +207,7 @@ func (l *Leaky) LeakRoutine(ctx context.Context, gate pourGate) {
 			}
 			firstEvent = false
 			// we overflowed
-			if l.orderEvent {
+			if l.Factory.orderEvent {
 				orderEvent[l.Mapkey].Done()
 			}
 		case ofw := <-l.Out:
@@ -237,7 +217,7 @@ func (l *Leaky) LeakRoutine(ctx context.Context, gate pourGate) {
 		case <-l.Suicide:
 			// don't wait defer to close the channel, in case we are blocked before returning
 			l.markDone()
-			metrics.BucketsCanceled.With(prometheus.Labels{"name": l.Name}).Inc()
+			metrics.BucketsCanceled.With(prometheus.Labels{"name": l.Factory.Spec.Name}).Inc()
 			l.logger.Debugf("Suicide triggered")
 			l.AllOut <- pipeline.Event{Type: pipeline.OVFLW, Overflow: pipeline.RuntimeAlert{Mapkey: l.Mapkey}}
 			l.logger.Tracef("Returning from leaky routine.")
@@ -254,14 +234,14 @@ func (l *Leaky) LeakRoutine(ctx context.Context, gate pourGate) {
 			alert = pipeline.RuntimeAlert{Mapkey: l.Mapkey}
 
 			if l.timedOverflow {
-				metrics.BucketsOverflow.With(prometheus.Labels{"name": l.Name}).Inc()
+				metrics.BucketsOverflow.With(prometheus.Labels{"name": l.Factory.Spec.Name}).Inc()
 
 				alert, err = NewAlert(l, ofw)
 				if err != nil {
 					log.Error(err)
 				}
-				for _, f := range l.BucketConfig.processors {
-					alert, ofw = f.OnBucketOverflow(l.BucketConfig, l, alert, ofw)
+				for _, f := range l.Factory.processors {
+					alert, ofw = f.OnBucketOverflow(l.Factory, l, alert, ofw)
 					if ofw == nil {
 						l.logger.Debugf("Overflow has been discarded (%T)", f)
 						break
@@ -270,7 +250,7 @@ func (l *Leaky) LeakRoutine(ctx context.Context, gate pourGate) {
 				l.logger.Infof("Timed Overflow")
 			} else {
 				l.logger.Debugf("bucket underflow, destroy")
-				metrics.BucketsUnderflow.With(prometheus.Labels{"name": l.Name}).Inc()
+				metrics.BucketsUnderflow.With(prometheus.Labels{"name": l.Factory.Spec.Name}).Inc()
 			}
 			if l.logger.Level >= log.TraceLevel {
 				// don't sdump if it's not going to be printed, it's expensive
@@ -320,9 +300,9 @@ func (l *Leaky) overflow(ofw *pipeline.Queue) {
 	if err != nil {
 		log.Errorf("%s", err)
 	}
-	l.logger.Tracef("Overflow hooks time : %v", l.BucketConfig.processors)
-	for _, f := range l.BucketConfig.processors {
-		alert, ofw = f.OnBucketOverflow(l.BucketConfig, l, alert, ofw)
+	l.logger.Tracef("Overflow hooks time : %v", l.Factory.processors)
+	for _, f := range l.Factory.processors {
+		alert, ofw = f.OnBucketOverflow(l.Factory, l, alert, ofw)
 		if ofw == nil {
 			l.logger.Debugf("Overflow has been discarded (%T)", f)
 			break
@@ -334,7 +314,7 @@ func (l *Leaky) overflow(ofw *pipeline.Queue) {
 	mt, _ := l.Ovflw_ts.MarshalText()
 	l.logger.Tracef("overflow time : %s", mt)
 
-	metrics.BucketsOverflow.With(prometheus.Labels{"name": l.Name}).Inc()
+	metrics.BucketsOverflow.With(prometheus.Labels{"name": l.Factory.Spec.Name}).Inc()
 
 	l.AllOut <- pipeline.Event{Overflow: alert, Type: pipeline.OVFLW, MarshaledTime: string(mt)}
 }

pkg/leakybucket/manager_load.go

@@ -25,6 +25,7 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
+// BucketSpec is the declarative YAML config for a scenario bucket. It holds all the possible user-provided fields.
 type BucketSpec struct {
 	FormatVersion       string                     `yaml:"format"`
 	Description         string                     `yaml:"description"`
@@ -53,8 +54,7 @@ type BucketSpec struct {
 	ScenarioVersion     string                     `yaml:"version,omitempty"`
 }
 
-// BucketFactory struct holds all fields for any bucket configuration. This is to have a
-// generic struct for buckets. This can be seen as a bucket factory.
+// BucketFactory is the compiled/validated, reusable template produced from a BucketSpec.
 type BucketFactory struct {
 	Spec BucketSpec
 
@@ -68,7 +68,7 @@ type BucketFactory struct {
 	duration            time.Duration       // internal representation of `Duration`
 	ret                 chan pipeline.Event // the bucket-specific output chan for overflows
 	processors          []Processor         // processors is the list of hooks for pour/overflow/create (cf. uniq, blackhole etc.)
-	hash                string
+	scenarioHash        string
 	Simulated           bool                // Set to true if the scenario instantiating the bucket was in the exclusion list
 	orderEvent          bool
 }
@@ -166,7 +166,7 @@ func loadBucketFactoriesFromFile(
 		f.Simulated = simcheck.IsSimulated(f.Spec.Name)
 
 		f.Spec.ScenarioVersion = item.State.LocalVersion
-		f.hash = item.State.LocalHash
+		f.scenarioHash = item.State.LocalHash
 
 		err = f.LoadBucket()
 		if err != nil {
@@ -353,8 +353,6 @@ func (f *BucketFactory) initDataFiles() {
 
 // LoadBucket validates and prepares a BucketFactory for runtime use (compile expressions, init processors, init data).
 func (f *BucketFactory) LoadBucket() error {
-	var err error
-
 	f.logger = bucketLogger(f)
 	f.logger.Infof("Adding %s bucket", f.Spec.Type)
 

pkg/leakybucket/manager_run.go

@@ -41,11 +41,11 @@ func GarbageCollectBuckets(deadline time.Time, bucketStore *BucketStore) {
 		const eps = 1e-9
 
 		tokat := val.Limiter.GetTokensCountAt(deadline)
-		tokcapa := float64(val.Capacity)
+		tokcapa := float64(val.Factory.Spec.Capacity)
 
 		// bucket actually underflowed based on log time, but no in real time
 		if tokat+eps >= tokcapa {
-			metrics.BucketsUnderflow.With(prometheus.Labels{"name": val.Name}).Inc()
+			metrics.BucketsUnderflow.With(prometheus.Labels{"name": val.Factory.Spec.Name}).Inc()
 			val.logger.Debugf("UNDERFLOW : first_ts:%s tokens_at:%f capcity:%f", val.First_ts, tokat, tokcapa)
 			toflush = append(toflush, key)
 			val.cancel()
@@ -135,7 +135,7 @@ func PourItemToBucket(
 			// holder.logger.Tracef("Successfully sent !")
 			if collector != nil {
 				evt := deepcopy.Copy(*parsed).(pipeline.Event)
-				collector.Add(bucket.Name, evt)
+				collector.Add(bucket.Factory.Spec.Name, evt)
 			}
 			holder.logger.Debugf("bucket '%s' is poured", holder.Spec.Name)
 			return nil
@@ -258,7 +258,7 @@ func PourItemToHolders(
 		}
 		// finally, pour the even into the bucket
 
-		if bucket.orderEvent {
+		if bucket.Factory.orderEvent {
 			if orderEvent == nil {
 				orderEvent = make(map[string]*sync.WaitGroup)
 			}
@@ -273,7 +273,7 @@ func PourItemToHolders(
 
 		err = PourItemToBucket(ctx, bucket, &holders[idx], buckets, &parsed, collector)
 
-		if bucket.orderEvent {
+		if bucket.Factory.orderEvent {
 			orderEvent[buckey].Wait()
 		}