diff --git a/go.mod b/go.mod index 723b8af3b..abb7788e9 100644 --- a/go.mod +++ b/go.mod @@ -94,7 +94,7 @@ require ( github.com/cenkalti/backoff/v4 v4.1.3 github.com/hashicorp/go-kms-wrapping/extras/kms/v2 v2.0.0-20221122211539-47c893099f13 github.com/hashicorp/go-version v1.3.0 - github.com/hashicorp/nodeenrollment v0.1.17 + github.com/hashicorp/nodeenrollment v0.1.18 github.com/kelseyhightower/envconfig v1.4.0 golang.org/x/exp v0.0.0-20220921164117-439092de6870 golang.org/x/net v0.0.0-20220722155237-a158d28d115b @@ -139,7 +139,7 @@ require ( github.com/hashicorp/go-secure-stdlib/tlsutil v0.1.1 // indirect github.com/hashicorp/golang-lru v0.5.4 // indirect github.com/hashicorp/vault/sdk v0.3.0 // indirect - github.com/hashicorp/yamux v0.0.0-20181012175058-2f1d1f20f75d // indirect + github.com/hashicorp/yamux v0.1.0 // indirect github.com/huandu/xstrings v1.3.2 // indirect github.com/imdario/mergo v0.3.12 // indirect github.com/jackc/chunkreader/v2 v2.0.1 // indirect diff --git a/go.sum b/go.sum index 7eac87371..8966004b2 100644 --- a/go.sum +++ b/go.sum @@ -744,16 +744,17 @@ github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+l github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4= github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ= -github.com/hashicorp/nodeenrollment v0.1.17 h1:ZaGNugd3EOZIdJxgC5bUA1CbHZ/OKZcgXnquzoFql6E= -github.com/hashicorp/nodeenrollment v0.1.17/go.mod h1:N5gYsm8mWiDfIw/j+1IQ6NBO1cWCmhPpvQ9GB1QUnsU= +github.com/hashicorp/nodeenrollment v0.1.18 h1:fqFMZ5e2cWupeQDxKledF9DGHsk9GoJl7K2Wrdiys6k= +github.com/hashicorp/nodeenrollment v0.1.18/go.mod h1:N5gYsm8mWiDfIw/j+1IQ6NBO1cWCmhPpvQ9GB1QUnsU= github.com/hashicorp/vault/api v1.3.1 h1:pkDkcgTh47PRjY1NEFeofqR4W/HkNUi9qIakESO2aRM= github.com/hashicorp/vault/api v1.3.1/go.mod h1:QeJoWxMFt+MsuWcYhmwRLwKEXrjwAFFywzhptMsTIUw= github.com/hashicorp/vault/sdk v0.1.13/go.mod h1:B+hVj7TpuQY1Y/GPbCpffmgd+tSEwvhkWnjtSYCaS2M= github.com/hashicorp/vault/sdk v0.3.0 h1:kR3dpxNkhh/wr6ycaJYqp6AFT/i2xaftbfnwZduTKEY= github.com/hashicorp/vault/sdk v0.3.0/go.mod h1:aZ3fNuL5VNydQk8GcLJ2TV8YCRVvyaakYkhZRoVuhj0= github.com/hashicorp/yamux v0.0.0-20180604194846-3520598351bb/go.mod h1:+NfK9FKeTrX5uv1uIXGdwYDTeHna2qgaIlx54MXqjAM= -github.com/hashicorp/yamux v0.0.0-20181012175058-2f1d1f20f75d h1:kJCB4vdITiW1eC1vq2e6IsrXKrZit1bv/TDYFGMp4BQ= github.com/hashicorp/yamux v0.0.0-20181012175058-2f1d1f20f75d/go.mod h1:+NfK9FKeTrX5uv1uIXGdwYDTeHna2qgaIlx54MXqjAM= +github.com/hashicorp/yamux v0.1.0 h1:DzDIF6Sd7GD2sX0kDFpHAsJMY4L+OfTvtuaQsOYXxzk= +github.com/hashicorp/yamux v0.1.0/go.mod h1:CtWFDAQgb7dxtzFs4tWbplKIe2jSi3+5vKbgIO0SLnQ= github.com/hinshun/vt10x v0.0.0-20180616224451-1954e6464174 h1:WlZsjVhE8Af9IcZDGgJGQpNflI3+MJSBhsgT5PCtzBQ= github.com/hinshun/vt10x v0.0.0-20180616224451-1954e6464174/go.mod h1:DqJ97dSdRW1W22yXSB90986pcOyQ7r45iio1KN2ez1A= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= diff --git a/internal/daemon/common/const.go b/internal/daemon/common/const.go index 27b5e2925..013d89603 100644 --- a/internal/daemon/common/const.go +++ b/internal/daemon/common/const.go @@ -1,3 +1,6 @@ package common -const ReverseGrpcConnectionAlpnValue = "the-downstream-dialer-plays-an-uno-reverse-card" +const ( + ReverseGrpcConnectionAlpnValue = "the-downstream-dialer-plays-an-uno-reverse-card" + DataPlaneProxyAlpnValue = "i-herd-you-like-proxies-so-i-put-a-proxy-in-your-proxy" +) diff --git a/internal/daemon/controller/controller.go b/internal/daemon/controller/controller.go index df1f17af1..c9d89258d 100644 --- a/internal/daemon/controller/controller.go +++ b/internal/daemon/controller/controller.go @@ -39,6 +39,7 @@ import ( host_plugin_assets "github.com/hashicorp/boundary/plugins/host" "github.com/hashicorp/boundary/sdk/pbs/plugin" external_host_plugins "github.com/hashicorp/boundary/sdk/plugins/host" + "github.com/hashicorp/boundary/version" "github.com/hashicorp/go-hclog" "github.com/hashicorp/go-secure-stdlib/mlock" "github.com/hashicorp/go-secure-stdlib/pluginutil/v2" @@ -56,7 +57,7 @@ type downstreamRouter interface { // ProcessPendingConnections starts a function that continually processes // incoming client connections. This only returns when the provided context // is done. - StartProcessingPendingConnections(context.Context, func() string) + StartProcessingPendingConnections(context.Context, func() string) error } // downstreamWorkersTicker defines an interface for a ticker that maintains the @@ -70,8 +71,8 @@ type downstreamWorkersTicker interface { var ( downstreamRouterFactory func() downstreamRouter - downstreamersFactory func(context.Context, string) (common.Downstreamers, error) - downstreamWorkersTickerFactory func(context.Context, string, common.Downstreamers, downstreamRouter) (downstreamWorkersTicker, error) + downstreamersFactory func(context.Context, string, string) (common.Downstreamers, error) + downstreamWorkersTickerFactory func(context.Context, string, string, common.Downstreamers, downstreamRouter) (downstreamWorkersTicker, error) commandClientFactory func(context.Context, *Controller) error ) @@ -389,7 +390,8 @@ func New(ctx context.Context, conf *Config) (*Controller, error) { } if downstreamersFactory != nil { - c.downstreamWorkers, err = downstreamersFactory(ctx, "root") + boundVer := version.Get().VersionNumber() + c.downstreamWorkers, err = downstreamersFactory(ctx, "root", boundVer) if err != nil { return nil, fmt.Errorf("unable to initialize downstream workers graph: %w", err) } @@ -479,7 +481,8 @@ func (c *Controller) Start() error { if downstreamWorkersTickerFactory != nil { // we'll use "root" to designate that this is the root of the graph (aka // a controller) - dswTicker, err := downstreamWorkersTickerFactory(c.baseContext, "root", c.downstreamWorkers, c.downstreamRoutes) + boundVer := version.Get().VersionNumber() + dswTicker, err := downstreamWorkersTickerFactory(c.baseContext, "root", boundVer, c.downstreamWorkers, c.downstreamRoutes) if err != nil { return fmt.Errorf("error creating downstream workers ticker: %w", err) } diff --git a/internal/daemon/controller/listeners.go b/internal/daemon/controller/listeners.go index 71a7387b6..7a7817dae 100644 --- a/internal/daemon/controller/listeners.go +++ b/internal/daemon/controller/listeners.go @@ -26,7 +26,7 @@ import ( // the function that handles a secondary connection over a provided listener var handleSecondaryConnection = closeListener -func closeListener(_ context.Context, l net.Listener, _ any, _ int) error { +func closeListener(_ context.Context, l, _ net.Listener, _ any) error { if l != nil { return l.Close() } @@ -243,18 +243,17 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error ln.GrpcServer = workerServer return func() { + err := handleSecondaryConnection(c.baseContext, metric.InstrumentClusterTrackingListener(multiplexingReverseGrpcListener, "reverse-grpc"), + nil, c.downstreamRoutes) + if err != nil { + event.WriteError(c.baseContext, op, err, event.WithInfoMsg("handleSecondaryConnection error")) + } go func() { err := splitListener.Start() if err != nil { event.WriteError(c.baseContext, op, err, event.WithInfoMsg("splitListener.Start() error")) } }() - go func() { - err := handleSecondaryConnection(c.baseContext, metric.InstrumentClusterTrackingListener(multiplexingReverseGrpcListener, "reverse-grpc"), c.downstreamRoutes, -1) - if err != nil { - event.WriteError(c.baseContext, op, err, event.WithInfoMsg("handleSecondaryConnection error")) - } - }() go func() { err := ln.GrpcServer.Serve(metric.InstrumentClusterTrackingListener(multiplexingAuthedListener, "grpc")) if err != nil { diff --git a/internal/daemon/worker/addressreceiver.go b/internal/daemon/worker/addressreceiver.go index b86136dbf..68a4636b6 100644 --- a/internal/daemon/worker/addressreceiver.go +++ b/internal/daemon/worker/addressreceiver.go @@ -16,9 +16,9 @@ func noopAddressReceivers(context.Context, *Worker) ([]addressReceiver, error) { type receiverType uint const ( - UnknownReceiverType receiverType = 0 - grpcResolverReceiverType receiverType = 1 - dialingListenerReceiverType receiverType = 2 + UnknownReceiverType receiverType = iota + grpcResolverReceiverType + secondaryConnectionReceiverType ) // String returns a string representation of the receiverType @@ -26,7 +26,7 @@ func (s receiverType) String() string { return [...]string{ "unknown", "grpcResolver", - "dialingListener", + "secondaryConnections", }[s] } diff --git a/internal/daemon/worker/controller_connection.go b/internal/daemon/worker/controller_connection.go index e8bb3355f..30e331d19 100644 --- a/internal/daemon/worker/controller_connection.go +++ b/internal/daemon/worker/controller_connection.go @@ -33,6 +33,7 @@ import ( "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/resolver" "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/structpb" ) const hcpbUrlSuffix = ".proxy.boundary.hashicorp.cloud:9202" @@ -77,8 +78,14 @@ func (w *Worker) StartControllerConnections() error { return nil } -func (w *Worker) controllerDialerFunc(extraAlpnProtos ...string) func(context.Context, string) (net.Conn, error) { - const op = "worker.(Worker).controllerDialerFunc" +// upstreamDialerFunc dials an upstream server. extraAlpnProtos can be provided +// to kms and pki connections and are used for identifying, on the server side, +// the intended purpose of the connection. upstreamDialerFunc takes an optional +// state struct which, if the connection is made using the node enrolment +// library, sends the state to the server and can be retrieved from the +// resulting protocol.Conn. +func (w *Worker) upstreamDialerFunc(state *structpb.Struct, extraAlpnProtos ...string) func(context.Context, string) (net.Conn, error) { + const op = "worker.(Worker).upstreamDialerFunc" return func(ctx context.Context, addr string) (net.Conn, error) { var conn net.Conn var err error @@ -90,6 +97,7 @@ func (w *Worker) controllerDialerFunc(extraAlpnProtos ...string) func(context.Co ctx, w.WorkerAuthStorage, addr, + nodeenrollment.WithState(state), nodeenrollment.WithWrapper(w.conf.WorkerAuthStorageKms), nodeenrollment.WithExtraAlpnProtos(extraAlpnProtos), // If the activation token hasn't been populated, this won't do @@ -202,7 +210,7 @@ func (w *Worker) createClientConn(addr string) error { grpc.WithUnaryInterceptor(metric.InstrumentClusterClient()), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(math.MaxInt32)), grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(math.MaxInt32)), - grpc.WithContextDialer(w.controllerDialerFunc()), + grpc.WithContextDialer(w.upstreamDialerFunc(nil)), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(defServiceConfig), // Don't have the resolver reach out for a service config from the diff --git a/internal/daemon/worker/handler.go b/internal/daemon/worker/handler.go index 1d8f36749..21a58172b 100644 --- a/internal/daemon/worker/handler.go +++ b/internal/daemon/worker/handler.go @@ -196,9 +196,9 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa } workerId := w.LastStatusSuccess().WorkerId - var ci *pbs.AuthorizeConnectionResponse + var acResp *pbs.AuthorizeConnectionResponse var connsLeft int32 - ci, connsLeft, err = sess.RequestAuthorizeConnection(ctx, workerId, connCancel) + acResp, connsLeft, err = sess.RequestAuthorizeConnection(ctx, workerId, connCancel) if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("unable to authorize connection")) if err = conn.Close(websocket.StatusInternalError, "unable to authorize connection"); err != nil { @@ -206,12 +206,12 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa } return } - event.WriteSysEvent(ctx, op, "connection successfully authorized", "session_id", sessionId, "connection_id", ci.GetConnectionId()) + event.WriteSysEvent(ctx, op, "connection successfully authorized", "session_id", sessionId, "connection_id", acResp.GetConnectionId()) // Wrapping the client websocket with a `net.Conn` implementation that // records the bytes that go across Read() and Write(). cc := &countingConn{Conn: websocket.NetConn(connCtx, conn, websocket.MessageBinary)} - err = sess.ApplyConnectionCounterCallbacks(ci.GetConnectionId(), cc.BytesRead, cc.BytesWritten) + err = sess.ApplyConnectionCounterCallbacks(acResp.GetConnectionId(), cc.BytesRead, cc.BytesWritten) if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("unable to set counter callbacks for session connection")) err = conn.Close(websocket.StatusInternalError, "unable to set counter callbacks for session connection") @@ -223,14 +223,14 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa defer func() { ccd := map[string]*session.ConnectionCloseData{ - ci.GetConnectionId(): { + acResp.GetConnectionId(): { SessionId: sess.GetId(), BytesUp: cc.BytesRead(), BytesDown: cc.BytesWritten(), }, } if sessionManager.RequestCloseConnections(ctx, ccd) { - event.WriteSysEvent(ctx, op, "connection closed", "session_id", sessionId, "connection_id", ci.GetConnectionId()) + event.WriteSysEvent(ctx, op, "connection closed", "session_id", sessionId, "connection_id", acResp.GetConnectionId()) } }() @@ -255,7 +255,7 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa } return } - protocolCtx := ci.GetProtocolContext() + protocolCtx := acResp.GetProtocolContext() if protocolCtx == nil { // TODO: Remove this if block once pre v0.12.0 controllers are no longer supported. if protocolCtx, err = GetProtocolContext(ctx, workerId, sess, endpointUrl.Scheme); err != nil { @@ -265,7 +265,7 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa } } - pDialer, err := proxyHandlers.GetEndpointDialer(ctx, endpointUrl.Host, workerId, ci, w.downstreamRoutes) + pDialer, err := proxyHandlers.GetEndpointDialer(ctx, endpointUrl.Host, workerId, acResp, w.downstreamRoutes) if err != nil { conn.Close(proxyHandlers.WebsocketStatusProtocolSetupError, "unable to get endpoint dialer") event.WriteError(ctx, op, err) @@ -273,13 +273,13 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa } // Verify the protocol has a supported proxy before calling RequestAuthorizeConnection - handleProxyFn, err := proxyHandlers.GetHandler(workerId, ci.GetProtocolContext()) + handleProxyFn, err := proxyHandlers.GetHandler(workerId, acResp.GetProtocolContext()) if err != nil { conn.Close(proxyHandlers.WebsocketStatusProtocolSetupError, "unable to get proxy handler") event.WriteError(ctx, op, err) return } - runProxy, err := handleProxyFn(ctx, cc, pDialer, ci.GetConnectionId(), protocolCtx) + runProxy, err := handleProxyFn(ctx, cc, pDialer, acResp.GetConnectionId(), protocolCtx) if err != nil { conn.Close(proxyHandlers.WebsocketStatusProtocolSetupError, "unable to setup proxying") event.WriteError(ctx, op, err) @@ -290,7 +290,7 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa // that we can establish the proxy. endpointAddr := pDialer.LastConnectionAddr() connectionInfo := &pbs.ConnectConnectionRequest{ - ConnectionId: ci.GetConnectionId(), + ConnectionId: acResp.GetConnectionId(), ClientTcpAddress: clientAddr.IP.String(), ClientTcpPort: uint32(clientAddr.Port), EndpointTcpAddress: endpointAddr.Ip(), @@ -299,7 +299,7 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa UserClientIp: userClientIp, } if err = sess.RequestConnectConnection(ctx, connectionInfo); err != nil { - event.WriteError(ctx, op, err, event.WithInfoMsg("error requesting connect connection", "session_id", sess.GetId(), "connection_id", ci.GetConnectionId())) + event.WriteError(ctx, op, err, event.WithInfoMsg("error requesting connect connection", "session_id", sess.GetId(), "connection_id", acResp.GetConnectionId())) if err = conn.Close(websocket.StatusInternalError, "unable to establish proxy"); err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection")) } diff --git a/internal/daemon/worker/listeners.go b/internal/daemon/worker/listeners.go index 1817a138f..cb48cd501 100644 --- a/internal/daemon/worker/listeners.go +++ b/internal/daemon/worker/listeners.go @@ -30,11 +30,16 @@ import ( ) // the function that handles a secondary connection over a provided listener -var handleSecondaryConnection = closeListener +var handleSecondaryConnection = closeListeners -func closeListener(_ context.Context, l net.Listener, _ any, _ int) error { +// closeListeners handles the secondary connection listeners by closing them. +// l is the grpc listener and l2 is the data plane listener. +func closeListeners(_ context.Context, l, l2 net.Listener, _ any) error { if l != nil { - return l.Close() + l.Close() + } + if l2 != nil { + l2.Close() } return nil } @@ -168,9 +173,9 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger, // Connections coming in here are authed by nodeenrollment and are for the // reverse grpc purpose - reverseGrpcListener, err := w.workerAuthSplitListener.GetListener(common.ReverseGrpcConnectionAlpnValue) + reverseGrpcListener, err := w.workerAuthSplitListener.GetListener(common.ReverseGrpcConnectionAlpnValue, nodee.WithNativeConns(true)) if err != nil { - return nil, fmt.Errorf("error instantiating non-worker split listener: %w", err) + return nil, fmt.Errorf("error instantiating reverse grpc split listener: %w", err) } // This wraps the reverse grpc pki worker connections with a listener which @@ -180,6 +185,20 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger, return nil, fmt.Errorf("%s: error creating reverse grpc pki worker tracking listener: %w", op, err) } + // Connections coming in here are authed by nodeenrollment and are for the + // multi-hop session-proxying + dataPlaneProxyListener, err := w.workerAuthSplitListener.GetListener(common.DataPlaneProxyAlpnValue, nodee.WithNativeConns(true)) + if err != nil { + return nil, fmt.Errorf("error instantiating websocket proxying split listener: %w", err) + } + + // This wraps the web socket proxying worker connections with a listener which + // adds the worker key id of the connections to the worker's pkiConnManager. + dataPlaneProxyTrackingListener, err := cluster.NewTrackingListener(w.baseContext, dataPlaneProxyListener, w.pkiConnManager) + if err != nil { + return nil, fmt.Errorf("%s: error creating web socket proxying tracking listener: %w", op, err) + } + statsHandler, err := metric.InstrumentClusterStatsHandler(w.baseContext) if err != nil { return nil, errors.Wrap(w.baseContext, err, op) @@ -214,10 +233,11 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger, } return func() { + handleSecondaryConnection(cancelCtx, metric.InstrumentWorkerClusterTrackingListener(revPkiWorkerTrackingListener, "reverse-grpc"), + metric.InstrumentWorkerClusterTrackingListener(dataPlaneProxyTrackingListener, "multihop-proxy-dataplane"), w.downstreamRoutes) go w.workerAuthSplitListener.Start() go httpServer.Serve(proxyListener) go ln.GrpcServer.Serve(metric.InstrumentWorkerClusterTrackingListener(pkiWorkerTrackingListener, "grpc")) - go handleSecondaryConnection(cancelCtx, metric.InstrumentWorkerClusterTrackingListener(revPkiWorkerTrackingListener, "reverse-grpc"), w.downstreamRoutes, -1) }, nil } diff --git a/internal/daemon/worker/status.go b/internal/daemon/worker/status.go index e4db0fa3b..cc0ac5222 100644 --- a/internal/daemon/worker/status.go +++ b/internal/daemon/worker/status.go @@ -21,7 +21,7 @@ import ( var firstStatusCheckPostHooks []func(context.Context, *Worker) error -var downstreamWorkersFactory func(ctx context.Context, workerId string) (downstreamers, error) +var downstreamWorkersFactory func(ctx context.Context, workerId string, ver string) (downstreamers, error) type LastStatusInformation struct { *pbs.StatusResponse @@ -305,7 +305,7 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess // If we have post hooks for after the first status check, run them now if w.everAuthenticated.CAS(authenticationStatusFirstAuthentication, authenticationStatusFirstStatusRpcSuccessful) { if downstreamWorkersFactory != nil { - w.downstreamWorkers, err = downstreamWorkersFactory(cancelCtx, w.LastStatusSuccess().WorkerId) + w.downstreamWorkers, err = downstreamWorkersFactory(cancelCtx, w.LastStatusSuccess().WorkerId, versionInfo.FullVersionNumber(false)) if err != nil { event.WriteError(cancelCtx, op, err) w.conf.ServerSideShutdownCh <- struct{}{} @@ -349,10 +349,10 @@ func (w *Worker) updateAddresses(cancelCtx context.Context, addrs []string, addr } // regardless of whether or not it's a new address, we need to set - // them for dialingListeners + // them for secondary connections for _, as := range *addressReceivers { switch { - case as.Type() == dialingListenerReceiverType: + case as.Type() == secondaryConnectionReceiverType: tmpAddrs := make([]string, len(addrs)) copy(tmpAddrs, addrs) if len(tmpAddrs) == 0 { diff --git a/internal/daemon/worker/worker.go b/internal/daemon/worker/worker.go index 57bf1906c..0bcef2a93 100644 --- a/internal/daemon/worker/worker.go +++ b/internal/daemon/worker/worker.go @@ -53,7 +53,7 @@ type downstreamRouter interface { // ProcessPendingConnections starts a function that continually processes // incoming client connections. This only returns when the provided context // is done. - StartProcessingPendingConnections(context.Context, func() string) + StartProcessingPendingConnections(context.Context, func() string) error } // downstreamers provides at least a minimum interface that must be met by a @@ -446,7 +446,9 @@ func (w *Worker) Start() error { } go func() { defer w.tickerWg.Done() - w.downstreamRoutes.StartProcessingPendingConnections(w.baseContext, servNameFn) + if err := w.downstreamRoutes.StartProcessingPendingConnections(w.baseContext, servNameFn); err != nil { + errors.Wrap(w.baseContext, err, op) + } }() go func() { defer w.tickerWg.Done()