OSS side of second downstream connections (see enterprise PR 69) (#2292)

Makefile

@@ -248,7 +248,7 @@ generate-database-dumps:
 test-ci: export CI_BUILD=1
 test-ci:
 	CGO_ENABLED=$(CGO_ENABLED) BUILD_TAGS='$(BUILD_TAGS)' sh -c "'$(CURDIR)/scripts/build.sh'"
-	~/.go/bin/go test ./... -v $(TESTARGS) -json -cover -timeout 120m | tparse
+	~/.go/bin/go test ./... -v $(TESTARGS) -json -cover -timeout 120m | tparse -follow
 
 .PHONY: test-sql
 test-sql:

internal/daemon/common/const.go

@@ -0,0 +1,3 @@
+package common
+
+const ReverseGrpcConnectionAlpnValue = "the-downstream-dialer-plays-an-uno-reverse-card"

internal/daemon/common/eventing_listener.go

@@ -0,0 +1,62 @@
+package common
+
+import (
+	"context"
+	"crypto/tls"
+	"net"
+
+	"github.com/hashicorp/boundary/internal/errors"
+	"github.com/hashicorp/boundary/internal/observability/event"
+	nodee "github.com/hashicorp/nodeenrollment"
+)
+
+// EventingListener simply sends an event when a worker has connected
+// successfully
+type eventingListener struct {
+	ctx    context.Context
+	baseLn net.Listener
+}
+
+func NewEventingListener(ctx context.Context, baseLn net.Listener) (net.Listener, error) {
+	const op = "common.(EventingListener).Accept"
+	switch {
+	case ctx == nil:
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "nil context")
+	case baseLn == nil:
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "nil base listener")
+	}
+	return &eventingListener{
+		ctx:    ctx,
+		baseLn: baseLn,
+	}, nil
+}
+
+func (e *eventingListener) Accept() (net.Conn, error) {
+	const op = "common.(EventingListener).Accept"
+	conn, err := e.baseLn.Accept()
+	if err != nil || conn == nil {
+		return conn, err
+	}
+
+	tlsConn, ok := conn.(*tls.Conn)
+	if !ok {
+		return conn, err
+	}
+
+	if tlsConn != nil && len(tlsConn.ConnectionState().PeerCertificates) > 0 {
+		keyId, err := nodee.KeyIdFromPkix(tlsConn.ConnectionState().PeerCertificates[0].SubjectKeyId)
+		if err == nil {
+			event.WriteSysEvent(e.ctx, op, "worker successfully authenticated", "key_id", keyId)
+		}
+	}
+
+	return conn, err
+}
+
+func (e *eventingListener) Close() error {
+	return e.baseLn.Close()
+}
+
+func (e *eventingListener) Addr() net.Addr {
+	return e.baseLn.Addr()
+}

internal/daemon/controller/intercepting_listener.go

@@ -1,145 +0,0 @@
-package controller
-
-import (
-	"crypto/tls"
-	"errors"
-	"fmt"
-	"net"
-	"strings"
-
-	"github.com/hashicorp/boundary/internal/observability/event"
-	nodee "github.com/hashicorp/nodeenrollment"
-	"github.com/hashicorp/nodeenrollment/protocol"
-)
-
-// tempError is an error that satisfies the temporary error interface that is
-// internally used by gRPC to determine whether an error should cause a listener
-// to die. Any error that isn't an accept error is wrapped in this since one
-// connection failing TLS wise doesn't mean we don't want to accept any more...
-type tempError struct {
-	error
-}
-
-// newTempError is a "temporary" error
-func newTempError(inner error) tempError {
-	return tempError{error: inner}
-}
-
-func (t tempError) Temporary() bool {
-	return true
-}
-
-// interceptingListener allows us to validate the nonce from a connection before
-// handing it off to the gRPC server. It is expected that the first thing a
-// connection sends after successful TLS validation is the nonce that was
-// contained in the KMS-encrypted TLS info. The reason is for replay attacks --
-// since the ALPN NextProtos are sent in the clear, we don't want anyone
-// sniffing the connection to simply replay the hello message and gain access.
-// By requiring the information encrypted within that message to match the first
-// bytes sent in the connection itself, we require that the service making the
-// incoming connection had to either be the service that did the initial
-// encryption, or had to be able to decrypt that against the same KMS key. This
-// means that KMS access is a requirement, and simple replay itself is not
-// sufficient.
-//
-// Note that this is semi-weak against a scenario where the value is decrypted
-// later since a controller restart would clear the cache. We could store a list
-// of seen nonces in the database, but since the original certificate was only
-// good for 3 minutes and 30 seconds anyways, the decryption would need to
-// happen within a short time window instead of much later. We can adjust this
-// window if we want (or even make it tunable), or store values in the DB as
-// well until the certificate expiration.
-type interceptingListener struct {
-	baseLn net.Listener
-	c      *Controller
-}
-
-func newInterceptingListener(c *Controller, baseLn net.Listener) *interceptingListener {
-	ret := &interceptingListener{
-		c:      c,
-		baseLn: baseLn,
-	}
-
-	return ret
-}
-
-func (m *interceptingListener) Accept() (net.Conn, error) {
-	const op = "controller.(interceptingListener).Accept"
-	ctx := m.c.baseContext
-	conn, err := m.baseLn.Accept()
-	if err != nil {
-		// We may already be closed but can't be sure, so just try anyways
-		if conn != nil {
-			if err := conn.Close(); err != nil {
-				event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
-			}
-		}
-		return nil, newTempError(err)
-	}
-
-	var tlsConn *tls.Conn
-	switch c := conn.(type) {
-	case *protocol.Conn:
-		// If we so choose, at this point we can pull out the client's
-		// NextProtos with c.ClientNextProtos
-		tlsConn = c.Conn
-	case *tls.Conn:
-		tlsConn = c
-	default:
-		err := fmt.Errorf("unknown connection type %T received", c)
-		event.WriteError(ctx, op, err)
-		return nil, newTempError(err)
-	}
-
-	switch {
-	case nodee.ContainsKnownAlpnProto(tlsConn.ConnectionState().NegotiatedProtocol):
-		keyId, err := nodee.KeyIdFromPkix(tlsConn.ConnectionState().PeerCertificates[0].SubjectKeyId)
-		if err != nil {
-			if err := conn.Close(); err != nil {
-				event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
-			}
-			return nil, newTempError(fmt.Errorf("error deriving key id from newly-authenticated node: %w", err))
-		}
-		event.WriteSysEvent(ctx, op, "worker successfully authenticated", "key_id", keyId)
-		return conn, nil
-
-	case strings.HasPrefix(tlsConn.ConnectionState().NegotiatedProtocol, "v1workerauth"):
-		nonce := make([]byte, 20)
-		read, err := conn.Read(nonce)
-		if err != nil {
-			if err := conn.Close(); err != nil {
-				event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
-			}
-			return nil, newTempError(fmt.Errorf("error reading nonce from connection: %w", err))
-		}
-		if read != len(nonce) {
-			if err := conn.Close(); err != nil {
-				event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
-			}
-			return nil, newTempError(fmt.Errorf("error reading nonce from worker, expected %d bytes, got %d", 20, read))
-		}
-		workerInfoRaw, found := m.c.workerAuthCache.Load(string(nonce))
-		if !found {
-			if err := conn.Close(); err != nil {
-				event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
-			}
-			return nil, newTempError(errors.New("did not find valid nonce for incoming worker"))
-		}
-		workerInfo := workerInfoRaw.(*workerAuthEntry)
-		workerInfo.conn = tlsConn
-		m.c.workerAuthCache.Delete(string(nonce))
-		event.WriteSysEvent(ctx, op, "worker successfully authed", "name", workerInfo.Name, "description", workerInfo.Description, "proxy_address", workerInfo.ProxyAddress)
-		return protocol.NewConn(tlsConn, workerInfo.clientNextProtos), nil
-
-	default:
-		return nil, newTempError(errors.New("unable to authenticate incoming connection"))
-	}
-}
-
-func (m *interceptingListener) Close() error {
-	return m.baseLn.Close()
-}
-
-func (m *interceptingListener) Addr() net.Addr {
-	return m.baseLn.Addr()
-}

internal/daemon/controller/kms_conn_router.go

@@ -0,0 +1,152 @@
+package controller
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"net"
+	"strings"
+
+	"github.com/hashicorp/boundary/internal/daemon/common"
+	"github.com/hashicorp/boundary/internal/errors"
+	"github.com/hashicorp/boundary/internal/observability/event"
+	nodeenet "github.com/hashicorp/nodeenrollment/net"
+)
+
+// tempError is an error that satisfies the temporary error interface that is
+// internally used by gRPC to determine whether an error should cause a listener
+// to die. Any error that isn't an accept error is wrapped in this since one
+// connection failing TLS wise doesn't mean we don't want to accept any more...
+type tempError struct {
+	error
+}
+
+// newTempError is a "temporary" error
+func newTempError(inner error) tempError {
+	return tempError{error: inner}
+}
+
+func (t tempError) Temporary() bool {
+	return true
+}
+
+// startKmsConnRouter allows us to validate the nonce from a connection before
+// handing it off to other purposes, such as the gRPC server. It is expected
+// that the first thing a connection sends after successful TLS validation is
+// the nonce that was contained in the KMS-encrypted TLS info. The reason is for
+// replay attacks -- since the ALPN NextProtos are sent in the clear, we don't
+// want anyone sniffing the connection to simply replay the hello message and
+// gain access. By requiring the information encrypted within that message to
+// match the first bytes sent in the connection itself, we require that the
+// service making the incoming connection had to either be the service that did
+// the initial encryption, or had to be able to decrypt that against the same
+// KMS key. This means that KMS access is a requirement, and simple replay
+// itself is not sufficient.
+func startKmsConnRouter(
+	ctx context.Context,
+	c *Controller,
+	baseLn net.Listener,
+	authedListener,
+	reverseGrpcListener *nodeenet.MultiplexingListener,
+) error {
+	const op = "controller.startKmsAuthRouter"
+	switch {
+	case c == nil:
+		return errors.New(ctx, errors.InvalidParameter, op, "nil Controller")
+	case baseLn == nil:
+		return errors.New(ctx, errors.InvalidParameter, op, "nil baseLn")
+	case authedListener == nil:
+		return errors.New(ctx, errors.InvalidParameter, op, "nil authedListener")
+	case reverseGrpcListener == nil:
+		return errors.New(ctx, errors.InvalidParameter, op, "nil reverseGrpcListener")
+	}
+	go func() {
+		for {
+			conn, err := baseLn.Accept()
+			if err != nil {
+				if err == net.ErrClosed {
+					// We're done
+					return
+				}
+				// Conn may already be closed but can't be sure, so just try anyways
+				if conn != nil {
+					if err := conn.Close(); err != nil {
+						event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
+					}
+				}
+				event.WriteError(ctx, op, err, event.WithInfoMsg("error accepting kms connection"))
+				continue
+			}
+
+			if conn == nil {
+				// No idea why this would happen, but be safe before we operate
+				// on it below
+				event.WriteError(ctx, op, fmt.Errorf("connection is nil"))
+				continue
+			}
+
+			tlsConn, ok := conn.(*tls.Conn)
+			if !ok {
+				// This should never happen, but is just to be safe
+				event.WriteError(ctx, op, fmt.Errorf("connection is not a *tls.Conn"))
+				_ = conn.Close()
+				continue
+			}
+
+			if !strings.HasPrefix(tlsConn.ConnectionState().NegotiatedProtocol, "v1workerauth") {
+				// If we're here it hasn't been handled by PKI and can't be
+				// handled here. We should never actually get here...
+				event.WriteError(ctx, op, fmt.Errorf("connection is not a kms connection"))
+				_ = conn.Close()
+				continue
+			}
+
+			nonce := make([]byte, 20)
+			read, err := conn.Read(nonce)
+			if err != nil {
+				if err := conn.Close(); err != nil {
+					event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
+				}
+				event.WriteError(ctx, op, err, event.WithInfoMsg("error reading nonce from connection"))
+				continue
+			}
+			if read != len(nonce) {
+				if err := conn.Close(); err != nil {
+					event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
+				}
+				event.WriteError(ctx, op, fmt.Errorf("error reading nonce from worker, expected %d bytes, got %d", 20, read))
+				continue
+			}
+
+			workerInfoRaw, found := c.workerAuthCache.Load(string(nonce))
+			if !found {
+				if err := conn.Close(); err != nil {
+					event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
+				}
+				event.WriteError(ctx, op, fmt.Errorf("did not find valid nonce for incoming worker"))
+				continue
+			}
+
+			workerInfo := workerInfoRaw.(*workerAuthEntry)
+			workerInfo.conn = tlsConn
+			c.workerAuthCache.Delete(string(nonce))
+			event.WriteSysEvent(ctx, op, "worker successfully authed", "name", workerInfo.Name, "description", workerInfo.Description, "proxy_address", workerInfo.ProxyAddress)
+
+			found = false
+			for _, proto := range workerInfo.clientNextProtos {
+				if proto == common.ReverseGrpcConnectionAlpnValue {
+					reverseGrpcListener.IngressConn(tlsConn, nil)
+					found = true
+					break
+				}
+			}
+			if found {
+				continue
+			}
+			// Didn't find specific ALPN values, so it's a normal grpc
+			// connection
+			authedListener.IngressConn(tlsConn, nil)
+		}
+	}()
+	return nil
+}

internal/daemon/controller/listeners.go

@@ -3,7 +3,6 @@ package controller
 import (
 	"context"
 	"crypto/tls"
-	"errors"
 	"fmt"
 	"math"
 	"net"
@@ -14,14 +13,27 @@ import (
 	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
 	"github.com/hashicorp/boundary/internal/cmd/base"
 	"github.com/hashicorp/boundary/internal/daemon/cluster/handlers"
+	"github.com/hashicorp/boundary/internal/daemon/common"
 	"github.com/hashicorp/boundary/internal/daemon/controller/internal/metric"
+	"github.com/hashicorp/boundary/internal/errors"
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/go-multierror"
 	"github.com/hashicorp/nodeenrollment/multihop"
+	nodeenet "github.com/hashicorp/nodeenrollment/net"
 	"github.com/hashicorp/nodeenrollment/protocol"
 	"google.golang.org/grpc"
 )
 
+// the function that handles a secondary connection over a provided listener
+var handleSecondaryConnection = closeListener
+
+func closeListener(_ context.Context, l net.Listener) error {
+	if l != nil {
+		return l.Close()
+	}
+	return nil
+}
+
 func (c *Controller) startListeners() error {
 	servers := make([]func(), 0, len(c.conf.Listeners))
 
@@ -112,7 +124,10 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 		return nil, fmt.Errorf("error fetching worker auth storage: %w", err)
 	}
 
-	l, err := protocol.NewInterceptingListener(
+	// The cluster listener is shut down at server shutdown time so we do not
+	// need to handle individual listener shutdown.
+	var l net.Listener
+	l, err = protocol.NewInterceptingListener(
 		&protocol.InterceptingListenerConfiguration{
 			Context:      c.baseContext,
 			Storage:      workerAuthStorage,
@@ -125,9 +140,63 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 		return nil, fmt.Errorf("error instantiating node auth listener: %w", err)
 	}
 
+	// Create split listener
+	splitListener, err := nodeenet.NewSplitListener(l)
+	if err != nil {
+		return nil, fmt.Errorf("error instantiating split listener: %w", err)
+	}
+
+	// This handles connections coming in on the cluster port that are
+	// authenticated via nodeenrollment but not with any extra purpose; these
+	// are normal worker connections
+	nodeeAuthedListener, err := splitListener.GetListener(nodeenet.AuthenticatedNonSpecificNextProto)
+	if err != nil {
+		return nil, fmt.Errorf("error instantiating node enrollment authed split listener: %w", err)
+	}
+	// This wraps the normal worker connections with something that events with
+	// the worker ID on successful auth/connection
+	eventingAuthedListener, err := common.NewEventingListener(c.baseContext, nodeeAuthedListener)
+	if err != nil {
+		return nil, fmt.Errorf("%s: error creating eventing listener: %w", op, err)
+	}
+	// Create a multiplexer to unify connections between PKI and KMS
+	multiplexingAuthedListener, err := nodeenet.NewMultiplexingListener(c.baseContext, nodeeAuthedListener.Addr())
+	if err != nil {
+		return nil, fmt.Errorf("error instantiating authed multiplexed listener: %w", err)
+	}
+	if err := multiplexingAuthedListener.IngressListener(eventingAuthedListener); err != nil {
+		return nil, fmt.Errorf("error adding authed evented listener to multiplexed listener: %w", err)
+	}
+	// Connections that came in on the cluster listener that are not authed via
+	// PKI need to be sent for KMS routing
+	nodeeUnauthedListener, err := splitListener.GetListener(nodeenet.UnauthenticatedNextProto)
+	if err != nil {
+		return nil, fmt.Errorf("error instantiating node enrollment unauthed split listener: %w", err)
+	}
+
+	// Connections coming in here are authed by nodeenrollment and are for the
+	// reverse grpc purpose
+	reverseGrpcListener, err := splitListener.GetListener(common.ReverseGrpcConnectionAlpnValue)
+	if err != nil {
+		return nil, fmt.Errorf("error instantiating reverse gprc connection split listener: %w", err)
+	}
+	// Create a multiplexer to unify reverse grpc connections between PKI and KMS
+	multiplexingReverseGrpcListener, err := nodeenet.NewMultiplexingListener(c.baseContext, reverseGrpcListener.Addr())
+	if err != nil {
+		return nil, fmt.Errorf("error instantiating reverse grpc multiplexed listener: %w", err)
+	}
+	if err := multiplexingReverseGrpcListener.IngressListener(reverseGrpcListener); err != nil {
+		return nil, fmt.Errorf("error adding reverse grpc listener to multiplexed listener: %w", err)
+	}
+
+	// Start routing KMS connections
+	if err := startKmsConnRouter(c.baseContext, c, nodeeUnauthedListener, multiplexingAuthedListener, multiplexingReverseGrpcListener); err != nil {
+		return nil, errors.Wrap(c.baseContext, err, op)
+	}
+
 	workerReqInterceptor, err := workerRequestInfoInterceptor(c.baseContext, c.conf.Eventer)
 	if err != nil {
-		return nil, fmt.Errorf("error getting sub-listener for worker proto: %w", err)
+		return nil, fmt.Errorf("error getting request interceptor for worker proto: %w", err)
 	}
 
 	workerServer := grpc.NewServer(
@@ -162,7 +231,11 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 
 	ln.GrpcServer = workerServer
 
-	return func() { go ln.GrpcServer.Serve(newInterceptingListener(c, l)) }, nil
+	return func() {
+		go splitListener.Start()
+		go handleSecondaryConnection(c.baseContext, multiplexingReverseGrpcListener)
+		go ln.GrpcServer.Serve(multiplexingAuthedListener)
+	}, nil
 }
 
 func (c *Controller) stopServersAndListeners() error {

internal/daemon/worker/addressreceiver.go

@@ -0,0 +1,45 @@
+package worker
+
+import (
+	"context"
+
+	"google.golang.org/grpc/resolver"
+	"google.golang.org/grpc/resolver/manual"
+)
+
+var extraAddressReceivers = noopAddressReceivers
+
+func noopAddressReceivers(context.Context, *Worker) ([]addressReceiver, error) {
+	return nil, nil
+}
+
+// addressReceiver allows the initializing and setting of addresses. Since a
+// main use case of this interface is to use it in grpc dialing it satisfies
+// a grpc resolver.Builder interface as well.
+type addressReceiver interface {
+	InitialAddresses([]string)
+	SetAddresses([]string)
+}
+
+// grpcResolverReceiver is an addressReceiver which wraps a grpc manual.Resolver
+// InitialAddresses is passed onto the Resolver's InitialState method and
+// SetAddresses  is passed onto the Resolver's UpdateState call.
+type grpcResolverReceiver struct {
+	*manual.Resolver
+}
+
+func (r grpcResolverReceiver) getState(addrs []string) resolver.State {
+	var rAddrs []resolver.Address
+	for _, a := range addrs {
+		rAddrs = append(rAddrs, resolver.Address{Addr: a})
+	}
+	return resolver.State{Addresses: rAddrs}
+}
+
+func (r *grpcResolverReceiver) InitialAddresses(addrs []string) {
+	r.Resolver.InitialState(r.getState(addrs))
+}
+
+func (r *grpcResolverReceiver) SetAddresses(addrs []string) {
+	r.Resolver.UpdateState(r.getState(addrs))
+}

internal/daemon/worker/auth_rotation_test.go

@@ -16,15 +16,12 @@ import (
 	"github.com/mr-tron/base58"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"google.golang.org/grpc/resolver"
 	"google.golang.org/protobuf/proto"
 )
 
 // TestRotationTicking ensures that we see new credential information for a
 // worker on both the controller side and worker side in a periodic fashion
 func TestRotationTicking(t *testing.T) {
-	t.Parallel()
-
 	require, assert := require.New(t), assert.New(t)
 	logger := hclog.New(&hclog.LoggerOptions{
 		Level: hclog.Trace,
@@ -41,10 +38,16 @@ func TestRotationTicking(t *testing.T) {
 
 	const rotationPeriod = 20 * time.Second
 
+	// names should not be set when using pki workers
+	wConf, err := config.DevWorker()
+	require.NoError(err)
+	wConf.Worker.Name = ""
+	wConf.Worker.InitialUpstreams = c.ClusterAddrs()
 	w := worker.NewTestWorker(t, &worker.TestWorkerOpts{
 		InitialUpstreams:   c.ClusterAddrs(),
 		Logger:             logger.Named("worker"),
 		AuthRotationPeriod: rotationPeriod,
+		Config:             wConf,
 	})
 	t.Cleanup(w.Shutdown)
 
@@ -109,7 +112,6 @@ func TestRotationTicking(t *testing.T) {
 		// Stop and start the client connections to ensure the new credentials
 		// are valid; if not, we won't establish a new connection and rotation
 		// will fail
-		w.Worker().Resolver().UpdateState(resolver.State{Addresses: []resolver.Address{}})
 		require.NotNil(w.Worker().GrpcClientConn)
 		require.NoError(w.Worker().GrpcClientConn.Close())
 		require.NoError(w.Worker().StartControllerConnections())

internal/daemon/worker/controller_connection.go

@@ -8,7 +8,6 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"encoding/pem"
-	"errors"
 	"fmt"
 	"math"
 	"math/big"
@@ -22,6 +21,7 @@ import (
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/cmd/base"
 	"github.com/hashicorp/boundary/internal/daemon/worker/internal/metric"
+	"github.com/hashicorp/boundary/internal/errors"
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/go-secure-stdlib/base62"
@@ -41,11 +41,11 @@ const hcpbUrlSuffix = ".proxy.boundary.hashicorp.cloud:9202"
 // connection client creation
 func (w *Worker) StartControllerConnections() error {
 	const op = "worker.(Worker).StartControllerConnections"
-	initialAddrs := make([]resolver.Address, 0, len(w.conf.RawConfig.Worker.InitialUpstreams))
+	initialAddrs := make([]string, 0, len(w.conf.RawConfig.Worker.InitialUpstreams))
 	for _, addr := range w.conf.RawConfig.Worker.InitialUpstreams {
 		switch {
 		case strings.HasPrefix(addr, "/"):
-			initialAddrs = append(initialAddrs, resolver.Address{Addr: addr})
+			initialAddrs = append(initialAddrs, addr)
 		default:
 			host, port, err := net.SplitHostPort(addr)
 			if err != nil && strings.Contains(err.Error(), "missing port in address") {
@@ -54,40 +54,39 @@ func (w *Worker) StartControllerConnections() error {
 			if err != nil {
 				return fmt.Errorf("error parsing upstream address: %w", err)
 			}
-			initialAddrs = append(initialAddrs, resolver.Address{Addr: net.JoinHostPort(host, port)})
+			initialAddrs = append(initialAddrs, net.JoinHostPort(host, port))
 		}
 	}
 
 	if len(initialAddrs) == 0 {
 		if w.conf.RawConfig.HcpbClusterId != "" {
 			clusterAddress := fmt.Sprintf("%s%s", w.conf.RawConfig.HcpbClusterId, hcpbUrlSuffix)
-			initialAddrs = append(initialAddrs, resolver.Address{Addr: clusterAddress})
+			initialAddrs = append(initialAddrs, clusterAddress)
 			event.WriteSysEvent(w.baseContext, op, fmt.Sprintf("Setting HCP Boundary cluster address %s as upstream address", clusterAddress))
 		} else {
-			return errors.New("no initial upstream addresses found")
+			return errors.New(w.baseContext, errors.InvalidParameter, op, "no initial upstream addresses found")
 		}
 	}
 
-	w.Resolver().InitialState(resolver.State{
-		Addresses: initialAddrs,
-	})
-	if err := w.createClientConn(initialAddrs[0].Addr); err != nil {
-		return fmt.Errorf("error making client connection to upstream address %s: %w", initialAddrs[0].Addr, err)
+	for _, ar := range w.addressReceivers {
+		ar.InitialAddresses(initialAddrs)
+	}
+	if err := w.createClientConn(initialAddrs[0]); err != nil {
+		return fmt.Errorf("error making client connection to upstream address %s: %w", initialAddrs[0], err)
 	}
-
 	return nil
 }
 
-func (w *Worker) controllerDialerFunc() func(context.Context, string) (net.Conn, error) {
+func (w *Worker) controllerDialerFunc(extraAlpnProtos ...string) func(context.Context, string) (net.Conn, error) {
 	const op = "worker.(Worker).controllerDialerFunc"
 	return func(ctx context.Context, addr string) (net.Conn, error) {
 		var conn net.Conn
 		var err error
 		switch {
 		case w.conf.WorkerAuthKms != nil && !w.conf.DevUsePkiForUpstream:
-			conn, err = w.v1KmsAuthDialFn(ctx, addr)
+			conn, err = w.v1KmsAuthDialFn(ctx, addr, extraAlpnProtos...)
 		default:
-			conn, err = protocol.Dial(ctx, w.WorkerAuthStorage, addr, nodeenrollment.WithWrapper(w.conf.WorkerAuthStorageKms))
+			conn, err = protocol.Dial(ctx, w.WorkerAuthStorage, addr, nodeenrollment.WithWrapper(w.conf.WorkerAuthStorageKms), nodeenrollment.WithExtraAlpnProtos(extraAlpnProtos))
 			// No error and a valid connection means the WorkerAuthRegistrationRequest was populated
 			// We can remove the stored workerAuthRequest file
 			if err == nil && conn != nil {
@@ -120,9 +119,9 @@ func (w *Worker) controllerDialerFunc() func(context.Context, string) (net.Conn,
 	}
 }
 
-func (w *Worker) v1KmsAuthDialFn(ctx context.Context, addr string) (net.Conn, error) {
+func (w *Worker) v1KmsAuthDialFn(ctx context.Context, addr string, extraAlpnProtos ...string) (net.Conn, error) {
 	const op = "worker.(Worker).v1KmsAuthDialFn"
-	tlsConf, authInfo, err := w.workerAuthTLSConfig()
+	tlsConf, authInfo, err := w.workerAuthTLSConfig(extraAlpnProtos...)
 	if err != nil {
 		return nil, fmt.Errorf("error creating tls config for worker auth: %w", err)
 	}
@@ -169,7 +168,15 @@ func (w *Worker) createClientConn(addr string) error {
 		]
 	  }
 	  `, defaultTimeout)
-	res := w.Resolver()
+	var res resolver.Builder
+	for _, v := range w.addressReceivers {
+		if rec, ok := v.(*grpcResolverReceiver); ok {
+			res = rec.Resolver
+		}
+	}
+	if res == nil {
+		return errors.New(w.baseContext, errors.Internal, op, "unable to find a resolver.Builder amongst the address receivers")
+	}
 	dialOpts := []grpc.DialOption{
 		grpc.WithResolvers(res),
 		grpc.WithUnaryInterceptor(metric.InstrumentClusterClient()),
@@ -204,7 +211,7 @@ func (w *Worker) createClientConn(addr string) error {
 	return nil
 }
 
-func (w *Worker) workerAuthTLSConfig() (*tls.Config, *base.WorkerAuthInfo, error) {
+func (w *Worker) workerAuthTLSConfig(extraAlpnProtos ...string) (*tls.Config, *base.WorkerAuthInfo, error) {
 	var err error
 	info := &base.WorkerAuthInfo{
 		Name:         w.conf.RawConfig.Worker.Name,
@@ -275,6 +282,7 @@ func (w *Worker) workerAuthTLSConfig() (*tls.Config, *base.WorkerAuthInfo, error
 	}
 	b64alpn := base64.RawStdEncoding.EncodeToString(marshaledEncInfo)
 	var nextProtos []string
+	nextProtos = append(nextProtos, extraAlpnProtos...)
 	var count int
 	for i := 0; i < len(b64alpn); i += 230 {
 		end := i + 230

internal/daemon/worker/listeners.go

@@ -3,7 +3,7 @@ package worker
 import (
 	"context"
 	"crypto/tls"
-	"errors"
+	stderrers "errors"
 	"fmt"
 	"log"
 	"math"
@@ -14,7 +14,9 @@ import (
 
 	"github.com/hashicorp/boundary/internal/cmd/base"
 	"github.com/hashicorp/boundary/internal/daemon/cluster/handlers"
+	"github.com/hashicorp/boundary/internal/daemon/common"
 	"github.com/hashicorp/boundary/internal/daemon/worker/session"
+	"github.com/hashicorp/boundary/internal/errors"
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/go-multierror"
@@ -27,6 +29,16 @@ import (
 	"google.golang.org/grpc"
 )
 
+// the function that handles a secondary connection over a provided listener
+var handleSecondaryConnection = closeListener
+
+func closeListener(_ context.Context, l net.Listener) error {
+	if l != nil {
+		return l.Close()
+	}
+	return nil
+}
+
 func (w *Worker) startListeners(sm session.Manager) error {
 	const op = "worker.(Worker).startListeners"
 
@@ -92,11 +104,11 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 	) (*types.FetchNodeCredentialsResponse, error) {
 		client := w.controllerMultihopConn.Load()
 		if client == nil {
-			return nil, temperror.New(errors.New("error fetching controller connection, client is nil"))
+			return nil, temperror.New(stderrers.New("error fetching controller connection, client is nil"))
 		}
 		multihopClient, ok := client.(multihop.MultihopServiceClient)
 		if !ok {
-			return nil, temperror.New(errors.New("client could not be understood as a multihop service client"))
+			return nil, temperror.New(stderrers.New("client could not be understood as a multihop service client"))
 		}
 		// log.Println("proxying fetch node credentials request")
 		return multihopClient.FetchNodeCredentials(ctx, req)
@@ -110,11 +122,11 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 	) (*types.GenerateServerCertificatesResponse, error) {
 		client := w.controllerMultihopConn.Load()
 		if client == nil {
-			return nil, temperror.New(errors.New("error fetching controller connection, client is nil"))
+			return nil, temperror.New(stderrers.New("error fetching controller connection, client is nil"))
 		}
 		multihopClient, ok := client.(multihop.MultihopServiceClient)
 		if !ok {
-			return nil, temperror.New(errors.New("client could not be understood as a multihop service client"))
+			return nil, temperror.New(stderrers.New("client could not be understood as a multihop service client"))
 		}
 		// log.Println("proxying generate server cert request")
 		return multihopClient.GenerateServerCertificates(ctx, req)
@@ -135,15 +147,30 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 		return nil, fmt.Errorf("error instantiating node auth listener: %w", err)
 	}
 
+	// Create split listener
 	w.workerAuthSplitListener, err = nodeenet.NewSplitListener(interceptingListener)
 	if err != nil {
 		return nil, fmt.Errorf("error instantiating split listener: %w", err)
 	}
-	workerListener, err := w.workerAuthSplitListener.GetListener(nodeenet.AuthenticatedNonSpecificNextProto)
+
+	// This handles connections coming in that are authenticated via
+	// nodeenrollment but not with any extra purpose; these are normal PKI
+	// worker connections
+	nodeeAuthListener, err := w.workerAuthSplitListener.GetListener(nodeenet.AuthenticatedNonSpecificNextProto)
 	if err != nil {
 		return nil, fmt.Errorf("error instantiating worker split listener: %w", err)
 	}
-	nonWorkerListener, err := w.workerAuthSplitListener.GetListener(nodeenet.UnauthenticatedNextProto)
+
+	// Connections that come into here are not authed by nodeenrollment so are
+	// proxy connections
+	proxyListener, err := w.workerAuthSplitListener.GetListener(nodeenet.UnauthenticatedNextProto)
+	if err != nil {
+		return nil, fmt.Errorf("error instantiating non-worker split listener: %w", err)
+	}
+
+	// Connections coming in here are authed by nodeenrollment and are for the
+	// reverse grpc purpose
+	reverseGrpcListener, err := w.workerAuthSplitListener.GetListener(common.ReverseGrpcConnectionAlpnValue)
 	if err != nil {
 		return nil, fmt.Errorf("error instantiating non-worker split listener: %w", err)
 	}
@@ -167,15 +194,16 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 
 	ln.GrpcServer = downstreamServer
 
+	eventingListener, err := common.NewEventingListener(cancelCtx, nodeeAuthListener)
+	if err != nil {
+		return nil, fmt.Errorf("%s: error creating eventing listener: %w", op, err)
+	}
+
 	return func() {
 		go w.workerAuthSplitListener.Start()
-		go httpServer.Serve(nonWorkerListener)
-		go ln.GrpcServer.Serve(
-			&eventingListener{
-				ctx:    cancelCtx,
-				baseLn: workerListener,
-			},
-		)
+		go httpServer.Serve(proxyListener)
+		go ln.GrpcServer.Serve(eventingListener)
+		go handleSecondaryConnection(cancelCtx, reverseGrpcListener)
 	}, nil
 }
 
@@ -256,45 +284,3 @@ func listenerCloseErrorCheck(lnType string, err error) error {
 
 	return err
 }
-
-type eventingListener struct {
-	ctx    context.Context
-	baseLn net.Listener
-}
-
-func (e *eventingListener) Accept() (net.Conn, error) {
-	const op = "worker.(eventingListener).Accept"
-	conn, err := e.baseLn.Accept()
-	if err != nil || conn == nil {
-		return conn, err
-	}
-
-	// This is all best-effort; anything going wrong here shouldn't disrupt the
-	// connection, so on error simply stop trying to get to an event
-	var tlsConn *tls.Conn
-	switch c := conn.(type) {
-	case *protocol.Conn:
-		// If we so choose, at this point we can pull out the client's
-		// NextProtos with c.ClientNextProtos
-		tlsConn = c.Conn
-	case *tls.Conn:
-		tlsConn = c
-	}
-
-	if tlsConn != nil && len(tlsConn.ConnectionState().PeerCertificates) > 0 {
-		keyId, err := nodee.KeyIdFromPkix(tlsConn.ConnectionState().PeerCertificates[0].SubjectKeyId)
-		if err == nil {
-			event.WriteSysEvent(e.ctx, op, "worker successfully authenticated", "key_id", keyId)
-		}
-	}
-
-	return conn, err
-}
-
-func (e *eventingListener) Close() error {
-	return e.baseLn.Close()
-}
-
-func (e *eventingListener) Addr() net.Addr {
-	return e.baseLn.Addr()
-}

internal/daemon/worker/status.go

@@ -14,7 +14,6 @@ import (
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/go-secure-stdlib/strutil"
-	"google.golang.org/grpc/resolver"
 )
 
 type LastStatusInformation struct {
@@ -23,7 +22,7 @@ type LastStatusInformation struct {
 	LastCalculatedUpstreams []string
 }
 
-func (w *Worker) startStatusTicking(cancelCtx context.Context, sessionManager session.Manager) {
+func (w *Worker) startStatusTicking(cancelCtx context.Context, sessionManager session.Manager, addrReceivers []addressReceiver) {
 	const op = "worker.(Worker).startStatusTicking"
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
 	// This function exists to desynchronize calls to controllers from
@@ -55,7 +54,7 @@ func (w *Worker) startStatusTicking(cancelCtx context.Context, sessionManager se
 				continue
 			}
 
-			w.sendWorkerStatus(cancelCtx, sessionManager)
+			w.sendWorkerStatus(cancelCtx, sessionManager, addrReceivers)
 			timer.Reset(getRandomInterval())
 		}
 	}
@@ -99,7 +98,7 @@ func (w *Worker) WaitForNextSuccessfulStatusUpdate() error {
 	return nil
 }
 
-func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager session.Manager) {
+func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager session.Manager, addressReceivers []addressReceiver) {
 	const op = "worker.(Worker).sendWorkerStatus"
 
 	// First send info as-is. We'll perform cleanup duties after we
@@ -193,15 +192,12 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess
 		}
 	} else {
 		w.updateTags.Store(false)
-		var addrs []resolver.Address
-		var newUpstreams []string
+		var addrs []string
 		// This may be empty if we are in a multiple hop scenario
 		if len(result.CalculatedUpstreams) > 0 {
-			addrs = make([]resolver.Address, 0, len(result.CalculatedUpstreams))
-			newUpstreams = make([]string, 0, len(result.CalculatedUpstreams))
+			addrs = make([]string, 0, len(result.CalculatedUpstreams))
 			for _, v := range result.CalculatedUpstreams {
-				addrs = append(addrs, resolver.Address{Addr: v.Address})
-				newUpstreams = append(newUpstreams, v.Address)
+				addrs = append(addrs, v.Address)
 			}
 
 		} else if w.conf.RawConfig.HcpbClusterId != "" {
@@ -212,27 +208,29 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess
 			if err != nil {
 				event.WriteError(hcpbWorkersCtx, op, err, event.WithInfoMsg("error fetching managed worker information"))
 			} else {
-				addrs = make([]resolver.Address, 0, len(workersResp.Workers))
-				newUpstreams = make([]string, 0, len(workersResp.Workers))
+				addrs = make([]string, 0, len(workersResp.Workers))
 				for _, v := range workersResp.Workers {
-					addrs = append(addrs, resolver.Address{Addr: v.Address})
-					newUpstreams = append(newUpstreams, v.Address)
+					addrs = append(addrs, v.Address)
 				}
 			}
 		}
 		if len(addrs) > 0 {
 			lastStatus := w.lastStatusSuccess.Load().(*LastStatusInformation)
 			// Compare upstreams; update resolver if there is a difference, and emit an event with old and new addresses
-			if lastStatus != nil && !strutil.EquivalentSlices(lastStatus.LastCalculatedUpstreams, newUpstreams) {
-				upstreamsMessage := fmt.Sprintf("Upstreams has changed; old upstreams were: %s, new upstreams are: %s", lastStatus.LastCalculatedUpstreams, newUpstreams)
+			if lastStatus != nil && !strutil.EquivalentSlices(lastStatus.LastCalculatedUpstreams, addrs) {
+				upstreamsMessage := fmt.Sprintf("Upstreams has changed; old upstreams were: %s, new upstreams are: %s", lastStatus.LastCalculatedUpstreams, addrs)
 				event.WriteSysEvent(cancelCtx, op, upstreamsMessage)
-				w.Resolver().UpdateState(resolver.State{Addresses: addrs})
+				for _, as := range addressReceivers {
+					as.SetAddresses(addrs)
+				}
 			} else if lastStatus == nil {
-				w.Resolver().UpdateState(resolver.State{Addresses: addrs})
-				event.WriteSysEvent(cancelCtx, op, fmt.Sprintf("Upstreams after first status set to: %s", newUpstreams))
+				for _, as := range addressReceivers {
+					as.SetAddresses(addrs)
+				}
+				event.WriteSysEvent(cancelCtx, op, fmt.Sprintf("Upstreams after first status set to: %s", addrs))
 			}
 		}
-		w.lastStatusSuccess.Store(&LastStatusInformation{StatusResponse: result, StatusTime: time.Now(), LastCalculatedUpstreams: newUpstreams})
+		w.lastStatusSuccess.Store(&LastStatusInformation{StatusResponse: result, StatusTime: time.Now(), LastCalculatedUpstreams: addrs})
 
 		for _, request := range result.GetJobsRequests() {
 			switch request.GetRequestType() {

internal/daemon/worker/worker.go

@@ -6,7 +6,6 @@ import (
 	"crypto/rand"
 	"crypto/tls"
 	"crypto/x509"
-	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -14,10 +13,12 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/hashicorp/boundary/internal/errors"
 	pb "github.com/hashicorp/boundary/internal/gen/controller/servers"
-
 	"github.com/hashicorp/nodeenrollment"
+	nodeenet "github.com/hashicorp/nodeenrollment/net"
 	"github.com/mr-tron/base58"
+	"google.golang.org/grpc/resolver/manual"
 
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/cmd/base"
@@ -29,14 +30,10 @@ import (
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-secure-stdlib/base62"
 	"github.com/hashicorp/go-secure-stdlib/mlock"
-	"github.com/hashicorp/nodeenrollment/multihop"
-	nodeenet "github.com/hashicorp/nodeenrollment/net"
 	nodeefile "github.com/hashicorp/nodeenrollment/storage/file"
 	"github.com/hashicorp/nodeenrollment/types"
 	ua "go.uber.org/atomic"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/resolver"
-	"google.golang.org/grpc/resolver/manual"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -57,6 +54,9 @@ type Worker struct {
 	// This is exported for tests.
 	GrpcClientConn *grpc.ClientConn
 
+	// receives address updates and contains the grpc resolver.
+	addressReceivers []addressReceiver
+
 	sessionManager session.Manager
 
 	controllerStatusConn *atomic.Value
@@ -64,8 +64,6 @@ type Worker struct {
 	lastStatusSuccess    *atomic.Value
 	workerStartTime      time.Time
 
-	controllerResolver *atomic.Value
-
 	controllerMultihopConn *atomic.Value
 
 	proxyListener *base.ServerListener
@@ -94,6 +92,7 @@ type Worker struct {
 }
 
 func New(conf *Config) (*Worker, error) {
+	const op = "worker.New"
 	metric.InitializeHttpCollectors(conf.PrometheusRegisterer)
 	metric.InitializeWebsocketCollectors(conf.PrometheusRegisterer)
 	metric.InitializeClusterClientCollectors(conf.PrometheusRegisterer)
@@ -105,7 +104,6 @@ func New(conf *Config) (*Worker, error) {
 		controllerStatusConn:   new(atomic.Value),
 		everAuthenticated:      ua.NewBool(false),
 		lastStatusSuccess:      new(atomic.Value),
-		controllerResolver:     new(atomic.Value),
 		controllerMultihopConn: new(atomic.Value),
 		tags:                   new(atomic.Value),
 		updateTags:             ua.NewBool(false),
@@ -116,7 +114,7 @@ func New(conf *Config) (*Worker, error) {
 	w.lastStatusSuccess.Store((*LastStatusInformation)(nil))
 	scheme := strconv.FormatInt(time.Now().UnixNano(), 36)
 	controllerResolver := manual.NewBuilderWithScheme(scheme)
-	w.controllerResolver.Store(controllerResolver)
+	w.addressReceivers = []addressReceiver{&grpcResolverReceiver{controllerResolver}}
 
 	if conf.RawConfig.Worker == nil {
 		conf.RawConfig.Worker = new(config.Worker)
@@ -178,6 +176,12 @@ func (w *Worker) Start() error {
 		return nil
 	}
 
+	extraAddrReceivers, err := extraAddressReceivers(w.baseContext, w)
+	if err != nil {
+		return errors.Wrap(w.baseContext, err, op)
+	}
+	w.addressReceivers = append(w.addressReceivers, extraAddrReceivers...)
+
 	if w.conf.WorkerAuthKms == nil || w.conf.DevUsePkiForUpstream {
 		// In this section, we look for existing worker credentials. The two
 		// variables below store whether to create new credentials and whether
@@ -191,7 +195,7 @@ func (w *Worker) Start() error {
 		w.WorkerAuthStorage, err = nodeefile.New(w.baseContext,
 			nodeefile.WithBaseDirectory(w.conf.RawConfig.Worker.AuthStoragePath))
 		if err != nil {
-			return fmt.Errorf("error loading worker auth storage directory: %w", err)
+			return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error loading worker auth storage directory"))
 		}
 
 		var createNodeAuthCreds bool
@@ -292,17 +296,16 @@ func (w *Worker) Start() error {
 	}
 
 	if err := w.StartControllerConnections(); err != nil {
-		return fmt.Errorf("error making controller connections: %w", err)
+		return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error making controller connections"))
 	}
 
-	var err error
 	w.sessionManager, err = session.NewManager(pbs.NewSessionServiceClient(w.GrpcClientConn))
 	if err != nil {
-		return fmt.Errorf("error creating session manager: %w", err)
+		return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error creating session manager"))
 	}
 
 	if err := w.startListeners(w.sessionManager); err != nil {
-		return fmt.Errorf("error starting worker listeners: %w", err)
+		return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error starting worker listeners"))
 	}
 
 	// Rather than deal with some of the potential error conditions for Add on
@@ -312,7 +315,7 @@ func (w *Worker) Start() error {
 	w.tickerWg.Add(2)
 	go func() {
 		defer w.tickerWg.Done()
-		w.startStatusTicking(w.baseContext, w.sessionManager)
+		w.startStatusTicking(w.baseContext, w.sessionManager, w.addressReceivers)
 	}()
 	go func() {
 		defer w.tickerWg.Done()
@@ -339,9 +342,6 @@ func (w *Worker) Shutdown() error {
 	// Stop listeners first to prevent new connections to the
 	// controller.
 	defer w.started.Store(false)
-	w.Resolver().UpdateState(resolver.State{Addresses: []resolver.Address{}})
-	w.baseCancel()
-
 	if err := w.stopServersAndListeners(); err != nil {
 		return fmt.Errorf("error stopping worker servers and listeners: %w", err)
 	}
@@ -370,7 +370,9 @@ func (w *Worker) Shutdown() error {
 
 	// Proceed with remainder of shutdown.
 	w.baseCancel()
-	w.Resolver().UpdateState(resolver.State{Addresses: []resolver.Address{}})
+	for _, ar := range w.addressReceivers {
+		ar.SetAddresses(nil)
+	}
 
 	w.started.Store(false)
 	w.tickerWg.Wait()
@@ -383,14 +385,6 @@ func (w *Worker) Shutdown() error {
 	return nil
 }
 
-func (w *Worker) Resolver() *manual.Resolver {
-	raw := w.controllerResolver.Load()
-	if raw == nil {
-		panic("nil resolver")
-	}
-	return raw.(*manual.Resolver)
-}
-
 func (w *Worker) ParseAndStoreTags(incoming map[string][]string) {
 	if len(incoming) == 0 {
 		w.tags.Store([]*pb.TagPair{})
@@ -409,40 +403,6 @@ func (w *Worker) ParseAndStoreTags(incoming map[string][]string) {
 	w.updateTags.Store(true)
 }
 
-// ControllerServerCoordinationServiceConn returns the underlying server coordination service client
-func (w *Worker) ControllerServerCoordinationServiceConn() (pbs.ServerCoordinationServiceClient, error) {
-	rawConn := w.controllerStatusConn.Load()
-	if rawConn == nil {
-		return nil, errors.New("unable to load controller service coordination service connection")
-	}
-	statusClient, ok := rawConn.(pbs.ServerCoordinationServiceClient)
-	if !ok {
-		return nil, fmt.Errorf("invalid service coordination service client %T", rawConn)
-	}
-	if statusClient == nil {
-		return nil, fmt.Errorf("service coordination service client is nil")
-	}
-
-	return statusClient, nil
-}
-
-// ControllerMultihopConn returns the underlying multihop service client
-func (w *Worker) ControllerMultihopConn() (multihop.MultihopServiceClient, error) {
-	rawConn := w.controllerMultihopConn.Load()
-	if rawConn == nil {
-		return nil, errors.New("unable to load controller multihop service connection")
-	}
-	multihopClient, ok := rawConn.(multihop.MultihopServiceClient)
-	if !ok {
-		return nil, fmt.Errorf("invalid multihop client %T", rawConn)
-	}
-	if multihopClient == nil {
-		return nil, fmt.Errorf("session multihop is nil")
-	}
-
-	return multihopClient, nil
-}
-
 func (w *Worker) getSessionTls(sessionManager session.Manager) func(hello *tls.ClientHelloInfo) (*tls.Config, error) {
 	const op = "worker.(Worker).getSessionTls"
 	return func(hello *tls.ClientHelloInfo) (*tls.Config, error) {
@@ -520,7 +480,7 @@ func (w *Worker) getSessionTls(sessionManager session.Manager) func(hello *tls.C
 			// Go will not run this without at least one peer certificate, but
 			// doesn't hurt to check
 			if len(cs.PeerCertificates) == 0 {
-				return errors.New("no peer certificates provided")
+				return errors.New(ctx, errors.InvalidParameter, op, "no peer certificates provided")
 			}
 			_, err := cs.PeerCertificates[0].Verify(verifyOpts)
 			return err

internal/daemon/worker/worker_test.go

@@ -138,8 +138,6 @@ func TestWorkerNew(t *testing.T) {
 }
 
 func TestSetupWorkerAuthStorage(t *testing.T) {
-	t.Parallel()
-
 	ctx := context.Background()
 
 	ts := db.TestWrapper(t)

internal/tests/cluster/worker_tagging_test.go

@@ -2,7 +2,6 @@ package cluster
 
 import (
 	"testing"
-	"time"
 
 	"github.com/hashicorp/boundary/api/targets"
 	"github.com/hashicorp/boundary/internal/cmd/config"
@@ -60,6 +59,7 @@ func TestWorkerTagging(t *testing.T) {
 	})
 	defer w1.Shutdown()
 	w1Addr := w1.ProxyAddrs()[0]
+	w1.Worker().WaitForNextSuccessfulStatusUpdate()
 
 	// Worker 2
 	conf, err = config.DevWorker()
@@ -77,6 +77,7 @@ func TestWorkerTagging(t *testing.T) {
 	})
 	defer w2.Shutdown()
 	w2Addr := w2.ProxyAddrs()[0]
+	w2.Worker().WaitForNextSuccessfulStatusUpdate()
 
 	// Worker 3
 	conf, err = config.DevWorker()
@@ -95,7 +96,8 @@ func TestWorkerTagging(t *testing.T) {
 	defer w3.Shutdown()
 	w3Addr := w3.ProxyAddrs()[0]
 
-	time.Sleep(10 * time.Second)
+	w3.Worker().WaitForNextSuccessfulStatusUpdate()
+
 	expectWorkers(t, c1, w1, w2, w3)
 
 	cases := []struct {