group collectors in a StateDumper{}

Author: mmetc

cmd/crowdsec/crowdsec.go

@@ -142,15 +142,13 @@ func runCrowdsec(
 	parsers *parser.Parsers,
 	hub *cwhub.Hub,
 	datasources []acquisitionTypes.DataSource,
-	pourCollector *leakybucket.PourCollector,
-	stageCollector *parser.StageParseCollector,
-	bucketOverflows []pipeline.Event,
+	sd *StateDumper,
 ) error {
 	inEvents = make(chan pipeline.Event)
 	logLines = make(chan pipeline.Event)
 
-	startParserRoutines(ctx, g, cConfig, parsers, stageCollector)
-	startBucketRoutines(ctx, g, cConfig, pourCollector)
+	startParserRoutines(ctx, g, cConfig, parsers, sd.StageParse)
+	startBucketRoutines(ctx, g, cConfig, sd.Pour)
 
 	apiClient, err := apiclient.GetLAPIClient()
 	if err != nil {
@@ -159,7 +157,7 @@ func runCrowdsec(
 
 	startHeartBeat(ctx, cConfig, apiClient)
 
-	startOutputRoutines(ctx, cConfig, parsers, apiClient, stageCollector, bucketOverflows)
+	startOutputRoutines(ctx, cConfig, parsers, apiClient, sd.StageParse, sd.BucketOverflows)
 
 	if err := startLPMetrics(ctx, cConfig, apiClient, hub, datasources); err != nil {
 		return err
@@ -182,9 +180,7 @@ func serveCrowdsec(
 	hub *cwhub.Hub,
 	datasources []acquisitionTypes.DataSource,
 	agentReady chan bool,
-	pourCollector *leakybucket.PourCollector,
-	stageCollector *parser.StageParseCollector,
-	bucketOverflows []pipeline.Event,
+	sd *StateDumper,
 ) {
 	cctx, cancel := context.WithCancel(ctx)
 
@@ -200,7 +196,7 @@ func serveCrowdsec(
 
 			agentReady <- true
 
-			if err := runCrowdsec(cctx, &g, cConfig, parsers, hub, datasources, pourCollector, stageCollector, bucketOverflows); err != nil {
+			if err := runCrowdsec(cctx, &g, cConfig, parsers, hub, datasources, sd); err != nil {
 				log.Fatalf("unable to start crowdsec routines: %s", err)
 			}
 		}()
@@ -217,11 +213,12 @@ func serveCrowdsec(
 		}
 
 		log.Debugf("everything is dead, return crowdsecTomb")
+		log.Debugf("sd.DumpDir == %s", sd.DumpDir)
 
-		if flags.DumpDir != "" {
-			log.Debugf("Dumping parser+bucket states to %s", flags.DumpDir)
+		if sd.DumpDir != "" {
+			log.Debugf("Dumping parser+bucket states to %s", sd.DumpDir)
 
-			if err := dumpAllStates(flags.DumpDir, pourCollector, stageCollector, bucketOverflows); err != nil {
+			if err := sd.Dump(); err != nil {
 				log.Fatal(err)
 			}
 

cmd/crowdsec/dump.go

@@ -12,21 +12,42 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
-func dumpAllStates(dir string, pourCollector *leakybucket.PourCollector, stageCollector *parser.StageParseCollector, bucketOverflows []pipeline.Event) error {
+type StateDumper struct {
+	DumpDir string
+	Pour *leakybucket.PourCollector
+	StageParse *parser.StageParseCollector
+	BucketOverflows []pipeline.Event
+}
+
+func NewStateDumper(dumpDir string) *StateDumper {
+	if dumpDir == "" {
+		return &StateDumper{}
+	}
+
+	return &StateDumper{
+		DumpDir:    dumpDir,
+		Pour:       leakybucket.NewPourCollector(),
+		StageParse: parser.NewStageParseCollector(),
+	}
+}
+
+func (sd *StateDumper) Dump() error {
+	dir := sd.DumpDir
+
 	err := os.MkdirAll(dir, 0o755)
 	if err != nil {
 		return err
 	}
 
-	if err := dumpCollector(dir, "parser-dump.yaml", stageCollector); err != nil {
+	if err := dumpCollector(dir, "parser-dump.yaml", sd.StageParse); err != nil {
 		return fmt.Errorf("dumping parser state: %w", err)
 	}
 
-	if err := dumpState(dir, "bucket-dump.yaml", bucketOverflows); err != nil {
+	if err := dumpState(dir, "bucket-dump.yaml", sd.BucketOverflows); err != nil {
 		return fmt.Errorf("dumping bucket overflow state: %w", err)
 	}
 
-	if err := dumpCollector(dir, "bucketpour-dump.yaml", pourCollector); err != nil {
+	if err := dumpCollector(dir, "bucketpour-dump.yaml", sd.Pour); err != nil {
 		return fmt.Errorf("dumping bucket pour state: %w", err)
 	}
 

cmd/crowdsec/main.go

@@ -26,7 +26,6 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/fflag"
 	"github.com/crowdsecurity/crowdsec/pkg/leakybucket"
 	"github.com/crowdsecurity/crowdsec/pkg/logging"
-	"github.com/crowdsecurity/crowdsec/pkg/parser"
 	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
@@ -214,18 +213,9 @@ func run(flags Flags) error {
 		return err
 	}
 
-	var (
-		pourCollector *leakybucket.PourCollector
-		stageCollector *parser.StageParseCollector
-		bucketOverflows []pipeline.Event
-	)
+	sd := NewStateDumper(flags.DumpDir)
 
-	if flags.DumpDir != "" {
-		pourCollector = leakybucket.NewPourCollector()
-		stageCollector = parser.NewStageParseCollector()
-	}
-
-	return StartRunSvc(ctx, cConfig, pourCollector, stageCollector, bucketOverflows)
+	return StartRunSvc(ctx, cConfig, sd)
 }
 
 func main() {

cmd/crowdsec/run_in_svc.go

@@ -16,9 +16,6 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/csconfig"
 	"github.com/crowdsecurity/crowdsec/pkg/database"
 	"github.com/crowdsecurity/crowdsec/pkg/fflag"
-	"github.com/crowdsecurity/crowdsec/pkg/leakybucket"
-	"github.com/crowdsecurity/crowdsec/pkg/parser"
-	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
 func isWindowsService() (bool, error) {
@@ -28,9 +25,7 @@ func isWindowsService() (bool, error) {
 func StartRunSvc(
 	ctx context.Context,
 	cConfig *csconfig.Config,
-	pourCollector *leakybucket.PourCollector,
-	stageCollector *parser.StageParseCollector,
-	bucketOverflows []pipeline.Event,
+	sd *StateDumper,
 ) error {
 	defer trace.CatchPanic("crowdsec/StartRunSvc")
 
@@ -72,5 +67,5 @@ func StartRunSvc(
 		}()
 	}
 
-	return Serve(ctx, cConfig, agentReady, pourCollector, stageCollector, bucketOverflows)
+	return Serve(ctx, cConfig, agentReady, sd)
 }

cmd/crowdsec/run_in_svc_windows.go

@@ -15,14 +15,13 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/csconfig"
 	"github.com/crowdsecurity/crowdsec/pkg/database"
 	"github.com/crowdsecurity/crowdsec/pkg/fflag"
-	"github.com/crowdsecurity/crowdsec/pkg/leakybucket"
 )
 
 func isWindowsService() (bool, error) {
 	return svc.IsWindowsService()
 }
 
-func StartRunSvc(ctx context.Context, cConfig *csconfig.Config, pourCollector *leakybucket.PourCollector) error {
+func StartRunSvc(ctx context.Context, cConfig *csconfig.Config, sd *StateDumper) error {
 	const svcName = "CrowdSec"
 	const svcDescription = "Crowdsec IPS/IDS"
 
@@ -37,7 +36,7 @@ func StartRunSvc(ctx context.Context, cConfig *csconfig.Config, pourCollector *l
 		return fmt.Errorf("failed to determine if we are running in windows service mode: %w", err)
 	}
 	if isRunninginService {
-		return runService(svcName, pourCollector)
+		return runService(svcName, sd)
 	}
 
 	switch flags.WinSvc {
@@ -62,15 +61,15 @@ func StartRunSvc(ctx context.Context, cConfig *csconfig.Config, pourCollector *l
 			return fmt.Errorf("failed to %s %s: %w", flags.WinSvc, svcName, err)
 		}
 	case "":
-		return WindowsRun(ctx, cConfig, pourCollector)
+		return WindowsRun(ctx, cConfig, sd)
 	default:
 		return fmt.Errorf("Invalid value for winsvc parameter: %s", flags.WinSvc)
 	}
 
 	return nil
 }
 
-func WindowsRun(ctx context.Context, cConfig *csconfig.Config, pourCollector *leakybucket.PourCollector) error {
+func WindowsRun(ctx context.Context, cConfig *csconfig.Config, sd *StateDumper) error {
 	if fflag.PProfBlockProfile.IsEnabled() {
 		runtime.SetBlockProfileRate(1)
 		runtime.SetMutexProfileFraction(1)
@@ -97,5 +96,5 @@ func WindowsRun(ctx context.Context, cConfig *csconfig.Config, pourCollector *le
 		registerPrometheus(cConfig.Prometheus)
 		go servePrometheus(cConfig.Prometheus, dbClient, agentReady)
 	}
-	return Serve(ctx, cConfig, agentReady, pourCollector)
+	return Serve(ctx, cConfig, agentReady, sd)
 }

cmd/crowdsec/serve.go

@@ -22,8 +22,6 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/cwhub"
 	"github.com/crowdsecurity/crowdsec/pkg/database"
 	"github.com/crowdsecurity/crowdsec/pkg/exprhelpers"
-	"github.com/crowdsecurity/crowdsec/pkg/leakybucket"
-	"github.com/crowdsecurity/crowdsec/pkg/parser"
 	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
@@ -35,16 +33,7 @@ func reloadHandler(ctx context.Context, _ os.Signal) (*csconfig.Config, error) {
 	crowdsecTomb = tomb.Tomb{}
 	pluginTomb = tomb.Tomb{}
 
-	var (
-		pourCollector *leakybucket.PourCollector
-		stageCollector *parser.StageParseCollector
-		bucketOverflows []pipeline.Event
-	)
-
-	if flags.DumpDir != "" {
-		pourCollector = leakybucket.NewPourCollector()
-		stageCollector = parser.NewStageParseCollector()
-	}
+	sd := NewStateDumper(flags.DumpDir)
 
 	cConfig, err := LoadConfig(flags.ConfigFile, flags.DisableAgent, flags.DisableAPI, false)
 	if err != nil {
@@ -90,7 +79,7 @@ func reloadHandler(ctx context.Context, _ os.Signal) (*csconfig.Config, error) {
 		}
 
 		agentReady := make(chan bool, 1)
-		serveCrowdsec(ctx, csParsers, cConfig, hub, datasources, agentReady, pourCollector, stageCollector, bucketOverflows)
+		serveCrowdsec(ctx, csParsers, cConfig, hub, datasources, agentReady, sd)
 	}
 
 	log.Info("Reload is finished")
@@ -318,9 +307,7 @@ func Serve(
 	ctx context.Context,
 	cConfig *csconfig.Config,
 	agentReady chan bool,
-	pourCollector *leakybucket.PourCollector,
-	stageCollector *parser.StageParseCollector,
-	bucketOverflows []pipeline.Event,
+	sd *StateDumper,
 ) error {
 	acquisTomb = tomb.Tomb{}
 	outputsTomb = tomb.Tomb{}
@@ -394,7 +381,7 @@ func Serve(
 
 		// if it's just linting, we're done
 		if !flags.TestMode {
-			serveCrowdsec(ctx, csParsers, cConfig, hub, datasources, agentReady, pourCollector, stageCollector, bucketOverflows)
+			serveCrowdsec(ctx, csParsers, cConfig, hub, datasources, agentReady, sd)
 		} else {
 			agentReady <- true
 		}

cmd/crowdsec/win_service.go

@@ -18,12 +18,11 @@ import (
 	"golang.org/x/sys/windows/svc/eventlog"
 
 	"github.com/crowdsecurity/crowdsec/pkg/csconfig"
-	"github.com/crowdsecurity/crowdsec/pkg/leakybucket"
 )
 
 type crowdsec_winservice struct {
 	config *csconfig.Config
-	pourCollector *leakybucket.PourCollector
+	stateDumper *StateDumper
 }
 
 func (m *crowdsec_winservice) Execute(args []string, r <-chan svc.ChangeRequest, changes chan<- svc.Status) (bool, uint32) {
@@ -63,7 +62,7 @@ func (m *crowdsec_winservice) Execute(args []string, r <-chan svc.ChangeRequest,
 		log.Fatal(err)
 	}
 
-	err = WindowsRun(ctx, cConfig, m.pourCollector)
+	err = WindowsRun(ctx, cConfig, m.stateDumper)
 	changes <- svc.Status{State: svc.Stopped}
 	if err != nil {
 		log.Fatal(err)
@@ -72,7 +71,7 @@ func (m *crowdsec_winservice) Execute(args []string, r <-chan svc.ChangeRequest,
 	return false, 0
 }
 
-func runService(name string, pourCollector *leakybucket.PourCollector) error {
+func runService(name string, sd *StateDumper) error {
 	// All the calls to logging before the logger is configured are pretty much useless, but we keep them for clarity
 	err := eventlog.InstallAsEventCreate("CrowdSec", eventlog.Error|eventlog.Warning|eventlog.Info)
 	if err != nil {
@@ -112,7 +111,7 @@ func runService(name string, pourCollector *leakybucket.PourCollector) error {
 	}
 
 	log.Infof("starting %s service", name)
-	winsvc := crowdsec_winservice{config: cConfig, pourCollector: pourCollector}
+	winsvc := crowdsec_winservice{config: cConfig, stateDumper: sd}
 
 	if err := svc.Run(name, &winsvc); err != nil {
 		return fmt.Errorf("%s service failed: %w", name, err)

pkg/parser/parsing_test.go

@@ -316,7 +316,7 @@ func testSubSet(testSet TestFile, pctx UnixParserCtx, nodes []Node) (bool, error
 	var results []pipeline.Event
 
 	for _, in := range testSet.Lines {
-		out, err := Parse(pctx, in, nodes, false)
+		out, err := Parse(pctx, in, nodes, nil)
 		if err != nil {
 			log.Errorf("Failed to process %s : %v", spew.Sdump(in), err)
 		}