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

implement MustNewClusterWithProxy and drop/undrop for partitioning

Browse source
This commit is contained in:
Matt Jaffee 2018-10-26 16:13:20 -05:00
parent af7ece74fd
commit 62c1bfe90e
No known key found for this signature in database
GPG key ID: 08A3DFFF987B11BF
2 changed files with 97 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
server/server_test.go
View file

@ -379,7 +379,11 @@ func (p uint64Slice) Len() int { return len(p) }
func (p uint64Slice) Less(i, j int) bool { return p[i] < p[j] }
func TestClusteringNodesReplica1(t *testing.T) {
cluster := test.MustRunCluster(t, 3)
cluster := test.MustNewCluster(t, 3)
err := cluster.Start()
if err != nil {
t.Fatalf("starting cluster: %v", err)
}
defer cluster.Close()
var wait = true
@ -407,15 +411,11 @@ func TestClusteringNodesReplica1(t *testing.T) {
config.Translation.MapSize = 100000
// this isn't necessary, but makes the test run way faster
fmt.Println("!!!!!!!!!!", config.Gossip.Port)
config.Gossip.Port = strconv.Itoa(int(cluster[2].Command.GossipTransport().URI.Port))
fmt.Println("!!!!!!!!!!!!!!!!", config.Gossip.Port)
cluster[2].Command = server.NewCommand(cluster[2].Stdin, cluster[2].Stdout, cluster[2].Stderr)
cluster[2].Command.Config = config
time.Sleep(time.Second * 40)
// Run new program.
if err := cluster[2].Start(); err != nil {
t.Fatalf("restarting node 2: %v", err)
@ -696,3 +696,12 @@ func TestClusterQueriesAfterRestart(t *testing.T) {
}
// TODO: confirm that things keep working if a node is hard-closed (no nodeLeave event) and immediately restarted with a different address.
func TestClusterPartitioning(t *testing.T) {
cluster := test.MustNewClusterWithProxy(t, 3)
err := cluster.Start()
if err != nil {
t.Fatalf("starting cluster with proxy: %v", err)
}
}

111
test/pilosa.go
View file

@ -31,11 +31,12 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/internal/udproxy"
"github.com/pilosa/pilosa/server"
"github.com/pkg/errors"
"github.com/jaffee/toxiproxy"
tox "github.com/jaffee/toxiproxy/client"
"github.com/Shopify/toxiproxy"
tox "github.com/Shopify/toxiproxy/client"
)
type portAllocator struct {
@ -69,9 +70,49 @@ func init() {
type Command struct {
*server.Command
proxies commandProxies
commandOptions []server.CommandOption
}
// Drop uses the proxies to drop all network traffic to/from this node.
func (com Command) Drop(tb testing.TB) {
if com.proxies.http == nil {
tb.Fatal("can't drop traffic if cluster wasn't created with proxy support.")
}
err := com.proxies.http.Disable()
if err != nil {
tb.Fatalf("disabling http proxy: %v", err)
}
err = com.proxies.memberTCP.Disable()
if err != nil {
tb.Fatalf("disabling memberlist tcp proxy: %v", err)
}
com.proxies.memberUDP.Drop()
}
// Undrop starts forwarding traffic to/from this node after a previous drop request.
func (com Command) Undrop(tb testing.TB) {
if com.proxies.http == nil {
tb.Fatal("can't drop traffic if cluster wasn't created with proxy support.")
}
err := com.proxies.http.Enable()
if err != nil {
tb.Fatalf("enabling http proxy: %v", err)
}
err = com.proxies.memberTCP.Enable()
if err != nil {
tb.Fatalf("enabling memberlist tcp proxy: %v", err)
}
com.proxies.memberUDP.Undrop()
}
type commandProxies struct {
http *tox.Proxy
memberTCP *tox.Proxy
memberUDP *udproxy.Proxy
}
func OptAllowedOrigins(origins []string) server.CommandOption {
return func(m *server.Command) error {
m.Config.Handler.AllowedOrigins = origins
@ -232,7 +273,6 @@ type Cluster []*Command
func (c Cluster) Start() error {
var gossipSeeds = make([]string, len(c))
for i, cc := range c {
cc.Config.Gossip.Port = "0"
cc.Config.Gossip.Seeds = gossipSeeds[:i]
if err := cc.Start(); err != nil {
return errors.Wrapf(err, "starting server %d", i)
@ -254,7 +294,17 @@ func (c Cluster) Close() error {
// MustNewCluster creates a new cluster
func MustNewCluster(tb testing.TB, size int, opts ...[]server.CommandOption) Cluster {
c, err := newCluster(size, opts...)
c, err := newCluster(size, false, opts...)
if err != nil {
tb.Fatalf("new cluster: %v", err)
}
return c
}
// MustNewClusterWithProxy returns a test cluster which has all connections
// going through a proxy to allow for testing network partitions.
func MustNewClusterWithProxy(tb testing.TB, size int, opts ...[]server.CommandOption) Cluster {
c, err := newCluster(size, true, opts...)
if err != nil {
tb.Fatalf("new cluster: %v", err)
}
@ -262,9 +312,7 @@ func MustNewCluster(tb testing.TB, size int, opts ...[]server.CommandOption) Clu
}
// newCluster creates a new cluster
func newCluster(size int, opts ...[]server.CommandOption) (Cluster, error) {
tclient := tox.NewClient(proxy)
func newCluster(size int, withproxy bool, opts ...[]server.CommandOption) (cluster Cluster, err error) {
if size == 0 {
return nil, errors.New("cluster must contain at least one node")
}
@ -272,33 +320,40 @@ func newCluster(size int, opts ...[]server.CommandOption) (Cluster, error) {
return nil, errors.New("Slice of CommandOptions must be of length 0, 1, or equal to the number of cluster nodes")
}
cluster := make(Cluster, size)
cluster = make(Cluster, size)
for i := 0; i < size; i++ {
var commandOpts []server.CommandOption
if len(opts) > 0 {
commandOpts = opts[i%len(opts)]
}
aport := strconv.Itoa(int(ports.Next()))
name := "node" + strconv.Itoa(i)
m := NewCommandNode(i == 0, commandOpts...)
m.Config.Bind = "localhost:" + strconv.Itoa(int(ports.Next()))
m.Config.Advertise = "localhost:" + aport
_, err := tclient.CreateProxy(name+aport, m.Config.Advertise, m.Config.Bind)
if err != nil {
return nil, errors.Wrap(err, "setting up toxiproxy")
}
gossipBindPort := strconv.Itoa(int(ports.Next()))
m.Config.Gossip.Port = gossipBindPort
gossipAdvertPort := strconv.Itoa(int(ports.Next()))
m.Config.Gossip.AdvertisePort = aport
_, err = tclient.CreateProxy(name+"-gossip"+gossipAdvertPort, "localhost:"+m.Config.Gossip.AdvertisePort, "localhost:"+m.Config.Gossip.Port)
if err != nil {
return nil, errors.Wrap(err, "setting up toxiproxy for gossip")
}
_, err = tclient.CreateProxy(name+"-gossipudp"+gossipAdvertPort, "localhost:"+m.Config.Gossip.AdvertisePort, "localhost:"+m.Config.Gossip.Port, tox.CreateWithProtocol("udp"))
if err != nil {
return nil, errors.Wrap(err, "setting up toxiproxy for udp gossip")
gossipBindPort := int(ports.Next())
m.Config.Gossip.Port = strconv.Itoa(gossipBindPort)
if withproxy {
tclient := tox.NewClient(proxy)
aport := strconv.Itoa(int(ports.Next()))
m.Config.Advertise = "localhost:" + aport
m.proxies.http, err = tclient.CreateProxy(name+aport, m.Config.Advertise, m.Config.Bind)
if err != nil {
return nil, errors.Wrap(err, "setting up toxiproxy")
}
gossipAdvertPort := int(ports.Next())
m.Config.Gossip.AdvertisePort = strconv.Itoa(gossipAdvertPort)
m.proxies.memberTCP, err = tclient.CreateProxy(name+"-gossip"+m.Config.Gossip.AdvertisePort, "localhost:"+m.Config.Gossip.AdvertisePort, "localhost:"+m.Config.Gossip.Port)
if err != nil {
return nil, errors.Wrap(err, "setting up toxiproxy for gossip")
}
m.proxies.memberUDP, err = udproxy.New("127.0.0.1", gossipAdvertPort, "127.0.0.1", gossipBindPort)
if err != nil {
return nil, errors.Wrap(err, "setting up proxy for udp gossip")
}
}
err = ioutil.WriteFile(path.Join(m.Config.DataDir, ".id"), []byte(name), 0600)
if err != nil {
return nil, errors.Wrap(err, "writing node id")
@ -310,8 +365,8 @@ func newCluster(size int, opts ...[]server.CommandOption) (Cluster, error) {
}
// runCluster creates and starts a new cluster
func runCluster(size int, opts ...[]server.CommandOption) (Cluster, error) {
cluster, err := newCluster(size, opts...)
func runCluster(size int, withproxy bool, opts ...[]server.CommandOption) (Cluster, error) {
cluster, err := newCluster(size, withproxy, opts...)
if err != nil {
return nil, errors.Wrap(err, "new cluster")
}
@ -323,7 +378,7 @@ func runCluster(size int, opts ...[]server.CommandOption) (Cluster, error) {
// MustRunCluster creates and starts a new cluster
func MustRunCluster(tb testing.TB, size int, opts ...[]server.CommandOption) Cluster {
c, err := runCluster(size, opts...)
c, err := runCluster(size, false, opts...)
if err != nil {
tb.Fatalf("run cluster: %v", err)
}