From ba5b46898ff662b59f6af5bfd8fdbde59b5df19d Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 6 Aug 2018 14:55:38 -0500 Subject: [PATCH] fix determineClusterState bug and add more test cases --- cluster.go | 2 +- server/server_test.go | 106 +++++++++++++++++++++++++++++++++++++++--- 2 files changed, 100 insertions(+), 8 deletions(-) diff --git a/cluster.go b/cluster.go index 12513c14e..86fdf7712 100644 --- a/cluster.go +++ b/cluster.go @@ -474,7 +474,7 @@ func (c *cluster) determineClusterState() (clusterState string) { if c.haveTopologyAgreement() && c.allNodesReady() { return ClusterStateNormal } - if len(c.nodeIDs())-len(c.Topology.nodeIDs) >= c.ReplicaN && c.allNodesReady() { + if len(c.Topology.nodeIDs)-len(c.nodeIDs()) < c.ReplicaN && c.allNodesReady() { return ClusterStateDegraded } return ClusterStateStarting diff --git a/server/server_test.go b/server/server_test.go index 6c937a116..9fa2df34d 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -21,6 +21,7 @@ import ( "math/rand" "reflect" "sort" + "strconv" "strings" "testing" "testing/quick" @@ -376,7 +377,7 @@ func (p uint64Slice) Swap(i, j int) { p[i], p[j] = p[j], p[i] } func (p uint64Slice) Len() int { return len(p) } func (p uint64Slice) Less(i, j int) bool { return p[i] < p[j] } -func TestClusteringNodes(t *testing.T) { +func TestClusteringNodesReplica1(t *testing.T) { cluster := test.MustRunCluster(t, 3) defer cluster.Close() @@ -396,16 +397,17 @@ func TestClusteringNodes(t *testing.T) { } // TODO: confirm that cluster stops accepting queries after one node closes - // TODO: implement and confirm that cluster keeps accepting queries if replication > 1 - // TODO: confirm that cluster stops accepting queries if 2 nodes fail and replication == 2 - // TODO: confirm that things keep working if a node is hard-closed (no nodeLeave event) and immediately restarted with a different address. - // TODO: confirm that a node can be removed using the remove endpoint after it has left teh cluster - // TODO: confirm that cluster still operates properly in state DEGRADED + if _, err := cluster[0].API.Query(context.Background(), &pilosa.QueryRequest{}); !strings.Contains(err.Error(), "not allowed in state STARTING") { + t.Fatalf("got unexpected error querying an incomplete cluster: %v", err) + } // Create new main with the same config. config := cluster[2].Command.Config // config.Bind = cluster[2].API.Node().URI.HostPort() - // config.Gossip.Port = strconv.Itoa(int(cluster[2].Command.GossipTransport().URI.Port)) + + // this isn't necessary, but makes the test run way faster + config.Gossip.Port = strconv.Itoa(int(cluster[2].Command.GossipTransport().URI.Port)) + cluster[2].Command = server.NewCommand(cluster[2].Stdin, cluster[2].Stdout, cluster[2].Stderr) cluster[2].Command.Config = config @@ -424,3 +426,93 @@ func TestClusteringNodes(t *testing.T) { time.Sleep(time.Millisecond) } } + +func TestClusteringNodesReplica2(t *testing.T) { + cluster := test.MustNewCluster(t, 3) + for _, c := range cluster { + c.Config.Cluster.ReplicaN = 2 + } + err := cluster.Start() + if err != nil { + t.Fatalf("starting cluster: %v", err) + } + + var wait = true + for wait { + wait = false + for _, node := range cluster { + if node.API.State() != pilosa.ClusterStateNormal { + wait = true + } + } + time.Sleep(time.Millisecond * 1) + } + + if err := cluster[2].Command.Close(); err != nil { + t.Fatalf("closing third node: %v", err) + } + + if cluster[0].API.State() != pilosa.ClusterStateDegraded { + t.Fatalf("expected state to be DEGRADED, but got %s", cluster[0].API.State()) + } + + // TODO: implement and confirm that cluster keeps accepting queries if replication > 1 + if _, err := cluster[0].API.CreateIndex(context.Background(), "anewindex", pilosa.IndexOptions{}); err != nil { + t.Fatalf("got unexpected error creating index: %v", err) + } + + // TODO: confirm that cluster stops accepting queries if 2 nodes fail and replication == 2 + if err := cluster[1].Command.Close(); err != nil { + t.Fatalf("closing 2nd node: %v", err) + } + + if cluster[0].API.State() != pilosa.ClusterStateStarting { + t.Fatalf("expected state to be Starting, but got %s", cluster[0].API.State()) + } + + if _, err := cluster[0].API.Query(context.Background(), &pilosa.QueryRequest{}); !strings.Contains(err.Error(), "not allowed in state STARTING") { + t.Fatalf("got unexpected error querying an incomplete cluster: %v", err) + } + + // Create new main with the same config. + config := cluster[2].Command.Config + // config.Bind = cluster[2].API.Node().URI.HostPort() + + // this isn't necessary, but makes the test run way faster + config.Gossip.Port = strconv.Itoa(int(cluster[2].Command.GossipTransport().URI.Port)) + + cluster[2].Command = server.NewCommand(cluster[2].Stdin, cluster[2].Stdout, cluster[2].Stderr) + cluster[2].Command.Config = config + + // Run new program. + if err := cluster[2].Start(); err != nil { + t.Fatalf("restarting node 2: %v", err) + } + + // Create new main with the same config. + config = cluster[1].Command.Config + // config.Bind = cluster[1].API.Node().URI.HostPort() + + // this isn't necessary, but makes the test run way faster + config.Gossip.Port = strconv.Itoa(int(cluster[1].Command.GossipTransport().URI.Port)) + + cluster[1].Command = server.NewCommand(cluster[1].Stdin, cluster[1].Stdout, cluster[1].Stderr) + cluster[1].Command.Config = config + + // Run new program. + if err := cluster[1].Start(); err != nil { + t.Fatalf("restarting node 2: %v", err) + } + + defer cluster.Close() + + for wait { + wait = false + for _, node := range cluster { + if node.API.State() != pilosa.ClusterStateNormal { + wait = true + } + } + time.Sleep(time.Millisecond) + } +}