From 011a7cf640bb7b3cd6e86f24246c75193640681e Mon Sep 17 00:00:00 2001 From: Irena Rindos Date: Wed, 16 Nov 2022 15:34:39 -0500 Subject: [PATCH] Dataplane routing (#2623) * addition of dataplane routing --- go.mod | 2 +- go.sum | 4 ++-- internal/daemon/common/const.go | 5 ++++- internal/daemon/controller/controller.go | 13 ++++++++----- internal/daemon/controller/listeners.go | 12 +++++------- internal/daemon/worker/listeners.go | 18 ++++++++++++++---- internal/daemon/worker/status.go | 4 ++-- internal/daemon/worker/worker.go | 6 ++++-- 8 files changed, 40 insertions(+), 24 deletions(-) diff --git a/go.mod b/go.mod index 93d9dadfa..725f41066 100644 --- a/go.mod +++ b/go.mod @@ -93,7 +93,7 @@ require ( github.com/DATA-DOG/go-sqlmock v1.5.0 github.com/cenkalti/backoff/v4 v4.1.0 github.com/hashicorp/go-kms-wrapping/extras/kms/v2 v2.0.0-20220711120347-32232bae6803 - github.com/hashicorp/nodeenrollment v0.1.17 + github.com/hashicorp/nodeenrollment v0.1.18 github.com/kelseyhightower/envconfig v1.4.0 golang.org/x/net v0.0.0-20220624214902-1bab6f366d9e ) diff --git a/go.sum b/go.sum index 77b9d4e56..2a0ab91a8 100644 --- a/go.sum +++ b/go.sum @@ -743,8 +743,8 @@ 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= diff --git a/internal/daemon/common/const.go b/internal/daemon/common/const.go index 27b5e2925..6c8c96046 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" + WebsocketProxyingAlpnValue = "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 9e6eeaf6c..9f03d9cfd 100644 --- a/internal/daemon/controller/controller.go +++ b/internal/daemon/controller/controller.go @@ -38,6 +38,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" @@ -55,7 +56,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 @@ -77,8 +78,8 @@ type downstreamers interface { var ( downstreamRouterFactory func() downstreamRouter - downstreamersFactory func(context.Context, string) (downstreamers, error) - downstreamWorkersTickerFactory func(context.Context, string, downstreamers, downstreamRouter) (downstreamWorkersTicker, error) + downstreamersFactory func(context.Context, string, string) (downstreamers, error) + downstreamWorkersTickerFactory func(context.Context, string, string, downstreamers, downstreamRouter) (downstreamWorkersTicker, error) commandClientFactory func(context.Context, *Controller) error ) @@ -393,7 +394,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) } @@ -483,7 +485,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 32b6af46c..6f4a27336 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() } @@ -242,18 +242,16 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error ln.GrpcServer = workerServer return func() { + err := handleSecondaryConnection(c.baseContext, multiplexingReverseGrpcListener, 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, multiplexingReverseGrpcListener, c.downstreamRoutes, -1) - if err != nil { - event.WriteError(c.baseContext, op, err, event.WithInfoMsg("handleSecondaryConnection error")) - } - }() go func() { err := ln.GrpcServer.Serve(multiplexingAuthedListener) if err != nil { diff --git a/internal/daemon/worker/listeners.go b/internal/daemon/worker/listeners.go index e091f738e..c0375ffcb 100644 --- a/internal/daemon/worker/listeners.go +++ b/internal/daemon/worker/listeners.go @@ -32,9 +32,12 @@ 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, l2 net.Listener, _ any) error { if l != nil { - return l.Close() + l.Close() + } + if l2 != nil { + l2.Close() } return nil } @@ -170,7 +173,14 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger, // 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) + return nil, fmt.Errorf("error instantiating reverse grpc split listener: %w", err) + } + + // Connections coming in here are authed by nodeenrollment and are for the + // multi-hop session-proxying + websocketProxyingListener, err := w.workerAuthSplitListener.GetListener(common.WebsocketProxyingAlpnValue) + if err != nil { + return nil, fmt.Errorf("error instantiating websocket proxying split listener: %w", err) } // This wraps the reverse grpc pki worker connections with a listener which @@ -213,10 +223,10 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger, } return func() { + handleSecondaryConnection(cancelCtx, revPkiWorkerTrackingListener, websocketProxyingListener, w.downstreamRoutes) go w.workerAuthSplitListener.Start() go httpServer.Serve(proxyListener) go ln.GrpcServer.Serve(pkiWorkerTrackingListener) - go handleSecondaryConnection(cancelCtx, revPkiWorkerTrackingListener, w.downstreamRoutes, -1) }, nil } diff --git a/internal/daemon/worker/status.go b/internal/daemon/worker/status.go index 09c1f1b62..c4864d876 100644 --- a/internal/daemon/worker/status.go +++ b/internal/daemon/worker/status.go @@ -20,7 +20,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 @@ -308,7 +308,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.VersionNumber()) if err != nil { event.WriteError(cancelCtx, op, err) w.conf.ServerSideShutdownCh <- struct{}{} diff --git a/internal/daemon/worker/worker.go b/internal/daemon/worker/worker.go index 1d3174249..8e2a49256 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 @@ -447,7 +447,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()