feature (downstreams): Add hooks for optional downstream workers/router (#2359)

This includes the required interfaces, factories, error codes, etc
This commit is contained in:
Jim authored and GitHub committed 2022-08-12 11:08:25 -04:00
1 parent 222a935223
commit 2818bb65cd
11 files changed
+307 -43

No files matched your search

+85
View File
@@ -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
}