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

refactor(worker): Ensure startListeners consistency with controller

Browse Source 
Implements changes on Worker's `startListeners` func to be consistent
with the changes on the Controller's side done in
b5ef468d33.
This commit is contained in:
Hugo Vieira committed 2022-03-29 16:45:56 +01:00
1 parent 1ebd7deccf
commit 657b099440
2 files changed
+191 -63
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/servers/worker/listeners.go
+57 -63
View File
@@ -5,12 +5,14 @@ import (
"crypto/tls"
"errors"
"fmt"
"log"
"net"
"net/http"
"os"
"sync"
"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"
@@ -18,7 +20,7 @@ import (
func (w *Worker) startListeners() error {
const op = "worker.(Worker).startListeners"
servers := make([]func(), 0, len(w.conf.Listeners))
e := event.SysEventer()
if e == nil {
return fmt.Errorf("%s: sys eventer not initialized", op)
@@ -27,70 +29,15 @@ func (w *Worker) startListeners() error {
if err != nil {
return fmt.Errorf("%s: unable to initialize std logger: %w", op, err)
}
for _, ln := range w.conf.Listeners {
for _, purpose := range ln.Config.Purpose {
switch purpose {
case "api", "cluster", "ops":
// We may have this in dev mode; ignore
continue
case "proxy":
// Do nothing; handle below
default:
return fmt.Errorf("unknown listener purpose %q", purpose)
}
handler, err := w.handler(HandlerProperties{
ListenerConfig: ln.Config,
})
if err != nil {
return fmt.Errorf("%s: %w", op, err)
}
cancelCtx := w.baseContext
server := &http.Server{
Handler: handler,
ReadHeaderTimeout: 10 * time.Second,
ReadTimeout: 30 * time.Second,
ErrorLog: logger,
BaseContext: func(net.Listener) context.Context {
return cancelCtx
},
}
ln.HTTPServer = server
if ln.Config.HTTPReadHeaderTimeout > 0 {
server.ReadHeaderTimeout = ln.Config.HTTPReadHeaderTimeout
}
if ln.Config.HTTPReadTimeout > 0 {
server.ReadTimeout = ln.Config.HTTPReadTimeout
}
if ln.Config.HTTPWriteTimeout > 0 {
server.WriteTimeout = ln.Config.HTTPWriteTimeout
}
if ln.Config.HTTPIdleTimeout > 0 {
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{
GetConfigForClient: w.getSessionTls,
})
if err != nil {
return fmt.Errorf("error getting tls listener: %w", err)
}
if l == nil {
return errors.New("could not get tls listener")
}
servers = append(servers, func() {
go server.Serve(l)
})
servers := make([]func(), 0, len(w.listeners))
for i := range w.listeners {
ln := w.listeners[i]
workerServer, err := w.configureForWorker(ln, logger)
if err != nil {
return fmt.Errorf("%s: failed to configure for worker: %w", op, err)
}
servers = append(servers, workerServer)
}
for _, s := range servers {
@@ -100,6 +47,53 @@ func (w *Worker) startListeners() error {
return nil
}
func (w *Worker) configureForWorker(ln *base.ServerListener, log *log.Logger) (func(), error) {
handler, err := w.handler(HandlerProperties{ListenerConfig: ln.Config})
if err != nil {
return nil, err
}
cancelCtx := w.baseContext
server := &http.Server{
Handler: handler,
ReadHeaderTimeout: 10 * time.Second,
ReadTimeout: 30 * time.Second,
ErrorLog: log,
BaseContext: func(net.Listener) context.Context {
return cancelCtx
},
}
ln.HTTPServer = server
if ln.Config.HTTPReadHeaderTimeout > 0 {
server.ReadHeaderTimeout = ln.Config.HTTPReadHeaderTimeout
}
if ln.Config.HTTPReadTimeout > 0 {
server.ReadTimeout = ln.Config.HTTPReadTimeout
}
if ln.Config.HTTPWriteTimeout > 0 {
server.WriteTimeout = ln.Config.HTTPWriteTimeout
}
if ln.Config.HTTPIdleTimeout > 0 {
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{
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
}
func (w *Worker) stopListeners() error {
serverWg := new(sync.WaitGroup)
for _, ln := range w.conf.Listeners {
internal/servers/worker/listeners_test.go
+134
View File
@@ -0,0 +1,134 @@
package worker
import (
"context"
"io"
"net"
"net/http"
"testing"
"github.com/hashicorp/go-secure-stdlib/listenerutil"
"github.com/stretchr/testify/require"
)
func TestStartListeners(t *testing.T) {
tests := []struct {
name string
listeners []*listenerutil.ListenerConfig
expErr bool
expErrMsg string
assertions func(t *testing.T, w *Worker, addrs []string)
}{
{
name: "one tcp listener, one unix listener",
listeners: []*listenerutil.ListenerConfig{
{
Type: "tcp",
Purpose: []string{"proxy"},
Address: "127.0.0.1:0",
TLSDisable: true,
},
{
Type: "unix",
Purpose: []string{"proxy"},
Address: "/tmp/boundary-worker-test-start-listeners.sock",
TLSDisable: true,
},
},
expErr: false,
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
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
},
},
{
name: "multiple tcp listeners, multiple unix listeners",
listeners: []*listenerutil.ListenerConfig{
{
Type: "tcp",
Purpose: []string{"proxy"},
Address: "127.0.0.1:0",
TLSDisable: true,
},
{
Type: "tcp",
Purpose: []string{"proxy"},
Address: "127.0.0.1:0",
TLSDisable: true,
},
{
Type: "unix",
Purpose: []string{"proxy"},
Address: "/tmp/boundary-worker-test-start-listeners0.sock",
TLSDisable: true,
},
{
Type: "unix",
Purpose: []string{"proxy"},
Address: "/tmp/boundary-worker-test-start-listeners1.sock",
TLSDisable: true,
},
},
expErr: false,
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
_, err = http.Get("http://" + addrs[1] + "/v1/proxy/")
require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation
for _, proxyAddr := range []string{addrs[2], addrs[3]} {
cl := http.Client{
Transport: &http.Transport{
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
}
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
conf := NewTestWorker(t, &TestWorkerOpts{DisableAutoStart: true}).w.conf
conf.RawConfig.SharedConfig.Listeners = tt.listeners
err := conf.SetupListeners(nil, conf.RawConfig.SharedConfig, []string{"proxy"})
require.NoError(t, err)
w, err := New(conf)
require.NoError(t, err)
w.baseContext = context.Background()
err = w.startListeners()
if tt.expErr {
require.EqualError(t, err, tt.expErrMsg)
return
}
require.NoError(t, err)
addrs := make([]string, 0)
for _, l := range w.listeners {
addrs = append(addrs, l.Mux.Addr().String())
}
if tt.assertions != nil {
tt.assertions(t, w, addrs)
}
})
}
}