feat: producer double close will not panic (#356)

* monitor safe close

* sts ch close

Author: crimson-gao

producer/monitor.go

@@ -25,6 +25,7 @@ type ProducerMetrics struct {
 type ProducerMonitor struct {
 	metrics atomic.Value // *ProducerMetrics
 	stopCh  chan struct{}
+	once    sync.Once
 	wg      sync.WaitGroup
 }
 
@@ -37,7 +38,9 @@ func newProducerMonitor() *ProducerMonitor {
 }
 
 func (m *ProducerMonitor) Stop() {
-	close(m.stopCh)
+	m.once.Do(func() {
+		close(m.stopCh)
+	})
 	m.wg.Wait()
 }
 
@@ -82,6 +85,7 @@ func (m *ProducerMonitor) getAndResetMetrics() *ProducerMetrics {
 }
 
 func (m *ProducerMonitor) reportThread(reportInterval time.Duration, logger log.Logger) {
+	m.wg.Add(1)
 	defer m.wg.Done()
 	ticker := time.NewTicker(reportInterval)
 	for {

producer/producer.go

@@ -28,6 +28,7 @@ type Producer struct {
 	logger                log.Logger
 	producerLogGroupSize  int64
 	monitor               *ProducerMonitor
+	stsCloseOnce          sync.Once
 }
 
 func NewProducer(producerConfig *ProducerConfig) (*Producer, error) {
@@ -288,7 +289,6 @@ func (producer *Producer) Start() {
 	producer.ioThreadPoolWaitGroup.Add(1)
 	go producer.threadPool.start(producer.ioWorkerWaitGroup, producer.ioThreadPoolWaitGroup)
 	if !producer.producerConfig.DisableRuntimeMetrics {
-		producer.monitor.wg.Add(1)
 		go producer.monitor.reportThread(time.Minute, producer.logger)
 	}
 }
@@ -333,7 +333,9 @@ func (producer *Producer) sendCloseProdcerSignal() {
 
 func (producer *Producer) closeStstokenChannel() {
 	if producer.producerConfig.StsTokenShutDown != nil {
-		close(producer.producerConfig.StsTokenShutDown)
-		level.Info(producer.logger).Log("msg", "producer closed ststoken")
+		producer.stsCloseOnce.Do(func() {
+			close(producer.producerConfig.StsTokenShutDown)
+			level.Info(producer.logger).Log("msg", "producer closed ststoken")
+		})
 	}
 }