diff --git a/Dockerfile-withgo b/Dockerfile-withgo new file mode 100644 index 000000000..5cb11e47b --- /dev/null +++ b/Dockerfile-withgo @@ -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"] diff --git a/Gopkg.lock b/Gopkg.lock index 4a6068840..280f6d056 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -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 diff --git a/ctl/server.go b/ctl/server.go index 7f384ec7b..90766c498 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -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") diff --git a/internal/clustertests/Dockerfile b/internal/clustertests/Dockerfile new file mode 100644 index 000000000..ecbfa7f21 --- /dev/null +++ b/internal/clustertests/Dockerfile @@ -0,0 +1,3 @@ +FROM ptest + +COPY . /go/src/github.com/pilosa/pilosa/internal/clustertests diff --git a/internal/clustertests/cluster_test.go b/internal/clustertests/cluster_test.go new file mode 100644 index 000000000..e63b95d0b --- /dev/null +++ b/internal/clustertests/cluster_test.go @@ -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) + } +} diff --git a/internal/clustertests/docker-compose.yml b/internal/clustertests/docker-compose.yml new file mode 100644 index 000000000..752c39177 --- /dev/null +++ b/internal/clustertests/docker-compose.yml @@ -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: diff --git a/internal/udproxy/udproxy.go b/internal/udproxy/udproxy.go deleted file mode 100644 index a79c0121f..000000000 --- a/internal/udproxy/udproxy.go +++ /dev/null @@ -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() -} diff --git a/internal/udproxy/udproxy_test.go b/internal/udproxy/udproxy_test.go deleted file mode 100644 index be746ea72..000000000 --- a/internal/udproxy/udproxy_test.go +++ /dev/null @@ -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 diff --git a/server/server_test.go b/server/server_test.go index f30c02774..d62605ce0 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -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) - } - -} diff --git a/test/pilosa.go b/test/pilosa.go index 3efecfe73..3bc43d593 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -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) }