diff --git a/api.go b/api.go index 417b45dfc..cf5431198 100644 --- a/api.go +++ b/api.go @@ -605,8 +605,8 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) // Validate that this handler owns the shard. - if !snap.OwnsShard(api.Node().ID, indexName, shard) { - api.server.logger.Printf("node %s does not own shard %d of index %s", api.Node().ID, shard, indexName) + if !snap.OwnsShard(api.NodeID(), indexName, shard) { + api.server.logger.Printf("node %s does not own shard %d of index %s", api.NodeID(), shard, indexName) return ErrClusterDoesNotOwnShard } @@ -807,6 +807,12 @@ func (api *API) Node() *topology.Node { return api.server.node() } +// NodeID gets the ID alone, so it doesn't have to do a complete lookup +// of the node, searching by its ID, to return the ID it searched for. +func (api *API) NodeID() string { + return api.server.nodeID +} + // PrimaryNode returns the coordinator node for the cluster. func (api *API) PrimaryNode() *topology.Node { // Create a snapshot of the cluster to use for node/partition calculations. @@ -1735,8 +1741,8 @@ func (api *API) validateShardOwnership(indexName string, shard uint64) error { // Create a snapshot of the cluster to use for node/partition calculations. snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) // Validate that this handler owns the shard. - if !snap.OwnsShard(api.Node().ID, indexName, shard) { - api.server.logger.Printf("node %s does not own shard %d of index %s", api.Node().ID, shard, indexName) + if !snap.OwnsShard(api.NodeID(), indexName, shard) { + api.server.logger.Printf("node %s does not own shard %d of index %s", api.NodeID(), shard, indexName) return ErrClusterDoesNotOwnShard } return nil @@ -2026,7 +2032,7 @@ func (api *API) PrimaryReplicaNodeURL() url.URL { // Create a snapshot of the cluster to use for node/partition calculations. snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) - node := snap.PrimaryReplicaNode(api.Node().ID) + node := snap.PrimaryReplicaNode(api.NodeID()) if node == nil { return url.URL{} } @@ -2137,7 +2143,7 @@ func (api *API) ReserveIDs(key IDAllocKey, session [32]byte, offset uint64, coun // Create a snapshot of the cluster to use for node/partition calculations. snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) - if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) { + if !snap.IsPrimaryFieldTranslationNode(api.NodeID()) { return api.holder.ida.reserve(key, session, offset, count) } @@ -2152,7 +2158,7 @@ func (api *API) CommitIDs(key IDAllocKey, session [32]byte, count uint64) error // Create a snapshot of the cluster to use for node/partition calculations. snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) - if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) { + if !snap.IsPrimaryFieldTranslationNode(api.NodeID()) { return api.holder.ida.commit(key, session, count) } @@ -2167,7 +2173,7 @@ func (api *API) ResetIDAlloc(index string) error { // Create a snapshot of the cluster to use for node/partition calculations. snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) - if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) { + if !snap.IsPrimaryFieldTranslationNode(api.NodeID()) { return api.holder.ida.reset(index) } diff --git a/cluster.go b/cluster.go index 715bcea86..37ba01b5f 100644 --- a/cluster.go +++ b/cluster.go @@ -709,25 +709,29 @@ func (c *cluster) addNodeBasicSorted(node *topology.Node) bool { // concurrent use, result may be modified. func (c *cluster) Nodes() []*topology.Node { nodes := c.noder.Nodes() + // duplicate the nodes since we're going to be altering them + copiedNodes := make([]topology.Node, len(nodes)) + result := make([]*topology.Node, len(nodes)) // Create a snapshot of the cluster to use for node/partition calculations. - snap := topology.NewClusterSnapshot(topology.NewLocalNoder(nodes), c.Hasher, c.ReplicaN) - primaryNode := snap.PrimaryFieldTranslationNode() + primary := topology.PrimaryNode(nodes, c.Hasher) // Set node states and IsPrimary. - for _, node := range nodes { - node.IsPrimary = node.ID == primaryNode.ID - + for i, node := range nodes { + copiedNodes[i] = *node + result[i] = &copiedNodes[i] + if node == primary { + copiedNodes[i].IsPrimary = true + } s, err := c.stator.NodeState(context.Background(), node.ID) if err != nil { // TODO should we delete this? - node.State = string(disco.NodeStateUnknown) + copiedNodes[i].State = string(disco.NodeStateUnknown) continue } - node.State = string(s) + copiedNodes[i].State = string(s) } - - return nodes + return result } // removeNodeBasicSorted removes a node from the cluster, maintaining the sort diff --git a/etcd/embed.go b/etcd/embed.go index 09bd82608..dd1062285 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -1127,9 +1127,11 @@ func (e *Etcd) RemoveShard(ctx context.Context, index, field string, shard uint6 // based on the etcd peers. func (e *Etcd) Nodes() []*topology.Node { peers := e.Peers() + // For N>1, this might actually reduce GC load. Maybe. + nodeData := make([]topology.Node, len(peers)) nodes := make([]*topology.Node, len(peers)) for i, peer := range peers { - node := &topology.Node{} + node := &nodeData[i] if meta, err := e.Metadata(context.Background(), peer.ID); err != nil { log.Println(err, "getting metadata") // TODO: handle this with a logger diff --git a/topology/hasher.go b/topology/hasher.go index a5c3f5964..41cd36b95 100644 --- a/topology/hasher.go +++ b/topology/hasher.go @@ -39,3 +39,13 @@ func (h *Jmphasher) Hash(key uint64, n int) int { func (h *Jmphasher) Name() string { return "jump-hash" } + +// PrimaryNode yields the node that would be selected as the primary from +// a list, for a given ID. It assumes the list is already in the +// expected order, as from Noder.Nodes(). +func PrimaryNode(nodes []*Node, hasher Hasher) *Node { + if len(nodes) == 0 { + return nil + } + return nodes[hasher.Hash(0, len(nodes))] +} diff --git a/topology/snapshot.go b/topology/snapshot.go index b7d865335..8ad6305c6 100644 --- a/topology/snapshot.go +++ b/topology/snapshot.go @@ -107,8 +107,14 @@ func (c *ClusterSnapshot) ShardNodes(index string, shard uint64) []*Node { } // OwnsShard returns true if a host owns a fragment. -func (c *ClusterSnapshot) OwnsShard(nodeID string, index string, shard uint64) bool { - return Nodes(c.ShardNodes(index, shard)).ContainsID(nodeID) +func (c *ClusterSnapshot) OwnsShard(nodeID string, index string, shard uint64) (ret bool) { + idx := c.Hasher.Hash(uint64(c.ShardToShardPartition(index, shard)), len(c.Nodes)) + for i := 0; i < c.ReplicaN; i++ { + if c.Nodes[(idx+i)%len(c.Nodes)].ID == nodeID { + return true + } + } + return false } // KeyNodes returns a list of nodes that own a key. @@ -147,11 +153,14 @@ func (c *ClusterSnapshot) IsPrimaryFieldTranslationNode(nodeID string) bool { } // PrimaryPartitionNode returns the primary node of the given partition. -func (c *ClusterSnapshot) PrimaryPartitionNode(partition int) *Node { - if nodes := c.PartitionNodes(partition); len(nodes) > 0 { - return nodes[0] +func (c *ClusterSnapshot) PrimaryPartitionNode(partitionID int) *Node { + // Determine primary owner node. + nodeIndex := c.PrimaryNodeIndex(partitionID) + if nodeIndex < 0 { + // no nodes anyway + return nil } - return nil + return c.Nodes[nodeIndex] } // IsPrimary returns true if the given node is the primary for the given