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

Dataplane routing (#2623)

Browse Source 
* addition of dataplane routing
This commit is contained in:
Irena Rindos authored and GitHub committed 2022-11-16 15:34:39 -05:00
1 parent 72410cfa97
commit 011a7cf640
8 files changed
+40 -24
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

go.mod
+1 -1
View File
@@ -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
)
go.sum
+2 -2
View File
@@ -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=
internal/daemon/common/const.go
+4 -1
View File
@@ -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"
)
internal/daemon/controller/controller.go
+8 -5
View File
@@ -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)
}
internal/daemon/controller/listeners.go
+5 -7
View File
@@ -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 {
internal/daemon/worker/listeners.go
+14 -4
View File
@@ -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
}
internal/daemon/worker/status.go
+2 -2
View File
@@ -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{}{}
internal/daemon/worker/worker.go
+4 -2
View File
@@ -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()