mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 09:05:55 +00:00
remove remaining references to "coordinator"
This commit is contained in:
Travis
parent
c16c543274
commit
2bbe1fdde0
19 changed files with 137 additions and 210 deletions
16
api.go
16
api.go
|
|
@ -795,14 +795,14 @@ func (api *API) TranslateData(ctx context.Context, indexName string, partition i
|
|||
}
|
||||
|
||||
// Hosts returns a list of the hosts in the cluster including their ID,
|
||||
// URL, and which is the coordinator.
|
||||
// URL, and which is the primary.
|
||||
func (api *API) Hosts(ctx context.Context) []*topology.Node {
|
||||
span, _ := tracing.StartSpanFromContext(ctx, "API.Hosts")
|
||||
defer span.Finish()
|
||||
return api.cluster.Nodes()
|
||||
}
|
||||
|
||||
// Node gets the ID, URI and coordinator status for this particular node.
|
||||
// Node gets the ID, URI and primary status for this particular node.
|
||||
func (api *API) Node() *topology.Node {
|
||||
return api.server.node()
|
||||
}
|
||||
|
|
@ -813,7 +813,7 @@ func (api *API) NodeID() string {
|
|||
return api.server.nodeID
|
||||
}
|
||||
|
||||
// PrimaryNode returns the coordinator node for the cluster.
|
||||
// PrimaryNode returns the primary node for the cluster.
|
||||
func (api *API) PrimaryNode() *topology.Node {
|
||||
// Create a snapshot of the cluster to use for node/partition calculations.
|
||||
snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
|
||||
|
|
@ -1372,7 +1372,7 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest,
|
|||
"field", req.Field)
|
||||
|
||||
// Unless explicitly ignoring key validation (meaning keys have been
|
||||
// translated to ids in a previous step at the coordinator node), then
|
||||
// translated to ids in a previous step at the primary node), then
|
||||
// check to see if keys need translation.
|
||||
if !options.IgnoreKeyCheck {
|
||||
// Translate row keys.
|
||||
|
|
@ -1511,7 +1511,7 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu
|
|||
"index", req.Index,
|
||||
"field", req.Field)
|
||||
// Unless explicitly ignoring key validation (meaning keys have been
|
||||
// translate to ids in a previous step at the coordinator node), then
|
||||
// translate to ids in a previous step at the primary node), then
|
||||
// check to see if keys need translation.
|
||||
if !options.IgnoreKeyCheck {
|
||||
// Translate column keys.
|
||||
|
|
@ -2147,7 +2147,7 @@ func (api *API) ReserveIDs(key IDAllocKey, session [32]byte, offset uint64, coun
|
|||
return api.holder.ida.reserve(key, session, offset, count)
|
||||
}
|
||||
|
||||
return nil, errors.New("cannot reserve IDs on a non-coordinator node")
|
||||
return nil, errors.New("cannot reserve IDs on a non-primary node")
|
||||
}
|
||||
|
||||
func (api *API) CommitIDs(key IDAllocKey, session [32]byte, count uint64) error {
|
||||
|
|
@ -2162,7 +2162,7 @@ func (api *API) CommitIDs(key IDAllocKey, session [32]byte, count uint64) error
|
|||
return api.holder.ida.commit(key, session, count)
|
||||
}
|
||||
|
||||
return errors.New("cannot commit IDs on a non-coordinator node")
|
||||
return errors.New("cannot commit IDs on a non-primary node")
|
||||
}
|
||||
|
||||
func (api *API) ResetIDAlloc(index string) error {
|
||||
|
|
@ -2177,7 +2177,7 @@ func (api *API) ResetIDAlloc(index string) error {
|
|||
return api.holder.ida.reset(index)
|
||||
}
|
||||
|
||||
return errors.New("cannot reset IDs on a non-coordinator node")
|
||||
return errors.New("cannot reset IDs on a non-primary node")
|
||||
}
|
||||
|
||||
// TranslateIndexDB is an internal function to load the index keys database
|
||||
|
|
|
|||
|
|
@ -224,7 +224,7 @@ func TestAPI_Import(t *testing.T) {
|
|||
|
||||
colKeys = colKeys[:N]
|
||||
|
||||
// Import data with keys to the coordinator (node0) and verify that it gets
|
||||
// Import data with keys to the primary and verify that it gets
|
||||
// translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher)
|
||||
req := &pilosa.ImportRequest{
|
||||
Index: indexName,
|
||||
|
|
@ -302,7 +302,7 @@ func TestAPI_ImportValue(t *testing.T) {
|
|||
)
|
||||
defer c.Close()
|
||||
|
||||
coord := c.GetCoordinator()
|
||||
coord := c.GetPrimary()
|
||||
m0 := c.GetNode(0)
|
||||
m1 := c.GetNode(1)
|
||||
|
||||
|
|
@ -329,7 +329,7 @@ func TestAPI_ImportValue(t *testing.T) {
|
|||
// Column keys are sharded so their order is not guaranteed.
|
||||
colKeys := []string{"col10", "col8", "col9", "col6", "col7", "col4", "col5", "col2", "col3", "col1"}
|
||||
|
||||
// Import data with keys to the coordinator (node0) and verify that it gets
|
||||
// Import data with keys to the primary and verify that it gets
|
||||
// translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher)
|
||||
req := &pilosa.ImportValueRequest{
|
||||
Index: index,
|
||||
|
|
|
|||
61
cluster.go
61
cluster.go
|
|
@ -173,28 +173,17 @@ func (c *cluster) abortAntiEntropy() {
|
|||
}
|
||||
}
|
||||
|
||||
func (c *cluster) coordinatorNode() *topology.Node {
|
||||
return c.unprotectedCoordinatorNode()
|
||||
func (c *cluster) primaryNode() *topology.Node {
|
||||
return c.unprotectedPrimaryNode()
|
||||
}
|
||||
|
||||
// unprotectedCoordinatorNode returns the coordinator node.
|
||||
func (c *cluster) unprotectedCoordinatorNode() *topology.Node {
|
||||
// unprotectedPrimaryNode returns the primary node.
|
||||
func (c *cluster) unprotectedPrimaryNode() *topology.Node {
|
||||
// Create a snapshot of the cluster to use for node/partition calculations.
|
||||
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
|
||||
return snap.PrimaryFieldTranslationNode()
|
||||
}
|
||||
|
||||
// isCoordinator is true if this node is the coordinator.
|
||||
func (c *cluster) isCoordinator() bool {
|
||||
return c.unprotectedIsCoordinator()
|
||||
}
|
||||
|
||||
func (c *cluster) unprotectedIsCoordinator() bool {
|
||||
// Create a snapshot of the cluster to use for node/partition calculations.
|
||||
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
|
||||
return snap.PrimaryFieldTranslationNode().ID == c.Node.ID
|
||||
}
|
||||
|
||||
func (c *cluster) applySchemaWithNewShards(schema *Schema) error {
|
||||
if schema == nil || len(schema.Indexes) == 0 {
|
||||
return nil
|
||||
|
|
@ -1127,7 +1116,7 @@ func (c *cluster) followResizeInstruction(ctx context.Context, instr *ResizeInst
|
|||
if err != nil {
|
||||
// For now it is an acceptable error if the fragment is not found
|
||||
// on the remote node. This occurs when a shard has been skipped and
|
||||
// therefore doesn't contain data. The coordinator correctly determined
|
||||
// therefore doesn't contain data. The primary correctly determined
|
||||
// the resize instruction to retrieve the shard, but it doesn't have data.
|
||||
// TODO: figure out a way to distinguish from "fragment not found" errors
|
||||
// which are true errors and which simply mean the fragment doesn't have data.
|
||||
|
|
@ -1336,21 +1325,21 @@ func (c *cluster) unprotectedPrimaryReplicaNode() *topology.Node {
|
|||
|
||||
// translateFieldKeys is basically a wrapper around
|
||||
// field.TranslateStore().TranslateKey(key), but in
|
||||
// the case where the local node is not coordinator, then this method will forward the translation
|
||||
// request to the coordinator.
|
||||
// the case where the local node is not primary, then this method will forward the translation
|
||||
// request to the primary.
|
||||
func (c *cluster) translateFieldKeys(ctx context.Context, field *Field, keys []string, writable bool) (ids []uint64, err error) {
|
||||
// Create a snapshot of the cluster to use for node/partition calculations.
|
||||
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
|
||||
|
||||
primary := snap.PrimaryFieldTranslationNode()
|
||||
if primary == nil {
|
||||
return nil, errors.Errorf("translating field(%s/%s) keys(%v) - cannot find coordinator node", field.Index(), field.Name(), keys)
|
||||
return nil, errors.Errorf("translating field(%s/%s) keys(%v) - cannot find primary node", field.Index(), field.Name(), keys)
|
||||
}
|
||||
|
||||
if c.Node.ID == primary.ID {
|
||||
ids, err = field.TranslateStore().TranslateKeys(keys, writable)
|
||||
} else {
|
||||
// If it's writable, then forward the request to the coordinator.
|
||||
// If it's writable, then forward the request to the primary.
|
||||
ids, err = c.InternalClient.TranslateKeysNode(ctx, &primary.URI, field.Index(), field.Name(), keys, writable)
|
||||
}
|
||||
|
||||
|
|
@ -1399,18 +1388,18 @@ func (c *cluster) findFieldKeys(ctx context.Context, field *Field, keys ...strin
|
|||
}
|
||||
|
||||
// It is possible that the missing keys exist, but have not been synced to the local replica.
|
||||
coordinator := c.coordinatorNode()
|
||||
if coordinator == nil {
|
||||
return nil, errors.Errorf("translating field(%s/%s) keys(%v) - cannot find coordinator node", field.Index(), field.Name(), keys)
|
||||
primary := c.primaryNode()
|
||||
if primary == nil {
|
||||
return nil, errors.Errorf("translating field(%s/%s) keys(%v) - cannot find primary node", field.Index(), field.Name(), keys)
|
||||
}
|
||||
if c.Node.ID == coordinator.ID {
|
||||
if c.Node.ID == primary.ID {
|
||||
// The local copy is the authoritative copy.
|
||||
return localTranslations, nil
|
||||
}
|
||||
|
||||
// Forward the missing keys to the coordinator.
|
||||
// The coordinator has the authoritative copy.
|
||||
remoteTranslations, err := c.InternalClient.FindFieldKeysNode(ctx, &coordinator.URI, field.Index(), field.Name(), missing...)
|
||||
// Forward the missing keys to the primary.
|
||||
// The primary has the authoritative copy.
|
||||
remoteTranslations, err := c.InternalClient.FindFieldKeysNode(ctx, &primary.URI, field.Index(), field.Name(), missing...)
|
||||
if err != nil {
|
||||
return nil, errors.Wrapf(err, "translating field(%s/%s) keys(%v) remotely", field.Index(), field.Name(), keys)
|
||||
}
|
||||
|
|
@ -1435,12 +1424,12 @@ func (c *cluster) createFieldKeys(ctx context.Context, field *Field, keys ...str
|
|||
return nil, errors.Wrap(ErrTranslatingKeyNotFound, "field is not keyed")
|
||||
}
|
||||
|
||||
// The coordinator is the only node that can create field keys, since it owns the authoritative copy.
|
||||
coordinator := c.coordinatorNode()
|
||||
if coordinator == nil {
|
||||
return nil, errors.Errorf("translating field(%s/%s) keys(%v) - cannot find coordinator node", field.Index(), field.Name(), keys)
|
||||
// The primary is the only node that can create field keys, since it owns the authoritative copy.
|
||||
primary := c.primaryNode()
|
||||
if primary == nil {
|
||||
return nil, errors.Errorf("translating field(%s/%s) keys(%v) - cannot find primary node", field.Index(), field.Name(), keys)
|
||||
}
|
||||
if c.Node.ID == coordinator.ID {
|
||||
if c.Node.ID == primary.ID {
|
||||
// The local copy is the authoritative copy.
|
||||
return field.TranslateStore().CreateKeys(keys...)
|
||||
}
|
||||
|
|
@ -1473,8 +1462,8 @@ func (c *cluster) createFieldKeys(ctx context.Context, field *Field, keys ...str
|
|||
return localTranslations, nil
|
||||
}
|
||||
|
||||
// Forward the missing keys to the coordinator to be created.
|
||||
remoteTranslations, err := c.InternalClient.CreateFieldKeysNode(ctx, &coordinator.URI, field.Index(), field.Name(), missing...)
|
||||
// Forward the missing keys to the primary to be created.
|
||||
remoteTranslations, err := c.InternalClient.CreateFieldKeysNode(ctx, &primary.URI, field.Index(), field.Name(), missing...)
|
||||
if err != nil {
|
||||
return nil, errors.Wrapf(err, "translating field(%s/%s) keys(%v) remotely", field.Index(), field.Name(), keys)
|
||||
}
|
||||
|
|
@ -1516,7 +1505,7 @@ func (c *cluster) translateFieldListIDs(field *Field, ids []uint64) (keys []stri
|
|||
|
||||
primary := snap.PrimaryFieldTranslationNode()
|
||||
if primary == nil {
|
||||
return nil, errors.Errorf("translating field(%s/%s) ids(%v) - cannot find coordinator node", field.Index(), field.Name(), ids)
|
||||
return nil, errors.Errorf("translating field(%s/%s) ids(%v) - cannot find primary node", field.Index(), field.Name(), ids)
|
||||
}
|
||||
|
||||
if c.Node.ID == primary.ID {
|
||||
|
|
@ -2018,7 +2007,7 @@ type DeleteViewMessage struct {
|
|||
View string
|
||||
}
|
||||
|
||||
// ResizeInstructionComplete is an internal message to the coordinator indicating
|
||||
// ResizeInstructionComplete is an internal message to the primary indicating
|
||||
// that the resize instructions performed on a single node have completed.
|
||||
type ResizeInstructionComplete struct {
|
||||
JobID int64
|
||||
|
|
|
|||
|
|
@ -613,44 +613,6 @@ func TestCluster_PreviousNode(t *testing.T) {
|
|||
})
|
||||
}
|
||||
|
||||
// NEXT: move this test to internal and unexport IsCoordinator
|
||||
func TestCluster_Coordinator(t *testing.T) {
|
||||
// TODO check if this test still makes sense
|
||||
t.Skip()
|
||||
|
||||
const urisCount = 2
|
||||
var uris []pnet.URI
|
||||
if err := port.GetPorts(func(ports []int) error {
|
||||
for i := 0; i < urisCount; i++ {
|
||||
uris = append(uris, NewTestURIFromHostPort(fmt.Sprintf("node%d", i), uint16(ports[i])))
|
||||
}
|
||||
return nil
|
||||
}, urisCount, 10); err != nil {
|
||||
t.Fatalf("getting ports: %v", err)
|
||||
}
|
||||
|
||||
node1 := &topology.Node{ID: "node1", URI: uris[0]}
|
||||
node2 := &topology.Node{ID: "node2", URI: uris[1]}
|
||||
noder := topology.NewLocalNoder([]*topology.Node{node1, node2})
|
||||
|
||||
c1 := *newCluster()
|
||||
c1.Node = node1
|
||||
// c1.Coordinator = node1.ID
|
||||
c1.noder = noder
|
||||
c2 := *newCluster()
|
||||
c2.Node = node2
|
||||
// c2.Coordinator = node1.ID
|
||||
c2.noder = noder
|
||||
|
||||
t.Run("IsCoordinator", func(t *testing.T) {
|
||||
if !c1.isCoordinator() {
|
||||
t.Errorf("!IsCoordinator error: %v", c1.Node)
|
||||
} else if c2.isCoordinator() {
|
||||
t.Errorf("IsCoordinator error: %v", c2.Node)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestAE(t *testing.T) {
|
||||
t.Run("AbortDoesn'tBlockUninitialized", func(t *testing.T) {
|
||||
c := newCluster()
|
||||
|
|
|
|||
|
|
@ -116,7 +116,7 @@ func Test_RandomQuery(t *testing.T) {
|
|||
timestamps = timestamps[:N]
|
||||
}
|
||||
|
||||
// Import data with keys to the coordinator (node0) and verify that it gets
|
||||
// Import data with keys to the primary and verify that it gets
|
||||
// translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher)
|
||||
req := &pilosa.ImportRequest{
|
||||
Index: indexes[i],
|
||||
|
|
|
|||
|
|
@ -261,7 +261,7 @@ func (cmd *ImportCommand) bufferBits(ctx context.Context, useColumnKeys, useRowK
|
|||
func (cmd *ImportCommand) importBits(ctx context.Context, useColumnKeys, useRowKeys bool, bits []pilosa.Bit) error {
|
||||
logger := log.New(cmd.Stderr, "", log.LstdFlags)
|
||||
|
||||
// If keys are used, all bits are sent to the primary translate store (i.e. coordinator).
|
||||
// If keys are used, all bits are sent to the primary translate store.
|
||||
if useColumnKeys || useRowKeys {
|
||||
logger.Printf("importing keys: n=%d", len(bits))
|
||||
if err := cmd.client.ImportK(ctx, cmd.Index, cmd.Field, bits, pilosa.OptImportOptionsClear(cmd.Clear)); err != nil {
|
||||
|
|
|
|||
|
|
@ -77,7 +77,7 @@ func TestAPI_SimplerOneNode_ImportColumnKey(t *testing.T) {
|
|||
// Keys are sharded so ordering is not guaranteed.
|
||||
colKeys := []string{"col10", "col8", "col9", "col6", "col7", "col4", "col5", "col2", "col3", "col1"}
|
||||
|
||||
// Import data with keys to the coordinator (node0) and verify that it gets
|
||||
// Import data with keys to the primary and verify that it gets
|
||||
// translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher)
|
||||
req := &pilosa.ImportRequest{
|
||||
Index: indexName,
|
||||
|
|
|
|||
|
|
@ -5554,7 +5554,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
|
|||
|
||||
// If this is the coordinating node then start with all nodes in the cluster.
|
||||
//
|
||||
// However, if this request is being sent from the coordinator then all
|
||||
// However, if this request is being sent from the primary then all
|
||||
// processing should be done locally so we start with just the local node.
|
||||
var nodes []*topology.Node
|
||||
if !opt.Remote {
|
||||
|
|
|
|||
|
|
@ -2960,11 +2960,11 @@ func TestExecutor_Execute_Remote_Row(t *testing.T) {
|
|||
hldr0 := c.GetHolder(0)
|
||||
hldr1 := c.GetHolder(1)
|
||||
|
||||
_, err := c.GetCoordinator().API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{})
|
||||
_, err := c.GetPrimary().API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{})
|
||||
if err != nil {
|
||||
t.Fatalf("creating index: %v", err)
|
||||
}
|
||||
_, err = c.GetCoordinator().API.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, pilosa.DefaultCacheSize))
|
||||
_, err = c.GetPrimary().API.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, pilosa.DefaultCacheSize))
|
||||
if err != nil {
|
||||
t.Fatalf("creating field: %v", err)
|
||||
}
|
||||
|
|
@ -2997,7 +2997,7 @@ func TestExecutor_Execute_Remote_Row(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("remote with timestamp", func(t *testing.T) {
|
||||
_, err = c.GetCoordinator().API.CreateField(context.Background(), "i", "z", pilosa.OptFieldTypeTime("Y"))
|
||||
_, err = c.GetPrimary().API.CreateField(context.Background(), "i", "z", pilosa.OptFieldTypeTime("Y"))
|
||||
if err != nil {
|
||||
t.Fatalf("creating field: %v", err)
|
||||
}
|
||||
|
|
@ -3012,7 +3012,7 @@ func TestExecutor_Execute_Remote_Row(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("remote topn", func(t *testing.T) {
|
||||
_, err = c.GetCoordinator().API.CreateField(context.Background(), "i", "fn", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 100))
|
||||
_, err = c.GetPrimary().API.CreateField(context.Background(), "i", "fn", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 100))
|
||||
if err != nil {
|
||||
t.Fatalf("creating field: %v", err)
|
||||
}
|
||||
|
|
@ -3075,7 +3075,7 @@ func TestExecutor_Execute_Remote_Row(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("remote groupBy on ints", func(t *testing.T) {
|
||||
_, err = c.GetCoordinator().API.CreateField(context.Background(), "i", "fint", pilosa.OptFieldTypeInt(-1000, 1000))
|
||||
_, err = c.GetPrimary().API.CreateField(context.Background(), "i", "fint", pilosa.OptFieldTypeInt(-1000, 1000))
|
||||
if err != nil {
|
||||
t.Fatalf("creating field: %v", err)
|
||||
}
|
||||
|
|
@ -3117,7 +3117,7 @@ func TestExecutor_Execute_Remote_Row(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("groupBy on ints with offset regression", func(t *testing.T) {
|
||||
_, err = c.GetCoordinator().API.CreateField(context.Background(), "i", "hint", pilosa.OptFieldTypeInt(1, 1000))
|
||||
_, err = c.GetPrimary().API.CreateField(context.Background(), "i", "hint", pilosa.OptFieldTypeInt(1, 1000))
|
||||
if err != nil {
|
||||
t.Fatalf("creating field: %v", err)
|
||||
}
|
||||
|
|
@ -3148,12 +3148,12 @@ func TestExecutor_Execute_Remote_Row(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("Row on ints with ASSIGN condition", func(t *testing.T) {
|
||||
_, err := c.GetCoordinator().API.CreateIndex(context.Background(), "intidx", pilosa.IndexOptions{})
|
||||
_, err := c.GetPrimary().API.CreateIndex(context.Background(), "intidx", pilosa.IndexOptions{})
|
||||
if err != nil {
|
||||
t.Fatalf("creating index: %v", err)
|
||||
}
|
||||
|
||||
_, err = c.GetCoordinator().API.CreateField(context.Background(), "intidx", "gint", pilosa.OptFieldTypeInt(-1000, 1000))
|
||||
_, err = c.GetPrimary().API.CreateField(context.Background(), "intidx", "gint", pilosa.OptFieldTypeInt(-1000, 1000))
|
||||
if err != nil {
|
||||
t.Fatalf("creating field: %v", err)
|
||||
}
|
||||
|
|
@ -3183,12 +3183,12 @@ func TestExecutor_Execute_Remote_Row(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("Row on decimals with ASSIGN condition", func(t *testing.T) {
|
||||
_, err := c.GetCoordinator().API.CreateIndex(context.Background(), "decidx", pilosa.IndexOptions{})
|
||||
_, err := c.GetPrimary().API.CreateIndex(context.Background(), "decidx", pilosa.IndexOptions{})
|
||||
if err != nil {
|
||||
t.Fatalf("creating index: %v", err)
|
||||
}
|
||||
|
||||
_, err = c.GetCoordinator().API.CreateField(context.Background(), "decidx", "fdec", pilosa.OptFieldTypeDecimal(0))
|
||||
_, err = c.GetPrimary().API.CreateField(context.Background(), "decidx", "fdec", pilosa.OptFieldTypeDecimal(0))
|
||||
if err != nil {
|
||||
t.Fatalf("creating field: %v", err)
|
||||
}
|
||||
|
|
@ -3217,19 +3217,19 @@ func TestExecutor_Execute_Remote_Row(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("Row on foreign key with ASSIGN condition", func(t *testing.T) {
|
||||
_, err := c.GetCoordinator().API.CreateIndex(context.Background(), "parent", pilosa.IndexOptions{Keys: true})
|
||||
_, err := c.GetPrimary().API.CreateIndex(context.Background(), "parent", pilosa.IndexOptions{Keys: true})
|
||||
if err != nil {
|
||||
t.Fatalf("creating index: %v", err)
|
||||
}
|
||||
_, err = c.GetCoordinator().API.CreateField(context.Background(), "parent", "general", pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, pilosa.DefaultCacheSize))
|
||||
_, err = c.GetPrimary().API.CreateField(context.Background(), "parent", "general", pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, pilosa.DefaultCacheSize))
|
||||
if err != nil {
|
||||
t.Fatalf("creating field: %v", err)
|
||||
}
|
||||
_, err = c.GetCoordinator().API.CreateIndex(context.Background(), "child", pilosa.IndexOptions{Keys: false})
|
||||
_, err = c.GetPrimary().API.CreateIndex(context.Background(), "child", pilosa.IndexOptions{Keys: false})
|
||||
if err != nil {
|
||||
t.Fatalf("creating index: %v", err)
|
||||
}
|
||||
_, err = c.GetCoordinator().API.CreateField(context.Background(), "child", "parentid",
|
||||
_, err = c.GetPrimary().API.CreateField(context.Background(), "child", "parentid",
|
||||
pilosa.OptFieldForeignIndex("parent"),
|
||||
pilosa.OptFieldTypeInt(-9223372036854775808, 9223372036854775807),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1755,7 +1755,7 @@ func (s *holderSyncer) resetTranslationSync() error {
|
|||
return errors.Wrap(err, "initialize index translate replication")
|
||||
}
|
||||
|
||||
// Connect to coordinator to stream field data.
|
||||
// Connect to primary to stream field data.
|
||||
if err := s.initializeFieldTranslateReplication(snap); err != nil {
|
||||
return errors.Wrap(err, "initialize field translate replication")
|
||||
}
|
||||
|
|
@ -1826,7 +1826,7 @@ func (s *holderSyncer) stopTranslationSync() error {
|
|||
|
||||
// setTranslateReadOnlyFlags updates all translation stores to enable or disable
|
||||
// writing new translation keys. Index stores are writable if the node owns the
|
||||
// partition. Field stores are writable if the node is the coordinator.
|
||||
// partition. Field stores are writable if the node is the primary.
|
||||
func (s *holderSyncer) setTranslateReadOnlyFlags(snap *topology.ClusterSnapshot) {
|
||||
s.Cluster.mu.RLock()
|
||||
isPrimaryFieldTranslator := snap.IsPrimaryFieldTranslationNode(s.Cluster.Node.ID)
|
||||
|
|
@ -1920,9 +1920,9 @@ func (s *holderSyncer) initializeIndexTranslateReplication(snap *topology.Cluste
|
|||
return nil
|
||||
}
|
||||
|
||||
// initializeFieldTranslateReplication connects the coordinator to stream field data.
|
||||
// initializeFieldTranslateReplication connects the primary to stream field data.
|
||||
func (s *holderSyncer) initializeFieldTranslateReplication(snap *topology.ClusterSnapshot) error {
|
||||
// Skip if coordinator.
|
||||
// Skip if primary.
|
||||
if snap.IsPrimaryFieldTranslationNode(s.Cluster.Node.ID) {
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -200,15 +200,15 @@ func (c *InternalClient) CreateIndex(ctx context.Context, index string, opt pilo
|
|||
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.CreateIndex")
|
||||
defer span.Finish()
|
||||
|
||||
// Get the coordinator node. Schema changes must go through
|
||||
// coordinator to avoid weird race conditions.
|
||||
// Get the primary node. Schema changes must go through
|
||||
// primary to avoid weird race conditions.
|
||||
nodes, err := c.Nodes(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("getting nodes: %s", err)
|
||||
}
|
||||
coord := getPrimaryNode(nodes)
|
||||
if coord == nil {
|
||||
return fmt.Errorf("could not find the coordinator node")
|
||||
return fmt.Errorf("could not find the primary node")
|
||||
}
|
||||
|
||||
// Encode query request.
|
||||
|
|
@ -437,13 +437,13 @@ func (c *InternalClient) ImportK(ctx context.Context, index, field string, bits
|
|||
return fmt.Errorf("Error Creating Payload: %s", err)
|
||||
}
|
||||
|
||||
// Get the coordinator node; all bits are sent to the
|
||||
// primary translate store (i.e. coordinator).
|
||||
// Get the primary node; all bits are sent to the
|
||||
// primary translate store (i.e. primary).
|
||||
// TODO... is that right^^?
|
||||
// RESPONSE: It looks like in ctl/import.go, we could change the
|
||||
// logic in ImportCommand.importBits() to only use ImportK
|
||||
// when useRowKeys = true. It's no longer necessary to
|
||||
// send column key translations to the coordinator (although
|
||||
// send column key translations to the primary (although
|
||||
// it should still work). As far as I know, the only thing
|
||||
// that uses ImportK is the pilosa import sub-command.
|
||||
nodes, err := c.Nodes(ctx)
|
||||
|
|
@ -452,7 +452,7 @@ func (c *InternalClient) ImportK(ctx context.Context, index, field string, bits
|
|||
}
|
||||
coord := getPrimaryNode(nodes)
|
||||
if coord == nil {
|
||||
return fmt.Errorf("could not find the coordinator node")
|
||||
return fmt.Errorf("could not find the primary node")
|
||||
}
|
||||
|
||||
// Import to node.
|
||||
|
|
@ -656,15 +656,15 @@ func (c *InternalClient) ImportValueK(ctx context.Context, index, field string,
|
|||
}
|
||||
}
|
||||
|
||||
// Get the coordinator node; all bits are sent to the
|
||||
// primary translate store (i.e. coordinator).
|
||||
// Get the primary node; all bits are sent to the
|
||||
// primary translate store.
|
||||
nodes, err := c.Nodes(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("getting nodes: %s", err)
|
||||
}
|
||||
coord := getPrimaryNode(nodes)
|
||||
if coord == nil {
|
||||
return fmt.Errorf("could not find the coordinator node")
|
||||
return fmt.Errorf("could not find the primary node")
|
||||
}
|
||||
|
||||
// Import to node.
|
||||
|
|
@ -966,15 +966,15 @@ func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, fiel
|
|||
return errors.Wrap(err, "marshaling")
|
||||
}
|
||||
|
||||
// Get the coordinator node. Schema changes must go through
|
||||
// coordinator to avoid weird race conditions.
|
||||
// Get the primary node. Schema changes must go through
|
||||
// primary to avoid weird race conditions.
|
||||
nodes, err := c.Nodes(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("getting nodes: %s", err)
|
||||
}
|
||||
coord := getPrimaryNode(nodes)
|
||||
if coord == nil {
|
||||
return fmt.Errorf("could not find the coordinator node")
|
||||
return fmt.Errorf("could not find the primary node")
|
||||
}
|
||||
|
||||
// Create URL & HTTP request.
|
||||
|
|
@ -1202,8 +1202,8 @@ func (c *InternalClient) SendMessage(ctx context.Context, uri *pnet.URI, msg []b
|
|||
return errors.Wrap(err, "draining SendMessage response body")
|
||||
}
|
||||
|
||||
// TranslateKeysNode function is mainly called to translate keys from coordinator node.
|
||||
// If coordinator node returns 404 error the function wraps it with pilosa.ErrTranslatingKeyNotFound.
|
||||
// TranslateKeysNode function is mainly called to translate keys from primary node.
|
||||
// If primary node returns 404 error the function wraps it with pilosa.ErrTranslatingKeyNotFound.
|
||||
func (c *InternalClient) TranslateKeysNode(ctx context.Context, uri *pnet.URI, index, field string, keys []string, writable bool) ([]uint64, error) {
|
||||
span, ctx := tracing.StartSpanFromContext(ctx, "TranslateKeysNode")
|
||||
defer span.Finish()
|
||||
|
|
@ -1608,7 +1608,7 @@ func (c *InternalClient) StartTransaction(ctx context.Context, id string, timeou
|
|||
// We're using the defaultURI here because this is only used by
|
||||
// tests, and we want to test requests against all hosts. A robust
|
||||
// client implementation would ensure that these requests go to
|
||||
// the coordinator.
|
||||
// the primary.
|
||||
u := uriPathToURL(c.defaultURI, "/transaction/"+id)
|
||||
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
|
||||
if err != nil {
|
||||
|
|
@ -1680,7 +1680,7 @@ func (c *InternalClient) GetTransaction(ctx context.Context, id string) (*pilosa
|
|||
// We're using the defaultURI here because this is only used by
|
||||
// tests, and we want to test requests against all hosts. A robust
|
||||
// client implementation would ensure that these requests go to
|
||||
// the coordinator.
|
||||
// the primary.
|
||||
u := uriPathToURL(c.defaultURI, "/transaction/"+id)
|
||||
req, err := http.NewRequest("GET", u.String(), nil)
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -764,7 +764,7 @@ func TestClient_ImportKeys(t *testing.T) {
|
|||
}
|
||||
})
|
||||
|
||||
// Import to node1 (ensure import is routed to coordinator for translation).
|
||||
// Import to node1 (ensure import is routed to primary for translation).
|
||||
t.Run("Import node1", func(t *testing.T) {
|
||||
if err := c1.ImportK(context.Background(), "keyed", "keyedf1", []pilosa.Bit{
|
||||
{RowKey: "green", ColumnKey: "eve"},
|
||||
|
|
@ -1226,8 +1226,8 @@ func TestClientTransactions(t *testing.T) {
|
|||
c := test.MustRunCluster(t, 3)
|
||||
defer c.Close()
|
||||
|
||||
coord := c.GetCoordinator()
|
||||
other := c.GetNonCoordinator()
|
||||
coord := c.GetPrimary()
|
||||
other := c.GetNonPrimary()
|
||||
|
||||
client0 := MustNewClient(coord.URL(), http.GetHTTPClient(nil))
|
||||
client1 := MustNewClient(other.URL(), http.GetHTTPClient(nil))
|
||||
|
|
@ -1357,10 +1357,10 @@ func TestClientTransactions(t *testing.T) {
|
|||
trns)
|
||||
}
|
||||
|
||||
// non-coordinator
|
||||
// non-primary
|
||||
if trns, err := client1.StartTransaction(context.Background(), "blah", time.Minute, false); err == nil ||
|
||||
!strings.Contains(err.Error(), pilosa.ErrNodeNotPrimary.Error()) {
|
||||
t.Fatalf("unexpected error starting on non-coordinator: %v", err)
|
||||
t.Fatalf("unexpected error starting on non-primary: %v", err)
|
||||
} else {
|
||||
test.CompareTransactions(t,
|
||||
nil,
|
||||
|
|
|
|||
|
|
@ -224,7 +224,6 @@ func (h *Handler) populateValidators() {
|
|||
h.validators = map[string]*queryValidationSpec{}
|
||||
h.validators["PostClusterResizeAbort"] = queryValidationSpecRequired()
|
||||
h.validators["PostClusterResizeRemoveNode"] = queryValidationSpecRequired()
|
||||
h.validators["PostClusterResizeSetCoordinator"] = queryValidationSpecRequired()
|
||||
h.validators["GetExport"] = queryValidationSpecRequired("index", "field", "shard")
|
||||
h.validators["GetIndexes"] = queryValidationSpecRequired()
|
||||
h.validators["GetIndex"] = queryValidationSpecRequired()
|
||||
|
|
|
|||
|
|
@ -943,7 +943,7 @@ func (s *Server) SendTo(node *topology.Node, m Message) error {
|
|||
}
|
||||
|
||||
// node returns the pilosa.node object. It is used by membership protocols to
|
||||
// get this node's name(ID), location(URI), and coordinator status.
|
||||
// get this node's name(ID), location(URI), and primary status.
|
||||
func (s *Server) node() *topology.Node {
|
||||
return s.cluster.Node.Clone()
|
||||
}
|
||||
|
|
@ -1113,7 +1113,7 @@ func (srv *Server) StartTransaction(ctx context.Context, id string, timeout time
|
|||
return nil, ErrNodeNotPrimary
|
||||
}
|
||||
if remote && (snap.IsPrimaryFieldTranslationNode(node.ID) || len(srv.cluster.Nodes()) == 1) {
|
||||
return nil, errors.New("unexpected remote start call to coordinator or single node cluster")
|
||||
return nil, errors.New("unexpected remote start call to primary or single node cluster")
|
||||
}
|
||||
|
||||
if remote {
|
||||
|
|
@ -1160,7 +1160,7 @@ func (srv *Server) FinishTransaction(ctx context.Context, id string, remote bool
|
|||
return nil, ErrNodeNotPrimary
|
||||
}
|
||||
if remote && (snap.IsPrimaryFieldTranslationNode(node.ID) || len(srv.cluster.Nodes()) == 1) {
|
||||
return nil, errors.New("unexpected remote finish call to coordinator or single node cluster")
|
||||
return nil, errors.New("unexpected remote finish call to primary or single node cluster")
|
||||
}
|
||||
|
||||
if remote {
|
||||
|
|
@ -1202,7 +1202,7 @@ func (srv *Server) GetTransaction(ctx context.Context, id string, remote bool) (
|
|||
}
|
||||
|
||||
if remote && (snap.IsPrimaryFieldTranslationNode(node.ID) || len(srv.cluster.Nodes()) == 1) {
|
||||
return nil, errors.New("unexpected remote get call to coordinator or single node cluster")
|
||||
return nil, errors.New("unexpected remote get call to primary or single node cluster")
|
||||
}
|
||||
|
||||
trns, err := srv.holder.GetTransaction(ctx, id)
|
||||
|
|
|
|||
|
|
@ -558,8 +558,8 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
|
|||
func TestClusterResize_RemoveNode(t *testing.T) {
|
||||
cluster := test.MustRunCluster(t, 3)
|
||||
defer cluster.Close()
|
||||
coord := cluster.GetCoordinator()
|
||||
other := cluster.GetNonCoordinator()
|
||||
coord := cluster.GetPrimary()
|
||||
other := cluster.GetNonPrimary()
|
||||
|
||||
mustNodeID := func(baseURL string) string {
|
||||
body := test.Do(t, "GET", fmt.Sprintf("%s/status", baseURL), "").Body
|
||||
|
|
@ -584,7 +584,7 @@ func TestClusterResize_RemoveNode(t *testing.T) {
|
|||
}
|
||||
})
|
||||
|
||||
t.Run("ErrorRemoveCoordinator", func(t *testing.T) {
|
||||
t.Run("ErrorRemovePrimary", func(t *testing.T) {
|
||||
nodeID := mustNodeID(coord.URL())
|
||||
resp := test.Do(t, "POST", coord.URL()+"/cluster/resize/remove-node", fmt.Sprintf(`{"id": "%s"}`, nodeID))
|
||||
|
||||
|
|
@ -596,7 +596,7 @@ func TestClusterResize_RemoveNode(t *testing.T) {
|
|||
}
|
||||
})
|
||||
|
||||
t.Run("ErrorRemoveOnNonCoordinator", func(t *testing.T) {
|
||||
t.Run("ErrorRemoveOnNonPrimary", func(t *testing.T) {
|
||||
nodeID := mustNodeID(other.URL())
|
||||
resp := test.Do(t, "POST", other.URL()+"/cluster/resize/remove-node", fmt.Sprintf(`{"id": "%s"}`, nodeID))
|
||||
|
||||
|
|
|
|||
|
|
@ -400,8 +400,8 @@ func TestTransactionsAPI(t *testing.T) {
|
|||
cluster := test.MustRunCluster(t, 3)
|
||||
defer cluster.Close()
|
||||
|
||||
coord := cluster.GetCoordinator().API
|
||||
other := cluster.GetNonCoordinator().API
|
||||
coord := cluster.GetPrimary().API
|
||||
other := cluster.GetNonPrimary().API
|
||||
ctx := context.Background()
|
||||
|
||||
// can fetch empty transactions
|
||||
|
|
@ -411,7 +411,7 @@ func TestTransactionsAPI(t *testing.T) {
|
|||
t.Fatalf("unexpectedly has transactions: %v", trnsMap)
|
||||
}
|
||||
|
||||
// can't fetch transactions from non-coordinator
|
||||
// can't fetch transactions from non-primary
|
||||
if _, err := other.Transactions(ctx); err != pilosa.ErrNodeNotPrimary {
|
||||
t.Errorf("api1 should return ErrNodeNotPrimary when asked for transactions but got: %v", err)
|
||||
}
|
||||
|
|
@ -442,7 +442,7 @@ func TestTransactionsAPI(t *testing.T) {
|
|||
test.CompareTransactions(t, &pilosa.Transaction{ID: id, Active: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns)
|
||||
}
|
||||
|
||||
// can't finish transaction on non-coordinator
|
||||
// can't finish transaction on non-primary
|
||||
if _, err := other.FinishTransaction(ctx, id, false); err != pilosa.ErrNodeNotPrimary {
|
||||
t.Errorf("unexpected error is not ErrNodeNotPrimary: %v", err)
|
||||
}
|
||||
|
|
@ -519,7 +519,7 @@ func TestTransactionsAPI(t *testing.T) {
|
|||
test.CompareTransactions(t, &pilosa.Transaction{ID: "exc", Active: true, Exclusive: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns)
|
||||
}
|
||||
|
||||
// LATER, test deadline extension on non-coordinator blocks active, exclusive transaction being returned
|
||||
// LATER, test deadline extension on non-primary blocks active, exclusive transaction being returned
|
||||
}
|
||||
|
||||
func TestMain_RecalculateCaches(t *testing.T) {
|
||||
|
|
@ -647,16 +647,16 @@ func TestClusteringNodesReplica1(t *testing.T) {
|
|||
t.Fatalf("starting cluster: %v", err)
|
||||
}
|
||||
|
||||
if err := cluster.GetNonCoordinator().Command.Close(); err != nil {
|
||||
if err := cluster.GetNonPrimary().Command.Close(); err != nil {
|
||||
t.Fatalf("closing third node: %v", err)
|
||||
}
|
||||
|
||||
if err := cluster.GetCoordinator().AwaitState(disco.ClusterStateDown, 30*time.Second); err != nil {
|
||||
if err := cluster.GetPrimary().AwaitState(disco.ClusterStateDown, 30*time.Second); err != nil {
|
||||
t.Fatalf("starting cluster: %v", err)
|
||||
}
|
||||
|
||||
// confirm that cluster stops accepting queries after one node closes
|
||||
if _, err := cluster.GetCoordinator().API.Query(context.Background(), &pilosa.QueryRequest{}); !strings.Contains(err.Error(), "not allowed in state DOWN") {
|
||||
if _, err := cluster.GetPrimary().API.Query(context.Background(), &pilosa.QueryRequest{}); !strings.Contains(err.Error(), "not allowed in state DOWN") {
|
||||
t.Fatalf("got unexpected error querying an incomplete cluster: %v", err)
|
||||
}
|
||||
}
|
||||
|
|
@ -675,7 +675,7 @@ func TestClusteringNodesReplica2(t *testing.T) {
|
|||
}
|
||||
defer cluster.Close()
|
||||
|
||||
coord, others := cluster.GetCoordinator(), cluster.GetNonCoordinators()
|
||||
coord, others := cluster.GetPrimary(), cluster.GetNonPrimaries()
|
||||
|
||||
if err := others[0].Close(); err != nil {
|
||||
t.Fatalf("closing third node: %v", err)
|
||||
|
|
@ -728,7 +728,7 @@ func TestRemoveNodeAfterItDies(t *testing.T) {
|
|||
cluster.Close()
|
||||
}()
|
||||
|
||||
coord, others := cluster.GetCoordinator(), cluster.GetNonCoordinators()
|
||||
coord, others := cluster.GetPrimary(), cluster.GetNonPrimaries()
|
||||
|
||||
err = coord.AwaitState(disco.ClusterStateNormal, 100*time.Millisecond)
|
||||
if err != nil {
|
||||
|
|
@ -789,7 +789,7 @@ func TestRemoveConcurrentIndexCreation(t *testing.T) {
|
|||
t.Fatalf("removing node: %v", err)
|
||||
}
|
||||
|
||||
err = cluster.GetCoordinator().AwaitState(disco.ClusterStateNormal, 100*time.Millisecond)
|
||||
err = cluster.GetPrimary().AwaitState(disco.ClusterStateNormal, 100*time.Millisecond)
|
||||
if err != nil {
|
||||
t.Fatalf("starting cluster: %v", err)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -55,7 +55,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.GetCoordinator().QueryAPI(t, &pilosa.QueryRequest{Index: index, Query: query})
|
||||
return c.GetPrimary().QueryAPI(t, &pilosa.QueryRequest{Index: index, Query: query})
|
||||
}
|
||||
|
||||
// QueryHTTP executes a PQL query through the HTTP endpoint. It fails
|
||||
|
|
@ -67,7 +67,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.GetCoordinator().Query(t, index, "", query)
|
||||
return c.GetPrimary().Query(t, index, "", query)
|
||||
}
|
||||
|
||||
// QueryGRPC executes a PQL query through the GRPC endpoint. It fails the
|
||||
|
|
@ -78,7 +78,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.GetCoordinator().Server.GRPCURI().Host, c.GetCoordinator().Server.GRPCURI().Port)}, nil)
|
||||
grpcClient, err := client.NewGRPCClient([]string{fmt.Sprintf("%s:%d", c.GetPrimary().Server.GRPCURI().Host, c.GetPrimary().Server.GRPCURI().Port)}, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("getting GRPC client: %v", err)
|
||||
}
|
||||
|
|
@ -132,6 +132,10 @@ func (c *Cluster) GetNode(n int) *Command {
|
|||
return c.Nodes[ids[n].idx]
|
||||
}
|
||||
|
||||
// GetPrimary gets the node which has been determined to be the primary.
|
||||
// This used to be node0 in tests, but since implementing etcd, the primary
|
||||
// can be any node in the cluster, so we have to use this method in tests which
|
||||
// need to act on the primary.
|
||||
func (c *Cluster) GetPrimary() *Command {
|
||||
for _, n := range c.Nodes {
|
||||
if n.IsPrimary() {
|
||||
|
|
@ -141,6 +145,7 @@ func (c *Cluster) GetPrimary() *Command {
|
|||
return nil
|
||||
}
|
||||
|
||||
// GetNonPrimary gets first first non-primary node in the list of nodes.
|
||||
func (c *Cluster) GetNonPrimary() *Command {
|
||||
for _, n := range c.Nodes {
|
||||
if !n.IsPrimary() {
|
||||
|
|
@ -150,6 +155,7 @@ func (c *Cluster) GetNonPrimary() *Command {
|
|||
return nil
|
||||
}
|
||||
|
||||
// GetNonPrimaries gets all nodes except the primary.
|
||||
func (c *Cluster) GetNonPrimaries() []*Command {
|
||||
rtn := make([]*Command, 0)
|
||||
for _, n := range c.Nodes {
|
||||
|
|
@ -160,24 +166,6 @@ func (c *Cluster) GetNonPrimaries() []*Command {
|
|||
return rtn
|
||||
}
|
||||
|
||||
// GetCoordinator gets the node which has been determined to be the coordinator.
|
||||
// This used to be node0 in tests, but since implementing etcd, the coordinator
|
||||
// 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 {
|
||||
return c.GetPrimary()
|
||||
}
|
||||
|
||||
// GetNonCoordinator gets first first non-coordinator node in the list of nodes.
|
||||
func (c *Cluster) GetNonCoordinator() *Command {
|
||||
return c.GetNonPrimary()
|
||||
}
|
||||
|
||||
// GetNonCoordinators gets all nodes except the coordinator.
|
||||
func (c *Cluster) GetNonCoordinators() []*Command {
|
||||
return c.GetNonPrimaries()
|
||||
}
|
||||
|
||||
// nodePlace represents a node's ID and its index into the c.Nodes slice.
|
||||
type nodePlace struct {
|
||||
id string
|
||||
|
|
@ -188,17 +176,6 @@ func (c *Cluster) GetHolder(n int) *Holder {
|
|||
return &Holder{Holder: c.GetNode(n).Server.Holder()}
|
||||
}
|
||||
|
||||
// GetCoordinatorHolder returns the Holder for the coordinator node.
|
||||
func (c *Cluster) GetCoordinatorHolder() *Holder {
|
||||
return &Holder{Holder: c.GetCoordinator().Server.Holder()}
|
||||
}
|
||||
|
||||
// GetNonCoordinatorHolder returns the Holder for the the first non-coordinator
|
||||
// node in the list of nodes.
|
||||
func (c *Cluster) GetNonCoordinatorHolder() *Holder {
|
||||
return &Holder{Holder: c.GetNonCoordinator().Server.Holder()}
|
||||
}
|
||||
|
||||
func (c *Cluster) Len() int {
|
||||
return len(c.Nodes)
|
||||
}
|
||||
|
|
@ -218,7 +195,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.GetCoordinator().API.ShardNodes(context.Background(), index, shard)
|
||||
nodes, err := c.GetPrimary().API.ShardNodes(context.Background(), index, shard)
|
||||
if err != nil {
|
||||
t.Fatalf("getting shard nodes: %v", err)
|
||||
}
|
||||
|
|
@ -261,7 +238,7 @@ func (c *Cluster) ImportKeyKey(t testing.TB, index, field string, valAndRecKeys
|
|||
importRequest.RowKeys[i] = vk[0]
|
||||
importRequest.ColumnKeys[i] = vk[1]
|
||||
}
|
||||
err := c.GetCoordinator().API.Import(context.Background(), nil, importRequest)
|
||||
err := c.GetPrimary().API.Import(context.Background(), nil, importRequest)
|
||||
if err != nil {
|
||||
t.Fatalf("importing keykey data: %v", err)
|
||||
}
|
||||
|
|
@ -291,7 +268,7 @@ func (c *Cluster) ImportTimeQuantumKey(t testing.TB, index, field string, entrie
|
|||
importRequest.Timestamps[i] = entry.Ts
|
||||
|
||||
}
|
||||
err := c.GetCoordinator().API.Import(context.Background(), nil, importRequest)
|
||||
err := c.GetPrimary().API.Import(context.Background(), nil, importRequest)
|
||||
if err != nil {
|
||||
t.Fatalf("importing keykey data: %v", err)
|
||||
}
|
||||
|
|
@ -317,7 +294,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.GetCoordinator().API.ImportValue(context.Background(), nil, importRequest); err != nil {
|
||||
if err := c.GetPrimary().API.ImportValue(context.Background(), nil, importRequest); err != nil {
|
||||
t.Fatalf("importing IntKey data: %v", err)
|
||||
}
|
||||
}
|
||||
|
|
@ -341,7 +318,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.GetCoordinator().API.ImportValue(context.Background(), nil, importRequest); err != nil {
|
||||
if err := c.GetPrimary().API.ImportValue(context.Background(), nil, importRequest); err != nil {
|
||||
t.Fatalf("importing IntID data: %v", err)
|
||||
}
|
||||
}
|
||||
|
|
@ -366,7 +343,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.GetCoordinator().API.Import(context.Background(), nil, importRequest)
|
||||
err := c.GetPrimary().API.Import(context.Background(), nil, importRequest)
|
||||
if err != nil {
|
||||
t.Fatalf("importing IDKey data: %v", err)
|
||||
}
|
||||
|
|
@ -375,11 +352,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.GetCoordinator().API.CreateIndex(context.Background(), index, iopts)
|
||||
idx, err := c.GetPrimary().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.GetCoordinator().API.Index(context.Background(), index)
|
||||
idx, err = c.GetPrimary().API.Index(context.Background(), index)
|
||||
if err != nil {
|
||||
t.Fatalf("getting index: %v", err)
|
||||
}
|
||||
|
|
@ -388,7 +365,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.GetCoordinator().API.CreateField(context.Background(), index, field, fopts...)
|
||||
f, err := c.GetPrimary().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 {
|
||||
|
|
@ -446,13 +423,13 @@ func (c *Cluster) Close() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (c *Cluster) CloseAndRemoveNonCoordinator() error {
|
||||
func (c *Cluster) CloseAndRemoveNonPrimary() error {
|
||||
for i, n := range c.Nodes {
|
||||
if !n.IsPrimary() {
|
||||
return c.CloseAndRemove(i)
|
||||
}
|
||||
}
|
||||
return errors.New("could not find non-coordinator node")
|
||||
return errors.New("could not find non-primary node")
|
||||
}
|
||||
|
||||
func (c *Cluster) CloseAndRemove(n int) error {
|
||||
|
|
|
|||
|
|
@ -30,10 +30,10 @@ func TestNewCluster(t *testing.T) {
|
|||
cluster := test.MustRunCluster(t, numNodes)
|
||||
defer cluster.Close()
|
||||
|
||||
coordinator := getCoordinator(cluster.Nodes[0])
|
||||
primary := getPrimary(cluster.Nodes[0])
|
||||
for i := 1; i < numNodes; i++ {
|
||||
if coordi := getCoordinator(cluster.Nodes[i]); coordi != coordinator {
|
||||
t.Fatalf("node %d does not have the same coordinator as node 0. '%v' and '%v' respectively", i, coordi, coordinator)
|
||||
if coordi := getPrimary(cluster.Nodes[i]); coordi != primary {
|
||||
t.Fatalf("node %d does not have the same primary as node 0. '%v' and '%v' respectively", i, coordi, primary)
|
||||
}
|
||||
}
|
||||
req, err := http.NewRequest(
|
||||
|
|
@ -82,12 +82,12 @@ func TestNewCluster(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func getCoordinator(m *test.Command) string {
|
||||
func getPrimary(m *test.Command) string {
|
||||
hosts := m.API.Hosts(context.Background())
|
||||
for _, host := range hosts {
|
||||
if host.IsPrimary {
|
||||
return host.ID
|
||||
}
|
||||
}
|
||||
panic("no coordinator in cluster")
|
||||
panic("no primary in cluster")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -469,8 +469,8 @@ func TestTranslation_Replication(t *testing.T) {
|
|||
)
|
||||
defer c.Close()
|
||||
|
||||
coord := c.GetCoordinator()
|
||||
other := c.GetNonCoordinator()
|
||||
coord := c.GetPrimary()
|
||||
other := c.GetNonPrimary()
|
||||
|
||||
ctx := context.Background()
|
||||
idx := "i"
|
||||
|
|
@ -512,8 +512,8 @@ func TestTranslation_Replication(t *testing.T) {
|
|||
// Verify the data exists
|
||||
coord.QueryExpect(t, idx, "", `Row(f=1)`, exp)
|
||||
|
||||
// Kill a non-coordinator node.
|
||||
if err := c.CloseAndRemoveNonCoordinator(); err != nil {
|
||||
// Kill a non-primary node.
|
||||
if err := c.CloseAndRemoveNonPrimary(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
|
@ -528,9 +528,9 @@ func TestTranslation_Replication(t *testing.T) {
|
|||
}
|
||||
|
||||
// Test key translation with multiple nodes.
|
||||
func TestTranslation_Coordinator(t *testing.T) {
|
||||
func TestTranslation_Primary(t *testing.T) {
|
||||
// Ensure that field key translations requests sent to
|
||||
// non-coordinator nodes are forwarded to the coordinator.
|
||||
// non-primary nodes are forwarded to the primary.
|
||||
t.Run("ForwardFieldKey", func(t *testing.T) {
|
||||
t.Skip("Short term skip to avoid go 1.13 test Should remove ASAP")
|
||||
// Start a 2-node cluster.
|
||||
|
|
@ -550,8 +550,8 @@ func TestTranslation_Coordinator(t *testing.T) {
|
|||
)
|
||||
defer c.Close()
|
||||
|
||||
node0 := c.GetCoordinator()
|
||||
node1 := c.GetNonCoordinator()
|
||||
node0 := c.GetPrimary()
|
||||
node1 := c.GetNonPrimary()
|
||||
|
||||
ctx := context.Background()
|
||||
idx := "i"
|
||||
|
|
@ -576,7 +576,7 @@ func TestTranslation_Coordinator(t *testing.T) {
|
|||
for i := range keys {
|
||||
pql := fmt.Sprintf(`Set(%d, %s="%s")`, i+1, fld, keys[i])
|
||||
|
||||
// Send a translation request to node1 (non-coordinator).
|
||||
// Send a translation request to node1 (non-primary).
|
||||
_, err := node1.API.Query(ctx,
|
||||
&pilosa.QueryRequest{Index: idx, Query: pql},
|
||||
)
|
||||
|
|
@ -632,8 +632,8 @@ func TestTranslation_TranslateIDsOnCluster(t *testing.T) {
|
|||
)
|
||||
defer c.Close()
|
||||
|
||||
coord := c.GetCoordinator()
|
||||
other := c.GetNonCoordinator()
|
||||
coord := c.GetPrimary()
|
||||
other := c.GetNonPrimary()
|
||||
|
||||
ctx := context.Background()
|
||||
idx, fld := "i", "f"
|
||||
|
|
@ -743,7 +743,7 @@ func TestTranslation_Cluster_CreateFind(t *testing.T) {
|
|||
}
|
||||
|
||||
// Obtain authoritative translations for the keys.
|
||||
translations, err := c.GetCoordinator().API.FindIndexKeys(ctx, "i", keyList...)
|
||||
translations, err := c.GetPrimary().API.FindIndexKeys(ctx, "i", keyList...)
|
||||
if err != nil {
|
||||
t.Errorf("obtaining authoritative translations: %v", err)
|
||||
return
|
||||
|
|
@ -829,7 +829,7 @@ func TestTranslation_Cluster_CreateFind(t *testing.T) {
|
|||
}
|
||||
|
||||
// Obtain authoritative translations for the keys.
|
||||
translations, err := c.GetCoordinator().API.FindFieldKeys(ctx, "i", "f", keyList...)
|
||||
translations, err := c.GetPrimary().API.FindFieldKeys(ctx, "i", "f", keyList...)
|
||||
if err != nil {
|
||||
t.Errorf("obtaining authoritative translations: %v", err)
|
||||
return
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue