From ea77db895a4bb1802e71c32ca254693960e27bf7 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Fri, 29 Jun 2018 14:43:50 -0500 Subject: [PATCH] fix syncholder test and a few bugs The http internal client's FragmentBlocks and Blockdata methods were being used incorrectly, and incorrect respectively. One was not being passed a node URI by monitorAntiEntropy, and the other was always using the defaultURI regardless of what was passed to it. Antientropy was doubling not working because of this. I think this crept in pretty recently, so hasn't actually affected anyone. I exposed a SyncData method on Server so that we can invoke the anti entropy task manually instead of trying to set up the interval so that it will run and then sleeping and waiting for it to run. Now that this test works the way it does, the other anti entropy test is obsolete, and I deleted it. --- fragment.go | 2 +- holder.go | 4 ++ holder_test.go | 118 +++++++++++++++++++------------------------- http/client.go | 5 +- http/client_test.go | 14 ------ server.go | 29 +++++++---- server_test.go | 51 ------------------- 7 files changed, 80 insertions(+), 143 deletions(-) delete mode 100644 server_test.go diff --git a/fragment.go b/fragment.go index 891c1570e..a5a40d68f 100644 --- a/fragment.go +++ b/fragment.go @@ -1752,7 +1752,7 @@ func (s *FragmentSyncer) syncFragment() error { } // Retrieve remote blocks. - blocks, err := s.Cluster.InternalClient.FragmentBlocks(context.Background(), nil, s.Fragment.index, s.Fragment.field, s.Fragment.shard) + blocks, err := s.Cluster.InternalClient.FragmentBlocks(context.Background(), &node.URI, s.Fragment.index, s.Fragment.field, s.Fragment.shard) if err != nil && err != ErrFragmentNotFound { return errors.Wrap(err, "getting blocks") } diff --git a/holder.go b/holder.go index bfb2ae758..eb6eacd57 100644 --- a/holder.go +++ b/holder.go @@ -564,6 +564,8 @@ func (h *Holder) logStartup() error { // 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 { + mu sync.Mutex + Holder *Holder Node *Node @@ -588,6 +590,8 @@ func (s *HolderSyncer) IsClosing() bool { // SyncHolder compares the holder on host with the local holder and resolves differences. 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() // Iterate over schema in sorted order. for _, di := range s.Holder.Schema() { diff --git a/holder_test.go b/holder_test.go index f758798a4..fe6a1a9f8 100644 --- a/holder_test.go +++ b/holder_test.go @@ -24,8 +24,6 @@ import ( "testing" "github.com/pilosa/pilosa" - "github.com/pilosa/pilosa/http" - "github.com/pilosa/pilosa/pql" "github.com/pilosa/pilosa/test" ) @@ -350,49 +348,40 @@ func TestHolder_DeleteIndex(t *testing.T) { // Ensure holder can sync with a remote holder. func TestHolderSyncer_SyncHolder(t *testing.T) { - t.Skip() // Until test.NewServer() works - - s := test.NewServer() - defer s.Close() - - uri, err := pilosa.NewURIFromAddress(s.URL) + c := test.MustNewCluster(t, 2) + c[0].Config.Cluster.ReplicaN = 2 + c[0].Config.AntiEntropy.Interval = 0 + c[1].Config.Cluster.ReplicaN = 2 + c[1].Config.AntiEntropy.Interval = 0 + err := c.Start() if err != nil { - t.Fatal(err) + t.Fatalf("starting cluster: %v", err) + } + defer c.Close() + + _, err = c[0].API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}) + if err != nil { + t.Fatalf("creating index i: %v", err) + } + _, err = c[0].API.CreateIndex(context.Background(), "y", pilosa.IndexOptions{}) + if err != nil { + t.Fatalf("creating index y: %v", err) + } + _, err = c[0].API.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, pilosa.DefaultCacheSize)) + if err != nil { + t.Fatalf("creating field f: %v", err) + } + _, err = c[0].API.CreateField(context.Background(), "i", "f0", pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, pilosa.DefaultCacheSize)) + if err != nil { + t.Fatalf("creating field f0: %v", err) + } + _, err = c[0].API.CreateField(context.Background(), "y", "z", pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, pilosa.DefaultCacheSize)) + if err != nil { + t.Fatalf("creating field z in y: %v", err) } - cluster := test.NewCluster(2) - client := http.GetHTTPClient(nil) - httpClient := http.NewInternalClientFromURI(uri, client) - cluster.InternalClient = httpClient - - // Create a local holder. - hldr0 := test.MustOpenHolder() - defer hldr0.Close() - - // Create a remote holder wrapped by an HTTP - hldr1 := test.MustOpenHolder() - defer hldr1.Close() - s.Handler.API.Holder = hldr1.Holder - s.Handler.Executor.ExecuteFn = func(ctx context.Context, index string, query *pql.Query, shards []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) { - e := pilosa.NewExecutor(pilosa.OptExecutorInternalQueryClient(httpClient)) - e.Holder = hldr1.Holder - e.Node = cluster.Nodes[1] - e.Cluster = cluster - return e.Execute(ctx, index, query, shards, opt) - } - - // Mock 2-node, fully replicated cluster. - cluster.ReplicaN = 2 - - cluster.Nodes[0].URI = pilosa.NewTestURIFromHostPort("localhost", 0) - cluster.Nodes[1].URI = *uri - - // Create fields on nodes. - for _, hldr := range []*test.Holder{hldr0, hldr1} { - hldr.MustCreateFieldIfNotExists("i", "f") - hldr.MustCreateFieldIfNotExists("i", "f0") - hldr.MustCreateFieldIfNotExists("y", "z") - } + hldr0 := &test.Holder{Holder: c[0].Server.Holder()} + hldr1 := &test.Holder{Holder: c[1].Server.Holder()} // Set data on the local holder. hldr0.SetBit("i", "f", 0, 10) @@ -414,42 +403,39 @@ func TestHolderSyncer_SyncHolder(t *testing.T) { hldr1.SetBit("y", "z", 10, (3*ShardWidth)+5) hldr1.SetBit("y", "z", 10, (3*ShardWidth)+7) - // Set highest shard. - hldr0.Index("i").SetRemoteMaxShard(1) - hldr0.Index("y").SetRemoteMaxShard(3) - - // Set up syncer. - syncer := pilosa.HolderSyncer{ - Holder: hldr0.Holder, - Node: cluster.Nodes[0], - Cluster: cluster, - Stats: pilosa.NopStatsClient, + err = c[0].Server.SyncData() + if err != nil { + t.Fatalf("syncing node 0: %v", err) } - - if err := syncer.SyncHolder(); err != nil { - t.Fatal(err) + err = c[1].Server.SyncData() + if err != nil { + t.Fatalf("syncing node 1: %v", err) } // Verify data is the same on both nodes. for i, hldr := range []*test.Holder{hldr0, hldr1} { 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) + t.Errorf("unexpected columns(%d/0): %+v", i, a) + } + if a := hldr.Row("i", "f", 2).Columns(); !reflect.DeepEqual(a, []uint64{20}) { + t.Errorf("unexpected columns(%d/2): %+v", i, a) + } + if a := hldr.Row("i", "f", 3).Columns(); !reflect.DeepEqual(a, []uint64{10}) { + t.Errorf("unexpected columns(%d/3): %+v", i, a) + } + if a := hldr.Row("i", "f", 120).Columns(); !reflect.DeepEqual(a, []uint64{10}) { + t.Errorf("unexpected columns(%d/120): %+v", i, a) + } + if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{4}) { + t.Errorf("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) + t.Errorf("unexpected columns(%d/d/f0): %+v", i, a) } if a := hldr.Row("y", "z", 10).Columns(); !reflect.DeepEqual(a, []uint64{(3 * ShardWidth) + 4, (3 * ShardWidth) + 5, (3 * ShardWidth) + 7}) { - t.Fatalf("unexpected columns(%d/y/z): %+v", i, a) + t.Errorf("unexpected columns(%d/y/z): %+v", i, a) } } } diff --git a/http/client.go b/http/client.go index 59efb4f72..345c572c1 100644 --- a/http/client.go +++ b/http/client.go @@ -717,6 +717,9 @@ func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, in // BlockData returns row/column id pairs for a block. func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index, field string, shard uint64, block int) ([]uint64, []uint64, error) { + if uri == nil { + panic("need to pass a URI to BlockData") + } buf, err := proto.Marshal(&internal.BlockDataRequest{ Index: index, Field: field, @@ -727,7 +730,7 @@ func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index, return nil, nil, errors.Wrap(err, "marshaling") } - u := uriPathToURL(c.defaultURI, "/fragment/block/data") + u := uriPathToURL(uri, "/fragment/block/data") req, err := http.NewRequest("GET", u.String(), bytes.NewReader(buf)) if err != nil { return nil, nil, errors.Wrap(err, "creating request") diff --git a/http/client_test.go b/http/client_test.go index fa9d94e4f..ec849ba10 100644 --- a/http/client_test.go +++ b/http/client_test.go @@ -30,20 +30,6 @@ import ( "github.com/pilosa/pilosa/test" ) -func createCluster(c *pilosa.Cluster) ([]*test.Server, []*test.Holder) { - numNodes := len(c.Nodes) - hldr := make([]*test.Holder, numNodes) - server := make([]*test.Server, numNodes) - for i := 0; i < numNodes; i++ { - hldr[i] = test.MustOpenHolder() - server[i] = test.NewServer() - server[i].Handler.API.Cluster = c - server[i].Handler.API.Cluster.Nodes[i].URI = server[i].HostURI() - server[i].Handler.API.Holder = hldr[i].Holder - } - return server, hldr -} - var defaultClient *gohttp.Client func init() { diff --git a/server.go b/server.go index e1ad60327..d64b9efa7 100644 --- a/server.go +++ b/server.go @@ -70,6 +70,7 @@ type Server struct { diagnosticInterval time.Duration maxWritesPerRequest int isCoordinator bool + syncer HolderSyncer primaryTranslateStore TranslateStore @@ -342,6 +343,12 @@ func (s *Server) Open() error { // buffered channel. s.cluster.listenForJoins() + s.syncer.Holder = s.holder + s.syncer.Node = s.cluster.Node + s.syncer.Cluster = s.cluster + s.syncer.Closing = s.closing + s.syncer.Stats = s.holder.Stats.WithTags("HolderSyncer") + // Start background monitoring. s.wg.Add(3) go func() { defer s.wg.Done(); s.monitorAntiEntropy() }() @@ -384,12 +391,22 @@ func (s *Server) loadNodeID() string { return nodeID } +// SyncData manually invokes the anti entropy process which makes sure that this +// node has the data from all replicas across the cluster. +func (s *Server) SyncData() error { + return errors.Wrap(s.syncer.SyncHolder(), "syncing holder") +} + func (s *Server) monitorAntiEntropy() { + if s.antiEntropyInterval == 0 { + return // anti entropy disabled + } ticker := time.NewTicker(s.antiEntropyInterval) defer ticker.Stop() s.logger.Printf("holder sync monitor initializing (%s interval)", s.antiEntropyInterval) + // Initialize syncer with local holder and remote client. for { // Wait for tick or a close. select { @@ -399,18 +416,10 @@ func (s *Server) monitorAntiEntropy() { s.holder.Stats.Count("AntiEntropy", 1, 1.0) } t := time.Now() - s.logger.Printf("holder sync beginning") - - // Initialize syncer with local holder and remote client. - var syncer HolderSyncer - syncer.Holder = s.holder - syncer.Node = s.cluster.Node - syncer.Cluster = s.cluster - syncer.Closing = s.closing - syncer.Stats = s.holder.Stats.WithTags("HolderSyncer") // Sync holders. - if err := syncer.SyncHolder(); err != nil { + s.logger.Printf("holder sync beginning") + if err := s.syncer.SyncHolder(); err != nil { s.logger.Printf("holder sync error: err=%s", err) continue } diff --git a/server_test.go b/server_test.go deleted file mode 100644 index 04f8a6171..000000000 --- a/server_test.go +++ /dev/null @@ -1,51 +0,0 @@ -// Copyright 2017 Pilosa Corp. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package pilosa_test - -import ( - "context" - "testing" - "time" - - "github.com/pilosa/pilosa" - "github.com/pilosa/pilosa/server" - "github.com/pilosa/pilosa/test" -) - -// TestMonitorAntiEntropy is a regression test which which caught a bug where -// pilosa.Server was not having its remoteClient field set by an option and so -// it was using a nil client in monitorAntiEntropy. -func TestMonitorAntiEntropy(t *testing.T) { - cluster := test.MustRunCluster(t, 3, []server.CommandOption{test.OptAntiEntropyInterval(time.Millisecond * 20)}) - client := cluster[1].Client() - err := client.CreateIndex(context.Background(), "balh", pilosa.IndexOptions{}) - if err != nil { - t.Fatalf("creating index: %v", err) - } - - err = client.CreateField(context.Background(), "balh", "fralh") - if err != nil { - t.Fatalf("creating field: %v", err) - } - - time.Sleep(time.Millisecond * 40) - for _, m := range cluster { - err := m.Close() - if err != nil { - t.Fatal(err) - } - } - -}