From a6a0c6a7c3a90a284860736004ced292ead496fa Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 2 Jul 2018 13:58:20 -0500 Subject: [PATCH 1/3] unexport Holder.view and prepare to unexport Holder.Fragment --- holder.go | 6 +- holder_internal_test.go | 152 ++++++++++++++++++++++++++++++++++++++++ holder_test.go | 114 ------------------------------ http/client_test.go | 4 +- 4 files changed, 158 insertions(+), 118 deletions(-) diff --git a/holder.go b/holder.go index bbd757ce4..e02daebb3 100644 --- a/holder.go +++ b/holder.go @@ -402,8 +402,8 @@ func (h *Holder) Field(index, name string) *Field { return idx.Field(name) } -// View returns the view for an index, field, and name. -func (h *Holder) View(index, field, name string) *View { +// view returns the view for an index, field, and name. +func (h *Holder) view(index, field, name string) *View { f := h.Field(index, field) if f == nil { return nil @@ -413,7 +413,7 @@ func (h *Holder) View(index, field, name string) *View { // Fragment returns the fragment for an index, field & shard. func (h *Holder) Fragment(index, field, view string, shard uint64) *Fragment { - v := h.View(index, field, view) + v := h.view(index, field, view) if v == nil { return nil } diff --git a/holder_internal_test.go b/holder_internal_test.go index 4bf9fae8b..fb725fe0a 100644 --- a/holder_internal_test.go +++ b/holder_internal_test.go @@ -18,6 +18,7 @@ import ( "io/ioutil" "os" "path/filepath" + "reflect" "strings" "testing" ) @@ -57,6 +58,43 @@ func newHolder() *tHolder { return h } +// MustCreateFieldIfNotExists returns a given field. Panic on error. +func (h *tHolder) MustCreateFieldIfNotExists(index, field string) *Field { + f, err := h.MustCreateIndexIfNotExists(index, IndexOptions{}).CreateFieldIfNotExists(field, FieldOptions{}) + if err != nil { + panic(err) + } + return f +} + +// MustCreateIndexIfNotExists returns a given index. Panic on error. +func (h *tHolder) MustCreateIndexIfNotExists(index string, opt IndexOptions) *Index { + idx, err := h.Holder.CreateIndexIfNotExists(index, opt) + if err != nil { + panic(err) + } + return idx +} + +// SetBit clears a bit on the given field. +func (h *tHolder) SetBit(index, field string, rowID, columnID uint64) { + f := h.MustCreateFieldIfNotExists(index, field) + _, err := f.SetBit(rowID, columnID, nil) + if err != nil { + panic(err) + } +} + +// Row returns a Row for a given field. +func (h *tHolder) Row(index, field string, rowID uint64) *Row { + f := h.MustCreateFieldIfNotExists(index, field) + row, err := f.Row(rowID) + if err != nil { + panic(err) + } + return row +} + func TestHolder_Optn(t *testing.T) { t.Run("ErrViewPermission", func(t *testing.T) { if os.Geteuid() == 0 { @@ -137,3 +175,117 @@ func TestHolder_Optn(t *testing.T) { }) } + +// Ensure holder can clean up orphaned fragments. +func TestHolderCleaner_CleanHolder(t *testing.T) { + cluster := NewTestCluster(2) + + // Create a local holder. + hldr0 := newHolder() + defer hldr0.Close() + + // Mock 2-node, fully replicated cluster. + cluster.ReplicaN = 2 + + cluster.Nodes[0].URI = NewTestURIFromHostPort("localhost", 0) + + // Create fields on nodes. + for _, hldr := range []*tHolder{hldr0} { + hldr.MustCreateFieldIfNotExists("i", "f") + hldr.MustCreateFieldIfNotExists("i", "f0") + hldr.MustCreateFieldIfNotExists("y", "z") + } + + // Set data on the local holder. + hldr0.SetBit("i", "f", 0, 10) + hldr0.SetBit("i", "f", 0, 4000) + hldr0.SetBit("i", "f", 2, 20) + hldr0.SetBit("i", "f", 3, 10) + hldr0.SetBit("i", "f", 120, 10) + hldr0.SetBit("i", "f", 200, 4) + + hldr0.SetBit("i", "f0", 9, ShardWidth+5) + + hldr0.SetBit("y", "z", 10, (2*ShardWidth)+4) + hldr0.SetBit("y", "z", 10, (2*ShardWidth)+5) + hldr0.SetBit("y", "z", 10, (2*ShardWidth)+7) + + // Set highest shard. + hldr0.Index("i").SetRemoteMaxShard(1) + hldr0.Index("y").SetRemoteMaxShard(2) + + // Keep replication the same and ensure we get the expected results. + cluster.ReplicaN = 2 + + // Set up cleaner for replication 2. + cleaner2 := HolderCleaner{ + Node: cluster.Nodes[0], + 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 []*tHolder{hldr0} { + if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{10, 4000}) { + t.Fatalf("unexpected columns(%d/0): %+v", i, a) + } else if a := hldr.Row("i", "f", 2).Columns(); !reflect.DeepEqual(a, []uint64{20}) { + t.Fatalf("unexpected columns(%d/2): %+v", i, a) + } else if a := hldr.Row("i", "f", 3).Columns(); !reflect.DeepEqual(a, []uint64{10}) { + t.Fatalf("unexpected columns(%d/3): %+v", i, a) + } else if a := hldr.Row("i", "f", 120).Columns(); !reflect.DeepEqual(a, []uint64{10}) { + t.Fatalf("unexpected columns(%d/120): %+v", i, a) + } else if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{4}) { + t.Fatalf("unexpected columns(%d/200): %+v", i, a) + } + + if a := hldr.Row("i", "f0", 9).Columns(); !reflect.DeepEqual(a, []uint64{ShardWidth + 5}) { + t.Fatalf("unexpected columns(%d/d/f0): %+v", i, a) + } + + if a := hldr.Row("y", "z", 10).Columns(); !reflect.DeepEqual(a, []uint64{(2 * ShardWidth) + 4, (2 * ShardWidth) + 5, (2 * ShardWidth) + 7}) { + t.Fatalf("unexpected columns(%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 := HolderCleaner{ + Node: cluster.Nodes[0], + 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 []*tHolder{hldr0} { + if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{10, 4000}) { + t.Fatalf("unexpected columns(%d/0): %+v", i, a) + } else if a := hldr.Row("i", "f", 2).Columns(); !reflect.DeepEqual(a, []uint64{20}) { + t.Fatalf("unexpected columns(%d/2): %+v", i, a) + } else if a := hldr.Row("i", "f", 3).Columns(); !reflect.DeepEqual(a, []uint64{10}) { + t.Fatalf("unexpected columns(%d/3): %+v", i, a) + } else if a := hldr.Row("i", "f", 120).Columns(); !reflect.DeepEqual(a, []uint64{10}) { + t.Fatalf("unexpected columns(%d/120): %+v", i, a) + } else if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{4}) { + t.Fatalf("unexpected columns(%d/200): %+v", i, a) + } + + f := hldr.Fragment("i", "f0", ViewStandard, 1) + if f != nil { + t.Fatalf("expected fragment to be deleted: (%d/i/f0): %+v", i, f) + } + + if a := hldr.Row("y", "z", 10).Columns(); !reflect.DeepEqual(a, []uint64{(2 * ShardWidth) + 4, (2 * ShardWidth) + 5, (2 * ShardWidth) + 7}) { + t.Fatalf("unexpected columns(%d/y/z): %+v", i, a) + } + } +} diff --git a/holder_test.go b/holder_test.go index 421f16012..078ab55f3 100644 --- a/holder_test.go +++ b/holder_test.go @@ -362,117 +362,3 @@ func TestHolderSyncer_SyncHolder(t *testing.T) { } } } - -// Ensure holder can clean up orphaned fragments. -func TestHolderCleaner_CleanHolder(t *testing.T) { - cluster := pilosa.NewTestCluster(2) - - // Create a local holder. - hldr0 := test.MustOpenHolder() - defer hldr0.Close() - - // Mock 2-node, fully replicated cluster. - cluster.ReplicaN = 2 - - cluster.Nodes[0].URI = pilosa.NewTestURIFromHostPort("localhost", 0) - - // Create fields on nodes. - for _, hldr := range []*test.Holder{hldr0} { - hldr.MustCreateFieldIfNotExists("i", "f") - hldr.MustCreateFieldIfNotExists("i", "f0") - hldr.MustCreateFieldIfNotExists("y", "z") - } - - // Set data on the local holder. - hldr0.SetBit("i", "f", 0, 10) - hldr0.SetBit("i", "f", 0, 4000) - hldr0.SetBit("i", "f", 2, 20) - hldr0.SetBit("i", "f", 3, 10) - hldr0.SetBit("i", "f", 120, 10) - hldr0.SetBit("i", "f", 200, 4) - - hldr0.SetBit("i", "f0", 9, ShardWidth+5) - - hldr0.SetBit("y", "z", 10, (2*ShardWidth)+4) - hldr0.SetBit("y", "z", 10, (2*ShardWidth)+5) - hldr0.SetBit("y", "z", 10, (2*ShardWidth)+7) - - // Set highest shard. - hldr0.Index("i").SetRemoteMaxShard(1) - hldr0.Index("y").SetRemoteMaxShard(2) - - // Keep replication the same and ensure we get the expected results. - cluster.ReplicaN = 2 - - // Set up cleaner for replication 2. - cleaner2 := pilosa.HolderCleaner{ - Node: cluster.Nodes[0], - 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} { - if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{10, 4000}) { - t.Fatalf("unexpected columns(%d/0): %+v", i, a) - } else if a := hldr.Row("i", "f", 2).Columns(); !reflect.DeepEqual(a, []uint64{20}) { - t.Fatalf("unexpected columns(%d/2): %+v", i, a) - } else if a := hldr.Row("i", "f", 3).Columns(); !reflect.DeepEqual(a, []uint64{10}) { - t.Fatalf("unexpected columns(%d/3): %+v", i, a) - } else if a := hldr.Row("i", "f", 120).Columns(); !reflect.DeepEqual(a, []uint64{10}) { - t.Fatalf("unexpected columns(%d/120): %+v", i, a) - } else if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{4}) { - t.Fatalf("unexpected columns(%d/200): %+v", i, a) - } - - if a := hldr.Row("i", "f0", 9).Columns(); !reflect.DeepEqual(a, []uint64{ShardWidth + 5}) { - t.Fatalf("unexpected columns(%d/d/f0): %+v", i, a) - } - - if a := hldr.Row("y", "z", 10).Columns(); !reflect.DeepEqual(a, []uint64{(2 * ShardWidth) + 4, (2 * ShardWidth) + 5, (2 * ShardWidth) + 7}) { - t.Fatalf("unexpected columns(%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{ - Node: cluster.Nodes[0], - 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} { - if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{10, 4000}) { - t.Fatalf("unexpected columns(%d/0): %+v", i, a) - } else if a := hldr.Row("i", "f", 2).Columns(); !reflect.DeepEqual(a, []uint64{20}) { - t.Fatalf("unexpected columns(%d/2): %+v", i, a) - } else if a := hldr.Row("i", "f", 3).Columns(); !reflect.DeepEqual(a, []uint64{10}) { - t.Fatalf("unexpected columns(%d/3): %+v", i, a) - } else if a := hldr.Row("i", "f", 120).Columns(); !reflect.DeepEqual(a, []uint64{10}) { - t.Fatalf("unexpected columns(%d/120): %+v", i, a) - } else if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{4}) { - t.Fatalf("unexpected columns(%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) - } - - if a := hldr.Row("y", "z", 10).Columns(); !reflect.DeepEqual(a, []uint64{(2 * ShardWidth) + 4, (2 * ShardWidth) + 5, (2 * ShardWidth) + 7}) { - t.Fatalf("unexpected columns(%d/y/z): %+v", i, a) - } - } -} diff --git a/http/client_test.go b/http/client_test.go index fc8e896b5..fb1105a27 100644 --- a/http/client_test.go +++ b/http/client_test.go @@ -306,7 +306,9 @@ func TestClient_FragmentBlocks(t *testing.T) { } // Verify data matches local blocks. - if a := hldr.Fragment("i", "f", pilosa.ViewStandard, 0).Blocks(); !reflect.DeepEqual(a, blocks) { + if a, err := cmd.API.FragmentBlocks(context.Background(), "i", "f", 0); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(a, blocks) { t.Fatalf("blocks mismatch:\n\nexp=%s\n\ngot=%s\n\n", spew.Sdump(a), spew.Sdump(blocks)) } } From 9995b0032e03a602dece88b9de59f710541470ad Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 2 Jul 2018 14:00:31 -0500 Subject: [PATCH 2/3] unexport Holder.Fragment, HolderSyncer and HolderCleaner --- api.go | 8 ++++---- cluster.go | 2 +- executor.go | 18 +++++++++--------- holder.go | 26 +++++++++++++------------- holder_internal_test.go | 6 +++--- server.go | 2 +- utils_internal_test.go | 4 ++-- 7 files changed, 33 insertions(+), 33 deletions(-) diff --git a/api.go b/api.go index 2c25dcd66..ed0a925a6 100644 --- a/api.go +++ b/api.go @@ -337,7 +337,7 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin } // Find the fragment. - f := api.holder.Fragment(indexName, fieldName, ViewStandard, shard) + f := api.holder.fragment(indexName, fieldName, ViewStandard, shard) if f == nil { return ErrFragmentNotFound } @@ -379,7 +379,7 @@ func (api *API) MarshalFragment(ctx context.Context, indexName string, fieldName } // Retrieve fragment from holder. - f := api.holder.Fragment(indexName, fieldName, ViewStandard, shard) + f := api.holder.fragment(indexName, fieldName, ViewStandard, shard) if f == nil { return nil, ErrFragmentNotFound } @@ -437,7 +437,7 @@ func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte, } // Retrieve fragment from holder. - f := api.holder.Fragment(req.Index, req.Field, ViewStandard, req.Shard) + f := api.holder.fragment(req.Index, req.Field, ViewStandard, req.Shard) if f == nil { return nil, ErrFragmentNotFound } @@ -461,7 +461,7 @@ func (api *API) FragmentBlocks(ctx context.Context, indexName string, fieldName } // Retrieve fragment from holder. - f := api.holder.Fragment(indexName, fieldName, ViewStandard, shard) + f := api.holder.fragment(indexName, fieldName, ViewStandard, shard) if f == nil { return nil, ErrFragmentNotFound } diff --git a/cluster.go b/cluster.go index 907d42c0d..a5108a2f9 100644 --- a/cluster.go +++ b/cluster.go @@ -450,7 +450,7 @@ func (c *cluster) setState(state string) { // been removed. // It's safe to do a cleanup after state changes back to normal. if doCleanup { - var cleaner HolderCleaner + var cleaner holderCleaner cleaner.Node = c.Node cleaner.Holder = c.holder cleaner.Cluster = c diff --git a/executor.go b/executor.go index 4194e7b4e..5b4e12351 100644 --- a/executor.go +++ b/executor.go @@ -427,7 +427,7 @@ func (e *executor) executeSumCountShard(ctx context.Context, index string, c *pq return ValCount{}, nil } - fragment := e.Holder.Fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard) + fragment := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard) if fragment == nil { return ValCount{}, nil } @@ -465,7 +465,7 @@ func (e *executor) executeMinShard(ctx context.Context, index string, c *pql.Cal return ValCount{}, nil } - fragment := e.Holder.Fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard) + fragment := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard) if fragment == nil { return ValCount{}, nil } @@ -503,7 +503,7 @@ func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Cal return ValCount{}, nil } - fragment := e.Holder.Fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard) + fragment := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard) if fragment == nil { return ValCount{}, nil } @@ -623,7 +623,7 @@ func (e *executor) executeTopNShard(ctx context.Context, index string, c *pql.Ca field = defaultField } - f := e.Holder.Fragment(index, field, ViewStandard, shard) + f := e.Holder.fragment(index, field, ViewStandard, shard) if f == nil { return nil, nil } @@ -693,7 +693,7 @@ func (e *executor) executeBitmapShard(ctx context.Context, index string, c *pql. return nil, fmt.Errorf("Row() must specify %v", rowLabel) } - frag := e.Holder.Fragment(index, fieldName, ViewStandard, shard) + frag := e.Holder.fragment(index, fieldName, ViewStandard, shard) if frag == nil { return NewRow(), nil } @@ -785,7 +785,7 @@ func (e *executor) executeRangeShard(ctx context.Context, index string, c *pql.C // Union bitmaps across all time-based views. row := &Row{} for _, view := range viewsByTimeRange(ViewStandard, startTime, endTime, q) { - f := e.Holder.Fragment(index, fieldName, view, shard) + f := e.Holder.fragment(index, fieldName, view, shard) if f == nil { continue } @@ -836,7 +836,7 @@ func (e *executor) executeBSIGroupRangeShard(ctx context.Context, index string, } // Retrieve fragment. - frag := e.Holder.Fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard) + frag := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard) if frag == nil { return NewRow(), nil } @@ -871,7 +871,7 @@ func (e *executor) executeBSIGroupRangeShard(ctx context.Context, index string, } // Retrieve fragment. - frag := e.Holder.Fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard) + frag := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard) if frag == nil { return NewRow(), nil } @@ -904,7 +904,7 @@ func (e *executor) executeBSIGroupRangeShard(ctx context.Context, index string, } // Retrieve fragment. - frag := e.Holder.Fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard) + frag := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard) if frag == nil { return NewRow(), nil } diff --git a/holder.go b/holder.go index e02daebb3..bad001787 100644 --- a/holder.go +++ b/holder.go @@ -411,8 +411,8 @@ func (h *Holder) view(index, field, name string) *View { return f.view(name) } -// Fragment returns the fragment for an index, field & shard. -func (h *Holder) Fragment(index, field, view string, shard uint64) *Fragment { +// fragment returns the fragment for an index, field & shard. +func (h *Holder) fragment(index, field, view string, shard uint64) *Fragment { v := h.view(index, field, view) if v == nil { return nil @@ -561,9 +561,9 @@ func (h *Holder) logStartup() error { return nil } -// HolderSyncer is an active anti-entropy tool that compares the local holder +// holderSyncer is an active anti-entropy tool that compares the local holder // with a remote holder based on block checksums and resolves differences. -type HolderSyncer struct { +type holderSyncer struct { mu sync.Mutex Holder *Holder @@ -579,7 +579,7 @@ type HolderSyncer struct { } // IsClosing returns true if the syncer has been marked to close. -func (s *HolderSyncer) IsClosing() bool { +func (s *holderSyncer) IsClosing() bool { select { case <-s.Closing: return true @@ -589,7 +589,7 @@ func (s *HolderSyncer) IsClosing() bool { } // SyncHolder compares the holder on host with the local holder and resolves differences. -func (s *HolderSyncer) SyncHolder() error { +func (s *holderSyncer) SyncHolder() error { s.mu.Lock() // only allow one instance of SyncHolder to be running at a time defer s.mu.Unlock() ti := time.Now() @@ -651,7 +651,7 @@ func (s *HolderSyncer) SyncHolder() error { } // syncIndex synchronizes index attributes with the rest of the cluster. -func (s *HolderSyncer) syncIndex(index string) error { +func (s *holderSyncer) syncIndex(index string) error { // Retrieve index reference. idx := s.Holder.Index(index) if idx == nil { @@ -694,7 +694,7 @@ func (s *HolderSyncer) syncIndex(index string) error { } // syncField synchronizes field attributes with the rest of the cluster. -func (s *HolderSyncer) syncField(index, name string) error { +func (s *holderSyncer) syncField(index, name string) error { // Retrieve field reference. f := s.Holder.Field(index, name) if f == nil { @@ -740,7 +740,7 @@ func (s *HolderSyncer) syncField(index, name string) error { } // syncFragment synchronizes a fragment with the rest of the cluster. -func (s *HolderSyncer) syncFragment(index, field, view string, shard uint64) error { +func (s *holderSyncer) syncFragment(index, field, view string, shard uint64) error { // Retrieve local field. f := s.Holder.Field(index, field) if f == nil { @@ -773,8 +773,8 @@ func (s *HolderSyncer) syncFragment(index, field, view string, shard uint64) err return nil } -// HolderCleaner removes fragments and data files that are no longer used. -type HolderCleaner struct { +// holderCleaner removes fragments and data files that are no longer used. +type holderCleaner struct { Node *Node Holder *Holder @@ -785,7 +785,7 @@ type HolderCleaner struct { } // IsClosing returns true if the cleaner has been marked to close. -func (c *HolderCleaner) IsClosing() bool { +func (c *holderCleaner) IsClosing() bool { select { case <-c.Closing: return true @@ -796,7 +796,7 @@ func (c *HolderCleaner) IsClosing() bool { // CleanHolder compares the holder with the cluster state and removes // any unnecessary fragments and files. -func (c *HolderCleaner) CleanHolder() error { +func (c *holderCleaner) CleanHolder() error { for _, index := range c.Holder.Indexes() { // Verify cleaner has not closed. if c.IsClosing() { diff --git a/holder_internal_test.go b/holder_internal_test.go index fb725fe0a..d7953b9ee 100644 --- a/holder_internal_test.go +++ b/holder_internal_test.go @@ -218,7 +218,7 @@ func TestHolderCleaner_CleanHolder(t *testing.T) { cluster.ReplicaN = 2 // Set up cleaner for replication 2. - cleaner2 := HolderCleaner{ + cleaner2 := holderCleaner{ Node: cluster.Nodes[0], Holder: hldr0.Holder, Cluster: cluster, @@ -255,7 +255,7 @@ func TestHolderCleaner_CleanHolder(t *testing.T) { cluster.ReplicaN = 1 // Set up cleaner for replication 1. - cleaner1 := HolderCleaner{ + cleaner1 := holderCleaner{ Node: cluster.Nodes[0], Holder: hldr0.Holder, Cluster: cluster, @@ -279,7 +279,7 @@ func TestHolderCleaner_CleanHolder(t *testing.T) { t.Fatalf("unexpected columns(%d/200): %+v", i, a) } - f := hldr.Fragment("i", "f0", ViewStandard, 1) + f := hldr.fragment("i", "f0", ViewStandard, 1) if f != nil { t.Fatalf("expected fragment to be deleted: (%d/i/f0): %+v", i, f) } diff --git a/server.go b/server.go index d14e3c37d..0e03365ae 100644 --- a/server.go +++ b/server.go @@ -70,7 +70,7 @@ type Server struct { diagnosticInterval time.Duration maxWritesPerRequest int isCoordinator bool - syncer HolderSyncer + syncer holderSyncer primaryTranslateStore TranslateStore diff --git a/utils_internal_test.go b/utils_internal_test.go index 939e8e04c..f7309961c 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -368,8 +368,8 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstructi srcNode := DecodeNode(src.Node) srcCluster := t.clusterByID(srcNode.ID) - srcFragment := srcCluster.holder.Fragment(src.Index, src.Field, src.View, src.Shard) - destFragment := destCluster.holder.Fragment(src.Index, src.Field, src.View, src.Shard) + srcFragment := srcCluster.holder.fragment(src.Index, src.Field, src.View, src.Shard) + destFragment := destCluster.holder.fragment(src.Index, src.Field, src.View, src.Shard) if destFragment == nil { // Create fragment on destination if it doesn't exist. f := destCluster.holder.Field(src.Index, src.Field) From 009242fef95e62af89746d47ae0f929e626a1110 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 2 Jul 2018 14:07:06 -0500 Subject: [PATCH 3/3] unexport more Holder stuff (gorename) --- api.go | 2 +- cluster.go | 4 ++-- holder.go | 46 +++++++++++++++++++++--------------------- server.go | 10 ++++----- utils_internal_test.go | 2 +- 5 files changed, 32 insertions(+), 32 deletions(-) diff --git a/api.go b/api.go index ed0a925a6..c745b2a0b 100644 --- a/api.go +++ b/api.go @@ -697,7 +697,7 @@ func (api *API) ImportValue(ctx context.Context, req internal.ImportValueRequest // MaxShards returns the maximum shard number for each index in a map. func (api *API) MaxShards(ctx context.Context) map[string]uint64 { - return api.holder.MaxShards() + return api.holder.maxShards() } // StatsWithTags returns an instance of whatever implementation of StatsClient diff --git a/cluster.go b/cluster.go index a5108a2f9..a3f5aa775 100644 --- a/cluster.go +++ b/cluster.go @@ -1150,7 +1150,7 @@ func (c *cluster) generateResizeJobByAction(nodeAction nodeAction) (*resizeJob, Node: EncodeNode(toCluster.unprotectedNodeByID(id)), Coordinator: EncodeNode(c.coordinatorNode()), Sources: sources, - Schema: c.holder.EncodeSchema(), // Include the schema to ensure it's in sync on the receiving node. + Schema: c.holder.encodeSchema(), // Include the schema to ensure it's in sync on the receiving node. ClusterStatus: c.Status(), } j.Instructions = append(j.Instructions, instr) @@ -1205,7 +1205,7 @@ func (c *cluster) followResizeInstruction(instr *internal.ResizeInstruction) err // Sync the schema received in the resize instruction. c.logger.Printf("Holder ApplySchema") - if err := c.holder.ApplySchema(instr.Schema); err != nil { + if err := c.holder.applySchema(instr.Schema); err != nil { return errors.Wrap(err, "applying schema") } diff --git a/holder.go b/holder.go index bad001787..8cf830a84 100644 --- a/holder.go +++ b/holder.go @@ -36,8 +36,8 @@ const ( // defaultCacheFlushInterval is the default value for Fragment.CacheFlushInterval. defaultCacheFlushInterval = 1 * time.Minute - // FileLimit is the maximum open file limit (ulimit -n) to automatically set. - FileLimit = 262144 // (512^2) + // fileLimit is the maximum open file limit (ulimit -n) to automatically set. + fileLimit = 262144 // (512^2) ) // Holder represents a container for indexes. @@ -50,7 +50,7 @@ type Holder struct { // opened channel is closed once Open() completes. opened chan struct{} - Broadcaster broadcaster + broadcaster broadcaster NewAttrStore func(string) AttrStore @@ -65,7 +65,7 @@ type Holder struct { Path string // The interval at which the cached row ids are persisted to disk. - CacheFlushInterval time.Duration + cacheFlushInterval time.Duration Logger Logger } @@ -78,12 +78,12 @@ func NewHolder() *Holder { opened: make(chan struct{}), - Broadcaster: NopBroadcaster, + broadcaster: NopBroadcaster, Stats: NopStatsClient, NewAttrStore: newNopAttrStore, - CacheFlushInterval: defaultCacheFlushInterval, + cacheFlushInterval: defaultCacheFlushInterval, Logger: NopLogger, } @@ -200,8 +200,8 @@ func (h *Holder) HasData() (bool, error) { return false, nil } -// MaxShards returns MaxShard map for all indexes. -func (h *Holder) MaxShards() map[string]uint64 { +// maxShards returns MaxShard map for all indexes. +func (h *Holder) maxShards() map[string]uint64 { a := make(map[string]uint64) for _, index := range h.Indexes() { a[index.Name()] = index.MaxShard() @@ -229,8 +229,8 @@ func (h *Holder) Schema() []*IndexInfo { return a } -// ApplySchema applies an internal Schema to Holder. -func (h *Holder) ApplySchema(schema *internal.Schema) error { +// applySchema applies an internal Schema to Holder. +func (h *Holder) applySchema(schema *internal.Schema) error { // Create indexes that don't exist. for _, index := range schema.Indexes { opt := IndexOptions{} @@ -257,15 +257,15 @@ func (h *Holder) ApplySchema(schema *internal.Schema) error { return nil } -// EncodeMaxShards creates and internal representation of max shards. -func (h *Holder) EncodeMaxShards() *internal.MaxShards { +// encodeMaxShards creates and internal representation of max shards. +func (h *Holder) encodeMaxShards() *internal.MaxShards { return &internal.MaxShards{ - Standard: h.MaxShards(), + Standard: h.maxShards(), } } -// EncodeSchema creates an internal representation of schema. -func (h *Holder) EncodeSchema() *internal.Schema { +// encodeSchema creates an internal representation of schema. +func (h *Holder) encodeSchema() *internal.Schema { return &internal.Schema{ Indexes: EncodeIndexes(h.Indexes()), } @@ -360,7 +360,7 @@ func (h *Holder) newIndex(path, name string) (*Index, error) { } index.Logger = h.Logger index.Stats = h.Stats.WithTags(fmt.Sprintf("index:%s", index.Name())) - index.broadcaster = h.Broadcaster + index.broadcaster = h.broadcaster index.NewAttrStore = h.NewAttrStore index.columnAttrStore = h.NewAttrStore(filepath.Join(index.path, ".data")) return index, nil @@ -423,7 +423,7 @@ func (h *Holder) fragment(index, field, view string, shard uint64) *Fragment { // monitorCacheFlush periodically flushes all fragment caches sequentially. // This is run in a goroutine. func (h *Holder) monitorCacheFlush() { - ticker := time.NewTicker(h.CacheFlushInterval) + ticker := time.NewTicker(h.cacheFlushInterval) defer ticker.Stop() for { @@ -476,11 +476,11 @@ func (h *Holder) setFileLimit() { return } // If the soft limit is lower than the FileLimit constant, we will try to change it. - if oldLimit.Cur < FileLimit { - newLimit.Cur = FileLimit + if oldLimit.Cur < fileLimit { + newLimit.Cur = fileLimit // If the hard limit is not high enough, we will try to change it too. - if oldLimit.Max < FileLimit { - newLimit.Max = FileLimit + if oldLimit.Max < fileLimit { + newLimit.Max = fileLimit } else { newLimit.Max = oldLimit.Max } @@ -508,8 +508,8 @@ func (h *Holder) setFileLimit() { if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, oldLimit); err != nil { h.Logger.Printf("ERROR checking open file limit: %s", err) } else { - if oldLimit.Cur < FileLimit { - h.Logger.Printf("WARNING: Tried to set open file limit to %d, but it is %d. You may consider running \"sudo ulimit -n %d\" before starting Pilosa to avoid \"too many open files\" error. See https://www.pilosa.com/docs/administration/#open-file-limits for more information.", FileLimit, oldLimit.Cur, FileLimit) + if oldLimit.Cur < fileLimit { + h.Logger.Printf("WARNING: Tried to set open file limit to %d, but it is %d. You may consider running \"sudo ulimit -n %d\" before starting Pilosa to avoid \"too many open files\" error. See https://www.pilosa.com/docs/administration/#open-file-limits for more information.", fileLimit, oldLimit.Cur, fileLimit) } } } diff --git a/server.go b/server.go index 0e03365ae..93a5dac36 100644 --- a/server.go +++ b/server.go @@ -298,7 +298,7 @@ func NewServer(opts ...ServerOption) (*Server, error) { s.executor.MaxWritesPerRequest = s.maxWritesPerRequest s.cluster.broadcaster = s s.cluster.maxWritesPerRequest = s.maxWritesPerRequest - s.holder.Broadcaster = s + s.holder.broadcaster = s err = s.cluster.setup() if err != nil { @@ -572,8 +572,8 @@ func (s *Server) LocalStatus() (proto.Message, error) { ns := internal.NodeStatus{ Node: EncodeNode(s.cluster.Node), - MaxShards: s.holder.EncodeMaxShards(), - Schema: s.holder.EncodeSchema(), + MaxShards: s.holder.encodeMaxShards(), + Schema: s.holder.encodeSchema(), } return &ns, nil @@ -606,12 +606,12 @@ func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error { } // Sync schema. - if err := s.holder.ApplySchema(ns.Schema); err != nil { + if err := s.holder.applySchema(ns.Schema); err != nil { return errors.Wrap(err, "applying schema") } // Sync maxShards. - oldmaxshards := s.holder.MaxShards() + oldmaxshards := s.holder.maxShards() for index, newMax := range ns.MaxShards.Standard { localIndex := s.holder.Index(index) // if we don't know about an index locally, log an error because diff --git a/utils_internal_test.go b/utils_internal_test.go index f7309961c..ca7bd1fa7 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -360,7 +360,7 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstructi destCluster := t.clusterByID(instrNode.ID) // Sync the schema received in the resize instruction. - if err := destCluster.holder.ApplySchema(instr.Schema); err != nil { + if err := destCluster.holder.applySchema(instr.Schema); err != nil { return err }