diff --git a/internal/daemon/controller/controller.go b/internal/daemon/controller/controller.go index 2b019ef97..3da8a9b9d 100644 --- a/internal/daemon/controller/controller.go +++ b/internal/daemon/controller/controller.go @@ -18,6 +18,7 @@ import ( "github.com/hashicorp/boundary/internal/daemon/controller/handlers/health" "github.com/hashicorp/boundary/internal/daemon/controller/internal/metric" "github.com/hashicorp/boundary/internal/db" + "github.com/hashicorp/boundary/internal/errors" pluginhost "github.com/hashicorp/boundary/internal/host/plugin" "github.com/hashicorp/boundary/internal/host/static" "github.com/hashicorp/boundary/internal/iam" @@ -42,6 +43,37 @@ import ( "google.golang.org/grpc" ) +// downstreamRouter defines a min interface which must be met by a +// Controller.downstreamRoutes field +type downstreamRouter interface { + // StartRouteMgmtTicking starts a ticker which manages the router's + // connections. + StartRouteMgmtTicking(context.Context, func() string, int) error +} + +// downstreamWorkersTicker defines an interface for a ticker that maintains the +// graph of the controller's downstream workers +type downstreamWorkersTicker interface { + // StartDownstreamWorkersTicking is used by a Controller to maintain their + // graph of downstream workers. + StartDownstreamWorkersTicking(context.Context, int) error +} + +// downstreamers provides at least a minimum interface that must be met by a +// Controller.downstreamWorkers field which is far better than allowing any (empty +// interface) +type downstreamers interface { + // Root returns the root of the downstreamers' graph + Root() string +} + +var ( + downstreamRouterFactory func() downstreamRouter + + downstreamersFactory func(context.Context, string) (downstreamers, error) + downstreamWorkersTickerFactory func(context.Context, string, downstreamers, downstreamRouter) (downstreamWorkersTicker, error) +) + type Controller struct { conf *Config logger hclog.Logger @@ -55,6 +87,10 @@ type Controller struct { workerAuthCache *sync.Map + // downstream workers and routes to those workers + downstreamWorkers downstreamers + downstreamRoutes downstreamRouter + apiListeners []*base.ServerListener clusterListener *base.ServerListener @@ -106,6 +142,10 @@ func New(ctx context.Context, conf *Config) (*Controller, error) { apiListeners: make([]*base.ServerListener, 0), } + if downstreamRouterFactory != nil { + c.downstreamRoutes = downstreamRouterFactory() + } + c.started.Store(false) if conf.SecureRandomReader == nil { @@ -316,6 +356,13 @@ func New(ctx context.Context, conf *Config) (*Controller, error) { return nil, fmt.Errorf("unable to ensure worker auth roots exist: %w", err) } + if downstreamersFactory != nil { + c.downstreamWorkers, err = downstreamersFactory(ctx, "root") + if err != nil { + return nil, fmt.Errorf("unable to initialize downstream workers graph: %w", err) + } + } + return c, nil } @@ -365,6 +412,44 @@ func (c *Controller) Start() error { c.started.Store(true) }() + if c.downstreamRoutes != nil { + c.tickerWg.Add(1) + go func() { + defer c.tickerWg.Done() + err := c.downstreamRoutes.StartRouteMgmtTicking( + c.baseContext, + func() string { + switch { + case c.conf.RawConfig.Controller.Name != "": + return c.conf.RawConfig.Controller.Name + default: + return "unknown controller name" + } + }, + -1, + ) + if err != nil { + errors.Wrap(c.baseContext, err, op) + } + }() + } + 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) + if err != nil { + return fmt.Errorf("error creating downstream workers ticker: %w", err) + } + c.tickerWg.Add(1) + go func() { + defer c.tickerWg.Done() + err := dswTicker.StartDownstreamWorkersTicking(c.baseContext, -1) + if err != nil { + event.WriteSysEvent(c.baseContext, op, "error starting/running downstream workers ticker", "err", err.Error()) + } + }() + } + return nil } diff --git a/internal/daemon/controller/listeners.go b/internal/daemon/controller/listeners.go index 404ebf84e..8b87e71a3 100644 --- a/internal/daemon/controller/listeners.go +++ b/internal/daemon/controller/listeners.go @@ -24,7 +24,7 @@ import ( // the function that handles a secondary connection over a provided listener var handleSecondaryConnection = closeListener -func closeListener(_ context.Context, l net.Listener) error { +func closeListener(_ context.Context, l net.Listener, _ any, _ int) error { if l != nil { return l.Close() } @@ -221,7 +221,7 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error return func() { go splitListener.Start() - go handleSecondaryConnection(c.baseContext, multiplexingReverseGrpcListener) + go handleSecondaryConnection(c.baseContext, multiplexingReverseGrpcListener, c.downstreamRoutes, -1) go ln.GrpcServer.Serve(multiplexingAuthedListener) }, nil } diff --git a/internal/daemon/worker/addressreceiver.go b/internal/daemon/worker/addressreceiver.go index 5e52e868b..b86136dbf 100644 --- a/internal/daemon/worker/addressreceiver.go +++ b/internal/daemon/worker/addressreceiver.go @@ -13,12 +13,30 @@ func noopAddressReceivers(context.Context, *Worker) ([]addressReceiver, error) { return nil, nil } +type receiverType uint + +const ( + UnknownReceiverType receiverType = 0 + grpcResolverReceiverType receiverType = 1 + dialingListenerReceiverType receiverType = 2 +) + +// String returns a string representation of the receiverType +func (s receiverType) String() string { + return [...]string{ + "unknown", + "grpcResolver", + "dialingListener", + }[s] +} + // 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) + Type() receiverType } // grpcResolverReceiver is an addressReceiver which wraps a grpc manual.Resolver @@ -28,6 +46,11 @@ type grpcResolverReceiver struct { *manual.Resolver } +// IsDialingListener always returns +func (_ *grpcResolverReceiver) Type() receiverType { + return grpcResolverReceiverType +} + func (r grpcResolverReceiver) getState(addrs []string) resolver.State { var rAddrs []resolver.Address for _, a := range addrs { diff --git a/internal/daemon/worker/listeners.go b/internal/daemon/worker/listeners.go index 0f7ea3ccf..3f332b4a0 100644 --- a/internal/daemon/worker/listeners.go +++ b/internal/daemon/worker/listeners.go @@ -30,7 +30,7 @@ import ( // the function that handles a secondary connection over a provided listener var handleSecondaryConnection = closeListener -func closeListener(_ context.Context, l net.Listener) error { +func closeListener(_ context.Context, l net.Listener, _ any, _ int) error { if l != nil { return l.Close() } @@ -193,7 +193,7 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger, go w.workerAuthSplitListener.Start() go httpServer.Serve(proxyListener) go ln.GrpcServer.Serve(eventingListener) - go handleSecondaryConnection(cancelCtx, reverseGrpcListener) + go handleSecondaryConnection(cancelCtx, reverseGrpcListener, w.downstreamRoutes, -1) }, nil } diff --git a/internal/daemon/worker/status.go b/internal/daemon/worker/status.go index 24e9c797e..2fa90c97c 100644 --- a/internal/daemon/worker/status.go +++ b/internal/daemon/worker/status.go @@ -7,10 +7,9 @@ import ( "math/rand" "time" - pb "github.com/hashicorp/boundary/internal/gen/controller/servers" - "github.com/hashicorp/boundary/internal/daemon/worker/common" "github.com/hashicorp/boundary/internal/daemon/worker/session" + pb "github.com/hashicorp/boundary/internal/gen/controller/servers" pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/go-secure-stdlib/strutil" @@ -18,13 +17,15 @@ import ( var firstStatusCheckPostHooks []func(context.Context, *Worker) error +var downstreamWorkersFactory func(ctx context.Context, workerId string) (downstreamers, error) + type LastStatusInformation struct { *pbs.StatusResponse StatusTime time.Time LastCalculatedUpstreams []string } -func (w *Worker) startStatusTicking(cancelCtx context.Context, sessionManager session.Manager, addrReceivers []addressReceiver) { +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 @@ -65,7 +66,11 @@ func (w *Worker) startStatusTicking(cancelCtx context.Context, sessionManager se // LastStatusSuccess reports the last time we sent a successful // status request. func (w *Worker) LastStatusSuccess() *LastStatusInformation { - return w.lastStatusSuccess.Load().(*LastStatusInformation) + s, ok := w.lastStatusSuccess.Load().(*LastStatusInformation) + if !ok { + return nil + } + return s } // WaitForNextSuccessfulStatusUpdate waits for the next successful status. It's @@ -100,7 +105,7 @@ func (w *Worker) WaitForNextSuccessfulStatusUpdate() error { return nil } -func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager session.Manager, addressReceivers []addressReceiver) { +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 @@ -228,17 +233,31 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess 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) - for _, as := range addressReceivers { + for _, as := range *addressReceivers { as.SetAddresses(addrs) } } else if lastStatus == nil { - for _, as := range addressReceivers { + for _, as := range *addressReceivers { as.SetAddresses(addrs) } event.WriteSysEvent(cancelCtx, op, fmt.Sprintf("Upstreams after first status set to: %s", addrs)) } } + // regardless of whether or not it's a new address, we need to set + // them for dialingListeners + for _, as := range *addressReceivers { + switch { + case as.Type() == dialingListenerReceiverType: + tmpAddrs := make([]string, len(addrs)) + copy(tmpAddrs, addrs) + if len(tmpAddrs) == 0 { + tmpAddrs = append(tmpAddrs, w.conf.RawConfig.Worker.InitialUpstreams...) + } + as.SetAddresses(tmpAddrs) + } + } + w.lastStatusSuccess.Store(&LastStatusInformation{StatusResponse: result, StatusTime: time.Now(), LastCalculatedUpstreams: addrs}) for _, request := range result.GetJobsRequests() { @@ -272,6 +291,14 @@ 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) + if err != nil { + event.WriteError(cancelCtx, op, err) + w.conf.ServerSideShutdownCh <- struct{}{} + return + } + } for _, fn := range firstStatusCheckPostHooks { if err := fn(cancelCtx, w); err != nil { // If we can't verify status we can't be expected to behave diff --git a/internal/daemon/worker/worker.go b/internal/daemon/worker/worker.go index ba8c048bd..8eb15ec01 100644 --- a/internal/daemon/worker/worker.go +++ b/internal/daemon/worker/worker.go @@ -39,6 +39,26 @@ import ( type randFn func(length int) (string, error) +// downstreamRouter defines a min interface which must be met by a +// Worker.downstreamRoutes field +type downstreamRouter interface { + // StartRouteMgmtTicking starts a ticker which manages the router's + // connections. + StartRouteMgmtTicking(context.Context, func() string, int) error +} + +// downstreamers provides at least a minimum interface that must be met by a +// Worker.downstreamWorkers field which is far better than allowing any (empty +// interface) +type downstreamers interface { + // Root returns the root of the downstreamers' graph + Root() string +} + +// downstreamRouterFactory provides a simple factory which a Worker can use to +// create its downstreamRouter +var downstreamRouterFactory func() downstreamRouter + const ( authenticationStatusNeverAuthenticated uint32 = iota authenticationStatusFirstAuthentication @@ -91,6 +111,10 @@ type Worker struct { WorkerAuthRegistrationRequest string workerAuthSplitListener *nodeenet.SplitListener + // downstream workers and routes to those workers + downstreamWorkers downstreamers + downstreamRoutes downstreamRouter + // Test-specific options (and possibly hidden dev-mode flags) TestOverrideX509VerifyDnsName string TestOverrideX509VerifyCertPool *x509.CertPool @@ -117,6 +141,10 @@ func New(conf *Config) (*Worker, error) { WorkerAuthCurrentKeyId: new(ua.String), } + if downstreamRouterFactory != nil { + w.downstreamRoutes = downstreamRouterFactory() + } + w.lastStatusSuccess.Store((*LastStatusInformation)(nil)) scheme := strconv.FormatInt(time.Now().UnixNano(), 36) controllerResolver := manual.NewBuilderWithScheme(scheme) @@ -182,12 +210,6 @@ 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 @@ -305,6 +327,7 @@ func (w *Worker) Start() error { 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 errors.Wrap(w.baseContext, err, op, errors.WithMsg("error creating session manager")) @@ -316,17 +339,34 @@ func (w *Worker) Start() error { // Rather than deal with some of the potential error conditions for Add on // the waitgroup vs. Done (in case a function exits immediately), we will - // always start rotation and simply exit early if we're using KMS, and - // always add 2 here. - w.tickerWg.Add(2) + // always start rotation and simply exit early if we're using KMS + w.tickerWg.Add(3) go func() { defer w.tickerWg.Done() - w.startStatusTicking(w.baseContext, w.sessionManager, w.addressReceivers) + w.startStatusTicking(w.baseContext, w.sessionManager, &w.addressReceivers) }() go func() { defer w.tickerWg.Done() w.startAuthRotationTicking(w.baseContext) }() + go func() { + defer w.tickerWg.Done() + if w.downstreamRoutes != nil { + err := w.downstreamRoutes.StartRouteMgmtTicking( + w.baseContext, + func() string { + if s := w.LastStatusSuccess(); s != nil { + return s.WorkerId + } + return "unknown worker id" + }, + -1, // indicates the ticker should run until cancelled. + ) + if err != nil { + errors.Wrap(w.baseContext, err, op) + } + } + }() w.workerStartTime = time.Now() w.started.Store(true) diff --git a/internal/errors/code.go b/internal/errors/code.go index 958b4c66d..3a1abbc07 100644 --- a/internal/errors/code.go +++ b/internal/errors/code.go @@ -21,28 +21,30 @@ const ( Unknown Code = 0 // Unknown will be equal to a zero value for Codes // General function errors are reserved Codes 100-999 - InvalidParameter Code = 100 // InvalidParameter represents an invalid parameter for an operation. - InvalidAddress Code = 101 // InvalidAddress represents an invalid host address for an operation - InvalidPublicId Code = 102 // InvalidPublicId represents an invalid public Id for an operation - InvalidFieldMask Code = 103 // InvalidFieldMask represents an invalid field mast for an operation - EmptyFieldMask Code = 104 // EmptyFieldMask represents an empty field mask for an operation - KeyNotFound Code = 105 // KeyNotFound represents that a key/version was not found in the KMS - TicketAlreadyRedeemed Code = 106 // TicketAlreadyRedeemed represents that the ticket version has already been redeemed - TicketNotFound Code = 107 // TicketNotFound represents that the ticket was not found - Io Code = 108 // Io represents that an io error occurred in an underlying call (i.e binary.Write) - InvalidTimeStamp Code = 109 // InvalidTimeStamp represents an invalid time stamp for an operation - SessionNotFound Code = 110 // SessionNotFound represents that the session was not found - InvalidSessionState Code = 111 // InvalidSessionState represents that the session was in an invalid state - TokenMismatch Code = 112 // TokenMismatch represents that there was a token mismatch - TooShort Code = 113 // TooShort represents an error that means the provided input is not meeting minimum length requirements - AccountAlreadyAssociated Code = 114 // AccountAlreadyAssociated represents an attempt to associate an account failed since it was already associated. - InvalidJobRunState Code = 115 // InvalidJobRunState represents that a JobRun was in an invalid state - InvalidDynamicCredential Code = 116 // InvalidDynamicCredential represents that a dynamic credential for a session was in an invalid state - JobAlreadyRunning Code = 117 // JobAlreadyRunning represents that a Job is already running when an attempt to run again was made - SubtypeAlreadyRegistered Code = 118 // SubtypeAlreadyRegistered represents that a value has already been registered in the subtype registry system. - NoPathFound = 119 // NoPathFound represents an error when no path is found to a worker - WorkerNotFound = 120 // WorkerNotFound represents an error when a worker is not found in the graph of downstream workers - CycleFound = 121 // CycleFound represents an error when a cycle is found between a parent and child worker + InvalidParameter Code = 100 // InvalidParameter represents an invalid parameter for an operation. + InvalidAddress Code = 101 // InvalidAddress represents an invalid host address for an operation + InvalidPublicId Code = 102 // InvalidPublicId represents an invalid public Id for an operation + InvalidFieldMask Code = 103 // InvalidFieldMask represents an invalid field mast for an operation + EmptyFieldMask Code = 104 // EmptyFieldMask represents an empty field mask for an operation + KeyNotFound Code = 105 // KeyNotFound represents that a key/version was not found in the KMS + TicketAlreadyRedeemed Code = 106 // TicketAlreadyRedeemed represents that the ticket version has already been redeemed + TicketNotFound Code = 107 // TicketNotFound represents that the ticket was not found + Io Code = 108 // Io represents that an io error occurred in an underlying call (i.e binary.Write) + InvalidTimeStamp Code = 109 // InvalidTimeStamp represents an invalid time stamp for an operation + SessionNotFound Code = 110 // SessionNotFound represents that the session was not found + InvalidSessionState Code = 111 // InvalidSessionState represents that the session was in an invalid state + TokenMismatch Code = 112 // TokenMismatch represents that there was a token mismatch + TooShort Code = 113 // TooShort represents an error that means the provided input is not meeting minimum length requirements + AccountAlreadyAssociated Code = 114 // AccountAlreadyAssociated represents an attempt to associate an account failed since it was already associated. + InvalidJobRunState Code = 115 // InvalidJobRunState represents that a JobRun was in an invalid state + InvalidDynamicCredential Code = 116 // InvalidDynamicCredential represents that a dynamic credential for a session was in an invalid state + JobAlreadyRunning Code = 117 // JobAlreadyRunning represents that a Job is already running when an attempt to run again was made + SubtypeAlreadyRegistered Code = 118 // SubtypeAlreadyRegistered represents that a value has already been registered in the subtype registry system. + NoPathFound = 119 // NoPathFound represents an error when no path is found to a worker + WorkerNotFound = 120 // WorkerNotFound represents an error when a worker is not found in the graph of downstream workers + CycleFound = 121 // CycleFound represents an error when a cycle is found between a parent and child worker + WorkerConnNotFound = 122 // WorkerConnNotFound represents an error when a connection to a worker is not found + KmsWorkerUnsupportedOperation = 123 // KmsWorkerUnsupportedOperation represents an error when a KMS worker is not supported for an operation AuthAttemptExpired Code = 198 // AuthAttemptExpired represents an expired authentication attempt AuthMethodInactive Code = 199 // AuthMethodInactive represents an error that means the auth method is not active. diff --git a/internal/errors/code_test.go b/internal/errors/code_test.go index 631bd1d5f..e60476574 100644 --- a/internal/errors/code_test.go +++ b/internal/errors/code_test.go @@ -327,6 +327,16 @@ func TestCode_Both_String_Info(t *testing.T) { c: WorkerNotFound, want: WorkerNotFound, }, + { + name: "WorkerConnNotFound", + c: WorkerConnNotFound, + want: WorkerConnNotFound, + }, + { + name: "KmsWorkerUnsupportedOperation", + c: KmsWorkerUnsupportedOperation, + want: KmsWorkerUnsupportedOperation, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { diff --git a/internal/errors/info.go b/internal/errors/info.go index fdb8a63f1..80e7c0e7d 100644 --- a/internal/errors/info.go +++ b/internal/errors/info.go @@ -264,4 +264,12 @@ var errorCodeInfo = map[Code]Info{ Message: "cycle found", Kind: State, }, + WorkerConnNotFound: { + Message: "worker connection not found", + Kind: State, + }, + KmsWorkerUnsupportedOperation: { + Message: "unsupported operation for a kms worker", + Kind: State, + }, } diff --git a/internal/util/is_nil.go b/internal/util/is_nil.go new file mode 100644 index 000000000..62a2c19d8 --- /dev/null +++ b/internal/util/is_nil.go @@ -0,0 +1,15 @@ +package util + +import "reflect" + +// IsNil checks if the interface is nil +func IsNil(i interface{}) bool { + if i == nil { + return true + } + switch reflect.TypeOf(i).Kind() { + case reflect.Ptr, reflect.Map, reflect.Chan, reflect.Slice: + return reflect.ValueOf(i).IsNil() + } + return false +} diff --git a/internal/util/is_nil_test.go b/internal/util/is_nil_test.go new file mode 100644 index 000000000..2e785eb07 --- /dev/null +++ b/internal/util/is_nil_test.go @@ -0,0 +1,54 @@ +package util_test + +import ( + "fmt" + "testing" + + "github.com/hashicorp/boundary/internal/util" + "github.com/stretchr/testify/assert" +) + +func Test_IsNil(t *testing.T) { + t.Parallel() + + var testErrNilPtr *testError + var testMapNilPtr map[string]struct{} + var testArrayNilPtr *[1]string + var testChanNilPtr *chan string + var testSliceNilPtr *[]string + + var testChanString chan string + + tc := []struct { + i interface{} + want bool + }{ + {i: &testError{}, want: false}, + {i: testError{}, want: false}, + {i: &map[string]struct{}{}, want: false}, + {i: map[string]struct{}{}, want: false}, + {i: [1]string{}, want: false}, + {i: &[1]string{}, want: false}, + {i: &testChanString, want: false}, + {i: "string", want: false}, + {i: []string{}, want: false}, + {i: nil, want: true}, + {i: testErrNilPtr, want: true}, + {i: testMapNilPtr, want: true}, + {i: testArrayNilPtr, want: true}, + {i: testChanNilPtr, want: true}, + {i: testChanString, want: true}, + {i: testSliceNilPtr, want: true}, + } + + for i, tc := range tc { + t.Run(fmt.Sprintf("test #%d", i+1), func(t *testing.T) { + assert := assert.New(t) + assert.Equal(tc.want, util.IsNil(tc.i)) + }) + } +} + +type testError struct{} + +func (*testError) Error() string { return "error" }