Merge pull request #1233 from jaffee/nil-client-bug

fix nil client bug in monitorAntiEntropy (and test)

broadcast_test.go

@@ -57,6 +57,7 @@ func testMessageMarshal(t *testing.T, m proto.Message) {
 // Ensure that BroadcastReceiver can register a BroadcastHandler.
 func TestBroadcast_BroadcastReceiver(t *testing.T) {
 	com := server.NewCommand(bytes.NewBuffer([]byte{}), ioutil.Discard, ioutil.Discard)
+	com.Config.Bind = "localhost:0"
 	err := com.SetupServer() // this test shouldn't need to import pilosa/server just to set up the Server, but it really shouldn't need to setup the Server at all. The Server should not be the implementation of Broadcast* TODO
 	if err != nil {
 		t.Fatalf("setting up server: %v", err)

server.go

@@ -163,6 +163,7 @@ func OptServerGCNotifier(gcn GCNotifier) ServerOption {
 func OptServerRemoteClient(c *http.Client) ServerOption {
 	return func(s *Server) error {
 		s.executor = NewExecutor(c)
+		s.remoteClient = c
 		s.defaultClient = NewInternalHTTPClientFromURI(nil, c)
 		s.Cluster.RemoteClient = c
 		return nil

server_test.go

@@ -0,0 +1,35 @@
+package pilosa_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/pilosa/pilosa"
+	"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.MustRunMainWithCluster(t, 3, test.OptAntiEntropyInterval(time.Millisecond*1))
+	client := cluster[1].Client()
+	err := client.CreateIndex(context.Background(), "balh", pilosa.IndexOptions{})
+	if err != nil {
+		t.Fatalf("creating index: %v", err)
+	}
+	err = client.CreateFrame(context.Background(), "balh", "fralh", pilosa.FrameOptions{})
+	if err != nil {
+		t.Fatalf("creating frame: %v", err)
+	}
+
+	time.Sleep(time.Millisecond * 2)
+	for _, m := range cluster {
+		err := m.Close()
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+}

test/pilosa.go

@@ -23,11 +23,13 @@ import (
 	"os"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/pilosa/pilosa"
 	"github.com/pilosa/pilosa/boltdb"
 	"github.com/pilosa/pilosa/gossip"
 	"github.com/pilosa/pilosa/server"
+	"github.com/pilosa/pilosa/toml"
 	"github.com/pkg/errors"
 )
 
@@ -41,8 +43,17 @@ type Main struct {
 	Stderr bytes.Buffer
 }
 
+type MainOpt func(m *Main) error
+
+func OptAntiEntropyInterval(dur time.Duration) MainOpt {
+	return func(m *Main) error {
+		m.Command.Config.AntiEntropy.Interval = toml.Duration(dur)
+		return nil
+	}
+}
+
 // NewMain returns a new instance of Main with a temporary data directory and random port.
-func NewMain() *Main {
+func NewMain(opts ...MainOpt) *Main {
 	path, err := ioutil.TempDir("", "pilosa-")
 	if err != nil {
 		panic(err)
@@ -55,6 +66,15 @@ func NewMain() *Main {
 	m.Command.Stdin = &m.Stdin
 	m.Command.Stdout = &m.Stdout
 	m.Command.Stderr = &m.Stderr
+
+	for _, opt := range opts {
+		err := opt(m)
+		if err != nil {
+			panic(err)
+		}
+
+	}
+
 	m.SetupServer()
 
 	if testing.Verbose() {
@@ -66,8 +86,8 @@ func NewMain() *Main {
 }
 
 // NewMainWithCluster returns a new instance of Main with clustering enabled.
-func NewMainWithCluster(isCoordinator bool) *Main {
-	m := NewMain()
+func NewMainWithCluster(isCoordinator bool, opts ...MainOpt) *Main {
+	m := NewMain(opts...)
 	m.Config.Cluster.Disabled = false
 	m.Config.Cluster.Coordinator = isCoordinator
 	return m
@@ -75,8 +95,8 @@ func NewMainWithCluster(isCoordinator bool) *Main {
 
 // MustRunMainWithCluster ruturns a running array of *Main where
 // all nodes are joined via memberlist (i.e. clustering enabled).
-func MustRunMainWithCluster(t *testing.T, size int) []*Main {
-	ma, err := runMainWithCluster(size)
+func MustRunMainWithCluster(t *testing.T, size int, opts ...MainOpt) []*Main {
+	ma, err := runMainWithCluster(size, opts...)
 	if err != nil {
 		t.Fatalf("new main array with cluster: %v", err)
 	}
@@ -85,7 +105,7 @@ func MustRunMainWithCluster(t *testing.T, size int) []*Main {
 
 // runMainWithCluster runs an array of *Main where all nodes are
 // joined via memberlist (i.e. clustering enabled).
-func runMainWithCluster(size int) ([]*Main, error) {
+func runMainWithCluster(size int, opts ...MainOpt) ([]*Main, error) {
 	if size == 0 {
 		return nil, errors.New("cluster must contain at least one node")
 	}
@@ -98,7 +118,7 @@ func runMainWithCluster(size int) ([]*Main, error) {
 	var gossipSeeds = make([]string, size)
 
 	for i := 0; i < size; i++ {
-		m := NewMainWithCluster(i == 0)
+		m := NewMainWithCluster(i == 0, opts...)
 		m.Config.Cluster.Disabled = false
 
 		gossipSeeds[i], err = m.RunWithTransport(gossipHost, gossipPort, gossipSeeds[:i])