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

add docker based cluster tests, remove proxy stuff, fix advertise

Browse source
This commit is contained in:
Matt Jaffee 2018-11-02 14:41:42 -05:00
parent ec08930c30
commit d86e9ed3ea
No known key found for this signature in database
GPG key ID: 08A3DFFF987B11BF
10 changed files with 236 additions and 365 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
Dockerfile-withgo Normal file
View file

@ -0,0 +1,22 @@
FROM golang:1.11
LABEL maintainer "dev@pilosa.com"
COPY . /go/src/github.com/pilosa/pilosa/
RUN cd /go/src/github.com/pilosa/pilosa \
&& CGO_ENABLED=0 make install-dep install FLAGS="-a"
RUN wget https://github.com/alexei-led/pumba/releases/download/0.6.0/pumba_linux_amd64 -O /pumba
RUN chmod +x /pumba
RUN cp /go/bin/pilosa /pilosa
COPY LICENSE /LICENSE
COPY NOTICE /NOTICE
EXPOSE 10101
VOLUME /data
ENTRYPOINT ["bash", "-c"]
CMD ["pilosa", "server", "--data-dir", "/data", "--bind", "http://0.0.0.0:10101"]

11
Gopkg.lock generated
View file

@ -189,6 +189,15 @@
revision = "c01d1270ff3e442a8a57cddc1c92dc1138598194"
version = "v1.2.0"
[[projects]]
name = "github.com/pilosa/go-pilosa"
packages = [
".",
"gopilosa_pbuf"
]
revision = "4e7807f5ad779407936744057cd17332046b6c3c"
version = "v1.1.0"
[[projects]]
name = "github.com/pkg/errors"
packages = ["."]
@ -324,6 +333,6 @@
[solve-meta]
analyzer-name = "dep"
analyzer-version = 1
inputs-digest = "8290156ce8b4066c46ab83d743f4c81df0a17e148415bb1ee8409a51ac4c3ba4"
inputs-digest = "be318fa4f2a72e7e849b2faff1ce9300deaaf2d24cf76100f4b538abed86295b"
solver-name = "gps-cdcl"
solver-version = 1

1
ctl/server.go
View file

@ -26,6 +26,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
flags := cmd.Flags()
flags.StringVarP(&srv.Config.DataDir, "data-dir", "d", srv.Config.DataDir, "Directory to store pilosa data files.")
flags.StringVarP(&srv.Config.Bind, "bind", "b", srv.Config.Bind, "Default URI on which pilosa should listen.")
flags.StringVarP(&srv.Config.Advertise, "advertise", "a", "", "Address to broadcast to other hosts and clients to be contacted on.")
flags.IntVarP(&srv.Config.MaxWritesPerRequest, "max-writes-per-request", "", srv.Config.MaxWritesPerRequest, "Number of write commands per request.")
flags.StringVar(&srv.Config.LogPath, "log-path", srv.Config.LogPath, "Log path")
flags.BoolVar(&srv.Config.Verbose, "verbose", srv.Config.Verbose, "Enable verbose logging")

3
internal/clustertests/Dockerfile Normal file
View file

@ -0,0 +1,3 @@
FROM ptest
COPY . /go/src/github.com/pilosa/pilosa/internal/clustertests

138
internal/clustertests/cluster_test.go Normal file
View file

@ -0,0 +1,138 @@
package clustertest
import (
"fmt"
"io"
"os"
"os/exec"
"testing"
"time"
"github.com/pilosa/go-pilosa"
pi "github.com/pilosa/pilosa"
)
func TestLongPauses(t *testing.T) {
t.Skip()
cli := getPilosaClient(t)
idx := pilosa.NewIndex("testidx")
err := cli.CreateIndex(idx)
if err != nil {
t.Fatalf("creating index: %v", err)
}
f := idx.Field("testf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 10))
err = cli.CreateField(f)
if err != nil {
t.Fatalf("creating field: %v", err)
}
data := make([]pilosa.Column, 1000)
for i := range data {
data[i].RowID = 0
data[i].ColumnID = uint64((i/10)*pi.ShardWidth + i%10)
}
err = cli.ImportField(f, &colIterator{cols: data}, pilosa.OptImportBatchSize(1000))
if err != nil {
t.Fatalf("importing: %v", err)
}
r, err := cli.Query(idx.Count(f.Row(0)))
if err != nil {
t.Fatalf("count querying: %v", err)
}
if r.Result().Count() != 1000 {
t.Fatalf("count after import is %d", r.Result().Count())
}
pcmd := exec.Command("/pumba", "pause", "clustertests_pilosa3_1", "--duration", "30s")
pcmd.Stdout = os.Stdout
pcmd.Stderr = os.Stderr
fmt.Println("pausing pilosa3 for 30s")
err = pcmd.Start()
if err != nil {
t.Fatalf("starting pumba command: %v", err)
}
err = pcmd.Wait()
if err != nil {
t.Fatalf("waiting on pumba pause cmd: %v", err)
}
fmt.Println("done with pause, waiting for stability")
time.Sleep(time.Second * 400)
fmt.Println("done waiting")
r, err = cli.Query(idx.Count(f.Row(0)))
if err != nil {
t.Fatalf("count querying: %v", err)
}
if r.Result().Count() != 1000 {
t.Fatalf("count after import is %d", r.Result().Count())
}
}
// Utils
func getPilosaClient(t *testing.T) *pilosa.Client {
cli, err := pilosa.NewClient("pilosa1:10101")
if err != nil {
time.Sleep(time.Millisecond * 40)
}
time.Sleep(time.Second * 2)
// start := time.Now()
// for i := 0; true; i++ {
// s, err := cli.Status()
// if i > 800 {
// t.Fatalf("couldn't connect to cluster after %d attempts and %v: state: %s, err: %v", i, time.Since(start), s.State, err)
// }
// if err != nil {
// time.Sleep(time.Millisecond * 40)
// continue
// }
// if s.State == "NORMAL" {
// break
// } else {
// time.Sleep(time.Millisecond * 40)
// }
// }
return cli
}
type colIterator struct {
cols []pilosa.Column
i uint
}
func (c *colIterator) NextRecord() (pilosa.Record, error) {
if int(c.i) >= len(c.cols) {
return nil, io.EOF
}
c.i++
return c.cols[c.i-1], nil
}
func TestColIterator(t *testing.T) {
data := make([]pilosa.Column, 3)
for i := range data {
data[i].RowID = 0
data[i].ColumnID = uint64((i/10)*pi.ShardWidth + i%10)
}
ci := colIterator{cols: data}
col := pilosa.Column{}
if rec, err := ci.NextRecord(); rec != col {
t.Fatalf("first record wrong: %v, err: %v", rec, err)
}
col.ColumnID = 1
if rec, err := ci.NextRecord(); rec != col || err != nil {
t.Fatalf("second record wrong: %v, err: %v", rec, err)
}
col.ColumnID = 2
if rec, err := ci.NextRecord(); rec != col || err != nil {
t.Fatalf("third record wrong: %v, err: %v", rec, err)
}
if rec, err := ci.NextRecord(); err != io.EOF {
t.Fatalf("should be EOF, but got %v, err: %v", rec, err)
}
}

53
internal/clustertests/docker-compose.yml Normal file
View file

@ -0,0 +1,53 @@
version: '2'
services:
pilosa1:
build:
context: ../..
dockerfile: Dockerfile-withgo
image: ptest
ports:
- "33455:10101"
environment:
- PILOSA_CLUSTER_COORDINATOR=true
- PILOSA_GOSSIP_SEEDS=pilosa1:14000
networks:
- pilosanet
command:
- "/pilosa server --bind pilosa1:10101"
pilosa2:
build:
context: ../..
dockerfile: Dockerfile-withgo
image: ptest
ports:
- "33456:10101"
environment:
- PILOSA_GOSSIP_SEEDS=pilosa1:14000
networks:
- pilosanet
command:
- "/pilosa server --bind pilosa2:10101"
pilosa3:
build:
context: .
image: ptest
ports:
- "33457:10101"
environment:
- PILOSA_GOSSIP_SEEDS=pilosa1:14000,pilosa2:14000
networks:
- pilosanet
command:
- "/pilosa server --bind pilosa3:10101"
client1:
build:
context: ../..
dockerfile: Dockerfile-withgo
networks:
- pilosanet
volumes:
- /var/run/docker.sock:/var/run/docker.sock
command:
- "go test -v github.com/pilosa/pilosa/internal/clustertests"
networks:
pilosanet:

151
internal/udproxy/udproxy.go
View file

@ -1,151 +0,0 @@
package udproxy
import (
"bytes"
"io"
"net"
"sync"
"time"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
type Proxy struct {
upstreamAddr *net.UDPAddr
conn *net.UDPConn
drop bool
dropLock sync.Mutex
quit chan struct{}
eg errgroup.Group
// map from client address to upstream connection. We must maintain a
// separate connection to upstream for each client connection so that we can
// differentiate data sent back from upstream.
upstreams map[*net.UDPAddr]*net.UDPConn
// TODO - need to track a per-connection timeout so that "upstreams" doesn't
// grow indefinitely.
}
func New(listenIP string, listenPort int, upstreamIP string, upstreamPort int) (*Proxy, error) {
uc, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP(listenIP), Port: listenPort})
if err != nil {
return nil, errors.Wrap(err, "listening")
}
p := &Proxy{
conn: uc,
upstreamAddr: &net.UDPAddr{IP: net.ParseIP(upstreamIP), Port: upstreamPort},
quit: make(chan struct{}),
upstreams: make(map[*net.UDPAddr]*net.UDPConn),
}
if p.upstreamAddr.IP == nil {
return nil, errors.Errorf("unable to parse upstream ip '%s'", upstreamIP)
}
p.eg.Go(p.run)
return p, nil
}
func (p *Proxy) Drop() {
// sleep to attempt to ensure all traffic that was supposed to pass, did
// pass.
time.Sleep(time.Millisecond * 3)
p.dropLock.Lock()
p.drop = true
p.dropLock.Unlock()
}
func (p *Proxy) Undrop() {
// this sleep is a cheap attempt to ensure everything that was supposed to
// be dropped was.
time.Sleep(time.Millisecond * 3)
p.dropLock.Lock()
p.drop = false
p.dropLock.Unlock()
}
func (p *Proxy) dropping() bool {
p.dropLock.Lock()
d := p.drop
p.dropLock.Unlock()
return d
}
func (p *Proxy) run() error {
buf := make([]byte, 65507)
for {
select {
case <-p.quit:
return nil
default:
}
err := p.conn.SetReadDeadline(time.Now().Add(time.Millisecond))
if err != nil {
return errors.Wrap(err, "setting read deadline (run)")
}
n, addr, err := p.conn.ReadFromUDP(buf)
if err, ok := err.(net.Error); ok && err.Timeout() {
continue
} else if err != nil {
return errors.Wrap(err, "reading from udp conn")
}
upConn := p.upstreams[addr]
if upConn == nil {
p.upstreams[addr], err = net.DialUDP("udp", &net.UDPAddr{}, p.upstreamAddr)
if err != nil {
return errors.Wrap(err, "creating new connection to upstream")
}
p.eg.Go(func() error {
return p.proxyBack(addr, p.upstreams[addr])
})
upConn = p.upstreams[addr]
}
if !p.dropping() {
_, err = io.Copy(upConn, bytes.NewBuffer(buf[:n]))
if err != nil {
return errors.Wrap(err, "writing to upstream conn")
}
}
}
}
func (p *Proxy) proxyBack(to *net.UDPAddr, from *net.UDPConn) error {
buf := make([]byte, 65507)
for {
select {
case <-p.quit:
return nil
default:
}
err := from.SetReadDeadline(time.Now().Add(time.Millisecond))
if err != nil {
return errors.Wrap(err, "setting read deadline (proxyBack)")
}
n, _, err := from.ReadFromUDP(buf)
if err, ok := err.(net.Error); ok && err.Timeout() {
continue
} else if err != nil {
return errors.Wrap(err, "reading from upstream")
}
if !p.dropping() {
_, err = io.Copy(addrWriter{c: p.conn, a: to}, bytes.NewBuffer(buf[:n]))
if err != nil {
return errors.Wrap(err, "writing back to client")
}
}
}
}
type addrWriter struct {
c *net.UDPConn
a *net.UDPAddr
}
func (a addrWriter) Write(b []byte) (n int, err error) {
return a.c.WriteTo(b, a.a)
}
func (p *Proxy) Close() error {
close(p.quit)
return p.eg.Wait()
}

83
internal/udproxy/udproxy_test.go
View file

@ -1,83 +0,0 @@
package udproxy_test
import (
"net"
"testing"
"github.com/pilosa/pilosa/internal/udproxy"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
func TestUDProxy(t *testing.T) {
p, err := udproxy.New("127.0.0.1", 12345, "127.0.0.1", 12346)
if err != nil {
t.Fatalf("creating proxy: %v", err)
}
uc, err := net.ListenUDP("udp", &net.UDPAddr{Port: 12346})
if err != nil {
t.Fatalf("listening udp upstream: %v", err)
}
resp := make([]byte, 8)
eg := errgroup.Group{}
eg.Go(func() error {
conn, err := net.DialUDP("udp", nil, &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 12345})
if err != nil {
t.Fatalf("connecting to proxy: %v", err)
}
_, err = conn.Write([]byte("hello!"))
if err != nil {
return errors.Wrap(err, "writing to proxy")
}
_, err = conn.Read(resp)
if err != nil {
return errors.Wrap(err, "reading from proxy")
}
p.Drop()
_, err = conn.Write([]byte("hello2"))
if err != nil {
return errors.Wrap(err, "writing to dropping proxy")
}
p.Undrop()
_, err = conn.Write([]byte("hello3"))
if err != nil {
return errors.Wrap(err, "writing to undropping proxy")
}
return nil
})
req := make([]byte, 10)
_, addr, err := uc.ReadFrom(req)
if err != nil {
t.Fatalf("upstream reading from proxy: %v", err)
}
if string(req[:6]) != "hello!" {
t.Fatalf("got unexpected request %s", req)
}
_, err = uc.WriteTo([]byte("goodbye"), addr)
if err != nil {
t.Fatalf("writing response: %v", err)
}
_, _, err = uc.ReadFrom(req)
if err != nil {
t.Fatalf("upstream reading from proxy: %v", err)
}
if string(req[:6]) != "hello3" {
t.Fatalf("got unexpected request %s", req)
}
eg.Wait()
if string(resp[:7]) != "goodbye" {
t.Fatalf("got unexpected response '%v", resp)
}
err = p.Close()
if err != nil {
t.Fatalf("err closing proxy: '%v'", err)
}
}
// TODO test dropping

15
server/server_test.go
View file

@ -379,11 +379,7 @@ 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.MustNewCluster(t, 3)
err := cluster.Start()
if err != nil {
t.Fatalf("starting cluster: %v", err)
}
cluster := test.MustRunCluster(t, 3)
defer cluster.Close()
var wait = true
@ -696,12 +692,3 @@ 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)
}
}

124
test/pilosa.go
View file

@ -19,100 +19,28 @@ import (
"context"
"fmt"
"io/ioutil"
"math/rand"
gohttp "net/http"
"os"
"path"
"strconv"
"strings"
"sync/atomic"
"testing"
"time"
"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/Shopify/toxiproxy"
tox "github.com/Shopify/toxiproxy/client"
)
type portAllocator struct {
port *uint32
}
var ports *portAllocator
// Next generates a new port and returns it
func (n *portAllocator) Next() (nextPort uint32) {
return atomic.AddUint32(n.port, 1)
}
var proxy string
func init() {
r := rand.New(rand.NewSource(int64(time.Now().Nanosecond())))
p := uint32(r.Uint32()%40000 + 20000)
ports = &portAllocator{
port: &p,
}
tport := strconv.Itoa(int(ports.Next()))
proxy = "localhost:" + tport
go toxiproxy.NewServer().Listen("localhost", tport)
}
////////////////////////////////////////////////////////////////////////////////////
// Command represents a test wrapper for server.Command.
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
@ -346,6 +274,7 @@ func (c Cluster) CreateField(t testing.TB, index string, iopts pilosa.IndexOptio
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)
@ -367,17 +296,7 @@ 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, 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...)
c, err := newCluster(size, opts...)
if err != nil {
tb.Fatalf("new cluster: %v", err)
}
@ -385,7 +304,7 @@ func MustNewClusterWithProxy(tb testing.TB, size int, opts ...[]server.CommandOp
}
// newCluster creates a new cluster
func newCluster(size int, withproxy bool, opts ...[]server.CommandOption) (cluster Cluster, err error) {
func newCluster(size int, opts ...[]server.CommandOption) (Cluster, error) {
if size == 0 {
return nil, errors.New("cluster must contain at least one node")
}
@ -393,41 +312,14 @@ func newCluster(size int, withproxy bool, opts ...[]server.CommandOption) (clust
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)]
}
name := "node" + strconv.Itoa(i)
m := NewCommandNode(i == 0, commandOpts...)
m.Config.Bind = "localhost:" + strconv.Itoa(int(ports.Next()))
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)
err := ioutil.WriteFile(path.Join(m.Config.DataDir, ".id"), []byte("node"+strconv.Itoa(i)), 0600)
if err != nil {
return nil, errors.Wrap(err, "writing node id")
}
@ -438,8 +330,8 @@ func newCluster(size int, withproxy bool, opts ...[]server.CommandOption) (clust
}
// runCluster creates and starts a new cluster
func runCluster(size int, withproxy bool, opts ...[]server.CommandOption) (Cluster, error) {
cluster, err := newCluster(size, withproxy, opts...)
func runCluster(size int, opts ...[]server.CommandOption) (Cluster, error) {
cluster, err := newCluster(size, opts...)
if err != nil {
return nil, errors.Wrap(err, "new cluster")
}
@ -451,7 +343,7 @@ func runCluster(size int, withproxy bool, opts ...[]server.CommandOption) (Clust
// MustRunCluster creates and starts a new cluster
func MustRunCluster(tb testing.TB, size int, opts ...[]server.CommandOption) Cluster {
c, err := runCluster(size, false, opts...)
c, err := runCluster(size, opts...)
if err != nil {
tb.Fatalf("run cluster: %v", err)
}