From 0e34409ff0bc12b32ae140ed6fc099618a8eb243 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Tue, 2 Feb 2021 11:47:06 +0100 Subject: [PATCH 1/2] Close etcd client after Revoke --- etcd/embed.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/etcd/embed.go b/etcd/embed.go index a0ddb8029..dbfd88024 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -743,9 +743,10 @@ func (e *Etcd) leaseKeepAlive(ttl int64) (clientv3.LeaseID, func(context.Context if cli, err := e.client(); err != nil { log.Printf("leaseKeepAlive: creates a new client: %v\n", err) } else { - if _, err := cli.Revoke(context.Background(), leaseResp.ID); err != nil { + if _, err := cli.Revoke(context.TODO(), leaseResp.ID); err != nil { log.Printf("leaseKeepAlive: revokes the lease (ID: %v): %v\n", leaseResp.ID, err) } + cli.Close() } return From a16a83445b51d4917b80dff4c8ab1264b51f504d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Tue, 2 Feb 2021 19:36:34 +0100 Subject: [PATCH 2/2] Apply stator --- cluster.go | 19 ++++++++----------- etcd/embed.go | 8 +++++--- executor_test.go | 17 ++++++++--------- server/server_test.go | 37 +++++++++++++++++++++---------------- test/cluster.go | 43 +------------------------------------------ 5 files changed, 43 insertions(+), 81 deletions(-) diff --git a/cluster.go b/cluster.go index d6982915d..470a2d829 100644 --- a/cluster.go +++ b/cluster.go @@ -512,8 +512,9 @@ func (c *cluster) unprotectedNodeByID(id string) *topology.Node { func (c *cluster) topologyContainsNode(id string) bool { c.Topology.mu.RLock() defer c.Topology.mu.RUnlock() - for _, nid := range c.Topology.nodeIDs { - if id == nid { + + for _, n := range c.noder.Nodes() { + if id == n.ID { return true } } @@ -2216,9 +2217,10 @@ func (c *cluster) nodeLeave(nodeID string) error { c.unprotectedCoordinatorNode().ID) } - if c.state != string(ClusterStateNormal) && c.state != string(ClusterStateDegraded) { - return fmt.Errorf("cluster must be '%s' or '%s' to remove a node but is '%s'", - ClusterStateNormal, ClusterStateDegraded, c.state) + state, err := c.stator.ClusterState(context.TODO()) + if err != nil || (state != disco.ClusterStateNormal && state != disco.ClusterStateDegraded) { + return fmt.Errorf("cluster must be '%s' or '%s' to remove a node but is '%s', error: %v", + ClusterStateNormal, ClusterStateDegraded, state, err) } // Ensure that node is in the cluster. @@ -2245,16 +2247,11 @@ func (c *cluster) nodeLeave(nodeID string) error { if err := c.removeNode(nodeID); err != nil { return errors.Wrap(err, "removing node") } - return c.unprotectedSetStateAndBroadcast(c.determineClusterState()) + return nil } else if err != nil { return errors.Wrap(err, "checking if holder has data") } - // If the cluster has data then change state to RESIZING and - // kick off the resizing process. - if err := c.unprotectedSetStateAndBroadcast(string(ClusterStateResizing)); err != nil { - return errors.Wrap(err, "broadcasting state") - } c.joiningLeavingNodes <- nodeAction{node: &topology.Node{ID: nodeID}, action: resizeJobActionRemove} return nil diff --git a/etcd/embed.go b/etcd/embed.go index dbfd88024..dcd261b60 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -728,7 +728,7 @@ func (e *Etcd) leaseKeepAlive(ttl int64) (clientv3.LeaseID, func(context.Context leaseResp, err := cli.Grant(context.TODO(), ttl) if err != nil { - return 0, nil, errors.Wrapf(err, "leaseKeepAlive: creates a new lease (TTL: %d)", ttl) + return 0, nil, errors.Wrapf(err, "leaseKeepAlive: creates a new lease (TTL: %v)", ttl) } keepaliveFunc := func(ctx context.Context, tick time.Duration) { @@ -744,7 +744,7 @@ func (e *Etcd) leaseKeepAlive(ttl int64) (clientv3.LeaseID, func(context.Context log.Printf("leaseKeepAlive: creates a new client: %v\n", err) } else { if _, err := cli.Revoke(context.TODO(), leaseResp.ID); err != nil { - log.Printf("leaseKeepAlive: revokes the lease (ID: %v): %v\n", leaseResp.ID, err) + log.Printf("leaseKeepAlive: revokes the lease (ID: %x): %v\n", leaseResp.ID, err) } cli.Close() } @@ -755,7 +755,7 @@ func (e *Etcd) leaseKeepAlive(ttl int64) (clientv3.LeaseID, func(context.Context log.Printf("leaseKeepAlive: creates a new client: %v\n", err) } else { if _, err = cli.KeepAliveOnce(ctx, leaseResp.ID); err != nil { - log.Printf("leaseKeepAlive: renews the lease (ID: %v): %v\n", leaseResp.ID, err) + log.Printf("leaseKeepAlive: renews the lease (ID: %x): %v\n", leaseResp.ID, err) } cli.Close() } @@ -768,10 +768,12 @@ func (e *Etcd) leaseKeepAlive(ttl int64) (clientv3.LeaseID, func(context.Context func (e *Etcd) client() (*clientv3.Client, error) { urls := e.e.Server.Cluster().ClientURLs() + cli, err := clientv3.NewFromURLs(urls) if err != nil { return nil, errors.Wrapf(err, "creates a new etcd client from URLs (%v)", urls) } + return cli, nil } diff --git a/executor_test.go b/executor_test.go index 18d9939ef..14c1d0fda 100644 --- a/executor_test.go +++ b/executor_test.go @@ -3526,8 +3526,9 @@ func TestExecutor_Execute_Existence(t *testing.T) { t.Fatal(err) } + node0 := c.GetNode(0) // Set bits. - if _, err := c.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `` + + if _, err := node0.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `` + fmt.Sprintf("Set(%d, f=%d)\n", 3, 10) + fmt.Sprintf("Set(%d, f=%d)\n", ShardWidth+1, 10) + fmt.Sprintf("Set(%d, f=%d)\n", ShardWidth+2, 20), @@ -3535,26 +3536,24 @@ func TestExecutor_Execute_Existence(t *testing.T) { t.Fatal(err) } - //index.Dump("after Set 3x") - - if res, err := c.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Row(f=10)`}); err != nil { + if res, err := node0.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Row(f=10)`}); err != nil { t.Fatal(err) } else if bits := res.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(bits, []uint64{3, ShardWidth + 1}) { t.Fatalf("unexpected columns: %+v", bits) } - if res, err := c.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Not(Row(f=10))`}); err != nil { + if res, err := node0.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Not(Row(f=10))`}); err != nil { t.Fatal(err) } else if bits := res.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(bits, []uint64{ShardWidth + 2}) { t.Fatalf("unexpected columns after Not: %+v", bits) } // Reopen cluster to ensure existence field is reloaded. - if err := c.GetNode(0).Reopen(); err != nil { + if err := node0.Reopen(); err != nil { t.Fatal(err) } - if err := c.AwaitState(string(pilosa.ClusterStateNormal), 10*time.Second); err != nil { + if err := node0.AwaitState(string(pilosa.ClusterStateNormal), 10*time.Second); err != nil { t.Fatalf("restarting cluster: %v", err) } @@ -6963,7 +6962,7 @@ toronto,3 { // 2019 All, this excludes userC (who likes pangolin & icecream) from the count. // UserC visited Paris and Toronto in 2019 query: `GroupBy( - Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59'), + Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59'), filter=Not(Intersect(Row(likes='pangolin'), Row(likes='icecream'))) )`, csvVerifier: `nairobi,1 @@ -6973,7 +6972,7 @@ toronto,2 }, { // After excluding UserC, this gets the sum of the networth of everyone per cities travelled query: `GroupBy( - Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59'), + Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59'), filter=Not(Intersect(Row(likes='pangolin'), Row(likes='icecream'))), aggregate=Sum(field=net_worth) )`, diff --git a/server/server_test.go b/server/server_test.go index c1bf488f2..ba93abc75 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -371,12 +371,13 @@ func TestConcurrentFieldCreation(t *testing.T) { cluster := test.MustRunCluster(t, 3) defer cluster.Close() - err := cluster.AwaitState(string(pilosa.ClusterStateNormal), 100*time.Millisecond) + node0 := cluster.GetNode(0) + err := node0.AwaitState(string(pilosa.ClusterStateNormal), 100*time.Millisecond) if err != nil { t.Fatalf("starting cluster: %v", err) } - api0 := cluster.GetNode(0).API + api0 := node0.API if _, err := api0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil { t.Fatalf("creating index: %v", err) } @@ -643,7 +644,7 @@ func TestClusteringNodesReplica1(t *testing.T) { cluster := test.MustRunCluster(t, 3) defer cluster.Close() - if err := cluster.AwaitState(string(disco.ClusterStateNormal), 100*time.Millisecond); err != nil { + if err := cluster.GetNode(0).AwaitState(string(disco.ClusterStateNormal), 100*time.Millisecond); err != nil { t.Fatalf("starting cluster: %v", err) } @@ -651,7 +652,7 @@ func TestClusteringNodesReplica1(t *testing.T) { t.Fatalf("closing third node: %v", err) } - if err := cluster.AwaitCoordinatorState(string(disco.ClusterStateDown), 60*time.Second); err != nil { + if err := cluster.GetCoordinator().AwaitState(string(disco.ClusterStateDown), 30*time.Second); err != nil { t.Fatalf("starting cluster: %v", err) } @@ -681,7 +682,7 @@ func TestClusteringNodesReplica2(t *testing.T) { t.Fatalf("closing third node: %v", err) } - err = cluster.AwaitCoordinatorState(string(disco.ClusterStateDegraded), 30*time.Second) + err = coord.AwaitState(string(disco.ClusterStateDegraded), 30*time.Second) if err != nil { t.Fatalf("after closing first server: %v", err) } @@ -699,7 +700,7 @@ func TestClusteringNodesReplica2(t *testing.T) { t.Fatalf("closing 2nd node: %v", err) } - err = cluster.AwaitCoordinatorState(string(pilosa.ClusterStateDown), 30*time.Second) + err = coord.AwaitState(string(pilosa.ClusterStateDown), 30*time.Second) if err != nil { t.Fatalf("after closing second server: %v", err) } @@ -710,6 +711,8 @@ func TestClusteringNodesReplica2(t *testing.T) { } func TestRemoveNodeAfterItDies(t *testing.T) { + t.Skip("TestRemoveNodeAfterItDies won't be supported unless we implement resizer.") + cluster := test.MustNewCluster(t, 3) for _, c := range cluster.Nodes { c.Config.Cluster.ReplicaN = 2 @@ -726,19 +729,20 @@ func TestRemoveNodeAfterItDies(t *testing.T) { cluster.Close() }() - err = cluster.AwaitState(string(pilosa.ClusterStateNormal), 100*time.Millisecond) + coord, others := cluster.GetCoordinator(), cluster.GetNonCoordinators() + + err = coord.AwaitState(string(pilosa.ClusterStateNormal), 100*time.Millisecond) if err != nil { t.Fatalf("starting cluster: %v", err) } - coord, others := cluster.GetCoordinator(), cluster.GetNonCoordinators() // prevent double-closing cluster.GetNode(2) from the deferred Close above disabled := others[0] if err := disabled.Close(); err != nil { t.Fatalf("closing third node: %v", err) } - err = cluster.AwaitCoordinatorState(string(pilosa.ClusterStateDegraded), 30*time.Second) + err = coord.AwaitState(string(pilosa.ClusterStateDegraded), 30*time.Second) if err != nil { t.Fatalf("starting cluster: %v", err) } @@ -747,7 +751,7 @@ func TestRemoveNodeAfterItDies(t *testing.T) { t.Fatalf("removing failed node: %v", err) } - err = cluster.AwaitCoordinatorState(string(pilosa.ClusterStateNormal), 30*time.Second) + err = coord.AwaitState(string(pilosa.ClusterStateNormal), 30*time.Second) if err != nil { t.Fatalf("removing disabled node: %v", err) } @@ -770,27 +774,28 @@ func TestRemoveConcurrentIndexCreation(t *testing.T) { } defer cluster.Close() - err = cluster.AwaitState(string(pilosa.ClusterStateNormal), 100*time.Millisecond) + node0 := cluster.GetNode(0) + err = node0.AwaitState(string(pilosa.ClusterStateNormal), 100*time.Millisecond) if err != nil { t.Fatalf("starting cluster: %v", err) } errc := make(chan error) go func() { - _, err := cluster.GetNode(0).API.CreateIndex(context.Background(), "blah", pilosa.IndexOptions{}) + _, err := node0.API.CreateIndex(context.Background(), "blah", pilosa.IndexOptions{}) errc <- err }() - if _, err := cluster.GetNode(0).API.RemoveNode(cluster.GetNode(2).API.Node().ID); err != nil { + if _, err := node0.API.RemoveNode(cluster.GetNode(2).API.Node().ID); err != nil { t.Fatalf("removing node: %v", err) } - err = cluster.AwaitCoordinatorState(string(pilosa.ClusterStateNormal), 100*time.Millisecond) + err = cluster.GetCoordinator().AwaitState(string(pilosa.ClusterStateNormal), 100*time.Millisecond) if err != nil { t.Fatalf("starting cluster: %v", err) } - hosts := cluster.GetNode(0).API.Hosts(context.Background()) + hosts := node0.API.Hosts(context.Background()) if len(hosts) != 2 { t.Fatalf("unexpected hosts: %v", hosts) } @@ -917,7 +922,7 @@ func TestClusterQueriesAfterRestart(t *testing.T) { defer cluster.Close() cmd1 := cluster.GetNode(1) - err := cluster.AwaitState(string(pilosa.ClusterStateNormal), 100*time.Millisecond) + err := cmd1.AwaitState(string(pilosa.ClusterStateNormal), 100*time.Millisecond) if err != nil { t.Fatalf("starting cluster: %v", err) } diff --git a/test/cluster.go b/test/cluster.go index b581f3eff..6a8325c77 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -438,7 +438,7 @@ func (c *Cluster) Start() error { return err } - return c.AwaitState(string(pilosa.ClusterStateNormal), 30*time.Second) + return c.GetNode(0).AwaitState(string(pilosa.ClusterStateNormal), 30*time.Second) } // Close stops a Cluster @@ -470,47 +470,6 @@ func (c *Cluster) CloseAndRemove(n int) error { return err } -// AwaitState waits for the cluster coordinator (assumed to be the first -// node) to reach a specified state. -func (c *Cluster) AwaitCoordinatorState(expectedState string, timeout time.Duration) error { - if len(c.Nodes) < 1 { - return errors.New("can't await coordinator state on an empty cluster") - } - onlyCoordinator := &Cluster{Nodes: []*Command{c.GetCoordinator()}} - return onlyCoordinator.AwaitState(expectedState, timeout) -} - -// ExceptionalState returns an error if any node in the cluster is not -// in the expected state. -func (c *Cluster) ExceptionalState(expectedState string) error { - for _, node := range c.Nodes { - state, err := node.API.State() - if err != nil || state != expectedState { - return fmt.Errorf("node %q: state %s: err %v", node.ID(), state, err) - } - } - return nil -} - -// AwaitState waits for the whole cluster to reach a specified state. -func (c *Cluster) AwaitState(expectedState string, timeout time.Duration) (err error) { - if len(c.Nodes) < 1 { - return errors.New("can't await state of an empty cluster") - } - startTime := time.Now() - var elapsed time.Duration - for elapsed = 0; elapsed <= timeout; elapsed = time.Since(startTime) { - // Counterintuitive: We're returning if the err *is* nil, - // meaning we've reached the expected state. - if err = c.ExceptionalState(expectedState); err == nil { - return err - } - time.Sleep(1 * time.Millisecond) - } - return fmt.Errorf("waited %v for cluster to reach state %q: %v", - elapsed, expectedState, err) -} - // MustNewCluster creates a new cluster. If opts contains only one // slice of command options, those options are used with every node. // If it is empty, default options are used. Otherwise, it must contain size