Fix issues with output reloading

CHANGELOG.next.asciidoc

@@ -75,6 +75,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix bug with `monitoring.cluster_uuid` setting not always being exposed via GET /state Beats API. {issue}16732[16732] {pull}17420[17420]
 - Fix building on FreeBSD by removing build flags from `add_cloudfoundry_metadata` processor. {pull}17486[17486]
 - Do not rotate log files on startup when interval is configured and rotateonstartup is disabled. {pull}17613[17613]
+- Fix goroutine leak and Elasticsearch output file descriptor leak when output reloading is in use. {issue}10491[10491] {pull}17381[17381]
 
 *Auditbeat*
 

libbeat/esleg/eslegclient/connection.go

@@ -63,11 +63,15 @@ type ConnectionSettings struct {
 	Parameters       map[string]string
 	CompressionLevel int
 	EscapeHTML       bool
-	Timeout          time.Duration
+
+	Timeout         time.Duration
+	IdleConnTimeout time.Duration
 }
 
 // NewConnection returns a new Elasticsearch client
 func NewConnection(s ConnectionSettings) (*Connection, error) {
+	s = settingsWithDefaults(s)
+
 	u, err := url.Parse(s.URL)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse elasticsearch URL: %v", err)
@@ -124,6 +128,7 @@ func NewConnection(s ConnectionSettings) (*Connection, error) {
 				DialTLS:         tlsDialer.Dial,
 				TLSClientConfig: s.TLS.ToConfig(),
 				Proxy:           proxy,
+				IdleConnTimeout: s.IdleConnTimeout,
 			},
 			Timeout: s.Timeout,
 		},
@@ -132,6 +137,15 @@ func NewConnection(s ConnectionSettings) (*Connection, error) {
 	}, nil
 }
 
+func settingsWithDefaults(s ConnectionSettings) ConnectionSettings {
+	settings := s
+	if settings.IdleConnTimeout == 0 {
+		settings.IdleConnTimeout = 1 * time.Minute
+	}
+
+	return settings
+}
+
 // NewClients returns a list of Elasticsearch clients based on the given
 // configuration. It accepts the same configuration parameters as the Elasticsearch
 // output, except for the output specific configuration options.  If multiple hosts
@@ -266,6 +280,7 @@ func (conn *Connection) Ping() (string, error) {
 
 // Close closes a connection.
 func (conn *Connection) Close() error {
+	conn.HTTP.CloseIdleConnections()
 	return nil
 }
 

libbeat/publisher/pipeline/batch.go

@@ -24,7 +24,13 @@ import (
 	"github.com/elastic/beats/v7/libbeat/publisher/queue"
 )
 
-type Batch struct {
+type Batch interface {
+	publisher.Batch
+
+	reduceTTL() bool
+}
+
+type batch struct {
 	original queue.Batch
 	ctx      *batchContext
 	ttl      int
@@ -38,17 +44,17 @@ type batchContext struct {
 
 var batchPool = sync.Pool{
 	New: func() interface{} {
-		return &Batch{}
+		return &batch{}
 	},
 }
 
-func newBatch(ctx *batchContext, original queue.Batch, ttl int) *Batch {
+func newBatch(ctx *batchContext, original queue.Batch, ttl int) *batch {
 	if original == nil {
 		panic("empty batch")
 	}
 
-	b := batchPool.Get().(*Batch)
-	*b = Batch{
+	b := batchPool.Get().(*batch)
+	*b = batch{
 		original: original,
 		ctx:      ctx,
 		ttl:      ttl,
@@ -57,45 +63,47 @@ func newBatch(ctx *batchContext, original queue.Batch, ttl int) *Batch {
 	return b
 }
 
-func releaseBatch(b *Batch) {
-	*b = Batch{} // clear batch
+func releaseBatch(b *batch) {
+	*b = batch{} // clear batch
 	batchPool.Put(b)
 }
 
-func (b *Batch) Events() []publisher.Event {
+func (b *batch) Events() []publisher.Event {
 	return b.events
 }
 
-func (b *Batch) ACK() {
-	b.ctx.observer.outBatchACKed(len(b.events))
+func (b *batch) ACK() {
+	if b.ctx != nil {
+		b.ctx.observer.outBatchACKed(len(b.events))
+	}
 	b.original.ACK()
 	releaseBatch(b)
 }
 
-func (b *Batch) Drop() {
+func (b *batch) Drop() {
 	b.original.ACK()
 	releaseBatch(b)
 }
 
-func (b *Batch) Retry() {
+func (b *batch) Retry() {
 	b.ctx.retryer.retry(b)
 }
 
-func (b *Batch) Cancelled() {
+func (b *batch) Cancelled() {
 	b.ctx.retryer.cancelled(b)
 }
 
-func (b *Batch) RetryEvents(events []publisher.Event) {
+func (b *batch) RetryEvents(events []publisher.Event) {
 	b.updEvents(events)
 	b.Retry()
 }
 
-func (b *Batch) CancelledEvents(events []publisher.Event) {
+func (b *batch) CancelledEvents(events []publisher.Event) {
 	b.updEvents(events)
 	b.Cancelled()
 }
 
-func (b *Batch) updEvents(events []publisher.Event) {
+func (b *batch) updEvents(events []publisher.Event) {
 	l1 := len(b.events)
 	l2 := len(events)
 	if l1 > l2 {
@@ -105,3 +113,33 @@ func (b *Batch) updEvents(events []publisher.Event) {
 
 	b.events = events
 }
+
+// reduceTTL reduces the time to live for all events that have no 'guaranteed'
+// sending requirements.  reduceTTL returns true if the batch is still alive.
+func (b *batch) reduceTTL() bool {
+	if b.ttl <= 0 {
+		return true
+	}
+
+	b.ttl--
+	if b.ttl > 0 {
+		return true
+	}
+
+	// filter for evens with guaranteed send flags
+	events := b.events[:0]
+	for _, event := range b.events {
+		if event.Guaranteed() {
+			events = append(events, event)
+		}
+	}
+	b.events = events
+
+	if len(b.events) > 0 {
+		b.ttl = -1 // we need infinite retry for all events left in this batch
+		return true
+	}
+
+	// all events have been dropped:
+	return false
+}

libbeat/publisher/pipeline/consumer.go

@@ -138,7 +138,7 @@ func (c *eventConsumer) loop(consumer queue.Consumer) {
 
 	var (
 		out    workQueue
-		batch  *Batch
+		batch  Batch
 		paused = true
 	)
 
@@ -154,7 +154,7 @@ func (c *eventConsumer) loop(consumer queue.Consumer) {
 		}
 
 		paused = c.paused()
-		if !paused && c.out != nil && batch != nil {
+		if c.out != nil && batch != nil {
 			out = c.out.workQueue
 		} else {
 			out = nil
@@ -195,6 +195,9 @@ func (c *eventConsumer) loop(consumer queue.Consumer) {
 			handleSignal(sig)
 		case out <- batch:
 			batch = nil
+			if paused {
+				out = nil
+			}
 		}
 	}
 }

libbeat/publisher/pipeline/controller.go

@@ -22,6 +22,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/common/reload"
 	"github.com/elastic/beats/v7/libbeat/outputs"
+	"github.com/elastic/beats/v7/libbeat/publisher"
 	"github.com/elastic/beats/v7/libbeat/publisher/queue"
 )
 
@@ -34,7 +35,8 @@ type outputController struct {
 	monitors Monitors
 	observer outputObserver
 
-	queue queue.Queue
+	queue     queue.Queue
+	workQueue workQueue
 
 	retryer  *retryer
 	consumer *eventConsumer
@@ -50,7 +52,7 @@ type outputGroup struct {
 	timeToLive int // event lifetime
 }
 
-type workQueue chan *Batch
+type workQueue chan publisher.Batch
 
 // outputWorker instances pass events from the shared workQueue to the outputs.Client
 // instances.
@@ -62,18 +64,19 @@ func newOutputController(
 	beat beat.Info,
 	monitors Monitors,
 	observer outputObserver,
-	b queue.Queue,
+	queue queue.Queue,
 ) *outputController {
 	c := &outputController{
-		beat:     beat,
-		monitors: monitors,
-		observer: observer,
-		queue:    b,
+		beat:      beat,
+		monitors:  monitors,
+		observer:  observer,
+		queue:     queue,
+		workQueue: makeWorkQueue(),
 	}
 
 	ctx := &batchContext{}
-	c.consumer = newEventConsumer(monitors.Logger, b, ctx)
-	c.retryer = newRetryer(monitors.Logger, observer, nil, c.consumer)
+	c.consumer = newEventConsumer(monitors.Logger, queue, ctx)
+	c.retryer = newRetryer(monitors.Logger, observer, c.workQueue, c.consumer)
 	ctx.observer = observer
 	ctx.retryer = c.retryer
 
@@ -86,27 +89,26 @@ func (c *outputController) Close() error {
 	c.consumer.sigPause()
 	c.consumer.close()
 	c.retryer.close()
+	close(c.workQueue)
 
 	if c.out != nil {
 		for _, out := range c.out.outputs {
 			out.Close()
 		}
-		close(c.out.workQueue)
 	}
 
 	return nil
 }
 
 func (c *outputController) Set(outGrp outputs.Group) {
-	// create new outputGroup with shared work queue
+	// create new output group with the shared work queue
 	clients := outGrp.Clients
-	queue := makeWorkQueue()
 	worker := make([]outputWorker, len(clients))
 	for i, client := range clients {
-		worker[i] = makeClientWorker(c.observer, queue, client)
+		worker[i] = makeClientWorker(c.observer, c.workQueue, client)
 	}
 	grp := &outputGroup{
-		workQueue:  queue,
+		workQueue:  c.workQueue,
 		outputs:    worker,
 		timeToLive: outGrp.Retry + 1,
 		batchSize:  outGrp.BatchSize,
@@ -119,7 +121,6 @@ func (c *outputController) Set(outGrp outputs.Group) {
 			c.retryer.sigOutputRemoved()
 		}
 	}
-	c.retryer.updOutput(queue)
 	for range clients {
 		c.retryer.sigOutputAdded()
 	}
@@ -136,12 +137,13 @@ func (c *outputController) Set(outGrp outputs.Group) {
 
 	// restart consumer (potentially blocked by retryer)
 	c.consumer.sigContinue()
+	c.consumer.sigUnWait()
 
 	c.observer.updateOutputGroup()
 }
 
 func makeWorkQueue() workQueue {
-	return workQueue(make(chan *Batch, 0))
+	return workQueue(make(chan publisher.Batch, 0))
 }
 
 // Reload the output