From 62c1bfe90ecfd2b074f02f36a31f63125f54432b Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Fri, 26 Oct 2018 16:13:20 -0500 Subject: [PATCH] implement MustNewClusterWithProxy and drop/undrop for partitioning --- server/server_test.go | 19 ++++++-- test/pilosa.go | 111 +++++++++++++++++++++++++++++++----------- 2 files changed, 97 insertions(+), 33 deletions(-) diff --git a/server/server_test.go b/server/server_test.go index 3900a7faa..f30c02774 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -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) + } + +} diff --git a/test/pilosa.go b/test/pilosa.go index 901e8ffd2..cf197e163 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -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) }