From fd9d4de31df0d07ff1c2802734096f72ba6f1785 Mon Sep 17 00:00:00 2001 From: Seebs Date: Fri, 10 Jun 2022 14:22:45 -0500 Subject: [PATCH] Remove most of the resize-related logic We had two different, incompatible-with-each-other, and both individually broken, partial implementations of resizing logic. There's the original pre-etcd resize, and then the etcd resize, and neither works, but there's conflicts between the ways they don't work. No attempt to fix this is likely to yield decent results, so instead, we yank them both out entirely, so if we decide to implement resizing (which we will) we won't be confused by stray code pertaining to resizing that's not really hooked up to anything. We're leaving the resize messages in protobuf to avoid renumbering protobuf messages. We rename some of our message types to UNUSED0, etcetera, so that any code still using the old names won't compile, to make sure we get rid of it, but we can't just drop the numbers without breaking rolling restart. The Resize_AddNode tests are removed not just because we don't have resizing, but because they were completely broken anyway and never worked at all. But there's no reason to fix them because they exist to fix the functionality we didn't have and are now removing the vestigial remains of. We also drop the one usage of the AddNode function of Noder, because it was used only by one test code fragment that was creatincg clusters, and that can be done more correctly. There were no other call sites at all. We mark the monitorAntiEntropy function to be ignored by code coverage because it's not actually being covered. There's a separate ticket for removing that entirely. --- api.go | 46 +- apimethod_string.go | 42 +- broadcast.go | 24 +- cluster.go | 1053 +---------------- cluster_internal_test.go | 365 ------ disco/disco.go | 43 - encoding/proto/proto.go | 168 --- etcd/embed.go | 151 +-- etcd/embed_test.go | 39 + http_handler.go | 80 -- internal_client.go | 2 +- .../App/Home/ClusterHealth/clusterStatus.ts | 7 +- pilosa.go | 6 +- server.go | 62 +- server/cluster_test.go | 461 +------- server/grpc.go | 2 - server/handler_test.go | 10 - server/server.go | 2 +- server/server_test.go | 94 -- topology/noder.go | 30 - utils_internal_test.go | 51 +- 21 files changed, 112 insertions(+), 2626 deletions(-) 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