From ab37bf5c7bbfb8c350755e9ab189dcb26d01f509 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Thu, 4 Feb 2021 20:26:41 +0100 Subject: [PATCH 1/2] Apply resizer interface (remove and add node) --- api.go | 74 ++- apimethod_string.go | 22 +- client.go | 6 + cluster.go | 1125 ++++++++++++++++++++------------------ cluster_internal_test.go | 98 +--- etcd/embed.go | 10 +- field.go | 8 + http/client.go | 33 ++ internal/private.pb.go | 182 ++++-- internal/public.pb.go | 190 +++++-- server.go | 50 +- utils_internal_test.go | 25 +- 12 files changed, 1041 insertions(+), 782 deletions(-) diff --git a/api.go b/api.go index 0d5309f09..aee5d2b8a 100644 --- a/api.go +++ b/api.go @@ -1745,21 +1745,19 @@ func (api *API) RemoveNode(id string) (*topology.Node, error) { return nil, errors.Wrap(err, "validating api method") } - removeNode := api.cluster.nodeByID(id) - if removeNode == nil { - if !api.cluster.topologyContainsNode(id) { - return nil, errors.Wrap(ErrNodeIDNotExists, "finding node to remove") - } - removeNode = &topology.Node{ - ID: id, - } + if api.cluster.disCo.ID() == id { + return nil, errors.Wrapf(ErrPreconditionFailed, "the node %s can not be removed", id) } - // Start the resize process (similar to NodeJoin) - err := api.cluster.nodeLeave(id) - if err != nil { - return removeNode, errors.Wrap(err, "calling node leave") + 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 } @@ -1769,14 +1767,17 @@ func (api *API) ResizeAbort() error { return errors.Wrap(err, "validating api method") } - err := api.cluster.completeCurrentJob(resizeJobStateAborted) - return errors.Wrap(err, "complete current job") + return api.cluster.resizeAbortAndBroadcast() } // State returns the cluster state which is usually "NORMAL", but could be // "STARTING", "RESIZING", or potentially others. See cluster.go for more // details. func (api *API) State() (string, error) { + if err := api.validate(apiState); err != nil { + return "", errors.Wrap(err, "validating api method") + } + return api.cluster.State() } @@ -2214,7 +2215,7 @@ const ( //apiSchema // not implemented apiSetCoordinator apiShardNodes - //apiState // not implemented + apiState //apiStatsWithTags // not implemented //apiVersion // not implemented apiViews @@ -2239,6 +2240,29 @@ var methodsResizing = map[apiMethod]struct{}{ apiFragmentData: {}, apiTranslateData: {}, apiResizeAbort: {}, + apiState: {}, +} + +var methodsDegraded = map[apiMethod]struct{}{ + apiExportCSV: {}, + apiFragmentBlockData: {}, + apiFragmentBlocks: {}, + apiField: {}, + apiFieldAttrDiff: {}, + apiIndex: {}, + apiIndexAttrDiff: {}, + apiQuery: {}, + apiRecalculateCaches: {}, + apiRemoveNode: {}, + apiShardNodes: {}, + // apiSchema: {}, + apiState: {}, + apiViews: {}, + apiStartTransaction: {}, + apiFinishTransaction: {}, + apiTransactions: {}, + apiGetTransaction: {}, + apiActiveQueries: {}, } var methodsNormal = map[apiMethod]struct{}{ @@ -2261,15 +2285,13 @@ var methodsNormal = map[apiMethod]struct{}{ apiRecalculateCaches: {}, apiRemoveNode: {}, apiShardNodes: {}, - apiViews: {}, - apiApplySchema: {}, - apiStartTransaction: {}, - apiFinishTransaction: {}, - apiTransactions: {}, - apiGetTransaction: {}, - apiActiveQueries: {}, - apiPastQueries: {}, - apiIDReserve: {}, - apiIDCommit: {}, - apiIDReset: {}, + // apiSchema: {}, + apiState: {}, + apiViews: {}, + apiApplySchema: {}, + apiStartTransaction: {}, + apiFinishTransaction: {}, + apiTransactions: {}, + apiGetTransaction: {}, + apiActiveQueries: {}, } diff --git a/apimethod_string.go b/apimethod_string.go index b694fcb9b..d7fe69dba 100644 --- a/apimethod_string.go +++ b/apimethod_string.go @@ -32,17 +32,23 @@ func _() { _ = x[apiResizeAbort-21] _ = x[apiSetCoordinator-22] _ = x[apiShardNodes-23] - _ = x[apiViews-24] - _ = x[apiApplySchema-25] - _ = x[apiStartTransaction-26] - _ = x[apiFinishTransaction-27] - _ = x[apiTransactions-28] - _ = x[apiGetTransaction-29] + _ = x[apiState-24] + _ = x[apiViews-25] + _ = x[apiApplySchema-26] + _ = x[apiStartTransaction-27] + _ = x[apiFinishTransaction-28] + _ = x[apiTransactions-29] + _ = x[apiGetTransaction-30] + _ = x[apiActiveQueries-31] + _ = x[apiPastQueries-32] + _ = x[apiIDReserve-33] + _ = x[apiIDCommit-34] + _ = x[apiIDReset-35] } -const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiTranslateDataapiFieldapiFieldAttrDiffapiImportapiImportValueapiIndexapiIndexAttrDiffapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSetCoordinatorapiShardNodesapiViewsapiApplySchemaapiStartTransactionapiFinishTransactionapiTransactionsapiGetTransaction" +const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiTranslateDataapiFieldapiFieldAttrDiffapiImportapiImportValueapiIndexapiIndexAttrDiffapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSetCoordinatorapiShardNodesapiStateapiViewsapiApplySchemaapiStartTransactionapiFinishTransactionapiTransactionsapiGetTransactionapiActiveQueriesapiPastQueriesapiIDReserveapiIDCommitapiIDReset" -var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 189, 197, 213, 222, 236, 244, 260, 268, 288, 301, 315, 332, 345, 353, 367, 386, 406, 421, 438} +var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 189, 197, 213, 222, 236, 244, 260, 268, 288, 301, 315, 332, 345, 353, 361, 375, 394, 414, 429, 446, 462, 476, 488, 499, 509} func (i apiMethod) String() string { if i < 0 || i >= apiMethod(len(_apiMethod_index)-1) { diff --git a/client.go b/client.go index ad53cdf4f..fd2bf45e0 100644 --- a/client.go +++ b/client.go @@ -93,6 +93,8 @@ type InternalClient interface { // InternalQueryClient is the internal interface for querying a node. type InternalQueryClient interface { + SchemaNode(ctx context.Context, uri *pnet.URI, views bool) ([]*IndexInfo, error) + QueryNode(ctx context.Context, uri *pnet.URI, index string, queryRequest *QueryRequest) (*QueryResponse, error) // Trasnlate keys on the particular node. The parameter writable informs TranslateStore if we can generate a new ID if any of keys does not exist. @@ -108,6 +110,10 @@ type InternalQueryClient interface { type nopInternalQueryClient struct{} +func (nopInternalQueryClient) SchemaNode(ctx context.Context, uri *pnet.URI, views bool) ([]*IndexInfo, error) { + return nil, nil +} + func (n nopInternalQueryClient) QueryNode(ctx context.Context, uri *pnet.URI, index string, queryRequest *QueryRequest) (*QueryResponse, error) { return nil, nil } diff --git a/cluster.go b/cluster.go index 51aff4e4d..f14a2e970 100644 --- a/cluster.go +++ b/cluster.go @@ -17,12 +17,12 @@ package pilosa import ( "context" "encoding/binary" + "encoding/json" "fmt" "hash/fnv" + "io" "io/ioutil" "math/rand" - "net/http" - "net/url" "os" "path/filepath" "sort" @@ -33,12 +33,10 @@ import ( "github.com/pilosa/pilosa/v2/disco" "github.com/pilosa/pilosa/v2/internal" "github.com/pilosa/pilosa/v2/logger" - pnet "github.com/pilosa/pilosa/v2/net" "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/topology" "github.com/pilosa/pilosa/v2/tracing" "github.com/pkg/errors" - uuid "github.com/satori/go.uuid" "golang.org/x/sync/errgroup" ) @@ -66,12 +64,29 @@ const ( defaultConfirmDownSleep = 1 * time.Second ) -// nodeAction represents a node that is joining or leaving the cluster. -type nodeAction struct { - node *topology.Node - action string +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 @@ -109,21 +124,15 @@ type cluster struct { // nolint: maligned holder *Holder broadcaster broadcaster - joiningLeavingNodes chan nodeAction - - // joining is held open until this node - // receives ClusterStatus from the coordinator. - joining chan struct{} - joined bool - abortAntiEntropyCh chan struct{} muAntiEntropy sync.Mutex translationSyncer TranslationSyncer - mu sync.RWMutex - jobs map[int64]*resizeJob - currentJob *resizeJob + mu sync.RWMutex + jobs map[int64]*resizeJob + currentJob *resizeJob + resizeCancel context.CancelFunc // Close management wg sync.WaitGroup @@ -144,10 +153,8 @@ func newCluster() *cluster { partitionN: topology.DefaultPartitionN, ReplicaN: 1, - joiningLeavingNodes: make(chan nodeAction, 10), // buffered channel - jobs: make(map[int64]*resizeJob), - closing: make(chan struct{}), - joining: make(chan struct{}), + jobs: make(map[int64]*resizeJob), + closing: make(chan struct{}), translationSyncer: NopTranslationSyncer, @@ -158,8 +165,10 @@ func newCluster() *cluster { confirmDownRetries: defaultConfirmDownRetries, confirmDownSleep: defaultConfirmDownSleep, - noder: topology.NewEmptyLocalNoder(), - stator: disco.NopStator, + disCo: disco.NopDisCo, + noder: topology.NewEmptyLocalNoder(), + stator: disco.NopStator, + resizer: disco.NopResizer, } } @@ -212,43 +221,453 @@ func (c *cluster) unprotectedIsCoordinator() bool { return snap.PrimaryFieldTranslationNode().ID == c.Node.ID } -// addNode adds a node to the Cluster and updates and saves the -// new topology. unprotected. -func (c *cluster) addNode(node *topology.Node) error { - // add to cluster - if !c.addNodeBasicSorted(node) { +func (c *cluster) applySchemaWithNewShards(schema *Schema) error { + if schema == nil || len(schema.Indexes) == 0 { return nil } - // add to topology - if c.Topology == nil { - return fmt.Errorf("Cluster.Topology is nil") + if err := c.holder.applySchema(schema); err != nil { + return errors.Wrap(err, "applying schema") } - if !c.Topology.addID(node.ID) { - return nil - } - c.Topology.nodeStates[node.ID] = node.State - // save topology - return c.saveTopology() + // Get and set the shards for each field. + for _, idx := range c.holder.indexes { + for _, fld := range idx.fields { + b, err := c.sharder.Shards(context.Background(), idx.name, fld.name) + if err != nil { + return errors.Wrapf(err, "getting shards for field: %s/%s", idx.name, fld.name) + } + fld.SetRemoteAvailableShards(b) + } + } + + return nil } -// removeNode removes a node from the Cluster and updates and saves the -// new topology. unprotected. -func (c *cluster) removeNode(nodeID string) error { - // remove from cluster - c.removeNodeBasicSorted(nodeID) +// 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 + } - // remove from topology - if c.Topology == nil { - return fmt.Errorf("Cluster.Topology is nil") - } - if !c.Topology.removeID(nodeID) { - return nil + if err := c.applySchemaWithNewShards(schema); err != nil { + return err + } } - // save topology - return c.saveTopology() + 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() + return &ResizeInstruction{ + Node: c.unprotectedNodeByID(myid), + Sources: fragmentSourcesByNode[myid], + TranslationSources: translationSourcesByNode[myid], + NodeStatus: c.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() + return &ResizeInstruction{ + Node: toCluster.unprotectedNodeByID(myid), + Sources: fragmentSourcesByNode[myid], + TranslationSources: translationSourcesByNode[myid], + NodeStatus: c.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()) + if err != nil { + return nil, err + } + + // TODO: replace following code by following code, + // after schemator is implemented + // indexes, err := c.holder.Schema() + // if err != nil { + // return nil, errors.Wrap(err, "getting schema") + // } + indexes := c.holder.Schema() + + return &ClusterStatus{ + State: string(state), + Nodes: c.Nodes(), + Schema: &Schema{Indexes: indexes}, + }, nil +} + +func (c *cluster) remoteSchema() (*Schema, error) { + for _, n := range c.noder.Nodes() { + if c.disCo.ID() == n.ID { + continue + } + + // TODO: replace following line by: + // ii, err := c.InternalClient.SchemaNode(context.Background(), &n.URI, true) + // after we + ii, err := c.InternalClient.SchemaNode(context.Background(), &n.URI, true) + if err != nil { + return nil, errors.Wrapf(err, "getting schema from %s (%v)", n.ID, n.URI) + } + + return &Schema{ii}, nil + } + return nil, nil } // nodeIDs returns the list of IDs in the cluster. @@ -275,21 +694,6 @@ func (c *cluster) State() (string, error) { return string(state), nil } -// unprotectedStatus returns the the cluster's status including what nodes it contains, its ID, and current state. -func (c *cluster) unprotectedStatus() *ClusterStatus { - state, err := c.stator.ClusterState(context.Background()) - if err != nil { - state = disco.ClusterStateUnknown - } - - return &ClusterStatus{ - ClusterID: c.id, - State: string(state), - Nodes: c.noder.Nodes(), - Schema: &Schema{Indexes: c.holder.Schema()}, - } -} - func (c *cluster) nodeByID(id string) *topology.Node { c.mu.RLock() defer c.mu.RUnlock() @@ -876,22 +1280,6 @@ func (c *cluster) setup() error { if err := c.loadTopology(); err != nil { return errors.Wrap(err, "loading topology") } - - c.id = c.Topology.clusterID - - // Only the coordinator needs to consider the .topology file. - if c.isCoordinator() { - err := c.considerTopology() - if err != nil { - return errors.Wrap(err, "considerTopology") - } - } - - // Add the local node to the cluster. - err := c.addNode(c.Node) - if err != nil { - return errors.Wrap(err, "adding local node") - } return nil } @@ -916,13 +1304,6 @@ func (c *cluster) close() error { return nil } -func (c *cluster) markAsJoined() { - if !c.joined { - c.joined = true - close(c.joining) - } -} - func (c *cluster) sendTo(node *topology.Node, m Message) error { if err := c.broadcaster.SendTo(node, m); err != nil { return errors.Wrap(err, "sending") @@ -930,121 +1311,6 @@ func (c *cluster) sendTo(node *topology.Node, m Message) error { return nil } -// unprotectedGenerateResizeJobByAction returns a resizeJob with instructions based on -// the difference between Cluster and a new Cluster with/without uri. -// Broadcaster is associated to the resizeJob here for use in broadcasting -// the resize instructions to other nodes in the cluster. -func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*resizeJob, error) { - j := newResizeJob(c.noder.Nodes(), nodeAction.node, nodeAction.action) - // A *new* node which is being added needs a schema update even if - // there's no data to send it. - var sendSchemaToNewNode string - j.Broadcaster = c.broadcaster - - // toCluster is a clone of Cluster with the new node added/removed for comparison. - toCluster := newCluster() - toCluster.noder.SetNodes(topology.Nodes(c.noder.Nodes()).Clone()) - toCluster.Hasher = c.Hasher - toCluster.partitionN = c.partitionN - toCluster.ReplicaN = c.ReplicaN - if nodeAction.action == resizeJobActionRemove { - toCluster.removeNodeBasicSorted(nodeAction.node.ID) - } else if nodeAction.action == resizeJobActionAdd { - toCluster.addNodeBasicSorted(nodeAction.node) - sendSchemaToNewNode = nodeAction.node.ID - } - - indexes := c.holder.Indexes() - - // 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 - } - - // 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 - } - } - - for _, node := range toCluster.noder.Nodes() { - dataToSend := len(fragmentSourcesByNode[node.ID]) != 0 || len(translationSourcesByNode[node.ID]) != 0 - // If we're adding a new node, that node needs to get a resize - // instruction even if there's no data it needs to read. - // Existing nodes already got the schema and are assumed to be - // up to date on it. - if !dataToSend && node.ID != sendSchemaToNewNode { - j.IDs[node.ID] = true - continue - } - - // Create a snapshot of the cluster to use for node/partition calculations. - snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN) - - instr := &ResizeInstruction{ - JobID: j.ID, - Node: toCluster.unprotectedNodeByID(node.ID), - Primary: snap.PrimaryFieldTranslationNode(), - Sources: fragmentSourcesByNode[node.ID], - TranslationSources: translationSourcesByNode[node.ID], - NodeStatus: c.nodeStatus(), // Include the NodeStatus in order to ensure that schema and availableShards are in sync on the receiving node. - ClusterStatus: c.unprotectedStatus(), - } - j.Instructions = append(j.Instructions, instr) - } - - return j, nil -} - // completeCurrentJob sets the state of the current resizeJob // then removes the pointer to currentJob. func (c *cluster) completeCurrentJob(state string) error { @@ -1067,179 +1333,155 @@ func (c *cluster) unprotectedCompleteCurrentJob(state string) error { return nil } -// followResizeInstruction is run by any node that receives a ResizeInstruction. -func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { - c.logger.Printf("follow resize instruction on %s", c.Node.ID) - // Make sure the cluster status on this node agrees with the Coordinator - // before attempting a resize. - if err := c.mergeClusterStatus(instr.ClusterStatus); err != nil { - return errors.Wrap(err, "merging cluster status") +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") } - c.logger.Printf("done MergeClusterStatus, start goroutine (%s)", c.Node.ID) + // 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.Printf("local field not found: %s/%s", is.Name, fs.Name) + continue + } - // The actual resizing runs in a goroutine because we don't want to block - // the distribution of other ResizeInstructions to the rest of the cluster. - go func() { + select { + case <-ctx.Done(): + return ctx.Err() - // Make sure the holder has opened. - c.holder.opened.Recv() + default: + // Get the shards for the field. + b, err := c.sharder.Shards(ctx, is.Name, f.name) + if err != nil { + return errors.Wrapf(err, "getting shards for field: %s/%s", is.Name, f.name) + } + f.SetRemoteAvailableShards(b) + } + } + } - // Prepare the return message. - complete := &ResizeInstructionComplete{ - JobID: instr.JobID, - Node: instr.Node, - Error: "", + // 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) } - // Stop processing on any error. - if err := func() error { - span, ctx := tracing.StartSpanFromContext(context.Background(), "Cluster.followResizeInstruction") - defer span.Finish() + select { + case <-ctx.Done(): + return ctx.Err() - // 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") + 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") } - // 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.Printf("local field not found: %s/%s", is.Name, fs.Name) - continue - } - if err := f.AddRemoteAvailableShards(fs.AvailableShards); err != nil { - return errors.Wrap(err, "adding remote available shards") - } - } + // Create the local fragment. + frag, err := v.CreateFragmentIfNotExists(src.Shard) + if err != nil { + return errors.Wrap(err, "creating fragment") } - // 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) - } - - // 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 coordinator 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") + // 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 coordinator 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) } - // 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) - } - - // 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") - } + // 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") } + } + } - return nil - }(); err != nil { - complete.Error = err.Error() + // 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) } - if err := c.sendTo(instr.Primary, complete); err != nil { - c.logger.Printf("sending resizeInstructionComplete error: err=%s", err) + 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") + } } - }() + } + return nil } -func (c *cluster) markResizeInstructionComplete(complete *ResizeInstructionComplete) error { - j := c.job(complete.JobID) - - // Abort the job if an error exists in the complete object. - if complete.Error != "" { - j.result <- resizeJobStateAborted - return errors.New(complete.Error) +func (c *cluster) resizeAbortAndBroadcast() error { + if err := c.resizeAbort(); err != nil { + return err } + return c.broadcaster.SendSync(&ResizeAbortMessage{}) +} - j.mu.Lock() - defer j.mu.Unlock() - - if j.isComplete() { - return fmt.Errorf("resize job %d is no longer running", j.ID) +func (c *cluster) resizeAbort() error { + if c.resizeCancel != nil { + c.resizeCancel() } - - // Mark host complete. - j.IDs[complete.Node.ID] = true - - if !j.nodesArePending() { - j.result <- resizeJobStateDone - } - return nil } @@ -1555,140 +1797,6 @@ func (c *cluster) loadTopology() error { return nil } -// saveTopology writes the current topology to disk. unprotected. -func (c *cluster) saveTopology() error { - if err := os.MkdirAll(c.Path, 0777); err != nil { - return errors.Wrap(err, "creating directory") - } - - if buf, err := proto.Marshal(encodeTopology(c.Topology)); err != nil { - return errors.Wrap(err, "marshalling") - } else if err := ioutil.WriteFile(filepath.Join(c.Path, ".topology"), buf, 0666); err != nil { - return errors.Wrap(err, "writing file") - } - return nil -} - -func (c *cluster) considerTopology() error { - // Create ClusterID if one does not already exist. - if c.id == "" { - u := uuid.NewV4() - c.id = u.String() - c.Topology.clusterID = c.id - } - - if c.Static { - return nil - } - - // If there is no .topology file, it's safe to proceed. - if len(c.Topology.nodeIDs) == 0 { - return nil - } - - // The local node (coordinator) must be in the .topology. - if !c.Topology.ContainsID(c.Node.ID) { - return fmt.Errorf("coordinator %s is not in topology: %v", c.Node.ID, c.Topology.nodeIDs) - } - - // Keep the cluster in state "STARTING" until hearing from all nodes. - // Topology contains 2+ hosts. - return nil -} - -// band aid to protect against false nodeLeave events from memberlist -// the test is the lightest weight endpoint of the node in question /version -// TODO provide more robust solution to false nodeLeave events -func (c *cluster) confirmNodeDown(uri pnet.URI) bool { - u := url.URL{ - Scheme: uri.Scheme, - Host: uri.HostPort(), - Path: "version", - } - req, err := http.NewRequest("GET", u.String(), nil) - if err != nil { - c.logger.Printf("bad request:%s %s", u.String(), err) - return false - } - for i := 0; i < c.confirmDownRetries; i++ { - ctx, cancel := context.WithTimeout(context.Background(), c.confirmDownSleep*2) - defer cancel() - resp, err := http.DefaultClient.Do(req.WithContext(ctx)) - var bod []byte - if err == nil { - bod, err = ioutil.ReadAll(resp.Body) - if resp.StatusCode == 200 { - return false - } - } - - c.logger.Printf("NodeLeave confirm with %s %d. err: '%v' bod: '%s'", uri.HostPort(), i, err, bod) - time.Sleep(c.confirmDownSleep) - } - return true -} - -// nodeLeave initiates the removal of a node from the cluster. -func (c *cluster) nodeLeave(nodeID string) error { - c.abortAntiEntropy() - // Technically there is a race condition here which could - // allow the anti-entropy process to re-start (and acquire - // the lock) before this lock has time to succeed. In that - // case, the user would have to wait through an entire - // anti-entropy cycle. We decided it wasn't worth the - // complexity (of, for example, implementing this with - // channels) to avoid that rare case. - c.muAntiEntropy.Lock() - defer c.muAntiEntropy.Unlock() - - c.mu.Lock() - defer c.mu.Unlock() - // Refuse the request if this is not the coordinator. - if !c.unprotectedIsCoordinator() { - return fmt.Errorf("node removal requests are only valid on the coordinator node: %s", - c.unprotectedCoordinatorNode().ID) - } - - state, err := c.stator.ClusterState(context.TODO()) - if err != nil || (state != disco.ClusterStateNormal && state != disco.ClusterStateDegraded) { - return fmt.Errorf("cluster must be '%s' or '%s' to remove a node but is '%s', error: %v", - ClusterStateNormal, ClusterStateDegraded, state, err) - } - - // Ensure that node is in the cluster. - if !c.topologyContainsNode(nodeID) { - return fmt.Errorf("Node is not a member of the cluster: %s", nodeID) - } - - // Prevent removing the coordinator node (this node). - if nodeID == c.Node.ID { - return fmt.Errorf("coordinator cannot be removed; first, make a different node the new coordinator") - } - - // See if resize job can be generated - if _, err := c.unprotectedGenerateResizeJobByAction( - nodeAction{ - node: &topology.Node{ID: nodeID}, - action: resizeJobActionRemove}, - ); err != nil { - return errors.Wrap(err, "generating job") - } - - // If the holder does not yet contain data, go ahead and remove the node. - if ok, err := c.holder.HasData(); !ok && err == nil { - if err := c.removeNode(nodeID); err != nil { - return errors.Wrap(err, "removing node") - } - return nil - } else if err != nil { - return errors.Wrap(err, "checking if holder has data") - } - - c.joiningLeavingNodes <- nodeAction{node: &topology.Node{ID: nodeID}, action: resizeJobActionRemove} - - return nil -} - func (c *cluster) nodeStatus() *NodeStatus { ns := &NodeStatus{ Node: c.Node, @@ -1714,53 +1822,6 @@ func (c *cluster) nodeStatus() *NodeStatus { return ns } -func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error { - c.mu.Lock() - defer c.mu.Unlock() - c.logger.Printf("merge cluster status: node=%s cluster=%v, topologySize=%v", c.Node.ID, cs, len(c.Topology.nodeIDs)) - // Ignore status updates from self (coordinator). - if c.unprotectedIsCoordinator() { - return nil - } - - // Set ClusterID. - c.unprotectedSetID(cs.ClusterID) - - officialNodes := cs.Nodes - - // Add all nodes from the coordinator. - for _, node := range officialNodes { - if err := c.addNode(node); err != nil { - return errors.Wrap(err, "adding node") - } - } - - // Remove any nodes not specified by the coordinator - // except for self. Generate a list to remove first - // so that nodes aren't removed mid-loop. - nodeIDsToRemove := []string{} - for _, node := range c.noder.Nodes() { - // Don't remove this node. - if node.ID == c.Node.ID { - continue - } - if topology.Nodes(officialNodes).ContainsID(node.ID) { - continue - } - nodeIDsToRemove = append(nodeIDsToRemove, node.ID) - } - - for _, nodeID := range nodeIDsToRemove { - if err := c.removeNode(nodeID); err != nil { - return errors.Wrap(err, "removing node") - } - } - - c.markAsJoined() - - return 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. diff --git a/cluster_internal_test.go b/cluster_internal_test.go index 2151b07a9..c382de9ec 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -19,20 +19,13 @@ import ( "fmt" "math/rand" "net" - "net/http" - "net/http/httptest" - "net/url" - "os" "reflect" - "strconv" "strings" "testing" "testing/quick" "time" "github.com/davecgh/go-spew/spew" - "github.com/gorilla/mux" - "github.com/pilosa/pilosa/v2/logger" pnet "github.com/pilosa/pilosa/v2/net" "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/test/port" @@ -673,16 +666,16 @@ func TestCluster_Topology(t *testing.T) { nodeinvalid := &topology.Node{ID: "nodeinvalid", URI: uris[3]} t.Run("AddNode", func(t *testing.T) { - err := c1.addNode(node1) + err := c1.addNode(node1.ID) if err != nil { t.Fatal(err) } // add the same host. - err = c1.addNode(node1) + err = c1.addNode(node1.ID) if err != nil { t.Fatal(err) } - err = c1.addNode(node2) + err = c1.addNode(node2.ID) if err != nil { t.Fatal(err) } @@ -1073,91 +1066,6 @@ func TestAE(t *testing.T) { }) } -func TestCluster_confirmNodeDownUp(t *testing.T) { - t.Skip("does a listen on :0, skip for now. TODO(jea) restore this.") - r := mux.NewRouter() - r.HandleFunc("/version", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusOK) - fmt.Fprintln(w, "ignored") - })) - server := httptest.NewServer(r) - // Close the server when test finishes - defer server.Close() - u, err := url.Parse(server.URL) - if err != nil { - t.Error("bad test setup") - } - uri := pnet.URI{} - host, port, _ := net.SplitHostPort(u.Host) - uri.Scheme = u.Scheme - uri.Host = host - iport, err := strconv.ParseUint(port, 0, 16) - if err != nil { - t.Error(err) - } - uri.Port = uint16(iport) - c := newCluster() - c.logger = logger.NewVerboseLogger(os.Stdout) - if c.confirmNodeDown(uri) { - t.Errorf("expected node to be up") - } -} - -func TestCluster_confirmNodeDownTimeout(t *testing.T) { - t.Skip("does a listen on :0, skip for now. TODO(jea) restore this.") - sleep := 50 * time.Millisecond - retries := 5 - if testing.Short() { - t.Skip() - } - r := mux.NewRouter() - r.HandleFunc("/version", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - time.Sleep(sleep * time.Duration(retries)) - fmt.Fprintln(w, "ignored") - })) - server := httptest.NewServer(r) - // Close the server when test finishes - defer server.Close() - u, err := url.Parse(server.URL) - if err != nil { - t.Error("bad test setup") - } - uri := pnet.URI{} - host, port, _ := net.SplitHostPort(u.Host) - uri.Scheme = u.Scheme - uri.Host = host - iport, err := strconv.ParseUint(port, 0, 16) - if err != nil { - t.Error(err) - } - uri.Port = uint16(iport) - c := newCluster() - c.confirmDownSleep = sleep - c.confirmDownRetries = retries - c.logger = logger.NewVerboseLogger(os.Stdout) - if !c.confirmNodeDown(uri) { - t.Errorf("expected node to be down") - } -} - -func TestCluster_confirmNodeDownDown(t *testing.T) { - if testing.Short() { - t.Skip() - } - uri := pnet.URI{} - uri.Scheme = "http" - uri.Host = "DoesntMatter" - uri.Port = 6666 - c := newCluster() - c.confirmDownSleep = 50 * time.Millisecond - c.confirmDownRetries = 5 - c.logger = logger.NewVerboseLogger(os.Stdout) - - if !c.confirmNodeDown(uri) { - t.Errorf("expected node to be down") - } -} - func TestCluster_GetNonPrimaryReplicas(t *testing.T) { c := newCluster() c.ReplicaN = 3 diff --git a/etcd/embed.go b/etcd/embed.go index 7a0bc2b79..e671442cd 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -41,12 +41,12 @@ import ( type Options struct { Name string `toml:"name"` Dir string `toml:"dir"` - LClientURL string `toml:"listen-client-address"` - AClientURL string `toml:"advertise-client-address"` - LPeerURL string `toml:"listen-peer-address"` - APeerURL string `toml:"advertise-peer-address"` - InitCluster string `toml:"initial-cluster"` + LClientURL string `toml:"listen-client-url"` + AClientURL string `toml:"advertise-client-url"` + LPeerURL string `toml:"listen-peer-url"` + APeerURL string `toml:"advertise-peer-url"` ClusterURL string `toml:"cluster-url"` + InitCluster string `toml:"initial-cluster"` ClusterName string `toml:"cluster-name"` HeartbeatTTL int64 `toml:"heartbeat-ttl"` diff --git a/field.go b/field.go index baec3a931..e5936503d 100644 --- a/field.go +++ b/field.go @@ -507,6 +507,14 @@ func (f *Field) unprotectedSaveAvailableShards() error { return nil } +// SetRemoteAvailableShards replaces remoteAvailableShards with the provided +// value. +func (f *Field) SetRemoteAvailableShards(b *roaring.Bitmap) { + f.mu.Lock() + defer f.mu.Unlock() + f.remoteAvailableShards = b +} + // RemoveAvailableShard removes a shard from the bitmap cache. // // NOTE: This can be overridden on the next sync so all nodes should be updated. diff --git a/http/client.go b/http/client.go index c29435521..f175a9da3 100644 --- a/http/client.go +++ b/http/client.go @@ -104,6 +104,39 @@ func (c *InternalClient) maxShardByIndex(ctx context.Context) (map[string]uint64 return rsp.Standard, nil } +// SchemaNode returns all index and field schema information from the specified +// node. +func (c *InternalClient) SchemaNode(ctx context.Context, uri *pnet.URI, views bool) ([]*pilosa.IndexInfo, error) { + span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Schema") + defer span.Finish() + + // TODO: /?views parameter will be ignored, till we implement schemator! + // Execute request against the host. + u := uri.Path(fmt.Sprintf("/schema?views=%v", views)) + + // Build request. + req, err := http.NewRequest("GET", u, nil) + if err != nil { + return nil, errors.Wrap(err, "creating request") + } + + req.Header.Set("User-Agent", "pilosa/"+pilosa.Version) + req.Header.Set("Accept", "application/json") + + // Execute request. + resp, err := c.executeRequest(req.WithContext(ctx)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + var rsp getSchemaResponse + if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil { + return nil, fmt.Errorf("json decode: %s", err) + } + return rsp.Indexes, nil +} + // Schema returns all index and field schema information. func (c *InternalClient) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error) { span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Schema") diff --git a/internal/private.pb.go b/internal/private.pb.go index a22b9c01a..1e87e9265 100644 --- a/internal/private.pb.go +++ b/internal/private.pb.go @@ -5411,7 +5411,10 @@ func (m *IndexMeta) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -5816,7 +5819,10 @@ func (m *FieldOptions) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -5899,7 +5905,10 @@ func (m *ImportResponse) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -6084,7 +6093,10 @@ func (m *BlockDataRequest) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -6287,7 +6299,10 @@ func (m *BlockDataResponse) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -6414,7 +6429,10 @@ func (m *Cache) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -6561,7 +6579,7 @@ func (m *MaxShards) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > postIndex { @@ -6578,7 +6596,10 @@ func (m *MaxShards) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -6712,7 +6733,10 @@ func (m *CreateShardMessage) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -6795,7 +6819,10 @@ func (m *DeleteIndexMessage) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -6933,7 +6960,10 @@ func (m *CreateIndexMessage) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -7103,7 +7133,10 @@ func (m *CreateFieldMessage) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -7218,7 +7251,10 @@ func (m *DeleteFieldMessage) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -7352,7 +7388,10 @@ func (m *DeleteAvailableShardMessage) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -7522,7 +7561,10 @@ func (m *Field) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -7607,7 +7649,10 @@ func (m *Schema) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -7779,7 +7824,10 @@ func (m *Index) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -7913,7 +7961,10 @@ func (m *URI) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -8120,7 +8171,10 @@ func (m *Node) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -8235,7 +8289,10 @@ func (m *NodeStateMessage) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -8341,7 +8398,10 @@ func (m *NodeEventMessage) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -8498,7 +8558,10 @@ func (m *NodeStatus) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -8634,7 +8697,10 @@ func (m *IndexStatus) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -8812,7 +8878,10 @@ func (m *FieldStatus) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -8997,7 +9066,10 @@ func (m *ClusterStatus) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -9150,7 +9222,10 @@ func (m *BSIGroup) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -9297,7 +9372,10 @@ func (m *CreateViewMessage) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -9444,7 +9522,10 @@ func (m *DeleteViewMessage) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -9726,7 +9807,10 @@ func (m *ResizeInstruction) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -9928,7 +10012,10 @@ func (m *ResizeSource) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -10066,7 +10153,10 @@ func (m *TranslationResizeSource) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -10204,7 +10294,10 @@ func (m *ResizeInstructionComplete) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -10319,7 +10412,10 @@ func (m *Topology) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -10370,7 +10466,10 @@ func (m *RecalculateCaches) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -10489,7 +10588,10 @@ func (m *TransactionMessage) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -10686,7 +10788,10 @@ func (m *Transaction) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { @@ -10737,7 +10842,10 @@ func (m *TransactionStats) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPrivate } if (iNdEx + skippy) > l { diff --git a/internal/public.pb.go b/internal/public.pb.go index e406e06ab..5f0c5277d 100644 --- a/internal/public.pb.go +++ b/internal/public.pb.go @@ -6748,7 +6748,10 @@ func (m *Row) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -6833,7 +6836,10 @@ func (m *RowMatrix) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -6956,7 +6962,10 @@ func (m *SignedRow) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -7115,7 +7124,10 @@ func (m *RowIdentifiers) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -7242,7 +7254,10 @@ func (m *IDList) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -7346,7 +7361,10 @@ func (m *ExtractedIDColumn) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -7463,7 +7481,10 @@ func (m *ExtractedIDMatrix) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -7546,7 +7567,10 @@ func (m *KeyList) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -7760,7 +7784,10 @@ func (m *ExtractedTableValue) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -7897,7 +7924,10 @@ func (m *ExtractedTableColumn) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -8012,7 +8042,10 @@ func (m *ExtractedTableField) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -8131,7 +8164,10 @@ func (m *ExtractedTable) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -8252,7 +8288,10 @@ func (m *Pair) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -8371,7 +8410,10 @@ func (m *PairField) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -8488,7 +8530,10 @@ func (m *PairsField) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -8558,7 +8603,10 @@ func (m *Int64) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -8728,7 +8776,10 @@ func (m *FieldRow) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -8851,7 +8902,10 @@ func (m *GroupCount) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -8987,7 +9041,10 @@ func (m *ValCount) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -9076,7 +9133,10 @@ func (m *Decimal) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -9212,7 +9272,10 @@ func (m *ColumnAttrSet) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -9396,7 +9459,10 @@ func (m *Attr) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -9481,7 +9547,10 @@ func (m *AttrMap) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -9774,7 +9843,10 @@ func (m *QueryRequest) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -9925,7 +9997,10 @@ func (m *QueryResponse) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -10538,7 +10613,10 @@ func (m *QueryResult) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -11022,7 +11100,10 @@ func (m *ImportRequest) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -11484,7 +11565,10 @@ func (m *ImportValueRequest) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -11654,7 +11738,10 @@ func (m *AtomicRecord) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -11737,7 +11824,10 @@ func (m *AtomicImportResponse) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -11904,7 +11994,10 @@ func (m *TranslateKeysRequest) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -12031,7 +12124,10 @@ func (m *TranslateKeysResponse) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -12222,7 +12318,10 @@ func (m *TranslateIDsRequest) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -12305,7 +12404,10 @@ func (m *TranslateIDsResponse) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -12422,7 +12524,10 @@ func (m *ImportRoaringRequestView) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -12616,7 +12721,10 @@ func (m *ImportRoaringRequest) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -12877,7 +12985,10 @@ func (m *ImportColumnAttrsRequest) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { @@ -12994,7 +13105,10 @@ func (m *GroupCounts) Unmarshal(dAtA []byte) error { if err != nil { return err } - if (skippy < 0) || (iNdEx+skippy) < 0 { + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) < 0 { return ErrInvalidLengthPublic } if (iNdEx + skippy) > l { diff --git a/server.go b/server.go index 9588e0ca2..74ebb76c2 100644 --- a/server.go +++ b/server.go @@ -604,12 +604,11 @@ func (s *Server) Open() error { s.holder.Activate() // if we joined existing cluster then broadcast "resize on add" message - // TODO - // 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 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") @@ -787,6 +786,7 @@ func (s *Server) receiveMessage(m Message) error { if err := f.AddRemoteAvailableShards(roaring.NewBitmap(obj.Shard)); err != nil { return errors.Wrap(err, "adding remote available shards") } + case *CreateIndexMessage: opt := obj.Meta idx, err := s.holder.CreateIndex(obj.Index, *opt) @@ -796,10 +796,12 @@ func (s *Server) receiveMessage(m Message) error { idx.mu.Lock() idx.createdAt = obj.CreatedAt idx.mu.Unlock() + case *DeleteIndexMessage: if err := s.holder.DeleteIndex(obj.Index); err != nil { return err } + case *CreateFieldMessage: idx := s.holder.Index(obj.Index) if idx == nil { @@ -813,16 +815,19 @@ func (s *Server) receiveMessage(m Message) error { fld.mu.Lock() fld.createdAt = obj.CreatedAt fld.mu.Unlock() + case *DeleteFieldMessage: idx := s.holder.Index(obj.Index) if err := idx.DeleteField(obj.Field); err != nil { return err } + case *DeleteAvailableShardMessage: f := s.holder.Field(obj.Index, obj.Field) if err := f.RemoveAvailableShard(obj.ShardID); err != nil { return err } + case *CreateViewMessage: f := s.holder.Field(obj.Index, obj.Field) if f == nil { @@ -831,6 +836,7 @@ func (s *Server) receiveMessage(m Message) error { if _, _, err := f.createViewIfNotExistsBase(obj.View); err != nil { return err } + case *DeleteViewMessage: f := s.holder.Field(obj.Index, obj.Field) if f == nil { @@ -840,31 +846,41 @@ func (s *Server) receiveMessage(m Message) error { if err != nil { return err } - case *ClusterStatus: - err := s.cluster.mergeClusterStatus(obj) - if err != nil { - return err - } - if !s.IsPrimary() { - if obj.Schema != nil { - s.holder.applyCreatedAt(obj.Schema.Indexes) + + 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(obj) + err := s.cluster.followResizeInstruction(context.Background(), obj) if err != nil { return err } - case *ResizeInstructionComplete: - err := s.cluster.markResizeInstructionComplete(obj) + + case *ResizeAbortMessage: + err := s.cluster.resizeAbort() if err != nil { return err } + case *RecalculateCaches: s.holder.recalculateCaches() + case *NodeStatus: s.handleRemoteStatus(obj) + case *TransactionMessage: err := s.handleTransactionMessage(obj) if err != nil { diff --git a/utils_internal_test.go b/utils_internal_test.go index 823afb2d4..26414ad11 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -264,7 +264,7 @@ func (t *ClusterCluster) addCluster(i int, saveTopology bool) (*cluster, error) // add nodes if saveTopology { for _, n := range t.common.Nodes { - if err := c.addNode(n); err != nil { + if err := c.addNode(n.ID); err != nil { return nil, err } } @@ -329,15 +329,6 @@ type bcast struct { func (b bcast) SendSync(m Message) error { switch obj := m.(type) { case *ClusterStatus: - // Apply the send message to all nodes (except the coordinator). - for _, c := range b.t.Clusters { - if c != b.c { - err := c.mergeClusterStatus(obj) - if err != nil { - return err - } - } - } b.t.mu.RLock() if obj.State == string(ClusterStateNormal) && b.t.resizing { close(b.t.resizeDone) @@ -367,21 +358,7 @@ func (b bcast) SendTo(to *topology.Node, m Message) error { if err != nil { return err } - case *ResizeInstructionComplete: - coord := b.t.clusterByID(to.ID) - // this used to be async, but that prevented us from checking - // its error status... - return coord.markResizeInstructionComplete(obj) case *ClusterStatus: - // Apply the send message to the node. - for _, c := range b.t.Clusters { - if c.Node.ID == to.ID { - err := c.mergeClusterStatus(obj) - if err != nil { - return err - } - } - } b.t.mu.RLock() if obj.State == string(ClusterStateNormal) && b.t.resizing { close(b.t.resizeDone) From bdbffe8d9626c447d616d1e632d1308b9bea3897 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Fri, 5 Feb 2021 12:48:22 +0100 Subject: [PATCH 2/2] Update cluster_internal_test.go --- cluster_internal_test.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/cluster_internal_test.go b/cluster_internal_test.go index c382de9ec..68c44a7fb 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -647,6 +647,8 @@ func TestCluster_Coordinator(t *testing.T) { } func TestCluster_Topology(t *testing.T) { + t.Skip("these tests don't really apply anymore; they were meant to tests the cluster and adding topology nodes.") + c1 := NewTestCluster(t, 1) // automatically creates Node{ID: "node0"} const urisCount = 4