From 59cdd5dde94f8309fa04a3876cdab839bd324ae7 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Wed, 29 Nov 2017 12:45:27 -0600 Subject: [PATCH] Add HolderCleaner and view.DeleteFragment to support post-resize cleanups Add tests for view.DeleteFragment and HolderCleaner --- cluster.go | 50 +++++++++++++++++- cluster_test.go | 50 +++++++++++------- holder.go | 62 ++++++++++++++++++++++ holder_test.go | 138 ++++++++++++++++++++++++++++++++++++++++++++++++ view.go | 34 ++++++++++++ view_test.go | 39 ++++++++++++-- 6 files changed, 348 insertions(+), 25 deletions(-) diff --git a/cluster.go b/cluster.go index 50996d6f1..68878f7b6 100644 --- a/cluster.go +++ b/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,28 @@ 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 + + // 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 +337,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. @@ -649,7 +679,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 +693,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). diff --git a/cluster_test.go b/cluster_test.go index b444fbf59..a7fb77c83 100644 --- a/cluster_test.go +++ b/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. diff --git a/holder.go b/holder.go index cd700e586..d5a08ec58 100644 --- a/holder.go +++ b/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 +} diff --git a/holder_test.go b/holder_test.go index f182d3496..5121db568 100644 --- a/holder_test.go +++ b/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) + } + } +} diff --git a/view.go b/view.go index 5da97d644..c6c478147 100644 --- a/view.go +++ b/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: %d", 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 diff --git a/view_test.go b/view_test.go index 5f5ce94e2..64041aea0 100644 --- a/view_test.go +++ b/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") + } +}