pkg/types -> new imports pt 2 (#4013)

* pkg/parser

* pkg/leakybucket

* lint

Author: mmetc

.golangci.yml

@@ -285,7 +285,7 @@ linters:
         - name: line-length-limit
           arguments:
             # lower this after refactoring
-            - 216
+            - 219
         - name: max-control-nesting
           arguments:
             # lower this after refactoring

pkg/leakybucket/bayesian.go

@@ -7,7 +7,7 @@ import (
 	"github.com/expr-lang/expr/vm"
 
 	"github.com/crowdsecurity/crowdsec/pkg/exprhelpers"
-	"github.com/crowdsecurity/crowdsec/pkg/types"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
 type RawBayesianCondition struct {
@@ -60,8 +60,8 @@ func (c *BayesianBucket) OnBucketInit(g *BucketFactory) error {
 	return err
 }
 
-func (c *BayesianBucket) AfterBucketPour(_ *BucketFactory) func(types.Event, *Leaky) *types.Event {
-	return func(msg types.Event, l *Leaky) *types.Event {
+func (c *BayesianBucket) AfterBucketPour(_ *BucketFactory) func(pipeline.Event, *Leaky) *pipeline.Event {
+	return func(msg pipeline.Event, l *Leaky) *pipeline.Event {
 		c.posterior = c.prior
 		l.logger.Debugf("starting bayesian evaluation with prior: %v", c.posterior)
 
@@ -85,7 +85,7 @@ func (c *BayesianBucket) AfterBucketPour(_ *BucketFactory) func(types.Event, *Le
 	}
 }
 
-func (b *BayesianEvent) bayesianUpdate(c *BayesianBucket, msg types.Event, l *Leaky) error {
+func (b *BayesianEvent) bayesianUpdate(c *BayesianBucket, msg pipeline.Event, l *Leaky) error {
 	var condition, ok bool
 
 	if b.conditionalFilterRuntime == nil {
@@ -148,7 +148,7 @@ func (b *BayesianEvent) compileCondition() (*vm.Program, error) {
 	}
 
 	// don't hold lock during compile
-	compiled, err := expr.Compile(name, exprhelpers.GetExprOptions(map[string]any{"queue": &types.Queue{}, "leaky": &Leaky{}, "evt": &types.Event{}})...)
+	compiled, err := expr.Compile(name, exprhelpers.GetExprOptions(map[string]any{"queue": &pipeline.Queue{}, "leaky": &Leaky{}, "evt": &pipeline.Event{}})...)
 	if err != nil {
 		return nil, fmt.Errorf("bayesian condition compile error: %w", err)
 	}

pkg/leakybucket/blackhole.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/crowdsecurity/crowdsec/pkg/types"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
 type HiddenKey struct {
@@ -30,8 +30,8 @@ func NewBlackhole(bucketFactory *BucketFactory) (*Blackhole, error) {
 	}, nil
 }
 
-func (bl *Blackhole) OnBucketOverflow(bucketFactory *BucketFactory) func(*Leaky, types.RuntimeAlert, *types.Queue) (types.RuntimeAlert, *types.Queue) {
-	return func(leaky *Leaky, alert types.RuntimeAlert, queue *types.Queue) (types.RuntimeAlert, *types.Queue) {
+func (bl *Blackhole) OnBucketOverflow(bucketFactory *BucketFactory) func(*Leaky, pipeline.RuntimeAlert, *pipeline.Queue) (pipeline.RuntimeAlert, *pipeline.Queue) {
+	return func(leaky *Leaky, alert pipeline.RuntimeAlert, queue *pipeline.Queue) (pipeline.RuntimeAlert, *pipeline.Queue) {
 		var blackholed = false
 		var tmp []HiddenKey
 		// search if we are blackholed and refresh the slice
@@ -54,7 +54,7 @@ func (bl *Blackhole) OnBucketOverflow(bucketFactory *BucketFactory) func(*Leaky,
 
 		if blackholed {
 			leaky.logger.Tracef("Event is blackholed (%s)", leaky.First_ts)
-			return types.RuntimeAlert{
+			return pipeline.RuntimeAlert{
 				Mapkey: leaky.Mapkey,
 			}, nil
 		}

pkg/leakybucket/bucket.go

@@ -16,7 +16,7 @@ import (
 	"github.com/crowdsecurity/go-cs-lib/trace"
 
 	"github.com/crowdsecurity/crowdsec/pkg/metrics"
-	"github.com/crowdsecurity/crowdsec/pkg/types"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
 // Leaky represents one instance of a bucket
@@ -27,13 +27,13 @@ type Leaky struct {
 	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
+	Queue *pipeline.Queue
 	// Leaky buckets are receiving message through a chan
-	In chan *types.Event `json:"-"`
+	In chan *pipeline.Event `json:"-"`
 	// Leaky buckets are pushing their overflows through a chan
-	Out chan *types.Queue `json:"-"`
+	Out chan *pipeline.Queue `json:"-"`
 	// shared for all buckets (the idea is to kill this afterward)
-	AllOut chan types.Event `json:"-"`
+	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)
@@ -53,7 +53,7 @@ type Leaky struct {
 	Leakspeed    time.Duration
 	BucketConfig *BucketFactory
 	Duration     time.Duration
-	Pour         func(*Leaky, types.Event) `json:"-"`
+	Pour         func(*Leaky, pipeline.Event) `json:"-"`
 	// Profiling when set to true enables profiling of bucket
 	Profiling           bool
 	timedOverflow       bool
@@ -107,9 +107,9 @@ func FromFactory(bucketFactory BucketFactory) *Leaky {
 		Name:            bucketFactory.Name,
 		Limiter:         limiter,
 		Uuid:            seed.Generate(),
-		Queue:           types.NewQueue(Qsize),
+		Queue:           pipeline.NewQueue(Qsize),
 		CacheSize:       bucketFactory.CacheSize,
-		Out:             make(chan *types.Queue, 1),
+		Out:             make(chan *pipeline.Queue, 1),
 		Suicide:         make(chan bool, 1),
 		AllOut:          bucketFactory.ret,
 		Capacity:        bucketFactory.Capacity,
@@ -118,7 +118,7 @@ func FromFactory(bucketFactory BucketFactory) *Leaky {
 		Pour:            Pour,
 		Reprocess:       bucketFactory.Reprocess,
 		Profiling:       bucketFactory.Profiling,
-		Mode:            types.LIVE,
+		Mode:            pipeline.LIVE,
 		scopeType:       bucketFactory.ScopeType,
 		scenarioVersion: bucketFactory.ScenarioVersion,
 		hash:            bucketFactory.hash,
@@ -249,19 +249,19 @@ func LeakRoutine(leaky *Leaky) error {
 			close(leaky.Signal)
 			metrics.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.AllOut <- pipeline.Event{Type: pipeline.OVFLW, Overflow: pipeline.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
+				alert pipeline.RuntimeAlert
 				err   error
 			)
 			leaky.Ovflw_ts = time.Now().UTC()
 			close(leaky.Signal)
 			ofw := leaky.Queue
-			alert = types.RuntimeAlert{Mapkey: leaky.Mapkey}
+			alert = pipeline.RuntimeAlert{Mapkey: leaky.Mapkey}
 
 			if leaky.timedOverflow {
 				metrics.BucketsOverflow.With(prometheus.Labels{"name": leaky.Name}).Inc()
@@ -285,10 +285,10 @@ func LeakRoutine(leaky *Leaky) error {
 			}
 			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.logger.Tracef("Overflow event: %s", spew.Sdump(pipeline.Event{Overflow: alert}))
 			}
 
-			leaky.AllOut <- types.Event{Overflow: alert, Type: types.OVFLW}
+			leaky.AllOut <- pipeline.Event{Overflow: alert, Type: pipeline.OVFLW}
 			leaky.logger.Tracef("Returning from leaky routine.")
 			return nil
 		case <-leaky.tomb.Dying():
@@ -297,15 +297,15 @@ func LeakRoutine(leaky *Leaky) error {
 				ofw := <-leaky.Out
 				leaky.overflow(ofw)
 			}
-			leaky.AllOut <- types.Event{Type: types.OVFLW, Overflow: types.RuntimeAlert{Mapkey: leaky.Mapkey}}
+			leaky.AllOut <- pipeline.Event{Type: pipeline.OVFLW, Overflow: pipeline.RuntimeAlert{Mapkey: leaky.Mapkey}}
 			return nil
 
 		}
 	End:
 	}
 }
 
-func Pour(leaky *Leaky, msg types.Event) {
+func Pour(leaky *Leaky, msg pipeline.Event) {
 	leaky.wgDumpState.Wait()
 	leaky.wgPour.Add(1)
 	defer leaky.wgPour.Done()
@@ -326,7 +326,7 @@ func Pour(leaky *Leaky, msg types.Event) {
 	}
 }
 
-func (leaky *Leaky) overflow(ofw *types.Queue) {
+func (leaky *Leaky) overflow(ofw *pipeline.Queue) {
 	close(leaky.Signal)
 	alert, err := NewAlert(leaky, ofw)
 	if err != nil {
@@ -348,5 +348,5 @@ func (leaky *Leaky) overflow(ofw *types.Queue) {
 
 	metrics.BucketsOverflow.With(prometheus.Labels{"name": leaky.Name}).Inc()
 
-	leaky.AllOut <- types.Event{Overflow: alert, Type: types.OVFLW, MarshaledTime: string(mt)}
+	leaky.AllOut <- pipeline.Event{Overflow: alert, Type: pipeline.OVFLW, MarshaledTime: string(mt)}
 }

pkg/leakybucket/buckets_test.go

@@ -24,12 +24,12 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/cwhub"
 	"github.com/crowdsecurity/crowdsec/pkg/exprhelpers"
 	"github.com/crowdsecurity/crowdsec/pkg/parser"
-	"github.com/crowdsecurity/crowdsec/pkg/types"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
 type TestFile struct {
-	Lines   []types.Event `yaml:"lines,omitempty"`
-	Results []types.Event `yaml:"results,omitempty"`
+	Lines   []pipeline.Event `yaml:"lines,omitempty"`
+	Results []pipeline.Event `yaml:"results,omitempty"`
 }
 
 func TestBucket(t *testing.T) {
@@ -174,8 +174,8 @@ func testOneBucket(t *testing.T, hub *cwhub.Hub, dir string, tomb *tomb.Tomb) er
 	return nil
 }
 
-func testFile(t *testing.T, file string, holders []BucketFactory, response chan types.Event, buckets *Buckets) bool {
-	var results []types.Event
+func testFile(t *testing.T, file string, holders []BucketFactory, response chan pipeline.Event, buckets *Buckets) bool {
+	var results []pipeline.Event
 
 	/* now we can load the test files */
 	// process the yaml
@@ -211,7 +211,7 @@ func testFile(t *testing.T, file string, holders []BucketFactory, response chan
 			latest_ts = ts
 		}
 
-		in.ExpectMode = types.TIMEMACHINE
+		in.ExpectMode = pipeline.TIMEMACHINE
 		log.Infof("Buckets input : %s", spew.Sdump(in))
 
 		ok, err := PourItemToHolders(in, holders, buckets)

pkg/leakybucket/conditional.go

@@ -8,7 +8,7 @@ import (
 	"github.com/expr-lang/expr/vm"
 
 	"github.com/crowdsecurity/crowdsec/pkg/exprhelpers"
-	"github.com/crowdsecurity/crowdsec/pkg/types"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
 var (
@@ -33,7 +33,7 @@ func (c *ConditionalOverflow) OnBucketInit(g *BucketFactory) error {
 	} else {
 		conditionalExprCacheLock.Unlock()
 		//release the lock during compile
-		compiledExpr, err = expr.Compile(g.ConditionalOverflow, exprhelpers.GetExprOptions(map[string]interface{}{"queue": &types.Queue{}, "leaky": &Leaky{}, "evt": &types.Event{}})...)
+		compiledExpr, err = expr.Compile(g.ConditionalOverflow, exprhelpers.GetExprOptions(map[string]interface{}{"queue": &pipeline.Queue{}, "leaky": &Leaky{}, "evt": &pipeline.Event{}})...)
 		if err != nil {
 			return fmt.Errorf("conditional compile error : %w", err)
 		}
@@ -47,8 +47,8 @@ func (c *ConditionalOverflow) OnBucketInit(g *BucketFactory) error {
 	return err
 }
 
-func (c *ConditionalOverflow) AfterBucketPour(b *BucketFactory) func(types.Event, *Leaky) *types.Event {
-	return func(msg types.Event, l *Leaky) *types.Event {
+func (c *ConditionalOverflow) AfterBucketPour(b *BucketFactory) func(pipeline.Event, *Leaky) *pipeline.Event {
+	return func(msg pipeline.Event, l *Leaky) *pipeline.Event {
 		var condition, ok bool
 
 		if c.ConditionalFilterRuntime != nil {

pkg/leakybucket/manager_load.go

@@ -23,6 +23,7 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/enrichment"
 	"github.com/crowdsecurity/crowdsec/pkg/exprhelpers"
 	"github.com/crowdsecurity/crowdsec/pkg/logging"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 	"github.com/crowdsecurity/crowdsec/pkg/types"
 )
 
@@ -63,7 +64,7 @@ type BucketFactory struct {
 	CancelOnFilter      string                 `yaml:"cancel_on,omitempty"` // a filter that, if matched, kills the bucket
 	leakspeed           time.Duration          // internal representation of `Leakspeed`
 	duration            time.Duration          // internal representation of `Duration`
-	ret                 chan types.Event       // the bucket-specific output chan for overflows
+	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.)
 	ScenarioVersion     string                 `yaml:"version,omitempty"`
 	hash                string
@@ -223,7 +224,7 @@ func compileScopeFilter(bucketFactory *BucketFactory) error {
 		return errors.New("filter is mandatory for non-IP, non-Range scope")
 	}
 
-	runTimeFilter, err := expr.Compile(bucketFactory.ScopeType.Filter, exprhelpers.GetExprOptions(map[string]any{"evt": &types.Event{}})...)
+	runTimeFilter, err := expr.Compile(bucketFactory.ScopeType.Filter, exprhelpers.GetExprOptions(map[string]any{"evt": &pipeline.Event{}})...)
 	if err != nil {
 		return fmt.Errorf("error compiling the scope filter: %w", err)
 	}
@@ -233,7 +234,7 @@ func compileScopeFilter(bucketFactory *BucketFactory) error {
 	return nil
 }
 
-func loadBucketFactoriesFromFile(item *cwhub.Item, hub *cwhub.Hub, buckets *Buckets, tomb *tomb.Tomb, response chan types.Event, orderEvent bool, simulationConfig csconfig.SimulationConfig) ([]BucketFactory, error) {
+func loadBucketFactoriesFromFile(item *cwhub.Item, hub *cwhub.Hub, buckets *Buckets, tomb *tomb.Tomb, response chan pipeline.Event, orderEvent bool, simulationConfig csconfig.SimulationConfig) ([]BucketFactory, error) {
 	itemPath := item.State.LocalPath
 
 	// process the yaml
@@ -312,9 +313,9 @@ func loadBucketFactoriesFromFile(item *cwhub.Item, hub *cwhub.Hub, buckets *Buck
 	return factories, nil
 }
 
-func LoadBuckets(cscfg *csconfig.CrowdsecServiceCfg, hub *cwhub.Hub, scenarios []*cwhub.Item, tomb *tomb.Tomb, buckets *Buckets, orderEvent bool) ([]BucketFactory, chan types.Event, error) {
+func LoadBuckets(cscfg *csconfig.CrowdsecServiceCfg, hub *cwhub.Hub, scenarios []*cwhub.Item, tomb *tomb.Tomb, buckets *Buckets, orderEvent bool) ([]BucketFactory, chan pipeline.Event, error) {
 	allFactories := []BucketFactory{}
-	response := make(chan types.Event, 1)
+	response := make(chan pipeline.Event, 1)
 
 	for _, item := range scenarios {
 		log.Debugf("Loading '%s'", item.State.LocalPath)
@@ -377,13 +378,13 @@ func LoadBucket(bucketFactory *BucketFactory, tomb *tomb.Tomb) error {
 		return errors.New("bucket without filter directive")
 	}
 
-	bucketFactory.RunTimeFilter, err = expr.Compile(bucketFactory.Filter, exprhelpers.GetExprOptions(map[string]any{"evt": &types.Event{}})...)
+	bucketFactory.RunTimeFilter, err = expr.Compile(bucketFactory.Filter, exprhelpers.GetExprOptions(map[string]any{"evt": &pipeline.Event{}})...)
 	if err != nil {
 		return fmt.Errorf("invalid filter '%s' in %s: %w", bucketFactory.Filter, bucketFactory.Filename, err)
 	}
 
 	if bucketFactory.GroupBy != "" {
-		bucketFactory.RunTimeGroupBy, err = expr.Compile(bucketFactory.GroupBy, exprhelpers.GetExprOptions(map[string]any{"evt": &types.Event{}})...)
+		bucketFactory.RunTimeGroupBy, err = expr.Compile(bucketFactory.GroupBy, exprhelpers.GetExprOptions(map[string]any{"evt": &pipeline.Event{}})...)
 		if err != nil {
 			return fmt.Errorf("invalid groupby '%s' in %s: %w", bucketFactory.GroupBy, bucketFactory.Filename, err)
 		}
@@ -411,7 +412,7 @@ func LoadBucket(bucketFactory *BucketFactory, tomb *tomb.Tomb) error {
 		bucketFactory.logger.Tracef("Adding a non duplicate filter")
 		bucketFactory.processors = append(bucketFactory.processors, &Uniq{})
 		// we're compiling and discarding the expression to be able to detect it during loading
-		_, err = expr.Compile(bucketFactory.Distinct, exprhelpers.GetExprOptions(map[string]any{"evt": &types.Event{}})...)
+		_, err = expr.Compile(bucketFactory.Distinct, exprhelpers.GetExprOptions(map[string]any{"evt": &pipeline.Event{}})...)
 		if err != nil {
 			return fmt.Errorf("invalid distinct '%s' in %s: %w", bucketFactory.Distinct, bucketFactory.Filename, err)
 		}
@@ -421,7 +422,7 @@ func LoadBucket(bucketFactory *BucketFactory, tomb *tomb.Tomb) error {
 		bucketFactory.logger.Tracef("Adding a cancel_on filter")
 		bucketFactory.processors = append(bucketFactory.processors, &CancelOnFilter{})
 		// we're compiling and discarding the expression to be able to detect it during loading
-		_, err = expr.Compile(bucketFactory.CancelOnFilter, exprhelpers.GetExprOptions(map[string]any{"evt": &types.Event{}})...)
+		_, err = expr.Compile(bucketFactory.CancelOnFilter, exprhelpers.GetExprOptions(map[string]any{"evt": &pipeline.Event{}})...)
 		if err != nil {
 			return fmt.Errorf("invalid cancel_on '%s' in %s: %w", bucketFactory.CancelOnFilter, bucketFactory.Filename, err)
 		}
@@ -455,7 +456,7 @@ func LoadBucket(bucketFactory *BucketFactory, tomb *tomb.Tomb) error {
 		bucketFactory.logger.Tracef("Adding conditional overflow")
 		bucketFactory.processors = append(bucketFactory.processors, &ConditionalOverflow{})
 		// we're compiling and discarding the expression to be able to detect it during loading
-		_, err = expr.Compile(bucketFactory.ConditionalOverflow, exprhelpers.GetExprOptions(map[string]any{"queue": &types.Queue{}, "leaky": &Leaky{}, "evt": &types.Event{}})...)
+		_, err = expr.Compile(bucketFactory.ConditionalOverflow, exprhelpers.GetExprOptions(map[string]any{"queue": &pipeline.Queue{}, "leaky": &Leaky{}, "evt": &pipeline.Event{}})...)
 		if err != nil {
 			return fmt.Errorf("invalid condition '%s' in %s: %w", bucketFactory.ConditionalOverflow, bucketFactory.Filename, err)
 		}