scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 09:05:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #13 from travisturner/disco-config-noder-primaryid

Browse source
This commit is contained in:
Antonio Navarro Perez 2021-02-04 10:38:15 +01:00 • committed by GitHub
commit add5065144
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 99 additions and 65 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

52
cluster.go
View file

@ -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)
}

34
etcd/embed.go
View file

@ -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
}

5
holder.go
View file

@ -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 {

27
server.go
View file

@ -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 {

4
server/handler_test.go
View file

@ -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

30
topology/noder.go
View file

@ -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

12
topology/snapshot.go
View file

@ -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
}