Merge branch 'master' into flusher-stop

CHANGELOG.md

@@ -4,9 +4,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [0.13.5] - 2020-12-08
+## [0.13.5] - 2020-12-09
 ### Fixed
 - Issue where flushers would retry indefinitely
+- Issue where flushers would improperly reuse the same http request multiple times
 
 ## [0.13.4] - 2020-12-07
 ### Added

operator/builtin/output/elastic/elastic.go

@@ -196,8 +196,8 @@ func (e *ElasticOutput) feedFlusher(ctx context.Context) {
 			continue
 		}
 
-		req := e.createRequest(entries)
 		e.flusher.Do(func(ctx context.Context) error {
+			req := e.createRequest(entries)
 			res, err := req.Do(ctx, e.client)
 			if err != nil {
 				return errors.NewError(

operator/builtin/output/forward/forward.go

@@ -131,13 +131,17 @@ func (f *ForwardOutput) feedFlusher(ctx context.Context) {
 			continue
 		}
 
-		req, err := f.createRequest(ctx, entries)
-		if err != nil {
-			f.Errorf("Failed to create request", zap.Error(err))
-			continue
-		}
-
 		f.flusher.Do(func(ctx context.Context) error {
+			req, err := f.createRequest(ctx, entries)
+			if err != nil {
+				f.Errorf("Failed to create request", zap.Error(err))
+				// drop these logs because we couldn't creat a request and a retry won't help
+				if err := clearer.MarkAllAsFlushed(); err != nil {
+					f.Errorf("Failed to mark entries as flushed after failing to create a request", zap.Error(err))
+				}
+				return nil
+			}
+
 			res, err := f.client.Do(req)
 			if err != nil {
 				return errors.Wrap(err, "send request")

operator/builtin/output/googlecloud/google_cloud.go

@@ -222,8 +222,8 @@ func (g *GoogleCloudOutput) feedFlusher(ctx context.Context) {
 			continue
 		}
 
-		req := g.createWriteRequest(entries)
 		g.flusher.Do(func(ctx context.Context) error {
+			req := g.createWriteRequest(entries)
 			_, err := g.client.WriteLogEntries(ctx, req)
 			if err != nil {
 				return err

operator/builtin/output/newrelic/newrelic.go

@@ -180,13 +180,17 @@ func (nro *NewRelicOutput) feedFlusher(ctx context.Context) {
 			continue
 		}
 
-		req, err := nro.newRequest(ctx, entries)
-		if err != nil {
-			nro.Errorw("Failed to create request from payload", zap.Error(err))
-			continue
-		}
-
 		nro.flusher.Do(func(ctx context.Context) error {
+			req, err := nro.newRequest(ctx, entries)
+			if err != nil {
+				nro.Errorw("Failed to create request from payload", zap.Error(err))
+				// drop these logs because we couldn't creat a request and a retry won't help
+				if err := clearer.MarkAllAsFlushed(); err != nil {
+					nro.Errorf("Failed to mark entries as flushed after failing to create a request", zap.Error(err))
+				}
+				return nil
+			}
+
 			res, err := nro.client.Do(req)
 			if err != nil {
 				return err

operator/builtin/output/otlp/otlp.go

@@ -127,8 +127,16 @@ func (o *OTLPOutput) feedFlusher(ctx context.Context) {
 			continue
 		}
 
-		req, err := o.createRequest(ctx, entries)
 		o.flusher.Do(func(ctx context.Context) error {
+			req, err := o.createRequest(ctx, entries)
+			if err != nil {
+				o.Errorf("Failed to create request", zap.Error(err))
+				// drop these logs because we couldn't creat a request and a retry won't help
+				if err := clearer.MarkAllAsFlushed(); err != nil {
+					o.Errorf("Failed to mark entries as flushed after failing to create a request", zap.Error(err))
+				}
+				return nil
+			}
 			res, err := o.client.Do(req)
 			if err != nil {
 				return errors.Wrap(err, "send request")