MM-14617 Dependency upgrades and adding modules support. (#10517)

* Dependency upgrades and adding modules support.

* Commenting out file tests playload verification portion.

* Fixing viper.

* Fixing hclog.

vendor/github.com/go-redis/redis/Makefile

@@ -15,7 +15,7 @@ bench: testdeps
 
 testdata/redis:
 	mkdir -p $@
-	wget -qO- https://github.com/antirez/redis/archive/unstable.tar.gz | tar xvz --strip-components=1 -C $@
+	wget -qO- https://github.com/antirez/redis/archive/5.0.tar.gz | tar xvz --strip-components=1 -C $@
 
 testdata/redis/src/redis-server: testdata/redis
 	sed -i.bak 's/libjemalloc.a/libjemalloc.a -lrt/g' $</src/Makefile

vendor/github.com/go-redis/redis/cluster.go

@@ -772,6 +772,11 @@ func cmdSlot(cmd Cmder, pos int) int {
 }
 
 func (c *ClusterClient) cmdSlot(cmd Cmder) int {
+	args := cmd.Args()
+	if args[0] == "cluster" && args[1] == "getkeysinslot" {
+		return args[2].(int)
+	}
+
 	cmdInfo := c.cmdInfo(cmd.Name())
 	return cmdSlot(cmd, cmdFirstKeyPos(cmd, cmdInfo))
 }
@@ -783,7 +788,7 @@ func (c *ClusterClient) cmdSlotAndNode(cmd Cmder) (int, *clusterNode, error) {
 	}
 
 	cmdInfo := c.cmdInfo(cmd.Name())
-	slot := cmdSlot(cmd, cmdFirstKeyPos(cmd, cmdInfo))
+	slot := c.cmdSlot(cmd)
 
 	if c.opt.ReadOnly && cmdInfo != nil && cmdInfo.ReadOnly {
 		if c.opt.RouteByLatency {
@@ -844,15 +849,12 @@ func (c *ClusterClient) Watch(fn func(*Tx) error, keys ...string) error {
 		if err == nil {
 			break
 		}
-
-		if internal.IsRetryableError(err, true) {
+		if err != Nil {
 			c.state.LazyReload()
-			continue
 		}
 
 		moved, ask, addr := internal.IsMovedError(err)
 		if moved || ask {
-			c.state.LazyReload()
 			node, err = c.nodes.GetOrCreate(addr)
 			if err != nil {
 				return err
@@ -860,7 +862,7 @@ func (c *ClusterClient) Watch(fn func(*Tx) error, keys ...string) error {
 			continue
 		}
 
-		if err == pool.ErrClosed {
+		if err == pool.ErrClosed || internal.IsReadOnlyError(err) {
 			node, err = c.slotMasterNode(slot)
 			if err != nil {
 				return err
@@ -868,6 +870,10 @@ func (c *ClusterClient) Watch(fn func(*Tx) error, keys ...string) error {
 			continue
 		}
 
+		if internal.IsRetryableError(err, true) {
+			continue
+		}
+
 		return err
 	}
 
@@ -932,6 +938,9 @@ func (c *ClusterClient) defaultProcess(cmd Cmder) error {
 		if err == nil {
 			break
 		}
+		if err != Nil {
+			c.state.LazyReload()
+		}
 
 		// If slave is loading - pick another node.
 		if c.opt.ReadOnly && internal.IsLoadingError(err) {
@@ -940,9 +949,23 @@ func (c *ClusterClient) defaultProcess(cmd Cmder) error {
 			continue
 		}
 
-		if internal.IsRetryableError(err, true) {
-			c.state.LazyReload()
+		var moved bool
+		var addr string
+		moved, ask, addr = internal.IsMovedError(err)
+		if moved || ask {
+			node, err = c.nodes.GetOrCreate(addr)
+			if err != nil {
+				break
+			}
+			continue
+		}
 
+		if err == pool.ErrClosed || internal.IsReadOnlyError(err) {
+			node = nil
+			continue
+		}
+
+		if internal.IsRetryableError(err, true) {
 			// First retry the same node.
 			if attempt == 0 {
 				continue
@@ -956,24 +979,6 @@ func (c *ClusterClient) defaultProcess(cmd Cmder) error {
 			continue
 		}
 
-		var moved bool
-		var addr string
-		moved, ask, addr = internal.IsMovedError(err)
-		if moved || ask {
-			c.state.LazyReload()
-
-			node, err = c.nodes.GetOrCreate(addr)
-			if err != nil {
-				break
-			}
-			continue
-		}
-
-		if err == pool.ErrClosed {
-			node = nil
-			continue
-		}
-
 		break
 	}
 
@@ -1522,40 +1527,46 @@ func (c *ClusterClient) txPipelineReadQueued(
 	return nil
 }
 
-func (c *ClusterClient) pubSub(channels []string) *PubSub {
+func (c *ClusterClient) pubSub() *PubSub {
 	var node *clusterNode
 	pubsub := &PubSub{
 		opt: c.opt.clientOptions(),
 
 		newConn: func(channels []string) (*pool.Conn, error) {
-			if node == nil {
-				var slot int
-				if len(channels) > 0 {
-					slot = hashtag.Slot(channels[0])
-				} else {
-					slot = -1
-				}
-
-				masterNode, err := c.slotMasterNode(slot)
-				if err != nil {
-					return nil, err
-				}
-				node = masterNode
+			if node != nil {
+				panic("node != nil")
 			}
-			return node.Client.newConn()
+
+			slot := hashtag.Slot(channels[0])
+
+			var err error
+			node, err = c.slotMasterNode(slot)
+			if err != nil {
+				return nil, err
+			}
+
+			cn, err := node.Client.newConn()
+			if err != nil {
+				return nil, err
+			}
+
+			return cn, nil
 		},
 		closeConn: func(cn *pool.Conn) error {
-			return node.Client.connPool.CloseConn(cn)
+			err := node.Client.connPool.CloseConn(cn)
+			node = nil
+			return err
 		},
 	}
 	pubsub.init()
+
 	return pubsub
 }
 
 // Subscribe subscribes the client to the specified channels.
 // Channels can be omitted to create empty subscription.
 func (c *ClusterClient) Subscribe(channels ...string) *PubSub {
-	pubsub := c.pubSub(channels)
+	pubsub := c.pubSub()
 	if len(channels) > 0 {
 		_ = pubsub.Subscribe(channels...)
 	}
@@ -1565,7 +1576,7 @@ func (c *ClusterClient) Subscribe(channels ...string) *PubSub {
 // PSubscribe subscribes the client to the given patterns.
 // Patterns can be omitted to create empty subscription.
 func (c *ClusterClient) PSubscribe(channels ...string) *PubSub {
-	pubsub := c.pubSub(channels)
+	pubsub := c.pubSub()
 	if len(channels) > 0 {
 		_ = pubsub.PSubscribe(channels...)
 	}

vendor/github.com/go-redis/redis/commands.go

@@ -270,6 +270,7 @@ type Cmdable interface {
 	ClusterResetHard() *StatusCmd
 	ClusterInfo() *StringCmd
 	ClusterKeySlot(key string) *IntCmd
+	ClusterGetKeysInSlot(slot int, count int) *StringSliceCmd
 	ClusterCountFailureReports(nodeID string) *IntCmd
 	ClusterCountKeysInSlot(slot int) *IntCmd
 	ClusterDelSlots(slots ...int) *StatusCmd
@@ -1452,10 +1453,11 @@ func (c *cmdable) XGroupDelConsumer(stream, group, consumer string) *IntCmd {
 type XReadGroupArgs struct {
 	Group    string
 	Consumer string
-	Streams  []string
-	Count    int64
-	Block    time.Duration
-	NoAck    bool
+	// List of streams and ids.
+	Streams []string
+	Count   int64
+	Block   time.Duration
+	NoAck   bool
 }
 
 func (c *cmdable) XReadGroup(a *XReadGroupArgs) *XStreamSliceCmd {
@@ -2402,6 +2404,12 @@ func (c *cmdable) ClusterKeySlot(key string) *IntCmd {
 	return cmd
 }
 
+func (c *cmdable) ClusterGetKeysInSlot(slot int, count int) *StringSliceCmd {
+	cmd := NewStringSliceCmd("cluster", "getkeysinslot", slot, count)
+	c.process(cmd)
+	return cmd
+}
+
 func (c *cmdable) ClusterCountFailureReports(nodeID string) *IntCmd {
 	cmd := NewIntCmd("cluster", "count-failure-reports", nodeID)
 	c.process(cmd)

vendor/github.com/go-redis/redis/internal/error.go

@@ -47,7 +47,8 @@ func IsBadConn(err error, allowTimeout bool) bool {
 		return false
 	}
 	if IsRedisError(err) {
-		return strings.HasPrefix(err.Error(), "READONLY ")
+		// #790
+		return IsReadOnlyError(err)
 	}
 	if allowTimeout {
 		if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
@@ -82,3 +83,7 @@ func IsMovedError(err error) (moved bool, ask bool, addr string) {
 func IsLoadingError(err error) bool {
 	return strings.HasPrefix(err.Error(), "LOADING ")
 }
+
+func IsReadOnlyError(err error) bool {
+	return strings.HasPrefix(err.Error(), "READONLY ")
+}

vendor/github.com/go-redis/redis/options.go

@@ -101,6 +101,9 @@ func (opt *Options) init() {
 	if opt.Network == "" {
 		opt.Network = "tcp"
 	}
+	if opt.Addr == "" {
+		opt.Addr = "localhost:6379"
+	}
 	if opt.Dialer == nil {
 		opt.Dialer = func() (net.Conn, error) {
 			netDialer := &net.Dialer{

vendor/github.com/go-redis/redis/pipeline.go

@@ -10,6 +10,7 @@ type pipelineExecer func([]Cmder) error
 
 type Pipeliner interface {
 	StatefulCmdable
+	Do(args ...interface{}) *Cmd
 	Process(cmd Cmder) error
 	Close() error
 	Discard() error
@@ -31,6 +32,12 @@ type Pipeline struct {
 	closed bool
 }
 
+func (c *Pipeline) Do(args ...interface{}) *Cmd {
+	cmd := NewCmd(args...)
+	_ = c.Process(cmd)
+	return cmd
+}
+
 // Process queues the cmd for later execution.
 func (c *Pipeline) Process(cmd Cmder) error {
 	c.mu.Lock()

vendor/github.com/go-redis/redis/ring.go

@@ -319,12 +319,12 @@ func (c *ringShards) Close() error {
 
 //------------------------------------------------------------------------------
 
-// Ring is a Redis client that uses constistent hashing to distribute
+// Ring is a Redis client that uses consistent hashing to distribute
 // keys across multiple Redis servers (shards). It's safe for
 // concurrent use by multiple goroutines.
 //
 // Ring monitors the state of each shard and removes dead shards from
-// the ring. When shard comes online it is added back to the ring. This
+// the ring. When a shard comes online it is added back to the ring. This
 // gives you maximum availability and partition tolerance, but no
 // consistency between different shards or even clients. Each client
 // uses shards that are available to the client and does not do any

vendor/github.com/go-redis/redis/sentinel.go

@@ -164,6 +164,24 @@ func (c *SentinelClient) Sentinels(name string) *SliceCmd {
 	return cmd
 }
 
+// Failover forces a failover as if the master was not reachable, and without
+// asking for agreement to other Sentinels.
+func (c *SentinelClient) Failover(name string) *StatusCmd {
+	cmd := NewStatusCmd("sentinel", "failover", name)
+	c.Process(cmd)
+	return cmd
+}
+
+// Reset resets all the masters with matching name. The pattern argument is a
+// glob-style pattern. The reset process clears any previous state in a master
+// (including a failover in progress), and removes every slave and sentinel
+// already discovered and associated with the master.
+func (c *SentinelClient) Reset(pattern string) *IntCmd {
+	cmd := NewIntCmd("sentinel", "reset", pattern)
+	c.Process(cmd)
+	return cmd
+}
+
 type sentinelFailover struct {
 	sentinelAddrs []string
 

vendor/github.com/go-redis/redis/tx.go

@@ -29,10 +29,10 @@ func (c *Client) newTx() *Tx {
 	return &tx
 }
 
-// Watch prepares a transcaction and marks the keys to be watched
+// Watch prepares a transaction and marks the keys to be watched
 // for conditional execution if there are any keys.
 //
-// The transaction is automatically closed when the fn exits.
+// The transaction is automatically closed when fn exits.
 func (c *Client) Watch(fn func(*Tx) error, keys ...string) error {
 	tx := c.newTx()
 	if len(keys) > 0 {