fix: data race in pattern aggregation push

pkg/pattern/aggregation/push.go

@@ -59,8 +59,9 @@ type Push struct {
 	contentType string
 	logger      log.Logger
 
-	// shutdown channels
-	quit chan struct{}
+	running  sync.WaitGroup
+	quit     chan struct{}
+	quitOnce sync.Once
 
 	// auth
 	username, password string
@@ -149,6 +150,7 @@ func NewPush(
 		metrics: metrics,
 	}
 
+	p.running.Add(1)
 	go p.run(pushPeriod)
 	return p, nil
 }
@@ -160,10 +162,10 @@ func (p *Push) WriteEntry(ts time.Time, e string, lbls labels.Labels) {
 
 // Stop will cancel any ongoing requests and stop the goroutine listening for requests
 func (p *Push) Stop() {
-	if p.quit != nil {
+	p.quitOnce.Do(func() {
 		close(p.quit)
-		p.quit = nil
-	}
+	})
+	p.running.Wait()
 }
 
 // buildPayload creates the snappy compressed protobuf to send to Loki
@@ -244,6 +246,8 @@ func (p *Push) buildPayload(ctx context.Context) ([]byte, error) {
 
 // run pulls lines out of the channel and sends them to Loki
 func (p *Push) run(pushPeriod time.Duration) {
+	p.running.Done()
+
 	ctx, cancel := context.WithCancel(context.Background())
 	pushTicker := time.NewTimer(pushPeriod)
 	defer pushTicker.Stop()

pkg/pattern/aggregation/push_test.go

@@ -169,6 +169,7 @@ func Test_Push(t *testing.T) {
 			lbls2,
 		)
 
+		p.running.Add(1)
 		go p.run(time.Nanosecond)
 
 		select {