Cherry-pick #6404 to 6.4: Log reconnect attempts (#8146)

Cherry-pick of PR #6404 to 6.4 branch. Original message: 

Closes #5175 

Logs an info line "Attempting to reconnect" at every reconnect attempt.

libbeat/monitoring/report/elasticsearch/client.go

@@ -152,3 +152,7 @@ func (c *publishClient) Publish(batch publisher.Batch) error {
 func (c *publishClient) Test(d testing.Driver) {
 	c.es.Test(d)
 }
+
+func (c *publishClient) String() string {
+	return "publish(" + c.es.String() + ")"
+}

libbeat/outputs/backoff.go

@@ -77,3 +77,7 @@ func (b *backoffClient) Test(d testing.Driver) {
 
 	c.Test(d)
 }
+
+func (b *backoffClient) String() string {
+	return "backoff(" + b.client.String() + ")"
+}

libbeat/outputs/console/console.go

@@ -160,3 +160,7 @@ func (c *console) writeBuffer(buf []byte) error {
 	}
 	return nil
 }
+
+func (c *console) String() string {
+	return "console"
+}

libbeat/outputs/elasticsearch/client.go

@@ -664,6 +664,10 @@ func (client *Client) Test(d testing.Driver) {
 	})
 }
 
+func (client *Client) String() string {
+	return "elasticsearch(" + client.Connection.URL + ")"
+}
+
 // Connect connects the client.
 func (conn *Connection) Connect() error {
 	var err error

libbeat/outputs/failover.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"math/rand"
+	"strings"
 
 	"github.com/elastic/beats/libbeat/publisher"
 	"github.com/elastic/beats/libbeat/testing"
@@ -109,3 +110,13 @@ func (f *failoverClient) Test(d testing.Driver) {
 		})
 	}
 }
+
+func (f *failoverClient) String() string {
+	names := make([]string, len(f.clients))
+
+	for i, client := range f.clients {
+		names[i] = client.String()
+	}
+
+	return "failover(" + strings.Join(names, ",") + ")"
+}

libbeat/outputs/fileout/file.go

@@ -35,6 +35,7 @@ func init() {
 }
 
 type fileOutput struct {
+	filePath string
 	beat     beat.Info
 	observer outputs.Observer
 	rotator  *file.Rotator
@@ -74,6 +75,8 @@ func (out *fileOutput) init(beat beat.Info, c config) error {
 		path = filepath.Join(c.Path, out.beat.Beat)
 	}
 
+	out.filePath = path
+
 	var err error
 	out.rotator, err = file.NewFileRotator(
 		path,
@@ -149,3 +152,7 @@ func (out *fileOutput) Publish(
 
 	return nil
 }
+
+func (out *fileOutput) String() string {
+	return "file(" + out.filePath + ")"
+}

libbeat/outputs/kafka/client.go

@@ -20,6 +20,7 @@ package kafka
 import (
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"sync/atomic"
 
@@ -141,6 +142,10 @@ func (c *client) Publish(batch publisher.Batch) error {
 	return nil
 }
 
+func (c *client) String() string {
+	return "kafka(" + strings.Join(c.hosts, ",") + ")"
+}
+
 func (c *client) getEventMessage(data *publisher.Event) (*message, error) {
 	event := &data.Content
 	msg := &message{partition: -1, data: *data}

libbeat/outputs/logstash/async.go

@@ -169,6 +169,10 @@ func (c *asyncClient) Publish(batch publisher.Batch) error {
 	return nil
 }
 
+func (c *asyncClient) String() string {
+	return "async(" + c.Client.String() + ")"
+}
+
 func (c *asyncClient) publishWindowed(
 	ref *msgRef,
 	events []publisher.Event,

libbeat/outputs/outputs.go

@@ -35,6 +35,8 @@ type Client interface {
 	// the publisher pipeline. The publisher pipeline (if configured by the output
 	// factory) will take care of retrying/dropping events.
 	Publish(publisher.Batch) error
+
+	String() string
 }
 
 // NetworkClient defines the required client capabilities for network based

libbeat/outputs/redis/backoff.go

@@ -112,3 +112,7 @@ func (b *backoffClient) resetFail() {
 	b.reason = failNone
 	b.backoff.Reset()
 }
+
+func (b *backoffClient) String() string {
+	return b.client.String()
+}

libbeat/outputs/redis/client.go

@@ -152,6 +152,10 @@ func (c *client) Publish(batch publisher.Batch) error {
 	return err
 }
 
+func (c *client) String() string {
+	return "redis(" + c.Client.String() + ")"
+}
+
 func (c *client) makePublish(
 	conn redis.Conn,
 ) (publishFn, error) {

libbeat/outputs/transport/client.go

@@ -247,3 +247,7 @@ func (c *Client) Test(d testing.Driver) {
 		d.Fatal("talk to server", err)
 	})
 }
+
+func (c *Client) String() string {
+	return c.network + "://" + c.host
+}

libbeat/publisher/pipeline/output.go

@@ -77,21 +77,33 @@ func (w *netClientWorker) Close() error {
 
 func (w *netClientWorker) run() {
 	for !w.closed.Load() {
+		reconnectAttempts := 0
+
 		// start initial connect loop from first batch, but return
 		// batch to pipeline for other outputs to catch up while we're trying to connect
 		for batch := range w.qu {
 			batch.Cancelled()
 
 			if w.closed.Load() {
+				logp.Info("Closed connection to %v", w.client)
 				return
 			}
 
+			if reconnectAttempts > 0 {
+				logp.Info("Attempting to reconnect to %v with %d reconnect attempt(s)", w.client, reconnectAttempts)
+			} else {
+				logp.Info("Connecting to %v", w.client)
+			}
+
 			err := w.client.Connect()
 			if err != nil {
-				logp.Err("Failed to connect: %v", err)
+				logp.Err("Failed to connect to %v: %v", w.client, err)
+				reconnectAttempts++
 				continue
 			}
 
+			logp.Info("Connection to %v established", w.client)
+			reconnectAttempts = 0
 			break
 		}
 

libbeat/publisher/pipeline/stress/out.go

@@ -106,3 +106,7 @@ func (t *testOutput) Publish(batch publisher.Batch) error {
 
 	return nil
 }
+
+func (t *testOutput) String() string {
+	return "test"
+}