Merge pull request #985 from travisturner/cleanup-fragments

Add HolderCleaner and view.DeleteFragment

cluster.go

@@ -284,6 +284,10 @@ func (c *Cluster) setState(state string) {
 		return
 	}
 
+	c.logger().Printf("Change cluster state from %s to %s", c.State, state)
+
+	var doCleanup bool
+
 	switch state {
 	case ClusterStateResizing:
 		c.prefect.SetRestricted()
@@ -291,10 +295,30 @@ func (c *Cluster) setState(state string) {
 		c.prefect.SetNormal()
 		// Don't change routing for these states:
 		// - ClusterStateStarting
+
+		// If state is RESIZING -> NORMAL then run cleanup.
+		if c.State == ClusterStateResizing {
+			doCleanup = true
+		}
 	}
 
-	c.logger().Printf("Change cluster state from %s to %s", c.State, state)
 	c.State = state
+
+	// 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.
+	if doCleanup {
+		var cleaner HolderCleaner
+		cleaner.URI = c.URI
+		cleaner.Holder = c.Holder
+		cleaner.Cluster = c
+		cleaner.Closing = c.closing
+
+		// Clean holder.
+		if err := cleaner.CleanHolder(); err != nil {
+			c.logger().Printf("holder clean error: err=%s", err)
+		}
+	}
 }
 
 func (c *Cluster) SetNodeState(state string) error {
@@ -315,11 +339,19 @@ func (c *Cluster) SetNodeState(state string) error {
 	return nil
 }
 
+// ReceiveNodeState set node state in Topology in order for the
+// Coordinator to keep track of, during startup, which nodes have
+// finished opening their Holder.
 func (c *Cluster) ReceiveNodeState(uri URI, state string) error {
 	if !c.IsCoordinator() {
 		return nil
 	}
 
+	// This method is really only useful during initial startup.
+	if c.State != ClusterStateStarting {
+		return nil
+	}
+
 	c.Topology.nodeStates[uri] = state
 
 	// Set cluster state to NORMAL.
@@ -499,7 +531,7 @@ func (c *Cluster) diff(other *Cluster) (action string, uri URI, err error) {
 				break
 			}
 		}
-	} else if len(c.Nodes) > len(other.Nodes) {
+	} else if lenFrom > lenTo {
 		// Removing a node.
 		if lenFrom-lenTo > 1 {
 			return action, uri, errors.New("removing more than one node at a time is not supported")
@@ -649,7 +681,7 @@ func (c *Cluster) PartitionNodes(partitionID int) []*Node {
 	return nodes
 }
 
-// OwnsSlices find the set of slices owned by the node per Index
+// OwnsSlices finds the set of slices owned by the node per Index
 func (c *Cluster) OwnsSlices(index string, maxSlice uint64, uri URI) []uint64 {
 	var slices []uint64
 	for i := uint64(0); i <= maxSlice; i++ {
@@ -663,6 +695,22 @@ func (c *Cluster) OwnsSlices(index string, maxSlice uint64, uri URI) []uint64 {
 	return slices
 }
 
+// ContainsSlices is like OwnsSlices, but it includes replicas.
+func (c *Cluster) ContainsSlices(index string, maxSlice uint64, uri URI) []uint64 {
+	var slices []uint64
+	for i := uint64(0); i <= maxSlice; i++ {
+		p := c.Partition(index, i)
+		// Determine the nodes for partition.
+		nodes := c.PartitionNodes(p)
+		for _, node := range nodes {
+			if node.URI == uri {
+				slices = append(slices, i)
+			}
+		}
+	}
+	return slices
+}
+
 // Hasher represents an interface to hash integers into buckets.
 type Hasher interface {
 	// Hashes the key into a number between [0,N).
@@ -1228,6 +1276,16 @@ func (u NodeSet) ToStrings() []string {
 	return other
 }
 
+// ContainsURI returns true if uri matches one of the nodesets's uris.
+func (n NodeSet) ContainsURI(uri URI) bool {
+	for _, nuri := range n {
+		if nuri == uri {
+			return true
+		}
+	}
+	return false
+}
+
 // Topology represents the list of hosts in the cluster.
 type Topology struct {
 	mu      sync.RWMutex
@@ -1252,12 +1310,7 @@ func (t *Topology) ContainsURI(uri URI) bool {
 }
 
 func (t *Topology) containsURI(uri URI) bool {
-	for _, turi := range t.NodeSet {
-		if turi == uri {
-			return true
-		}
-	}
-	return false
+	return NodeSet(t.NodeSet).ContainsURI(uri)
 }
 
 func (t *Topology) positionByURI(uri URI) int {
@@ -1511,9 +1564,25 @@ func (c *Cluster) MergeClusterStatus(cs *internal.ClusterStatus) error {
 		return nil
 	}
 
-	for _, uri := range decodeURIs(cs.NodeSet) {
-		c.AddNode(uri)
+	officialURIs := decodeURIs(cs.NodeSet)
+
+	// Add all nodes from the coordinator.
+	for _, uri := range officialURIs {
+		if err := c.AddNode(uri); err != nil {
+			return err
+		}
 	}
+
+	// Remove any nodes not specified by the coordinator.
+	for _, uri := range c.NodeSet() {
+		if NodeSet(officialURIs).ContainsURI(uri) {
+			continue
+		}
+		if err := c.RemoveNode(uri); err != nil {
+			return err
+		}
+	}
+
 	c.setState(cs.State)
 
 	return nil

cluster_test.go

@@ -102,6 +102,17 @@ func TestCluster_OwnsSlices(t *testing.T) {
 	}
 }
 
+// Ensure ContainsSlices can find the actual slice list for node and index.
+func TestCluster_ContainsSlices(t *testing.T) {
+	c := test.NewCluster(5)
+	c.ReplicaN = 3
+	slices := c.ContainsSlices("test", 10, test.NewURIFromHostPort("host2", 0))
+
+	if !reflect.DeepEqual(slices, []uint64{0, 2, 3, 5, 6, 9, 10}) {
+		t.Fatalf("unexpected slices for node's index: %v", slices)
+	}
+}
+
 func TestCluster_Nodes(t *testing.T) {
 	uri0 := test.NewURIFromHostPort("node0", 0)
 	uri1 := test.NewURIFromHostPort("node1", 0)
@@ -407,6 +418,7 @@ func TestCluster_ResizeStates(t *testing.T) {
 	t.Run("Multiple nodes, with data", func(t *testing.T) {
 		tc := test.NewTestCluster(0)
 		tc.AddNode(false)
+		node0 := tc.Clusters[0]
 
 		// Open TestCluster.
 		if err := tc.Open(); err != nil {
@@ -441,15 +453,23 @@ func TestCluster_ResizeStates(t *testing.T) {
 		tc.SetFieldValue("i", "fields", 1300000, "fld0", -99)
 		tc.SetFieldValue("i", "fields", 1300000, "fld0", 99)
 
-		// AddNode needs to block until the resize process has completed.
-		if err := tc.AddNode(false); err != nil {
-			t.Fatal(err)
-		}
+		// Before starting the resize, get the CheckSum to use for
+		// comparison later.
+		node0Frame := node0.Holder.Frame("i", "f")
+		node0View := node0Frame.View("standard")
+		node0Fragment := node0View.Fragment(1)
+		node0Checksum := node0Fragment.Checksum()
 
-		node0 := tc.Clusters[0]
+		node0Frame = node0.Holder.Frame("i", "fields")
+		node0View = node0Frame.View("field_fld0")
+		node0Fragment = node0View.Fragment(1)
+		node0ChecksumFld := node0Fragment.Checksum()
+
+		// AddNode needs to block until the resize process has completed.
+		tc.AddNode(false)
 		node1 := tc.Clusters[1]
 
-		// Ensure that nodes comes up in state NORMAL.
+		// Ensure that nodes come up in state NORMAL.
 		if node0.State != pilosa.ClusterStateNormal {
 			t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State)
 		} else if node1.State != pilosa.ClusterStateNormal {
@@ -469,34 +489,24 @@ func TestCluster_ResizeStates(t *testing.T) {
 
 		// Bits
 		// Verify that node-1 contains the fragment (i/f/standard/1) transferred from node-0.
-		node0Frame := node0.Holder.Frame("i", "f")
-		node0View := node0Frame.View("standard")
-		node0Fragment := node0View.Fragment(1)
-
 		node1Frame := node1.Holder.Frame("i", "f")
 		node1View := node1Frame.View("standard")
 		node1Fragment := node1View.Fragment(1)
 
 		// Ensure checksums are the same.
-		orig := node0Fragment.Checksum()
-		if chksum := node1Fragment.Checksum(); !bytes.Equal(chksum, orig) {
-			t.Fatalf("expected standard view checksum to match: %x - %x", chksum, orig)
+		if chksum := node1Fragment.Checksum(); !bytes.Equal(chksum, node0Checksum) {
+			t.Fatalf("expected standard view checksum to match: %x - %x", chksum, node0Checksum)
 		}
 
 		// Values
 		// Verify that node-1 contains the fragment (i/fields/field_fld0/1) transferred from node-0.
-		node0Frame = node0.Holder.Frame("i", "fields")
-		node0View = node0Frame.View("field_fld0")
-		node0Fragment = node0View.Fragment(1)
-
 		node1Frame = node1.Holder.Frame("i", "fields")
 		node1View = node1Frame.View("field_fld0")
 		node1Fragment = node1View.Fragment(1)
 
 		// Ensure checksums are the same.
-		orig = node0Fragment.Checksum()
-		if chksum := node1Fragment.Checksum(); !bytes.Equal(chksum, orig) {
-			t.Fatalf("expected field view checksum to match: %x - %x", chksum, orig)
+		if chksum := node1Fragment.Checksum(); !bytes.Equal(chksum, node0ChecksumFld) {
+			t.Fatalf("expected checksum to match: %x - %x", chksum, node0ChecksumFld)
 		}
 
 		// Close TestCluster.

holder.go

@@ -712,3 +712,65 @@ func (s *HolderSyncer) syncFragment(index, frame, view string, slice uint64) err
 
 	return nil
 }
+
+// HolderCleaner removes fragments and data files that are no longer used.
+type HolderCleaner struct {
+	URI URI
+
+	Holder  *Holder
+	Cluster *Cluster
+
+	// Signals that the sync should stop.
+	Closing <-chan struct{}
+}
+
+// IsClosing returns true if the cleaner has been marked to close.
+func (c *HolderCleaner) IsClosing() bool {
+	select {
+	case <-c.Closing:
+		return true
+	default:
+		return false
+	}
+}
+
+// CleanHolder compares the holder with the cluster state and removes
+// any unnecessary fragments and files.
+func (c *HolderCleaner) CleanHolder() error {
+	for _, index := range c.Holder.Indexes() {
+		// Verify cleaner has not closed.
+		if c.IsClosing() {
+			return nil
+		}
+
+		// Get the fragments that node is responsible for (based on hash(index, node)).
+		containedSlices := c.Cluster.ContainsSlices(index.Name(), index.MaxSlice(), c.URI)
+
+		// Get the fragments registered in memory.
+		for _, frame := range index.Frames() {
+			for _, view := range frame.Views() {
+				for _, fragment := range view.Fragments() {
+					fragSlice := fragment.Slice()
+					// Ignore fragments that should be present.
+					if uint64InSlice(fragSlice, containedSlices) {
+						continue
+					}
+					// Delete fragment.
+					if err := view.DeleteFragment(fragSlice); err != nil {
+						return err
+					}
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func uint64InSlice(i uint64, s []uint64) bool {
+	for _, o := range s {
+		if i == o {
+			return true
+		}
+	}
+	return false
+}

holder_test.go

@@ -506,3 +506,141 @@ func TestHolderSyncer_SyncHolder(t *testing.T) {
 		}
 	}
 }
+
+// Ensure holder can clean up orphaned fragments.
+func TestHolderCleaner_CleanHolder(t *testing.T) {
+	cluster := test.NewCluster(2)
+
+	// Create a local holder.
+	hldr0 := test.MustOpenHolder()
+	defer hldr0.Close()
+
+	// Mock 2-node, fully replicated cluster.
+	cluster.ReplicaN = 2
+
+	cluster.Nodes[0].URI = test.NewURIFromHostPort("localhost", 0)
+
+	// Create frames on nodes.
+	for _, hldr := range []*test.Holder{hldr0} {
+		hldr.MustCreateFrameIfNotExists("i", "f")
+		hldr.MustCreateFrameIfNotExists("i", "f0")
+		hldr.MustCreateFrameIfNotExists("y", "z")
+	}
+
+	// Set data on the local holder.
+	f := hldr0.MustCreateFragmentIfNotExists("i", "f", pilosa.ViewStandard, 0)
+	if _, err := f.SetBit(0, 10); err != nil {
+		t.Fatal(err)
+	} else if _, err := f.SetBit(0, 4000); err != nil {
+		t.Fatal(err)
+	} else if _, err := f.SetBit(2, 20); err != nil {
+		t.Fatal(err)
+	} else if _, err := f.SetBit(3, 10); err != nil {
+		t.Fatal(err)
+	} else if _, err := f.SetBit(120, 10); err != nil {
+		t.Fatal(err)
+	} else if _, err := f.SetBit(200, 4); err != nil {
+		t.Fatal(err)
+	}
+
+	f = hldr0.MustCreateFragmentIfNotExists("i", "f0", pilosa.ViewStandard, 1)
+	if _, err := f.SetBit(9, SliceWidth+5); err != nil {
+		t.Fatal(err)
+	}
+
+	f = hldr0.MustCreateFragmentIfNotExists("y", "z", pilosa.ViewStandard, 2)
+	if _, err := f.SetBit(10, (2*SliceWidth)+4); err != nil {
+		t.Fatal(err)
+	} else if _, err := f.SetBit(10, (2*SliceWidth)+5); err != nil {
+		t.Fatal(err)
+	} else if _, err := f.SetBit(10, (2*SliceWidth)+7); err != nil {
+		t.Fatal(err)
+	}
+
+	// Set highest slice.
+	hldr0.Index("i").SetRemoteMaxSlice(1)
+	hldr0.Index("y").SetRemoteMaxSlice(2)
+
+	// Keep replication the same and ensure we get the expected results.
+	cluster.ReplicaN = 2
+
+	// Set up cleaner for replication 2.
+	cleaner2 := pilosa.HolderCleaner{
+		URI:     cluster.Nodes[0].URI,
+		Holder:  hldr0.Holder,
+		Cluster: cluster,
+	}
+
+	if err := cleaner2.CleanHolder(); err != nil {
+		t.Fatal(err)
+	}
+
+	// Verify data is the same on both nodes.
+	for i, hldr := range []*test.Holder{hldr0} {
+		f := hldr.Fragment("i", "f", pilosa.ViewStandard, 0)
+		if a := f.Row(0).Bits(); !reflect.DeepEqual(a, []uint64{10, 4000}) {
+			t.Fatalf("unexpected bits(%d/0): %+v", i, a)
+		} else if a := f.Row(2).Bits(); !reflect.DeepEqual(a, []uint64{20}) {
+			t.Fatalf("unexpected bits(%d/2): %+v", i, a)
+		} else if a := f.Row(3).Bits(); !reflect.DeepEqual(a, []uint64{10}) {
+			t.Fatalf("unexpected bits(%d/3): %+v", i, a)
+		} else if a := f.Row(120).Bits(); !reflect.DeepEqual(a, []uint64{10}) {
+			t.Fatalf("unexpected bits(%d/120): %+v", i, a)
+		} else if a := f.Row(200).Bits(); !reflect.DeepEqual(a, []uint64{4}) {
+			t.Fatalf("unexpected bits(%d/200): %+v", i, a)
+		}
+
+		f = hldr.Fragment("i", "f0", pilosa.ViewStandard, 1)
+		a := f.Row(9).Bits()
+		if !reflect.DeepEqual(a, []uint64{SliceWidth + 5}) {
+			t.Fatalf("unexpected bits(%d/i/f0): %+v", i, a)
+		}
+		if a := f.Row(9).Bits(); !reflect.DeepEqual(a, []uint64{SliceWidth + 5}) {
+			t.Fatalf("unexpected bits(%d/d/f0): %+v", i, a)
+		}
+		f = hldr.Fragment("y", "z", pilosa.ViewStandard, 2)
+		if a := f.Row(10).Bits(); !reflect.DeepEqual(a, []uint64{(2 * SliceWidth) + 4, (2 * SliceWidth) + 5, (2 * SliceWidth) + 7}) {
+			t.Fatalf("unexpected bits(%d/y/z): %+v", i, a)
+		}
+	}
+
+	// Change replication factor to ensure we have fragments to remove.
+	cluster.ReplicaN = 1
+
+	// Set up cleaner for replication 1.
+	cleaner1 := pilosa.HolderCleaner{
+		URI:     cluster.Nodes[0].URI,
+		Holder:  hldr0.Holder,
+		Cluster: cluster,
+	}
+
+	if err := cleaner1.CleanHolder(); err != nil {
+		t.Fatal(err)
+	}
+
+	// Verify data is the same on both nodes.
+	for i, hldr := range []*test.Holder{hldr0} {
+		f := hldr.Fragment("i", "f", pilosa.ViewStandard, 0)
+		if a := f.Row(0).Bits(); !reflect.DeepEqual(a, []uint64{10, 4000}) {
+			t.Fatalf("unexpected bits(%d/0): %+v", i, a)
+		} else if a := f.Row(2).Bits(); !reflect.DeepEqual(a, []uint64{20}) {
+			t.Fatalf("unexpected bits(%d/2): %+v", i, a)
+		} else if a := f.Row(3).Bits(); !reflect.DeepEqual(a, []uint64{10}) {
+			t.Fatalf("unexpected bits(%d/3): %+v", i, a)
+		} else if a := f.Row(120).Bits(); !reflect.DeepEqual(a, []uint64{10}) {
+			t.Fatalf("unexpected bits(%d/120): %+v", i, a)
+		} else if a := f.Row(200).Bits(); !reflect.DeepEqual(a, []uint64{4}) {
+			t.Fatalf("unexpected bits(%d/200): %+v", i, a)
+		}
+
+		f = hldr.Fragment("i", "f0", pilosa.ViewStandard, 1)
+		if f != nil {
+			t.Fatalf("expected fragment to be deleted: (%d/i/f0): %+v", i, f)
+		}
+
+		f = hldr.Fragment("y", "z", pilosa.ViewStandard, 2)
+		if a := f.Row(10).Bits(); !reflect.DeepEqual(a, []uint64{(2 * SliceWidth) + 4, (2 * SliceWidth) + 5, (2 * SliceWidth) + 7}) {
+			t.Fatalf("unexpected bits(%d/y/z): %+v", i, a)
+		}
+	}
+}

view.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"log"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -119,6 +120,9 @@ func (v *View) Open() error {
 	return nil
 }
 
+// logger returns a logger instance for the view.
+func (v *View) logger() *log.Logger { return log.New(v.LogOutput, "", log.LstdFlags) }
+
 // openFragments opens and initializes the fragments inside the view.
 func (v *View) openFragments() error {
 	file, err := os.Open(filepath.Join(v.path, "fragments"))
@@ -270,6 +274,36 @@ func (v *View) newFragment(path string, slice uint64) *Fragment {
 	return frag
 }
 
+// DeleteFragment removes the fragment from the view.
+func (v *View) DeleteFragment(slice uint64) error {
+
+	fragment := v.fragments[slice]
+	if fragment == nil {
+		return ErrFragmentNotFound
+	}
+
+	v.logger().Printf("delete fragment: (%s/%s/%s) %d", v.index, v.frame, v.name, slice)
+
+	// Close data files before deletion.
+	if err := fragment.Close(); err != nil {
+		return err
+	}
+
+	// Delete fragment file.
+	if err := os.Remove(fragment.Path()); err != nil {
+		return err
+	}
+
+	// Delete fragment cache file.
+	if err := os.Remove(fragment.CachePath()); err != nil {
+		v.logger().Printf("no cache file to delete for slice %d", slice)
+	}
+
+	delete(v.fragments, slice)
+
+	return nil
+}
+
 // SetBit sets a bit within the view.
 func (v *View) SetBit(rowID, columnID uint64) (changed bool, err error) {
 	slice := columnID / SliceWidth

view_test.go

@@ -17,6 +17,7 @@ package pilosa_test
 import (
 	"io/ioutil"
 	"os"
+	"testing"
 
 	"github.com/pilosa/pilosa"
 	"github.com/pilosa/pilosa/test"
@@ -30,14 +31,13 @@ type View struct {
 
 // NewView returns a new instance of View with a temporary path.
 func NewView(index, frame, name string) *View {
-	file, err := ioutil.TempFile("", "pilosa-view-")
+	path, err := ioutil.TempDir("", "pilosa-view-")
 	if err != nil {
 		panic(err)
 	}
-	file.Close()
 
 	v := &View{
-		View:         pilosa.NewView(file.Name(), index, frame, name, pilosa.DefaultCacheSize),
+		View:         pilosa.NewView(path, index, frame, name, pilosa.DefaultCacheSize),
 		RowAttrStore: test.MustOpenAttrStore(),
 	}
 	v.View.RowAttrStore = v.RowAttrStore.AttrStore
@@ -93,3 +93,36 @@ func (v *View) MustClearBits(rowID uint64, columnIDs ...uint64) {
 		}
 	}
 }
+
+// Ensure view can open and retrieve a fragment.
+func TestView_DeleteFragment(t *testing.T) {
+	v := MustOpenView("i", "f", "v")
+	defer v.Close()
+
+	slice := uint64(9)
+
+	// Create fragment.
+	fragment, err := v.CreateFragmentIfNotExists(slice)
+	if err != nil {
+		t.Fatal(err)
+	} else if fragment == nil {
+		t.Fatal("expected fragment")
+	}
+
+	err = v.DeleteFragment(slice)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if v.Fragment(slice) != nil {
+		t.Fatal("fragment still exists in view")
+	}
+
+	// Recreate fragment with same slice, verify that the old fragment was not reused.
+	fragment2, err := v.CreateFragmentIfNotExists(slice)
+	if err != nil {
+		t.Fatal(err)
+	} else if fragment == fragment2 {
+		t.Fatal("failed to create new fragment")
+	}
+}