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

remove port_mapper because it can't work with our unrestartable server

Browse source 
Long story short: Once we create a server and start it, we can't start
it again. We can't close it and restart it, and we can't just start
it without closing it.

Unfortunately, if the server's config needs to change, we have a Problem
here.

This ultimately means that the retry logic for GetListeners can't actually
retry successfully; if we fail on the first attempt, we necessarily fail
on any later attempts also, and if we try to fix that, we get panics.

But!

We don't actually NEED to retry. We just need to ensure that we can
open a :0 port, extract the actual port number, and use that in places
where the port number mattered, without having to rebind it.

The only actual place we needed to rebind things was opening gRPC
servers, so we introduce a gRPC Listener that can be used instead of
trying to bind to a specified port.

In a bunch of other cases where we had similar logic to try to allocate
and then use a port, we can switch to just using a provided listener.
For instance, net/http has `Serve(net.Listener, handler)`, not just
ListenAndServe(addr, handler).

This should eliminate the weird CI failures from eaddrinuse.

NOT fixed: server/cluster_test.go/TestClusterResize_AddNode isn't working
right now. The new node isn't actually being added to the existing cluster.
I attempted this but was outsmarted by it, and I think fixing the
rest of this is worth it as a separate thing.
This commit is contained in:
Seebs 2021-02-16 14:12:36 -06:00
parent 4200de481d
commit 4f5f3e30ea
13 changed files with 205 additions and 233 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
cluster_internal_test.go
View file

@ -27,7 +27,6 @@ import (
"github.com/davecgh/go-spew/spew"
pnet "github.com/pilosa/pilosa/v2/net"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/test/port"
"github.com/pilosa/pilosa/v2/testhook"
"github.com/pilosa/pilosa/v2/topology"
)
@ -498,13 +497,9 @@ func TestCluster_ContainsShards(t *testing.T) {
func TestCluster_Nodes(t *testing.T) {
const urisCount = 4
var uris []pnet.URI
if err := port.GetPorts(func(ports []int) error {
for i := 0; i < urisCount; i++ {
uris = append(uris, NewTestURIFromHostPort(fmt.Sprintf("node%d", i), uint16(ports[i])))
}
return nil
}, urisCount, 10); err != nil {
t.Fatalf("getting ports: %v", err)
arbitraryPorts := []int{17384, 17385, 17386, 17387}
for i := 0; i < urisCount; i++ {
uris = append(uris, NewTestURIFromHostPort(fmt.Sprintf("node%d", i), uint16(arbitraryPorts[i])))
}
node0 := &topology.Node{ID: "node0", URI: uris[0]}

14
http/handler_test.go
View file

@ -22,7 +22,6 @@ import (
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/http"
"github.com/pilosa/pilosa/v2/test"
"github.com/pilosa/pilosa/v2/test/port"
)
func TestHandlerOptions(t *testing.T) {
@ -35,15 +34,10 @@ func TestHandlerOptions(t *testing.T) {
t.Fatalf("expected error making handler without options, got nil")
}
var ln net.Listener
err = port.GetPort(func(p int) error {
ln, err = net.Listen("tcp", port.ColonZeroString(p))
if err != nil {
t.Fatal(err)
}
return err
}, 10)
ln, err := net.Listen("tcp", ":0")
if err != nil {
t.Fatalf("creating listener: %v", err)
}
_, err = http.NewHandler(http.OptHandlerListener(ln, ln.Addr().String()))
if err == nil {

14
main_test.go
View file

@ -16,24 +16,28 @@ package pilosa_test
import (
"fmt"
"net"
"net/http"
"testing"
_ "net/http/pprof"
"github.com/pilosa/pilosa/v2/test/port"
"github.com/pilosa/pilosa/v2/testhook"
)
func TestMain(m *testing.M) {
l, err := net.Listen("tcp", ":0")
if err != nil {
panic(err)
}
port := l.Addr().(*net.TCPAddr).Port
fmt.Printf("pilosa/ TestMain: online stack-traces: curl http://localhost:%v/debug/pprof/goroutine?debug=2\n", port)
go func() {
err := port.GetPort(func(port int) error {
fmt.Printf("pilosa/ TestMain: online stack-traces: curl http://localhost:%v/debug/pprof/goroutine?debug=2\n", port)
return http.ListenAndServe(fmt.Sprintf("127.0.0.1:%v", port), nil)
}, 10)
err := http.Serve(l, nil)
if err != nil {
panic(err)
}
}()
testhook.RunTestsWithHooks(m)
}

40
pg/pgtest/server.go
View file

@ -37,13 +37,8 @@ func (f ShutdownFunc) Finish(tb testing.TB, name string) {
}
}
// ServeTCP creates a TCP listener and serves postgres wire protocol on it.
func ServeTCP(addr string, server *pg.Server) (net.Addr, ShutdownFunc, error) {
listener, err := net.Listen("tcp", addr)
if err != nil {
return nil, nil, errors.Wrap(err, "listening on TCP")
}
// ServeListener serves postgres wire protocol on a listener.
func ServeListener(listener net.Listener, server *pg.Server) (net.Addr, ShutdownFunc, error) {
laddr := listener.Addr()
ctx, cancel := context.WithCancel(context.Background())
@ -58,6 +53,37 @@ func ServeTCP(addr string, server *pg.Server) (net.Addr, ShutdownFunc, error) {
nil
}
// ServeTCP creates a TCP listener and serves postgres wire protocol on it.
func ServeTCP(addr string, server *pg.Server) (net.Addr, ShutdownFunc, error) {
listener, err := net.Listen("tcp", addr)
if err != nil {
return nil, nil, errors.Wrap(err, "listening on TCP")
}
return ServeListener(listener, server)
}
// ServeTLSListener sets up TLS on the server and invokes ServeListener.
func ServeTLSListener(listener net.Listener, server *pg.Server) (net.Addr, ShutdownFunc, error) {
err := SetupTLS(server)
if err != nil {
return nil, nil, errors.Wrap(err, "server TLS setup failed")
}
var tries int = 5
var netAddr net.Addr
var shutdown ShutdownFunc
for i := 0; i < tries; i++ {
if i > 0 {
fmt.Printf("--- try serving TLS again: %d\n", i)
}
if netAddr, shutdown, err = ServeListener(listener, server); err == nil {
break
}
}
return netAddr, shutdown, err
}
// ServeTLS sets up TLS on the server and invokes ServeTCP.
func ServeTLS(addr string, server *pg.Server) (net.Addr, ShutdownFunc, error) {
err := SetupTLS(server)

34
pg/server_test.go
View file

@ -31,7 +31,6 @@ import (
"github.com/pilosa/pilosa/v2/logger"
"github.com/pilosa/pilosa/v2/pg"
"github.com/pilosa/pilosa/v2/pg/pgtest"
"github.com/pilosa/pilosa/v2/test/port"
)
// TestStartupTimeout tests that an incoming connection that does nothing times out and gets closed.
@ -111,14 +110,7 @@ func TestPQConnect(t *testing.T) {
Logger: logger.NopLogger,
}
var addr net.Addr
var shutdown pgtest.ShutdownFunc
var err error
err = port.GetPort(func(p int) error {
addr, shutdown, err = pgtest.ServeTCP(port.ColonZeroString(p), server)
return err
}, 10)
addr, shutdown, err := pgtest.ServeTCP(":0", server)
if err != nil {
t.Fatalf("starting postgres server: %v", err)
}
@ -150,13 +142,7 @@ func TestPQConnectSSL(t *testing.T) {
Logger: logger.NopLogger,
}
var addr net.Addr
var shutdown pgtest.ShutdownFunc
var err error
err = port.GetPort(func(p int) error {
addr, shutdown, err = pgtest.ServeTLS(port.ColonZeroString(p), server)
return err
}, 10)
addr, shutdown, err := pgtest.ServeTLS(":0", server)
if err != nil {
t.Fatalf("starting postgres server: %v", err)
}
@ -221,13 +207,7 @@ func TestPSQLQuery(t *testing.T) {
Logger: logger.NopLogger,
}
var addr net.Addr
var shutdown pgtest.ShutdownFunc
var err error
err = port.GetPort(func(p int) error {
addr, shutdown, err = pgtest.ServeTCP(port.ColonZeroString(p), server)
return err
}, 10)
addr, shutdown, err := pgtest.ServeTCP(":0", server)
if err != nil {
t.Fatalf("starting postgres server: %v", err)
}
@ -289,13 +269,7 @@ func TestPSQLQuery(t *testing.T) {
CancellationManager: pg.NewLocalCancellationManager(rand.Reader),
}
var addr net.Addr
var shutdown pgtest.ShutdownFunc
var err error
err = port.GetPort(func(p int) error {
addr, shutdown, err = pgtest.ServeTCP(port.ColonZeroString(p), server)
return err
}, 10)
addr, shutdown, err := pgtest.ServeTCP(":0", server)
if err != nil {
t.Fatalf("starting postgres server: %v", err)
}

13
rbf/db_test.go
View file

@ -18,6 +18,7 @@ import (
"context"
"fmt"
"math/rand"
"net"
"net/http"
"os"
"testing"
@ -27,7 +28,6 @@ import (
"github.com/pilosa/pilosa/v2/rbf"
rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg"
"github.com/pilosa/pilosa/v2/test/port"
"golang.org/x/sync/errgroup"
)
@ -351,11 +351,14 @@ func TestDB_MultiTx(t *testing.T) {
// better diagnosis of deadlocks/hung situations versus just really slow "Quick" tests.
func TestMain(m *testing.M) {
l, err := net.Listen("tcp", ":0")
if err != nil {
panic(err)
}
port := l.Addr().(*net.TCPAddr).Port
fmt.Printf("rbf/ TestMain: online stack-traces: curl http://localhost:%v/debug/pprof/goroutine?debug=2\n", port)
go func() {
err := port.GetPort(func(port int) error {
fmt.Printf("rbf/ TestMain: online stack-traces: curl http://localhost:%v/debug/pprof/goroutine?debug=2\n", port)
return http.ListenAndServe(fmt.Sprintf("127.0.0.1:%v", port), nil)
}, 10)
err := http.Serve(l, nil)
if err != nil {
panic(err)
}

33
server/cluster_test.go
View file

@ -29,7 +29,6 @@ import (
"github.com/pilosa/pilosa/v2/disco"
"github.com/pilosa/pilosa/v2/server"
"github.com/pilosa/pilosa/v2/test"
"github.com/pilosa/pilosa/v2/test/port"
)
// Ensure program can send/receive broadcast messages.
@ -185,19 +184,27 @@ func TestClusterResize_AddNode(t *testing.T) {
// Configure node1
m1 := test.NewCommandNode(t)
if err := port.GetListeners(func(lsns []*net.TCPListener) error {
portsCfg := test.GenPortsConfig(test.NewPorts(lsns))
m1.Config.Etcd = portsCfg[0].Etcd
m1.Config.Name = portsCfg[0].Name
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
m1.Config.BindGRPC = portsCfg[0].BindGRPC
return m1.Start()
}, 3, 10); err != nil {
t.Fatalf("starting second main: %v", err)
lsns := make([]*net.TCPListener, 3)
for i := range lsns {
l, err := net.Listen("tcp", ":0")
if err != nil {
t.Fatal(err)
}
lsns[i] = l.(*net.TCPListener)
}
portsCfg := test.GenPortsConfig(test.NewPorts(lsns))
m1.Config.Etcd = portsCfg[0].Etcd
m1.Config.Name = portsCfg[0].Name
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
m1.Config.BindGRPC = portsCfg[0].BindGRPC
m1.Config.GRPCListener = portsCfg[0].GRPCListener
err := m1.Start()
if err != nil {
t.Fatal(err)
}
defer m1.Close()
state0, err0 := m0.API.State()

6
server/config.go
View file

@ -66,6 +66,12 @@ type Config struct {
// BindGRPC is the host:port on which Pilosa will bind for gRPC.
BindGRPC string `toml:"bind-grpc"`
// GRPCListener is an already-bound listener to use for gRPC.
// This is for use by test infrastructure, where it's useful to
// be able to dynamically generate the bindings by actually binding
// to :0, and avoid "address already in use" errors.
GRPCListener *net.TCPListener
// Advertise is the address advertised by the server to other nodes
// in the cluster. It should be reachable by all other nodes and should
// route to an interface that Bind is listening on.

22
server/server.go
View file

@ -296,16 +296,18 @@ func (m *Command) SetupServer() error {
if err != nil {
return errors.Wrap(err, "processing bind grpc address")
}
// create gRPC listener
m.grpcLn, err = net.Listen("tcp", grpcURI.HostPort())
if err != nil {
return errors.Wrap(err, "creating grpc listener")
}
// If grpc port is 0, get auto-allocated port from listener
if grpcURI.Port == 0 {
grpcURI.SetPort(uint16(m.grpcLn.Addr().(*net.TCPAddr).Port))
if m.Config.GRPCListener == nil {
// create gRPC listener
m.grpcLn, err = net.Listen("tcp", grpcURI.HostPort())
if err != nil {
return errors.Wrap(err, "creating grpc listener")
}
// If grpc port is 0, get auto-allocated port from listener
if grpcURI.Port == 0 {
grpcURI.SetPort(uint16(m.grpcLn.Addr().(*net.TCPAddr).Port))
}
} else {
m.grpcLn = m.Config.GRPCListener
}
// Setup TLS

13
server/server_test.go
View file

@ -22,6 +22,7 @@ import (
"fmt"
"io/ioutil"
"math/rand"
"net"
nethttp "net/http"
"os"
"reflect"
@ -37,7 +38,6 @@ import (
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/server"
"github.com/pilosa/pilosa/v2/test"
"github.com/pilosa/pilosa/v2/test/port"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
@ -1190,11 +1190,14 @@ Set("h", adec=100.22)
}
func TestMain(m *testing.M) {
l, err := net.Listen("tcp", ":0")
if err != nil {
panic(err)
}
port := l.Addr().(*net.TCPAddr).Port
fmt.Printf("server/ TestMain: online stack-traces: curl http://localhost:%v/debug/pprof/goroutine?debug=2\n", port)
go func() {
err := port.GetPort(func(port int) error {
fmt.Printf("server/ TestMain: online stack-traces: curl http://localhost:%v/debug/pprof/goroutine?debug=2\n", port)
return nethttp.ListenAndServe(fmt.Sprintf("127.0.0.1:%v", port), nil)
}, 10)
err := nethttp.Serve(l, nil)
if err != nil {
panic(err)
}

50
test/cluster.go
View file

@ -18,7 +18,6 @@ import (
"context"
"fmt"
"math"
"net"
"sort"
"strings"
"testing"
@ -30,7 +29,6 @@ import (
"github.com/pilosa/pilosa/v2/proto"
"github.com/pilosa/pilosa/v2/server"
"github.com/pilosa/pilosa/v2/storage"
"github.com/pilosa/pilosa/v2/test/port"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
@ -45,6 +43,7 @@ func (*ModHasher) Name() string { return "mod" }
// Cluster represents a Pilosa cluster (multiple Command instances)
type Cluster struct {
Nodes []*Command
tb testing.TB
}
// Query executes an API.Query through one of the cluster's node's API. It fails
@ -376,40 +375,21 @@ func (c *Cluster) CreateField(t testing.TB, index string, iopts pilosa.IndexOpti
// Start runs a Cluster
func (c *Cluster) Start() error {
var eg errgroup.Group
err := port.GetListeners(
func(lsns []*net.TCPListener) (err0 error) {
sliceOfPorts := NewPorts(lsns)
defer func() {
if err0 != nil {
// going to retry. Close the still open listeners
for _, ports := range sliceOfPorts {
_ = ports.Close()
}
}
}()
portsCfg := GenPortsConfig(sliceOfPorts)
for i, cc := range c.Nodes {
cc := cc
cc.Config.Etcd = portsCfg[i].Etcd
cc.Config.Name = portsCfg[i].Name
cc.Config.Cluster.Name = portsCfg[i].Cluster.Name
cc.Config.BindGRPC = portsCfg[i].BindGRPC
eg.Go(func() error {
return cc.Start()
})
}
return eg.Wait()
}, 3*len(c.Nodes), 10)
err := GetPortsGenConfigs(c.tb, c.Nodes)
if err != nil {
return err
return errors.Wrap(err, "configuring cluster ports")
}
var eg errgroup.Group
for _, cc := range c.Nodes {
cc := cc
eg.Go(func() error {
return cc.Start()
})
}
err = eg.Wait()
if err != nil {
return errors.Wrap(err, "starting cluster")
}
return c.GetNode(0).AwaitState(disco.ClusterStateNormal, 30*time.Second)
}
@ -488,7 +468,7 @@ func newCluster(tb testing.TB, size int, 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 := &Cluster{Nodes: make([]*Command, size)}
cluster := &Cluster{Nodes: make([]*Command, size), tb: tb}
for i := 0; i < size; i++ {
var commandOpts []server.CommandOption
if len(opts) > 0 {

91
test/disco.go
View file

@ -19,10 +19,13 @@ import (
"io/ioutil"
"net"
"strings"
"testing"
"time"
"github.com/pilosa/pilosa/v2/etcd"
"github.com/pilosa/pilosa/v2/server"
"github.com/pilosa/pilosa/v2/testhook"
"github.com/pkg/errors"
)
type Ports struct {
@ -32,16 +35,90 @@ type Ports struct {
LsnP *net.TCPListener
PortP int
LsnG *net.TCPListener
Grpc int
}
func (ports *Ports) Close() error {
err := ports.LsnC.Close()
err2 := ports.LsnP.Close()
err3 := ports.LsnG.Close()
if err != nil {
return err
}
return err2
if err2 != nil {
return err2
}
return err3
}
// listenerPortURL builds a TCP listener and corresponding http://localhost:%d
// URL, and returns those.
func listenerWithURL() (listener *net.TCPListener, url string, err error) {
l, err := net.Listen("tcp", ":0")
if err != nil {
return listener, url, err
}
listener = l.(*net.TCPListener)
port := listener.Addr().(*net.TCPAddr).Port
url = fmt.Sprintf("http://localhost:%d", port)
return listener, url, err
}
// GetPortsGenConfigs creates listener ports, and updates the configurations
// of servers to match these created ports, including cross-references
// like updating the InitCluster values in the Etcd configs.
func GetPortsGenConfigs(tb testing.TB, nodes []*Command) error {
peerUrls := make([]string, len(nodes))
for i := range nodes {
if nodes[i].Config == nil {
nodes[i].Config = &server.Config{}
}
config := nodes[i].Config
name := fmt.Sprintf("server%d", i)
clusterName := fmt.Sprintf("cluster-%s", tb.Name())
discoDir, err := testhook.TempDir(tb, "disco.")
if err != nil {
return errors.Wrap(err, "creating temp directory")
}
clientListener, clientURL, err := listenerWithURL()
if err != nil {
return errors.Wrap(err, "creating client listener")
}
peerListener, peerURL, err := listenerWithURL()
if err != nil {
return errors.Wrap(err, "creating peer listener")
}
grpcListener, grpcUrl, err := listenerWithURL()
if err != nil {
return errors.Wrap(err, "creating gRPC listener")
}
// for grpc, we don't want the http part...
colon := strings.LastIndexByte(grpcUrl, ':')
if colon != -1 {
grpcUrl = grpcUrl[colon:]
}
config.Name = name
config.Cluster.Name = clusterName
config.BindGRPC = grpcUrl
config.GRPCListener = grpcListener
config.Etcd = etcd.Options{
Dir: discoDir,
LClientURL: clientURL,
AClientURL: clientURL,
LPeerURL: peerURL,
APeerURL: peerURL,
HeartbeatTTL: 5 * int64(time.Second),
LPeerSocket: []*net.TCPListener{peerListener},
LClientSocket: []*net.TCPListener{clientListener},
}
peerUrls[i] = fmt.Sprintf("%s=%s", name, peerURL)
}
allPeerUrls := strings.Join(peerUrls, ",")
for i := range nodes {
nodes[i].Config.Etcd.InitCluster = allPeerUrls
}
return nil
}
//GenPortsConfig creates specific configuration for etcd.
@ -63,8 +140,9 @@ func GenPortsConfig(ports []Ports) []*server.Config {
}
cfgs[i] = &server.Config{
Name: name,
BindGRPC: fmt.Sprintf(":%d", ports[i].Grpc),
Name: name,
BindGRPC: fmt.Sprintf(":%d", ports[i].Grpc),
GRPCListener: ports[i].LsnG,
Etcd: etcd.Options{
Dir: discoDir,
LClientURL: lClientURL,
@ -102,12 +180,9 @@ func NewPorts(lsn []*net.TCPListener) []Ports {
PortC: ports[i],
LsnP: lsn[i+1],
PortP: ports[i+1],
Grpc: ports[i+2],
Grpc: ports[i+2],
LsnG: lsn[i+2],
})
// make Grpc port available to
// be rebound.
lsn[i+2].Close()
}
return out

97
test/port/port_mapper.go
View file

@ -1,97 +0,0 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package port
import (
"fmt"
"log"
"net"
"strings"
"syscall"
)
func ColonZeroString(port int) string {
return fmt.Sprintf(":%d", port)
}
func GetPort(wrapper func(int) error, retries int) error {
f := func(ports []int) error {
return wrapper(ports[0])
}
return GetPorts(f, 1, retries)
}
func GetPorts(wrapper func([]int) error, requestedPorts, retries int) error {
for i := 0; i < retries; i++ {
// get all requested ports
listeners := make([]*net.TCPListener, requestedPorts)
ports := make([]int, requestedPorts)
for i := 0; i < requestedPorts; i++ {
l, err := net.Listen("tcp", ":0")
if err != nil {
log.Println("[port_mapper] error getting a free port", err)
return GetPorts(wrapper, requestedPorts, retries-1)
}
ports[i] = l.Addr().(*net.TCPAddr).Port
listeners[i] = l.(*net.TCPListener)
}
for _, l := range listeners {
if err := l.Close(); err != nil {
log.Println("[port_mapper] error closing the listener", err)
}
}
// send to wrapper and check output error
err := wrapper(ports)
if (err != nil) && (err == syscall.EADDRINUSE || strings.Contains(err.Error(), "address already in use")) {
log.Printf("[port_mapper: %+v] address already in use error calling the wrapper: %v\n", ports, err)
// only retry on address already in use error
continue
}
return err
}
return nil
}
func GetListeners(wrapper func([]*net.TCPListener) error, requestedPorts, retries int) error {
for i := 0; i < retries; i++ {
// get all requested ports
listeners := make([]*net.TCPListener, requestedPorts)
ports := make([]int, requestedPorts)
for i := 0; i < requestedPorts; i++ {
l, err := net.Listen("tcp", ":0")
if err != nil {
log.Println("[port_mapper] error getting a free port", err)
return GetListeners(wrapper, requestedPorts, retries-1)
}
ports[i] = l.Addr().(*net.TCPAddr).Port
listeners[i] = l.(*net.TCPListener)
}
// send to wrapper and check output error
err := wrapper(listeners)
if (err != nil) && (err == syscall.EADDRINUSE || strings.Contains(err.Error(), "address already in use")) {
log.Printf("[port_mapper: %+v] address already in use error calling the wrapper: %v\n", ports, err)
// only retry on address already in use error
continue
}
return err
}
return nil
}