pkg/leakybucket: replace Signal chan with explicit read/done chans (#4277)

* pkg/leakybucket: replace Signal chan with explicit read/done chans

* fix: remove %w of nil err

Author: mmetc

pkg/leakybucket/bucket.go

@@ -43,8 +43,10 @@ type Leaky struct {
 	CacheSize int
 	// the unique identifier of the bucket (a hash)
 	Mapkey string
-	// chan for signaling
-	Signal       chan bool `json:"-"`
+	ready        chan struct{} // closed when LeakRoutine is ready
+	readyOnce    sync.Once     // use to prevent double close
+	done         chan struct{} // closed when LeakRoutine has stopped processing
+	doneOnce     sync.Once     // use to prevent double close
 	Suicide      chan bool `json:"-"`
 	Reprocess    bool
 	Simulated    bool
@@ -150,6 +152,8 @@ func (l *Leaky) LeakRoutine(ctx context.Context, gate pourGate) {
 		firstEvent         = true
 	)
 
+	defer l.markDone()
+
 	defer func() {
 		if durationTicker != nil {
 			durationTicker.Stop()
@@ -170,13 +174,12 @@ func (l *Leaky) LeakRoutine(ctx context.Context, gate pourGate) {
 	// and preventing them from being destroyed
 	processors := deepcopy.Copy(l.BucketConfig.processors).([]Processor)
 
-	l.Signal <- true
+	l.markReady()
 
 	for _, f := range processors {
 		err := f.OnBucketInit(l.BucketConfig)
 		if err != nil {
 			l.logger.Errorf("Problem at bucket initializiation. Bail out %T : %v", f, err)
-			close(l.Signal)
 			return
 		}
 	}
@@ -237,7 +240,8 @@ func (l *Leaky) LeakRoutine(ctx context.Context, gate pourGate) {
 			return
 		// suiciiiide
 		case <-l.Suicide:
-			close(l.Signal)
+			// don't wait defer to close the channel, in case we are blocked before returning
+			l.markDone()
 			metrics.BucketsCanceled.With(prometheus.Labels{"name": l.Name}).Inc()
 			l.logger.Debugf("Suicide triggered")
 			l.AllOut <- pipeline.Event{Type: pipeline.OVFLW, Overflow: pipeline.RuntimeAlert{Mapkey: l.Mapkey}}
@@ -250,7 +254,7 @@ func (l *Leaky) LeakRoutine(ctx context.Context, gate pourGate) {
 				err   error
 			)
 			l.Ovflw_ts = time.Now().UTC()
-			close(l.Signal)
+			l.markDone()
 			ofw := l.Queue
 			alert = pipeline.RuntimeAlert{Mapkey: l.Mapkey}
 
@@ -316,7 +320,7 @@ func Pour(l *Leaky, gate pourGate, msg pipeline.Event) {
 }
 
 func (l *Leaky) overflow(ofw *pipeline.Queue) {
-	close(l.Signal)
+	l.markDone()
 	alert, err := NewAlert(l, ofw)
 	if err != nil {
 		log.Errorf("%s", err)
@@ -339,3 +343,16 @@ func (l *Leaky) overflow(ofw *pipeline.Queue) {
 
 	l.AllOut <- pipeline.Event{Overflow: alert, Type: pipeline.OVFLW, MarshaledTime: string(mt)}
 }
+
+func (l *Leaky) markReady() {
+	l.readyOnce.Do(func() {
+		close(l.ready)
+	})
+}
+
+func (l *Leaky) markDone() {
+	l.doneOnce.Do(func() {
+		close(l.done)
+	})
+}
+

pkg/leakybucket/manager_load.go

@@ -262,7 +262,7 @@ func (f *BucketFactory) LoadBucket() error {
 
 	impl, ok := bucketTypes[f.Type]
 	if !ok {
-		return fmt.Errorf("invalid type '%s' in %s: %w", f.Type, f.Filename, err)
+		return fmt.Errorf("invalid type '%s' in %s", f.Type, f.Filename)
 	}
 
 	f.processors = impl.BuildProcessors(f)

pkg/leakybucket/manager_run.go

@@ -87,18 +87,16 @@ func PourItemToBucket(
 
 		/* check if leak routine is up */
 		select {
-		case _, ok := <-bucket.Signal:
-			if !ok {
-				// the bucket was found and dead, get a new one and continue
-				bucket.logger.Tracef("Bucket %s found dead, cleanup the body", buckey)
-				bucketStore.Delete(buckey)
-				sigclosed += 1
-				bucket, err = LoadOrStoreBucketFromHolder(ctx, buckey, bucketStore, holder, parsed.ExpectMode)
-				if err != nil {
-					return err
-				}
-				continue
+		case <-bucket.done:
+			// the bucket was found and dead, get a new one and continue
+			bucket.logger.Tracef("Bucket %s found dead, cleanup the body", buckey)
+			bucketStore.Delete(buckey)
+			sigclosed += 1
+			bucket, err = LoadOrStoreBucketFromHolder(ctx, buckey, bucketStore, holder, parsed.ExpectMode)
+			if err != nil {
+				return err
 			}
+			continue
 			// holder.logger.Tracef("Signal exists, try to pour :)")
 		default:
 			// nothing to read, but not closed, try to pour
@@ -177,7 +175,8 @@ func LoadOrStoreBucketFromHolder(
 	}
 	fresh_bucket.In = make(chan *pipeline.Event)
 	fresh_bucket.Mapkey = partitionKey
-	fresh_bucket.Signal = make(chan bool, 1)
+	fresh_bucket.ready = make(chan struct{})
+	fresh_bucket.done = make(chan struct{})
 	actual, stored := buckets.LoadOrStore(partitionKey, fresh_bucket)
 	if !stored {
 		go func() {
@@ -187,7 +186,7 @@ func LoadOrStoreBucketFromHolder(
 		}()
 		leaky = fresh_bucket
 		// once the created goroutine is ready to process event, we can return it
-		<-fresh_bucket.Signal
+		<-fresh_bucket.ready
 	} else {
 		holder.logger.Debugf("Unexpectedly found exisint bucket for %s", partitionKey)
 		leaky = actual