Using specific pass for each host in redis output

libbeat/outputs/redis/redis.go

@@ -19,6 +19,10 @@ package redis
 
 import (
 	"errors"
+	"fmt"
+	"net/url"
+	"strconv"
+	"strings"
 	"time"
 
 	"github.com/elastic/beats/libbeat/beat"
@@ -42,6 +46,8 @@ const (
 	defaultWaitRetry    = 1 * time.Second
 	defaultMaxWaitRetry = 60 * time.Second
 	defaultPort         = 6379
+	redisScheme         = "redis"
+	tlsRedisScheme      = "rediss"
 )
 
 func init() {
@@ -118,19 +124,57 @@ func makeRedis(
 	}
 
 	clients := make([]outputs.NetworkClient, len(hosts))
-	for i, host := range hosts {
-		enc, err := codec.CreateEncoder(beat, config.Codec)
+	for i, h := range hosts {
+		if parts := strings.SplitN(h, "://", 2); len(parts) != 2 {
+			h = fmt.Sprintf("%s://%s", redisScheme, h)
+		}
+
+		hostUrl, err := url.Parse(h)
+		if err != nil {
+			return outputs.Fail(err)
+		}
+
+		if hostUrl.Host == "" {
+			return outputs.Fail(fmt.Errorf("invalid redis url host %s", hostUrl.Host))
+		}
+
+		if hostUrl.Scheme != redisScheme && hostUrl.Scheme != tlsRedisScheme {
+			return outputs.Fail(fmt.Errorf("invalid redis url scheme %s", hostUrl.Scheme))
+		}
+
+		if hostUrl.Scheme == tlsRedisScheme && transp.TLS == nil {
+			transp.TLS = &transport.TLSConfig{}
+		}
+
+		var host string
+		var port int
+		if hostUrl.Port() != "" {
+			parts := strings.SplitN(hostUrl.Host, ":", 2)
+			host = parts[0]
+			port, _ = strconv.Atoi(parts[1])
+		} else {
+			host = strings.ReplaceAll(hostUrl.Host, ":", "")
+			port = defaultPort
+		}
+
+		conn, err := transport.NewClient(transp, "tcp", host, port)
 		if err != nil {
 			return outputs.Fail(err)
 		}
 
-		conn, err := transport.NewClient(transp, "tcp", host, defaultPort)
+		pass := config.Password
+		hostPass, passSet := hostUrl.User.Password()
+		if passSet {
+			pass = hostPass
+		}
+
+		enc, err := codec.CreateEncoder(beat, config.Codec)
 		if err != nil {
 			return outputs.Fail(err)
 		}
 
 		client := newClient(conn, observer, config.Timeout,
-			config.Password, config.Db, key, dataType, config.Index, enc)
+			pass, config.Db, key, dataType, config.Index, enc)
 		clients[i] = newBackoffClient(client, config.Backoff.Init, config.Backoff.Max)
 	}