pkg/leakybucket: remove global bucketStore, unused parameters + tags (#4286)

Author: mmetc

cmd/crowdsec/crowdsec.go

@@ -82,7 +82,7 @@ func startParserRoutines(ctx context.Context, g *errgroup.Group, cConfig *csconf
 	}
 }
 
-func startBucketRoutines(ctx context.Context, g *errgroup.Group, cConfig *csconfig.Config, pourCollector *leakybucket.PourCollector) {
+func startBucketRoutines(ctx context.Context, g *errgroup.Group, cConfig *csconfig.Config, pourCollector *leakybucket.PourCollector, bucketStore *leakybucket.BucketStore) {
 	for idx := range cConfig.Crowdsec.BucketsRoutinesCount {
 		log.WithField("idx", idx).Info("Starting bucket routine")
 		g.Go(func() error {
@@ -98,7 +98,7 @@ func startHeartBeat(ctx context.Context, _ *csconfig.Config, apiClient *apiclien
 	apiClient.HeartBeat.StartHeartBeat(ctx)
 }
 
-func startOutputRoutines(ctx context.Context, cConfig *csconfig.Config, parsers *parser.Parsers, apiClient *apiclient.ApiClient, sd *StateDumper) {
+func startOutputRoutines(ctx context.Context, cConfig *csconfig.Config, parsers *parser.Parsers, apiClient *apiclient.ApiClient, sd *StateDumper, bucketStore *leakybucket.BucketStore) {
 	for idx := range cConfig.Crowdsec.OutputRoutinesCount {
 		log.WithField("idx", idx).Info("Starting output routine")
 		outputsTomb.Go(func() error {
@@ -144,12 +144,13 @@ func runCrowdsec(
 	hub *cwhub.Hub,
 	datasources []acquisitionTypes.DataSource,
 	sd *StateDumper,
+	bucketStore *leakybucket.BucketStore,
 ) error {
 	inEvents = make(chan pipeline.Event)
 	logLines = make(chan pipeline.Event)
 
 	startParserRoutines(ctx, g, cConfig, parsers, sd.StageParse)
-	startBucketRoutines(ctx, g, cConfig, sd.Pour)
+	startBucketRoutines(ctx, g, cConfig, sd.Pour, bucketStore)
 
 	apiClient, err := apiclient.GetLAPIClient()
 	if err != nil {
@@ -158,7 +159,7 @@ func runCrowdsec(
 
 	startHeartBeat(ctx, cConfig, apiClient)
 
-	startOutputRoutines(ctx, cConfig, parsers, apiClient, sd)
+	startOutputRoutines(ctx, cConfig, parsers, apiClient, sd, bucketStore)
 
 	if err := startLPMetrics(ctx, cConfig, apiClient, hub, datasources); err != nil {
 		return err
@@ -187,6 +188,8 @@ func serveCrowdsec(
 
 	var g errgroup.Group
 
+	bucketStore := leakybucket.NewBucketStore()
+
 	crowdsecTomb.Go(func() error {
 		defer trace.CatchPanic("crowdsec/serveCrowdsec")
 
@@ -197,7 +200,7 @@ func serveCrowdsec(
 
 			agentReady <- true
 
-			if err := runCrowdsec(cctx, &g, cConfig, parsers, hub, datasources, sd); err != nil {
+			if err := runCrowdsec(cctx, &g, cConfig, parsers, hub, datasources, sd, bucketStore); err != nil {
 				log.Fatalf("unable to start crowdsec routines: %s", err)
 			}
 		}()

cmd/crowdsec/main.go

@@ -43,7 +43,6 @@ var (
 	dataSources []acquisitionTypes.DataSource
 	// the state of the buckets
 	holders []leakybucket.BucketFactory
-	bucketStore *leakybucket.BucketStore
 
 	logLines   chan pipeline.Event
 	inEvents  chan pipeline.Event
@@ -54,13 +53,11 @@ var (
 func LoadBuckets(cConfig *csconfig.Config, hub *cwhub.Hub) error {
 	var err error
 
-	bucketStore = leakybucket.NewBucketStore()
-
 	scenarios := hub.GetInstalledByType(cwhub.SCENARIOS, false)
 
 	log.Infof("Loading %d scenario files", len(scenarios))
 
-	holders, outEvents, err = leakybucket.LoadBuckets(cConfig.Crowdsec, hub, scenarios, bucketStore, flags.OrderEvent)
+	holders, outEvents, err = leakybucket.LoadBuckets(cConfig.Crowdsec, hub, scenarios, flags.OrderEvent)
 	if err != nil {
 		return err
 	}

pkg/leakybucket/bucketstore_test.go

@@ -123,7 +123,7 @@ func testOneBucket(t *testing.T, hub *cwhub.Hub, dir string) {
 
 	cscfg := &csconfig.CrowdsecServiceCfg{}
 
-	holders, response, err := LoadBuckets(cscfg, hub, scenarios, bucketStore, false)
+	holders, response, err := LoadBuckets(cscfg, hub, scenarios, false)
 	require.NoError(t, err)
 
 	testFile(t, filepath.Join(dir, "test.json"), holders, response, bucketStore)

pkg/leakybucket/manager_load.go

@@ -58,18 +58,18 @@ type BucketSpec struct {
 type BucketFactory struct {
 	Spec BucketSpec
 
-	logger              *log.Entry                 // logger is bucket-specific logger (used by Debug as well)
-	BucketName          string                     `yaml:"-"`
-	Filename            string                     `yaml:"-"`
-	RunTimeFilter       *vm.Program                `json:"-"`
-	RunTimeGroupBy      *vm.Program                `json:"-"`
-	DataDir             string                     `yaml:"-"`
-	leakspeed           time.Duration              // internal representation of `Leakspeed`
-	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.)
+	logger              *log.Entry          // logger is bucket-specific logger (used by Debug as well)
+	BucketName          string
+	Filename            string
+	RunTimeFilter       *vm.Program         `json:"-"`
+	RunTimeGroupBy      *vm.Program         `json:"-"`
+	DataDir             string
+	leakspeed           time.Duration       // internal representation of `Leakspeed`
+	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
-	Simulated           bool `yaml:"simulated"` // Set to true if the scenario instantiating the bucket was in the exclusion list
+	Simulated           bool                // Set to true if the scenario instantiating the bucket was in the exclusion list
 	orderEvent          bool
 }
 
@@ -104,7 +104,6 @@ type SimulationChecker interface {
 func loadBucketFactoriesFromFile(
 	item *cwhub.Item,
 	hub *cwhub.Hub,
-	bucketStore *BucketStore,
 	response chan pipeline.Event,
 	orderEvent bool,
 	simcheck SimulationChecker,
@@ -186,7 +185,6 @@ func LoadBuckets(
 	cscfg *csconfig.CrowdsecServiceCfg,
 	hub *cwhub.Hub,
 	scenarios []*cwhub.Item,
-	bucketStore *BucketStore,
 	orderEvent bool,
 ) ([]BucketFactory, chan pipeline.Event, error) {
 	allFactories := []BucketFactory{}
@@ -195,7 +193,7 @@ func LoadBuckets(
 	for _, item := range scenarios {
 		log.Debugf("Loading '%s'", item.State.LocalPath)
 
-		factories, err := loadBucketFactoriesFromFile(item, hub, bucketStore, response, orderEvent, &cscfg.SimulationConfig)
+		factories, err := loadBucketFactoriesFromFile(item, hub, response, orderEvent, &cscfg.SimulationConfig)
 		if err != nil {
 			return nil, nil, err
 		}