diff --git a/redis/error.go b/redis/error.go index 8442ce2..c5b71ed 100644 --- a/redis/error.go +++ b/redis/error.go @@ -73,6 +73,8 @@ var ( ErrClusterDown = ErrResult.NewSubtype("clusterdown", ErrTraitNotSent) // ErrLoading - redis didn't finish start ErrLoading = ErrResult.NewSubtype("loading", ErrTraitNotSent) + // ErrMasterDown - MASTERDOWN response: replica lost its master and replica-serve-stale-data is 'no' + ErrMasterDown = ErrResult.NewSubtype("masterdown", ErrTraitNotSent) // ErrExecEmpty - EXEC returns nil (WATCH failed) (it is strange, cause we don't support WATCH) ErrExecEmpty = ErrResult.NewSubtype("exec_empty") // ErrExecAbort - EXEC returns EXECABORT diff --git a/redis/reader.go b/redis/reader.go index ac05ffd..9f9bb01 100644 --- a/redis/reader.go +++ b/redis/reader.go @@ -54,6 +54,9 @@ func ReadResponse(b *bufio.Reader) (interface{}, int) { if strings.HasPrefix(txt, "LOADING") { return ErrLoading.New(txt), len(line) } + if strings.HasPrefix(txt, "MASTERDOWN") { + return ErrMasterDown.New(txt), len(line) + } if strings.HasPrefix(txt, "EXECABORT") { return ErrExecAbort.New(txt), len(line) } diff --git a/redis/reader_test.go b/redis/reader_test.go index 18b1761..87ff387 100644 --- a/redis/reader_test.go +++ b/redis/reader_test.go @@ -145,6 +145,12 @@ func TestReadResponse_Correct(t *testing.T) { assert.Equal(t, "LOADING", err.Message()) } + res = readLines("-MASTERDOWN Link with MASTER is down and replica-serve-stale-data is set to 'no'.\r\n") + if checkErrType(t, res, ErrMasterDown) { + err := res.(*errorx.Error) + assert.True(t, err.HasTrait(ErrTraitNotSent)) + } + for i := -1000; i <= 1000; i++ { res = readLines(fmt.Sprintf(":%d\r\n", i)) assert.Equal(t, int64(i), res) diff --git a/rediscluster/cluster.go b/rediscluster/cluster.go index 8280db2..6a86b9f 100644 --- a/rediscluster/cluster.go +++ b/rediscluster/cluster.go @@ -50,7 +50,13 @@ const ( const ( defaultCheckInterval = 5 * time.Second - defaultWaitToMigrate = 20 * time.Millisecond + // A failover takes cluster-node-timeout plus an election of a second or two, and the + // client notices the new master one CheckInterval later; 60 seconds cover the default + // 15-second node timeout several times over and stay well below the ~160 seconds + // (repl-ping-replica-period + node-timeout × cluster-replica-validity-factor) after + // which redis itself stops considering the replica fresh enough for promotion. + defaultReplicaLinkDownTolerance = 60 * time.Second + defaultWaitToMigrate = 20 * time.Millisecond forceInterval = 100 * time.Millisecond @@ -112,6 +118,11 @@ type Opts struct { ForceMinLatencyReplica bool // WeightProvider - enables to explicitly set weights of replicas (has higher priority than LatencyOrientedRR) WeightProvider WeightProvider + // ReplicaLinkDownTolerance - a replica whose link with its master is down keeps receiving + // reads while its master_link_down_since_seconds stays below this value. A replica that + // has never synced since it started is never read regardless of it. + // default: 60 seconds; negative: a replica with a broken link is never read + ReplicaLinkDownTolerance time.Duration // Enable connection with TLS TLSEnabled bool // Config for TLS connection @@ -241,6 +252,10 @@ func NewCluster(ctx context.Context, initAddrs []string, opts Opts) (*Cluster, e cluster.opts.WaitToMigrate = 100 * time.Millisecond } + if cluster.opts.ReplicaLinkDownTolerance == 0 { + cluster.opts.ReplicaLinkDownTolerance = defaultReplicaLinkDownTolerance + } + cluster.latencyAwareness = disabled if cluster.opts.LatencyOrientedRR { cluster.latencyAwareness = enabled diff --git a/rediscluster/replica_health_internal_test.go b/rediscluster/replica_health_internal_test.go new file mode 100644 index 0000000..076fc8b --- /dev/null +++ b/rediscluster/replica_health_internal_test.go @@ -0,0 +1,55 @@ +package rediscluster + +import ( + "strings" + "testing" + "time" +) + +func TestReplicaHealthy(t *testing.T) { + info := func(lines ...string) []byte { + return []byte("# Replication\r\nrole:slave\r\n" + strings.Join(lines, "\r\n") + "\r\n# Persistence\r\nloading:0\r\n") + } + for _, tc := range []struct { + name string + info []byte + linkDownTolerance time.Duration + want bool + }{ + {name: "link up", info: info("master_link_status:up"), linkDownTolerance: time.Minute, want: true}, + {name: "link down for a while", info: info("master_link_status:down", "master_link_down_since_seconds:3"), linkDownTolerance: time.Minute, want: true}, + {name: "link down for the whole linkDownTolerance", info: info("master_link_status:down", "master_link_down_since_seconds:60"), linkDownTolerance: time.Minute, want: false}, + {name: "never synced since start", info: info("master_link_status:down", "master_link_down_since_seconds:-1"), linkDownTolerance: time.Minute, want: false}, + {name: "link down without duration", info: info("master_link_status:down"), linkDownTolerance: time.Minute, want: false}, + {name: "negative tolerance", info: info("master_link_status:down", "master_link_down_since_seconds:0"), linkDownTolerance: -1, want: false}, + {name: "loading", info: []byte("# Replication\r\nmaster_link_status:up\r\n# Persistence\r\nloading:1\r\n"), linkDownTolerance: time.Minute, want: false}, + {name: "async loading is not loading", info: []byte("# Replication\r\nmaster_link_status:up\r\n# Persistence\r\nloading:0\r\nasync_loading:1\r\n"), linkDownTolerance: time.Minute, want: true}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := replicaHealthy(tc.info, tc.linkDownTolerance); got != tc.want { + t.Errorf("replicaHealthy(%q, %v) = %v, want %v", tc.info, tc.linkDownTolerance, got, tc.want) + } + }) + } +} + +func TestInfoField(t *testing.T) { + info := []byte("loading:0\r\nasync_loading:1\r\nmaster_link_status:down\r\nmaster_link_down_since_seconds:-1\r\n") + for _, tc := range []struct { + field string + want string + ok bool + }{ + {field: "loading", want: "0", ok: true}, + {field: "async_loading", want: "1", ok: true}, + {field: "master_link_status", want: "down", ok: true}, + {field: "master_link_down_since_seconds", want: "-1", ok: true}, + {field: "link_status", ok: false}, + {field: "role", ok: false}, + } { + got, ok := infoField(info, tc.field) + if ok != tc.ok || string(got) != tc.want { + t.Errorf("infoField(%q) = %q, %v; want %q, %v", tc.field, got, ok, tc.want, tc.ok) + } + } +} diff --git a/rediscluster/slotrange.go b/rediscluster/slotrange.go index 359a597..b60de2c 100644 --- a/rediscluster/slotrange.go +++ b/rediscluster/slotrange.go @@ -3,6 +3,7 @@ package rediscluster import ( "bytes" "math" + "strconv" "sync/atomic" "time" @@ -142,7 +143,7 @@ func (c *Cluster) updateMappings(slotRanges []redisclusterutil.SlotsRange) { c.nodeWait.promises = make(map[string]*[]connThen, 1) c.nodeWait.Unlock() - go newConfig.setConnRoles() + go newConfig.setConnRoles(c.opts.ReplicaLinkDownTolerance) var sh uint32 for i := 0; i < redisclusterutil.NumSlots; i++ { @@ -213,7 +214,13 @@ func (c *Cluster) updateMappings(slotRanges []redisclusterutil.SlotsRange) { }) } -func (s *shard) setReplicaInfo(res interface{}, n uint64) { +func (s *shard) replicaInfoFuture(linkDownTolerance time.Duration) redis.FuncFuture { + return func(res interface{}, n uint64) { + s.setReplicaInfo(res, n, linkDownTolerance) + } +} + +func (s *shard) setReplicaInfo(res interface{}, n uint64, linkDownTolerance time.Duration) { haserr := false if err := redis.AsError(res); err != nil { haserr = true @@ -222,8 +229,8 @@ func (s *shard) setReplicaInfo(res interface{}, n uint64) { haserr = !(ok && str == "OK") } else if buf, ok := res.([]byte); !ok { haserr = true - } else if bytes.Contains(buf, []byte("master_link_status:down")) || bytes.Contains(buf, []byte("loading:1")) { - haserr = true + } else { + haserr = !replicaHealthy(buf, linkDownTolerance) } for { oldstate := atomic.LoadUint32(&s.good) @@ -242,7 +249,59 @@ func (s *shard) setReplicaInfo(res interface{}, n uint64) { } } -func (cfg *clusterConfig) setConnRoles() { +// replicaHealthy tells whether INFO output describes a replica worth reading from. +// master_link_down_since_seconds is -1 for a replica that has never synced since it +// started, and its dataset is then anything from empty to the RDB it booted from. +func replicaHealthy(info []byte, linkDownTolerance time.Duration) bool { + if infoHas(info, "loading", "1") { + return false + } + if !infoHas(info, "master_link_status", "down") { + return true + } + since, ok := infoInt(info, "master_link_down_since_seconds") + if !ok || since < 0 { + return false + } + return time.Duration(since)*time.Second < linkDownTolerance +} + +// infoField returns the value of a named INFO field. The name is matched as a whole +// line prefix: `loading` must not be found inside `async_loading`. +func infoField(info []byte, field string) ([]byte, bool) { + key := []byte(field + ":") + for off := 0; ; { + i := bytes.Index(info[off:], key) + if i < 0 { + return nil, false + } + i += off + if i == 0 || info[i-1] == '\n' { + value := info[i+len(key):] + if end := bytes.IndexAny(value, "\r\n"); end >= 0 { + value = value[:end] + } + return value, true + } + off = i + 1 + } +} + +func infoHas(info []byte, field, value string) bool { + got, ok := infoField(info, field) + return ok && string(got) == value +} + +func infoInt(info []byte, field string) (int64, bool) { + value, ok := infoField(info, field) + if !ok { + return 0, false + } + v, err := strconv.ParseInt(string(value), 10, 64) + return v, err == nil +} + +func (cfg *clusterConfig) setConnRoles(linkDownTolerance time.Duration) { for _, sh := range cfg.shards { for i, addr := range sh.addr { node := cfg.nodes[addr] @@ -254,7 +313,7 @@ func (cfg *clusterConfig) setConnRoles() { conn.Send(Request{"READWRITE", nil}, nil, 0) } else { conn.SendBatch([]Request{{"READONLY", nil}, {"INFO", nil}}, - redis.FuncFuture(sh.setReplicaInfo), uint64(i*2)) + sh.replicaInfoFuture(linkDownTolerance), uint64(i*2)) } } }