diff --git a/api.go b/api.go index d0a5c396d..1324206eb 100644 --- a/api.go +++ b/api.go @@ -125,7 +125,6 @@ var validAPIMethods = map[disco.ClusterState]map[apiMethod]struct{}{ // has on a node under load, this is set to effectively allow all requests // in a DEGRADED state. disco.ClusterStateDegraded: appendMap(methodsCommon, methodsNormal), - disco.ClusterStateResizing: appendMap(methodsCommon, methodsResizing), // Ideally, this would be just `methodsCommon`, but in an attempt to reduce // the influence that state (determined by etcd) has on a node under load, // this is set to effectively allow all requests in a DOWN state. @@ -2272,40 +2271,8 @@ func (api *API) indexField(indexName string, fieldName string, shard uint64) (*I return index, field, nil } -// RemoveNode puts the cluster into the "RESIZING" state and begins the job of -// removing the given node. -func (api *API) RemoveNode(id string) (*topology.Node, error) { - if err := api.validate(apiRemoveNode); err != nil { - return nil, errors.Wrap(err, "validating api method") - } - - if api.cluster.disCo.ID() == id { - return nil, errors.Wrapf(ErrPreconditionFailed, "cannot issue node removal request to the node being removed, id=%s", id) - } - - removeNode := api.cluster.nodeByID(id) - if removeNode == nil { - return nil, errors.Wrap(ErrNodeIDNotExists, "finding node to remove") - } - - if err := api.cluster.removeNode(id); err != nil { - return nil, errors.Wrapf(err, "removing node %s", id) - } - - return removeNode, nil -} - -// ResizeAbort stops the current resize job. -func (api *API) ResizeAbort() error { - if err := api.validate(apiResizeAbort); err != nil { - return errors.Wrap(err, "validating api method") - } - - return api.cluster.resizeAbortAndBroadcast() -} - // State returns the cluster state which is usually "NORMAL", but could be -// "STARTING", "RESIZING", or potentially others. See disco.go for more +// "STARTING", or potentially others. See disco.go for more // details. func (api *API) State() (disco.ClusterState, error) { if err := api.validate(apiState); err != nil { @@ -3257,8 +3224,6 @@ const ( //apiMaxShards // not implemented apiQuery apiRecalculateCaches - apiRemoveNode - apiResizeAbort apiSchema apiShardNodes apiState @@ -3286,14 +3251,6 @@ var methodsCommon = map[apiMethod]struct{}{ apiState: {}, } -var methodsResizing = map[apiMethod]struct{}{ - apiFragmentData: {}, - apiTranslateData: {}, - apiFieldTranslateData: {}, - apiResizeAbort: {}, - apiSchema: {}, -} - // var methodsDegraded = map[apiMethod]struct{}{ // apiExportCSV: {}, // apiFragmentBlockData: {}, @@ -3330,7 +3287,6 @@ var methodsNormal = map[apiMethod]struct{}{ apiIndex: {}, apiQuery: {}, apiRecalculateCaches: {}, - apiRemoveNode: {}, apiShardNodes: {}, apiSchema: {}, apiViews: {}, diff --git a/apimethod_string.go b/apimethod_string.go index 24703b05a..eebc4cab6 100644 --- a/apimethod_string.go +++ b/apimethod_string.go @@ -27,31 +27,29 @@ func _() { _ = x[apiIndex-16] _ = x[apiQuery-17] _ = x[apiRecalculateCaches-18] - _ = x[apiRemoveNode-19] - _ = x[apiResizeAbort-20] - _ = x[apiSchema-21] - _ = x[apiShardNodes-22] - _ = x[apiState-23] - _ = x[apiViews-24] - _ = x[apiApplySchema-25] - _ = x[apiStartTransaction-26] - _ = x[apiFinishTransaction-27] - _ = x[apiTransactions-28] - _ = x[apiGetTransaction-29] - _ = x[apiActiveQueries-30] - _ = x[apiPastQueries-31] - _ = x[apiIDReserve-32] - _ = x[apiIDCommit-33] - _ = x[apiIDReset-34] - _ = x[apiPartitionNodes-35] - _ = x[apiIngestOperations-36] - _ = x[apiIngestNodeOperations-37] - _ = x[apiMutexCheck-38] + _ = x[apiSchema-19] + _ = x[apiShardNodes-20] + _ = x[apiState-21] + _ = x[apiViews-22] + _ = x[apiApplySchema-23] + _ = x[apiStartTransaction-24] + _ = x[apiFinishTransaction-25] + _ = x[apiTransactions-26] + _ = x[apiGetTransaction-27] + _ = x[apiActiveQueries-28] + _ = x[apiPastQueries-29] + _ = x[apiIDReserve-30] + _ = x[apiIDCommit-31] + _ = x[apiIDReset-32] + _ = x[apiPartitionNodes-33] + _ = x[apiIngestOperations-34] + _ = x[apiIngestNodeOperations-35] + _ = x[apiMutexCheck-36] } -const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiTranslateDataapiFieldTranslateDataapiFieldapiImportapiImportValueapiIndexapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSchemaapiShardNodesapiStateapiViewsapiApplySchemaapiStartTransactionapiFinishTransactionapiTransactionsapiGetTransactionapiActiveQueriesapiPastQueriesapiIDReserveapiIDCommitapiIDResetapiPartitionNodesapiIngestOperationsapiIngestNodeOperationsapiMutexCheck" +const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiTranslateDataapiFieldTranslateDataapiFieldapiImportapiImportValueapiIndexapiQueryapiRecalculateCachesapiSchemaapiShardNodesapiStateapiViewsapiApplySchemaapiStartTransactionapiFinishTransactionapiTransactionsapiGetTransactionapiActiveQueriesapiPastQueriesapiIDReserveapiIDCommitapiIDResetapiPartitionNodesapiIngestOperationsapiIngestNodeOperationsapiMutexCheck" -var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 189, 210, 218, 227, 241, 249, 257, 277, 290, 304, 313, 326, 334, 342, 356, 375, 395, 410, 427, 443, 457, 469, 480, 490, 507, 526, 549, 562} +var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 189, 210, 218, 227, 241, 249, 257, 277, 286, 299, 307, 315, 329, 348, 368, 383, 400, 416, 430, 442, 453, 463, 480, 499, 522, 535} func (i apiMethod) String() string { if i < 0 || i >= apiMethod(len(_apiMethod_index)-1) { diff --git a/broadcast.go b/broadcast.go index 0ad8e6b40..ecab7b946 100644 --- a/broadcast.go +++ b/broadcast.go @@ -60,16 +60,16 @@ const ( messageTypeCreateView messageTypeDeleteView messageTypeClusterStatus - messageTypeResizeInstruction - messageTypeResizeInstructionComplete + messageTypeUNUSED0 // used to be ResizeInstruction + messageTypeUNUSED1 // used to be ResizeInstructionComplete messageTypeNodeState messageTypeRecalculateCaches messageTypeLoadSchemaMessage messageTypeNodeEvent messageTypeNodeStatus messageTypeTransaction - messageTypeResizeNodeMessage - messageTypeResizeAbortMessage + messageTypeUNUSED2 // used to be ResizeNodeMessage + messageTypeUNUSED3 // used to be ResizeAbortMessage messageTypeUpdateField ) @@ -102,10 +102,6 @@ func getMessage(typ byte) Message { return &DeleteViewMessage{} case messageTypeClusterStatus: return &ClusterStatus{} - case messageTypeResizeInstruction: - return &ResizeInstruction{} - case messageTypeResizeInstructionComplete: - return &ResizeInstructionComplete{} case messageTypeNodeState: return &NodeStateMessage{} case messageTypeRecalculateCaches: @@ -118,10 +114,6 @@ func getMessage(typ byte) Message { return &NodeStatus{} case messageTypeTransaction: return &TransactionMessage{} - case messageTypeResizeNodeMessage: - return &ResizeNodeMessage{} - case messageTypeResizeAbortMessage: - return &ResizeAbortMessage{} case messageTypeUpdateField: return &UpdateFieldMessage{} default: @@ -147,10 +139,6 @@ func getMessageType(m Message) byte { return messageTypeDeleteView case *ClusterStatus: return messageTypeClusterStatus - case *ResizeInstruction: - return messageTypeResizeInstruction - case *ResizeInstructionComplete: - return messageTypeResizeInstructionComplete case *NodeStateMessage: return messageTypeNodeState case *RecalculateCaches: @@ -163,10 +151,6 @@ func getMessageType(m Message) byte { return messageTypeNodeStatus case *TransactionMessage: return messageTypeTransaction - case *ResizeNodeMessage: - return messageTypeResizeNodeMessage - case *ResizeAbortMessage: - return messageTypeResizeAbortMessage case *UpdateFieldMessage: return messageTypeUpdateField default: diff --git a/cluster.go b/cluster.go index e70ab6e79..b37d3eca0 100644 --- a/cluster.go +++ b/cluster.go @@ -3,10 +3,7 @@ package pilosa import ( "context" - "encoding/json" "fmt" - "io" - "math/rand" "sync" "time" @@ -15,42 +12,15 @@ import ( "github.com/molecula/featurebase/v3/logger" "github.com/molecula/featurebase/v3/roaring" "github.com/molecula/featurebase/v3/topology" - "github.com/molecula/featurebase/v3/tracing" "github.com/pkg/errors" "golang.org/x/sync/errgroup" ) const ( - resizeJobActionAdd = "ADD" - resizeJobActionRemove = "REMOVE" - defaultConfirmDownRetries = 10 defaultConfirmDownSleep = 1 * time.Second ) -type ResizeNodeMessage struct { - NodeID string - Action string -} - -type ResizeNodeProgress struct { - FromID string - ToID string - Done bool - Error string -} - -func (p ResizeNodeProgress) applyJSON(fn func([]byte) error) error { - data, err := json.Marshal(p) - if err != nil { - return err - } - - return fn(data) -} - -type ResizeAbortMessage struct{} - // cluster represents a collection of nodes. type cluster struct { // nolint: maligned noder topology.Noder @@ -79,11 +49,8 @@ type cluster struct { // nolint: maligned // Distributed Consensus disCo disco.DisCo stator disco.Stator - resizer disco.Resizer sharder disco.Sharder - // Required for cluster Resize. - Static bool // Static is primarily used for testing. holder *Holder broadcaster broadcaster @@ -92,9 +59,7 @@ type cluster struct { // nolint: maligned translationSyncer TranslationSyncer - mu sync.RWMutex - jobs map[int64]*resizeJob - resizeCancel context.CancelFunc + mu sync.RWMutex // Close management wg sync.WaitGroup @@ -117,7 +82,6 @@ func newCluster() *cluster { partitionN: topology.DefaultPartitionN, ReplicaN: 1, - jobs: make(map[int64]*resizeJob), closing: make(chan struct{}), translationSyncer: NopTranslationSyncer, @@ -129,10 +93,9 @@ func newCluster() *cluster { confirmDownRetries: defaultConfirmDownRetries, confirmDownSleep: defaultConfirmDownSleep, - disCo: disco.NopDisCo, - noder: topology.NewEmptyLocalNoder(), - stator: disco.NopStator, - resizer: disco.NopResizer, + disCo: disco.NopDisCo, + noder: topology.NewEmptyLocalNoder(), + stator: disco.NopStator, } } @@ -196,399 +159,6 @@ func (c *cluster) applySchemaWithNewShards(schema *Schema) error { return nil } -// addNode adds a node to the Cluster and starts resizing process -func (c *cluster) addNode(id string) error { - // If this method is being called on the node which was just added, then the - // node will be completely empty. That means that it won't have the current - // schema with which to calculate its resize intructions (in - // c.resizeNodeOnAdd, which calls c.generateResizeInstructionOnAdd). Because - // of this, we need to request and apply the current schema from etcd before - // we can proceed with the resize process. - if id == c.disCo.ID() { - schema, err := c.remoteSchema() - if err != nil { - return err - } - - if err := c.applySchemaWithNewShards(schema); err != nil { - return err - } - } - - eg := &errgroup.Group{} - for _, n := range c.noder.Nodes() { - if err := c.sendTo(n, &ResizeNodeMessage{NodeID: id, Action: resizeJobActionAdd}); err != nil { - return errors.Wrap(err, "broadcasting resize message") - } - - nodeID := n.ID - eg.Go(func() error { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - err := c.resizer.Watch(ctx, nodeID, func(data []byte) error { - var progress ResizeNodeProgress - if err := json.Unmarshal(data, &progress); err != nil { - return errors.Wrapf(err, "watching progress node %s", nodeID) - } - if progress.Error != "" { - return errors.Errorf("watching progress node %s: %s", nodeID, progress.Error) - } - if progress.Done { - return io.EOF - } - return nil - }) - if err == io.EOF { - err = nil - } - return err - }) - } - - // Wait for all background resize threads to return. If there were any - // errors, then we need to delete the node (which we were attempting to add) - // from the etcd cluster. - go func() { - if err := eg.Wait(); err != nil { - c.logger.Printf("Stop watching all peers: %+v", err) - - if err := c.disCo.DeleteNode(context.Background(), id); err != nil { - // resizing failed, so we have to delete the new node. - c.logger.Printf("Cannot delete the node %s: %+v", id, err) - } - } - }() - - return nil -} - -func (c *cluster) resizeNodeOnAdd(addNodeID string) error { - ctx, cancel := context.WithCancel(context.Background()) - - // set status to RESIZING - progressFunc, err := c.resizer.Resize(context.Background()) - if err != nil { - cancel() - return errors.Wrapf(err, "setting RESIZING state on %s", c.disCo.ID()) - } - - c.resizeCancel = cancel - // start async. data balancing - go func() { - progress := ResizeNodeProgress{ToID: addNodeID, FromID: c.disCo.ID()} - defer func() { - err := progress.applyJSON(progressFunc) - if err != nil { - c.logger.Printf("updating resize progress (%s -> %s): %+v", c.disCo.ID(), addNodeID, err) - } - if c.resizeCancel != nil { - c.resizeCancel() - } - err = c.resizer.DoneResize() - if err != nil { - c.logger.Printf("done resize (%s -> %s): %+v", c.disCo.ID(), addNodeID, err) - } - }() - - instr, err := c.generateResizeInstructionOnAdd(addNodeID) - if err != nil { - progress.Error = errors.Wrapf(err, "generating resize instruction (%s -> %s)", c.disCo.ID(), addNodeID).Error() - c.logger.Printf(progress.Error) - return - } - - if err = c.followResizeInstruction(ctx, instr); err != nil { - progress.Error = errors.Wrapf(err, "following resize instruction (%s -> %s)", c.disCo.ID(), addNodeID).Error() - c.logger.Printf(progress.Error) - return - } - progress.Done = true - }() - - return nil -} - -func (c *cluster) generateResizeInstructionOnAdd(addNodeID string) (*ResizeInstruction, error) { - fromCluster := newCluster() - for _, n := range topology.Nodes(c.noder.Nodes()).Clone() { - if n.ID == addNodeID { - continue - } - fromCluster.noder.AppendNode(n) - } - fromCluster.Hasher = c.Hasher - fromCluster.partitionN = c.partitionN - fromCluster.ReplicaN = c.ReplicaN - - // fragmentSourcesByNode is a map of Node.ID to sources of fragment data. - // It is initialized with all the nodes in toCluster. - fragmentSourcesByNode := make(map[string][]*ResizeSource) - for _, n := range c.noder.Nodes() { - fragmentSourcesByNode[n.ID] = nil - } - - indexes := c.holder.Indexes() - // Add to fragmentSourcesByNode the instructions for each index. - for _, idx := range indexes { - fragSources, err := fromCluster.fragSources(c, idx) - if err != nil { - return nil, errors.Wrap(err, "getting sources") - } - - for nodeid, sources := range fragSources { - fragmentSourcesByNode[nodeid] = append(fragmentSourcesByNode[nodeid], sources...) - } - } - - // translationSourcesByNode is a map of Node.ID to sources of partitioned - // key translation data for indexes. - // It is initialized with all the nodes in toCluster. - translationSourcesByNode := make(map[string][]*TranslationResizeSource) - for _, n := range c.noder.Nodes() { - translationSourcesByNode[n.ID] = nil - } - - if len(indexes) > 0 { - // Add to translationSourcesByNode the instructions for the cluster. - translationNodes, err := fromCluster.translationNodes(c) - if err != nil { - return nil, errors.Wrap(err, "getting translation sources") - } - - // Create a list of TranslationResizeSource for each index, - // using translationNodes as a template. - translationSources := make(map[string][]*TranslationResizeSource) - for _, idx := range indexes { - // Only include indexes with keys. - if !idx.Keys() { - continue - } - indexName := idx.Name() - for node, resizeNodes := range translationNodes { - for i := range resizeNodes { - translationSources[node] = append(translationSources[node], - &TranslationResizeSource{ - Node: resizeNodes[i].node, - Index: indexName, - PartitionID: resizeNodes[i].partitionID, - }) - } - } - } - - for nodeid, sources := range translationSources { - translationSourcesByNode[nodeid] = sources - } - } - - status, err := c.unprotectedStatus() - if err != nil { - return nil, errors.Wrap(err, "getting cluster status") - } - - myid := c.disCo.ID() - nodeStatus, err := c.nodeStatus() - if err != nil { - return nil, errors.Wrap(err, "getting node status") - } - return &ResizeInstruction{ - Node: c.unprotectedNodeByID(myid), - Sources: fragmentSourcesByNode[myid], - TranslationSources: translationSourcesByNode[myid], - NodeStatus: nodeStatus, // Include the NodeStatus in order to ensure that schema and availableShards are in sync on the receiving node. - ClusterStatus: status, - }, nil -} - -// removeNode removes a node from the Cluster and starts resizing process. -func (c *cluster) removeNode(id string) error { - eg := &errgroup.Group{} - for _, n := range c.noder.Nodes() { - // Don't send the resize message to the node being removed. - if n.ID == id { - continue - } - - if err := c.sendTo(n, &ResizeNodeMessage{NodeID: id, Action: resizeJobActionRemove}); err != nil { - return errors.Wrap(err, "broadcasting resize message") - } - - nodeID := n.ID - eg.Go(func() error { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - err := c.resizer.Watch(ctx, nodeID, func(data []byte) error { - var progress ResizeNodeProgress - if err := json.Unmarshal(data, &progress); err != nil { - return errors.Wrapf(err, "watching progress node %s", nodeID) - } - if progress.Error != "" { - return errors.Errorf("watching progress node %s: %s", nodeID, progress.Error) - } - if progress.Done { - return io.EOF - } - return nil - }) - if err == io.EOF { - err = nil - } - return err - }) - } - - // monitor all background resize threads - go func() { - if err := eg.Wait(); err != nil { - c.logger.Printf("Stop watching all peers: %+v", err) - return - } - - if err := c.disCo.DeleteNode(context.Background(), id); err != nil { - // it's ok, we can delete the node - c.logger.Printf("Cannot delete the node %s: %+v", id, err) - } - }() - - return nil -} - -func (c *cluster) resizeNodeOnRemove(removeNodeID string) error { - ctx, cancel := context.WithCancel(context.Background()) - - // set status to RESIZING - progressFunc, err := c.resizer.Resize(context.Background()) - if err != nil { - cancel() - return errors.Wrapf(err, "setting RESIZING state on %s", c.disCo.ID()) - } - - c.resizeCancel = cancel - // start async. data balancing - go func() { - progress := ResizeNodeProgress{FromID: removeNodeID, ToID: c.disCo.ID()} - defer func() { - err := progress.applyJSON(progressFunc) - if err != nil { - c.logger.Printf("updating resize progress (%s <- %s): %+v", c.disCo.ID(), removeNodeID, err) - } - if c.resizeCancel != nil { - c.resizeCancel() - } - err = c.resizer.DoneResize() - if err != nil { - c.logger.Printf("done resize (%s <- %s): %+v", c.disCo.ID(), removeNodeID, err) - } - }() - - instr, err := c.generateResizeInstructionOnRemove(removeNodeID) - if err != nil { - progress.Error = errors.Wrapf(err, "generating resize instruction (%s <- %s)", c.disCo.ID(), removeNodeID).Error() - c.logger.Printf(progress.Error) - return - } - - if err = c.followResizeInstruction(ctx, instr); err != nil { - progress.Error = errors.Wrapf(err, "following resize instruction (%s <- %s)", c.disCo.ID(), removeNodeID).Error() - c.logger.Printf(progress.Error) - return - } - progress.Done = true - }() - - return nil -} - -func (c *cluster) generateResizeInstructionOnRemove(removeNodeID string) (*ResizeInstruction, error) { - toCluster := newCluster() - toCluster.noder.SetNodes(topology.Nodes(c.noder.Nodes()).Clone()) - toCluster.Hasher = c.Hasher - toCluster.partitionN = c.partitionN - toCluster.ReplicaN = c.ReplicaN - toCluster.removeNodeBasicSorted(removeNodeID) - - // fragmentSourcesByNode is a map of Node.ID to sources of fragment data. - // It is initialized with all the nodes in toCluster. - fragmentSourcesByNode := make(map[string][]*ResizeSource) - for _, n := range toCluster.noder.Nodes() { - fragmentSourcesByNode[n.ID] = nil - } - - indexes := c.holder.Indexes() - // Add to fragmentSourcesByNode the instructions for each index. - for _, idx := range indexes { - fragSources, err := c.fragSources(toCluster, idx) - if err != nil { - return nil, errors.Wrap(err, "getting sources") - } - - for nodeid, sources := range fragSources { - fragmentSourcesByNode[nodeid] = append(fragmentSourcesByNode[nodeid], sources...) - } - } - - // translationSourcesByNode is a map of Node.ID to sources of partitioned - // key translation data for indexes. - // It is initialized with all the nodes in toCluster. - translationSourcesByNode := make(map[string][]*TranslationResizeSource) - for _, n := range toCluster.noder.Nodes() { - translationSourcesByNode[n.ID] = nil - } - - if len(indexes) > 0 { - // Add to translationSourcesByNode the instructions for the cluster. - translationNodes, err := c.translationNodes(toCluster) - if err != nil { - return nil, errors.Wrap(err, "getting translation sources") - } - - // Create a list of TranslationResizeSource for each index, - // using translationNodes as a template. - translationSources := make(map[string][]*TranslationResizeSource) - for _, idx := range indexes { - // Only include indexes with keys. - if !idx.Keys() { - continue - } - indexName := idx.Name() - for node, resizeNodes := range translationNodes { - for i := range resizeNodes { - translationSources[node] = append(translationSources[node], - &TranslationResizeSource{ - Node: resizeNodes[i].node, - Index: indexName, - PartitionID: resizeNodes[i].partitionID, - }) - } - } - } - - for nodeid, sources := range translationSources { - translationSourcesByNode[nodeid] = sources - } - } - - status, err := c.unprotectedStatus() - if err != nil { - return nil, errors.Wrap(err, "getting cluster status") - } - - myid := c.disCo.ID() - nodeStatus, err := c.nodeStatus() - if err != nil { - return nil, errors.Wrap(err, "getting node status") - } - return &ResizeInstruction{ - Node: toCluster.unprotectedNodeByID(myid), - Sources: fragmentSourcesByNode[myid], - TranslationSources: translationSourcesByNode[myid], - NodeStatus: nodeStatus, // Include the NodeStatus in order to ensure that schema and availableShards are in sync on the receiving node. - ClusterStatus: status, - }, nil -} - // unprotectedStatus returns the the cluster's status including what nodes it contains, its ID, and current state. func (c *cluster) unprotectedStatus() (*ClusterStatus, error) { state, err := c.stator.ClusterState(context.Background()) @@ -659,30 +229,6 @@ func (c *cluster) nodePositionByID(nodeID string) int { return -1 } -// addNodeBasicSorted adds a node to the cluster, sorted by id. Returns a -// pointer to the node and true if the node was added or updated. unprotected. -func (c *cluster) addNodeBasicSorted(node *topology.Node) bool { - n := c.unprotectedNodeByID(node.ID) - - if n != nil { - nn := &topology.Node{ - ID: node.ID, - URI: node.URI, - GRPCURI: node.GRPCURI, - IsPrimary: node.IsPrimary, - State: node.State, - } - if n.State != node.State || n.IsPrimary != node.IsPrimary || n.URI != node.URI { - *n = *nn - return true - } - return false - } - - c.noder.AppendNode(node) - return true -} - // Nodes returns a copy of the slice of nodes in the cluster. Safe for // concurrent use, result may be modified. func (c *cluster) Nodes() []*topology.Node { @@ -704,275 +250,6 @@ func (c *cluster) Nodes() []*topology.Node { return result } -// removeNodeBasicSorted removes a node from the cluster, maintaining the sort -// order. Returns true if the node was removed. unprotected. -func (c *cluster) removeNodeBasicSorted(nodeID string) bool { - return c.noder.RemoveNode(nodeID) -} - -// frag is a struct of basic fragment information. -type frag struct { - field string - view string - shard uint64 -} - -func fragsDiff(a, b []frag) []frag { - m := make(map[frag]uint64) - - for _, y := range b { - m[y]++ - } - - var ret []frag - for _, x := range a { - if m[x] > 0 { - m[x]-- - continue - } - ret = append(ret, x) - } - - return ret -} - -type fragsByHost map[string][]frag - -type viewsByField map[string][]string - -func (a viewsByField) addView(field, view string) { - a[field] = append(a[field], view) -} - -func (c *cluster) fragsByHost(idx *Index) fragsByHost { - // fieldViews is a map of field to slice of views. - fieldViews := make(viewsByField) - - for _, field := range idx.Fields() { - for _, view := range field.views() { - fieldViews.addView(field.Name(), view.name) - } - } - return c.fragCombos(idx.Name(), idx.AvailableShards(includeRemote), fieldViews) -} - -// fragCombos returns a map (by uri) of lists of fragments for a given index -// by creating every combination of field/view specified in `fieldViews` up -// for the given set of shards with data. -func (c *cluster) fragCombos(idx string, availableShards *roaring.Bitmap, fieldViews viewsByField) fragsByHost { - // Create a snapshot of the cluster to use for node/partition calculations. - snap := c.NewSnapshot() - - t := make(fragsByHost) - _ = availableShards.ForEach(func(i uint64) error { - nodes := snap.ShardNodes(idx, i) - for _, n := range nodes { - // for each field/view combination: - for field, views := range fieldViews { - for _, view := range views { - t[n.ID] = append(t[n.ID], frag{field, view, i}) - } - } - } - return nil - }) - return t -} - -// diff compares c with another cluster and determines if a node is being -// added or removed. An error is returned for any case other than where -// exactly one node is added or removed. unprotected. -func (c *cluster) diff(other *cluster) (action string, nodeID string, err error) { - cNodes := c.noder.Nodes() - otherNodes := other.noder.Nodes() - lenFrom := len(cNodes) - lenTo := len(otherNodes) - // Determine if a node is being added or removed. - if lenFrom == lenTo { - return "", "", errors.New("clusters are the same size") - } - if lenFrom < lenTo { - // Adding a node. - if lenTo-lenFrom > 1 { - return "", "", errors.New("adding more than one node at a time is not supported") - } - action = resizeJobActionAdd - // Determine the node ID that is being added. - for _, n := range otherNodes { - if c.unprotectedNodeByID(n.ID) == nil { - nodeID = n.ID - break - } - } - } else if lenFrom > lenTo { - // Removing a node. - if lenFrom-lenTo > 1 { - return "", "", errors.New("removing more than one node at a time is not supported") - } - action = resizeJobActionRemove - // Determine the node ID that is being removed. - for _, n := range cNodes { - if other.unprotectedNodeByID(n.ID) == nil { - nodeID = n.ID - break - } - } - } - return action, nodeID, nil -} - -// fragSources returns a list of ResizeSources - for each node in the `to` cluster - -// required to move from cluster `c` to cluster `to`. unprotected. -func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*ResizeSource, error) { - m := make(map[string][]*ResizeSource) - - // Determine if a node is being added or removed. - action, diffNodeID, err := c.diff(to) - if err != nil { - return nil, errors.Wrap(err, "diffing") - } - - // Initialize the map with all the nodes in `to`. - for _, n := range to.noder.Nodes() { - m[n.ID] = nil - } - - // If a node is being added, the source can be confined to the - // primary fragments (i.e. no need to use replicas as source data). - // In this case, source fragments can be based on a cluster with - // replica = 1. - // If a node is being removed, however, then it will most likely - // require that a replica fragment be the source data. - srcCluster := c - if action == resizeJobActionAdd && c.ReplicaN > 1 { - srcCluster = newCluster() - srcCluster.noder.SetNodes(topology.Nodes(c.noder.Nodes()).Clone()) - srcCluster.Hasher = c.Hasher - srcCluster.partitionN = c.partitionN - srcCluster.ReplicaN = 1 - } - - // Represents the fragment location for the from/to clusters. - fFrags := c.fragsByHost(idx) - tFrags := to.fragsByHost(idx) - - // srcFrags is the frag map based on a source cluster of replica = 1. - srcFrags := srcCluster.fragsByHost(idx) - - // srcNodesByFrag is the inverse representation of srcFrags. - srcNodesByFrag := make(map[frag]string) - for nodeID, frags := range srcFrags { - // If a node is being removed, don't consider it as a source. - if action == resizeJobActionRemove && nodeID == diffNodeID { - continue - } - for _, frag := range frags { - srcNodesByFrag[frag] = nodeID - } - } - - // Get the frag diff for each nodeID. - diffs := make(fragsByHost) - for nodeID, frags := range tFrags { - if _, ok := fFrags[nodeID]; ok { - diffs[nodeID] = fragsDiff(frags, fFrags[nodeID]) - } else { - diffs[nodeID] = frags - } - } - - // Get the ResizeSource for each diff. - for nodeID, diff := range diffs { - m[nodeID] = []*ResizeSource{} - for _, frag := range diff { - // If there is no valid source node ID for a fragment, - // it likely means that the replica factor was not - // high enough for the remaining nodes to contain - // the fragment. - srcNodeID, ok := srcNodesByFrag[frag] - if !ok { - return nil, errors.New("not enough data to perform resize (replica factor may need to be increased)") - } - - src := &ResizeSource{ - Node: c.unprotectedNodeByID(srcNodeID), - Index: idx.Name(), - Field: frag.field, - View: frag.view, - Shard: frag.shard, - } - - m[nodeID] = append(m[nodeID], src) - } - } - - return m, nil -} - -// translationNodes returns a list of translationResizeNodes - for each node -// in the `to` cluster - required to move from cluster `c` to cluster `to`. unprotected. -// Because the parition scheme for every index is the same, this is used as a template -// to create index-specific `TranslationResizeSource`s. -func (c *cluster) translationNodes(to *cluster) (map[string][]*translationResizeNode, error) { - m := make(map[string][]*translationResizeNode) - - // Determine if a node is being added or removed. - action, diffNodeID, err := c.diff(to) - if err != nil { - return nil, errors.Wrap(err, "diffing") - } - - // Initialize the map with all the nodes in `to`. - for _, n := range to.noder.Nodes() { - m[n.ID] = nil - } - - // Create a snapshot of the cluster to use for node/partition calculations. - fSnap := c.NewSnapshot() - toSnap := topology.NewClusterSnapshot(to.noder, c.Hasher, c.partitionAssigner, to.ReplicaN) - - for pid := 0; pid < c.partitionN; pid++ { - fNodes := fSnap.PartitionNodes(pid) - tNodes := toSnap.PartitionNodes(pid) - - // For `to` cluster, we include all nodes containing a - // replica for the partition. The source for each replica - // will be the primary in the `from` cluster. For the `from` - // cluster, we only need the first node, unless that node is - // being removed, then we use the second node. If no second - // node exists in that case, then we have to raise an error - // indicating that not enough replicas exist to support - // the resize. - if len(tNodes) > 0 { - var foundPrimary bool - for i := range fNodes { - if action == resizeJobActionRemove && fNodes[i].ID == diffNodeID { - continue - } - // We only need to add the source if the nodes differ; - // in other words if the primary partition is on the - // same node, it doesn't need to retrieve it. - for n := range tNodes { - if tNodes[n].ID != fNodes[i].ID { - m[tNodes[n].ID] = append(m[tNodes[n].ID], - &translationResizeNode{ - node: fNodes[i], - partitionID: pid, - }) - } - } - foundPrimary = true - break - } - if !foundPrimary { - return nil, ErrResizeNoReplicas - } - } - } - - return m, nil -} - // shardDistributionByIndex returns a map of [nodeID][primaryOrReplica][]uint64, // where the int slices are lists of shards. func (c *cluster) shardDistributionByIndex(indexName string) map[string]map[string][]uint64 { @@ -1014,283 +291,6 @@ func (c *cluster) close() error { return nil } -func (c *cluster) sendTo(node *topology.Node, m Message) error { - if err := c.broadcaster.SendTo(node, m); err != nil { - return errors.Wrap(err, "sending") - } - return nil -} - -func (c *cluster) followResizeInstruction(ctx context.Context, instr *ResizeInstruction) error { - // Make sure the holder has opened. - c.holder.opened.Recv() - - span, _ := tracing.StartSpanFromContext(ctx, "Cluster.followResizeInstruction") - defer span.Finish() - - // Sync the NodeStatus received in the resize instruction. - // Sync schema. - c.logger.Debugf("holder applySchema") - if err := c.holder.applySchema(instr.NodeStatus.Schema); err != nil { - return errors.Wrap(err, "applying schema") - } - - // Sync available shards. - for _, is := range instr.NodeStatus.Indexes { - for _, fs := range is.Fields { - f := c.holder.Field(is.Name, fs.Name) - // if we don't know about a field locally, log an error because - // fields should be created and synced prior to shard creation - if f == nil { - c.logger.Errorf("local field not found: %s/%s", is.Name, fs.Name) - continue - } - - select { - case <-ctx.Done(): - return ctx.Err() - - default: - err := f.loadAvailableShards() - if err != nil { - return errors.Wrapf(err, "getting shards for field: %s/%s", is.Name, f.name) - } - - } - } - } - - // Request each source file in ResizeSources. - for _, src := range instr.Sources { - srcURI := src.Node.URI - c.logger.Printf("get shard %d for index %s from host %s", src.Shard, src.Index, srcURI) - // Retrieve field. - f := c.holder.Field(src.Index, src.Field) - if f == nil { - return newNotFoundError(ErrFieldNotFound, src.Field) - } - - select { - case <-ctx.Done(): - return ctx.Err() - - default: - // Create view. - var v *view - if err := func() (err error) { - v, err = f.createViewIfNotExists(src.View) - return err - }(); err != nil { - return errors.Wrap(err, "creating view") - } - - // Create the local fragment. - frag, err := v.CreateFragmentIfNotExists(src.Shard) - if err != nil { - return errors.Wrap(err, "creating fragment") - } - - // Stream shard from remote node. - c.logger.Printf("retrieve shard %d for index %s from host %s", src.Shard, src.Index, srcURI) - rd, err := c.InternalClient.RetrieveShardFromURI(ctx, src.Index, src.Field, src.View, src.Shard, srcURI) - 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 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. - if err == ErrFragmentNotFound { - continue - } - return errors.Wrap(err, "retrieving shard") - } else if rd == nil { - return fmt.Errorf("shard %v doesn't exist on host: %s", src.Shard, srcURI) - } - - // Write to local field and always close reader. - if err := func() error { - defer rd.Close() - _, err := frag.ReadFrom(rd) - return err - }(); err != nil { - return errors.Wrap(err, "copying remote shard") - } - } - } - - // Request each translation source file in TranslationResizeSources. - for _, src := range instr.TranslationSources { - srcURI := src.Node.URI - - idx := c.holder.Index(src.Index) - if idx == nil { - return newNotFoundError(ErrIndexNotFound, src.Index) - } - - select { - case <-ctx.Done(): - return ctx.Err() - - default: - // Retrieve partition from remote node. - c.logger.Printf("retrieve translate partition %d for index %s from host %s", src.PartitionID, src.Index, srcURI) - rd, err := c.InternalClient.RetrieveTranslatePartitionFromURI(ctx, src.Index, src.PartitionID, srcURI) - if err != nil { - return errors.Wrap(err, "retrieving translate partition") - } else if rd == nil { - return fmt.Errorf("partition %d doesn't exist on host: %s", src.PartitionID, src.Node.URI) - } - - // Write to local store and always close reader. - if err := func() error { - defer rd.Close() - // Get the translate store for this index/partition. - store := idx.TranslateStore(src.PartitionID) - _, err = store.ReadFrom(rd) - return errors.Wrap(err, "reading from reader") - }(); err != nil { - return errors.Wrap(err, "copying remote partition") - } - } - } - - // fire off translation sync - _ = c.translationSyncer.Reset() - - return nil -} - -func (c *cluster) resizeAbortAndBroadcast() error { - if err := c.resizeAbort(); err != nil { - return err - } - return c.broadcaster.SendSync(&ResizeAbortMessage{}) -} - -func (c *cluster) resizeAbort() error { - if c.resizeCancel != nil { - c.resizeCancel() - } - // fire off translation sync - _ = c.translationSyncer.Reset() - return nil -} - -type resizeJob struct { - ID int64 - IDs map[string]bool - Instructions []*ResizeInstruction - Broadcaster broadcaster - - action string - result chan string - - Logger logger.Logger -} - -// newResizeJob returns a new instance of resizeJob. -func newResizeJob(existingNodes []*topology.Node, node *topology.Node, action string) *resizeJob { - - // Build a map of uris to track their resize status. - // The value for a node will be set to true after that node - // has indicated that it has completed all resize instructions. - ids := make(map[string]bool) - - if action == resizeJobActionRemove { - for _, n := range existingNodes { - // Exclude the removed node from the map. - if n.ID == node.ID { - continue - } - ids[n.ID] = false - } - } else if action == resizeJobActionAdd { - for _, n := range existingNodes { - ids[n.ID] = false - } - // Include the added node in the map for tracking. - ids[node.ID] = false - } - - return &resizeJob{ - ID: rand.Int63(), - IDs: ids, - action: action, - result: make(chan string), - Logger: logger.NopLogger, - } -} - -/////////////////////////////////////////// -// Cluster implements the Noder interface. -// This is temporary and should be removed once etcd is fully implemented as -// noder. - -// SetNodes implements the Noder interface. -func (c *cluster) SetNodes(nodes []*topology.Node) {} - -// AppendNode implements the Noder interface. -func (c *cluster) AppendNode(node *topology.Node) {} - -// RemoveNode implements the Noder interface. -func (c *cluster) RemoveNode(nodeID string) bool { - return false -} - -// SetNodeState implements the Noder interface. -func (c *cluster) SetNodeState(nodeID string, state string) {} - -/////////////////////////////////////////// - -func (c *cluster) nodeStatus() (*NodeStatus, error) { - indexes, err := c.holder.Schema() - if err != nil { - return nil, errors.Wrap(err, "getting schema") - } - ns := &NodeStatus{ - Node: c.Node, - Schema: &Schema{Indexes: indexes}, - } - var availableShards *roaring.Bitmap - for _, idx := range ns.Schema.Indexes { - is := &IndexStatus{Name: idx.Name, CreatedAt: idx.CreatedAt} - for _, f := range idx.Fields { - if field := c.holder.Field(idx.Name, f.Name); field != nil { - availableShards = field.AvailableShards(includeRemote) - } else { - availableShards = roaring.NewBitmap() - } - is.Fields = append(is.Fields, &FieldStatus{ - Name: f.Name, - CreatedAt: f.CreatedAt, - AvailableShards: availableShards, - }) - } - ns.Indexes = append(ns.Indexes, is) - } - return ns, nil -} - -// unprotectedPreviousNode returns the node listed before the current node in c.Nodes. -// If there is only one node in the cluster, returns nil. -// If the current node is the first node in the list, returns the last node. -func (c *cluster) unprotectedPreviousNode() *topology.Node { - cNodes := c.noder.Nodes() - if len(cNodes) <= 1 { - return nil - } - - pos := c.nodePositionByID(c.Node.ID) - if pos == -1 { - return nil - } else if pos == 0 { - return cNodes[len(cNodes)-1] - } else { - return cNodes[pos-1] - } -} - // PrimaryReplicaNode returns the node listed before the current node in c.Nodes. // This is different than "previous node" as the first node always returns nil. func (c *cluster) PrimaryReplicaNode() *topology.Node { @@ -1922,43 +922,6 @@ type ClusterStatus struct { Schema *Schema } -// ResizeInstruction contains the instruction provided to a node -// during a cluster resize operation. -type ResizeInstruction struct { - JobID int64 - Node *topology.Node - Primary *topology.Node - Sources []*ResizeSource - TranslationSources []*TranslationResizeSource - NodeStatus *NodeStatus - ClusterStatus *ClusterStatus -} - -// ResizeSource is the source of data for a node acting on a -// ResizeInstruction. -type ResizeSource struct { - Node *topology.Node `protobuf:"bytes,1,opt,name=Node" json:"Node,omitempty"` - Index string `protobuf:"bytes,2,opt,name=Index,proto3" json:"Index,omitempty"` - Field string `protobuf:"bytes,3,opt,name=Field,proto3" json:"Field,omitempty"` - View string `protobuf:"bytes,4,opt,name=View,proto3" json:"View,omitempty"` - Shard uint64 `protobuf:"varint,5,opt,name=Shard,proto3" json:"Shard,omitempty"` -} - -// TranslationResizeSource is the source of translation data for -// a node acting on a ResizeInstruction. -type TranslationResizeSource struct { - Node *topology.Node - Index string - PartitionID int -} - -// translateResizeNode holds the node/partition pairs used -// to create a TranslationResizeSource for each index. -type translationResizeNode struct { - node *topology.Node - partitionID int -} - // Schema contains information about indexes and their configuration. type Schema struct { Indexes []*IndexInfo `json:"indexes"` @@ -2026,14 +989,6 @@ type DeleteViewMessage struct { View string } -// 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 - Node *topology.Node - Error string -} - // NodeStateMessage is an internal message for broadcasting a node's state. type NodeStateMessage struct { NodeID string `protobuf:"bytes,1,opt,name=NodeID,proto3" json:"NodeID,omitempty"` diff --git a/cluster_internal_test.go b/cluster_internal_test.go index 6b992644c..ba70740bc 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -2,11 +2,9 @@ package pilosa import ( - "context" "fmt" "math/rand" "reflect" - "strings" "testing" "testing/quick" "time" @@ -19,59 +17,6 @@ import ( . "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck ) -// Ensure that fragCombos creates the correct fragment mapping. -func TestFragCombos(t *testing.T) { - t.Skip() // skipping due to change partitioning strategy - uri0, err := pnet.NewURIFromAddress("host0") - if err != nil { - t.Fatal(err) - } - uri1, err := pnet.NewURIFromAddress("host1") - if err != nil { - t.Fatal(err) - } - - node0 := &topology.Node{ID: "node0", URI: *uri0} - node1 := &topology.Node{ID: "node1", URI: *uri1} - - c := newCluster() - c.addNodeBasicSorted(node0) - c.addNodeBasicSorted(node1) - - tests := []struct { - idx string - availableShards *roaring.Bitmap - fieldViews viewsByField - expected fragsByHost - }{ - { - idx: "i", - availableShards: roaring.NewBitmap(0, 1, 2), - fieldViews: viewsByField{"f": []string{"v1", "v2"}}, - expected: fragsByHost{ - "node0": []frag{{"f", "v1", uint64(0)}, {"f", "v2", uint64(0)}}, - "node1": []frag{{"f", "v1", uint64(1)}, {"f", "v2", uint64(1)}, {"f", "v1", uint64(2)}, {"f", "v2", uint64(2)}}, - }, - }, - { - idx: "foo", - availableShards: roaring.NewBitmap(0, 1, 2, 3), - fieldViews: viewsByField{"f": []string{"v0"}}, - expected: fragsByHost{ - "node0": []frag{{"f", "v0", uint64(1)}, {"f", "v0", uint64(2)}}, - "node1": []frag{{"f", "v0", uint64(0)}, {"f", "v0", uint64(3)}}, - }, - }, - } - for _, test := range tests { - actual := c.fragCombos(test.idx, test.availableShards, test.fieldViews) - if !reflect.DeepEqual(actual, test.expected) { - t.Errorf("expected: %v, but got: %v", test.expected, actual) - } - - } -} - // newHolderWithTempPath returns a new instance of Holder. func newHolderWithTempPath(tb testing.TB, backend string) *Holder { path, err := testhook.TempDirInDir(tb, *TempDir, "pilosa-holder-") @@ -109,251 +54,6 @@ func newIndexWithTempPath(tb testing.TB, name string) *Index { return index } -// Ensure that fragSources creates the correct fragment mapping. -func TestFragSources(t *testing.T) { - t.Skip() // skipping due to change partitioning strategy - - uri0, err := pnet.NewURIFromAddress("host0") - if err != nil { - t.Fatal(err) - } - uri1, err := pnet.NewURIFromAddress("host1") - if err != nil { - t.Fatal(err) - } - uri2, err := pnet.NewURIFromAddress("host2") - if err != nil { - t.Fatal(err) - } - uri3, err := pnet.NewURIFromAddress("host3") - if err != nil { - t.Fatal(err) - } - - node0 := &topology.Node{ID: "node0", URI: *uri0} - node1 := &topology.Node{ID: "node1", URI: *uri1} - node2 := &topology.Node{ID: "node2", URI: *uri2} - node3 := &topology.Node{ID: "node3", URI: *uri3} - - c1 := newCluster() - c1.ReplicaN = 1 - c1.addNodeBasicSorted(node0) - c1.addNodeBasicSorted(node1) - - c2 := newCluster() - c2.ReplicaN = 1 - c2.addNodeBasicSorted(node0) - c2.addNodeBasicSorted(node1) - c2.addNodeBasicSorted(node2) - - c3 := newCluster() - c3.ReplicaN = 2 - c3.addNodeBasicSorted(node0) - c3.addNodeBasicSorted(node1) - - c4 := newCluster() - c4.ReplicaN = 2 - c4.addNodeBasicSorted(node0) - c4.addNodeBasicSorted(node1) - c4.addNodeBasicSorted(node2) - - c5 := newCluster() - c5.ReplicaN = 2 - c5.addNodeBasicSorted(node0) - c5.addNodeBasicSorted(node1) - c5.addNodeBasicSorted(node2) - c5.addNodeBasicSorted(node3) - - idx := newIndexWithTempPath(t, "i") - - field, err := idx.CreateFieldIfNotExists("f", OptFieldTypeDefault()) - if err != nil { - t.Fatal(err) - } - - // Obtain transaction. - var shard uint64 - tx := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard}) - defer tx.Rollback() - - _, err = field.SetBit(tx, 1, 101, nil) - if err != nil { - t.Fatal(err) - } - PanicOn(tx.Commit()) - - shard = 1 - tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard}) - defer tx.Rollback() - _, err = field.SetBit(tx, 1, ShardWidth*shard+1, nil) - if err != nil { - t.Fatal(err) - } - PanicOn(tx.Commit()) - - shard = 2 - tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard}) - defer tx.Rollback() - - _, err = field.SetBit(tx, 1, ShardWidth*shard+1, nil) - if err != nil { - t.Fatal(err) - } - PanicOn(tx.Commit()) - - shard = 3 - tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard}) - defer tx.Rollback() - - _, err = field.SetBit(tx, 1, ShardWidth*shard+1, nil) - if err != nil { - t.Fatal(err) - } - PanicOn(tx.Commit()) - - tests := []struct { - from *cluster - to *cluster - idx *Index - expected map[string][]*ResizeSource - err string - }{ - { - from: c1, - to: c2, - idx: idx, - expected: map[string][]*ResizeSource{ - "node0": {}, - "node1": {}, - "node2": { - {&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(0)}, - {&topology.Node{ID: "node1", URI: pnet.URI{Scheme: "http", Host: "host1", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(2)}, - }, - }, - err: "", - }, - { - from: c4, - to: c3, - idx: idx, - expected: map[string][]*ResizeSource{ - "node0": { - {&topology.Node{ID: "node1", URI: pnet.URI{Scheme: "http", Host: "host1", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(1)}, - }, - "node1": { - {&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(0)}, - {&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(2)}, - }, - }, - err: "", - }, - { - from: c5, - to: c4, - idx: idx, - expected: map[string][]*ResizeSource{ - "node0": { - {&topology.Node{ID: "node2", URI: pnet.URI{Scheme: "http", Host: "host2", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(0)}, - {&topology.Node{ID: "node2", URI: pnet.URI{Scheme: "http", Host: "host2", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(2)}, - }, - "node1": { - {&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(3)}, - }, - "node2": {}, - }, - err: "", - }, - { - from: c2, - to: c4, - idx: idx, - expected: nil, - err: "clusters are the same size", - }, - { - from: c1, - to: c5, - idx: idx, - expected: nil, - err: "adding more than one node at a time is not supported", - }, - { - from: c5, - to: c1, - idx: idx, - expected: nil, - err: "removing more than one node at a time is not supported", - }, - } - for _, test := range tests { - - actual, err := (test.from).fragSources(test.to, test.idx) - if test.err != "" { - if !strings.Contains(err.Error(), test.err) { - t.Fatalf("expected error: %s, got: %s", test.err, err.Error()) - } - } else { - if err != nil { - t.Fatal(err) - } - if !reflect.DeepEqual(actual, test.expected) { - t.Errorf("expected: %v, but got: %v", test.expected, actual) - } - } - } -} - -// Ensure that fragSources creates the correct fragment mapping. -func TestResizeJob(t *testing.T) { - - uri0, err := pnet.NewURIFromAddress("host0") - if err != nil { - t.Fatal(err) - } - uri1, err := pnet.NewURIFromAddress("host1") - if err != nil { - t.Fatal(err) - } - uri2, err := pnet.NewURIFromAddress("host2") - if err != nil { - t.Fatal(err) - } - - node0 := &topology.Node{ID: "node0", URI: *uri0} - node1 := &topology.Node{ID: "node1", URI: *uri1} - node2 := &topology.Node{ID: "node2", URI: *uri2} - - tests := []struct { - existingNodes []*topology.Node - node *topology.Node - action string - expectedIDs map[string]bool - }{ - { - existingNodes: []*topology.Node{node0, node1}, - node: node2, - action: resizeJobActionAdd, - expectedIDs: map[string]bool{node0.ID: false, node1.ID: false, node2.ID: false}, - }, - { - existingNodes: []*topology.Node{node0, node1, node2}, - node: node2, - action: resizeJobActionRemove, - expectedIDs: map[string]bool{node0.ID: false, node1.ID: false}, - }, - } - for _, test := range tests { - - actual := newResizeJob(test.existingNodes, test.node, test.action) - if err != nil { - t.Fatal(err) - } - if !reflect.DeepEqual(actual.IDs, test.expectedIDs) { - t.Errorf("expected: %v, but got: %v", test.expectedIDs, actual.IDs) - } - } -} - // Ensure the cluster can fairly distribute partitions across the nodes. func TestCluster_Owners(t *testing.T) { c := cluster{ @@ -502,60 +202,6 @@ func TestCluster_Nodes(t *testing.T) { }) } -func TestCluster_PreviousNode(t *testing.T) { - node0 := &topology.Node{ID: "node0"} - node1 := &topology.Node{ID: "node1"} - node2 := &topology.Node{ID: "node2"} - - t.Run("OneNode", func(t *testing.T) { - c := newCluster() - c.addNodeBasicSorted(node0) - - c.Node = node0 - if prev := c.unprotectedPreviousNode(); prev != nil { - t.Errorf("expected: nil, but got: %v", prev) - } - }) - - t.Run("TwoNode", func(t *testing.T) { - c := newCluster() - c.addNodeBasicSorted(node0) - c.addNodeBasicSorted(node1) - - c.Node = node0 - if prev := c.unprotectedPreviousNode(); prev != node1 { - t.Errorf("expected: node1, but got: %v", prev) - } - - c.Node = node1 - if prev := c.unprotectedPreviousNode(); prev != node0 { - t.Errorf("expected: node0, but got: %v", prev) - } - }) - - t.Run("ThreeNode", func(t *testing.T) { - c := newCluster() - c.addNodeBasicSorted(node0) - c.addNodeBasicSorted(node1) - c.addNodeBasicSorted(node2) - - c.Node = node0 - if prev := c.unprotectedPreviousNode(); prev != node2 { - t.Errorf("expected: node2, but got: %v", prev) - } - - c.Node = node1 - if prev := c.unprotectedPreviousNode(); prev != node0 { - t.Errorf("expected: node0, but got: %v", prev) - } - - c.Node = node2 - if prev := c.unprotectedPreviousNode(); prev != node1 { - t.Errorf("expected: node1, but got: %v", prev) - } - }) -} - func TestAE(t *testing.T) { t.Run("AbortDoesn'tBlockUninitialized", func(t *testing.T) { c := newCluster() @@ -615,14 +261,3 @@ func TestAE(t *testing.T) { } }) } - -func TestTranslateIndexKey(t *testing.T) { - c := newCluster() - node0 := &topology.Node{ID: "node0"} - c.addNodeBasicSorted(node0) - c.holder = newHolderWithTempPath(t, "rbf") - _, e := c.translateIndexKey(context.Background(), "i", "a", false) - if e == nil { - t.Fatal("expecting error") - } -} diff --git a/disco/disco.go b/disco/disco.go index 799f4e82f..585c7a43a 100644 --- a/disco/disco.go +++ b/disco/disco.go @@ -57,7 +57,6 @@ const ( ClusterStateStarting ClusterState = "STARTING" // cluster is starting and some internal services are not ready yet. ClusterStateDegraded ClusterState = "DEGRADED" // cluster is running but we've lost some # of hosts >0 but < replicaN. Only read queries are allowed. ClusterStateNormal ClusterState = "NORMAL" // cluster is up and running. - ClusterStateResizing ClusterState = "RESIZING" // cluster is replicating data to other nodes. ClusterStateDown ClusterState = "DOWN" // cluster is unable to serve queries. ) @@ -67,11 +66,9 @@ const ( NodeStateUnknown NodeState = "UNKNOWN" NodeStateStarting NodeState = "STARTING" NodeStateStarted NodeState = "STARTED" - NodeStateResizing NodeState = "RESIZING" ) type Stator interface { - // Started will mark the actual node as already started. // It must be called after all initialization processes // are up and running. @@ -83,15 +80,7 @@ type Stator interface { // - If all nodes are up and running: "NORMAL" // - If number of DOWN nodes is lower than number of replicas: "DEGRADED" // - If number of unresponsive nodes is greater than (or equal to) the number of replicas: "DOWN" - // - If any of the nodes started a resize operation, or a new - // node was specifically added or removed from the cluster: "RESIZING" ClusterState(context.Context) (ClusterState, error) - - // NodeState returns the specific state of a node given its ID. - NodeState(context.Context, string) (NodeState, error) - - // NodeStates will return all the states by node ID of the actual nodes in the cluster. - NodeStates(context.Context) (map[string]NodeState, error) } // Schema is a map of all indexes, each of those being a map of fields, then @@ -141,21 +130,6 @@ type Metadator interface { SetMetadata(ctx context.Context, metadata []byte) error } -// Resizer triggers resizing the node and changes cluster state into RESIZING. -// We can also return some kind of handler from Resize function (e.g. key-value) -type Resizer interface { - // Resize will trigger a resize event. Node state will change to RESIZE state. - // The returned function can be used to send info about the resize process to other nodes. - Resize(ctx context.Context) (func([]byte) error, error) - - // DoneResize will mark the resize event as done. This will be called when all the resize actions are done. - DoneResize() error - - // Watch will give information about a resize event in another node, using its peerID. - // onUpdate function will be called per each event sent by the node in RESIZE state. - Watch(ctx context.Context, peerID string, onUpdate func([]byte) error) error -} - // Sharder is an interface used to maintain the set of availableShards bitmaps // per field. type Sharder interface { @@ -217,14 +191,6 @@ func (n *nopStator) Started(ctx context.Context) error { return nil } -func (n *nopStator) NodeState(context.Context, string) (NodeState, error) { - return NodeStateUnknown, nil -} - -func (n *nopStator) NodeStates(context.Context) (map[string]NodeState, error) { - return nil, nil -} - // NopMetadator represents a Metadator that doesn't do anything. var NopMetadator Metadator = &nopMetadator{} @@ -237,15 +203,6 @@ func (*nopMetadator) SetMetadata(context.Context, []byte) error { return nil } -// NopResizer represents a Resizer that doesn't do anything. -var NopResizer Resizer = &nopResizer{} - -type nopResizer struct{} - -func (*nopResizer) Resize(context.Context) (func([]byte) error, error) { return nil, nil } -func (*nopResizer) DoneResize() error { return nil } -func (*nopResizer) Watch(context.Context, string, func([]byte) error) error { return nil } - // NopSharder represents a Sharder that doesn't do anything. var NopSharder Sharder = &nopSharder{} diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index 384bcfe5a..cb38aee26 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -119,22 +119,6 @@ func (s Serializer) Unmarshal(buf []byte, m pilosa.Message) error { } s.decodeClusterStatus(msg, mt) return nil - case *pilosa.ResizeInstruction: - msg := &pb.ResizeInstruction{} - err := proto.Unmarshal(buf, msg) - if err != nil { - return errors.Wrap(err, "unmarshaling ResizeInstruction") - } - s.decodeResizeInstruction(msg, mt) - return nil - case *pilosa.ResizeInstructionComplete: - msg := &pb.ResizeInstructionComplete{} - err := proto.Unmarshal(buf, msg) - if err != nil { - return errors.Wrap(err, "unmarshaling ResizeInstructionComplete") - } - s.decodeResizeInstructionComplete(msg, mt) - return nil case *pilosa.NodeStateMessage: msg := &pb.NodeStateMessage{} err := proto.Unmarshal(buf, msg) @@ -312,23 +296,6 @@ func (s Serializer) Unmarshal(buf []byte, m pilosa.Message) error { *mt = s.decodeRowMatrix(msg) return nil - case *pilosa.ResizeNodeMessage: - msg := &pb.ResizeNodeMessage{} - err := proto.Unmarshal(buf, msg) - if err != nil { - return errors.Wrap(err, "unmarshaling ResizeNodeMessage") - } - decodeResizeNodeMessage(msg, mt) - return nil - - case *pilosa.ResizeAbortMessage: - msg := &pb.ResizeAbortMessage{} - err := proto.Unmarshal(buf, msg) - if err != nil { - return errors.Wrap(err, "unmarshaling ResizeAbortMessage") - } - decodeResizeAbortMessage(msg, mt) - return nil case *ingest.ShardedRequest: msg := &pb.ShardedIngestRequest{} err := proto.Unmarshal(buf, msg) @@ -368,10 +335,6 @@ func (s Serializer) encodeToProto(m pilosa.Message) proto.Message { return s.encodeDeleteViewMessage(mt) case *pilosa.ClusterStatus: return s.encodeClusterStatus(mt) - case *pilosa.ResizeInstruction: - return s.encodeResizeInstruction(mt) - case *pilosa.ResizeInstructionComplete: - return s.encodeResizeInstructionComplete(mt) case *pilosa.NodeStateMessage: return s.encodeNodeStateMessage(mt) case *pilosa.RecalculateCaches: @@ -414,10 +377,6 @@ func (s Serializer) encodeToProto(m pilosa.Message) proto.Message { return s.encodeTransactionMessage(mt) case *pilosa.AtomicRecord: return s.encodeAtomicRecord(mt) - case *pilosa.ResizeNodeMessage: - return s.encodeResizeNodeMessage(mt) - case *pilosa.ResizeAbortMessage: - return s.encodeResizeAbortMessage(mt) case *ingest.ShardedRequest: return s.encodeShardedIngestRequest(mt) } @@ -606,52 +565,6 @@ func (s Serializer) encodeQueryResponse(m *pilosa.QueryResponse) *pb.QueryRespon return resp } -func (s Serializer) encodeResizeInstruction(m *pilosa.ResizeInstruction) *pb.ResizeInstruction { - return &pb.ResizeInstruction{ - JobID: m.JobID, - Node: s.encodeNode(m.Node), - Primary: s.encodeNode(m.Primary), - Sources: s.encodeResizeSources(m.Sources), - TranslationSources: s.encodeTranslationResizeSources(m.TranslationSources), - NodeStatus: s.encodeNodeStatus(m.NodeStatus), - ClusterStatus: s.encodeClusterStatus(m.ClusterStatus), - } -} - -func (s Serializer) encodeResizeSources(srcs []*pilosa.ResizeSource) []*pb.ResizeSource { - new := make([]*pb.ResizeSource, 0, len(srcs)) - for _, src := range srcs { - new = append(new, s.encodeResizeSource(src)) - } - return new -} - -func (s Serializer) encodeResizeSource(m *pilosa.ResizeSource) *pb.ResizeSource { - return &pb.ResizeSource{ - Node: s.encodeNode(m.Node), - Index: m.Index, - Field: m.Field, - View: m.View, - Shard: m.Shard, - } -} - -func (s Serializer) encodeTranslationResizeSources(srcs []*pilosa.TranslationResizeSource) []*pb.TranslationResizeSource { - new := make([]*pb.TranslationResizeSource, 0, len(srcs)) - for _, src := range srcs { - new = append(new, s.encodeTranslationResizeSource(src)) - } - return new -} - -func (s Serializer) encodeTranslationResizeSource(m *pilosa.TranslationResizeSource) *pb.TranslationResizeSource { - return &pb.TranslationResizeSource{ - Node: s.encodeNode(m.Node), - Index: m.Index, - PartitionID: int32(m.PartitionID), - } -} - func (s Serializer) encodeSchema(m *pilosa.Schema) *pb.Schema { return &pb.Schema{ Indexes: s.encodeIndexInfos(m.Indexes), @@ -838,14 +751,6 @@ func (s Serializer) encodeDeleteViewMessage(m *pilosa.DeleteViewMessage) *pb.Del } } -func (s Serializer) encodeResizeInstructionComplete(m *pilosa.ResizeInstructionComplete) *pb.ResizeInstructionComplete { - return &pb.ResizeInstructionComplete{ - JobID: m.JobID, - Node: s.encodeNode(m.Node), - Error: m.Error, - } -} - func (s Serializer) encodeNodeStateMessage(m *pilosa.NodeStateMessage) *pb.NodeStateMessage { return &pb.NodeStateMessage{ NodeID: m.NodeID, @@ -1034,52 +939,6 @@ func (s Serializer) encodeShardIngestOperation(op *ingest.Operation) *pb.ShardIn return out } -func (s Serializer) decodeResizeInstruction(ri *pb.ResizeInstruction, m *pilosa.ResizeInstruction) { - m.JobID = ri.JobID - m.Node = &topology.Node{} - s.decodeNode(ri.Node, m.Node) - m.Primary = &topology.Node{} - s.decodeNode(ri.Primary, m.Primary) - m.Sources = make([]*pilosa.ResizeSource, len(ri.Sources)) - s.decodeResizeSources(ri.Sources, m.Sources) - m.TranslationSources = make([]*pilosa.TranslationResizeSource, len(ri.TranslationSources)) - s.decodeTranslationResizeSources(ri.TranslationSources, m.TranslationSources) - m.NodeStatus = &pilosa.NodeStatus{} - s.decodeNodeStatus(ri.NodeStatus, m.NodeStatus) - m.ClusterStatus = &pilosa.ClusterStatus{} - s.decodeClusterStatus(ri.ClusterStatus, m.ClusterStatus) -} - -func (s Serializer) decodeResizeSources(srcs []*pb.ResizeSource, m []*pilosa.ResizeSource) { - for i := range srcs { - m[i] = &pilosa.ResizeSource{} - s.decodeResizeSource(srcs[i], m[i]) - } -} - -func (s Serializer) decodeResizeSource(rs *pb.ResizeSource, m *pilosa.ResizeSource) { - m.Node = &topology.Node{} - s.decodeNode(rs.Node, m.Node) - m.Index = rs.Index - m.Field = rs.Field - m.View = rs.View - m.Shard = rs.Shard -} - -func (s Serializer) decodeTranslationResizeSources(srcs []*pb.TranslationResizeSource, m []*pilosa.TranslationResizeSource) { - for i := range srcs { - m[i] = &pilosa.TranslationResizeSource{} - s.decodeTranslationResizeSource(srcs[i], m[i]) - } -} - -func (s Serializer) decodeTranslationResizeSource(rs *pb.TranslationResizeSource, m *pilosa.TranslationResizeSource) { - m.Node = &topology.Node{} - s.decodeNode(rs.Node, m.Node) - m.Index = rs.Index - m.PartitionID = int(rs.PartitionID) -} - func (s Serializer) decodeSchema(sc *pb.Schema, m *pilosa.Schema) { m.Indexes = make([]*pilosa.IndexInfo, len(sc.Indexes)) s.decodeIndexes(sc.Indexes, m.Indexes) @@ -1252,13 +1111,6 @@ func (s Serializer) decodeDeleteViewMessage(pb *pb.DeleteViewMessage, m *pilosa. m.View = pb.View } -func (s Serializer) decodeResizeInstructionComplete(pb *pb.ResizeInstructionComplete, m *pilosa.ResizeInstructionComplete) { - m.JobID = pb.JobID - m.Node = &topology.Node{} - s.decodeNode(pb.Node, m.Node) - m.Error = pb.Error -} - func (s Serializer) decodeNodeStateMessage(pb *pb.NodeStateMessage, m *pilosa.NodeStateMessage) { m.NodeID = pb.NodeID m.State = pb.State @@ -1986,26 +1838,6 @@ func (s Serializer) encodeDecimal(p *pql.Decimal) *pb.Decimal { return retval } -func (s Serializer) encodeResizeNodeMessage(m *pilosa.ResizeNodeMessage) *pb.ResizeNodeMessage { - return &pb.ResizeNodeMessage{ - NodeID: m.NodeID, - Action: m.Action, - } -} - -func (s Serializer) encodeResizeAbortMessage(*pilosa.ResizeAbortMessage) *pb.ResizeAbortMessage { - return &pb.ResizeAbortMessage{} -} - -func decodeResizeNodeMessage(pb *pb.ResizeNodeMessage, m *pilosa.ResizeNodeMessage) { - m.NodeID = pb.NodeID - m.Action = pb.Action -} - -func decodeResizeAbortMessage(pb *pb.ResizeAbortMessage, m *pilosa.ResizeAbortMessage) { - -} - func (s Serializer) decodeShardedIngestRequest(req *pb.ShardedIngestRequest) (*ingest.ShardedRequest, error) { if req == nil || len(req.Ops) == 0 { return &ingest.ShardedRequest{}, nil diff --git a/etcd/embed.go b/etcd/embed.go index 312678d35..2b9d19b25 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -6,9 +6,7 @@ import ( "context" "encoding/json" "fmt" - "log" "net" - "os" "path" "sort" "strings" @@ -59,7 +57,6 @@ var ( _ disco.Schemator = &Etcd{} _ disco.Stator = &Etcd{} _ disco.Metadator = &Etcd{} - _ disco.Resizer = &Etcd{} _ disco.Sharder = &Etcd{} ) @@ -69,7 +66,6 @@ const ( nodePrefix = "/node/" heartbeatPrefix = nodePrefix + "heartbeat/" schemaPrefix = "/schema/" - resizePrefix = nodePrefix + "resize/" metadataPrefix = nodePrefix + "metadata/" shardPrefix = "/shard/" ) @@ -93,15 +89,11 @@ var ( // reach us out of order. type nodeData struct { heartbeatState string - resizeState string metadata []byte topologyNode *topology.Node } func (n *nodeData) computedState() disco.NodeState { - if n.resizeState != "" { - return disco.NodeStateResizing - } if n.heartbeatState != "" { return disco.NodeState(n.heartbeatState) } @@ -116,7 +108,7 @@ type Etcd struct { cli *clientv3.Client cliMu sync.Mutex - heartbeatLeasedKV, resizeLeasedKV *leasedKV + heartbeatLeasedKV *leasedKV // We have a watcher running. watchCancel() cancels its context. watchCancel func() @@ -164,10 +156,6 @@ func (e *Etcd) Close() error { e.watchCancel() } if e.e != nil { - if e.resizeLeasedKV != nil { - e.resizeLeasedKV.Stop() - e.resizeLeasedKV = nil - } if e.heartbeatLeasedKV != nil { e.heartbeatLeasedKV.Stop() } @@ -262,7 +250,7 @@ func (e *Etcd) retryClient(fn func(cli *clientv3.Client) error) (err error) { return errors.Wrap(err, "exhausted all retries") } -func (e *Etcd) parseOptions() *embed.Config { +func (e *Etcd) parseOptions() (*embed.Config, error) { cfg := embed.NewConfig() cfg.LogLevel = "error" cfg.Logger = "zap" @@ -293,8 +281,7 @@ func (e *Etcd) parseOptions() *embed.Config { //check for multiple nodes in the cluster and error if present nodes := strings.Split(e.options.InitCluster, ",") if len(nodes) > 1 { - e.logger.Errorf("Multiple cluster nodes detected - this version of FeatureBase only supports single node. %+v", e.options.InitCluster) - os.Exit(1) + return nil, fmt.Errorf("multiple cluster nodes detected - this version of FeatureBase only supports single node. %+v", e.options.InitCluster) } // %% end sonarcloud ignore %% } @@ -305,24 +292,7 @@ func (e *Etcd) parseOptions() *embed.Config { } if e.options.ClusterURL != "" { - cfg.ClusterState = embed.ClusterStateFlagExisting - cli, err := clientv3.NewFromURL(e.options.ClusterURL) - if err != nil { - panic(err) - } - defer cli.Close() - - log.Println("Cluster Members:") - mIDs, mNames, mURLs := memberList(cli) - - for i, id := range mIDs { - log.Printf("\tid: %d, name: %s, url: %s\n", id, mNames[i], mURLs[i]) - cfg.InitialCluster += "," + mNames[i] + "=" + mURLs[i] - } - - log.Println("Joining Cluster:") - id, name := memberAdd(cli, e.options.APeerURL) - log.Printf("\tid: %d, name: %s\n", id, name) + return nil, errors.New("joining an existing cluster is unsupported") } // can only use tls if not using pre-configured listeners cfg.ClientTLSInfo = transport.TLSInfo{ @@ -336,12 +306,15 @@ func (e *Etcd) parseOptions() *embed.Config { KeyFile: e.options.PeerKeyFile, } - return cfg + return cfg, nil } // Start starts etcd and hearbeat func (e *Etcd) Start(ctx context.Context) (_ disco.InitialClusterState, err error) { - opts := e.parseOptions() + opts, err := e.parseOptions() + if err != nil { + return disco.InitialClusterStateNew, err + } state := disco.InitialClusterState(opts.ClusterState) e.e, err = embed.StartEtcd(opts) @@ -403,24 +376,6 @@ func (e *Etcd) startHeartbeatAndWatcher(ctx context.Context) error { return nil } -func (e *Etcd) NodeState(ctx context.Context, peerID string) (disco.NodeState, error) { - return e.nodeState(ctx, peerID) -} - -func (e *Etcd) nodeState(ctx context.Context, peerID string) (disco.NodeState, error) { - e.nodeMu.Lock() - defer e.nodeMu.Unlock() - err := e.populateNodeStates(ctx) - return e.nodeStates[peerID], err -} - -func (e *Etcd) NodeStates(ctx context.Context) (map[string]disco.NodeState, error) { - e.nodeMu.Lock() - defer e.nodeMu.Unlock() - err := e.populateNodeStates(ctx) - return e.nodeStates, err -} - func (e *Etcd) Started(ctx context.Context) (err error) { return e.heartbeatLeasedKV.Set(ctx, string(disco.NodeStateStarted)) } @@ -465,7 +420,6 @@ func (e *Etcd) ClusterState(ctx context.Context) (out disco.ClusterState, err er var ( heartbeats int = 0 - resize bool starting bool ) e.nodeMu.Lock() @@ -480,8 +434,6 @@ func (e *Etcd) ClusterState(ctx context.Context) (out disco.ClusterState, err er switch state { case disco.NodeStateStarting: starting = true - case disco.NodeStateResizing: - resize = true case disco.NodeStateUnknown: continue } @@ -489,10 +441,6 @@ func (e *Etcd) ClusterState(ctx context.Context) (out disco.ClusterState, err er heartbeats++ } - if resize { - return disco.ClusterStateResizing, nil - } - if starting { return disco.ClusterStateStarting, nil } @@ -508,57 +456,6 @@ func (e *Etcd) ClusterState(ctx context.Context) (out disco.ClusterState, err er return disco.ClusterStateNormal, nil } -func (e *Etcd) Resize(ctx context.Context) (func([]byte) error, error) { - key := path.Join(resizePrefix, e.e.Server.ID().String()) - if e.resizeLeasedKV == nil { - e.resizeLeasedKV = newLeasedKV(e, key, e.options.HeartbeatTTL) - } - - if err := e.resizeLeasedKV.Start(""); err != nil { - return nil, errors.Wrap(err, "Resize: creates a new hearbeat") - } - - return func(value []byte) error { - log.Println("Update progress:", key, string(value)) - return e.putKey(ctx, key, string(value), clientv3.WithIgnoreLease()) - }, nil -} - -func (e *Etcd) DoneResize() error { - if e.resizeLeasedKV != nil { - e.resizeLeasedKV.Stop() - } - - e.resizeLeasedKV = nil - return nil -} - -func (e *Etcd) Watch(ctx context.Context, peerID string, onUpdate func([]byte) error) error { - key := path.Join(resizePrefix, peerID) - for resp := range e.cli.Watch(ctx, key) { - if err := resp.Err(); err != nil { - return errors.Wrapf(err, "Watch: key (%s) response", key) - } - - for _, ev := range resp.Events { - switch ev.Type { - case mvccpb.PUT: - if onUpdate != nil && ev.Kv.Value != nil { - if err := onUpdate(ev.Kv.Value); err != nil { - return err - } - } - - case mvccpb.DELETE: - // nothing to watch - key was deleted - return errors.WithMessagef(disco.ErrKeyDeleted, "Watch key %s", key) - } - } - } - - return nil -} - // parseNodeKey reads heartbeatPrefix + "23" and yields (heartbeatPrefix, "23", nil). func parseNodeKey(key []byte) (prefix string, peerID string, err error) { // we're looking for things starting with nodePrefix @@ -596,12 +493,6 @@ func (e *Etcd) deleteNodeData(key []byte, revision int64) error { e.knownNodes[peerID].metadata = nil e.knownNodes[peerID].topologyNode = &topology.Node{} e.nodeStatesDirty = true - case resizePrefix: - if e.knownNodes[peerID] == nil { - e.knownNodes[peerID] = &nodeData{} - } - e.knownNodes[peerID].resizeState = "" - e.nodeStatesDirty = true default: return fmt.Errorf("node watch: invalid prefix %q", prefix) } @@ -609,7 +500,7 @@ func (e *Etcd) deleteNodeData(key []byte, revision int64) error { } // putNodeData does the actual updating of the node state maps, etc, -// given an incoming heartbeat, metadata, or resizing change. It requires +// given an incoming heartbeat or metadata change. It requires // that you already hold the node mutex. func (e *Etcd) putNodeData(key []byte, value []byte, revision int64) (err error) { prefix, peerID, err := parseNodeKey(key) @@ -640,12 +531,6 @@ func (e *Etcd) putNodeData(key []byte, value []byte, revision int64) (err error) // This saves us one remake of the node later, probably. e.knownNodes[peerID].topologyNode.State = e.knownNodes[peerID].computedState() e.nodeStatesDirty = true - case resizePrefix: - if e.knownNodes[peerID] == nil { - e.knownNodes[peerID] = &nodeData{} - } - e.knownNodes[peerID].resizeState = string(value) - e.nodeStatesDirty = true default: return fmt.Errorf("node watch: invalid prefix %q", prefix) } @@ -728,7 +613,7 @@ func (e *Etcd) watchNodesOnce(ctx context.Context, cli *clientv3.Client) (err er } } -// WatchNodes monitors changes to /heartbeat/, /resizing/, and /metadata/; +// WatchNodes monitors changes to /heartbeat/ and /metadata/; // basically, it catches changes to cluster state, but ignores the schema. func (e *Etcd) WatchNodes() { ctx, cancel := context.WithCancel(context.Background()) @@ -1170,17 +1055,3 @@ func (e *Etcd) NodeIDs() []string { } return ids } - -// SetNodes implements the Noder interface as NOP -// (because we can't force to set nodes for etcd). -func (e *Etcd) SetNodes(nodes []*topology.Node) {} - -// AppendNode implements the Noder interface as NOP -// (because resizer is responsible for adding new nodes). -func (e *Etcd) AppendNode(node *topology.Node) {} - -// RemoveNode implements the Noder interface as NOP -// (because resizer is responsible for removing existing nodes) -func (e *Etcd) RemoveNode(nodeID string) bool { - return false -} diff --git a/etcd/embed_test.go b/etcd/embed_test.go index e1e4581b4..c737c62f9 100644 --- a/etcd/embed_test.go +++ b/etcd/embed_test.go @@ -1,11 +1,14 @@ package etcd import ( + "context" "net/url" "testing" "time" pilosa "github.com/molecula/featurebase/v3" + "github.com/molecula/featurebase/v3/disco" + "github.com/molecula/featurebase/v3/logger" "go.etcd.io/etcd/server/v3/embed" ) @@ -43,3 +46,39 @@ func TestRestartEtcd(t *testing.T) { } e.Close() } + +func TestParseOptions(t *testing.T) { + var e = &Etcd{options: Options{ClusterURL: "http://foo"}, logger: logger.NewLogfLogger(t)} + curl, _ := url.Parse(pilosa.EtcdUnixSocket(t)) + e.options.LClientURL = curl.String() + curl, _ = url.Parse(pilosa.EtcdUnixSocket(t)) + e.options.LPeerURL = curl.String() + + e.options.ClusterURL = "http://foo" + _, err := e.parseOptions() + if err == nil { + t.Fatalf("cluster URL should be rejected") + } + e.options.ClusterURL = "" + e.options.InitCluster = "a,b" + _, err = e.parseOptions() + if AllowCluster() { + if err != nil { + t.Fatalf("expect options parsing to succeed") + } + } else { + t.Logf("no-allow: %v", err) + if err == nil { + t.Fatalf("should have failed to parse a multi-node cluster in non-clustered build") + } + } + + // verify failure on start with invalid options + state, err := e.Start(context.Background()) + if err == nil { + t.Fatalf("should have gotten error starting etcd with invalid options") + } + if state != disco.InitialClusterStateNew { + t.Fatalf("expected cluster state of %q, got %q", disco.InitialClusterStateNew, state) + } +} diff --git a/http_handler.go b/http_handler.go index f0a9d8d48..2f6af1288 100644 --- a/http_handler.go +++ b/http_handler.go @@ -279,8 +279,6 @@ func (h *Handler) Close() error { func (h *Handler) populateValidators() { h.validators = map[string]*queryValidationSpec{} - h.validators["PostClusterResizeAbort"] = queryValidationSpecRequired() - h.validators["PostClusterResizeRemoveNode"] = queryValidationSpecRequired() h.validators["GetExport"] = queryValidationSpecRequired("index", "field", "shard") h.validators["GetIndexes"] = queryValidationSpecRequired() h.validators["GetIndex"] = queryValidationSpecRequired() @@ -479,8 +477,6 @@ var latticeRoutes = []string{"/tables", "/query", "/querybuilder", "/signin"} // // newRouter creates a new mux http router. func newRouter(handler *Handler) http.Handler { router := mux.NewRouter() - router.HandleFunc("/cluster/resize/abort", handler.chkAuthZ(handler.handlePostClusterResizeAbort, authz.Admin)).Methods("POST").Name("PostClusterResizeAbort") - router.HandleFunc("/cluster/resize/remove-node", handler.chkAuthZ(handler.handlePostClusterResizeRemoveNode, authz.Admin)).Methods("POST").Name("PostClusterResizeRemoveNode") // TODO: figure out how to protect these if needed router.PathPrefix("/debug/pprof/").Handler(http.DefaultServeMux).Methods("GET") @@ -2830,82 +2826,6 @@ func parseUint64Slice(s string) ([]uint64, error) { return a, nil } -// handlePostClusterResizeRemoveNode handles POST /cluster/resize/remove-node request. -func (h *Handler) handlePostClusterResizeRemoveNode(w http.ResponseWriter, r *http.Request) { - if !validHeaderAcceptJSON(r.Header) { - http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable) - return - } - - // Decode request. - var req removeNodeRequest - err := json.NewDecoder(r.Body).Decode(&req) - if err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - - removeNode, err := h.api.RemoveNode(req.ID) - if err != nil { - if errors.Cause(err) == ErrNodeIDNotExists { - http.Error(w, "removing node: "+err.Error(), http.StatusNotFound) - } else { - http.Error(w, "removing node: "+err.Error(), http.StatusInternalServerError) - } - return - } - - // Encode response. - w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(removeNodeResponse{ - Remove: removeNode, - }); err != nil { - h.logger.Errorf("response encoding error: %s", err) - } -} - -type removeNodeRequest struct { - ID string `json:"id"` -} - -type removeNodeResponse struct { - Remove *topology.Node `json:"remove"` -} - -// handlePostClusterResizeAbort handles POST /cluster/resize/abort request. -func (h *Handler) handlePostClusterResizeAbort(w http.ResponseWriter, r *http.Request) { - - if !validHeaderAcceptJSON(r.Header) { - http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable) - return - } - err := h.api.ResizeAbort() - var msg string - if err != nil { - switch errors.Cause(err) { - case ErrNodeNotPrimary: - http.Error(w, err.Error(), http.StatusBadRequest) - return - case ErrResizeNotRunning: - msg = err.Error() - default: - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - } - // Encode response. - w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(clusterResizeAbortResponse{ - Info: msg, - }); err != nil { - h.logger.Errorf("response encoding error: %s", err) - } -} - -type clusterResizeAbortResponse struct { - Info string `json:"info"` -} - func (h *Handler) handleRecalculateCaches(w http.ResponseWriter, r *http.Request) { err := h.api.RecalculateCaches(r.Context()) if err != nil { diff --git a/internal_client.go b/internal_client.go index e529e33f0..dae937f9a 100644 --- a/internal_client.go +++ b/internal_client.go @@ -2387,7 +2387,7 @@ func (c *InternalClient) FieldTranslateDataReader(ctx context.Context, index, fi return resp.Body, nil } -// Status returns pilosa cluster state as a string ("NORMAL", "DEGRADED", "DOWN", "RESIZING", ...) +// Status returns pilosa cluster state as a string ("NORMAL", "DEGRADED", "DOWN", ...) func (c *InternalClient) Status(ctx context.Context) (string, error) { span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Status") defer span.Finish() diff --git a/lattice/src/App/Home/ClusterHealth/clusterStatus.ts b/lattice/src/App/Home/ClusterHealth/clusterStatus.ts index 204225a13..6919a954c 100644 --- a/lattice/src/App/Home/ClusterHealth/clusterStatus.ts +++ b/lattice/src/App/Home/ClusterHealth/clusterStatus.ts @@ -1,4 +1,4 @@ -type clusterStatuses = 'NORMAL' | 'DEGRADED' | 'STARTING' | 'RESIZING' | 'DOWN' | 'UNKNOWN'; +type clusterStatuses = 'NORMAL' | 'DEGRADED' | 'STARTING' | 'DOWN' | 'UNKNOWN'; export const CLUSTER_STATUS: { [key in clusterStatuses]: { label: string; status: string }; @@ -16,11 +16,6 @@ export const CLUSTER_STATUS: { label: 'Some nodes are up, but not enough to answer queries.', status: 'error' }, - RESIZING: { - label: - 'Cluster is resizing. Most endpoints are unavailable until the resize completes.', - status: 'info' - }, DOWN: { label: 'Cluster is unable to serve queries.', status: 'disabled' diff --git a/pilosa.go b/pilosa.go index 218e37491..bda9c7ccc 100644 --- a/pilosa.go +++ b/pilosa.go @@ -62,10 +62,8 @@ var ( // ErrPreconditionFailed is returned when specified index/field createdAt timestamps don't match ErrPreconditionFailed = errors.New("precondition failed") - ErrNodeIDNotExists = errors.New("node with provided ID does not exist") - ErrNodeNotPrimary = errors.New("node is not the primary") - ErrResizeNotRunning = errors.New("no resize job currently running") - ErrResizeNoReplicas = errors.New("not enough data to perform resize (replica factor may need to be increased)") + ErrNodeIDNotExists = errors.New("node with provided ID does not exist") + ErrNodeNotPrimary = errors.New("node is not the primary") ErrNotImplemented = errors.New("not implemented") ErrFieldsArgumentRequired = errors.New("fields argument required") diff --git a/server.go b/server.go index baa5f8968..c77369deb 100644 --- a/server.go +++ b/server.go @@ -59,7 +59,6 @@ type Server struct { // nolint: maligned disCo disco.DisCo stator disco.Stator metadator disco.Metadator - resizer disco.Resizer noder topology.Noder sharder disco.Sharder schemator disco.Schemator @@ -402,7 +401,6 @@ func OptServerMaxQueryMemory(v int64) ServerOption { func OptServerDisCo(disCo disco.DisCo, stator disco.Stator, metadator disco.Metadator, - resizer disco.Resizer, noder topology.Noder, sharder disco.Sharder, schemator disco.Schemator) ServerOption { @@ -411,7 +409,6 @@ func OptServerDisCo(disCo disco.DisCo, s.disCo = disCo s.stator = stator s.metadator = metadator - s.resizer = resizer s.noder = noder s.sharder = sharder s.schemator = schemator @@ -455,7 +452,6 @@ func NewServer(opts ...ServerOption) (*Server, error) { disCo: disco.NopDisCo, stator: disco.NopStator, metadator: disco.NopMetadator, - resizer: disco.NopResizer, noder: topology.NewEmptyLocalNoder(), sharder: disco.NopSharder, schemator: disco.NopSchemator, @@ -525,7 +521,6 @@ func NewServer(opts ...ServerOption) (*Server, error) { s.cluster.holder = s.holder s.cluster.disCo = s.disCo s.cluster.stator = s.stator - s.cluster.resizer = s.resizer s.cluster.noder = s.noder s.cluster.sharder = s.sharder @@ -596,6 +591,11 @@ func (s *Server) Open() error { if err != nil { return errors.Wrap(err, "starting DisCo") } + // I'm pretty sure this can't happen, because the path that would have led to it + // happening now generates an error already, but let's be careful. + if initState == disco.InitialClusterStateExisting { + return errors.New("disco reports existing cluster, but this is not supported") + } // Set node ID. s.nodeID = s.disCo.ID() @@ -648,12 +648,6 @@ func (s *Server) Open() error { } // bring up the background tasks for the holder. s.holder.Activate() - // if we joined existing cluster then broadcast "resize on add" message - if initState == disco.InitialClusterStateExisting { - if err := s.cluster.addNode(s.nodeID); err != nil { - return errors.Wrap(err, "adding a node to the existing cluster") - } - } if err := s.stator.Started(context.Background()); err != nil { return errors.Wrap(err, "setting nodeState") @@ -904,6 +898,9 @@ func (s *Server) TTLRemoval(ctx context.Context) { } func (s *Server) monitorAntiEntropy() { + // %% begin sonarcloud ignore %% + // This code isn't really used anymore because of problems with the design, + // but we haven't taken it out yet. But there's no code coverage of it. if s.antiEntropyInterval == 0 || s.cluster.ReplicaN <= 1 { return // anti entropy disabled } @@ -920,25 +917,23 @@ func (s *Server) monitorAntiEntropy() { select { case <-s.closing: return - case <-s.cluster.abortAntiEntropyCh: // receive here so we don't block resizing + case <-s.cluster.abortAntiEntropyCh: + // receive here so we don't block resizing + // ... note that resizing is gone now, but I don't know whether we still need this. continue case <-ticker.C: s.holder.Stats.Count(MetricAntiEntropy, 1, 1.0) } t := time.Now() - state, err := s.cluster.State() + // We used to check for resizing before doing anti-entropy, but resizing is out + // so we don't otherwise care about state. + _, err := s.cluster.State() if err != nil { s.logger.Printf("cluster state error: err=%s", err) continue } - if state == disco.ClusterStateResizing { - continue // don't launch anti-entropy during resize. - // the cluster sets its state to resizing and *then* sends to - // abortAntiEntropyCh before starting to resize - } - // Sync holders. s.logger.Infof("holder sync beginning") s.cluster.muAntiEntropy.Lock() @@ -966,6 +961,7 @@ func (s *Server) monitorAntiEntropy() { break } } + // %% end sonarcloud ignore %% } // receiveMessage represents an implementation of BroadcastHandler. @@ -1027,34 +1023,6 @@ func (s *Server) receiveMessage(m Message) error { return err } - case *ResizeNodeMessage: - switch obj.Action { - case resizeJobActionRemove: - if err := s.cluster.resizeNodeOnRemove(obj.NodeID); err != nil { - return errors.Wrapf(err, "resizing node %s on remove %s", s.cluster.disCo.ID(), obj.NodeID) - } - - case resizeJobActionAdd: - if err := s.cluster.resizeNodeOnAdd(obj.NodeID); err != nil { - return errors.Wrapf(err, "resizing node %s on remove %s", s.cluster.disCo.ID(), obj.NodeID) - } - - default: - return fmt.Errorf("incorrect resizing node action: %s", obj.Action) - } - - case *ResizeInstruction: - err := s.cluster.followResizeInstruction(context.Background(), obj) - if err != nil { - return err - } - - case *ResizeAbortMessage: - err := s.cluster.resizeAbort() - if err != nil { - return err - } - case *RecalculateCaches: s.holder.recalculateCaches() diff --git a/server/cluster_test.go b/server/cluster_test.go index 14d743465..441a0b282 100644 --- a/server/cluster_test.go +++ b/server/cluster_test.go @@ -3,16 +3,12 @@ package server_test import ( "context" - "encoding/json" "fmt" - "net" - "net/http" "reflect" - "strings" "testing" "time" - "github.com/molecula/featurebase/v3" + pilosa "github.com/molecula/featurebase/v3" "github.com/molecula/featurebase/v3/disco" "github.com/molecula/featurebase/v3/server" "github.com/molecula/featurebase/v3/test" @@ -103,7 +99,7 @@ func TestMain_SendReceiveMessage(t *testing.T) { } // Ensure that an empty node comes up in a NORMAL state. -func TestClusterResize_EmptyNode(t *testing.T) { +func TestCluster_EmptyNode(t *testing.T) { m0 := test.RunCommand(t) defer m0.Close() @@ -114,7 +110,9 @@ func TestClusterResize_EmptyNode(t *testing.T) { } // Ensure that a cluster of empty nodes comes up in a NORMAL state. -func TestClusterResize_EmptyNodes(t *testing.T) { +// Do not combine this with TestCluster_EmptyNode; for non-clustering +// builds, this test gets skipped but we still want that test. +func TestCluster_EmptyNodes(t *testing.T) { clus := test.MustRunCluster(t, 3) defer clus.Close() @@ -127,455 +125,6 @@ func TestClusterResize_EmptyNodes(t *testing.T) { } } -// Ensure that adding a node correctly resizes the cluster. -func TestClusterResize_AddNode(t *testing.T) { - t.Run("NoData", func(t *testing.T) { - clus := test.MustRunCluster(t, 3) - defer clus.Close() - - clus.GetNode(0).AssertState(t, disco.ClusterStateNormal, 1*time.Second) - clus.GetNode(1).AssertState(t, disco.ClusterStateNormal, 1*time.Second) - }) - t.Run("WithIndex", func(t *testing.T) { - // Configure node0 - m0 := test.MustRunCluster(t, 1).GetNode(0) - defer m0.Close() - - // Create a client for each node. - client0 := m0.Client() - - // Create indexes and fields on one node. - if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { - t.Fatal(err) - } else if err := client0.CreateField(context.Background(), "i", "f"); err != nil { - t.Fatal(err) - } - - // Configure node1 - m1 := test.NewCommandNode(t) - lsns := make([]*net.TCPListener, 3) - for i := range lsns { - l, err := net.Listen("tcp", "localhost:0") - if err != nil { - t.Fatal(err) - } - lsns[i] = l.(*net.TCPListener) - } - portsCfg := test.GenPortsConfig(t, test.NewPorts(lsns)) - - m1.Config.Etcd = portsCfg[0].Etcd - m1.Config.Name = portsCfg[0].Name - m1.Config.Cluster.Name = portsCfg[0].Cluster.Name - m1.Config.BindGRPC = portsCfg[0].BindGRPC - m1.Config.GRPCListener = portsCfg[0].GRPCListener - - err := m1.Start() - if err != nil { - t.Fatal(err) - } - - defer m1.Close() - - m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - }) - t.Run("ContinuousShards", func(t *testing.T) { - - // Configure node0 - c := test.MustRunCluster(t, 3) - defer c.Close() - - m0 := c.GetNode(0) - defer m0.Close() - - // Create a client for each node. - client0 := m0.Client() - - // Create indexes and fields on one node. - if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { - t.Fatal(err) - } else if err := client0.CreateField(context.Background(), "i", "f"); err != nil { - t.Fatal(err) - } - - col := pilosa.ShardWidth + 20 - - // Write data on first node. - if _, err := m0.Queryf(t, "i", "", ` - Set(1, f=1) - Set(%d, f=1) - `, col); err != nil { - t.Fatal(err) - } - // exp is the expected result for the Row queries that follow. - exp := fmt.Sprintf(`{"results":[{"columns":[1,%d]}]}`, col) - - // Verify the data exists on the single node. - m0.QueryExpect(t, "i", "", `Row(f=1)`, exp) - - // Configure node1 - m1 := c.GetNode(1) - defer m1.Close() - - m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - - // Verify the data exists on both nodes. - m0.QueryExpect(t, "i", "", `Row(f=1)`, exp) - m1.QueryExpect(t, "i", "", `Row(f=1)`, exp) - }) - - t.Run("OneShard", func(t *testing.T) { - // Configure node0 - c := test.MustRunCluster(t, 3) - defer c.Close() - - // Configure node0 - m0 := c.GetNode(0) - defer m0.Close() - - // Create a client for each node. - client0 := m0.Client() - - // Create indexes and fields on one node. - if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { - t.Fatal(err) - } else if err := client0.CreateField(context.Background(), "i", "f"); err != nil { - t.Fatal(err) - } - - // Write data on first node. - if _, err := m0.Query(t, "i", "", ` - Set(1, f=1) - `); err != nil { - t.Fatal(err) - } - // exp is the expected result for the Row queries that follow. - exp := `{"results":[{"columns":[1]}]}` - - // Verify the data exists on the single node. - m0.QueryExpect(t, "i", "", `Row(f=1)`, exp) - - // Configure node1 - m1 := c.GetNode(1) - defer m1.Close() - - m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - - // Verify the data exists on both nodes. - m0.QueryExpect(t, "i", "", `Row(f=1)`, exp) - m1.QueryExpect(t, "i", "", `Row(f=1)`, exp) - }) - - t.Run("SkippedShard", func(t *testing.T) { - // same reason as the ContinuousShards test above. - c := test.MustRunCluster(t, 3) - defer c.Close() - - // Configure node0 - m0 := c.GetNode(0) - defer m0.Close() - - // Create a client for each node. - client0 := m0.Client() - - // Create indexes and fields on one node. - if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { - t.Fatal(err) - } else if err := client0.CreateField(context.Background(), "i", "f"); err != nil { - t.Fatal(err) - } - - col := pilosa.ShardWidth*2 + 20 - - // Write data on first node. Note that no data is placed on shard 1. - if _, err := m0.Queryf(t, "i", "", ` - Set(1, f=1) - Set(%d, f=1) - `, col); err != nil { - t.Fatal(err) - } - - // exp is the expected result for the Row queries that follow. - exp := fmt.Sprintf(`{"results":[{"columns":[1,%d]}]}`, col) - - // Verify the data exists on the single node. - m0.QueryExpect(t, "i", "", `Row(f=1)`, exp) - - // Configure node1 - m1 := c.GetNode(1) - defer m1.Close() - - m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - - // Verify the data exists on both nodes. - m0.QueryExpect(t, "i", "", `Row(f=1)`, exp) - m1.QueryExpect(t, "i", "", `Row(f=1)`, exp) - }) -} - -// Ensure that adding a node correctly resizes the cluster. -func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) { - t.Run("WithIndex", func(t *testing.T) { - c := test.MustRunCluster(t, 3) - defer c.Close() - - // Configure node0 - m0 := c.GetNode(0) - defer m0.Close() - - // Create a client for each node. - client0 := m0.Client() - - // Create indexes and fields on one node. - if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { - t.Fatal(err) - } else if err := client0.CreateField(context.Background(), "i", "f"); err != nil { - t.Fatal(err) - } - - errc := make(chan error) - go func() { - _, err := m0.API.CreateIndex(context.Background(), "blah", pilosa.IndexOptions{}) - errc <- err - }() - - // Configure node1 - m1 := c.GetNode(1) - defer m1.Close() - - m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - - if err := <-errc; err != nil { - t.Fatalf("error from index creation: %v", err) - } - }) - - t.Run("ContinuousShards", func(t *testing.T) { - c := test.MustRunCluster(t, 3) - defer c.Close() - - // Configure node0 - m0 := c.GetNode(0) - defer m0.Close() - - // Create a client for each node. - client0 := m0.Client() - - // Create indexes and fields on one node. - if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { - t.Fatal(err) - } else if err := client0.CreateField(context.Background(), "i", "f"); err != nil { - t.Fatal(err) - } - - col := pilosa.ShardWidth + 20 - - // Write data on first node. - if _, err := m0.Queryf(t, "i", "", ` - Set(1, f=1) - Set(%d, f=1) - `, col); err != nil { - t.Fatal(err) - } - - // exp is the expected result for the Row queries that follow. - exp := fmt.Sprintf(`{"results":[{"columns":[1,%d]}]}`, col) - - // Verify the data exists on the single node. - m0.QueryExpect(t, "i", "", `Row(f=1)`, exp) - - // Configure node1 - m1 := c.GetNode(1) - defer m1.Close() - - m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - - // Verify the data exists on both nodes. - m0.QueryExpect(t, "i", "", `Row(f=1)`, exp) - m1.QueryExpect(t, "i", "", `Row(f=1)`, exp) - }) - - t.Run("SkippedShard", func(t *testing.T) { - c := test.MustRunCluster(t, 3) - defer c.Close() - - // Configure node0 - m0 := c.GetNode(0) - defer m0.Close() - - // Create a client for each node. - client0 := m0.Client() - - // Create indexes and fields on one node. - if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { - t.Fatal(err) - } else if err := client0.CreateField(context.Background(), "i", "f"); err != nil { - t.Fatal(err) - } - - col := pilosa.ShardWidth*2 + 20 - - // Write data on first node. Note that no data is placed on shard 1. - if _, err := m0.Queryf(t, "i", "", ` - Set(1, f=1) - Set(%d, f=1) - `, col); err != nil { - t.Fatal(err) - } - - // exp is the expected result for the Row queries that follow. - exp := fmt.Sprintf(`{"results":[{"columns":[1,%d]}]}`, col) - - // Verify the data exists on the single node. - m0.QueryExpect(t, "i", "", `Row(f=1)`, exp) - - // Configure node1 - m1 := c.GetNode(1) - defer m1.Close() - - m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - - // Verify the data exists on both nodes. - m0.QueryExpect(t, "i", "", `Row(f=1)`, exp) - m1.QueryExpect(t, "i", "", `Row(f=1)`, exp) - }) - - t.Run("WithIndexKeys", func(t *testing.T) { - c := test.MustRunCluster(t, 3) - defer c.Close() - - // Configure node0 - m0 := c.GetNode(0) - defer m0.Close() - - // Create a client for each node. - client0 := m0.Client() - - // Create indexes and fields on one node. - if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{Keys: true}); err != nil && err != pilosa.ErrIndexExists { - t.Fatal(err) - } else if err := client0.CreateField(context.Background(), "i", "f"); err != nil { - t.Fatal(err) - } - - // Write data on first node. - if _, err := m0.Query(t, "i", "", ` - Set('col1', f=1) - Set('col2', f=1) - `); err != nil { - t.Fatal(err) - } - - // exp is the expected result for the Row queries that follow. - exp := `{"results":[{"columns":[],"keys":["col2","col1"]}]}` - - // Verify the data exists on the single node. - m0.QueryExpect(t, "i", "", `Row(f=1)`, exp) - - // Configure node1 - m1 := c.GetNode(1) - defer m1.Close() - - m0.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - m1.AssertState(t, disco.ClusterStateNormal, 1*time.Second) - - m0.QueryExpect(t, "i", "", `Row(f=1)`, exp) - m1.QueryExpect(t, "i", "", `Row(f=1)`, exp) - }) -} - -func TestClusterResize_RemoveNode(t *testing.T) { - cluster := test.MustRunCluster(t, 3) - defer cluster.Close() - coord := cluster.GetPrimary() - other := cluster.GetNonPrimary() - - mustNodeID := func(baseURL string) string { - body := test.Do(t, "GET", fmt.Sprintf("%s/status", baseURL), "").Body - var resp map[string]interface{} - err := json.Unmarshal([]byte(body), &resp) - if err != nil { - panic(err) - } - if localID, ok := resp["localID"].(string); ok { - return localID - } - panic("localID should be a string") - } - - t.Run("ErrorRemoveInvalidNode", func(t *testing.T) { - resp := test.Do(t, "POST", coord.URL()+"/cluster/resize/remove-node", `{"id": "invalid-node-id"}`) - expBody := "removing node: finding node to remove: node with provided ID does not exist" - if resp.StatusCode != http.StatusNotFound { - t.Fatalf("expected StatusCode %d but got %d", http.StatusNotFound, resp.StatusCode) - } else if strings.TrimSpace(resp.Body) != expBody { - t.Fatalf("expected Body '%s' but got '%s'", expBody, strings.TrimSpace(resp.Body)) - } - }) - - 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)) - - expBody := fmt.Sprintf("removing node: cannot issue node removal request to the node being removed, id=%s: precondition failed", nodeID) - if resp.StatusCode != http.StatusInternalServerError { - t.Fatalf("expected StatusCode %d but got %d", http.StatusInternalServerError, resp.StatusCode) - } else if strings.TrimSpace(resp.Body) != expBody { - t.Fatalf("expected Body '%s' but got '%s'", expBody, strings.TrimSpace(resp.Body)) - } - }) - - 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)) - - expBody := fmt.Sprintf(`removing node: cannot issue node removal request to the node being removed, id=%s: precondition failed`, nodeID) - if resp.StatusCode != http.StatusInternalServerError { - t.Fatalf("expected StatusCode %d but got %d", http.StatusInternalServerError, resp.StatusCode) - } else if strings.TrimSpace(resp.Body) != expBody { - t.Fatalf("expected Body '%s' but got '%s'", expBody, strings.TrimSpace(resp.Body)) - } - }) - - t.Run("ErrorRemoveWithoutReplicas", func(t *testing.T) { - t.Skip("TODO: Unskip the test if you understand it") - client0 := coord.Client() - - // Create indexes and fields on one node. - if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { - t.Fatal(err) - } else if err := client0.CreateField(context.Background(), "i", "f"); err != nil { - t.Fatal(err) - } - - // This is an attempt to ensure there is data on both nodes, but is not guaranteed. - // TODO: Deterministic node IDs would ensure consistent results - setColumns := "" - for i := 0; i < 20; i++ { - setColumns += fmt.Sprintf("Set(%d, f=1) ", i*pilosa.ShardWidth) - } - - if _, err := coord.Query(t, "i", "", setColumns); err != nil { - t.Fatal(err) - } - - nodeID := mustNodeID(other.URL()) - resp := test.Do(t, "POST", coord.URL()+"/cluster/resize/remove-node", fmt.Sprintf(`{"id": "%s"}`, nodeID)) - expBody := "not enough data to perform resize" - if resp.StatusCode != http.StatusInternalServerError { - t.Fatalf("expected StatusCode %d but got %d", http.StatusInternalServerError, resp.StatusCode) - } else if !strings.Contains(resp.Body, expBody) { - t.Fatalf("expected to contain '%s' but got '%s'", expBody, strings.TrimSpace(resp.Body)) - } - }) -} - func TestClusterMutualTLS(t *testing.T) { commandOpts := make([][]server.CommandOption, 3) configs := make([]*server.Config, 3) diff --git a/server/grpc.go b/server/grpc.go index 4c100bb80..60e3c8765 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -125,8 +125,6 @@ func errToStatusError(err error) error { return status.Error(codes.Aborted, err.Error()) case pilosa.ErrClusterDoesNotOwnShard, - pilosa.ErrResizeNoReplicas, - pilosa.ErrResizeNotRunning, pilosa.ErrNodeNotPrimary, pilosa.ErrTooManyWrites, pilosa.ErrNodeIDNotExists: diff --git a/server/handler_test.go b/server/handler_test.go index 72d10ae04..792729d48 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -526,16 +526,6 @@ func TestHandler_Endpoints(t *testing.T) { mustJSONDecode(t, w.Body) }) - t.Run("Abort no resize job", func(t *testing.T) { - w := httptest.NewRecorder() - h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/cluster/resize/abort", nil)) - if w.Code != gohttp.StatusInternalServerError { - bod, err := ioutil.ReadAll(w.Body) - t.Fatalf("unexpected status code: %d, bod: %s, readerr: %v", w.Code, bod, err) - } - // TODO need to test aborting a cluster resize job. this may not be the right place - }) - hldr.SetBit("i0", "f0", 30, (1*pilosa.ShardWidth)+1) hldr.SetBit("i0", "f0", 30, (1*pilosa.ShardWidth)+2) hldr.SetBit("i0", "f0", 30, (3*pilosa.ShardWidth)+4) diff --git a/server/server.go b/server/server.go index e60ccedda..c2972e272 100644 --- a/server/server.go +++ b/server/server.go @@ -459,7 +459,7 @@ func (m *Command) SetupServer() error { } e := petcd.NewEtcd(m.Config.Etcd, m.logger, m.Config.Cluster.ReplicaN, version) - discoOpt := pilosa.OptServerDisCo(e, e, e, e, e, e, e) + discoOpt := pilosa.OptServerDisCo(e, e, e, e, e, e) serverOptions := []pilosa.ServerOption{ pilosa.OptServerAntiEntropyInterval(time.Duration(m.Config.AntiEntropy.Interval)), diff --git a/server/server_test.go b/server/server_test.go index 8ef2290da..b201fb0e6 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -660,100 +660,6 @@ 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 - } - err := cluster.Start() - if err != nil { - t.Fatalf("starting cluster: %v", err) - } - // The anonymous function is necessary so that the slice - // passed to Close() as a receiver is the modified value - // of cluster, because we're removing the last entry from it - // below. - defer func() { - cluster.Close() - }() - - coord, others := cluster.GetPrimary(), cluster.GetNonPrimaries() - - err = cluster.AwaitState(disco.ClusterStateNormal, 100*time.Millisecond) - if err != nil { - t.Fatalf("starting cluster: %v", err) - } - - // 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.AwaitPrimaryState(disco.ClusterStateDegraded, 30*time.Second) - if err != nil { - t.Fatalf("degrading cluster: %v", err) - } - - if _, err := coord.API.RemoveNode(disabled.API.Node().ID); err != nil { - t.Fatalf("removing failed node: %v", err) - } - - err = cluster.AwaitPrimaryState(disco.ClusterStateNormal, 30*time.Second) - if err != nil { - t.Fatalf("removing disabled node: %v", err) - } - - hosts := coord.API.Hosts(context.Background()) - if len(hosts) != 2 { - t.Fatalf("unexpected hosts: %v", hosts) - } -} - -func TestRemoveConcurrentIndexCreation(t *testing.T) { - t.Skip("TestRemoveConcurrentIndexCreation won't be supported under etcd. Under RESIZING, creating/updating schema not allowed now.") - cluster := test.MustNewCluster(t, 3) - for _, c := range cluster.Nodes { - c.Config.Cluster.ReplicaN = 2 - } - err := cluster.Start() - if err != nil { - t.Fatalf("starting cluster: %v", err) - } - defer cluster.Close() - - node0 := cluster.GetNode(0) - err = cluster.AwaitState(disco.ClusterStateNormal, 100*time.Millisecond) - if err != nil { - t.Fatalf("starting cluster: %v", err) - } - - errc := make(chan error) - go func() { - _, err := node0.API.CreateIndex(context.Background(), "blah", pilosa.IndexOptions{}) - errc <- err - }() - - if _, err := node0.API.RemoveNode(cluster.GetNode(2).API.Node().ID); err != nil { - t.Fatalf("removing node: %v", err) - } - - err = cluster.AwaitPrimaryState(disco.ClusterStateNormal, 100*time.Millisecond) - if err != nil { - t.Fatalf("starting cluster: %v", err) - } - - hosts := node0.API.Hosts(context.Background()) - if len(hosts) != 2 { - t.Fatalf("unexpected hosts: %v", hosts) - } - if err := <-errc; err != nil { - t.Fatalf("error from index creation: %v", err) - } -} - // Ensure program imports timestamps as UTC. func TestMain_ImportTimestamp(t *testing.T) { m := test.RunCommand(t) diff --git a/topology/noder.go b/topology/noder.go index 9e9ab3942..b1c3bccd4 100644 --- a/topology/noder.go +++ b/topology/noder.go @@ -10,9 +10,6 @@ import ( type Noder interface { Nodes() []*Node // Remember: this has to be sorted correctly!! PrimaryNodeID(hasher Hasher) string - SetNodes([]*Node) - AppendNode(*Node) - RemoveNode(nodeID string) bool } // localNoder is a simple implementation of the Noder interface @@ -67,30 +64,3 @@ func (n *localNoder) PrimaryNodeID(hasher Hasher) string { } return primaryNode.ID } - -// SetNodes implements the Noder interface. -func (n *localNoder) SetNodes(nodes []*Node) { - n.nodes = nodes -} - -// AppendNode implements the Noder interface. -func (n *localNoder) AppendNode(node *Node) { - n.nodes = append(n.nodes, node) - - // All hosts must be merged in the same order on all nodes in the cluster. - sort.Sort(ByID(n.nodes)) -} - -// RemoveNode implements the Noder interface. -func (n *localNoder) RemoveNode(nodeID string) bool { - i := NodePositionByID(n.nodes, nodeID) - if i < 0 { - return false - } - - copy(n.nodes[i:], n.nodes[i+1:]) - n.nodes[len(n.nodes)-1] = nil - n.nodes = n.nodes[:len(n.nodes)-1] - - return true -} diff --git a/utils_internal_test.go b/utils_internal_test.go index 80667d71d..797253dcf 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + "github.com/molecula/featurebase/v3/etcd" pnet "github.com/molecula/featurebase/v3/net" "github.com/molecula/featurebase/v3/testhook" "github.com/molecula/featurebase/v3/topology" @@ -15,6 +16,9 @@ import ( // NewTestCluster returns a cluster with n nodes and uses a mod-based hasher. func NewTestCluster(tb testing.TB, n int) *cluster { + if n > 1 && etcd.AllowCluster() { + tb.Skipf("cluster size %d not supported in unclustered mode", n) + } path, err := testhook.TempDir(tb, "pilosa-cluster-") if err != nil { panic(err) @@ -26,12 +30,15 @@ func NewTestCluster(tb testing.TB, n int) *cluster { c.Hasher = NewTestModHasher() c.Path = path + nodes := make([]*topology.Node, 0, n) + for i := 0; i < n; i++ { - c.noder.AppendNode(&topology.Node{ + nodes = append(nodes, &topology.Node{ ID: fmt.Sprintf("node%d", i), URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)), }) } + c.noder = topology.NewLocalNoder(nodes) cNodes := c.noder.Nodes() @@ -65,48 +72,6 @@ func (*TestModHasher) Hash(key uint64, n int) int { return int(key) % n } func (*TestModHasher) Name() string { return "mod" } -var _ = NewTestClusterWithReplication // happy linter - -func NewTestClusterWithReplication(tb testing.TB, nNodes, nReplicas, partitionN int) (c *cluster, cleaner func()) { - path, err := testhook.TempDir(tb, "pilosa-cluster-") - if err != nil { - panic(err) - } - - // holder - h := NewHolder(path, mustHolderConfig()) - - // cluster - availableShardFileFlushDuration.Set(100 * time.Millisecond) - c = newCluster() - c.holder = h - c.ReplicaN = nReplicas - c.Hasher = &topology.Jmphasher{} - c.Path = path - c.partitionN = partitionN - - for i := 0; i < nNodes; i++ { - nodeID := fmt.Sprintf("node%d", i) - c.noder.AppendNode(&topology.Node{ - ID: nodeID, - URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)), - }) - } - - cNodes := c.noder.Nodes() - - c.Node = cNodes[0] - - if err := c.holder.Open(); err != nil { - panic(err) - } - - return c, func() { - c.holder.Close() - c.close() - } -} - func TestReplaceFirstFromBack(t *testing.T) { for name, test := range map[string]struct { input string