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) - } - } - -}