cmd/crowdsec: rename pipeline channels (#4175)

Author: mmetc

cmd/crowdsec/crowdsec.go

@@ -74,7 +74,7 @@ func startParserRoutines(ctx context.Context, g *errgroup.Group, cConfig *csconf
 		log.WithField("idx", idx).Info("Starting parser routine")
 		g.Go(func() error {
 			defer trace.CatchPanic("crowdsec/runParse/"+strconv.Itoa(idx))
-			runParse(ctx, inputLineChan, inputEventChan, *parsers.Ctx, parsers.Nodes)
+			runParse(ctx, logLines, inEvents, *parsers.Ctx, parsers.Nodes)
 			return nil
 		})
 	}
@@ -85,7 +85,7 @@ func startBucketRoutines(ctx context.Context, g *errgroup.Group, cConfig *csconf
 		log.WithField("idx", idx).Info("Starting bucket routine")
 		g.Go(func() error {
 			defer trace.CatchPanic("crowdsec/runPour/"+strconv.Itoa(idx))
-			runPour(ctx, inputEventChan, holders, buckets, cConfig)
+			runPour(ctx, inEvents, holders, buckets, cConfig)
 			return nil
 		})
 	}
@@ -101,7 +101,7 @@ func startOutputRoutines(ctx context.Context, cConfig *csconfig.Config, parsers
 		log.WithField("idx", idx).Info("Starting output routine")
 		outputsTomb.Go(func() error {
 			defer trace.CatchPanic("crowdsec/runOutput/"+strconv.Itoa(idx))
-			return runOutput(ctx, inputEventChan, outputEventChan, buckets, *parsers.PovfwCtx, parsers.Povfwnodes, apiClient)
+			return runOutput(ctx, inEvents, outEvents, buckets, *parsers.PovfwCtx, parsers.Povfwnodes, apiClient)
 		})
 	}
 }
@@ -135,8 +135,8 @@ func startLPMetrics(ctx context.Context, cConfig *csconfig.Config, apiClient *ap
 
 // runCrowdsec starts the log processor service
 func runCrowdsec(ctx context.Context, g *errgroup.Group, cConfig *csconfig.Config, parsers *parser.Parsers, hub *cwhub.Hub, datasources []acquisition.DataSource) error {
-	inputEventChan = make(chan pipeline.Event)
-	inputLineChan = make(chan pipeline.Event)
+	inEvents = make(chan pipeline.Event)
+	logLines = make(chan pipeline.Event)
 
 	startParserRoutines(ctx, g, cConfig, parsers)
 	startBucketRoutines(ctx, g, cConfig)
@@ -156,7 +156,7 @@ func runCrowdsec(ctx context.Context, g *errgroup.Group, cConfig *csconfig.Confi
 
 	log.Info("Starting processing data")
 
-	if err := acquisition.StartAcquisition(ctx, dataSources, inputLineChan, &acquisTomb); err != nil {
+	if err := acquisition.StartAcquisition(ctx, dataSources, logLines, &acquisTomb); err != nil {
 		return fmt.Errorf("starting acquisition error: %w", err)
 	}
 

cmd/crowdsec/main.go

@@ -44,9 +44,9 @@ var (
 	holders []leakybucket.BucketFactory
 	buckets *leakybucket.Buckets
 
-	inputLineChan   chan pipeline.Event
-	inputEventChan  chan pipeline.Event
-	outputEventChan chan pipeline.Event // the buckets init returns its own chan that is used for multiplexing
+	logLines   chan pipeline.Event
+	inEvents  chan pipeline.Event
+	outEvents chan pipeline.Event // the buckets init returns its own chan that is used for multiplexing
 	pluginBroker      csplugin.PluginBroker
 )
 
@@ -59,7 +59,7 @@ func LoadBuckets(cConfig *csconfig.Config, hub *cwhub.Hub) error {
 
 	log.Infof("Loading %d scenario files", len(scenarios))
 
-	holders, outputEventChan, err = leakybucket.LoadBuckets(cConfig.Crowdsec, hub, scenarios, buckets, flags.OrderEvent)
+	holders, outEvents, err = leakybucket.LoadBuckets(cConfig.Crowdsec, hub, scenarios, buckets, flags.OrderEvent)
 	if err != nil {
 		return err
 	}

cmd/crowdsec/parse.go

@@ -24,7 +24,7 @@ func parseEvent(
 	- one that is treated as a log and can go to scenarios
 	- another one that will go directly to LAPI*/
 	if event.Type == pipeline.APPSEC {
-		outputEventChan <- event
+		outEvents <- event
 		return nil
 	}
 	if event.Line.Module == "" {

cmd/crowdsec/serve.go

@@ -105,7 +105,7 @@ func ShutdownCrowdsecRoutines(cancel context.CancelFunc, g *errgroup.Group) erro
 	if len(dataSources) > 0 {
 		acquisTomb.Kill(nil)
 		log.Debugf("waiting for acquisition to finish")
-		drainChan(inputLineChan)
+		drainChan(logLines)
 
 		if err := acquisTomb.Wait(); err != nil {
 			log.Warningf("Acquisition returned error : %s", err)
@@ -114,7 +114,7 @@ func ShutdownCrowdsecRoutines(cancel context.CancelFunc, g *errgroup.Group) erro
 	}
 
 	log.Debugf("acquisition is finished, wait for parser/bucket/ouputs.")
-	drainChan(inputEventChan)
+	drainChan(inEvents)
 
 	outputsTomb.Kill(nil)