Libbeat output reloading cause a goroutine leak and a possibly a fd leak

[ph]

Central management relies on the reloading implementation inside libbeat, when an input, module or output is found in the incoming configurations from CM, CM will take care of reloading the appropriate part of the code.

Lets take the following scenario:

Filebeat is configured with a simple input listening to events from the /var/log/syslog path and send events to Elasticsearch.

Now assume that the configuration changes every second when we reload the output, we have to migrate in flight events in the queue to the new output, when a new Elasticsearch output is created the first thing it will do is to ping the Elasticsearch host.

The current problem is when we swap the output, we keep a goroutine of the previous output and also keep a reference to the TCP Transport pool created for the output, in short we leak a goroutine and a fd on every reload of the output.

[ph]

Adding pprof goroutine trace to identify reports.

(pprof) list netClientWorker
Total: 3089
ROUTINE ======================== github.com/elastic/beats/libbeat/publisher/pipeline.(*netClientWorker).run in /Users/ph/go/src/github.com/elastic/beats/libbeat/publisher/pipeline/output.go
         0       2178 (flat, cum) 70.51% of Total
         .          .     79:   for !w.closed.Load() {
         .          .     80:           reconnectAttempts := 0
         .          .     81:
         .          .     82:           // start initial connect loop from first batch, but return
         .          .     83:           // batch to pipeline for other outputs to catch up while we're trying to connect
         .       1738     84:           for batch := range w.qu {
         .          .     85:                   batch.Cancelled()
         .          .     86:
         .          .     87:                   if w.closed.Load() {
         .          .     88:                           logp.Info("Closed connection to %v", w.client)
         .          .     89:                           return
         .          .     90:                   }
         .          .     91:
         .          .     92:                   if reconnectAttempts > 0 {
         .          .     93:                           logp.Info("Attempting to reconnect to %v with %d reconnect attempt(s)", w.client, reconnectAttempts)
         .          .     94:                   } else {
         .          .     95:                           logp.Info("Connecting to %v", w.client)
         .          .     96:                   }
         .          .     97:
         .          .     98:                   err := w.client.Connect()
         .          .     99:                   if err != nil {
         .          .    100:                           logp.Err("Failed to connect to %v: %v", w.client, err)
         .          .    101:                           reconnectAttempts++
         .          .    102:                           continue
         .          .    103:                   }
         .          .    104:
         .          .    105:                   logp.Info("Connection to %v established", w.client)
         .          .    106:                   reconnectAttempts = 0
         .          .    107:                   break
         .          .    108:           }
         .          .    109:
         .          .    110:           // send loop
         .        440    111:           for batch := range w.qu {
         .          .    112:                   if w.closed.Load() {
         .          .    113:                           if batch != nil {
         .          .    114:                                   batch.Cancelled()
         .          .    115:                           }
         .          .    116:                           return
(pprof)

[ph]

Looking at reloadable part this doesn't look so bad to fix.

[urso]

Initial attempt to fix this issue in #10599. We did run into issues trying to fix this issue, ultimately creating #11068 as a prerequisite.