Add FullFetch to resource updating ticker (#3947)

* Add full fetch for controllers w/o refresh tokens

internal/clientcache/internal/daemon/options.go

@@ -5,13 +5,19 @@ package daemon
 
 import (
 	"context"
+	"fmt"
+	"time"
 
 	"github.com/hashicorp/boundary/internal/clientcache/internal/cache"
 )
 
 type options struct {
-	withDebug                   bool
-	withBoundaryTokenReaderFunc cache.BoundaryTokenReaderFn
+	withDebug                              bool
+	withRefreshInterval                    time.Duration
+	withFullFetchInterval                  time.Duration
+	testWithIntervalRandomizationFactor    float64
+	testWithIntervalRandomizationFactorSet bool
+	withBoundaryTokenReaderFunc            cache.BoundaryTokenReaderFn
 }
 
 // Option - how options are passed as args
@@ -40,6 +46,41 @@ func WithDebug(_ context.Context, debug bool) Option {
 	}
 }
 
+// withRefreshInterval provides an optional refresh interval.
+func withRefreshInterval(_ context.Context, d time.Duration) Option {
+	return func(o *options) error {
+		if d <= 0 {
+			return fmt.Errorf("provided refresh interval %q must be positive", d)
+		}
+		o.withRefreshInterval = d
+		return nil
+	}
+}
+
+// withFullFetchInterval provides an optional full fetch interval.
+func withFullFetchInterval(_ context.Context, d time.Duration) Option {
+	return func(o *options) error {
+		if d <= 0 {
+			return fmt.Errorf("provided full fetch interval %q must be positive", d)
+		}
+		o.withFullFetchInterval = d
+		return nil
+	}
+}
+
+// testWithIntervalRandomizationFactor provides an optional test interval
+// randomization factor.
+func testWithIntervalRandomizationFactor(_ context.Context, f float64) Option {
+	return func(o *options) error {
+		if f < 0 {
+			return fmt.Errorf("testWithIntervalRandomizationFactor must be non negative")
+		}
+		o.testWithIntervalRandomizationFactor = f
+		o.testWithIntervalRandomizationFactorSet = true
+		return nil
+	}
+}
+
 // WithBoundaryTokenReaderFunc provides an option for specifying a BoundaryTokenReaderFn
 func WithBoundaryTokenReaderFunc(_ context.Context, fn cache.BoundaryTokenReaderFn) Option {
 	return func(o *options) error {

internal/clientcache/internal/daemon/options_test.go

@@ -6,6 +6,7 @@ package daemon
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/hashicorp/boundary/api/authtokens"
 	"github.com/hashicorp/boundary/internal/clientcache/internal/cache"
@@ -30,6 +31,54 @@ func Test_GetOpts(t *testing.T) {
 		testOpts.withDebug = true
 		assert.Equal(t, opts, testOpts)
 	})
+	t.Run("withRefreshInterval", func(t *testing.T) {
+		opts, err := getOpts(withRefreshInterval(ctx, time.Second))
+		require.NoError(t, err)
+		testOpts := getDefaultOptions()
+		testOpts.withRefreshInterval = time.Second
+		assert.Equal(t, opts, testOpts)
+	})
+	t.Run("withRefreshInterval_negative", func(t *testing.T) {
+		opts, err := getOpts(withRefreshInterval(ctx, -time.Second))
+		require.ErrorContains(t, err, "must be positive")
+		testOpts := getDefaultOptions()
+		assert.Equal(t, opts, testOpts)
+	})
+	t.Run("withFullFetchInterval", func(t *testing.T) {
+		opts, err := getOpts(withFullFetchInterval(ctx, time.Second))
+		require.NoError(t, err)
+		testOpts := getDefaultOptions()
+		testOpts.withFullFetchInterval = time.Second
+		assert.Equal(t, opts, testOpts)
+	})
+	t.Run("withFullFetchInterval_negative", func(t *testing.T) {
+		opts, err := getOpts(withFullFetchInterval(ctx, -time.Second))
+		require.ErrorContains(t, err, "must be positive")
+		testOpts := getDefaultOptions()
+		assert.Equal(t, opts, testOpts)
+	})
+	t.Run("testWithIntervalRandomizationFactor", func(t *testing.T) {
+		opts, err := getOpts(testWithIntervalRandomizationFactor(ctx, 0.2))
+		require.NoError(t, err)
+		testOpts := getDefaultOptions()
+		testOpts.testWithIntervalRandomizationFactor = 0.2
+		testOpts.testWithIntervalRandomizationFactorSet = true
+		assert.Equal(t, opts, testOpts)
+	})
+	t.Run("testWithIntervalRandomizationFactor zero", func(t *testing.T) {
+		opts, err := getOpts(testWithIntervalRandomizationFactor(ctx, 0))
+		require.NoError(t, err)
+		testOpts := getDefaultOptions()
+		testOpts.testWithIntervalRandomizationFactor = 0
+		testOpts.testWithIntervalRandomizationFactorSet = true
+		assert.Equal(t, opts, testOpts)
+	})
+	t.Run("testWithIntervalRandomizationFactor negative", func(t *testing.T) {
+		opts, err := getOpts(testWithIntervalRandomizationFactor(ctx, -0.2))
+		require.ErrorContains(t, err, "must be non negative")
+		testOpts := getDefaultOptions()
+		assert.Equal(t, opts, testOpts)
+	})
 	t.Run("WithBoundaryTokenReaderFunc", func(t *testing.T) {
 		var f cache.BoundaryTokenReaderFn = func(ctx context.Context, addr, token string) (*authtokens.AuthToken, error) {
 			return nil, nil

internal/clientcache/internal/daemon/server.go

@@ -60,14 +60,15 @@ type CacheServer struct {
 }
 
 type Config struct {
-	ContextCancel          context.CancelFunc
-	RefreshIntervalSeconds int64
-	DatabaseUrl            string
-	StoreDebug             bool
-	LogLevel               string
-	LogFormat              string
-	LogWriter              io.Writer
-	DotDirectory           string
+	ContextCancel     context.CancelFunc
+	RefreshInterval   time.Duration
+	FullFetchInterval time.Duration
+	DatabaseUrl       string
+	StoreDebug        bool
+	LogLevel          string
+	LogFormat         string
+	LogWriter         io.Writer
+	DotDirectory      string
 }
 
 func (sc *Config) validate(ctx context.Context) error {
@@ -231,7 +232,14 @@ func (s *CacheServer) Serve(ctx context.Context, cmd Commander, opt ...Option) e
 	if err != nil {
 		return errors.Wrap(ctx, err, op)
 	}
-	tic, err := newRefreshTicker(ctx, s.conf.RefreshIntervalSeconds, refreshService)
+	ticOptions := []Option{}
+	if s.conf.RefreshInterval > 0 {
+		ticOptions = append(ticOptions, withRefreshInterval(ctx, s.conf.RefreshInterval))
+	}
+	if s.conf.FullFetchInterval > 0 {
+		ticOptions = append(ticOptions, withFullFetchInterval(ctx, s.conf.FullFetchInterval))
+	}
+	tic, err := newRefreshTicker(ctx, refreshService, ticOptions...)
 	if err != nil {
 		return errors.Wrap(ctx, err, op)
 	}
@@ -239,10 +247,14 @@ func (s *CacheServer) Serve(ctx context.Context, cmd Commander, opt ...Option) e
 	tickingCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	var tickerWg sync.WaitGroup
-	tickerWg.Add(1)
+	tickerWg.Add(2)
 	go func() {
 		defer tickerWg.Done()
-		tic.start(tickingCtx)
+		tic.startRefresh(tickingCtx)
+	}()
+	go func() {
+		defer tickerWg.Done()
+		tic.startFullFetch(tickingCtx)
 	}()
 
 	mux := http.NewServeMux()

internal/clientcache/internal/daemon/testing.go

@@ -33,11 +33,12 @@ func NewTestServer(t *testing.T, cmd Commander, opt ...Option) *TestServer {
 	dotDir := t.TempDir()
 
 	cfg := &Config{
-		ContextCancel:          cancel,
-		RefreshIntervalSeconds: DefaultRefreshIntervalSeconds,
-		StoreDebug:             opts.withDebug,
-		LogWriter:              io.Discard,
-		DotDirectory:           dotDir,
+		ContextCancel:     cancel,
+		RefreshInterval:   defaultRefreshInterval,
+		FullFetchInterval: defaultFullFetchInterval,
+		StoreDebug:        opts.withDebug,
+		LogWriter:         io.Discard,
+		DotDirectory:      dotDir,
 	}
 
 	s, err := New(ctx, cfg)

internal/clientcache/internal/daemon/ticker.go

@@ -5,6 +5,7 @@ package daemon
 
 import (
 	"context"
+	"math/rand"
 	"time"
 
 	"github.com/hashicorp/boundary/internal/clientcache/internal/cache"
@@ -14,37 +15,70 @@ import (
 )
 
 const (
-	DefaultRefreshIntervalSeconds = 15
+	DefaultRefreshIntervalSeconds = 60
 	defaultRefreshInterval        = DefaultRefreshIntervalSeconds * time.Second
+
+	DefaultFullFetchIntervalSeconds = 60 * 5
+	defaultFullFetchInterval        = DefaultFullFetchIntervalSeconds * time.Second
+
+	defaultRandomizationFactor float64 = 0.2
 )
 
 type refreshService interface {
 	Refresh(context.Context, ...cache.Option) error
+	FullFetch(context.Context, ...cache.Option) error
 }
 
 type refreshTicker struct {
-	tickerCtx       context.Context
-	refreshInterval time.Duration
-	refreshChan     chan struct{}
-	refresher       refreshService
+	tickerCtx           context.Context
+	refreshInterval     time.Duration
+	fullFetchInterval   time.Duration
+	randomizationFactor float64
+	rand                *rand.Rand
+	refreshChan         chan struct{}
+	refresher           refreshService
 }
 
-func newRefreshTicker(ctx context.Context, refreshIntervalSeconds int64, refresh refreshService) (*refreshTicker, error) {
+// newRefreshTicker creates a new refresh ticker. Accepted options are
+// withRefreshInterval, and withFullFetchInterval.
+// testWithIntervalRandomizationFactor is provided for tests.
+// The intervals can deviate +/- randomizationFactor * configured interval.
+// For example, an interval of 10 seconds and a randomization factor of 0.2 will
+// encounter intervals of 8 to 12 seconds since 10 * .2 is 2 seconds and 10 +/- 2
+// is 8 to 12.
+func newRefreshTicker(ctx context.Context, refresh refreshService, opt ...Option) (*refreshTicker, error) {
 	const op = "daemon.newRefreshTicker"
 	switch {
-	case refreshIntervalSeconds == 0:
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "refresh interval seconds is missing")
 	case util.IsNil(refresh):
 		return nil, errors.New(ctx, errors.InvalidParameter, op, "refreshing function is missing")
 	}
 
-	refreshInterval := defaultRefreshInterval
-	if refreshIntervalSeconds != 0 {
-		refreshInterval = time.Duration(refreshIntervalSeconds) * time.Second
+	opts, err := getOpts(opt...)
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op)
 	}
+
+	refreshInterval := defaultRefreshInterval
+	if opts.withRefreshInterval != 0 {
+		refreshInterval = opts.withRefreshInterval
+	}
+
+	fullFetchInterval := defaultFullFetchInterval
+	if opts.withFullFetchInterval != 0 {
+		fullFetchInterval = opts.withFullFetchInterval
+	}
+
+	randomizationFactor := defaultRandomizationFactor
+	if opts.testWithIntervalRandomizationFactorSet {
+		randomizationFactor = opts.testWithIntervalRandomizationFactor
+	}
+
 	return &refreshTicker{
-		refreshInterval: refreshInterval,
-		refresher:       refresh,
+		refreshInterval:     refreshInterval,
+		fullFetchInterval:   fullFetchInterval,
+		randomizationFactor: randomizationFactor,
+		rand:                rand.New(rand.NewSource(time.Now().UnixNano())),
+		refresher:           refresh,
 
 		// We make this channel size 1 so if something happens midway through the refresh
 		// we can immediately refresh again immediately to pick up something that might have been
@@ -64,8 +98,30 @@ func (rt *refreshTicker) refresh() {
 	}
 }
 
-func (rt *refreshTicker) start(ctx context.Context) {
-	const op = "daemon.(refreshTicker).start"
+func (rt *refreshTicker) startFullFetch(ctx context.Context) {
+	const op = "daemon.(refreshTicker).startFullFetch"
+	timer := time.NewTimer(0)
+	for {
+		// full fetch doesn't listen to the refreshChan since completely new
+		// users are handled by refresh since we don't know if the new user
+		// is for a boundary instance that supports refresh tokens.
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+		}
+		if err := rt.refresher.FullFetch(ctx); err != nil {
+			event.WriteError(rt.tickerCtx, op, err)
+		}
+		timer.Reset(rt.nextIntervalWithRandomness(rt.fullFetchInterval))
+	}
+}
+
+// startRefresh repeatedly calls Refresh on the this refreshTicker's refresher
+// at some interval.  This interval can be overwritten by using the withDuration
+// option when calling startRefresh
+func (rt *refreshTicker) startRefresh(ctx context.Context) {
+	const op = "daemon.(refreshTicker).startRefresh"
 	timer := time.NewTimer(0)
 	for {
 		select {
@@ -77,6 +133,25 @@ func (rt *refreshTicker) start(ctx context.Context) {
 		if err := rt.refresher.Refresh(ctx); err != nil {
 			event.WriteError(rt.tickerCtx, op, err)
 		}
-		timer.Reset(rt.refreshInterval)
+		timer.Reset(rt.nextIntervalWithRandomness(rt.refreshInterval))
 	}
 }
+
+// nextIntervalWithRandomness applies the ticker's randomization factor to the
+// provided duration and returns a newly calculated duration.  If the tickers
+// randomization factor is 0 the returned duration is the same as the provided.
+func (rt *refreshTicker) nextIntervalWithRandomness(d time.Duration) time.Duration {
+	if rt.randomizationFactor == 0 {
+		return d
+	}
+
+	randFactor := rt.rand.Float64()
+	randFactor = randFactor * rt.randomizationFactor
+
+	randDur := randFactor * float64(d)
+	// Half a chance to be faster, not slower
+	if rt.rand.Float32() > 0.5 {
+		randDur = -1 * randDur
+	}
+	return d + time.Duration(randDur)
+}

internal/clientcache/internal/daemon/ticker_test.go

@@ -18,11 +18,11 @@ func TestTickerRefresh(t *testing.T) {
 
 	refresher := &fakeRefresher{make(chan struct{})}
 
-	rt, err := newRefreshTicker(ctx, 60, refresher)
+	rt, err := newRefreshTicker(ctx, refresher)
 	require.NoError(t, err)
 
 	tickerCtx, tickerCancel := context.WithCancel(ctx)
-	go rt.start(tickerCtx)
+	go rt.startRefresh(tickerCtx)
 	t.Cleanup(func() {
 		tickerCancel()
 	})
@@ -47,6 +47,56 @@ func TestTickerRefresh(t *testing.T) {
 	}
 }
 
+func TestTickerFullFetch(t *testing.T) {
+	ctx := context.Background()
+
+	refresher := &fakeRefresher{make(chan struct{})}
+
+	rt, err := newRefreshTicker(ctx, refresher)
+	require.NoError(t, err)
+
+	tickerCtx, tickerCancel := context.WithCancel(ctx)
+	go rt.startFullFetch(tickerCtx)
+	t.Cleanup(func() {
+		tickerCancel()
+	})
+
+	// let the normal start ticker refresh things,
+	<-refresher.called
+
+	testCtx, testCancel := context.WithTimeout(ctx, 100*time.Millisecond)
+	defer testCancel()
+	rt.refresh()
+	select {
+	case <-refresher.called:
+		assert.Fail(t, "full fetch called as a result of refresh()")
+	case <-testCtx.Done():
+	}
+}
+
+func TestNextIntervalWithRandomness(t *testing.T) {
+	ctx := context.Background()
+	refresher := &fakeRefresher{make(chan struct{})}
+	rt, err := newRefreshTicker(ctx, refresher, testWithIntervalRandomizationFactor(ctx, .2))
+	require.NoError(t, err)
+
+	interval := 10 * time.Second
+	max, min := interval, interval
+	for i := 0; i < 100; i++ {
+		got := rt.nextIntervalWithRandomness(10 * time.Second)
+		if got > max {
+			max = got
+		}
+		if got < min {
+			min = got
+		}
+	}
+	assert.GreaterOrEqual(t, min, 8*time.Second)
+	assert.LessOrEqual(t, max, 12*time.Second)
+	assert.Less(t, min, interval)
+	assert.Greater(t, max, interval)
+}
+
 type fakeRefresher struct {
 	called chan struct{}
 }
@@ -55,3 +105,8 @@ func (r *fakeRefresher) Refresh(context.Context, ...cache.Option) error {
 	r.called <- struct{}{}
 	return nil
 }
+
+func (r *fakeRefresher) FullFetch(context.Context, ...cache.Option) error {
+	r.called <- struct{}{}
+	return nil
+}