refact: drop parserTomb, lpMetricsTomb (#4138)

Author: mmetc

cmd/crowdsec/crowdsec.go

@@ -5,10 +5,10 @@ import (
 	"fmt"
 	"os"
 	"strconv"
-	"sync"
 	"time"
 
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/crowdsecurity/go-cs-lib/trace"
 
@@ -69,40 +69,25 @@ func initCrowdsec(ctx context.Context, cConfig *csconfig.Config, hub *cwhub.Hub,
 	return csParsers, datasources, nil
 }
 
-func startParserRoutines(cConfig *csconfig.Config, parsers *parser.Parsers) {
-	// start go-routines for parsing, buckets pour and outputs.
-	parserWg := &sync.WaitGroup{}
-
-	parsersTomb.Go(func() error {
-		parserWg.Add(1)
-
-		for range cConfig.Crowdsec.ParserRoutinesCount {
-			parsersTomb.Go(func() error {
-				defer trace.CatchPanic("crowdsec/runParse")
-
-				if err := runParse(inputLineChan, inputEventChan, *parsers.Ctx, parsers.Nodes); err != nil {
-					// this error will never happen as parser.Parse is not able to return errors
-					return err
-				}
-
-				return nil
-			})
-		}
-
-		parserWg.Done()
-
-		return nil
-	})
-	parserWg.Wait()
+func startParserRoutines(ctx context.Context, g *errgroup.Group, cConfig *csconfig.Config, parsers *parser.Parsers) {
+	for idx := range cConfig.Crowdsec.ParserRoutinesCount {
+		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)
+			return nil
+		})
+	}
 }
 
-func startBucketRoutines(ctx context.Context, cConfig *csconfig.Config) {
+func startBucketRoutines(ctx context.Context, g *errgroup.Group, cConfig *csconfig.Config) {
 	for idx := range cConfig.Crowdsec.BucketsRoutinesCount {
-		log.Infof("Starting bucket routine %d", idx)
-		go func() {
+		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)
-		}()
+			return nil
+		})
 	}
 }
 
@@ -112,24 +97,13 @@ func startHeartBeat(ctx context.Context, _ *csconfig.Config, apiClient *apiclien
 }
 
 func startOutputRoutines(ctx context.Context, cConfig *csconfig.Config, parsers *parser.Parsers, apiClient *apiclient.ApiClient) {
-	outputWg := &sync.WaitGroup{}
-
-	outputsTomb.Go(func() error {
-		outputWg.Add(1)
-
-		for range cConfig.Crowdsec.OutputRoutinesCount {
-			outputsTomb.Go(func() error {
-				defer trace.CatchPanic("crowdsec/runOutput")
-
-				return runOutput(ctx, inputEventChan, outputEventChan, buckets, *parsers.PovfwCtx, parsers.Povfwnodes, apiClient)
-			})
-		}
-
-		outputWg.Done()
-
-		return nil
-	})
-	outputWg.Wait()
+	for idx := range cConfig.Crowdsec.OutputRoutinesCount {
+		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)
+		})
+	}
 }
 
 func startLPMetrics(ctx context.Context, cConfig *csconfig.Config, apiClient *apiclient.ApiClient, hub *cwhub.Hub, datasources []acquisition.DataSource) error {
@@ -142,9 +116,9 @@ func startLPMetrics(ctx context.Context, cConfig *csconfig.Config, apiClient *ap
 		hub,
 	)
 
-	lpMetricsTomb.Go(func() error {
-		return mp.Run(ctx, &lpMetricsTomb)
-	})
+	go func() {
+		mp.Run(ctx)
+	}()
 
 	if cConfig.Prometheus != nil && cConfig.Prometheus.Enabled {
 		aggregated := false
@@ -161,13 +135,12 @@ func startLPMetrics(ctx context.Context, cConfig *csconfig.Config, apiClient *ap
 }
 
 // runCrowdsec starts the log processor service
-func runCrowdsec(ctx context.Context, cConfig *csconfig.Config, parsers *parser.Parsers, hub *cwhub.Hub, datasources []acquisition.DataSource) error {
+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)
 
-	startParserRoutines(cConfig, parsers)
-
-	startBucketRoutines(ctx, cConfig)
+	startParserRoutines(ctx, g, cConfig, parsers)
+	startBucketRoutines(ctx, g, cConfig)
 
 	apiClient, err := apiclient.GetLAPIClient()
 	if err != nil {
@@ -195,6 +168,7 @@ func runCrowdsec(ctx context.Context, cConfig *csconfig.Config, parsers *parser.
 func serveCrowdsec(ctx context.Context, parsers *parser.Parsers, cConfig *csconfig.Config, hub *cwhub.Hub, datasources []acquisition.DataSource, agentReady chan bool) {
 	cctx, cancel := context.WithCancel(ctx)
 
+	var g errgroup.Group
 
 	crowdsecTomb.Go(func() error {
 		defer trace.CatchPanic("crowdsec/serveCrowdsec")
@@ -206,7 +180,7 @@ func serveCrowdsec(ctx context.Context, parsers *parser.Parsers, cConfig *csconf
 
 			agentReady <- true
 
-			if err := runCrowdsec(cctx, cConfig, parsers, hub, datasources); err != nil {
+			if err := runCrowdsec(cctx, &g, cConfig, parsers, hub, datasources); err != nil {
 				log.Fatalf("unable to start crowdsec routines: %s", err)
 			}
 		}()
@@ -218,7 +192,7 @@ func serveCrowdsec(ctx context.Context, parsers *parser.Parsers, cConfig *csconf
 		waitOnTomb()
 		log.Debugf("Shutting down crowdsec routines")
 
-		if err := ShutdownCrowdsecRoutines(cancel); err != nil {
+		if err := ShutdownCrowdsecRoutines(cancel, &g); err != nil {
 			return fmt.Errorf("unable to shutdown crowdsec routines: %w", err)
 		}
 

cmd/crowdsec/lpmetrics.go

@@ -13,7 +13,6 @@ import (
 	io_prometheus_client "github.com/prometheus/client_model/go"
 
 	"github.com/sirupsen/logrus"
-	"gopkg.in/tomb.v2"
 
 	"github.com/crowdsecurity/go-cs-lib/ptr"
 	"github.com/crowdsecurity/go-cs-lib/trace"
@@ -403,11 +402,11 @@ func (m *MetricsProvider) sendMetrics(ctx context.Context, met *models.AllMetric
 	}
 }
 
-func (m *MetricsProvider) Run(ctx context.Context, myTomb *tomb.Tomb) error {
+func (m *MetricsProvider) Run(ctx context.Context) {
 	defer trace.CatchPanic("crowdsec/MetricsProvider.Run")
 
 	if m.interval == time.Duration(0) {
-		return nil
+		return
 	}
 
 	ticker := time.NewTicker(1) // Send on start
@@ -418,9 +417,9 @@ func (m *MetricsProvider) Run(ctx context.Context, myTomb *tomb.Tomb) error {
 			met := m.metricsPayload()
 			m.sendMetrics(ctx, met)
 			ticker.Reset(m.interval)
-		case <-myTomb.Dying():
+		case <-ctx.Done():
 			ticker.Stop()
-			return nil
+			return
 		}
 	}
 }

cmd/crowdsec/main.go

@@ -31,12 +31,10 @@ import (
 var (
 	// tombs for the parser, buckets and outputs.
 	acquisTomb    tomb.Tomb
-	parsersTomb   tomb.Tomb
 	outputsTomb   tomb.Tomb
 	apiTomb       tomb.Tomb
 	crowdsecTomb  tomb.Tomb
 	pluginTomb    tomb.Tomb
-	lpMetricsTomb tomb.Tomb
 
 	flags Flags
 

cmd/crowdsec/output.go

@@ -132,12 +132,11 @@ func runOutput(ctx context.Context, input chan pipeline.Event, overflow chan pip
 			}
 
 			if event.Overflow.Reprocess {
-				log.Debugf("Overflow being reprocessed.")
 				select {
 				case input <- event:
-					log.Debugf("reprocessing overflow event")
-				case <-parsersTomb.Dead():
-					log.Debugf("parsing is dead, skipping")
+					log.Debug("Reprocessing overflow event")
+				case <-ctx.Done():
+					log.Debug("Reprocessing overflow event: parsing is dead, skipping")
 				}
 			}
 			if dumpStates {

cmd/crowdsec/parse.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -11,12 +12,12 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
-func runParse(input chan pipeline.Event, output chan pipeline.Event, parserCTX parser.UnixParserCtx, nodes []parser.Node) error {
+func runParse(ctx context.Context, input chan pipeline.Event, output chan pipeline.Event, parserCTX parser.UnixParserCtx, nodes []parser.Node) {
 	for {
 		select {
-		case <-parsersTomb.Dying():
+		case <-ctx.Done():
 			log.Infof("Killing parser routines")
-			return nil
+			return
 		case event := <-input:
 			if !event.Process {
 				continue

cmd/crowdsec/serve.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/sync/errgroup"
 	"gopkg.in/tomb.v2"
 
 	"github.com/crowdsecurity/go-cs-lib/csdaemon"
@@ -27,12 +28,10 @@ import (
 func reloadHandler(ctx context.Context, _ os.Signal) (*csconfig.Config, error) {
 	// re-initialize tombs
 	acquisTomb = tomb.Tomb{}
-	parsersTomb = tomb.Tomb{}
 	outputsTomb = tomb.Tomb{}
 	apiTomb = tomb.Tomb{}
 	crowdsecTomb = tomb.Tomb{}
 	pluginTomb = tomb.Tomb{}
-	lpMetricsTomb = tomb.Tomb{}
 
 	cConfig, err := LoadConfig(flags.ConfigFile, flags.DisableAgent, flags.DisableAPI, false)
 	if err != nil {
@@ -86,7 +85,19 @@ func reloadHandler(ctx context.Context, _ os.Signal) (*csconfig.Config, error) {
 	return cConfig, nil
 }
 
-func ShutdownCrowdsecRoutines(cancel context.CancelFunc) error {
+func waitErrGroup(g *errgroup.Group, timeout time.Duration) error {
+	done := make(chan error, 1)
+	go func() { done <- g.Wait() }()
+
+	select {
+	case err := <-done:
+		return err
+	case <-time.After(timeout):
+		return context.DeadlineExceeded
+	}
+}
+
+func ShutdownCrowdsecRoutines(cancel context.CancelFunc, g *errgroup.Group) error {
 	var reterr error
 
 	log.Debugf("Shutting down crowdsec sub-routines")
@@ -103,20 +114,8 @@ func ShutdownCrowdsecRoutines(cancel context.CancelFunc) error {
 	}
 
 	log.Debugf("acquisition is finished, wait for parser/bucket/ouputs.")
-	parsersTomb.Kill(nil)
 	drainChan(inputEventChan)
 
-	if err := parsersTomb.Wait(); err != nil {
-		log.Warningf("Parsers returned error : %s", err)
-		reterr = err
-	}
-
-	log.Debugf("parsers is done")
-	time.Sleep(1 * time.Second) // ugly workaround for now to ensure PourItemtoholders are finished
-	cancel()
-
-	log.Debugf("buckets is done")
-	time.Sleep(1 * time.Second) // ugly workaround for now
 	outputsTomb.Kill(nil)
 
 	done := make(chan error, 1)
@@ -139,13 +138,14 @@ func ShutdownCrowdsecRoutines(cancel context.CancelFunc) error {
 		log.Warningf("Outputs didn't finish in time, some events may have not been flushed")
 	}
 
-	lpMetricsTomb.Kill(nil)
+	cancel()
 
-	if err := lpMetricsTomb.Wait(); err != nil {
-		log.Warningf("Metrics returned error : %s", err)
-		reterr = err
+	if err := waitErrGroup(g, 3 * time.Second); err != nil {
+		log.WithError(err).Warn("timeout waiting for parser/bucket routines")
 	}
 
+	log.Debugf("parsers are done")
+	log.Debugf("buckets are done")
 	log.Debugf("metrics are done")
 
 	// He's dead, Jim.
@@ -304,12 +304,10 @@ func HandleSignals(ctx context.Context, cConfig *csconfig.Config) error {
 
 func Serve(ctx context.Context, cConfig *csconfig.Config, agentReady chan bool) error {
 	acquisTomb = tomb.Tomb{}
-	parsersTomb = tomb.Tomb{}
 	outputsTomb = tomb.Tomb{}
 	apiTomb = tomb.Tomb{}
 	crowdsecTomb = tomb.Tomb{}
 	pluginTomb = tomb.Tomb{}
-	lpMetricsTomb = tomb.Tomb{}
 
 	if cConfig.API.Server != nil && cConfig.API.Server.DbConfig != nil {
 		dbCfg := cConfig.API.Server.DbConfig