From 32b5c5bceac517a8a6bb81998ee3b0a15032d05f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Mon, 25 Jan 2021 19:54:25 +0100 Subject: [PATCH] Replace Node(0) by GetCoordinator --- test/cluster.go | 39 ++++++++++++++++++++++-------------- topology/snapshot.go | 12 +++++++++-- translator_test.go | 47 ++++++++++++++++++++++---------------------- 3 files changed, 58 insertions(+), 40 deletions(-) diff --git a/test/cluster.go b/test/cluster.go index c7e372321..a72ccc497 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -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 { diff --git a/topology/snapshot.go b/topology/snapshot.go index 172152066..cb79b7582 100644 --- a/topology/snapshot.go +++ b/topology/snapshot.go @@ -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 diff --git a/translator_test.go b/translator_test.go index e47ee9233..b57b3e397 100644 --- a/translator_test.go +++ b/translator_test.go @@ -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)