diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7d10318..020e117 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -2,15 +2,31 @@ name: CI on: [push] +concurrency: + group: ci-${{ github.ref }} + cancel-in-progress: true + jobs: build: runs-on: ubuntu-latest + timeout-minutes: 25 strategy: + fail-fast: false matrix: - go: ["1.19", "1.20", "1.21", "1.22", "1.23", "1.24"] + go: ["1.19", "1.20", "1.21", "1.22", "1.23", "1.24", "1.25", "1.26", "1.27"] stage: [testredis, testconn, testcluster] + exclude: + # Cluster suite spawns a 7-node redis cluster and takes ~7 minutes per + # matrix cell, while its behaviour does not depend on the go version. + - {go: "1.20", stage: testcluster} + - {go: "1.21", stage: testcluster} + - {go: "1.22", stage: testcluster} + - {go: "1.23", stage: testcluster} + - {go: "1.24", stage: testcluster} + - {go: "1.25", stage: testcluster} + - {go: "1.26", stage: testcluster} steps: @@ -43,4 +59,3 @@ jobs: - name: Build run: make ${{ matrix.stage }} - diff --git a/Makefile b/Makefile index 94f4e02..93c9ab0 100644 --- a/Makefile +++ b/Makefile @@ -13,7 +13,7 @@ test: testcluster testconn testredis rm redis-$(REDIS_VERSION) -rf testredis: /tmp/redis-server/redis-server - PATH=/tmp/redis-server/:${PATH} go test ./redis + PATH=/tmp/redis-server/:${PATH} go test -count 1 ./redis testconn: /tmp/redis-server/redis-server killall redis-server || true diff --git a/redis/example_test.go b/redis/example_test.go index fc82c5a..d65a3f0 100644 --- a/redis/example_test.go +++ b/redis/example_test.go @@ -51,9 +51,9 @@ func ExampleAsError() { } func ExampleScanner() { - defer runServer(46231)() + defer runServer(21050)() ctx := context.Background() - conn, _ := redisconn.Connect(ctx, "127.0.0.1:46231", redisconn.Opts{ + conn, _ := redisconn.Connect(ctx, "127.0.0.1:21050", redisconn.Opts{ Logger: redisconn.NoopLogger{}, }) sync := redis.Sync{conn} @@ -79,9 +79,9 @@ func ExampleScanner() { } func ExampleSync() { - defer runServer(46231)() + defer runServer(21050)() ctx := context.Background() - conn, _ := redisconn.Connect(ctx, "127.0.0.1:46231", redisconn.Opts{ + conn, _ := redisconn.Connect(ctx, "127.0.0.1:21050", redisconn.Opts{ Logger: redisconn.NoopLogger{}, }) sync := redis.Sync{conn} @@ -114,7 +114,7 @@ func ExampleSync() { // OK // OK // ["1" "2"] - // redispipe.result: WRONGTYPE Operation against a key holding the wrong kind of value {request: Req("HSET", ["key1" "field1" "val1"]), address: 127.0.0.1:46231} + // redispipe.result: WRONGTYPE Operation against a key holding the wrong kind of value {request: Req("HSET", ["key1" "field1" "val1"]), address: 127.0.0.1:21050} // // ['\x02' '\x01' "2" "1"] } diff --git a/rediscluster/bench/bench_test.go b/rediscluster/bench/bench_test.go index a57c943..4e6e97e 100644 --- a/rediscluster/bench/bench_test.go +++ b/rediscluster/bench/bench_test.go @@ -29,11 +29,11 @@ func benchCluster(port int) func() { } func BenchmarkSerialGetSet(b *B) { - defer benchCluster(45000)() + defer benchCluster(21080)() rng := rand.New(rand.NewSource(1)) b.Run("radix_pause0", func(b *B) { rdxv2, err := radix.NewCluster( - []string{"127.0.0.1:45000"}, + []string{"127.0.0.1:21080"}, radix.ClusterPoolFunc(func(network, addr string) (radix.Client, error) { return radix.NewPool(network, addr, 4, radix.PoolPipelineWindow(0, 0)) @@ -72,7 +72,7 @@ func BenchmarkSerialGetSet(b *B) { }) b.Run("redispipe", func(b *B) { - pipe, err := rediscluster.NewCluster(context.Background(), []string{"127.0.0.1:45000"}, rediscluster.Opts{ + pipe, err := rediscluster.NewCluster(context.Background(), []string{"127.0.0.1:21080"}, rediscluster.Opts{ Logger: rediscluster.NoopLogger{}, HostOpts: redisconn.Opts{ Logger: redisconn.NoopLogger{}, @@ -96,7 +96,7 @@ func BenchmarkSerialGetSet(b *B) { }) b.Run("redispipe_pause0", func(b *B) { - pipe, err := rediscluster.NewCluster(context.Background(), []string{"127.0.0.1:45000"}, rediscluster.Opts{ + pipe, err := rediscluster.NewCluster(context.Background(), []string{"127.0.0.1:21080"}, rediscluster.Opts{ Logger: rediscluster.NoopLogger{}, HostOpts: redisconn.Opts{ Logger: redisconn.NoopLogger{}, @@ -122,7 +122,7 @@ func BenchmarkSerialGetSet(b *B) { } func BenchmarkParallelGetSet(b *B) { - defer benchCluster(45000)() + defer benchCluster(21080)() parallel := runtime.GOMAXPROCS(0) * 8 i := uint32(1) @@ -137,7 +137,7 @@ func BenchmarkParallelGetSet(b *B) { } b.Run("radix", func(b *B) { - rdx2, err := radix.NewCluster([]string{"127.0.0.1:45000"}) + rdx2, err := radix.NewCluster([]string{"127.0.0.1:21080"}) defer rdx2.Close() if err != nil { b.Fatal(err) @@ -170,7 +170,7 @@ func BenchmarkParallelGetSet(b *B) { }) b.Run("redispipe", func(b *B) { - pipe, err := rediscluster.NewCluster(context.Background(), []string{"127.0.0.1:45000"}, rediscluster.Opts{ + pipe, err := rediscluster.NewCluster(context.Background(), []string{"127.0.0.1:21080"}, rediscluster.Opts{ Logger: rediscluster.NoopLogger{}, HostOpts: redisconn.Opts{ Logger: redisconn.NoopLogger{}, @@ -196,7 +196,7 @@ func BenchmarkParallelGetSet(b *B) { func newRedigo() *redigo.Cluster { c, err := redigo.NewCluster(&redigo.Options{ - StartNodes: []string{"127.0.0.1:45000"}, + StartNodes: []string{"127.0.0.1:21080"}, ConnTimeout: time.Minute, KeepAlive: 128, AliveTime: time.Minute, diff --git a/rediscluster/cluster_test.go b/rediscluster/cluster_test.go index d340c6f..3c6bf47 100644 --- a/rediscluster/cluster_test.go +++ b/rediscluster/cluster_test.go @@ -39,7 +39,7 @@ type Suite struct { func (s *Suite) SetupSuite() { testbed.InitDir(".") - s.cl = testbed.NewCluster(43210) + s.cl = testbed.NewCluster(21100) s.keys = make([]string, NumSlots) cnt := 0 for i := 0; cnt < NumSlots; i++ { @@ -81,6 +81,70 @@ func (s *Suite) AsError(v interface{}) *errorx.Error { return v.(*errorx.Error) } +func containsEvent(events []string, event string) bool { + for _, ev := range events { + if ev == event { + return true + } + } + return false +} + +// waitDebugEvent waits for event to be emitted by cluster's background control loop. +func (s *Suite) waitDebugEvent(event string, timeout time.Duration) { + deadline := time.Now().Add(timeout) + for !containsEvent(DebugEvents(), event) { + if time.Now().After(deadline) { + s.r().Contains(DebugEvents(), event) + } + time.Sleep(10 * time.Millisecond) + } +} + +// waitNoDebugEvent waits for a window of quiet duration without event being emitted. +// Control loop iteration started before the state change under test may still emit +// the event, so a single window is retried until timeout. +func (s *Suite) waitNoDebugEvent(event string, quiet, timeout time.Duration) { + deadline := time.Now().Add(timeout) + for { + DebugEventsReset() + time.Sleep(quiet) + events := DebugEvents() + if !containsEvent(events, event) { + return + } + if time.Now().After(deadline) { + s.r().NotContains(events, event) + } + } +} + +// waitReplicated waits until a replica acknowledges writes to the shard serving slot. +// Node roles are not stable across tests, so the master is the node that accepts a write. +func (s *Suite) waitReplicated(slot int, timeout time.Duration) { + // WAIT only accounts for writes issued by the calling connection, hence the probe. + probe := slotkey("replprobe", s.keys[slot]) + deadline := time.Now().Add(timeout) + for { + for i := range s.cl.Node { + node := &s.cl.Node[i] + if !node.RunningNow() { + continue + } + if redis.AsError(node.Do("SET", probe, "1")) != nil { + continue + } + if n, ok := node.Do("WAIT", 1, 100).(int64); ok && n >= 1 { + return + } + } + if time.Now().After(deadline) { + s.r().Fail("no replica caught up with master of slot " + strconv.Itoa(slot)) + } + time.Sleep(10 * time.Millisecond) + } +} + var defopts = redisconn.Opts{ IOTimeout: 200 * time.Millisecond, } @@ -109,10 +173,45 @@ func TestCluster(t *testing.T) { func (s *Suite) TestConnectDisconnected() { s.cl.Stop() - _, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, clustopts) + _, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, clustopts) s.r().NotNil(err) } +func isConnectivityError(res interface{}) bool { + rerr := redis.AsErrorx(res) + return rerr != nil && rerr.HasTrait(redis.ErrTraitConnectivity) +} + +// doRetrying repeats a request while it fails with a connectivity error, which a +// shard is allowed to answer with while slots migrate between nodes. +func (s *Suite) doRetrying(sconn redis.SyncCtx, ctx context.Context, cmd string, args ...interface{}) interface{} { + var res interface{} + for i := 0; i < 20; i++ { + if res = sconn.Do(ctx, cmd, args...); !isConnectivityError(res) { + break + } + time.Sleep(50 * time.Millisecond) + } + return res +} + +// sendManyRetrying is doRetrying for a batch of requests. +func (s *Suite) sendManyRetrying(sconn redis.SyncCtx, ctx context.Context, reqs []redis.Request) []interface{} { + var ress []interface{} + for i := 0; i < 20; i++ { + ress = sconn.SendMany(ctx, reqs) + retry := false + for _, res := range ress { + retry = retry || isConnectivityError(res) + } + if !retry { + break + } + time.Sleep(50 * time.Millisecond) + } + return ress +} + func slotkey(prefix, slot string, suffix ...string) string { if len(suffix) == 0 { return prefix + "{" + slot + "}" @@ -137,7 +236,7 @@ func (s *Suite) slotnode(slot int) *testbed.Node { } func (s *Suite) TestBasicOps() { - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, clustopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, clustopts) s.r().Nil(err) defer cl.Close() scl := redis.SyncCtx{cl} @@ -167,21 +266,21 @@ func (s *Suite) Test_justToCover() { opts.CheckInterval = 0 opts.MovedRetries = 11 opts.WaitToMigrate = time.Microsecond - cl, err = NewCluster(s.ctx, []string{"never-known-lost-my-host.badubadu.duba:43210"}, opts) + cl, err = NewCluster(s.ctx, []string{"no-such-host-xyzzy.invalid:21100"}, opts) s.r().Nil(cl) s.r().Error(err) opts.CheckInterval = 11 * time.Minute opts.MovedRetries = 1 opts.WaitToMigrate = time.Second - cl, err = NewCluster(s.ctx, []string{"127.0.0.1:43200"}, opts) + cl, err = NewCluster(s.ctx, []string{"127.0.0.1:21090"}, opts) s.r().Nil(cl) s.r().Error(err) opts = clustopts opts.ConnsPerHost = 1 opts.Handle = new(struct{}) - cl, err = NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) + cl, err = NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() @@ -237,7 +336,7 @@ func (c *cancelledFuture) Resolve(res interface{}, n uint64) { } func (s *Suite) TestSendMany() { - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, clustopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, clustopts) s.r().Nil(err) defer cl.Close() scl := redis.SyncCtx{cl} @@ -267,7 +366,7 @@ func (s *Suite) TestSendMany() { func (s *Suite) TestTransactionNormal() { // copy fo connection test - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, clustopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, clustopts) s.r().Nil(err) defer cl.Close() @@ -318,7 +417,7 @@ func (s *Suite) TestScan() { opts := clustopts opts.HostOpts.IOTimeout = time.Second - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() @@ -358,14 +457,15 @@ func (a alwaysZero) Current() uint32 { func (s *Suite) TestFallbackToSlaveStop() { opts := longcheckopts opts.RoundRobinSeed = alwaysZero{} - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() sconn := redis.SyncCtx{cl.WithPolicy(MasterAndSlaves)} key := slotkey("toslave", s.keys[1], "stop") - sconn.Do(s.ctx, "SET", key, "1") + s.r().Equal("OK", sconn.Do(s.ctx, "SET", key, "1")) + s.waitReplicated(1, 30*time.Second) s.cl.Node[0].Stop() // test read from replica @@ -388,14 +488,15 @@ func (s *Suite) TestFallbackToSlaveStop() { func (s *Suite) TestFallbackToSlaveTimeout() { opts := longcheckopts opts.RoundRobinSeed = alwaysZero{} - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() sconn := redis.SyncCtx{cl.WithPolicy(PreferSlaves)} key := slotkey("toslave", s.keys[1], "timeout") - sconn.Do(s.ctx, "SET", key, "1") + s.r().Equal("OK", sconn.Do(s.ctx, "SET", key, "1")) + s.waitReplicated(1, 30*time.Second) s.cl.Node[0].Pause() // test read from replica @@ -418,7 +519,7 @@ func (s *Suite) TestFallbackToSlaveTimeout() { } func (s *Suite) TestGetMoved() { - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, longcheckopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, longcheckopts) s.r().Nil(err) defer cl.Close() @@ -436,7 +537,7 @@ func (s *Suite) TestGetMoved() { } func (s *Suite) TestSetMoved() { - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, longcheckopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, longcheckopts) s.r().Nil(err) defer cl.Close() @@ -456,7 +557,7 @@ func (s *Suite) TestSetMoved() { } func (s *Suite) TestMasterOnly() { - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, clustopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, clustopts) s.r().Nil(err) defer cl.Close() @@ -466,20 +567,17 @@ func (s *Suite) TestMasterOnly() { err = redisclusterutil.SetMasterOnly(cl, "", []uint16{1, 2}) s.r().Nil(err) - time.Sleep(clustopts.CheckInterval * 2) - s.Contains(DebugEvents(), "automatic masteronly") + s.waitDebugEvent("automatic masteronly", 10*time.Second) err = redisclusterutil.UnsetMasterOnly(cl, "", []uint16{1, 2}) s.r().Nil(err) - DebugEventsReset() - time.Sleep(clustopts.CheckInterval * 2) - s.NotContains(DebugEvents(), "automatic masteronly") + s.waitNoDebugEvent("automatic masteronly", clustopts.CheckInterval*2, 10*time.Second) }() } } func (s *Suite) TestAsk() { - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, longcheckopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, longcheckopts) s.r().Nil(err) defer cl.Close() @@ -508,7 +606,7 @@ func (s *Suite) TestAskTransaction() { opts := longcheckopts opts.MovedRetries = 4 - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() @@ -575,7 +673,7 @@ func (s *Suite) TestMovedTransaction() { opts := longcheckopts opts.MovedRetries = 4 - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() @@ -609,11 +707,18 @@ func (s *Suite) fillMany(sconn redis.SyncCtx, prefix string) { for _, res := range ress { s.r().Equal("OK", res) } - time.Sleep(10 * time.Millisecond) + // Tests read filled keys with MasterAndSlaves policy. + for _, slot := range []int{0, 5500, 11000} { + s.waitReplicated(slot, 30*time.Second) + } } func (s *Suite) TestAllReturns_Good() { - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, clustopts) + opts := clustopts + opts.HostOpts.IOTimeout = 2 * time.Second + // ReconnectPause defaults to DialTimeout*2, which follows IOTimeout. + opts.HostOpts.ReconnectPause = 100 * time.Millisecond + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() @@ -627,6 +732,7 @@ func (s *Suite) TestAllReturns_Good() { for i := 0; i < N; i++ { go func(i int) { + defer func() { ch <- struct{}{} }() for j := 0; j < K; j++ { skey := s.keys[(i*N+j)*127%NumSlots] key := slotkey("allgood", skey) @@ -657,7 +763,6 @@ func (s *Suite) TestAllReturns_Good() { return } } - ch <- struct{}{} }(i) } @@ -678,7 +783,12 @@ Loop: func (s *Suite) TestAllReturns_GoodMoving() { opts := clustopts opts.CheckInterval = 4 * time.Second - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) + // MIGRATE blocks both source and destination server, and this test runs it + // in a loop against a cluster loaded with N goroutines. + opts.HostOpts.IOTimeout = 2 * time.Second + // ReconnectPause defaults to DialTimeout*2, which follows IOTimeout. + opts.HostOpts.ReconnectPause = 100 * time.Millisecond + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() @@ -703,7 +813,7 @@ func (s *Suite) TestAllReturns_GoodMoving() { for j := 0; atomic.LoadUint32(&stop) == 0; j++ { skey := s.keys[(i*N+j)*127%NumSlots] key := slotkey("allgoodmove", skey) - res := sconn.Do(ctx, "GET", key) + res := s.doRetrying(sconn, ctx, "GET", key) if !s.Equal([]byte(skey), res) { log.Println("Res ", res) atomic.AddUint32(&bad, 1) @@ -722,7 +832,7 @@ func (s *Suite) TestAllReturns_GoodMoving() { redis.Req("SET", keya, keyb), redis.Req("GET", keyb), } - ress := sconn.SendMany(ctx, reqs) + ress := s.sendManyRetrying(sconn, ctx, reqs) if !s.Equal("OK", ress[0]) { log.Println("Ress[0] ", ress[0]) @@ -768,7 +878,7 @@ func (s *Suite) TestAllReturns_Bad() { s.ctx, s.ctxcancel = context.WithTimeout(context.Background(), 10*time.Minute) DebugDisable = true - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, clustopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, clustopts) s.r().Nil(err) defer cl.Close() @@ -924,9 +1034,9 @@ Loop: func (s *Suite) TestConnectWithDeadAddresses() { addrs := []string{ - "127.0.0.1:43200", // dead - "127.0.0.1:43210", // live - "127.0.0.1:43201", // dead + "127.0.0.1:21090", // dead + "127.0.0.1:21100", // live + "127.0.0.1:21091", // dead } cl, err := NewCluster(s.ctx, addrs, clustopts) s.Nil(err) @@ -941,7 +1051,7 @@ func (s *Suite) TestConnectWithDeadAddresses() { func (s *Suite) TestConnectWithUnresolvableAddresses() { addrs := []string{ "no-such-host-xyzzy.invalid:6379", // guaranteed NXDOMAIN - "127.0.0.1:43210", // live + "127.0.0.1:21100", // live } cl, err := NewCluster(s.ctx, addrs, clustopts) s.Nil(err) diff --git a/redisconn/bench/bench_test.go b/redisconn/bench/bench_test.go index 144b434..8b711f3 100644 --- a/redisconn/bench/bench_test.go +++ b/redisconn/bench/bench_test.go @@ -25,9 +25,9 @@ func benchServer(port int) func() { } func BenchmarkSerialGetSet(b *B) { - defer benchServer(45678)() + defer benchServer(21070)() b.Run("radix", func(b *B) { - rdxv2, err := radix.Dial("tcp", "127.0.0.1:45678") + rdxv2, err := radix.Dial("tcp", "127.0.0.1:21070") if err != nil { b.Fatal(err) return @@ -59,7 +59,7 @@ func BenchmarkSerialGetSet(b *B) { }) b.Run("redispipe", func(b *B) { - pipe, err := redisconn.Connect(context.Background(), "127.0.0.1:45678", redisconn.Opts{ + pipe, err := redisconn.Connect(context.Background(), "127.0.0.1:21070", redisconn.Opts{ Logger: redisconn.NoopLogger{}, }) defer pipe.Close() @@ -79,7 +79,7 @@ func BenchmarkSerialGetSet(b *B) { }) b.Run("redispipe_pause0", func(b *B) { - pipe, err := redisconn.Connect(context.Background(), "127.0.0.1:45678", redisconn.Opts{ + pipe, err := redisconn.Connect(context.Background(), "127.0.0.1:21070", redisconn.Opts{ Logger: redisconn.NoopLogger{}, WritePause: -1, }) @@ -101,7 +101,7 @@ func BenchmarkSerialGetSet(b *B) { } func BenchmarkParallelGetSet(b *B) { - defer benchServer(45678)() + defer benchServer(21070)() parallel := runtime.GOMAXPROCS(0) * 2 do := func(b *B, fn func()) { @@ -114,7 +114,7 @@ func BenchmarkParallelGetSet(b *B) { } b.Run("radix", func(b *B) { - rdx2, err := radix.NewPool("tcp", "127.0.0.1:45678", parallel) + rdx2, err := radix.NewPool("tcp", "127.0.0.1:21070", parallel) if err != nil { b.Fatal(err) } @@ -152,7 +152,7 @@ func BenchmarkParallelGetSet(b *B) { }) b.Run("redispipe", func(b *B) { - pipe, err := redisconn.Connect(context.Background(), "127.0.0.1:45678", redisconn.Opts{ + pipe, err := redisconn.Connect(context.Background(), "127.0.0.1:21070", redisconn.Opts{ Logger: redisconn.NoopLogger{}, }) if err != nil { @@ -173,7 +173,7 @@ func BenchmarkParallelGetSet(b *B) { } func newRedigo() redigo.Conn { - c, err := redigo.Dial("tcp", "127.0.0.1:45678") + c, err := redigo.Dial("tcp", "127.0.0.1:21070") if err != nil { panic(err) } diff --git a/redisconn/conn_test.go b/redisconn/conn_test.go index d0d4f8f..0bcd7c3 100644 --- a/redisconn/conn_test.go +++ b/redisconn/conn_test.go @@ -29,8 +29,8 @@ type Suite struct { func (s *Suite) SetupSuite() { testbed.InitDir(".") - s.s.Port = 45678 - s.s.TlsPort = 55678 + s.s.Port = 21060 + s.s.TlsPort = 21061 s.s.Start() } @@ -115,8 +115,10 @@ func (s *Suite) TestConnects() { } func (s *Suite) TestConnectsTls() { + // A TLS handshake does not fit into the IO timeout the other tests use. + const tlsTimeout = time.Second tlsopts := Opts{ - IOTimeout: defopts.IOTimeout, + IOTimeout: tlsTimeout, TLSEnabled: true, TLSConfig: &tls.Config{ InsecureSkipVerify: true, @@ -125,7 +127,7 @@ func (s *Suite) TestConnectsTls() { conn, err := Connect(s.ctx, s.s.TlsAddr(), tlsopts) s.r().Nil(err) defer conn.Close() - s.goodPing(conn, 0) + s.goodPing(conn, tlsTimeout) } func (s *Suite) TestConnectsDb() { diff --git a/testbed/cluster.go b/testbed/cluster.go index 1ba738e..bdf9ed3 100644 --- a/testbed/cluster.go +++ b/testbed/cluster.go @@ -3,10 +3,13 @@ package testbed import ( "bytes" "crypto/tls" + "fmt" "log" + "sync/atomic" "time" "github.com/joomcode/redispipe/rediscluster/redisclusterutil" + "github.com/joomcode/redispipe/redisdumb" ) // Node is wrapper for Server with its NodeId @@ -18,6 +21,8 @@ type Node struct { // Cluster is a tool for starting/stopping redis cluster for tests. type Cluster struct { Node []Node + + lastNotOk atomic.Value } // NewCluster instantiate cluster of 6 nodes (3 masters and 3 slaves). @@ -41,7 +46,7 @@ func NewCluster(startport uint16) *Cluster { cl.Node[i].Args = []string{ "--cluster-enabled", "yes", "--cluster-config-file", "node-" + cl.Node[i].PortStr(effectivePort) + ".conf", - "--cluster-node-timeout", "200", + "--cluster-node-timeout", "1000", "--cluster-slave-validity-factor", "1000", "--slave-serve-stale-data", "yes", "--cluster-require-full-coverage", "no", @@ -102,10 +107,42 @@ func RaiseClusterPanic() { panic("cluster didn't stabilize") } +func dumpResult(res interface{}) string { + if buf, ok := res.([]byte); ok { + return string(buf) + } + return fmt.Sprintf("%v", res) +} + +// DumpState reports what every node thinks about the cluster. +func (cl *Cluster) DumpState() { + if reason, ok := cl.lastNotOk.Load().(string); ok { + log.Printf("cluster is not ok: %s", reason) + } + for i := range cl.Node { + n := &cl.Node[i] + if !n.RunningNow() { + log.Printf("node %d (%s): not running", i, n.Addr()) + continue + } + // The node's own connection belongs to the goroutine that polls it. + conn := redisdumb.Conn{ + Addr: n.Conn.Addr, + TlsAddr: n.Conn.TlsAddr, + TLSEnabled: n.Conn.TLSEnabled, + TLSConfig: n.Conn.TLSConfig, + } + log.Printf("node %d (%s): %s\n%s", i, n.Addr(), dumpResult(conn.Do("CLUSTER INFO")), dumpResult(conn.Do("CLUSTER NODES"))) + } +} + // WaitClusterOk wait for cluster configuration to be stable. func (cl *Cluster) WaitClusterOk() { i := 0 - t := time.AfterFunc(30*time.Second, RaiseClusterPanic) + t := time.AfterFunc(60*time.Second, func() { + cl.DumpState() + RaiseClusterPanic() + }) defer t.Stop() for !cl.ClusterOk() { if i++; i == 10 { @@ -117,6 +154,14 @@ func (cl *Cluster) WaitClusterOk() { // ClusterOk checks cluster configuration. func (cl *Cluster) ClusterOk() bool { + reason := cl.clusterNotOkReason() + cl.lastNotOk.Store(reason) + return reason == "" +} + +// clusterNotOkReason returns an empty string when the cluster configuration is stable, +// and what stands in the way otherwise. +func (cl *Cluster) clusterNotOkReason() string { stopped := []int{} for i := range cl.Node { if !cl.Node[i].RunningNow() { @@ -124,6 +169,7 @@ func (cl *Cluster) ClusterOk() bool { } } var hashsum uint64 + var hashnode int for i := range cl.Node { if !cl.Node[i].RunningNow() { continue @@ -131,24 +177,24 @@ func (cl *Cluster) ClusterOk() bool { res := cl.Node[i].Do("CLUSTER INFO") buf, ok := res.([]byte) if !ok { - return false + return fmt.Sprintf("node %d: CLUSTER INFO: %v", i, res) } if !bytes.Contains(buf, []byte("cluster_state:ok")) { - return false + return fmt.Sprintf("node %d: cluster state is not ok", i) } res = cl.Node[i].Do("INFO REPLICATION") buf, ok = res.([]byte) if !ok { - return false + return fmt.Sprintf("node %d: INFO REPLICATION: %v", i, res) } if !bytes.Contains(buf, []byte("role:master")) && !bytes.Contains(buf, []byte("master_link_status:up")) { - return false + return fmt.Sprintf("node %d: replica is not in sync with its master", i) } res = cl.Node[i].Do("CLUSTER NODES") buf, ok = res.([]byte) if !ok { - return false + return fmt.Sprintf("node %d: CLUSTER NODES: %v", i, res) } masters := 0 for _, line := range bytes.Split(buf, []byte("\n")) { @@ -156,7 +202,7 @@ func (cl *Cluster) ClusterOk() bool { for _, j := range stopped { if bytes.HasPrefix(line, cl.Node[j].NodeId) { if !bytes.Contains(line, []byte("fail ")) { - return false + return fmt.Sprintf("node %d: stopped node %d is not marked as failed", i, j) } hasStopped = true } @@ -165,18 +211,21 @@ func (cl *Cluster) ClusterOk() bool { masters++ } } - if masters != 3+(len(cl.Node)-6) { - return false - + if want := 3 + (len(cl.Node) - 6); masters != want { + return fmt.Sprintf("node %d: sees %d masters instead of %d", i, masters, want) + } + infos, err := redisclusterutil.ParseClusterNodes(res) + if err != nil { + return fmt.Sprintf("node %d: unparsable CLUSTER NODES: %v", i, err) } - infos, _ := redisclusterutil.ParseClusterNodes(res) hash := infos.HashSum() if hash != hashsum && hashsum != 0 { - return false + return fmt.Sprintf("node %d: sees a configuration different from node %d", i, hashnode) } hashsum = hash + hashnode = i } - return true + return "" } // AttemptFailover tries to issue CLUSTER FAILOVER FORCE to slaves of falled masters. @@ -217,8 +266,18 @@ func (cl *Cluster) CancelMoveSlot(slot int) { // FinishMoveSlot finalizes slot migration func (cl *Cluster) FinishMoveSlot(slot, from, to int) { - cl.Node[to].Do("CLUSTER SETSLOT", slot, "NODE", cl.Node[to].NodeId) - cl.Node[from].Do("CLUSTER SETSLOT", slot, "NODE", cl.Node[to].NodeId) + cl.Node[to].DoSure("CLUSTER SETSLOT", slot, "NODE", cl.Node[to].NodeId) + cl.Node[from].DoSure("CLUSTER SETSLOT", slot, "NODE", cl.Node[to].NodeId) + // The rest of the masters would learn the new owner from the epoch bump below, + // but a next migration of the same slot may outrun it, and then neither the old + // nor the new owner claims the slot and the cluster never agrees again. + // Replicas answer with an error, which is why these are not DoSure. + for i := range cl.Node { + if i == to || i == from || !cl.Node[i].RunningNow() { + continue + } + cl.Node[i].Do("CLUSTER SETSLOT", slot, "NODE", cl.Node[to].NodeId) + } cl.Node[to].Do("CLUSTER BUMPEPOCH", "BROADCAST") // proprietary extension cl.Node[to].Do("CLUSTER BUMPEPOCH") } @@ -262,7 +321,7 @@ func (cl *Cluster) StartSeventhNode() { cl.Node[6].Args = []string{ "--cluster-enabled", "yes", "--cluster-config-file", "node-" + cl.Node[6].PortStr(effectivePort) + ".conf", - "--cluster-node-timeout", "200", + "--cluster-node-timeout", "1000", "--cluster-slave-validity-factor", "1000", "--slave-serve-stale-data", "yes", "--cluster-require-full-coverage", "no",