Merge pull request #1053 from jaffee/test-cluster-helper

add NewServerCluster(size int) method to pilosa/test

Gopkg.lock

@@ -163,6 +163,12 @@
   revision = "16398bac157da96aa88f98a2df640c7f32af1da2"
   version = "v1.0.1"
 
+[[projects]]
+  name = "github.com/pkg/errors"
+  packages = ["."]
+  revision = "645ef00459ed84a119197bfb8d8205042c6df63d"
+  version = "v0.8.0"
+
 [[projects]]
   name = "github.com/rakyll/statik"
   packages = ["fs"]
@@ -262,6 +268,6 @@
 [solve-meta]
   analyzer-name = "dep"
   analyzer-version = 1
-  inputs-digest = "2e353a12454268d89afe6d06c6021631b577bfd62dede36458f34397ab34fa17"
+  inputs-digest = "d91110a10c830f7a9cc439b9578840d97d9921e84d08242316da8d4a18c68c56"
   solver-name = "gps-cdcl"
   solver-version = 1

gossip/gossip.go

@@ -26,6 +26,7 @@ import (
 	"github.com/hashicorp/memberlist"
 	"github.com/pilosa/pilosa"
 	"github.com/pilosa/pilosa/internal"
+	"github.com/pkg/errors"
 )
 
 // Ensure GossipNodeSet implements interfaces.
@@ -76,7 +77,7 @@ func (g *GossipNodeSet) Open() error {
 	}
 	ml, err := memberlist.Create(g.config.memberlistConfig)
 	if err != nil {
-		return err
+		return errors.Wrap(err, "creating memberlist")
 	}
 	g.memberlist = ml
 	g.broadcasts = &memberlist.TransmitLimitedQueue{
@@ -90,7 +91,7 @@ func (g *GossipNodeSet) Open() error {
 	nodes := []*pilosa.Node{&pilosa.Node{Scheme: "gossip", Host: g.config.gossipSeed}} //TODO: support a list of seeds
 	err = g.joinWithRetry(pilosa.Nodes(nodes).Hosts())
 	if err != nil {
-		return err
+		return errors.Wrap(err, "joinWithRetry")
 	}
 	return nil
 }

test/pilosa.go

@@ -8,41 +8,114 @@ import (
 	"testing"
 
 	"github.com/pilosa/pilosa/server"
+	"github.com/pkg/errors"
 )
 
 func MustNewRunningServer(t *testing.T) *server.Command {
-	s := server.NewCommand(&bytes.Buffer{}, ioutil.Discard, ioutil.Discard)
-	s.Config.Bind = ":0"
-	port := strconv.Itoa(MustOpenPort(t))
-	s.Config.GossipPort = port
-	s.Config.GossipSeed = "localhost:" + port
-	td, err := ioutil.TempDir("", "")
+	s, err := newServer()
 	if err != nil {
-		t.Fatalf("error creating temp data directory: %v", err)
+		t.Fatalf("getting new server: %v", err)
 	}
-	s.Config.DataDir = td
+
 	err = s.Run()
 	if err != nil {
-		t.Fatalf("error running new pilosa server: %v", err)
+		t.Fatalf("running new pilosa server: %v", err)
 	}
 	return s
 }
 
-func MustOpenPort(t *testing.T) int {
+func newServer() (*server.Command, error) {
+	s := server.NewCommand(&bytes.Buffer{}, ioutil.Discard, ioutil.Discard)
+
+	port, err := findPort()
+	if err != nil {
+		return nil, errors.Wrap(err, "getting port")
+	}
+	s.Config.Bind = "localhost:" + strconv.Itoa(port)
+
+	gport, err := findPort()
+	if err != nil {
+		return nil, errors.Wrap(err, "getting gossip port")
+	}
+	s.Config.GossipPort = strconv.Itoa(gport)
+
+	s.Config.GossipSeed = "localhost:" + s.Config.GossipPort
+	s.Config.Cluster.Type = "gossip"
+	s.Config.Metric.Diagnostics = false
+	td, err := ioutil.TempDir("", "")
+	if err != nil {
+		return nil, errors.Wrap(err, "temp dir")
+	}
+	s.Config.DataDir = td
+	return s, nil
+}
+
+func findPort() (int, error) {
 	addr, err := net.ResolveTCPAddr("tcp", ":0")
 	if err != nil {
-		t.Fatalf("resolving new port addr: %v", err)
+		return 0, errors.Wrap(err, "resolving new port addr")
 	}
-
 	l, err := net.ListenTCP("tcp", addr)
 	if err != nil {
-		t.Fatalf("listening to get new port: %v", err)
+		return 0, errors.Wrap(err, "listening to get new port")
 	}
-	defer func() {
-		err := l.Close()
-		if err != nil {
-			t.Logf("error closing listener in MustOpenPort: %v", err)
-		}
-	}()
-	return l.Addr().(*net.TCPAddr).Port
+	port := l.Addr().(*net.TCPAddr).Port
+	err = l.Close()
+	if err != nil {
+		return port, errors.Wrap(err, "closing listener")
+	}
+	return port, nil
+
+}
+
+func MustFindPort(t *testing.T) int {
+	port, err := findPort()
+	if err != nil {
+		t.Fatalf("allocating new port: %v", err)
+	}
+	return port
+}
+
+type Cluster struct {
+	Servers []*server.Command
+}
+
+func MustNewServerCluster(t *testing.T, size int) *Cluster {
+	cluster, err := NewServerCluster(size)
+	if err != nil {
+		t.Fatalf("new cluster: %v", err)
+	}
+	return cluster
+}
+
+func NewServerCluster(size int) (cluster *Cluster, err error) {
+	cluster = &Cluster{
+		Servers: make([]*server.Command, size),
+	}
+	hosts := make([]string, size)
+	for i := 0; i < size; i++ {
+		s, err := newServer()
+		if err != nil {
+			return nil, errors.Wrap(err, "new server")
+		}
+		cluster.Servers[i] = s
+		hosts[i] = s.Config.Bind
+		s.Config.GossipSeed = cluster.Servers[0].Config.GossipSeed
+
+	}
+
+	for _, s := range cluster.Servers {
+		s.Config.Cluster.Hosts = hosts
+	}
+	for i, s := range cluster.Servers {
+		err := s.Run()
+		if err != nil {
+			for j := 0; j <= i; j++ {
+				cluster.Servers[j].Close()
+			}
+			return nil, errors.Wrapf(err, "starting server %d of %d. Config: %#v", i+1, size, s.Config)
+		}
+	}
+
+	return cluster, nil
 }

test/pilosa_test.go

@@ -0,0 +1,48 @@
+package test_test
+
+import (
+	"net/http"
+	"testing"
+
+	"encoding/json"
+
+	"github.com/pilosa/pilosa/test"
+)
+
+func TestNewCluster(t *testing.T) {
+	cluster := test.MustNewServerCluster(t, 3)
+	response, err := http.Get("http://" + cluster.Servers[0].Server.Addr().String() + "/status")
+	if err != nil {
+		t.Fatalf("getting schema: %v", err)
+	}
+	dec := json.NewDecoder(response.Body)
+	body := struct {
+		Status struct {
+			Nodes []struct {
+				Host   string
+				Schema string
+				State  string
+			}
+		}
+	}{}
+
+	err = dec.Decode(&body)
+	if err != nil {
+		t.Fatalf("decoding status response: %v", err)
+	}
+
+	bytes, err := json.MarshalIndent(body, "", "  ")
+	if err != nil {
+		t.Fatalf("encoding: %v", err)
+	}
+
+	if len(body.Status.Nodes) != 3 {
+		t.Fatalf("wrong number of nodes in status: %s", bytes)
+	}
+
+	for i, node := range body.Status.Nodes {
+		if node.State != "UP" {
+			t.Fatalf("node %d should be up but is %s", i, node.State)
+		}
+	}
+}