diff --git a/internal/clientcache/cmd/daemon/start.go b/internal/clientcache/cmd/daemon/start.go index 16b28f753..ea0a4f26f 100644 --- a/internal/clientcache/cmd/daemon/start.go +++ b/internal/clientcache/cmd/daemon/start.go @@ -13,6 +13,7 @@ import ( "path/filepath" "strings" "sync" + "time" "github.com/hashicorp/boundary/internal/clientcache/internal/daemon" "github.com/hashicorp/boundary/internal/cmd/base" @@ -46,12 +47,13 @@ type server interface { type StartCommand struct { *base.Command - flagRefreshIntervalSeconds int64 - flagDatabaseUrl string - flagLogLevel string - flagLogFormat string - flagStoreDebug bool - flagBackground bool + flagRefreshInterval time.Duration + flagFullFetchInterval time.Duration + flagDatabaseUrl string + flagLogLevel string + flagLogFormat string + flagStoreDebug bool + flagBackground bool } func (c *StartCommand) Synopsis() string { @@ -95,12 +97,17 @@ func (c *StartCommand) Flags() *base.FlagSets { Target: &c.flagDatabaseUrl, Usage: `If set, specifies the URL used to connect to the sqlite database (store) for caching. This can refer to a file on disk (file://) from which a URL will be read; an env var (env://) from which the URL will be read; or a direct database URL.`, }) - f.Int64Var(&base.Int64Var{ + f.DurationVar(&base.DurationVar{ Name: "refresh-interval-seconds", - Target: &c.flagRefreshIntervalSeconds, - Usage: `If set, specifies the number of seconds between cache refreshes. Default: 5 minutes`, - Aliases: []string{"r"}, - Default: daemon.DefaultRefreshIntervalSeconds, + Target: &c.flagRefreshInterval, + Usage: `If set, specifies the number of seconds between refresh token supported cache refreshes. Default: 1 minute`, + Default: daemon.DefaultRefreshIntervalSeconds * time.Second, + }) + f.DurationVar(&base.DurationVar{ + Name: "full-fetch-interval-seconds", + Target: &c.flagFullFetchInterval, + Usage: `If set, specifies the number of seconds between full cache refresh for boundary instances which do not support refresh tokens. Default: 5 minutes`, + Default: daemon.DefaultFullFetchIntervalSeconds * time.Second, }) f.BoolVar(&base.BoolVar{ Name: "store-debug", @@ -175,14 +182,15 @@ func (c *StartCommand) Run(args []string) int { writers = append(writers, logFile) cfg := &daemon.Config{ - ContextCancel: cancel, - RefreshIntervalSeconds: c.flagRefreshIntervalSeconds, - DatabaseUrl: c.flagDatabaseUrl, - StoreDebug: c.flagStoreDebug, - LogLevel: c.flagLogLevel, - LogFormat: c.flagLogFormat, - LogWriter: io.MultiWriter(writers...), - DotDirectory: dotDir, + ContextCancel: cancel, + RefreshInterval: c.flagRefreshInterval, + FullFetchInterval: c.flagFullFetchInterval, + DatabaseUrl: c.flagDatabaseUrl, + StoreDebug: c.flagStoreDebug, + LogLevel: c.flagLogLevel, + LogFormat: c.flagLogFormat, + LogWriter: io.MultiWriter(writers...), + DotDirectory: dotDir, } srv, err := daemon.New(ctx, cfg) diff --git a/internal/clientcache/internal/cache/options.go b/internal/clientcache/internal/cache/options.go index 7a4e04b78..c6780a7b1 100644 --- a/internal/clientcache/internal/cache/options.go +++ b/internal/clientcache/internal/cache/options.go @@ -10,6 +10,8 @@ import ( type options struct { withUpdateLastAccessedTime bool withDbType dbw.DbType + withAuthTokenId string + withUserId string withTargetRetrievalFunc TargetRetrievalFunc withSessionRetrievalFunc SessionRetrievalFunc } @@ -42,6 +44,22 @@ func WithUpdateLastAccessedTime(b bool) Option { } } +// withUserId provides an option for providing an auth token id +func withAuthTokenId(id string) Option { + return func(o *options) error { + o.withAuthTokenId = id + return nil + } +} + +// withUserId provides an option for providing a user id +func withUserId(id string) Option { + return func(o *options) error { + o.withUserId = id + return nil + } +} + // WithTargetRetrievalFunc provides an option for specifying a targetRetrievalFunc func WithTargetRetrievalFunc(fn TargetRetrievalFunc) Option { return func(o *options) error { diff --git a/internal/clientcache/internal/cache/options_test.go b/internal/clientcache/internal/cache/options_test.go index 7724918f2..7f60342a4 100644 --- a/internal/clientcache/internal/cache/options_test.go +++ b/internal/clientcache/internal/cache/options_test.go @@ -32,6 +32,20 @@ func Test_GetOpts(t *testing.T) { testOpts.withUpdateLastAccessedTime = true assert.Equal(t, opts, testOpts) }) + t.Run("withUserId", func(t *testing.T) { + opts, err := getOpts(withUserId("u123")) + require.NoError(t, err) + testOpts := getDefaultOptions() + testOpts.withUserId = "u123" + assert.Equal(t, opts, testOpts) + }) + t.Run("withAuthTokenId", func(t *testing.T) { + opts, err := getOpts(withAuthTokenId("at123")) + require.NoError(t, err) + testOpts := getDefaultOptions() + testOpts.withAuthTokenId = "at123" + assert.Equal(t, opts, testOpts) + }) t.Run("WithTargetRetrievalFunc", func(t *testing.T) { var f TargetRetrievalFunc = func(ctx context.Context, addr, authTok string, refreshTok RefreshTokenValue) ([]*targets.Target, []string, RefreshTokenValue, error) { return nil, nil, "", nil diff --git a/internal/clientcache/internal/cache/refresh.go b/internal/clientcache/internal/cache/refresh.go index 342f0ab7e..37cee1ec2 100644 --- a/internal/clientcache/internal/cache/refresh.go +++ b/internal/clientcache/internal/cache/refresh.go @@ -88,11 +88,50 @@ func (r *RefreshService) cleanAndPickAuthTokens(ctx context.Context, u *user) (m return ret, nil } -// Refresh iterates over all tokens in the cache, attempts to read the -// resources for those tokens from boundary and updates the cache with -// the values retrieved there. Refresh accepts the options -// WithTarget which overwrites the default function used to retrieve the -// targets from a boundary address. +// refreshableUsers filters the provided users down to the set of users who can +// have their resources refreshed using refresh tokens. A user is returned by +// this method if it has no resources in storage at all or it has a refresh +// token for any resource. +func (r *RefreshService) refreshableUsers(ctx context.Context, in []*user) ([]*user, error) { + const op = "cache.(RefreshService).refreshableUsers" + var ret []*user + for _, u := range in { + rt, err := r.repo.listRefreshTokens(ctx, u) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + if len(rt) > 0 { + ret = append(ret, u) + continue + } + + // If any resource exists for this user, since it doesn't have a refresh + // token yet it is not refreshable. + ts, err := r.repo.searchTargets(ctx, "true", nil, withUserId(u.Id)) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + if len(ts) > 0 { + continue + } + ss, err := r.repo.searchSessions(ctx, "true", nil, withUserId(u.Id)) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + if len(ss) > 0 { + continue + } + ret = append(ret, u) + } + return ret, nil +} + +// Refresh iterates over all tokens in the cache that are for users who either +// have a refresh token or which do not have any resources in the cache yet. It +// then attempts to read those user's resources from boundary and updates the +// cache with the values retrieved there. Refresh accepts the options +// WithTargetRetrievalFunc and WithSessionRetrievalFunc which overwrites the +// default functions used to retrieve those resources from boundary. func (r *RefreshService) Refresh(ctx context.Context, opt ...Option) error { const op = "cache.(RefreshService).Refresh" if err := r.repo.cleanExpiredOrOrphanedAuthTokens(ctx); err != nil { @@ -106,6 +145,12 @@ func (r *RefreshService) Refresh(ctx context.Context, opt ...Option) error { if err != nil { return errors.Wrap(ctx, err, op) } + + us, err = r.refreshableUsers(ctx, us) + if err != nil { + return errors.Wrap(ctx, err, op) + } + var retErr error for _, u := range us { tokens, err := r.cleanAndPickAuthTokens(ctx, u) @@ -124,3 +169,55 @@ func (r *RefreshService) Refresh(ctx context.Context, opt ...Option) error { } return retErr } + +// FullFetch is used for users of boundary instances which do not support +// refresh tokens. For these users a call to FullFetch retrieves all resources +// for that user and updates the database with the result. Users are identified +// as being in this group if they have at least one resource in the cache and +// do not have any refresh tokens. Users who do not have any resources nor +// refresh tokens are updated using Refresh. +func (r *RefreshService) FullFetch(ctx context.Context, opt ...Option) error { + const op = "cache.(RefreshService).Refresh" + if err := r.repo.cleanExpiredOrOrphanedAuthTokens(ctx); err != nil { + return errors.Wrap(ctx, err, op) + } + if err := r.repo.syncKeyringlessTokensWithDb(ctx); err != nil { + return errors.Wrap(ctx, err, op) + } + + us, err := r.repo.listUsers(ctx) + if err != nil { + return errors.Wrap(ctx, err, op) + } + + removeUsers, err := r.refreshableUsers(ctx, us) + if err != nil { + return errors.Wrap(ctx, err, op) + } + removedUserIds := make(map[string]struct{}, len(removeUsers)) + for _, ru := range removeUsers { + removedUserIds[ru.Id] = struct{}{} + } + + var retErr error + for _, u := range us { + if _, ok := removedUserIds[u.Id]; ok { + continue + } + + tokens, err := r.cleanAndPickAuthTokens(ctx, u) + if err != nil { + retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op)) + continue + } + + if err := r.repo.fullFetchTargets(ctx, u, tokens, opt...); err != nil { + retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op)) + } + if err := r.repo.fullFetchSessions(ctx, u, tokens, opt...); err != nil { + retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op)) + } + + } + return retErr +} diff --git a/internal/clientcache/internal/cache/refresh_test.go b/internal/clientcache/internal/cache/refresh_test.go index ba3cc86f8..5525aa55a 100644 --- a/internal/clientcache/internal/cache/refresh_test.go +++ b/internal/clientcache/internal/cache/refresh_test.go @@ -52,6 +52,12 @@ func testStaticResourceRetrievalFunc[T any](t *testing.T, ret [][]T, removed [][ } } +func testFullFetchRetrievalFunc[T any](t *testing.T, ret []T) func(context.Context, string, string, RefreshTokenValue) ([]T, []string, RefreshTokenValue, error) { + return func(_ context.Context, _, _ string, _ RefreshTokenValue) ([]T, []string, RefreshTokenValue, error) { + return ret, nil, "", nil + } +} + // testErroringForRefreshTokenRetrievalFunc returns a refresh token error when // the refresh token is not empty. This is useful for testing behavior when // the refresh token has expired or is otherwise invalid. @@ -303,8 +309,6 @@ func TestCleanAndPickTokens(t *testing.T) { func TestRefresh(t *testing.T) { ctx := context.Background() - s, err := db.Open(ctx) - require.NoError(t, err) boundaryAddr := "address" u := &user{Id: "u1", Address: boundaryAddr} @@ -317,15 +321,18 @@ func TestRefresh(t *testing.T) { boundaryAuthTokens := []*authtokens.AuthToken{at} atMap := make(map[ringToken]*authtokens.AuthToken) - r, err := NewRepository(ctx, s, &sync.Map{}, mapBasedAuthTokenKeyringLookup(atMap), sliceBasedAuthTokenBoundaryReader(boundaryAuthTokens)) - require.NoError(t, err) - rs, err := NewRefreshService(ctx, r) - require.NoError(t, err) atMap[ringToken{"k", "t"}] = at - require.NoError(t, r.AddKeyringToken(ctx, boundaryAddr, KeyringToken{KeyringType: "k", TokenName: "t", AuthTokenId: at.Id})) t.Run("set targets", func(t *testing.T) { + s, err := db.Open(ctx) + require.NoError(t, err) + r, err := NewRepository(ctx, s, &sync.Map{}, mapBasedAuthTokenKeyringLookup(atMap), sliceBasedAuthTokenBoundaryReader(boundaryAuthTokens)) + require.NoError(t, err) + rs, err := NewRefreshService(ctx, r) + require.NoError(t, err) + require.NoError(t, r.AddKeyringToken(ctx, boundaryAddr, KeyringToken{KeyringType: "k", TokenName: "t", AuthTokenId: at.Id})) + retTargets := []*targets.Target{ target("1"), target("2"), @@ -359,6 +366,14 @@ func TestRefresh(t *testing.T) { }) t.Run("set sessions", func(t *testing.T) { + s, err := db.Open(ctx) + require.NoError(t, err) + r, err := NewRepository(ctx, s, &sync.Map{}, mapBasedAuthTokenKeyringLookup(atMap), sliceBasedAuthTokenBoundaryReader(boundaryAuthTokens)) + require.NoError(t, err) + rs, err := NewRefreshService(ctx, r) + require.NoError(t, err) + require.NoError(t, r.AddKeyringToken(ctx, boundaryAddr, KeyringToken{KeyringType: "k", TokenName: "t", AuthTokenId: at.Id})) + retSess := []*sessions.Session{ session("1"), session("2"), @@ -391,8 +406,16 @@ func TestRefresh(t *testing.T) { }) t.Run("error propogates up", func(t *testing.T) { + s, err := db.Open(ctx) + require.NoError(t, err) + r, err := NewRepository(ctx, s, &sync.Map{}, mapBasedAuthTokenKeyringLookup(atMap), sliceBasedAuthTokenBoundaryReader(boundaryAuthTokens)) + require.NoError(t, err) + rs, err := NewRefreshService(ctx, r) + require.NoError(t, err) + require.NoError(t, r.AddKeyringToken(ctx, boundaryAddr, KeyringToken{KeyringType: "k", TokenName: "t", AuthTokenId: at.Id})) + innerErr := errors.New("test error") - err := rs.Refresh(ctx, + err = rs.Refresh(ctx, WithSessionRetrievalFunc(testStaticResourceRetrievalFunc[*sessions.Session](t, nil, nil)), WithTargetRetrievalFunc(func(ctx context.Context, addr, token string, refreshTok RefreshTokenValue) ([]*targets.Target, []string, RefreshTokenValue, error) { require.Equal(t, boundaryAddr, addr) @@ -411,6 +434,14 @@ func TestRefresh(t *testing.T) { }) t.Run("tokens that are no longer in the ring is deleted", func(t *testing.T) { + s, err := db.Open(ctx) + require.NoError(t, err) + r, err := NewRepository(ctx, s, &sync.Map{}, mapBasedAuthTokenKeyringLookup(atMap), sliceBasedAuthTokenBoundaryReader(boundaryAuthTokens)) + require.NoError(t, err) + rs, err := NewRefreshService(ctx, r) + require.NoError(t, err) + require.NoError(t, r.AddKeyringToken(ctx, boundaryAddr, KeyringToken{KeyringType: "k", TokenName: "t", AuthTokenId: at.Id})) + // Remove the token from the keyring, see that we can still see the // token and then user until a Refresh happens which causes them to be // cleaned up. @@ -439,6 +470,178 @@ func TestRefresh(t *testing.T) { }) } +func TestFullFetch(t *testing.T) { + ctx := context.Background() + + boundaryAddr := "address" + u := &user{Id: "u1", Address: boundaryAddr} + at := &authtokens.AuthToken{ + Id: "at_1", + Token: "at_1_token", + UserId: u.Id, + ExpirationTime: time.Now().Add(time.Minute), + } + + boundaryAuthTokens := []*authtokens.AuthToken{at} + atMap := make(map[ringToken]*authtokens.AuthToken) + + atMap[ringToken{"k", "t"}] = at + + t.Run("set targets", func(t *testing.T) { + s, err := db.Open(ctx) + require.NoError(t, err) + r, err := NewRepository(ctx, s, &sync.Map{}, mapBasedAuthTokenKeyringLookup(atMap), sliceBasedAuthTokenBoundaryReader(boundaryAuthTokens)) + require.NoError(t, err) + rs, err := NewRefreshService(ctx, r) + require.NoError(t, err) + require.NoError(t, r.AddKeyringToken(ctx, boundaryAddr, KeyringToken{KeyringType: "k", TokenName: "t", AuthTokenId: at.Id})) + + retTargets := []*targets.Target{ + target("1"), + target("2"), + target("3"), + target("4"), + } + // Since this user doesn't have any resources, the user's data will still + // only get updated with a call to Refresh. + assert.NoError(t, rs.FullFetch(ctx, + WithSessionRetrievalFunc(testFullFetchRetrievalFunc[*sessions.Session](t, nil)), + WithTargetRetrievalFunc(testFullFetchRetrievalFunc[*targets.Target](t, retTargets)))) + + got, err := r.ListTargets(ctx, at.Id) + assert.NoError(t, err) + assert.Empty(t, got) + + assert.NoError(t, rs.Refresh(ctx, + WithSessionRetrievalFunc(testFullFetchRetrievalFunc[*sessions.Session](t, nil)), + WithTargetRetrievalFunc(testFullFetchRetrievalFunc[*targets.Target](t, retTargets[:2])))) + + got, err = r.ListTargets(ctx, at.Id) + assert.NoError(t, err) + assert.ElementsMatch(t, got, retTargets[:2]) + + // now a full fetch will work since the user has resources and no refresh token + assert.NoError(t, rs.FullFetch(ctx, + WithSessionRetrievalFunc(testFullFetchRetrievalFunc[*sessions.Session](t, nil)), + WithTargetRetrievalFunc(testFullFetchRetrievalFunc[*targets.Target](t, retTargets)))) + + got, err = r.ListTargets(ctx, at.Id) + assert.NoError(t, err) + assert.ElementsMatch(t, got, retTargets) + }) + + t.Run("set sessions", func(t *testing.T) { + s, err := db.Open(ctx) + require.NoError(t, err) + r, err := NewRepository(ctx, s, &sync.Map{}, mapBasedAuthTokenKeyringLookup(atMap), sliceBasedAuthTokenBoundaryReader(boundaryAuthTokens)) + require.NoError(t, err) + rs, err := NewRefreshService(ctx, r) + require.NoError(t, err) + require.NoError(t, r.AddKeyringToken(ctx, boundaryAddr, KeyringToken{KeyringType: "k", TokenName: "t", AuthTokenId: at.Id})) + + retSess := []*sessions.Session{ + session("1"), + session("2"), + session("3"), + session("4"), + } + assert.NoError(t, rs.FullFetch(ctx, + WithTargetRetrievalFunc(testFullFetchRetrievalFunc[*targets.Target](t, nil)), + WithSessionRetrievalFunc(testFullFetchRetrievalFunc[*sessions.Session](t, retSess)))) + + got, err := r.ListSessions(ctx, at.Id) + assert.NoError(t, err) + assert.Empty(t, got) + + assert.NoError(t, rs.Refresh(ctx, + WithTargetRetrievalFunc(testFullFetchRetrievalFunc[*targets.Target](t, nil)), + WithSessionRetrievalFunc(testFullFetchRetrievalFunc[*sessions.Session](t, retSess[:2])))) + got, err = r.ListSessions(ctx, at.Id) + assert.NoError(t, err) + assert.ElementsMatch(t, got, retSess[:2]) + + assert.NoError(t, rs.FullFetch(ctx, + WithTargetRetrievalFunc(testFullFetchRetrievalFunc[*targets.Target](t, nil)), + WithSessionRetrievalFunc(testFullFetchRetrievalFunc[*sessions.Session](t, retSess)))) + got, err = r.ListSessions(ctx, at.Id) + assert.NoError(t, err) + assert.ElementsMatch(t, got, retSess) + }) + + t.Run("error propogates up", func(t *testing.T) { + s, err := db.Open(ctx) + require.NoError(t, err) + r, err := NewRepository(ctx, s, &sync.Map{}, mapBasedAuthTokenKeyringLookup(atMap), sliceBasedAuthTokenBoundaryReader(boundaryAuthTokens)) + require.NoError(t, err) + rs, err := NewRefreshService(ctx, r) + require.NoError(t, err) + require.NoError(t, r.AddKeyringToken(ctx, boundaryAddr, KeyringToken{KeyringType: "k", TokenName: "t", AuthTokenId: at.Id})) + + assert.NoError(t, rs.Refresh(ctx, + WithTargetRetrievalFunc(testFullFetchRetrievalFunc[*targets.Target](t, []*targets.Target{target("1")})), + WithSessionRetrievalFunc(testFullFetchRetrievalFunc[*sessions.Session](t, []*sessions.Session{session("1")})))) + + innerErr := errors.New("test error") + err = rs.FullFetch(ctx, + WithSessionRetrievalFunc(testFullFetchRetrievalFunc[*sessions.Session](t, nil)), + WithTargetRetrievalFunc(func(ctx context.Context, addr, token string, refreshTok RefreshTokenValue) ([]*targets.Target, []string, RefreshTokenValue, error) { + require.Equal(t, boundaryAddr, addr) + require.Equal(t, at.Token, token) + return nil, nil, "", innerErr + })) + assert.ErrorContains(t, err, innerErr.Error()) + + err = rs.FullFetch(ctx, + WithTargetRetrievalFunc(testFullFetchRetrievalFunc[*targets.Target](t, nil)), + WithSessionRetrievalFunc(func(ctx context.Context, addr, token string, refreshTok RefreshTokenValue) ([]*sessions.Session, []string, RefreshTokenValue, error) { + require.Equal(t, boundaryAddr, addr) + require.Equal(t, at.Token, token) + return nil, nil, "", innerErr + })) + assert.ErrorContains(t, err, innerErr.Error()) + }) + + t.Run("tokens that are no longer in the ring is deleted", func(t *testing.T) { + s, err := db.Open(ctx) + require.NoError(t, err) + r, err := NewRepository(ctx, s, &sync.Map{}, mapBasedAuthTokenKeyringLookup(atMap), sliceBasedAuthTokenBoundaryReader(boundaryAuthTokens)) + require.NoError(t, err) + rs, err := NewRefreshService(ctx, r) + require.NoError(t, err) + require.NoError(t, r.AddKeyringToken(ctx, boundaryAddr, KeyringToken{KeyringType: "k", TokenName: "t", AuthTokenId: at.Id})) + + assert.NoError(t, rs.Refresh(ctx, + WithTargetRetrievalFunc(testFullFetchRetrievalFunc[*targets.Target](t, []*targets.Target{target("1")})), + WithSessionRetrievalFunc(testFullFetchRetrievalFunc[*sessions.Session](t, []*sessions.Session{session("1")})))) + + // Remove the token from the keyring, see that we can still see the + // token and then user until a Refresh happens which causes them to be + // cleaned up. + delete(atMap, ringToken{"k", "t"}) + + ps, err := r.listTokens(ctx, u) + require.NoError(t, err) + assert.Len(t, ps, 1) + + us, err := r.listUsers(ctx) + require.NoError(t, err) + assert.Len(t, us, 1) + + rs.FullFetch(ctx, + WithSessionRetrievalFunc(testFullFetchRetrievalFunc[*sessions.Session](t, nil)), + WithTargetRetrievalFunc(testFullFetchRetrievalFunc[*targets.Target](t, nil))) + + ps, err = r.listTokens(ctx, u) + require.NoError(t, err) + assert.Empty(t, ps) + + // And since the last token was deleted, the user also was deleted + us, err = r.listUsers(ctx) + require.NoError(t, err) + assert.Empty(t, us) + }) +} + func target(suffix string) *targets.Target { return &targets.Target{ Id: fmt.Sprintf("target_%s", suffix), diff --git a/internal/clientcache/internal/cache/repository_refresh_token.go b/internal/clientcache/internal/cache/repository_refresh_token.go index 0f53ca5ec..e7058e792 100644 --- a/internal/clientcache/internal/cache/repository_refresh_token.go +++ b/internal/clientcache/internal/cache/repository_refresh_token.go @@ -5,6 +5,7 @@ package cache import ( "context" + "database/sql" "github.com/hashicorp/boundary/internal/db" "github.com/hashicorp/boundary/internal/errors" @@ -43,6 +44,22 @@ func (r *Repository) lookupRefreshToken(ctx context.Context, u *user, resourceTy return rt.RefreshToken, nil } +// listRefreshTokens returns all refresh tokens associated with a specific user +func (r *Repository) listRefreshTokens(ctx context.Context, u *user) ([]*refreshToken, error) { + const op = "cache.(Repository).listRefreshTokens" + switch { + case util.IsNil(u): + return nil, errors.New(ctx, errors.InvalidParameter, op, "user is nil") + case u.Id == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "user id is empty") + } + var ret []*refreshToken + if err := r.rw.SearchWhere(ctx, &ret, "user_id = @user_id", []any{sql.Named("user_id", u.Id)}); err != nil { + return nil, errors.Wrap(ctx, err, op) + } + return ret, nil +} + // deleteRefreshToken deletes the refresh token for the provided user and resource type func (r *Repository) deleteRefreshToken(ctx context.Context, u *user, rType resourceType) error { const op = "cache.(Repository).deleteRefreshToken" diff --git a/internal/clientcache/internal/cache/repository_refresh_token_test.go b/internal/clientcache/internal/cache/repository_refresh_token_test.go index bbc729f93..24340d9ea 100644 --- a/internal/clientcache/internal/cache/repository_refresh_token_test.go +++ b/internal/clientcache/internal/cache/repository_refresh_token_test.go @@ -15,6 +15,62 @@ import ( "github.com/stretchr/testify/require" ) +func TestListRefreshToken(t *testing.T) { + ctx := context.Background() + s, err := cachedb.Open(ctx) + require.NoError(t, err) + + r, err := NewRepository(ctx, s, &sync.Map{}, + mapBasedAuthTokenKeyringLookup(map[ringToken]*authtokens.AuthToken{}), + sliceBasedAuthTokenBoundaryReader(nil)) + require.NoError(t, err) + + t.Run("nil user", func(t *testing.T) { + got, err := r.listRefreshTokens(ctx, nil) + assert.ErrorContains(t, err, "user is nil") + assert.Nil(t, got) + }) + t.Run("empty user id", func(t *testing.T) { + got, err := r.listRefreshTokens(ctx, &user{Address: ""}) + assert.ErrorContains(t, err, "user id is empty") + assert.Nil(t, got) + }) + t.Run("empty response", func(t *testing.T) { + got, err := r.listRefreshTokens(ctx, &user{Id: "u123"}) + assert.NoError(t, err) + assert.Empty(t, got) + }) + t.Run("got target response", func(t *testing.T) { + known := &user{Id: "withtargetresponse", Address: "addr"} + require.NoError(t, r.rw.Create(ctx, known)) + + token := RefreshTokenValue("something") + r.rw.DoTx(ctx, 1, db.ExpBackoff{}, func(r db.Reader, w db.Writer) error { + require.NoError(t, upsertRefreshToken(ctx, w, known, targetResourceType, token)) + return nil + }) + + got, err := r.listRefreshTokens(ctx, known) + assert.NoError(t, err) + assert.Len(t, got, 1) + }) + t.Run("got multiple responses", func(t *testing.T) { + known := &user{Id: "with multiple responses", Address: "addr"} + require.NoError(t, r.rw.Create(ctx, known)) + + token := RefreshTokenValue("something") + r.rw.DoTx(ctx, 1, db.ExpBackoff{}, func(r db.Reader, w db.Writer) error { + require.NoError(t, upsertRefreshToken(ctx, w, known, targetResourceType, token)) + require.NoError(t, upsertRefreshToken(ctx, w, known, sessionResourceType, token)) + return nil + }) + + got, err := r.listRefreshTokens(ctx, known) + assert.NoError(t, err) + assert.Len(t, got, 2) + }) +} + func TestLookupRefreshToken(t *testing.T) { ctx := context.Background() s, err := cachedb.Open(ctx) diff --git a/internal/clientcache/internal/cache/repository_sessions.go b/internal/clientcache/internal/cache/repository_sessions.go index a28acb147..3ef6dd82c 100644 --- a/internal/clientcache/internal/cache/repository_sessions.go +++ b/internal/clientcache/internal/cache/repository_sessions.go @@ -43,6 +43,9 @@ func defaultSessionFunc(ctx context.Context, addr, authTok string, refreshTok Re return l.Items, l.RemovedIds, RefreshTokenValue(l.RefreshToken), nil } +// refreshSessions uses attempts to refresh the sessions for the provided user +// using the provided tokens. If available, it uses the refresh tokens in +// storage to retrieve and apply only the delta. func (r *Repository) refreshSessions(ctx context.Context, u *user, tokens map[AuthToken]string, opt ...Option) error { const op = "cache.(Repository).refreshSessions" switch { @@ -101,7 +104,7 @@ func (r *Repository) refreshSessions(ctx context.Context, u *user, tokens map[Au } event.WriteSysEvent(ctx, op, fmt.Sprintf("updating %d sessions for user %v", len(resp), u)) - _, err = r.rw.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(r db.Reader, w db.Writer) error { + _, err = r.rw.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(_ db.Reader, w db.Writer) error { switch { case oldRefreshToken == "": if _, err := w.Exec(ctx, "delete from session where user_id = @user_id", @@ -115,28 +118,9 @@ func (r *Repository) refreshSessions(ctx context.Context, u *user, tokens map[Au } } - for _, s := range resp { - item, err := json.Marshal(s) - if err != nil { - return err - } - newSession := &Session{ - UserId: u.Id, - Id: s.Id, - Type: s.Type, - Status: s.Status, - Endpoint: s.Endpoint, - Item: string(item), - } - onConflict := db.OnConflict{ - Target: db.Columns{"user_id", "id"}, - Action: db.UpdateAll(true), - } - if err := w.Create(ctx, newSession, db.WithOnConflict(&onConflict)); err != nil { - return err - } + if err := upsertSessions(ctx, w, u, resp); err != nil { + return err } - if err := upsertRefreshToken(ctx, w, u, resourceType, newRefreshToken); err != nil { return err } @@ -148,13 +132,117 @@ func (r *Repository) refreshSessions(ctx context.Context, u *user, tokens map[Au return nil } +// fullFetchSessions fetches all sessions for the provided user and sets the +// cache to match the values returned. If the response includes a refresh +// token it will save that as well. +func (r *Repository) fullFetchSessions(ctx context.Context, u *user, tokens map[AuthToken]string, opt ...Option) error { + const op = "cache.(Repository).fullFetchSessions" + switch { + case util.IsNil(u): + return errors.New(ctx, errors.InvalidParameter, op, "user is nil") + case u.Id == "": + return errors.New(ctx, errors.InvalidParameter, op, "user id is missing") + case u.Address == "": + return errors.New(ctx, errors.InvalidParameter, op, "user boundary address is missing") + } + const resourceType = sessionResourceType + + opts, err := getOpts(opt...) + if err != nil { + return errors.Wrap(ctx, err, op) + } + if opts.withSessionRetrievalFunc == nil { + opts.withSessionRetrievalFunc = defaultSessionFunc + } + + // Find and use a token for retrieving sessions + var gotResponse bool + var resp []*sessions.Session + var newRefreshToken RefreshTokenValue + var retErr error + for at, t := range tokens { + resp, _, newRefreshToken, err = opts.withSessionRetrievalFunc(ctx, u.Address, t, "") + if err != nil { + retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op, errors.WithMsg("for token %q", at.Id))) + continue + } + gotResponse = true + break + } + + if retErr != nil { + if saveErr := r.SaveError(ctx, u, resourceType, retErr); saveErr != nil { + return stderrors.Join(err, errors.Wrap(ctx, saveErr, op)) + } + } + if !gotResponse { + return retErr + } + + event.WriteSysEvent(ctx, op, fmt.Sprintf("updating %d sessions for user %v", len(resp), u)) + _, err = r.rw.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(_ db.Reader, w db.Writer) error { + if _, err := w.Exec(ctx, "delete from session where user_id = @user_id", + []any{sql.Named("user_id", u.Id)}); err != nil { + return err + } + + if err := upsertSessions(ctx, w, u, resp); err != nil { + return err + } + if err := upsertRefreshToken(ctx, w, u, resourceType, newRefreshToken); err != nil { + return err + } + return nil + }) + if err != nil { + return errors.Wrap(ctx, err, op) + } + return nil +} + +// upsertSessions upserts the provided sessions to be stored for the provided user. +func upsertSessions(ctx context.Context, w db.Writer, u *user, in []*sessions.Session) error { + const op = "cache.upsertSessions" + switch { + case util.IsNil(w): + return errors.New(ctx, errors.InvalidParameter, op, "writer is nil") + case !w.IsTx(ctx): + return errors.New(ctx, errors.InvalidParameter, op, "writer isn't in a transaction") + case util.IsNil(u): + return errors.New(ctx, errors.InvalidParameter, op, "user is nil") + } + + for _, s := range in { + item, err := json.Marshal(s) + if err != nil { + return errors.Wrap(ctx, err, op) + } + newSession := &Session{ + UserId: u.Id, + Id: s.Id, + Type: s.Type, + Status: s.Status, + Endpoint: s.Endpoint, + Item: string(item), + } + onConflict := db.OnConflict{ + Target: db.Columns{"user_id", "id"}, + Action: db.UpdateAll(true), + } + if err := w.Create(ctx, newSession, db.WithOnConflict(&onConflict)); err != nil { + return errors.Wrap(ctx, err, op) + } + } + return nil +} + func (r *Repository) ListSessions(ctx context.Context, authTokenId string) ([]*sessions.Session, error) { const op = "cache.(Repository).ListSessions" switch { case authTokenId == "": return nil, errors.New(ctx, errors.InvalidParameter, op, "auth token id is missing") } - ret, err := r.searchSessions(ctx, authTokenId, "true", nil) + ret, err := r.searchSessions(ctx, "true", nil, withAuthTokenId(authTokenId)) if err != nil { return nil, errors.Wrap(ctx, err, op) } @@ -174,26 +262,39 @@ func (r *Repository) QuerySessions(ctx context.Context, authTokenId, query strin if err != nil { return nil, errors.Wrap(ctx, err, op, errors.WithCode(errors.InvalidParameter)) } - ret, err := r.searchSessions(ctx, authTokenId, w.Condition, w.Args) + ret, err := r.searchSessions(ctx, w.Condition, w.Args, withAuthTokenId(authTokenId)) if err != nil { return nil, errors.Wrap(ctx, err, op) } return ret, nil } -func (r *Repository) searchSessions(ctx context.Context, authTokenId, condition string, searchArgs []any) ([]*sessions.Session, error) { +func (r *Repository) searchSessions(ctx context.Context, condition string, searchArgs []any, opt ...Option) ([]*sessions.Session, error) { const op = "cache.(Repository).searchSessions" switch { - case authTokenId == "": - return nil, errors.New(ctx, errors.InvalidParameter, op, "auth token id is missing") case condition == "": return nil, errors.New(ctx, errors.InvalidParameter, op, "condition is missing") } - condition = fmt.Sprintf("%s and user_id in (select user_id from auth_token where id = ?)", condition) - args := append(searchArgs, authTokenId) + opts, err := getOpts(opt...) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + switch { + case opts.withAuthTokenId != "" && opts.withUserId != "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "both user id and auth token id were provided") + case opts.withAuthTokenId == "" && opts.withUserId == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "neither user id nor auth token id were provided") + case opts.withAuthTokenId != "": + condition = fmt.Sprintf("%s and user_id in (select user_id from auth_token where id = ?)", condition) + searchArgs = append(searchArgs, opts.withAuthTokenId) + case opts.withUserId != "": + condition = fmt.Sprintf("%s and user_id = ?", condition) + searchArgs = append(searchArgs, opts.withUserId) + } + var cachedSessions []*Session - if err := r.rw.SearchWhere(ctx, &cachedSessions, condition, args, db.WithLimit(-1)); err != nil { + if err := r.rw.SearchWhere(ctx, &cachedSessions, condition, searchArgs, db.WithLimit(-1)); err != nil { return nil, errors.Wrap(ctx, err, op) } diff --git a/internal/clientcache/internal/cache/repository_sessions_test.go b/internal/clientcache/internal/cache/repository_sessions_test.go index 369d07ee5..e558c1515 100644 --- a/internal/clientcache/internal/cache/repository_sessions_test.go +++ b/internal/clientcache/internal/cache/repository_sessions_test.go @@ -426,6 +426,7 @@ func TestDefaultSessionRetrievalFunc(t *testing.T) { tc.Client().SetToken(tc.Token().Token) tarClient := targets.NewClient(tc.Client()) _ = worker.NewTestWorker(t, &worker.TestWorkerOpts{ + Name: "test", InitialUpstreams: tc.ClusterAddrs(), WorkerAuthKms: tc.Config().WorkerAuthKms, }) @@ -433,8 +434,10 @@ func TestDefaultSessionRetrievalFunc(t *testing.T) { tar1, err := tarClient.Create(tc.Context(), "tcp", "p_1234567890", targets.WithName("tar1"), targets.WithTcpTargetDefaultPort(1), targets.WithAddress("address")) require.NoError(t, err) require.NotNil(t, tar1) + + require.NoError(t, tc.WaitForNextWorkerStatusUpdate("test")) _, err = tarClient.AuthorizeSession(tc.Context(), tar1.Item.Id) - assert.NoError(t, err) + require.NoError(t, err) got, removed, refTok, err := defaultSessionFunc(tc.Context(), tc.ApiAddrs()[0], tc.Token().Token, "") assert.NoError(t, err) diff --git a/internal/clientcache/internal/cache/repository_targets.go b/internal/clientcache/internal/cache/repository_targets.go index 26c4077d8..614d21dae 100644 --- a/internal/clientcache/internal/cache/repository_targets.go +++ b/internal/clientcache/internal/cache/repository_targets.go @@ -43,6 +43,9 @@ func defaultTargetFunc(ctx context.Context, addr, authTok string, refreshTok Ref return l.Items, l.RemovedIds, RefreshTokenValue(l.RefreshToken), nil } +// refreshTargets uses attempts to refresh the targets for the provided user +// using the provided tokens. If available, it uses the refresh tokens in +// storage to retrieve and apply only the delta. func (r *Repository) refreshTargets(ctx context.Context, u *user, tokens map[AuthToken]string, opt ...Option) error { const op = "cache.(Repository).refreshTargets" switch { @@ -112,28 +115,9 @@ func (r *Repository) refreshTargets(ctx context.Context, u *user, tokens map[Aut } } - for _, t := range resp { - item, err := json.Marshal(t) - if err != nil { - return err - } - newTarget := &Target{ - UserId: u.Id, - Id: t.Id, - Name: t.Name, - Description: t.Description, - Address: t.Address, - Item: string(item), - } - onConflict := db.OnConflict{ - Target: db.Columns{"user_id", "id"}, - Action: db.UpdateAll(true), - } - if err := w.Create(ctx, newTarget, db.WithOnConflict(&onConflict)); err != nil { - return err - } + if err := upsertTargets(ctx, w, u, resp); err != nil { + return err } - if newRefreshToken != "" || oldRefreshToken != "" { if err := upsertRefreshToken(ctx, w, u, resourceType, newRefreshToken); err != nil { return err @@ -147,13 +131,116 @@ func (r *Repository) refreshTargets(ctx context.Context, u *user, tokens map[Aut return nil } +// fullFetchTargets fetches all targets for the provided user and sets the +// cache to match the values returned. If the response includes a refresh +// token it will save that as well. +func (r *Repository) fullFetchTargets(ctx context.Context, u *user, tokens map[AuthToken]string, opt ...Option) error { + const op = "cache.(Repository).fullFetchTargets" + switch { + case util.IsNil(u): + return errors.New(ctx, errors.InvalidParameter, op, "user is nil") + case u.Id == "": + return errors.New(ctx, errors.InvalidParameter, op, "user id is missing") + } + const resourceType = targetResourceType + + opts, err := getOpts(opt...) + if err != nil { + return errors.Wrap(ctx, err, op) + } + if opts.withTargetRetrievalFunc == nil { + opts.withTargetRetrievalFunc = defaultTargetFunc + } + + // Find and use a token for retrieving targets + var gotResponse bool + var resp []*targets.Target + var newRefreshToken RefreshTokenValue + var retErr error + for at, t := range tokens { + resp, _, newRefreshToken, err = opts.withTargetRetrievalFunc(ctx, u.Address, t, "") + if err != nil { + retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op, errors.WithMsg("for token %q", at.Id))) + continue + } + gotResponse = true + break + } + if retErr != nil { + if saveErr := r.SaveError(ctx, u, resourceType, retErr); saveErr != nil { + return stderrors.Join(err, errors.Wrap(ctx, saveErr, op)) + } + } + if !gotResponse { + return retErr + } + + event.WriteSysEvent(ctx, op, fmt.Sprintf("updating %d targets for user %v", len(resp), u)) + _, err = r.rw.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(r db.Reader, w db.Writer) error { + if _, err := w.Exec(ctx, "delete from target where user_id = @user_id", + []any{sql.Named("user_id", u.Id)}); err != nil { + return err + } + + if err := upsertTargets(ctx, w, u, resp); err != nil { + return err + } + if newRefreshToken != "" { + if err := upsertRefreshToken(ctx, w, u, resourceType, newRefreshToken); err != nil { + return err + } + } + return nil + }) + if err != nil { + return errors.Wrap(ctx, err, op) + } + return nil +} + +// upsertTargets upserts the provided targets to be stored for the provided user. +func upsertTargets(ctx context.Context, w db.Writer, u *user, in []*targets.Target) error { + const op = "cache.upsertTargets" + switch { + case util.IsNil(w): + return errors.New(ctx, errors.InvalidParameter, op, "writer is nil") + case !w.IsTx(ctx): + return errors.New(ctx, errors.InvalidParameter, op, "writer isn't in a transaction") + case util.IsNil(u): + return errors.New(ctx, errors.InvalidParameter, op, "user is nil") + } + + for _, t := range in { + item, err := json.Marshal(t) + if err != nil { + return errors.Wrap(ctx, err, op) + } + newTarget := &Target{ + UserId: u.Id, + Id: t.Id, + Name: t.Name, + Description: t.Description, + Address: t.Address, + Item: string(item), + } + onConflict := db.OnConflict{ + Target: db.Columns{"user_id", "id"}, + Action: db.UpdateAll(true), + } + if err := w.Create(ctx, newTarget, db.WithOnConflict(&onConflict)); err != nil { + return errors.Wrap(ctx, err, op) + } + } + return nil +} + func (r *Repository) ListTargets(ctx context.Context, authTokenId string) ([]*targets.Target, error) { const op = "cache.(Repository).ListTargets" switch { case authTokenId == "": return nil, errors.New(ctx, errors.InvalidParameter, op, "auth token id is missing") } - ret, err := r.searchTargets(ctx, authTokenId, "true", nil) + ret, err := r.searchTargets(ctx, "true", nil, withAuthTokenId(authTokenId)) if err != nil { return nil, errors.Wrap(ctx, err, op) } @@ -173,26 +260,39 @@ func (r *Repository) QueryTargets(ctx context.Context, authTokenId, query string if err != nil { return nil, errors.Wrap(ctx, err, op, errors.WithCode(errors.InvalidParameter)) } - ret, err := r.searchTargets(ctx, authTokenId, w.Condition, w.Args) + ret, err := r.searchTargets(ctx, w.Condition, w.Args, withAuthTokenId(authTokenId)) if err != nil { return nil, errors.Wrap(ctx, err, op) } return ret, nil } -func (r *Repository) searchTargets(ctx context.Context, authTokenId, condition string, searchArgs []any) ([]*targets.Target, error) { +func (r *Repository) searchTargets(ctx context.Context, condition string, searchArgs []any, opt ...Option) ([]*targets.Target, error) { const op = "cache.(Repository).searchTargets" switch { - case authTokenId == "": - return nil, errors.New(ctx, errors.InvalidParameter, op, "user id is missing") case condition == "": return nil, errors.New(ctx, errors.InvalidParameter, op, "condition is missing") } - condition = fmt.Sprintf("%s and user_id in (select user_id from auth_token where id = ?)", condition) - args := append(searchArgs, authTokenId) + opts, err := getOpts(opt...) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + switch { + case opts.withAuthTokenId != "" && opts.withUserId != "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "both user id and auth token id were provided") + case opts.withAuthTokenId == "" && opts.withUserId == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "neither user id nor auth token id were provided") + case opts.withAuthTokenId != "": + condition = fmt.Sprintf("%s and user_id in (select user_id from auth_token where id = ?)", condition) + searchArgs = append(searchArgs, opts.withAuthTokenId) + case opts.withUserId != "": + condition = fmt.Sprintf("%s and user_id = ?", condition) + searchArgs = append(searchArgs, opts.withUserId) + } + var cachedTargets []*Target - if err := r.rw.SearchWhere(ctx, &cachedTargets, condition, args, db.WithLimit(-1)); err != nil { + if err := r.rw.SearchWhere(ctx, &cachedTargets, condition, searchArgs, db.WithLimit(-1)); err != nil { return nil, errors.Wrap(ctx, err, op) } diff --git a/internal/clientcache/internal/cache/repository_targets_test.go b/internal/clientcache/internal/cache/repository_targets_test.go index a6a87b92b..4eaf85466 100644 --- a/internal/clientcache/internal/cache/repository_targets_test.go +++ b/internal/clientcache/internal/cache/repository_targets_test.go @@ -432,5 +432,6 @@ func TestDefaultTargetRetrievalFunc(t *testing.T) { assert.NoError(t, err) assert.NotEmpty(t, refTok2) assert.Empty(t, removed2) + // Note: This will sometimes fail until PR 3897 is merged assert.Empty(t, got2) } diff --git a/internal/clientcache/internal/daemon/options.go b/internal/clientcache/internal/daemon/options.go index 5abeee384..2f667ddeb 100644 --- a/internal/clientcache/internal/daemon/options.go +++ b/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 { diff --git a/internal/clientcache/internal/daemon/options_test.go b/internal/clientcache/internal/daemon/options_test.go index 16c21b5a7..49e3c2b25 100644 --- a/internal/clientcache/internal/daemon/options_test.go +++ b/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 diff --git a/internal/clientcache/internal/daemon/server.go b/internal/clientcache/internal/daemon/server.go index dd09f51f8..866710e06 100644 --- a/internal/clientcache/internal/daemon/server.go +++ b/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() diff --git a/internal/clientcache/internal/daemon/testing.go b/internal/clientcache/internal/daemon/testing.go index e873838f2..822c74b2d 100644 --- a/internal/clientcache/internal/daemon/testing.go +++ b/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) diff --git a/internal/clientcache/internal/daemon/ticker.go b/internal/clientcache/internal/daemon/ticker.go index caff76d80..94637febe 100644 --- a/internal/clientcache/internal/daemon/ticker.go +++ b/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) +} diff --git a/internal/clientcache/internal/daemon/ticker_test.go b/internal/clientcache/internal/daemon/ticker_test.go index e389ce2f3..cf688b385 100644 --- a/internal/clientcache/internal/daemon/ticker_test.go +++ b/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 +}