CyberCoveLLC/threshold
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove ALPN Muxer (#1965)

Browse Source 
We never ended up using this and it makes it much harder to reason about the listeners.

An explanation: for the moment I separated out all of the listeners in the struct, even though one should suffice, because when I did it that way first I was hitting some really weird and hard to find intermittent behavior where it seemed like the listeners were getting crossed. So I re-did it this way with each listener very explicitly referenced. I'm happy to make a follow-up PR that attempts to combine them all into a single listener without hitting the issues I did, if desired, but at least doing it via a separate PR means that it won't be something with removing the ALPN muxer that is at fault.
This commit is contained in:
Jeff Mitchell authored and GitHub committed 2022-04-01 12:06:41 -04:00
1 parent ce7292d9c8
commit 07dc3db974
13 files changed
+191 -832
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/servers/controller/listeners.go
+11 -38
View File
@@ -14,7 +14,6 @@ import (
grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
"github.com/hashicorp/boundary/internal/cmd/base"
pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
"github.com/hashicorp/boundary/internal/libs/alpnmux"
"github.com/hashicorp/boundary/internal/servers/controller/handlers/workers"
"github.com/hashicorp/go-multierror"
"google.golang.org/grpc"
@@ -42,7 +41,7 @@ func (c *Controller) startListeners() error {
for i := range c.apiListeners {
ln := c.apiListeners[i]
apiServers, err := c.configureForAPI(ln)
apiServers, err := c.configureForApi(ln)
if err != nil {
return fmt.Errorf("failed to configure listener for api mode: %w", err)
}
@@ -62,7 +61,7 @@ func (c *Controller) startListeners() error {
return nil
}
func (c *Controller) configureForAPI(ln *base.ServerListener) ([]func(), error) {
func (c *Controller) configureForApi(ln *base.ServerListener) ([]func(), error) {
apiServers := make([]func(), 0)
handler, err := c.apiHandler(HandlerProperties{
@@ -97,39 +96,15 @@ func (c *Controller) configureForAPI(ln *base.ServerListener) ([]func(), error)
server.IdleTimeout = ln.Config.HTTPIdleTimeout
}
switch ln.Config.TLSDisable {
case true:
l, err := ln.Mux.RegisterProto(alpnmux.NoProto, nil)
if err != nil {
return nil, fmt.Errorf("error getting non-tls listener: %w", err)
}
if l == nil {
return nil, errors.New("could not get non-tls listener")
}
apiServers = append(apiServers, func() { go server.Serve(l) })
default:
for _, v := range []string{"", "http/1.1", "h2"} {
l := ln.Mux.GetListener(v)
if l == nil {
return nil, fmt.Errorf("could not get tls proto %q listener", v)
}
apiServers = append(apiServers, func() { go server.Serve(l) })
}
}
apiServers = append(apiServers, func() { go server.Serve(ln.ApiListener) })
return apiServers, nil
}
func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error) {
// Clear out in case this is a second start of the controller
ln.Mux.UnregisterProto(alpnmux.DefaultProto)
l, err := ln.Mux.RegisterProto(alpnmux.DefaultProto, &tls.Config{
l := tls.NewListener(ln.ClusterListener, &tls.Config{
GetConfigForClient: c.validateWorkerTls,
})
if err != nil {
return nil, fmt.Errorf("error getting sub-listener for worker proto: %w", err)
}
workerReqInterceptor, err := workerRequestInfoInterceptor(c.baseContext, c.conf.Eventer)
if err != nil {
@@ -153,11 +128,9 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
pbs.RegisterServerCoordinationServiceServer(workerServer, workerService)
pbs.RegisterSessionServiceServer(workerServer, workerService)
interceptor := newInterceptingListener(c, l)
ln.ALPNListener = interceptor
ln.GrpcServer = workerServer
return func() { go ln.GrpcServer.Serve(ln.ALPNListener) }, nil
return func() { go ln.GrpcServer.Serve(newInterceptingListener(c, l)) }, nil
}
func (c *Controller) stopServersAndListeners() error {
@@ -183,12 +156,12 @@ func (c *Controller) stopClusterGrpcServerAndListener() error {
if c.clusterListener.GrpcServer == nil {
return fmt.Errorf("no cluster grpc server")
}
if c.clusterListener.Mux == nil {
return fmt.Errorf("no cluster listener mux")
if c.clusterListener.ClusterListener == nil {
return fmt.Errorf("no cluster listener")
}
c.clusterListener.GrpcServer.GracefulStop()
err := c.clusterListener.Mux.Close()
err := c.clusterListener.ClusterListener.Close()
return listenerCloseErrorCheck(c.clusterListener.Config.Type, err)
}
@@ -204,7 +177,7 @@ func (c *Controller) stopHttpServersAndListeners() error {
ln.HTTPServer.Shutdown(ctx)
cancel()
err := ln.Mux.Close() // The HTTP Shutdown call should close this, but just in case.
err := ln.ApiListener.Close() // The HTTP Shutdown call should close this, but just in case.
err = listenerCloseErrorCheck(ln.Config.Type, err)
if err != nil {
multierror.Append(closeErrors, err)
@@ -231,11 +204,11 @@ func (c *Controller) stopAnyListeners() error {
var closeErrors *multierror.Error
for i := range c.apiListeners {
ln := c.apiListeners[i]
if ln == nil || ln.Mux == nil {
if ln == nil || ln.ApiListener == nil {
continue
}
err := ln.Mux.Close()
err := ln.ApiListener.Close()
err = listenerCloseErrorCheck(ln.Config.Type, err)
if err != nil {
multierror.Append(closeErrors, err)
internal/servers/controller/listeners_test.go
+35 -41
View File
@@ -20,7 +20,6 @@ import (
"time"
"github.com/hashicorp/boundary/internal/cmd/base"
"github.com/hashicorp/boundary/internal/libs/alpnmux"
"github.com/hashicorp/go-secure-stdlib/base62"
"github.com/hashicorp/go-secure-stdlib/configutil/v2"
"github.com/hashicorp/go-secure-stdlib/listenerutil"
@@ -256,9 +255,9 @@ func TestStartListeners(t *testing.T) {
apiAddrs := make([]string, 0)
for _, l := range c.apiListeners {
apiAddrs = append(apiAddrs, l.Mux.Addr().String())
apiAddrs = append(apiAddrs, l.ApiListener.Addr().String())
}
tt.assertions(t, c, apiAddrs, c.clusterListener.Mux.Addr().String())
tt.assertions(t, c, apiAddrs, c.clusterListener.ClusterListener.Addr().String())
})
}
}
@@ -297,7 +296,7 @@ func TestStopClusterGrpcServerAndListener(t *testing.T) {
}
},
expErr: true,
expErrStr: "no cluster listener mux",
expErrStr: "no cluster listener",
},
{
name: "listener already closed",
@@ -308,15 +307,14 @@ func TestStopClusterGrpcServerAndListener(t *testing.T) {
return &Controller{
clusterListener: &base.ServerListener{
ALPNListener: l,
GrpcServer: grpc.NewServer(),
Mux: alpnmux.New(l),
Config: &listenerutil.ListenerConfig{Type: "tcp"},
ClusterListener: l,
GrpcServer: grpc.NewServer(),
Config: &listenerutil.ListenerConfig{Type: "tcp"},
},
}
},
assertions: func(t *testing.T, c *Controller) {
require.ErrorIs(t, c.clusterListener.Mux.Close(), net.ErrClosed)
require.ErrorIs(t, c.clusterListener.ClusterListener.Close(), net.ErrClosed)
},
expErr: false,
},
@@ -339,15 +337,14 @@ func TestStopClusterGrpcServerAndListener(t *testing.T) {
return &Controller{
clusterListener: &base.ServerListener{
ALPNListener: l,
GrpcServer: grpcServer,
Mux: alpnmux.New(l),
Config: &listenerutil.ListenerConfig{Type: "tcp"},
ClusterListener: l,
GrpcServer: grpcServer,
Config: &listenerutil.ListenerConfig{Type: "tcp"},
},
}
},
assertions: func(t *testing.T, c *Controller) {
require.ErrorIs(t, c.clusterListener.Mux.Close(), net.ErrClosed)
require.ErrorIs(t, c.clusterListener.ClusterListener.Close(), net.ErrClosed)
},
expErr: false,
},
@@ -421,20 +418,19 @@ func TestStopHttpServersAndListeners(t *testing.T) {
baseContext: context.Background(),
apiListeners: []*base.ServerListener{
{
ALPNListener: l1,
HTTPServer: s1,
Mux: alpnmux.New(l1),
Config: &listenerutil.ListenerConfig{Type: "tcp"},
ApiListener: l1,
HTTPServer: s1,
Config: &listenerutil.ListenerConfig{Type: "tcp"},
},
},
}
},
assertions: func(t *testing.T, c *Controller) {
// Asserts the HTTP Servers are closed.
require.ErrorIs(t, c.apiListeners[0].HTTPServer.Serve(c.apiListeners[0].ALPNListener), http.ErrServerClosed)
require.ErrorIs(t, c.apiListeners[0].HTTPServer.Serve(c.apiListeners[0].ApiListener), http.ErrServerClosed)
// Asserts the underlying listeners are closed.
require.ErrorIs(t, c.apiListeners[0].Mux.Close(), net.ErrClosed)
require.ErrorIs(t, c.apiListeners[0].ApiListener.Close(), net.ErrClosed)
},
expErr: false,
},
@@ -468,28 +464,26 @@ func TestStopHttpServersAndListeners(t *testing.T) {
baseContext: context.Background(),
apiListeners: []*base.ServerListener{
{
ALPNListener: l1,
HTTPServer: s1,
Mux: alpnmux.New(l1),
Config: &listenerutil.ListenerConfig{Type: "tcp"},
ApiListener: l1,
HTTPServer: s1,
Config: &listenerutil.ListenerConfig{Type: "tcp"},
},
{
ALPNListener: l2,
HTTPServer: s2,
Mux: alpnmux.New(l2),
Config: &listenerutil.ListenerConfig{Type: "tcp"},
ApiListener: l2,
HTTPServer: s2,
Config: &listenerutil.ListenerConfig{Type: "tcp"},
},
},
}
},
assertions: func(t *testing.T, c *Controller) {
// Asserts the HTTP Servers are closed.
require.ErrorIs(t, c.apiListeners[0].HTTPServer.Serve(c.apiListeners[0].ALPNListener), http.ErrServerClosed)
require.ErrorIs(t, c.apiListeners[1].HTTPServer.Serve(c.apiListeners[1].ALPNListener), http.ErrServerClosed)
require.ErrorIs(t, c.apiListeners[0].HTTPServer.Serve(c.apiListeners[0].ApiListener), http.ErrServerClosed)
require.ErrorIs(t, c.apiListeners[1].HTTPServer.Serve(c.apiListeners[1].ApiListener), http.ErrServerClosed)
// Asserts the underlying listeners are closed.
require.ErrorIs(t, c.apiListeners[0].Mux.Close(), net.ErrClosed)
require.ErrorIs(t, c.apiListeners[1].Mux.Close(), net.ErrClosed)
require.ErrorIs(t, c.apiListeners[0].ApiListener.Close(), net.ErrClosed)
require.ErrorIs(t, c.apiListeners[1].ApiListener.Close(), net.ErrClosed)
},
expErr: false,
},
@@ -601,10 +595,10 @@ func TestStopAnyListeners(t *testing.T) {
expErr: false,
},
{
name: "listeners with nil mux",
name: "non-empty listeners with nil listeners",
controllerFn: func(t *testing.T) *Controller {
return &Controller{apiListeners: []*base.ServerListener{
{Mux: nil}, {Mux: nil}, {Mux: nil},
{ClusterListener: nil}, {ClusterListener: nil}, {ClusterListener: nil},
}}
},
expErr: false,
@@ -624,23 +618,23 @@ func TestStopAnyListeners(t *testing.T) {
return &Controller{apiListeners: []*base.ServerListener{
{
Config: &listenerutil.ListenerConfig{Type: "tcp"},
Mux: alpnmux.New(l1),
Config: &listenerutil.ListenerConfig{Type: "tcp"},
ApiListener: l1,
},
{
Config: &listenerutil.ListenerConfig{Type: "tcp"},
Mux: alpnmux.New(l2),
Config: &listenerutil.ListenerConfig{Type: "tcp"},
ApiListener: l2,
},
{
Config: &listenerutil.ListenerConfig{Type: "tcp"},
Mux: alpnmux.New(l3),
Config: &listenerutil.ListenerConfig{Type: "tcp"},
ApiListener: l3,
},
}}
},
assertions: func(t *testing.T, c *Controller) {
for i := range c.apiListeners {
ln := c.apiListeners[i]
require.ErrorIs(t, ln.Mux.Close(), net.ErrClosed)
require.ErrorIs(t, ln.ApiListener.Close(), net.ErrClosed)
}
},
expErr: false,
internal/servers/controller/testing.go
+9 -1
View File
@@ -219,7 +219,15 @@ func (tc *TestController) addrs(purpose string) []string {
addrs := make([]string, 0, len(tc.b.Listeners))
for _, listener := range tc.b.Listeners {
if listener.Config.Purpose[0] == purpose {
addr := listener.Mux.Addr()
var addr net.Addr
switch purpose {
case "api":
addr = listener.ApiListener.Addr()
case "cluster":
addr = listener.ClusterListener.Addr()
case "ops":
addr = listener.OpsListener.Addr()
}
switch {
case strings.HasPrefix(addr.String(), "/"):
switch purpose {
internal/servers/worker/listeners.go
+7 -17
View File
@@ -12,7 +12,6 @@ import (
"time"
"github.com/hashicorp/boundary/internal/cmd/base"
"github.com/hashicorp/boundary/internal/libs/alpnmux"
"github.com/hashicorp/boundary/internal/observability/event"
"github.com/hashicorp/go-multierror"
)
@@ -24,7 +23,7 @@ func (w *Worker) startListeners() error {
if e == nil {
return fmt.Errorf("%s: sys eventer not initialized", op)
}
logger, err := e.StandardLogger(w.baseContext, "listeners", event.ErrorType)
logger, err := e.StandardLogger(w.baseContext, "worker.listeners: ", event.ErrorType)
if err != nil {
return fmt.Errorf("%s: unable to initialize std logger: %w", op, err)
}
@@ -46,7 +45,7 @@ func (w *Worker) startListeners() error {
return nil
}
func (w *Worker) configureForWorker(ln *base.ServerListener, log *log.Logger) (func(), error) {
func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger) (func(), error) {
handler, err := w.handler(HandlerProperties{ListenerConfig: ln.Config})
if err != nil {
return nil, err
@@ -57,7 +56,7 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, log *log.Logger) (f
Handler: handler,
ReadHeaderTimeout: 10 * time.Second,
ReadTimeout: 30 * time.Second,
ErrorLog: log,
ErrorLog: logger,
BaseContext: func(net.Listener) context.Context {
return cancelCtx
},
@@ -77,18 +76,9 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, log *log.Logger) (f
server.IdleTimeout = ln.Config.HTTPIdleTimeout
}
// Clear out in case this is a second start of the controller
ln.Mux.UnregisterProto(alpnmux.DefaultProto)
ln.Mux.UnregisterProto(alpnmux.NoProto)
l, err := ln.Mux.RegisterProto(alpnmux.DefaultProto, &tls.Config{
l := tls.NewListener(ln.ProxyListener, &tls.Config{
GetConfigForClient: w.getSessionTls,
})
if err != nil {
return nil, fmt.Errorf("error getting tls listener: %w", err)
}
if l == nil {
return nil, errors.New("could not get tls listener")
}
return func() { go server.Serve(l) }, nil
}
@@ -120,7 +110,7 @@ func (w *Worker) stopHttpServersAndListeners() error {
ln.HTTPServer.Shutdown(ctx)
cancel()
err := ln.Mux.Close()
err := ln.ProxyListener.Close()
err = listenerCloseErrorCheck(ln.Config.Type, err)
if err != nil {
multierror.Append(closeErrors, err)
@@ -136,11 +126,11 @@ func (w *Worker) stopHttpServersAndListeners() error {
func (w *Worker) stopAnyListeners() error {
var closeErrors *multierror.Error
for _, ln := range w.listeners {
if ln == nil || ln.Mux == nil {
if ln == nil || ln.ProxyListener == nil {
continue
}
err := ln.Mux.Close()
err := ln.ProxyListener.Close()
err = listenerCloseErrorCheck(ln.Config.Type, err)
if err != nil {
multierror.Append(closeErrors, err)
internal/servers/worker/listeners_test.go
+39 -33
View File
@@ -12,12 +12,20 @@ import (
"time"
"github.com/hashicorp/boundary/internal/cmd/base"
"github.com/hashicorp/boundary/internal/libs/alpnmux"
"github.com/hashicorp/go-secure-stdlib/listenerutil"
"github.com/stretchr/testify/require"
)
func TestStartListeners(t *testing.T) {
testNonTlsRejected := func(t *testing.T, resp *http.Response, err error) {
require.NoError(t, err)
require.NotNil(t, resp)
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
require.NoError(t, err)
require.Equal(t, "Client sent an HTTP request to an HTTPS server.\n", string(body))
}
tests := []struct {
name string
listeners []*listenerutil.ListenerConfig
@@ -45,16 +53,16 @@ func TestStartListeners(t *testing.T) {
assertions: func(t *testing.T, w *Worker, addrs []string) {
require.Len(t, addrs, 2)
_, err := http.Get("http://" + addrs[0] + "/v1/proxy/")
require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation
resp, err := http.Get("http://" + addrs[0] + "/v1/proxy/")
testNonTlsRejected(t, resp, err)
cl := http.Client{
Transport: &http.Transport{
Dial: func(network, addr string) (net.Conn, error) { return net.Dial("unix", addrs[1]) },
},
}
_, err = cl.Get("http://anything.domain/v1/proxy/")
require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation
resp, err = cl.Get("http://anything.domain/v1/proxy/")
testNonTlsRejected(t, resp, err)
},
},
{
@@ -89,11 +97,11 @@ func TestStartListeners(t *testing.T) {
assertions: func(t *testing.T, w *Worker, addrs []string) {
require.Len(t, addrs, 4)
_, err := http.Get("http://" + addrs[0] + "/v1/proxy/")
require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation
resp, err := http.Get("http://" + addrs[0] + "/v1/proxy/")
testNonTlsRejected(t, resp, err)
_, err = http.Get("http://" + addrs[1] + "/v1/proxy/")
require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation
resp, err = http.Get("http://" + addrs[1] + "/v1/proxy/")
testNonTlsRejected(t, resp, err)
for _, proxyAddr := range []string{addrs[2], addrs[3]} {
cl := http.Client{
@@ -101,8 +109,8 @@ func TestStartListeners(t *testing.T) {
Dial: func(network, addr string) (net.Conn, error) { return net.Dial("unix", proxyAddr) },
},
}
_, err = cl.Get("http://anything.domain/v1/proxy")
require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation
resp, err = cl.Get("http://anything.domain/v1/proxy")
testNonTlsRejected(t, resp, err)
}
},
},
@@ -130,7 +138,7 @@ func TestStartListeners(t *testing.T) {
addrs := make([]string, 0)
for _, l := range w.listeners {
addrs = append(addrs, l.Mux.Addr().String())
addrs = append(addrs, l.ProxyListener.Addr().String())
}
if tt.assertions != nil {
tt.assertions(t, w, addrs)
@@ -204,28 +212,26 @@ func TestStopHttpServersAndListeners(t *testing.T) {
baseContext: context.Background(),
listeners: []*base.ServerListener{
{
ALPNListener: l1,
HTTPServer: s1,
Mux: alpnmux.New(l1),
Config: &listenerutil.ListenerConfig{Type: "tcp"},
ProxyListener: l1,
HTTPServer: s1,
Config: &listenerutil.ListenerConfig{Type: "tcp"},
},
{
ALPNListener: l2,
HTTPServer: s2,
Mux: alpnmux.New(l2),
Config: &listenerutil.ListenerConfig{Type: "tcp"},
ProxyListener: l2,
HTTPServer: s2,
Config: &listenerutil.ListenerConfig{Type: "tcp"},
},
},
}
},
assertions: func(t *testing.T, w *Worker) {
// Asserts the HTTP Servers are closed.
require.ErrorIs(t, w.listeners[0].HTTPServer.Serve(w.listeners[0].ALPNListener), http.ErrServerClosed)
require.ErrorIs(t, w.listeners[1].HTTPServer.Serve(w.listeners[1].ALPNListener), http.ErrServerClosed)
require.ErrorIs(t, w.listeners[0].HTTPServer.Serve(w.listeners[0].ProxyListener), http.ErrServerClosed)
require.ErrorIs(t, w.listeners[1].HTTPServer.Serve(w.listeners[1].ProxyListener), http.ErrServerClosed)
// Asserts the underlying listeners are closed.
require.ErrorIs(t, w.listeners[0].Mux.Close(), net.ErrClosed)
require.ErrorIs(t, w.listeners[1].Mux.Close(), net.ErrClosed)
require.ErrorIs(t, w.listeners[0].ProxyListener.Close(), net.ErrClosed)
require.ErrorIs(t, w.listeners[1].ProxyListener.Close(), net.ErrClosed)
},
expErr: false,
},
@@ -278,10 +284,10 @@ func TestStopAnyListeners(t *testing.T) {
expErr: false,
},
{
name: "listeners with nil mux",
name: "non-empty but nil listeners",
workerFn: func(t *testing.T) *Worker {
return &Worker{listeners: []*base.ServerListener{
{Mux: nil}, {Mux: nil}, {Mux: nil},
{ProxyListener: nil}, {ProxyListener: nil}, {ProxyListener: nil},
}}
},
expErr: false,
@@ -301,23 +307,23 @@ func TestStopAnyListeners(t *testing.T) {
return &Worker{listeners: []*base.ServerListener{
{
Config: &listenerutil.ListenerConfig{Type: "tcp"},
Mux: alpnmux.New(l1),
Config: &listenerutil.ListenerConfig{Type: "tcp"},
ProxyListener: l1,
},
{
Config: &listenerutil.ListenerConfig{Type: "tcp"},
Mux: alpnmux.New(l2),
Config: &listenerutil.ListenerConfig{Type: "tcp"},
ProxyListener: l2,
},
{
Config: &listenerutil.ListenerConfig{Type: "tcp"},
Mux: alpnmux.New(l3),
Config: &listenerutil.ListenerConfig{Type: "tcp"},
ProxyListener: l3,
},
}}
},
assertions: func(t *testing.T, w *Worker) {
for i := range w.listeners {
ln := w.listeners[i]
require.ErrorIs(t, ln.Mux.Close(), net.ErrClosed)
require.ErrorIs(t, ln.ProxyListener.Close(), net.ErrClosed)
}
},
expErr: false,
internal/servers/worker/testing.go
+1 -1
View File
@@ -68,7 +68,7 @@ func (tw *TestWorker) ProxyAddrs() []string {
for _, listener := range tw.b.Listeners {
if listener.Config.Purpose[0] == "proxy" {
tcpAddr, ok := listener.Mux.Addr().(*net.TCPAddr)
tcpAddr, ok := listener.ProxyListener.Addr().(*net.TCPAddr)
if !ok {
tw.t.Fatal("could not parse address as a TCP addr")
}