Make cloudwatchlogs's pusher wait for the final flush to complete before returning

plugins/outputs/cloudwatchlogs/cloudwatchlogs.go

@@ -61,17 +61,23 @@ type CloudWatchLogs struct {
 
 	Log telegraf.Logger `toml:"-"`
 
-	cwDests map[Target]*cwDest
+	pusherStopChan  chan struct{}
+	pusherWaitGroup sync.WaitGroup
+	cwDests         map[Target]*cwDest
 }
 
 func (c *CloudWatchLogs) Connect() error {
 	return nil
 }
 
 func (c *CloudWatchLogs) Close() error {
+	close(c.pusherStopChan)
+	c.pusherWaitGroup.Wait()
+
 	for _, d := range c.cwDests {
 		d.Stop()
 	}
+
 	return nil
 }
 
@@ -129,7 +135,7 @@ func (c *CloudWatchLogs) getDest(t Target) *cwDest {
 	client.Handlers.Build.PushBackNamed(handlers.NewRequestCompressionHandler([]string{"PutLogEvents"}))
 	client.Handlers.Build.PushBackNamed(handlers.NewCustomHeaderHandler("User-Agent", agentinfo.UserAgent(t.Group)))
 
-	pusher := NewPusher(t, client, c.ForceFlushInterval.Duration, maxRetryTimeout, c.Log)
+	pusher := NewPusher(t, client, c.ForceFlushInterval.Duration, maxRetryTimeout, c.Log, c.pusherStopChan, &c.pusherWaitGroup)
 	cwd := &cwDest{pusher: pusher, retryer: logThrottleRetryer}
 	c.cwDests[t] = cwd
 	return cwd
@@ -290,7 +296,6 @@ func (cd *cwDest) Publish(events []logs.LogEvent) error {
 }
 
 func (cd *cwDest) Stop() {
-	cd.pusher.Stop()
 	cd.retryer.Stop()
 	cd.stopped = true
 }
@@ -365,6 +370,7 @@ func init() {
 	outputs.Add("cloudwatchlogs", func() telegraf.Output {
 		return &CloudWatchLogs{
 			ForceFlushInterval: internal.Duration{Duration: defaultFlushTimeout},
+			pusherStopChan:     make(chan struct{}),
 			cwDests:            make(map[Target]*cwDest),
 		}
 	})

plugins/outputs/cloudwatchlogs/pusher.go

@@ -50,14 +50,15 @@ type pusher struct {
 	sequenceToken       *string
 	lastValidTime       int64
 	needSort            bool
-	stop                chan struct{}
+	stop                <-chan struct{}
 	lastSentTime        time.Time
 
 	initNonBlockingChOnce sync.Once
 	startNonBlockCh       chan struct{}
+	wg                    *sync.WaitGroup
 }
 
-func NewPusher(target Target, service CloudWatchLogsService, flushTimeout time.Duration, retryDuration time.Duration, logger telegraf.Logger) *pusher {
+func NewPusher(target Target, service CloudWatchLogsService, flushTimeout time.Duration, retryDuration time.Duration, logger telegraf.Logger, stop <-chan struct{}, wg *sync.WaitGroup) *pusher {
 	p := &pusher{
 		Target:          target,
 		Service:         service,
@@ -67,10 +68,12 @@ func NewPusher(target Target, service CloudWatchLogsService, flushTimeout time.D
 		events:          make([]*cloudwatchlogs.InputLogEvent, 0, 10),
 		eventsCh:        make(chan logs.LogEvent, 100),
 		flushTimer:      time.NewTimer(flushTimeout),
-		stop:            make(chan struct{}),
+		stop:            stop,
 		startNonBlockCh: make(chan struct{}),
+		wg:              wg,
 	}
 	p.putRetentionPolicy()
+	p.wg.Add(1)
 	go p.start()
 	return p
 }
@@ -120,11 +123,9 @@ func hasValidTime(e logs.LogEvent) bool {
 	return true
 }
 
-func (p *pusher) Stop() {
-	close(p.stop)
-}
-
 func (p *pusher) start() {
+	defer p.wg.Done()
+
 	ec := make(chan logs.LogEvent)
 
 	// Merge events from both blocking and non-blocking channel
@@ -299,7 +300,15 @@ func (p *pusher) send() {
 		}
 
 		p.Log.Warnf("Retried %v time, going to sleep %v before retrying.", retryCount, wait)
-		time.Sleep(wait)
+
+		select {
+		case <-p.stop:
+			p.Log.Errorf("Stop requested after %v retries to %v/%v failed for PutLogEvents, request dropped.", retryCount, p.Group, p.Stream)
+			p.reset()
+			return
+		case <-time.After(wait):
+		}
+
 		retryCount++
 	}