diff --git a/cluster.go b/cluster.go index c6b0c75e5..c4a295072 100644 --- a/cluster.go +++ b/cluster.go @@ -50,9 +50,8 @@ const ( ClusterStateResizing = disco.ClusterStateResizing ClusterStateDown = disco.ClusterStateDown - // NodeState represents the state of a node during startup. - nodeStateReady = "READY" - nodeStateDown = "DOWN" + // nodeStateDown represents the state of a node which is unavailable. + nodeStateDown = "DOWN" // resizeJob states. resizeJobStateRunning = "RUNNING" @@ -213,24 +212,6 @@ func (c *cluster) unprotectedIsCoordinator() bool { return snap.PrimaryFieldTranslationNode().ID == c.Node.ID } -// unprotectedSendSync is used in place of c.broadcaster.SendSync (which is -// Server.SendSync) because Server.SendSync needs to obtain a cluster lock to -// get the list of nodes. TODO: the reference loop from -// Server->cluster->broadcaster(Server) will likely continue to cause confusion -// and should be refactored. -func (c *cluster) unprotectedSendSync(m Message) error { - var eg errgroup.Group - for _, node := range c.noder.Nodes() { - node := node - // Don't send to myself. - if node.ID == c.Node.ID { - continue - } - eg.Go(func() error { return c.broadcaster.SendTo(node, m) }) - } - return eg.Wait() -} - // addNode adds a node to the Cluster and updates and saves the // new topology. unprotected. func (c *cluster) addNode(node *topology.Node) error { @@ -818,20 +799,6 @@ func (c *cluster) partitionNodes(partitionID int) []*topology.Node { return nodes } -func (c *cluster) primaryPartitionNode(partition int) *topology.Node { - c.mu.RLock() - defer c.mu.RUnlock() - return c.unprotectedPrimaryPartitionNode(partition) -} - -// unprotectedPrimaryPartition returns tprimary node of partition. -func (c *cluster) unprotectedPrimaryPartitionNode(partition int) *topology.Node { - if nodes := c.partitionNodes(partition); len(nodes) > 0 { - return nodes[0] - } - return nil -} - func (t *Topology) IsPrimary(nodeID string, partitionID int) bool { primary := t.PrimaryNodeIndex(partitionID) return nodeID == t.nodeIDs[primary] @@ -1651,6 +1618,11 @@ func (t *Topology) Nodes() []*topology.Node { return nodes } +// PrimaryNodeID implements the Noder interface. +func (t *Topology) PrimaryNodeID(topology.Hasher) string { + return "" +} + // SetNodes implements the Noder interface. func (t *Topology) SetNodes(nodes []*topology.Node) {} @@ -2466,11 +2438,14 @@ func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...s // TODO: use local replicas to short-circuit network traffic + // Create a snapshot of the cluster to use for node/partition calculations. + snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN) + // Group keys by node. keysByNode := make(map[*topology.Node][]string) for partitionID, keys := range keysByPartition { // Find the primary node for this partition. - primary := c.primaryPartitionNode(partitionID) + primary := snap.PrimaryPartitionNode(partitionID) if primary == nil { return nil, errors.Errorf("translating index(%s) keys(%v) on partition(%d) - cannot find primary node", indexName, keys, partitionID) } @@ -2572,12 +2547,15 @@ func (c *cluster) createIndexKeys(ctx context.Context, indexName string, keys .. // TODO: use local replicas to short-circuit network traffic + // Create a snapshot of the cluster to use for node/partition calculations. + snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN) + // Group keys by node. // Delete remote keys from the by-partition map so that it can be used for local translation. keysByNode := make(map[*topology.Node][]string) for partitionID, keys := range keysByPartition { // Find the primary node for this partition. - primary := c.primaryPartitionNode(partitionID) + primary := snap.PrimaryPartitionNode(partitionID) if primary == nil { return nil, errors.Errorf("translating index(%s) keys(%v) on partition(%d) - cannot find primary node", indexName, keys, partitionID) } diff --git a/etcd/embed.go b/etcd/embed.go index dcd261b60..7a0bc2b79 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -1044,16 +1044,15 @@ func (e *Etcd) RemoveShard(ctx context.Context, index, field string, shard uint6 return nil } -// Nodes implements the Noder interface. -func (n *Etcd) Nodes() []*topology.Node { - // If we have looked up nodes within a certain time, then we're going to - // use the cached value for now. This is temporary and will be addressed - // correctly in #1133. - peers := n.Peers() +// Nodes implements the Noder interface. It returns the sorted list of nodes +// based on the etcd peers. +func (e *Etcd) Nodes() []*topology.Node { + peers := e.Peers() nodes := make([]*topology.Node, len(peers)) for i, peer := range peers { node := &topology.Node{} - if meta, err := n.Metadata(context.Background(), peer.ID); err != nil { + + if meta, err := e.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") @@ -1070,16 +1069,31 @@ func (n *Etcd) Nodes() []*topology.Node { return nodes } +// PrimaryNodeID implements the Noder interface. +func (e *Etcd) PrimaryNodeID(hasher topology.Hasher) string { + return topology.PrimaryNodeID(e.NodeIDs(), hasher) +} + +// NodeIDs returns the list of node IDs in the etcd cluster. +func (e *Etcd) NodeIDs() []string { + peers := e.Peers() + ids := make([]string, len(peers)) + for i, peer := range peers { + ids[i] = peer.ID + } + return ids +} + // SetNodes implements the Noder interface as NOP // (because we can't force to set nodes for etcd). -func (n *Etcd) SetNodes(nodes []*topology.Node) {} +func (e *Etcd) SetNodes(nodes []*topology.Node) {} // AppendNode implements the Noder interface as NOP // (because resizer is responsible for adding new nodes). -func (n *Etcd) AppendNode(node *topology.Node) {} +func (e *Etcd) AppendNode(node *topology.Node) {} // RemoveNode implements the Noder interface as NOP // (because resizer is responsible for removing existing nodes) -func (n *Etcd) RemoveNode(nodeID string) bool { +func (e *Etcd) RemoveNode(nodeID string) bool { return false } diff --git a/holder.go b/holder.go index 48ca449f1..d697b3ab4 100644 --- a/holder.go +++ b/holder.go @@ -1823,6 +1823,11 @@ type holderCleaner struct { Closing <-chan struct{} } +// TODO: this is here to satisfy the linter since holderCleaner was removed from +// the gossip implementation of removeNode. But presumably we will use it once +// we have ported over the etcd implementation. +var _ holderCleaner + // IsClosing returns true if the cleaner has been marked to close. func (c *holderCleaner) IsClosing() bool { select { diff --git a/server.go b/server.go index dbd66bf19..71262c829 100644 --- a/server.go +++ b/server.go @@ -422,7 +422,7 @@ func NewServer(opts ...ServerOption) (*Server, error) { stator: disco.NopStator, metadator: disco.NopMetadator, resizer: disco.NopResizer, - noder: topology.NewLocalNoder(nil), + noder: topology.NewEmptyLocalNoder(), sharder: disco.NopSharder, confirmDownRetries: defaultConfirmDownRetries, @@ -565,14 +565,12 @@ func (s *Server) Open() error { // Set node ID. s.nodeID = s.disCo.ID() - // TODO we cannot set IsPrimary here because we don't have all the needed info node := &topology.Node{ - ID: s.nodeID, - URI: s.uri, - GRPCURI: s.grpcURI, - State: nodeStateDown, - // TODO set primary - IsPrimary: false, + ID: s.nodeID, + URI: s.uri, + GRPCURI: s.grpcURI, + State: nodeStateDown, + IsPrimary: s.IsPrimary(), } // Set metadata for this node. @@ -1036,10 +1034,7 @@ func (s *Server) mergeRemoteStatus(ns *NodeStatus) error { // IsPrimary returns if this node is primary right now or not. func (s *Server) IsPrimary() bool { - if primary := s.cluster.PrimaryReplicaNode(); primary != nil { - return s.nodeID == primary.ID - } - return false + return s.nodeID == s.noder.PrimaryNodeID(s.cluster.Hasher) } // monitorDiagnostics periodically polls the Pilosa Indexes for cluster info. @@ -1141,7 +1136,7 @@ func (s *Server) monitorRuntime() { } func (srv *Server) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool, remote bool) (*Transaction, error) { - snap := topology.NewClusterSnapshot(srv.cluster, srv.cluster.Hasher, srv.cluster.partitionN) + snap := topology.NewClusterSnapshot(srv.cluster.noder, srv.cluster.Hasher, srv.cluster.partitionN) node := srv.node() if !remote && !snap.IsPrimaryFieldTranslationNode(node.ID) && len(srv.cluster.Nodes()) > 1 { return nil, ErrNodeNotCoordinator @@ -1188,7 +1183,7 @@ func (srv *Server) StartTransaction(ctx context.Context, id string, timeout time } func (srv *Server) FinishTransaction(ctx context.Context, id string, remote bool) (*Transaction, error) { - snap := topology.NewClusterSnapshot(srv.cluster, srv.cluster.Hasher, srv.cluster.partitionN) + snap := topology.NewClusterSnapshot(srv.cluster.noder, srv.cluster.Hasher, srv.cluster.partitionN) node := srv.node() if !remote && !snap.IsPrimaryFieldTranslationNode(node.ID) && len(srv.cluster.Nodes()) > 1 { return nil, ErrNodeNotCoordinator @@ -1218,7 +1213,7 @@ func (srv *Server) FinishTransaction(ctx context.Context, id string, remote bool } func (srv *Server) Transactions(ctx context.Context) (map[string]*Transaction, error) { - snap := topology.NewClusterSnapshot(srv.cluster, srv.cluster.Hasher, srv.cluster.partitionN) + snap := topology.NewClusterSnapshot(srv.cluster.noder, srv.cluster.Hasher, srv.cluster.partitionN) node := srv.node() if !snap.IsPrimaryFieldTranslationNode(node.ID) && len(srv.cluster.Nodes()) > 1 { return nil, ErrNodeNotCoordinator @@ -1228,7 +1223,7 @@ func (srv *Server) Transactions(ctx context.Context) (map[string]*Transaction, e } func (srv *Server) GetTransaction(ctx context.Context, id string, remote bool) (*Transaction, error) { - snap := topology.NewClusterSnapshot(srv.cluster, srv.cluster.Hasher, srv.cluster.partitionN) + snap := topology.NewClusterSnapshot(srv.cluster.noder, srv.cluster.Hasher, srv.cluster.partitionN) node := srv.node() if !remote && !snap.IsPrimaryFieldTranslationNode(node.ID) && len(srv.cluster.Nodes()) > 1 { diff --git a/server/handler_test.go b/server/handler_test.go index 8cc8fb095..2f7486534 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -1060,8 +1060,8 @@ func TestHandler_Endpoints(t *testing.T) { } body := mustJSONDecodeSlice(t, w.Body) bmap := body[0].(map[string]interface{}) - if bmap["isPrimary"] != false { - t.Fatalf("expected false primary, got: %+v", bmap) + if bmap["isPrimary"] != true { + t.Fatalf("expected true primary, got: %+v", bmap) } // invalid argument should return BadRequest diff --git a/topology/noder.go b/topology/noder.go index f0499b997..63067e199 100644 --- a/topology/noder.go +++ b/topology/noder.go @@ -22,6 +22,7 @@ import ( // nodes in a cluster can be maintained outside of the cluster struct. type Noder interface { Nodes() []*Node // Remember: this has to be sorted correctly!! + PrimaryNodeID(hasher Hasher) string SetNodes([]*Node) AppendNode(*Node) RemoveNode(nodeID string) bool @@ -46,11 +47,40 @@ func NewEmptyLocalNoder() *localNoder { return &localNoder{} } +// NewIDNoder is a helper function for wrapping an existing slice of Node IDs +// with something which implements Noder. +func NewIDNoder(ids []string) *localNoder { + nodes := make([]*Node, len(ids)) + for i, id := range ids { + node := &Node{ + ID: id, + } + nodes[i] = node + } + + // Nodes must be sorted. + sort.Sort(ByID(nodes)) + + return &localNoder{ + nodes: nodes, + } +} + // Nodes implements the Noder interface. func (n *localNoder) Nodes() []*Node { return n.nodes } +// PrimaryNodeID implements the Noder interface. +func (n *localNoder) PrimaryNodeID(hasher Hasher) string { + snap := NewClusterSnapshot(NewLocalNoder(n.nodes), hasher, 1) + primaryNode := snap.PrimaryFieldTranslationNode() + if primaryNode == nil { + return "" + } + return primaryNode.ID +} + // SetNodes implements the Noder interface. func (n *localNoder) SetNodes(nodes []*Node) { n.nodes = nodes diff --git a/topology/snapshot.go b/topology/snapshot.go index decccccfc..fc1a2d83f 100644 --- a/topology/snapshot.go +++ b/topology/snapshot.go @@ -280,3 +280,15 @@ func NodePositionByID(nodes []*Node, nodeID string) int { } return -1 } + +// PrimaryNodeID returns the ID of the primary node, given a list of node IDs +// and a hasher. The order of the node IDs provided does not matter because this +// function will re-order them in a deterministic way. +func PrimaryNodeID(nodeIDs []string, hasher Hasher) string { + snap := NewClusterSnapshot(NewIDNoder(nodeIDs), hasher, 1) + primaryNode := snap.PrimaryFieldTranslationNode() + if primaryNode == nil { + return "" + } + return primaryNode.ID +}