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

Merge pull request #1586 from pilosa/1492-ae-and-resize

Browse source 
Fix - prevent anti-entropy and cluster resize from running simultaneously
This commit is contained in:
Matthew Jaffee 2018-08-20 11:10:36 -05:00 • committed by GitHub
commit e426c4215e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 213 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
api.go
View file

@ -428,7 +428,7 @@ func (api *API) FragmentBlocks(_ context.Context, indexName, fieldName, viewName
// Hosts returns a list of the hosts in the cluster including their ID,
// URL, and which is the coordinator.
func (api *API) Hosts(_ context.Context) []*Node {
return api.cluster.Nodes
return api.cluster.Nodes()
}
// Node gets the ID, URI and coordinator status for this particular node.

101
cluster.go
View file

@ -169,7 +169,7 @@ type nodeAction struct {
type cluster struct { // nolint: maligned
id string
Node *Node
Nodes []*Node
nodes []*Node
// Hashing algorithm used to assign partitions to nodes.
Hasher Hasher
@ -204,6 +204,8 @@ type cluster struct { // nolint: maligned
joining chan struct{}
joined bool
abortAntiEntropyCh chan struct{}
mu sync.RWMutex
jobs map[int64]*resizeJob
currentJob *resizeJob
@ -235,6 +237,33 @@ func newCluster() *cluster {
}
}
// initializeAntiEntropy is called by the anti entropy routine when it starts.
// If the AE channel is created without a routine reading from it, cluster will
// block indefinitely when calling abortAntiEntropy().
func (c *cluster) initializeAntiEntropy() {
c.mu.Lock()
c.abortAntiEntropyCh = make(chan struct{})
c.mu.Unlock()
}
// abortAntiEntropyQ checks whether the cluster wants to abort the anti entropy
// process (so that it can resize). It does not block.
func (c *cluster) abortAntiEntropyQ() bool {
select {
case <-c.abortAntiEntropyCh:
return true
default:
return false
}
}
// abortAntiEntropy blocks until the anti-entropy routine calls abortAntiEntropyQ
func (c *cluster) abortAntiEntropy() {
if c.abortAntiEntropyCh != nil {
c.abortAntiEntropyCh <- struct{}{}
}
}
func (c *cluster) coordinatorNode() *Node {
c.mu.RLock()
defer c.mu.RUnlock()
@ -302,7 +331,7 @@ func (c *cluster) unprotectedUpdateCoordinator(n *Node) bool {
c.Coordinator = n.ID
changed = true
}
for _, node := range c.Nodes {
for _, node := range c.nodes {
if node.ID == n.ID {
node.IsCoordinator = true
} else {
@ -359,7 +388,7 @@ func (c *cluster) removeNode(nodeID string) error {
// nodeIDs returns the list of IDs in the cluster.
func (c *cluster) nodeIDs() []string {
return Nodes(c.Nodes).IDs()
return Nodes(c.nodes).IDs()
}
func (c *cluster) unprotectedSetID(id string) {
@ -405,6 +434,10 @@ func (c *cluster) unprotectedSetState(state string) {
c.state = state
if state == ClusterStateResizing {
c.abortAntiEntropy()
}
// TODO: consider NOT running cleanup on an active node that has
// been removed.
// It's safe to do a cleanup after state changes back to normal.
@ -484,7 +517,7 @@ func (c *cluster) unprotectedStatus() *ClusterStatus {
return &ClusterStatus{
ClusterID: c.id,
State: c.state,
Nodes: c.Nodes,
Nodes: c.nodes,
}
}
@ -496,7 +529,7 @@ func (c *cluster) nodeByID(id string) *Node {
// unprotectedNodeByID returns a node reference by ID.
func (c *cluster) unprotectedNodeByID(id string) *Node {
for _, n := range c.Nodes {
for _, n := range c.nodes {
if n.ID == id {
return n
}
@ -517,7 +550,7 @@ func (c *cluster) topologyContainsNode(id string) bool {
// nodePositionByID returns the position of the node in slice c.Nodes.
func (c *cluster) nodePositionByID(nodeID string) int {
for i, n := range c.Nodes {
for i, n := range c.nodes {
if n.ID == nodeID {
return i
}
@ -533,14 +566,24 @@ func (c *cluster) addNodeBasicSorted(node *Node) bool {
return false
}
c.Nodes = append(c.Nodes, node)
c.nodes = append(c.nodes, node)
// All hosts must be merged in the same order on all nodes in the cluster.
sort.Sort(byID(c.Nodes))
sort.Sort(byID(c.nodes))
return true
}
// Nodes returns a copy of the slice of nodes in the cluster. Safe for
// concurrent use, result may be modified.
func (c *cluster) Nodes() []*Node {
c.mu.Lock()
defer c.mu.Unlock()
ret := make([]*Node, len(c.nodes))
copy(ret, c.nodes)
return ret
}
// removeNodeBasicSorted removes a node from the cluster, maintaining the sort
// order. Returns true if the node was removed. unprotected.
func (c *cluster) removeNodeBasicSorted(nodeID string) bool {
@ -549,9 +592,9 @@ func (c *cluster) removeNodeBasicSorted(nodeID string) bool {
return false
}
copy(c.Nodes[i:], c.Nodes[i+1:])
c.Nodes[len(c.Nodes)-1] = nil
c.Nodes = c.Nodes[:len(c.Nodes)-1]
copy(c.nodes[i:], c.nodes[i+1:])
c.nodes[len(c.nodes)-1] = nil
c.nodes = c.nodes[:len(c.nodes)-1]
return true
}
@ -625,8 +668,8 @@ func (c *cluster) fragCombos(idx string, maxShard uint64, fieldViews viewsByFiel
// added or removed. An error is returned for any case other than where
// exactly one node is added or removed. unprotected.
func (c *cluster) diff(other *cluster) (action string, nodeID string, err error) {
lenFrom := len(c.Nodes)
lenTo := len(other.Nodes)
lenFrom := len(c.nodes)
lenTo := len(other.nodes)
// Determine if a node is being added or removed.
if lenFrom == lenTo {
return "", "", errors.New("clusters are the same size")
@ -638,7 +681,7 @@ func (c *cluster) diff(other *cluster) (action string, nodeID string, err error)
}
action = resizeJobActionAdd
// Determine the node ID that is being added.
for _, n := range other.Nodes {
for _, n := range other.nodes {
if c.unprotectedNodeByID(n.ID) == nil {
nodeID = n.ID
break
@ -651,7 +694,7 @@ func (c *cluster) diff(other *cluster) (action string, nodeID string, err error)
}
action = resizeJobActionRemove
// Determine the node ID that is being removed.
for _, n := range c.Nodes {
for _, n := range c.nodes {
if other.unprotectedNodeByID(n.ID) == nil {
nodeID = n.ID
break
@ -673,7 +716,7 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*ResizeSour
}
// Initialize the map with all the nodes in `to`.
for _, n := range to.Nodes {
for _, n := range to.nodes {
m[n.ID] = nil
}
@ -686,7 +729,7 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*ResizeSour
srcCluster := c
if action == resizeJobActionAdd && c.ReplicaN > 1 {
srcCluster = newCluster()
srcCluster.Nodes = Nodes(c.Nodes).Clone()
srcCluster.nodes = Nodes(c.nodes).Clone()
srcCluster.Hasher = c.Hasher
srcCluster.partitionN = c.partitionN
srcCluster.ReplicaN = 1
@ -776,19 +819,19 @@ func (c *cluster) partitionNodes(partitionID int) []*Node {
// Default replica count to between one and the number of nodes.
// The replica count can be zero if there are no nodes.
replicaN := c.ReplicaN
if replicaN > len(c.Nodes) {
replicaN = len(c.Nodes)
if replicaN > len(c.nodes) {
replicaN = len(c.nodes)
} else if replicaN == 0 {
replicaN = 1
}
// Determine primary owner node.
nodeIndex := c.Hasher.Hash(uint64(partitionID), len(c.Nodes))
nodeIndex := c.Hasher.Hash(uint64(partitionID), len(c.nodes))
// Collect nodes around the ring.
nodes := make([]*Node, replicaN)
for i := 0; i < replicaN; i++ {
nodes[i] = c.Nodes[(nodeIndex+i)%len(c.Nodes)]
nodes[i] = c.nodes[(nodeIndex+i)%len(c.nodes)]
}
return nodes
@ -1093,12 +1136,12 @@ func (c *cluster) unprotectedGenerateResizeJob(nodeAction nodeAction) (*resizeJo
// 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.Nodes, nodeAction.node, nodeAction.action)
j := newResizeJob(c.nodes, nodeAction.node, nodeAction.action)
j.Broadcaster = c.broadcaster
// toCluster is a clone of Cluster with the new node added/removed for comparison.
toCluster := newCluster()
toCluster.Nodes = Nodes(c.Nodes).Clone()
toCluster.nodes = Nodes(c.nodes).Clone()
toCluster.Hasher = c.Hasher
toCluster.partitionN = c.partitionN
toCluster.ReplicaN = c.ReplicaN
@ -1111,7 +1154,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
// multiIndex is a map of sources initialized with all the nodes in toCluster.
multiIndex := make(map[string][]*ResizeSource)
for _, n := range toCluster.Nodes {
for _, n := range toCluster.nodes {
multiIndex[n.ID] = nil
}
@ -1761,7 +1804,7 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
// except for self. Generate a list to remove first
// so that nodes aren't removed mid-loop.
nodeIDsToRemove := []string{}
for _, node := range c.Nodes {
for _, node := range c.nodes {
// Don't remove this node.
if node.ID == c.Node.ID {
continue
@ -1793,7 +1836,7 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
// 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.
func (c *cluster) unprotectedPreviousNode() *Node {
if len(c.Nodes) <= 1 {
if len(c.nodes) <= 1 {
return nil
}
@ -1801,9 +1844,9 @@ func (c *cluster) unprotectedPreviousNode() *Node {
if pos == -1 {
return nil
} else if pos == 0 {
return c.Nodes[len(c.Nodes)-1]
return c.nodes[len(c.nodes)-1]
} else {
return c.Nodes[pos-1]
return c.nodes[pos-1]
}
}
@ -1817,7 +1860,7 @@ func (c *cluster) setStatic(hosts []string) error {
if err != nil {
return errors.Wrap(err, "getting URI")
}
c.Nodes = append(c.Nodes, &Node{URI: *uri})
c.nodes = append(c.nodes, &Node{URI: *uri})
}
return nil
}

71
cluster_internal_test.go
View file

@ -22,6 +22,7 @@ import (
"strings"
"testing"
"testing/quick"
"time"
"github.com/davecgh/go-spew/spew"
"github.com/pkg/errors"
@ -316,7 +317,7 @@ func TestResizeJob(t *testing.T) {
// Ensure the cluster can fairly distribute partitions across the nodes.
func TestCluster_Owners(t *testing.T) {
c := cluster{
Nodes: []*Node{
nodes: []*Node{
{URI: NewTestURIFromHostPort("serverA", 1000)},
{URI: NewTestURIFromHostPort("serverB", 1000)},
{URI: NewTestURIFromHostPort("serverC", 1000)},
@ -326,12 +327,12 @@ func TestCluster_Owners(t *testing.T) {
}
// Verify nodes are distributed.
if a := c.partitionNodes(0); !reflect.DeepEqual(a, []*Node{c.Nodes[0], c.Nodes[1]}) {
if a := c.partitionNodes(0); !reflect.DeepEqual(a, []*Node{c.nodes[0], c.nodes[1]}) {
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
}
// Verify nodes go around the ring.
if a := c.partitionNodes(2); !reflect.DeepEqual(a, []*Node{c.Nodes[2], c.Nodes[0]}) {
if a := c.partitionNodes(2); !reflect.DeepEqual(a, []*Node{c.nodes[2], c.nodes[0]}) {
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
}
}
@ -384,7 +385,7 @@ func TestHasher(t *testing.T) {
func TestCluster_ContainsShards(t *testing.T) {
c := NewTestCluster(5)
c.ReplicaN = 3
shards := c.containsShards("test", 10, c.Nodes[2])
shards := c.containsShards("test", 10, c.nodes[2])
if !reflect.DeepEqual(shards, []uint64{0, 2, 3, 5, 6, 9, 10}) {
t.Fatalf("unexpected shars for node's index: %v", shards)
@ -804,13 +805,71 @@ func TestCluster_ResizeStates(t *testing.T) {
})
}
func TestAE(t *testing.T) {
t.Run("AbortDoesn'tBlockUninitialized", func(t *testing.T) {
c := newCluster()
ch := make(chan struct{})
go func() {
c.abortAntiEntropy()
close(ch)
}()
select {
case <-ch:
return
case <-time.After(time.Second):
t.Fatalf("aborting anti entropy on a new cluster blocked")
}
})
t.Run("AbortBlocksInitialized", func(t *testing.T) {
c := newCluster()
c.initializeAntiEntropy()
ch := make(chan struct{})
go func() {
c.abortAntiEntropy()
close(ch)
}()
select {
case <-ch:
t.Fatalf("aborting anti entropy on an initialized cluster didn't block")
case <-time.After(time.Microsecond * 100):
}
})
t.Run("AbortAntiEntropyQ", func(t *testing.T) {
c := newCluster()
c.initializeAntiEntropy()
if c.abortAntiEntropyQ() {
t.Fatalf("abortAntiEntropyQ should report false when abort not called")
}
go func() {
for {
if c.abortAntiEntropyQ() {
break
}
}
}()
ch := make(chan struct{})
go func() {
c.abortAntiEntropy()
close(ch)
}()
select {
case <-ch:
case <-time.After(time.Second):
t.Fatalf("abort should not have blocked this long")
}
})
}
// Ensures that coordinator can be changed.
func TestCluster_UpdateCoordinator(t *testing.T) {
t.Run("UpdateCoordinator", func(t *testing.T) {
c := NewTestCluster(2)
oldNode := c.Nodes[0]
newNode := c.Nodes[1]
oldNode := c.nodes[0]
newNode := c.nodes[1]
// Update coordinator to the same value.
if c.updateCoordinator(oldNode) {

8
executor.go
View file

@ -1225,7 +1225,7 @@ func (e *executor) executeSetRowAttrs(ctx context.Context, index string, c *pql.
}
// Execute on remote nodes in parallel.
nodes := Nodes(e.Cluster.Nodes).FilterID(e.Node.ID)
nodes := Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
resp := make(chan error, len(nodes))
for _, node := range nodes {
go func(node *Node) {
@ -1311,7 +1311,7 @@ func (e *executor) executeBulkSetRowAttrs(ctx context.Context, index string, cal
}
// Execute on remote nodes in parallel.
nodes := Nodes(e.Cluster.Nodes).FilterID(e.Node.ID)
nodes := Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
resp := make(chan error, len(nodes))
for _, node := range nodes {
go func(node *Node) {
@ -1360,7 +1360,7 @@ func (e *executor) executeSetColumnAttrs(ctx context.Context, index string, c *p
}
// Execute on remote nodes in parallel.
nodes := Nodes(e.Cluster.Nodes).FilterID(e.Node.ID)
nodes := Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
resp := make(chan error, len(nodes))
for _, node := range nodes {
go func(node *Node) {
@ -1431,7 +1431,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
// processing should be done locally so we start with just the local node.
var nodes []*Node
if !opt.Remote {
nodes = Nodes(e.Cluster.Nodes).Clone()
nodes = Nodes(e.Cluster.nodes).Clone()
} else {
nodes = []*Node{e.Cluster.unprotectedNodeByID(e.Node.ID)}
}

9
holder.go
View file

@ -599,8 +599,11 @@ type holderSyncer struct {
Closing <-chan struct{}
}
// IsClosing returns true if the syncer has been marked to close.
// IsClosing returns true if the syncer has been asked to close.
func (s *holderSyncer) IsClosing() bool {
if s.Cluster.abortAntiEntropyQ() {
return true
}
select {
case <-s.Closing:
return true
@ -688,7 +691,7 @@ func (s *holderSyncer) syncIndex(index string) error {
s.Stats.CountWithCustomTags("ColumnAttrStoreBlocks", int64(len(blks)), 1.0, []string{indexTag})
// Sync with every other host.
for _, node := range Nodes(s.Cluster.Nodes).FilterID(s.Node.ID) {
for _, node := range Nodes(s.Cluster.nodes).FilterID(s.Node.ID) {
// Retrieve attributes from differing blocks.
// Skip update and recomputation if no attributes have changed.
m, err := s.Cluster.InternalClient.ColumnAttrDiff(context.Background(), &node.URI, index, blks)
@ -732,7 +735,7 @@ func (s *holderSyncer) syncField(index, name string) error {
s.Stats.CountWithCustomTags("RowAttrStoreBlocks", int64(len(blks)), 1.0, []string{indexTag, fieldTag})
// Sync with every other host.
for _, node := range Nodes(s.Cluster.Nodes).FilterID(s.Node.ID) {
for _, node := range Nodes(s.Cluster.nodes).FilterID(s.Node.ID) {
// Retrieve attributes from differing blocks.
// Skip update and recomputation if no attributes have changed.
m, err := s.Cluster.InternalClient.RowAttrDiff(context.Background(), &node.URI, index, name, blks)

6
holder_internal_test.go
View file

@ -183,7 +183,7 @@ func TestHolderCleaner_CleanHolder(t *testing.T) {
// Mock 2-node, fully replicated cluster.
cluster.ReplicaN = 2
cluster.Nodes[0].URI = NewTestURIFromHostPort("localhost", 0)
cluster.nodes[0].URI = NewTestURIFromHostPort("localhost", 0)
// Create fields on nodes.
for _, hldr := range []*tHolder{hldr0} {
@ -215,7 +215,7 @@ func TestHolderCleaner_CleanHolder(t *testing.T) {
// Set up cleaner for replication 2.
cleaner2 := holderCleaner{
Node: cluster.Nodes[0],
Node: cluster.nodes[0],
Holder: hldr0.Holder,
Cluster: cluster,
}
@ -252,7 +252,7 @@ func TestHolderCleaner_CleanHolder(t *testing.T) {
// Set up cleaner for replication 1.
cleaner1 := holderCleaner{
Node: cluster.Nodes[0],
Node: cluster.nodes[0],
Holder: hldr0.Holder,
Cluster: cluster,
}

40
server.go
View file

@ -237,11 +237,12 @@ func OptServerClusterHasher(h Hasher) ServerOption {
// NewServer returns a new instance of Server.
func NewServer(opts ...ServerOption) (*Server, error) {
s := &Server{
closing: make(chan struct{}),
cluster: newCluster(),
holder: NewHolder(),
diagnostics: newDiagnosticsCollector(defaultDiagnosticServer),
systemInfo: newNopSystemInfo(),
closing: make(chan struct{}),
cluster: newCluster(),
holder: NewHolder(),
diagnostics: newDiagnosticsCollector(defaultDiagnosticServer),
systemInfo: newNopSystemInfo(),
defaultClient: nopInternalClient{},
gcNotifier: NopGCNotifier,
@ -251,6 +252,9 @@ func NewServer(opts ...ServerOption) (*Server, error) {
logger: NopLogger,
}
s.executor = newExecutor(optExecutorInternalQueryClient(s.defaultClient))
s.cluster.InternalClient = s.defaultClient
s.diagnostics.server = s
for _, opt := range opts {
@ -412,6 +416,8 @@ func (s *Server) monitorAntiEntropy() {
if s.antiEntropyInterval == 0 {
return // anti entropy disabled
}
s.cluster.initializeAntiEntropy()
ticker := time.NewTicker(s.antiEntropyInterval)
defer ticker.Stop()
@ -423,11 +429,17 @@ func (s *Server) monitorAntiEntropy() {
select {
case <-s.closing:
return
case <-s.cluster.abortAntiEntropyCh: // receive here so we don't block resizing
continue
case <-ticker.C:
s.holder.Stats.Count("AntiEntropy", 1, 1.0)
}
t := time.Now()
if s.cluster.State() == ClusterStateResizing {
continue // don't launch anti-entropy during resize.
// the cluster sets its state to resizing and *then* sends to
// abortAntiEntropyCh before starting to resize
}
// Sync holders.
s.logger.Printf("holder sync beginning")
if err := s.syncer.SyncHolder(); err != nil {
@ -439,6 +451,18 @@ func (s *Server) monitorAntiEntropy() {
s.logger.Printf("holder sync complete")
dif := time.Since(t)
s.holder.Stats.Histogram("AntiEntropyDuration", float64(dif), 1.0)
// Drain tick channel since we just finished anti-entropy. If the AE
// process took a long time, we don't want them to pile up on each
// other.
for {
select {
case <-ticker.C:
continue
default:
}
break
}
}
}
@ -537,7 +561,7 @@ func (s *Server) SendSync(m Message) error {
return fmt.Errorf("marshaling message: %v", err)
}
msg = append([]byte{getMessageType(m)}, msg...)
for _, node := range s.cluster.Nodes {
for _, node := range s.cluster.nodes {
node := node
s.logger.Printf("SendSync to: %s", node.URI)
// Don't forward the message to ourselves.
@ -637,7 +661,7 @@ func (s *Server) monitorDiagnostics() {
s.diagnostics.SetVersion(Version)
s.diagnostics.Set("Host", s.uri.Host)
s.diagnostics.Set("Cluster", strings.Join(s.cluster.nodeIDs(), ","))
s.diagnostics.Set("NumNodes", len(s.cluster.Nodes))
s.diagnostics.Set("NumNodes", len(s.cluster.nodes))
s.diagnostics.Set("NumCPU", runtime.NumCPU())
s.diagnostics.Set("NodeID", s.nodeID)
s.diagnostics.Set("ClusterID", s.cluster.id)

27
server_internal_test.go
View file

@ -15,8 +15,10 @@
package pilosa
import (
"io/ioutil"
"runtime"
"testing"
"time"
)
// Ensure the file handle count is working
@ -33,3 +35,28 @@ func TestCountOpenFiles(t *testing.T) {
t.Error("countOpenFiles returned invalid value 0.")
}
}
func TestMonitorAntiEntropyZero(t *testing.T) {
td, err := ioutil.TempDir("", "")
if err != nil {
t.Fatalf("getting temp dir: %v", err)
}
s, err := NewServer(OptServerDataDir(td),
OptServerAntiEntropyInterval(0))
if err != nil {
t.Fatalf("making new server: %v", err)
}
ch := make(chan struct{})
go func() {
s.monitorAntiEntropy()
close(ch)
}()
select {
case <-ch:
case <-time.After(time.Second):
t.Fatalf("monitorAntiEntropy should have returned immediately with duration 0")
}
}

6
utils_internal_test.go
View file

@ -40,14 +40,14 @@ func NewTestCluster(n int) *cluster {
c.Topology = newTopology()
for i := 0; i < n; i++ {
c.Nodes = append(c.Nodes, &Node{
c.nodes = append(c.nodes, &Node{
ID: fmt.Sprintf("node%d", i),
URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
})
}
c.Node = c.Nodes[0]
c.Coordinator = c.Nodes[0].ID
c.Node = c.nodes[0]
c.Coordinator = c.nodes[0].ID
c.SetState(ClusterStateNormal)
return c