scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-05 08:10:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge remote-tracking branch 'upstream/cluster-resize' into cluster-resize

Browse source
This commit is contained in:
Travis Turner 2018-01-25 09:35:29 -06:00
commit 267cb05039
No known key found for this signature in database
GPG key ID: 7F08008DFD9314C9
7 changed files with 195 additions and 292 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
cluster.go
View file

@ -175,7 +175,7 @@ type Cluster struct {
// Required for cluster Resize.
Static bool // Static is primarily used for testing in a non-gossip environment.
State string
state string
Coordinator URI
Holder *Holder
Broadcaster Broadcaster
@ -302,9 +302,21 @@ func (c *Cluster) setID(id string) {
c.Topology.ClusterID = c.ID
}
func (c *Cluster) State() string {
c.mu.RLock()
defer c.mu.RUnlock()
return c.state
}
func (c *Cluster) SetState(state string) {
c.mu.Lock()
defer c.mu.Unlock()
c.setState(state)
}
func (c *Cluster) setState(state string) {
// Ignore cases where the state hasn't changed.
if state == c.State {
if state == c.state {
return
}
@ -321,12 +333,12 @@ func (c *Cluster) setState(state string) {
// - ClusterStateStarting
// If state is RESIZING -> NORMAL then run cleanup.
if c.State == ClusterStateResizing {
if c.state == ClusterStateResizing {
doCleanup = true
}
}
c.State = state
c.state = state
// TODO: consider NOT running cleanup on an active node that has
// been removed.
@ -373,7 +385,7 @@ func (c *Cluster) ReceiveNodeState(uri URI, state string) error {
}
// This method is really only useful during initial startup.
if c.State != ClusterStateStarting {
if c.State() != ClusterStateStarting {
return nil
}
@ -397,7 +409,7 @@ func (c *Cluster) ReceiveNodeState(uri URI, state string) error {
func (c *Cluster) Status() *internal.ClusterStatus {
return &internal.ClusterStatus{
ClusterID: c.ID,
State: c.State,
State: c.state,
NodeSet: encodeURIs(c.NodeSet()),
}
}
@ -763,7 +775,7 @@ func (h *jmphasher) Hash(key uint64, n int) int {
func (c *Cluster) Open() error {
// Cluster always comes up in state STARTING until cluster membership is determined.
c.State = ClusterStateStarting
c.state = ClusterStateStarting
// Load topology file if it exists.
if err := c.loadTopology(); err != nil {
@ -820,7 +832,7 @@ func (c *Cluster) markAsJoined() {
}
func (c *Cluster) needTopologyAgreement() bool {
return c.State == ClusterStateStarting && !URISlicesAreEqual(c.Topology.NodeSet, c.NodeSet())
return c.State() == ClusterStateStarting && !URISlicesAreEqual(c.Topology.NodeSet, c.NodeSet())
}
func (c *Cluster) haveTopologyAgreement() bool {
@ -886,7 +898,7 @@ func (c *Cluster) handleNodeAction(nodeAction nodeAction) error {
}
func (c *Cluster) setStateAndBroadcast(state string) error {
c.setState(state)
c.SetState(state)
// Broadcast cluster status changes to the cluster.
c.logger().Printf("broadcasting ClusterStatus: %s", state)
return c.Broadcaster.SendSync(c.Status())
@ -1618,7 +1630,7 @@ func (c *Cluster) NodeLeave(uri URI) error {
return fmt.Errorf("Node removal requests are only valid on the Coordinator node: %s", c.Coordinator)
}
if c.State != ClusterStateNormal {
if c.State() != ClusterStateNormal {
return fmt.Errorf("Cluster must be in state %s to remove a node. Current state: %s", ClusterStateNormal, c.State)
}
@ -1683,7 +1695,7 @@ func (c *Cluster) MergeClusterStatus(cs *internal.ClusterStatus) error {
}
}
c.setState(cs.State)
c.SetState(cs.State)
c.markAsJoined()

36
cluster_test.go
View file

@ -255,8 +255,8 @@ func TestCluster_ResizeStates(t *testing.T) {
node := tc.Clusters[0]
// Ensure that node comes up in state NORMAL.
if node.State != pilosa.ClusterStateNormal {
t.Errorf("expected state: %v, but got: %v", pilosa.ClusterStateNormal, node.State)
if node.State() != pilosa.ClusterStateNormal {
t.Errorf("expected state: %v, but got: %v", pilosa.ClusterStateNormal, node.State())
}
expectedTop := &pilosa.Topology{
@ -292,8 +292,8 @@ func TestCluster_ResizeStates(t *testing.T) {
}
// Ensure that node comes up in state NORMAL.
if node.State != pilosa.ClusterStateNormal {
t.Errorf("expected state: %v, but got: %v", pilosa.ClusterStateNormal, node.State)
if node.State() != pilosa.ClusterStateNormal {
t.Errorf("expected state: %v, but got: %v", pilosa.ClusterStateNormal, node.State())
}
// Close TestCluster.
@ -344,10 +344,10 @@ func TestCluster_ResizeStates(t *testing.T) {
node1 := tc.Clusters[1]
// Ensure that nodes comes up in state NORMAL.
if node0.State != pilosa.ClusterStateNormal {
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State)
} else if node1.State != pilosa.ClusterStateNormal {
t.Errorf("expected node1 state: %v, but got: %v", pilosa.ClusterStateNormal, node1.State)
if node0.State() != pilosa.ClusterStateNormal {
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State())
} else if node1.State() != pilosa.ClusterStateNormal {
t.Errorf("expected node1 state: %v, but got: %v", pilosa.ClusterStateNormal, node1.State())
}
expectedTop := &pilosa.Topology{
@ -388,8 +388,8 @@ func TestCluster_ResizeStates(t *testing.T) {
}
// Ensure that node is in state STARTING before the other node joins.
if node0.State != pilosa.ClusterStateStarting {
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateStarting, node0.State)
if node0.State() != pilosa.ClusterStateStarting {
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateStarting, node0.State())
}
// Expect an error by adding a node not in the topology.
@ -403,10 +403,10 @@ func TestCluster_ResizeStates(t *testing.T) {
node2 := tc.Clusters[2]
// Ensure that node comes up in state NORMAL.
if node0.State != pilosa.ClusterStateNormal {
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State)
} else if node2.State != pilosa.ClusterStateNormal {
t.Errorf("expected node1 state: %v, but got: %v", pilosa.ClusterStateNormal, node2.State)
if node0.State() != pilosa.ClusterStateNormal {
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State())
} else if node2.State() != pilosa.ClusterStateNormal {
t.Errorf("expected node1 state: %v, but got: %v", pilosa.ClusterStateNormal, node2.State())
}
// Close TestCluster.
@ -470,10 +470,10 @@ func TestCluster_ResizeStates(t *testing.T) {
node1 := tc.Clusters[1]
// Ensure that nodes come up in state NORMAL.
if node0.State != pilosa.ClusterStateNormal {
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State)
} else if node1.State != pilosa.ClusterStateNormal {
t.Errorf("expected node1 state: %v, but got: %v", pilosa.ClusterStateNormal, node1.State)
if node0.State() != pilosa.ClusterStateNormal {
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State())
} else if node1.State() != pilosa.ClusterStateNormal {
t.Errorf("expected node1 state: %v, but got: %v", pilosa.ClusterStateNormal, node1.State())
}
expectedTop := &pilosa.Topology{

2
server.go
View file

@ -510,7 +510,7 @@ func (s *Server) ClusterStatus() (proto.Message, error) {
// HandleRemoteStatus receives incoming NodeStatus from remote nodes.
func (s *Server) HandleRemoteStatus(pb proto.Message) error {
// Ignore NodeStatus messages until the cluster is in a Normal state.
if s.Cluster.State != ClusterStateNormal {
if s.Cluster.State() != ClusterStateNormal {
return nil
}

71
server/cluster_test.go
View file

@ -24,15 +24,16 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/gossip"
"github.com/pilosa/pilosa/test"
)
// Ensure program can send/receive broadcast messages.
func TestMain_SendReceiveMessage(t *testing.T) {
m0 := MustRunMain()
m0 := test.MustRunMain()
defer m0.Close()
m1 := MustRunMain()
m1 := test.MustRunMain()
defer m1.Close()
// Update cluster config
@ -205,18 +206,18 @@ func TestMain_SendReceiveMessage(t *testing.T) {
// Ensure that an empty node comes up in a NORMAL state.
func TestClusterResize_EmptyNode(t *testing.T) {
m0 := MustRunMain()
m0 := test.MustRunMain()
defer m0.Close()
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
t.Fatalf("unexpected cluster state: %s", m0.Server.Cluster.State)
if m0.Server.Cluster.State() != pilosa.ClusterStateNormal {
t.Fatalf("unexpected cluster state: %s", m0.Server.Cluster.State())
}
}
// Ensure that a cluster of empty nodes comes up in a NORMAL state.
func TestClusterResize_EmptyNodes(t *testing.T) {
// Configure node0
m0 := NewMain()
m0 := test.NewMain()
defer m0.Close()
gossipHost := "localhost"
@ -227,7 +228,7 @@ func TestClusterResize_EmptyNodes(t *testing.T) {
}
// Configure node1
m1 := NewMain()
m1 := test.NewMain()
defer m1.Close()
seed, coord, err = m1.RunWithTransport(gossipHost, gossipPort, seed, &coord)
@ -235,10 +236,10 @@ func TestClusterResize_EmptyNodes(t *testing.T) {
t.Fatal(err)
}
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State)
} else if m1.Server.Cluster.State != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State)
if m0.Server.Cluster.State() != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State())
} else if m1.Server.Cluster.State() != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State())
}
}
@ -246,7 +247,7 @@ func TestClusterResize_EmptyNodes(t *testing.T) {
func TestClusterResize_AddNode(t *testing.T) {
t.Run("NoData", func(t *testing.T) {
// Configure node0
m0 := NewMain()
m0 := test.NewMain()
defer m0.Close()
seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
@ -255,7 +256,7 @@ func TestClusterResize_AddNode(t *testing.T) {
}
// Configure node1
m1 := NewMain()
m1 := test.NewMain()
defer m1.Close()
var eg errgroup.Group
@ -272,15 +273,15 @@ func TestClusterResize_AddNode(t *testing.T) {
time.Sleep(1 * time.Second)
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State)
} else if m1.Server.Cluster.State != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State)
if m0.Server.Cluster.State() != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State())
} else if m1.Server.Cluster.State() != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State())
}
})
t.Run("WithIndex", func(t *testing.T) {
// Configure node0
m0 := NewMain()
m0 := test.NewMain()
defer m0.Close()
seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
@ -299,7 +300,7 @@ func TestClusterResize_AddNode(t *testing.T) {
}
// Configure node1
m1 := NewMain()
m1 := test.NewMain()
defer m1.Close()
var eg errgroup.Group
@ -317,16 +318,16 @@ func TestClusterResize_AddNode(t *testing.T) {
// Give the cluster time to settle.
time.Sleep(1 * time.Second)
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State)
} else if m1.Server.Cluster.State != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State)
if m0.Server.Cluster.State() != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State())
} else if m1.Server.Cluster.State() != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State())
}
})
t.Run("ContinuousSlices", func(t *testing.T) {
// Configure node0
m0 := NewMain()
m0 := test.NewMain()
defer m0.Close()
seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
@ -354,7 +355,7 @@ func TestClusterResize_AddNode(t *testing.T) {
}
// Configure node1
m1 := NewMain()
m1 := test.NewMain()
defer m1.Close()
var eg errgroup.Group
@ -372,16 +373,16 @@ func TestClusterResize_AddNode(t *testing.T) {
// Give the cluster time to settle.
time.Sleep(1 * time.Second)
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State)
} else if m1.Server.Cluster.State != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State)
if m0.Server.Cluster.State() != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State())
} else if m1.Server.Cluster.State() != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State())
}
})
t.Run("SkippedSlice", func(t *testing.T) {
// Configure node0
m0 := NewMain()
m0 := test.NewMain()
defer m0.Close()
seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
@ -409,7 +410,7 @@ func TestClusterResize_AddNode(t *testing.T) {
}
// Configure node1
m1 := NewMain()
m1 := test.NewMain()
defer m1.Close()
var eg errgroup.Group
@ -427,10 +428,10 @@ func TestClusterResize_AddNode(t *testing.T) {
// Give the cluster time to settle.
time.Sleep(1 * time.Second)
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State)
} else if m1.Server.Cluster.State != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State)
if m0.Server.Cluster.State() != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State())
} else if m1.Server.Cluster.State() != pilosa.ClusterStateNormal {
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State())
}
})
}

230
server/server_test.go
View file

@ -15,26 +15,19 @@
package server_test
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"io/ioutil"
"math/rand"
"net/http"
"os"
"reflect"
"runtime"
"sort"
"strings"
"testing"
"testing/quick"
"github.com/BurntSushi/toml"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/gossip"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
)
@ -45,7 +38,7 @@ func TestMain_Set_Quick(t *testing.T) {
}
if err := quick.Check(func(cmds []SetCommand) bool {
m := MustRunMain()
m := test.MustRunMain()
defer m.Close()
// Create client.
@ -121,7 +114,7 @@ func TestMain_Set_Quick(t *testing.T) {
// Ensure program can set row attributes and retrieve them.
func TestMain_SetRowAttrs(t *testing.T) {
m := MustRunMain()
m := test.MustRunMain()
defer m.Close()
// Create frames.
@ -198,7 +191,7 @@ func TestMain_SetRowAttrs(t *testing.T) {
// Ensure program can set column attributes and retrieve them.
func TestMain_SetColumnAttrs(t *testing.T) {
m := MustRunMain()
m := test.MustRunMain()
defer m.Close()
// Create frames.
@ -242,7 +235,7 @@ func TestMain_SetColumnAttrs(t *testing.T) {
// Ensure program can set column attributes with columnLabel option.
func TestMain_SetColumnAttrsWithColumnOption(t *testing.T) {
m := MustRunMain()
m := test.MustRunMain()
defer m.Close()
// Create frames.
@ -276,7 +269,7 @@ func TestMain_SetColumnAttrsWithColumnOption(t *testing.T) {
// Ensure program can set bits on one cluster and then restore to a second cluster.
func TestMain_FrameRestore(t *testing.T) {
mains1 := NewMainArrayWithCluster(2)
mains1 := test.NewMainArrayWithCluster(2)
m0 := mains1[0]
// Create frames.
@ -309,7 +302,7 @@ func TestMain_FrameRestore(t *testing.T) {
}
// Start second cluster.
mains2 := NewMainArrayWithCluster(2)
mains2 := test.NewMainArrayWithCluster(2)
m2 := mains2[0]
defer m2.Close()
@ -378,191 +371,6 @@ func TestCountOpenFiles(t *testing.T) {
}
}
// Main represents a test wrapper for main.Main.
type Main struct {
*server.Command
Stdin bytes.Buffer
Stdout bytes.Buffer
Stderr bytes.Buffer
}
// NewMain returns a new instance of Main with a temporary data directory and random port.
func NewMain() *Main {
path, err := ioutil.TempDir("", "pilosa-")
if err != nil {
panic(err)
}
m := &Main{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr)}
m.Server.Network = *test.Network
m.Config.DataDir = path
m.Config.Bind = "localhost:0"
m.Config.Cluster.Type = "static"
m.Command.Stdin = &m.Stdin
m.Command.Stdout = &m.Stdout
m.Command.Stderr = &m.Stderr
if testing.Verbose() {
m.Command.Stdout = io.MultiWriter(os.Stdout, m.Command.Stdout)
m.Command.Stderr = io.MultiWriter(os.Stderr, m.Command.Stderr)
}
return m
}
func NewMainArrayWithCluster(size int) []*Main {
cluster, err := test.NewServerCluster(size)
if err != nil {
panic(err)
}
mainArray := make([]*Main, size)
for i := 0; i < size; i++ {
mainArray[i] = &Main{Command: cluster.Servers[i]}
}
return mainArray
}
// MustRunMain returns a new, running Main. Panic on error.
func MustRunMain() *Main {
m := NewMain()
m.Config.Metric.Diagnostics = false // Disable diagnostics.
if err := m.Run(); err != nil {
panic(err)
}
return m
}
// Close closes the program and removes the underlying data directory.
func (m *Main) Close() error {
defer os.RemoveAll(m.Config.DataDir)
return m.Command.Close()
}
// Reopen closes the program and reopens it.
func (m *Main) Reopen() error {
if err := m.Command.Close(); err != nil {
return err
}
// Create new main with the same config.
config := m.Config
m.Command = server.NewCommand(os.Stdin, os.Stdout, os.Stderr)
m.Server.Network = *test.Network
m.Config = config
// Run new program.
if err := m.Run(); err != nil {
return err
}
return nil
}
// RunWithTransport runs Main and returns the dynamically allocated gossip port.
func (m *Main) RunWithTransport(host string, bindPort int, joinSeed string, coordinator *pilosa.URI) (seed string, coord pilosa.URI, err error) {
defer close(m.Started)
m.Config.Cluster.Type = "gossip"
/*
TEST:
- SetupServer (just static settings from config)
- OpenListener (sets Server.Name to use in gossip)
- NewTransport (gossip)
- SetupNetworking (does the gossip or static stuff) - uses Server.Name
- Open server
PRODUCTION:
- SetupServer (just static settings from config)
- SetupNetworking (does the gossip or static stuff) - calls NewTransport
- Open server - calls OpenListener
*/
// SetupServer
err = m.SetupServer()
if err != nil {
return seed, coord, err
}
// Open server listener.
// This is used to set Server.Name, which is used as the node
// name for identifying a memberlist node.
err = m.Server.OpenListener()
if err != nil {
return seed, coord, err
}
// Open gossip transport to use in SetupServer.
transport, err := gossip.NewTransport(host, bindPort)
if err != nil {
return seed, coord, err
}
m.GossipTransport = transport
if joinSeed != "" {
m.Config.Gossip.Seed = joinSeed
} else {
m.Config.Gossip.Seed = transport.URI.String()
}
seed = m.Config.Gossip.Seed
// SetupNetworking
err = m.SetupNetworking()
if err != nil {
return seed, coord, err
}
if err = m.Server.BroadcastReceiver.Start(m.Server); err != nil {
return seed, coord, err
}
if coordinator != nil {
coord = *coordinator
} else {
coord = m.Server.URI
}
m.Server.Cluster.Coordinator = coord
m.Server.Cluster.Static = false
// Initialize server.
err = m.Server.Open()
if err != nil {
return seed, coord, err
}
return seed, coord, nil
}
// URL returns the base URL string for accessing the running program.
func (m *Main) URL() string { return "http://" + m.Server.Addr().String() }
// Client returns a client to connect to the program.
func (m *Main) Client() *pilosa.InternalHTTPClient {
client, err := pilosa.NewInternalHTTPClient(m.Server.URI.HostPort(), pilosa.GetHTTPClient(nil))
if err != nil {
panic(err)
}
return client
}
// Query executes a query against the program through the HTTP API.
func (m *Main) Query(index, rawQuery, query string) (string, error) {
resp := MustDo("POST", m.URL()+fmt.Sprintf("/index/%s/query?", index)+rawQuery, query)
if resp.StatusCode != http.StatusOK {
return "", fmt.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
}
return resp.Body, nil
}
// CreateDefinition.
func (m *Main) CreateDefinition(index, def, query string) (string, error) {
resp := MustDo("POST", m.URL()+fmt.Sprintf("/index/%s/input-definition/%s", index, def), query)
if resp.StatusCode != http.StatusOK {
return "", fmt.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
}
return resp.Body, nil
}
// SetCommand represents a command to set a bit.
type SetCommand struct {
ID uint64
@ -619,32 +427,6 @@ func ParseConfig(s string) (pilosa.Config, error) {
return c, err
}
// MustDo executes http.Do() with an http.NewRequest(). Panic on error.
func MustDo(method, urlStr string, body string) *httpResponse {
req, err := http.NewRequest(method, urlStr, strings.NewReader(body))
if err != nil {
panic(err)
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
panic(err)
}
defer resp.Body.Close()
buf, err := ioutil.ReadAll(resp.Body)
if err != nil {
panic(err)
}
return &httpResponse{Response: resp, Body: string(buf)}
}
// httpResponse is a wrapper for http.Response that holds the Body as a string.
type httpResponse struct {
*http.Response
Body string
}
// MustMarshalJSON marshals v into a string. Panic on error.
func MustMarshalJSON(v interface{}) string {
buf, err := json.Marshal(v)

14
test/cluster.go
View file

@ -7,6 +7,7 @@ import (
"io/ioutil"
"path/filepath"
"sort"
"sync"
"time"
"github.com/gogo/protobuf/proto"
@ -77,6 +78,8 @@ type TestCluster struct {
common *commonClusterSettings
mu sync.RWMutex
resizing bool
resizeDone chan struct{}
}
@ -184,8 +187,11 @@ func (t *TestCluster) AddNode(saveTopology bool) error {
}
// Wait for the AddNode job to finish.
if c.State != pilosa.ClusterStateNormal {
if c.State() != pilosa.ClusterStateNormal {
t.resizeDone = make(chan struct{})
t.mu.Lock()
t.resizing = true
t.mu.Unlock()
<-t.resizeDone
}
}
@ -266,7 +272,7 @@ func NewTestCluster(n int) *TestCluster {
// SetState sets the state of the cluster on each node.
func (t *TestCluster) SetState(state string) {
for _, c := range t.Clusters {
c.State = state
c.SetState(state)
}
}
@ -314,9 +320,11 @@ func (t *TestCluster) SendSync(pb proto.Message) error {
for _, c := range t.Clusters {
c.MergeClusterStatus(obj)
}
if obj.State == pilosa.ClusterStateNormal && t.resizeDone != nil {
t.mu.RLock()
if obj.State == pilosa.ClusterStateNormal && t.resizing {
close(t.resizeDone)
}
t.mu.RUnlock()
}
return nil

100
test/pilosa.go
View file

@ -2,9 +2,12 @@ package test
import (
"bytes"
"fmt"
"io"
"io/ioutil"
"net/http"
"os"
"strings"
"testing"
"github.com/pilosa/pilosa"
@ -47,12 +50,53 @@ func NewMain() *Main {
return m
}
func NewMainArrayWithCluster(size int) []*Main {
cluster, err := NewServerCluster(size)
if err != nil {
panic(err)
}
mainArray := make([]*Main, size)
for i := 0; i < size; i++ {
mainArray[i] = cluster.Servers[i]
}
return mainArray
}
// MustRunMain returns a new, running Main. Panic on error.
func MustRunMain() *Main {
m := NewMain()
m.Config.Metric.Diagnostics = false // Disable diagnostics.
if err := m.Run(); err != nil {
panic(err)
}
return m
}
// Close closes the program and removes the underlying data directory.
func (m *Main) Close() error {
defer os.RemoveAll(m.Config.DataDir)
return m.Command.Close()
}
// Reopen closes the program and reopens it.
func (m *Main) Reopen() error {
if err := m.Command.Close(); err != nil {
return err
}
// Create new main with the same config.
config := m.Config
m.Command = server.NewCommand(os.Stdin, os.Stdout, os.Stderr)
m.Server.Network = *Network
m.Config = config
// Run new program.
if err := m.Run(); err != nil {
return err
}
return nil
}
// RunWithTransport runs Main and returns the dynamically allocated gossip port.
func (m *Main) RunWithTransport(host string, bindPort int, joinSeed string, coordinator *pilosa.URI) (seed string, coord pilosa.URI, err error) {
defer close(m.Started)
@ -128,6 +172,36 @@ func (m *Main) RunWithTransport(host string, bindPort int, joinSeed string, coor
return seed, coord, nil
}
// URL returns the base URL string for accessing the running program.
func (m *Main) URL() string { return "http://" + m.Server.Addr().String() }
// Client returns a client to connect to the program.
func (m *Main) Client() *pilosa.InternalHTTPClient {
client, err := pilosa.NewInternalHTTPClient(m.Server.URI.HostPort(), pilosa.GetHTTPClient(nil))
if err != nil {
panic(err)
}
return client
}
// Query executes a query against the program through the HTTP API.
func (m *Main) Query(index, rawQuery, query string) (string, error) {
resp := MustDo("POST", m.URL()+fmt.Sprintf("/index/%s/query?", index)+rawQuery, query)
if resp.StatusCode != http.StatusOK {
return "", fmt.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
}
return resp.Body, nil
}
// CreateDefinition.
func (m *Main) CreateDefinition(index, def, query string) (string, error) {
resp := MustDo("POST", m.URL()+fmt.Sprintf("/index/%s/input-definition/%s", index, def), query)
if resp.StatusCode != http.StatusOK {
return "", fmt.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
}
return resp.Body, nil
}
////////////////////////////////////////////////////////////////////////////////////
type Cluster struct {
@ -169,3 +243,29 @@ func NewServerCluster(size int) (cluster *Cluster, err error) {
return cluster, nil
}
// MustDo executes http.Do() with an http.NewRequest(). Panic on error.
func MustDo(method, urlStr string, body string) *httpResponse {
req, err := http.NewRequest(method, urlStr, strings.NewReader(body))
if err != nil {
panic(err)
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
panic(err)
}
defer resp.Body.Close()
buf, err := ioutil.ReadAll(resp.Body)
if err != nil {
panic(err)
}
return &httpResponse{Response: resp, Body: string(buf)}
}
// httpResponse is a wrapper for http.Response that holds the Body as a string.
type httpResponse struct {
*http.Response
Body string
}