mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 17:15:56 +00:00
Replace Node(0) by GetCoordinator
This commit is contained in:
Kuba Podgórski
parent
ace4dea46f
commit
32b5c5bcea
3 changed files with 58 additions and 40 deletions
|
|
@ -57,7 +57,7 @@ func (c *Cluster) Query(t testing.TB, index, query string) pilosa.QueryResponse
|
|||
t.Fatal("must have at least one node in cluster to query")
|
||||
}
|
||||
|
||||
return c.GetNode(0).QueryAPI(t, &pilosa.QueryRequest{Index: index, Query: query})
|
||||
return c.GetCoordinator().QueryAPI(t, &pilosa.QueryRequest{Index: index, Query: query})
|
||||
}
|
||||
|
||||
// QueryHTTP executes a PQL query through the HTTP endpoint. It fails
|
||||
|
|
@ -69,7 +69,7 @@ func (c *Cluster) QueryHTTP(t testing.TB, index, query string) (string, error) {
|
|||
t.Fatal("must have at least one node in cluster to query")
|
||||
}
|
||||
|
||||
return c.GetNode(0).Query(t, index, "", query)
|
||||
return c.GetCoordinator().Query(t, index, "", query)
|
||||
}
|
||||
|
||||
// QueryGRPC executes a PQL query through the GRPC endpoint. It fails the
|
||||
|
|
@ -80,7 +80,7 @@ func (c *Cluster) QueryGRPC(t testing.TB, index, query string) *proto.TableRespo
|
|||
t.Fatal("must have at least one node in cluster to query")
|
||||
}
|
||||
|
||||
grpcClient, err := client.NewGRPCClient([]string{fmt.Sprintf("%s:%d", c.GetNode(0).Server.GRPCURI().Host, c.GetNode(0).Server.GRPCURI().Port)}, nil)
|
||||
grpcClient, err := client.NewGRPCClient([]string{fmt.Sprintf("%s:%d", c.GetCoordinator().Server.GRPCURI().Host, c.GetCoordinator().Server.GRPCURI().Port)}, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("getting GRPC client: %v", err)
|
||||
}
|
||||
|
|
@ -139,9 +139,18 @@ func (c *Cluster) GetNode(n int) *Command {
|
|||
// can be any node in the cluster, so we have to use this method in tests which
|
||||
// need to act on the coordinator.
|
||||
func (c *Cluster) GetCoordinator() *Command {
|
||||
for i := range c.Nodes {
|
||||
if c.Nodes[i].IsCoordinator() {
|
||||
return c.Nodes[i]
|
||||
for _, n := range c.Nodes {
|
||||
if n.IsCoordinator() {
|
||||
return n
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Cluster) GetNonCoordinator() *Command {
|
||||
for _, n := range c.Nodes {
|
||||
if !n.IsCoordinator() {
|
||||
return n
|
||||
}
|
||||
}
|
||||
return nil
|
||||
|
|
@ -176,7 +185,7 @@ func (c *Cluster) ImportBits(t testing.TB, index, field string, rowcols [][2]uin
|
|||
rowIDs[i] = bit[0]
|
||||
colIDs[i] = bit[1]
|
||||
}
|
||||
nodes, err := c.GetNode(0).API.ShardNodes(context.Background(), index, shard)
|
||||
nodes, err := c.GetCoordinator().API.ShardNodes(context.Background(), index, shard)
|
||||
if err != nil {
|
||||
t.Fatalf("getting shard nodes: %v", err)
|
||||
}
|
||||
|
|
@ -219,7 +228,7 @@ func (c *Cluster) ImportKeyKey(t testing.TB, index, field string, valAndRecKeys
|
|||
importRequest.RowKeys[i] = vk[0]
|
||||
importRequest.ColumnKeys[i] = vk[1]
|
||||
}
|
||||
err := c.GetNode(0).API.Import(context.Background(), nil, importRequest)
|
||||
err := c.GetCoordinator().API.Import(context.Background(), nil, importRequest)
|
||||
if err != nil {
|
||||
t.Fatalf("importing keykey data: %v", err)
|
||||
}
|
||||
|
|
@ -249,7 +258,7 @@ func (c *Cluster) ImportTimeQuantumKey(t testing.TB, index, field string, entrie
|
|||
importRequest.Timestamps[i] = entry.Ts
|
||||
|
||||
}
|
||||
err := c.GetNode(0).API.Import(context.Background(), nil, importRequest)
|
||||
err := c.GetCoordinator().API.Import(context.Background(), nil, importRequest)
|
||||
if err != nil {
|
||||
t.Fatalf("importing keykey data: %v", err)
|
||||
}
|
||||
|
|
@ -275,7 +284,7 @@ func (c *Cluster) ImportIntKey(t testing.TB, index, field string, pairs []IntKey
|
|||
importRequest.Values[i] = pair.Val
|
||||
importRequest.ColumnKeys[i] = pair.Key
|
||||
}
|
||||
if err := c.GetNode(0).API.ImportValue(context.Background(), nil, importRequest); err != nil {
|
||||
if err := c.GetCoordinator().API.ImportValue(context.Background(), nil, importRequest); err != nil {
|
||||
t.Fatalf("importing IntKey data: %v", err)
|
||||
}
|
||||
}
|
||||
|
|
@ -299,7 +308,7 @@ func (c *Cluster) ImportIntID(t testing.TB, index, field string, pairs []IntID)
|
|||
importRequest.Values[i] = pair.Val
|
||||
importRequest.ColumnIDs[i] = pair.ID
|
||||
}
|
||||
if err := c.GetNode(0).API.ImportValue(context.Background(), nil, importRequest); err != nil {
|
||||
if err := c.GetCoordinator().API.ImportValue(context.Background(), nil, importRequest); err != nil {
|
||||
t.Fatalf("importing IntID data: %v", err)
|
||||
}
|
||||
}
|
||||
|
|
@ -324,7 +333,7 @@ func (c *Cluster) ImportIDKey(t testing.TB, index, field string, pairs []KeyID)
|
|||
importRequest.RowIDs[i] = pair.ID
|
||||
importRequest.ColumnKeys[i] = pair.Key
|
||||
}
|
||||
err := c.GetNode(0).API.Import(context.Background(), nil, importRequest)
|
||||
err := c.GetCoordinator().API.Import(context.Background(), nil, importRequest)
|
||||
if err != nil {
|
||||
t.Fatalf("importing IDKey data: %v", err)
|
||||
}
|
||||
|
|
@ -333,11 +342,11 @@ func (c *Cluster) ImportIDKey(t testing.TB, index, field string, pairs []KeyID)
|
|||
// CreateField creates the index (if necessary) and field specified.
|
||||
func (c *Cluster) CreateField(t testing.TB, index string, iopts pilosa.IndexOptions, field string, fopts ...pilosa.FieldOption) *pilosa.Field {
|
||||
t.Helper()
|
||||
idx, err := c.GetNode(0).API.CreateIndex(context.Background(), index, iopts)
|
||||
idx, err := c.GetCoordinator().API.CreateIndex(context.Background(), index, iopts)
|
||||
if err != nil && !strings.Contains(err.Error(), "index already exists") {
|
||||
t.Fatalf("creating index: %v", err)
|
||||
} else if err != nil { // index exists
|
||||
idx, err = c.GetNode(0).API.Index(context.Background(), index)
|
||||
idx, err = c.GetCoordinator().API.Index(context.Background(), index)
|
||||
if err != nil {
|
||||
t.Fatalf("getting index: %v", err)
|
||||
}
|
||||
|
|
@ -346,7 +355,7 @@ func (c *Cluster) CreateField(t testing.TB, index string, iopts pilosa.IndexOpti
|
|||
t.Logf("existing index options:\n%v\ndon't match given opts:\n%v\n in pilosa/test.Cluster.CreateField", idx.Options(), iopts)
|
||||
}
|
||||
|
||||
f, err := c.GetNode(0).API.CreateField(context.Background(), index, field, fopts...)
|
||||
f, err := c.GetCoordinator().API.CreateField(context.Background(), index, field, fopts...)
|
||||
// we'll assume the field doesn't exist because checking if the options
|
||||
// match seems painful.
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -139,13 +139,21 @@ func (c *ClusterSnapshot) PartitionNodes(partitionID int) []*Node {
|
|||
// field keys. The primary could be any node in the cluster, but we arbitrarily
|
||||
// define it to be the node responsible for partition 0.
|
||||
func (c *ClusterSnapshot) PrimaryFieldTranslationNode() *Node {
|
||||
return c.PrimaryPartitionNode(0)
|
||||
for _, n := range c.Nodes {
|
||||
if n.IsCoordinator {
|
||||
return n
|
||||
}
|
||||
}
|
||||
return nil
|
||||
|
||||
// return c.PrimaryPartitionNode(0)
|
||||
}
|
||||
|
||||
// IsPrimaryFieldTranslationNode returns true if nodeID represents the primary
|
||||
// node responsible for field translation.
|
||||
func (c *ClusterSnapshot) IsPrimaryFieldTranslationNode(nodeID string) bool {
|
||||
return c.PrimaryFieldTranslationNode().ID == nodeID
|
||||
return c.IsCoordinatorNode(nodeID)
|
||||
//c.PrimaryFieldTranslationNode().ID == nodeID
|
||||
}
|
||||
|
||||
// IsCoordinatorNode returns true if nodeID represents the coordinator
|
||||
|
|
|
|||
|
|
@ -483,28 +483,28 @@ func TestTranslation_Replication(t *testing.T) {
|
|||
)
|
||||
defer c.Close()
|
||||
|
||||
node0 := c.GetNode(0)
|
||||
node1 := c.GetNode(1)
|
||||
coord := c.GetCoordinator()
|
||||
other := c.GetNonCoordinator()
|
||||
|
||||
ctx := context.Background()
|
||||
idx := "i"
|
||||
field := "f"
|
||||
|
||||
// Create an index with keys.
|
||||
if _, err := node0.API.CreateIndex(ctx, idx,
|
||||
if _, err := coord.API.CreateIndex(ctx, idx,
|
||||
pilosa.IndexOptions{
|
||||
Keys: true,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if _, err := node0.API.CreateField(ctx, idx, field); err != nil {
|
||||
if _, err := coord.API.CreateField(ctx, idx, field); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Write data on first node.
|
||||
// these keys are a minimal example to reproduce the problem for the case of a 3-node cluster with replication factor 2
|
||||
if _, err := node0.Queryf(t, idx, "", `
|
||||
if _, err := coord.Queryf(t, idx, "", `
|
||||
Set("x1", f=1)
|
||||
Set("x2", f=1)
|
||||
`); err != nil {
|
||||
|
|
@ -513,14 +513,14 @@ func TestTranslation_Replication(t *testing.T) {
|
|||
|
||||
exp := `{"results":[{"attrs":{},"columns":[],"keys":["x1","x2"]}]}`
|
||||
|
||||
if !test.CheckClusterState(node0, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", node0.API.State())
|
||||
} else if !test.CheckClusterState(node1, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", node1.API.State())
|
||||
if !test.CheckClusterState(coord, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", coord.API.State())
|
||||
} else if !test.CheckClusterState(other, pilosa.ClusterStateNormal, 1000) {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", other.API.State())
|
||||
}
|
||||
|
||||
// Verify the data exists
|
||||
node0.QueryExpect(t, idx, "", `Row(f=1)`, exp)
|
||||
coord.QueryExpect(t, idx, "", `Row(f=1)`, exp)
|
||||
|
||||
// Kill one node.
|
||||
if err := c.CloseAndRemove(1); err != nil {
|
||||
|
|
@ -528,7 +528,7 @@ func TestTranslation_Replication(t *testing.T) {
|
|||
}
|
||||
|
||||
// Verify the data exists with one node down
|
||||
node0.QueryExpect(t, idx, "", `Row(f=1)`, exp)
|
||||
coord.QueryExpect(t, idx, "", `Row(f=1)`, exp)
|
||||
})
|
||||
}
|
||||
|
||||
|
|
@ -557,8 +557,8 @@ func TestTranslation_Coordinator(t *testing.T) {
|
|||
)
|
||||
defer c.Close()
|
||||
|
||||
node0 := c.GetNode(0)
|
||||
node1 := c.GetNode(1)
|
||||
node0 := c.GetCoordinator()
|
||||
node1 := c.GetNonCoordinator()
|
||||
|
||||
ctx := context.Background()
|
||||
idx := "i"
|
||||
|
|
@ -643,23 +643,24 @@ func TestTranslation_TranslateIDsOnCluster(t *testing.T) {
|
|||
)
|
||||
defer c.Close()
|
||||
|
||||
node0 := c.GetNode(0)
|
||||
node3 := c.GetNode(3)
|
||||
coord := c.GetCoordinator()
|
||||
other := c.GetNonCoordinator()
|
||||
|
||||
ctx := context.Background()
|
||||
idx, fld := "i", "f"
|
||||
// Create an index with keys.
|
||||
if _, err := node0.API.CreateIndex(ctx, idx, pilosa.IndexOptions{Keys: true}); err != nil {
|
||||
if _, err := coord.API.CreateIndex(ctx, idx, pilosa.IndexOptions{Keys: true}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Create an index with keys.
|
||||
if _, err := node0.API.CreateField(ctx, idx, fld, pilosa.OptFieldKeys()); err != nil {
|
||||
if _, err := coord.API.CreateField(ctx, idx, fld, pilosa.OptFieldKeys()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
keys := []string{"k0", "k1", "k2", "k3", "k4", "k5", "k6", "k7", "k8", "k9"}
|
||||
// write a new key and get id
|
||||
req, err := node0.API.Serializer.Marshal(&pilosa.TranslateKeysRequest{
|
||||
req, err := coord.API.Serializer.Marshal(&pilosa.TranslateKeysRequest{
|
||||
Index: idx,
|
||||
Field: fld,
|
||||
Keys: keys,
|
||||
|
|
@ -668,20 +669,20 @@ func TestTranslation_TranslateIDsOnCluster(t *testing.T) {
|
|||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if buf, err := node0.API.TranslateKeys(ctx, bytes.NewReader(req)); err != nil {
|
||||
if buf, err := coord.API.TranslateKeys(ctx, bytes.NewReader(req)); err != nil {
|
||||
t.Fatal(err)
|
||||
} else {
|
||||
var (
|
||||
respKeys pilosa.TranslateKeysResponse
|
||||
respIDs pilosa.TranslateIDsResponse
|
||||
)
|
||||
if err = node0.API.Serializer.Unmarshal(buf, &respKeys); err != nil {
|
||||
if err = other.API.Serializer.Unmarshal(buf, &respKeys); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ids := respKeys.IDs
|
||||
|
||||
// translate ids
|
||||
req, err = node3.API.Serializer.Marshal(&pilosa.TranslateIDsRequest{
|
||||
req, err = other.API.Serializer.Marshal(&pilosa.TranslateIDsRequest{
|
||||
Index: idx,
|
||||
Field: fld,
|
||||
IDs: ids,
|
||||
|
|
@ -689,10 +690,10 @@ func TestTranslation_TranslateIDsOnCluster(t *testing.T) {
|
|||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if buf, err = node3.API.TranslateIDs(ctx, bytes.NewReader(req)); err != nil {
|
||||
if buf, err = other.API.TranslateIDs(ctx, bytes.NewReader(req)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = node3.API.Serializer.Unmarshal(buf, &respIDs); err != nil {
|
||||
if err = other.API.Serializer.Unmarshal(buf, &respIDs); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if !reflect.DeepEqual(respIDs.Keys, keys) {
|
||||
t.Fatalf("TranslateIDs(%+v): expected: %+v, got: %+v", ids, keys, respIDs.Keys)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue