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

internal/daemon/controller+worker: propagate downstream timeout (#5083)

Browse Source 
Propagate the downstream worker timeout into the
downstream receiver and downstream ticker
so we have a configurable consistent timeout
value to use for worker to worker interactions.

Also change the atomic.Pointer to an atomic.Int64
for consistency with other values.
This commit is contained in:
Johan Brandhorst-Satzkorn authored and GitHub committed 2024-09-11 12:19:11 -04:00
1 parent 83b61275d5
commit 41de35f2b1
4 files changed
+35 -36
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/daemon/controller/controller.go
+17 -18
View File
@@ -10,7 +10,6 @@ import (
"strings"
"sync"
"sync/atomic"
"time"
"github.com/hashicorp/boundary/internal/alias"
talias "github.com/hashicorp/boundary/internal/alias/target"
@@ -89,10 +88,10 @@ type downstreamWorkersTicker interface {
}
var (
downstreamReceiverFactory func() downstreamReceiver
downstreamReceiverFactory func(*atomic.Int64) (downstreamReceiver, error)
downstreamersFactory func(context.Context, string, string) (common.Downstreamers, error)
downstreamWorkersTickerFactory func(context.Context, string, string, common.Downstreamers, downstreamReceiver, time.Duration) (downstreamWorkersTicker, error)
downstreamWorkersTickerFactory func(context.Context, string, string, common.Downstreamers, downstreamReceiver, *atomic.Int64) (downstreamWorkersTicker, error)
commandClientFactory func(context.Context, *Controller) error
extControllerFactory func(ctx context.Context, c *Controller, r db.Reader, w db.Writer, kms *kms.Kms) (intglobals.ControllerExtension, error)
)
@@ -124,7 +123,7 @@ type Controller struct {
// because they are casted to time.Duration.
workerStatusGracePeriod *atomic.Int64
livenessTimeToStale *atomic.Int64
getDownstreamWorkersTimeout *atomic.Pointer[time.Duration]
getDownstreamWorkersTimeout *atomic.Int64
apiGrpcServer *grpc.Server
apiGrpcServerListener grpcServerListener
@@ -190,11 +189,7 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
downstreamConnManager: cluster.NewDownstreamManager(),
workerStatusGracePeriod: new(atomic.Int64),
livenessTimeToStale: new(atomic.Int64),
getDownstreamWorkersTimeout: new(atomic.Pointer[time.Duration]),
}
if downstreamReceiverFactory != nil {
c.downstreamConns = downstreamReceiverFactory()
getDownstreamWorkersTimeout: new(atomic.Int64),
}
c.started.Store(false)
@@ -243,11 +238,17 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
switch conf.RawConfig.Controller.GetDownstreamWorkersTimeoutDuration {
case 0:
to := server.DefaultLiveness
c.getDownstreamWorkersTimeout.Store(&to)
c.getDownstreamWorkersTimeout.Store(int64(server.DefaultLiveness))
default:
to := conf.RawConfig.Controller.GetDownstreamWorkersTimeoutDuration
c.getDownstreamWorkersTimeout.Store(&to)
c.getDownstreamWorkersTimeout.Store(int64(conf.RawConfig.Controller.GetDownstreamWorkersTimeoutDuration))
}
if downstreamReceiverFactory != nil {
var err error
c.downstreamConns, err = downstreamReceiverFactory(c.getDownstreamWorkersTimeout)
if err != nil {
return nil, fmt.Errorf("%s: unable to initialize downstream receiver: %w", op, err)
}
}
clusterListeners := make([]*base.ServerListener, 0)
@@ -591,7 +592,7 @@ func (c *Controller) Start() error {
// we'll use "root" to designate that this is the root of the graph (aka
// a controller)
boundVer := version.Get().VersionNumber()
dswTicker, err := downstreamWorkersTickerFactory(c.baseContext, "root", boundVer, c.downstreamWorkers, c.downstreamConns, *c.getDownstreamWorkersTimeout.Load())
dswTicker, err := downstreamWorkersTickerFactory(c.baseContext, "root", boundVer, c.downstreamWorkers, c.downstreamConns, c.getDownstreamWorkersTimeout)
if err != nil {
return fmt.Errorf("error creating downstream workers ticker: %w", err)
}
@@ -709,11 +710,9 @@ func (c *Controller) ReloadTimings(newConfig *config.Config) error {
switch newConfig.Controller.GetDownstreamWorkersTimeoutDuration {
case 0:
to := server.DefaultLiveness
c.getDownstreamWorkersTimeout.Store(&to)
c.getDownstreamWorkersTimeout.Store(int64(server.DefaultLiveness))
default:
to := newConfig.Controller.GetDownstreamWorkersTimeoutDuration
c.getDownstreamWorkersTimeout.Store(&to)
c.getDownstreamWorkersTimeout.Store(int64(newConfig.Controller.GetDownstreamWorkersTimeoutDuration))
}
return nil
internal/daemon/worker/status_test.go
+1
View File
@@ -24,6 +24,7 @@ func TestWorkerWaitForNextSuccessfulStatusUpdate(t *testing.T) {
})
err := event.InitSysEventer(testLogger, testLock, "TestWorkerWaitForNextSuccessfulStatusUpdate", event.WithEventerConfig(testConfig))
require.NoError(t, err)
t.Cleanup(func() { event.TestResetSystEventer(t) })
for _, name := range []string{"ok", "timeout"} {
t.Run(name, func(t *testing.T) {
require := require.New(t)
internal/daemon/worker/worker.go
+15 -16
View File
@@ -97,7 +97,7 @@ type recorderManager interface {
// reverseConnReceiverFactory provides a simple factory which a Worker can use to
// create its reverseConnReceiver
var reverseConnReceiverFactory func() reverseConnReceiver
var reverseConnReceiverFactory func(*atomic.Int64) (reverseConnReceiver, error)
var recordingStorageFactory func(
ctx context.Context,
@@ -189,7 +189,7 @@ type Worker struct {
// because they are casted to time.Duration.
successfulStatusGracePeriod *atomic.Int64
statusCallTimeoutDuration *atomic.Int64
getDownstreamWorkersTimeoutDuration *atomic.Pointer[time.Duration]
getDownstreamWorkersTimeoutDuration *atomic.Int64
// AuthRotationNextRotation is useful in tests to understand how long to
// sleep
@@ -232,18 +232,13 @@ func New(ctx context.Context, conf *Config) (*Worker, error) {
localStorageState: new(atomic.Value),
successfulStatusGracePeriod: new(atomic.Int64),
statusCallTimeoutDuration: new(atomic.Int64),
getDownstreamWorkersTimeoutDuration: new(atomic.Pointer[time.Duration]),
getDownstreamWorkersTimeoutDuration: new(atomic.Int64),
upstreamConnectionState: new(atomic.Value),
downstreamWorkers: new(atomic.Pointer[downstreamersContainer]),
}
w.operationalState.Store(server.UnknownOperationalState)
w.localStorageState.Store(server.UnknownLocalStorageState)
if reverseConnReceiverFactory != nil {
w.downstreamReceiver = reverseConnReceiverFactory()
}
w.lastStatusSuccess.Store((*LastStatusInformation)(nil))
scheme := strconv.FormatInt(time.Now().UnixNano(), 36)
controllerResolver := manual.NewBuilderWithScheme(scheme)
@@ -338,15 +333,21 @@ func New(ctx context.Context, conf *Config) (*Worker, error) {
}
switch conf.RawConfig.Worker.GetDownstreamWorkersTimeoutDuration {
case 0:
to := server.DefaultLiveness
w.getDownstreamWorkersTimeoutDuration.Store(&to)
w.getDownstreamWorkersTimeoutDuration.Store(int64(server.DefaultLiveness))
default:
to := conf.RawConfig.Worker.GetDownstreamWorkersTimeoutDuration
w.getDownstreamWorkersTimeoutDuration.Store(&to)
w.getDownstreamWorkersTimeoutDuration.Store(int64(conf.RawConfig.Worker.GetDownstreamWorkersTimeoutDuration))
}
// FIXME: This is really ugly, but works.
session.CloseCallTimeout.Store(w.successfulStatusGracePeriod.Load())
if reverseConnReceiverFactory != nil {
var err error
w.downstreamReceiver, err = reverseConnReceiverFactory(w.getDownstreamWorkersTimeoutDuration)
if err != nil {
return nil, fmt.Errorf("%s: error creating reverse connection receiver: %w", op, err)
}
}
if eventListenerFactory != nil {
var err error
w.storageEventListener, err = eventListenerFactory(w)
@@ -429,11 +430,9 @@ func (w *Worker) Reload(ctx context.Context, newConf *config.Config) {
}
switch newConf.Worker.GetDownstreamWorkersTimeoutDuration {
case 0:
to := server.DefaultLiveness
w.getDownstreamWorkersTimeoutDuration.Store(&to)
w.getDownstreamWorkersTimeoutDuration.Store(int64(server.DefaultLiveness))
default:
to := newConf.Worker.GetDownstreamWorkersTimeoutDuration
w.getDownstreamWorkersTimeoutDuration.Store(&to)
w.getDownstreamWorkersTimeoutDuration.Store(int64(newConf.Worker.GetDownstreamWorkersTimeoutDuration))
}
// See comment about this in worker.go
session.CloseCallTimeout.Store(w.successfulStatusGracePeriod.Load())
internal/daemon/worker/worker_test.go
+2 -2
View File
@@ -181,7 +181,7 @@ func TestWorkerNew(t *testing.T) {
}
if util.IsNil(tt.in.Eventer) {
require.NoError(t, event.InitSysEventer(hclog.Default(), &sync.Mutex{}, "worker_test", event.WithEventerConfig(&event.EventerConfig{})))
defer event.TestResetSystEventer(t)
t.Cleanup(func() { event.TestResetSystEventer(t) })
tt.in.Eventer = event.SysEventer()
}
@@ -332,7 +332,7 @@ func TestSetupWorkerAuthStorage(t *testing.T) {
func Test_Worker_getSessionTls(t *testing.T) {
require.NoError(t, event.InitSysEventer(hclog.Default(), &sync.Mutex{}, "worker_test", event.WithEventerConfig(&event.EventerConfig{})))
defer event.TestResetSystEventer(t)
t.Cleanup(func() { event.TestResetSystEventer(t) })
conf := &Config{
Server: &base.Server{