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/cache.go b/etcd/cache.go index 47543779a..dc7b8a539 100644 --- a/etcd/cache.go +++ b/etcd/cache.go @@ -16,9 +16,6 @@ package etcd import ( "context" - "encoding/json" - "log" - "sort" "sync" "time" @@ -35,8 +32,12 @@ type EtcdWithCache struct { peerMetadataMu sync.RWMutex peerMetadata map[string][]byte - stateMu sync.Mutex + stateMu sync.Mutex // cluster state cache updates + peersMu sync.Mutex // peer-list cache updates + nodes []*topology.Node // unmarshalled Node data + nodesTTL int // seconds + nodesLastRequest time.Time // last time requested nodeStates map[string]nodeState nodeStateTTL int // seconds nodeStateFrequency int // max requests per second allowed before using the cache @@ -63,6 +64,7 @@ func NewEtcdWithCache(opt Options, replicas int) *EtcdWithCache { nodeStateFrequency: 1, clusterStateTTL: 6, clusterStateFrequency: 1, + nodesTTL: 6, peerMetadata: make(map[string][]byte), nodeStates: make(map[string]nodeState), @@ -151,27 +153,17 @@ func (c *EtcdWithCache) NodeState(ctx context.Context, peerID string) (disco.Nod return ns.val, nil } -// Nodes implements the Noder interface. +// Nodes caches the result of the underlying implementation's node list. func (c *EtcdWithCache) Nodes() []*topology.Node { - peers := c.Peers() - nodes := make([]*topology.Node, len(peers)) - for i, peer := range peers { - node := &topology.Node{} - if meta, err := c.Metadata(context.Background(), peer.ID); err != nil { - log.Println(err, "getting metadata") // TODO: handle this with a logger - } else if err := json.Unmarshal(meta, node); err != nil { - log.Println(err, "unmarshaling json metadata") - } + c.peersMu.Lock() + defer c.peersMu.Unlock() - node.ID = peer.ID - - nodes[i] = node + now := time.Now() + if now.Sub(c.nodesLastRequest) > (time.Duration(c.nodesTTL) * time.Second) { + c.nodes = c.Etcd.Nodes() + c.nodesLastRequest = now } - - // Nodes must be sorted. - sort.Sort(topology.ByID(nodes)) - - return nodes + return c.nodes } // SetNodes implements the Noder interface as NOP 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/field.go b/field.go index 34a6aa72e..0bfdede5d 100644 --- a/field.go +++ b/field.go @@ -1187,6 +1187,12 @@ func (f *Field) createViewIfNotExistsBase(cvm *CreateViewMessage) (*view, bool, f.mu.Lock() defer f.mu.Unlock() + // If we already have this view, we can probably assume etcd already + // has it. + if view := f.viewMap[cvm.View]; view != nil { + return view, false, nil + } + // Create the view in etcd as the system of record. // Don't persist views related to the existence field. if f.name != existenceFieldName { @@ -1195,9 +1201,6 @@ func (f *Field) createViewIfNotExistsBase(cvm *CreateViewMessage) (*view, bool, } } - if view := f.viewMap[cvm.View]; view != nil { - return view, false, nil - } view := f.newView(f.viewPath(cvm.View), cvm.View) if err := view.openEmpty(); err != nil { diff --git a/holder.go b/holder.go index 91d0e1211..c6c9977c7 100644 --- a/holder.go +++ b/holder.go @@ -1947,7 +1947,7 @@ func (s *holderSyncer) initializeIndexTranslateReplication(snap *topology.Cluste // initializeFieldTranslateReplication connects the coordinator to stream field data. func (s *holderSyncer) initializeFieldTranslateReplication(snap *topology.ClusterSnapshot) error { // Skip if coordinator. - if !snap.IsPrimaryFieldTranslationNode(s.Cluster.Node.ID) { + if snap.IsPrimaryFieldTranslationNode(s.Cluster.Node.ID) { return nil } 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