make NewRouter tolerant to the cluster's degraded state (resolve issue #13)

* DiscoveryAllBuckets: don't cancel requests to other replicasets if request to some replicaset failed.
* add test for degraded cluster (TestDegradedCluster)
* AddReplicaset: don't fail if there are no available RW instances
* NewRouter: don't fail if DiscoveryAllBuckets has returned an error

Author: nurzhan-saktaganov

CHANGELOG.md

@@ -3,9 +3,15 @@
 BUG FIXES:
 - Fixed go.mod and go.sum files with go mod tidy.
 
+CHANGES:
+* DiscoveryAllBuckets: don't cancel requests to other replicasets if request to some replicaset failed.
+* AddReplicaset: don't fail if there are no available RW instances.
+* NewRouter: don't fail if DiscoveryAllBuckets has returned an error.
+
 TESTS:
 
 - Added etcd v2 provider tests.
+- Add test for degraded cluster (TestDegradedCluster).
 
 ## v2.0.1
 

discovery.go

@@ -222,7 +222,7 @@ func (r *Router) DiscoveryAllBuckets(ctx context.Context) error {
 
 	r.log().Infof(ctx, "Start discovery all buckets")
 
-	errGr, ctx := errgroup.WithContext(ctx)
+	var errGr errgroup.Group
 
 	view := r.getConsistentView()
 	nameToReplicasetRef := r.getNameToReplicaset()
@@ -236,6 +236,7 @@ func (r *Router) DiscoveryAllBuckets(ctx context.Context) error {
 			for {
 				resp, err := rs.bucketsDiscovery(ctx, bucketsDiscoveryPaginationFrom)
 				if err != nil {
+					r.log().Errorf(ctx, "can't bucketsDiscovery for rs %s: %v", rs.info, err)
 					return err
 				}
 

tarantool_test.go

@@ -639,3 +639,48 @@ func TestReplicaset_Pooler(t *testing.T) {
 		}
 	})
 }
+
+func TestDegradedCluster(t *testing.T) {
+	ctx := context.Background()
+
+	// create a topology to imitate cluster with several replicasets:
+	// 1 fake replicaset (imitates an unavailable replicaset) + all replicasets of topology
+	topologyDegraded := map[vshardrouter.ReplicasetInfo][]vshardrouter.InstanceInfo{
+		{
+			// add fake replicaset
+			Name:   "storage_0",
+			UUID:   uuid.New(),
+			Weight: 1,
+		}: {
+			{
+				Name: "storage_0_a",
+				UUID: uuid.New(),
+				Addr: "127.0.0.1:2998",
+			},
+			{
+				Name: "storage_0_b",
+				UUID: uuid.New(),
+				Addr: "127.0.0.1:2999",
+			},
+		},
+	}
+
+	// add all replicasets of topology
+	for k, v := range topology {
+		topologyDegraded[k] = v
+	}
+
+	router, err := vshardrouter.NewRouter(ctx, vshardrouter.Config{
+		TopologyProvider: static.NewProvider(topologyDegraded),
+		DiscoveryTimeout: 5 * time.Second,
+		DiscoveryMode:    vshardrouter.DiscoveryModeOn,
+		TotalBucketCount: totalBucketCount,
+		User:             username,
+	})
+	require.NoError(t, err, "NewRouter created successfully")
+
+	for bucketID := uint64(1); bucketID <= totalBucketCount; bucketID++ {
+		_, err := router.Route(ctx, bucketID)
+		require.NoErrorf(t, err, "bucket %d resolved successfully")
+	}
+}

topology.go

@@ -130,13 +130,11 @@ func (r *Router) AddReplicaset(ctx context.Context, rsInfo ReplicasetInfo, insta
 		r.log().Infof(ctx, "[replicaset %s ] instance %s %s in role %s", rsInfo, instName, connectStatus, instConnInfo.ConnRole)
 	}
 
-	isConnected, err := conn.ConnectedNow(pool.RW)
-	if err != nil {
-		return fmt.Errorf("cant check rs pool conntected rw now with error: %s", err)
-	}
-
-	if !isConnected {
-		return fmt.Errorf("got connected now as false, storage must be configured first")
+	switch isConnected, err := conn.ConnectedNow(pool.RW); {
+	case err != nil:
+		r.log().Errorf(ctx, "cant check rs pool conntected rw now with error: %v", err)
+	case !isConnected:
+		r.log().Errorf(ctx, "got connected now as false to pool.RW")
 	}
 
 	replicaset.conn = conn

vshard.go

@@ -241,7 +241,7 @@ func NewRouter(ctx context.Context, cfg Config) (*Router, error) {
 
 	err = router.DiscoveryAllBuckets(ctx)
 	if err != nil {
-		return nil, err
+		router.log().Errorf(ctx, "router.DiscoveryAllBuckets failed: %v", err)
 	}
 
 	if cfg.DiscoveryMode == DiscoveryModeOn {
@@ -255,7 +255,7 @@ func NewRouter(ctx context.Context, cfg Config) (*Router, error) {
 		router.cancelDiscovery = cancelFunc
 	}
 
-	return router, err
+	return router, nil
 }
 
 // BucketSet Set a bucket to a replicaset.