scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-06 00:25:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'disco-config-noder' of github.com:travisturner/privilosa into disco-config-noder

Browse source
This commit is contained in:
Travis 2021-02-02 15:54:01 -06:00
commit 28a19cdff7
No known key found for this signature in database
GPG key ID: 37080CC2042BA34E
5 changed files with 45 additions and 82 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
cluster.go
View file

@ -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

11
etcd/embed.go
View file

@ -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) {
@ -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 {
log.Printf("leaseKeepAlive: revokes the lease (ID: %v): %v\n", leaseResp.ID, err)
if _, err := cli.Revoke(context.TODO(), leaseResp.ID); err != nil {
log.Printf("leaseKeepAlive: revokes the lease (ID: %x): %v\n", leaseResp.ID, err)
}
cli.Close()
}
return
@ -754,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()
}
@ -767,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
}

17
executor_test.go
View file

@ -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)
)`,

37
server/server_test.go
View file

@ -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)
}

43
test/cluster.go
View file

@ -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