diff --git a/internal/clientcache/cmd/daemon/start.go b/internal/clientcache/cmd/daemon/start.go index f6ff5f844..117c890ea 100644 --- a/internal/clientcache/cmd/daemon/start.go +++ b/internal/clientcache/cmd/daemon/start.go @@ -50,7 +50,7 @@ type StartCommand struct { *base.Command flagRefreshInterval time.Duration - flagFullFetchInterval time.Duration + flagRecheckSupportInterval time.Duration flagMaxSearchStaleness time.Duration flagMaxSearchRefreshTimeout time.Duration flagDatabaseUrl string @@ -104,25 +104,26 @@ func (c *StartCommand) Flags() *base.FlagSets { f.DurationVar(&base.DurationVar{ Name: "refresh-interval", Target: &c.flagRefreshInterval, - Usage: `Specifies the interval between refresh token supported cache refreshes. Default: 1 minute`, + Usage: `Specifies the interval between refresh token supported cache refreshes.`, Default: daemon.DefaultRefreshInterval, }) f.DurationVar(&base.DurationVar{ - Name: "full-fetch-interval", - Target: &c.flagFullFetchInterval, - Usage: `Specifies the interval between full cache refresh for boundary instances which do not support refresh tokens. Default: 5 minutes`, - Default: daemon.DefaultFullFetchInterval, + Name: "recheck-support-interval", + Target: &c.flagRecheckSupportInterval, + Usage: `Specifies the interval between checking if a boundary instances is supported when it previously was not.`, + Default: daemon.DefaultRecheckSupportInterval, + Hidden: true, }) f.DurationVar(&base.DurationVar{ Name: "max-search-staleness", Target: &c.flagMaxSearchStaleness, - Usage: `Specifies the duration of time that can pass since the resource was last updated before performing a search waits for the resources being refreshed first. Default: 1 minute`, + Usage: `Specifies the duration of time that can pass since the resource was last updated before performing a search waits for the resources being refreshed first.`, Default: daemon.DefaultSearchStaleness, }) f.DurationVar(&base.DurationVar{ Name: "max-search-refresh-timeout", Target: &c.flagMaxSearchRefreshTimeout, - Usage: `If a search request triggers a best effort refresh, this specifies how long the refresh should run before timing out. Default: 1 second`, + Usage: `If a search request triggers a best effort refresh, this specifies how long the refresh should run before timing out.`, Default: daemon.DefaultSearchRefreshTimeout, }) f.BoolVar(&base.BoolVar{ @@ -196,16 +197,16 @@ func (c *StartCommand) Run(args []string) int { writers = append(writers, lf) cfg := &daemon.Config{ - ContextCancel: cancel, - RefreshInterval: c.flagRefreshInterval, - FullFetchInterval: c.flagFullFetchInterval, - MaxSearchStaleness: c.flagMaxSearchStaleness, - DatabaseUrl: c.flagDatabaseUrl, - StoreDebug: c.flagStoreDebug, - LogLevel: c.flagLogLevel, - LogFormat: c.flagLogFormat, - LogWriter: io.MultiWriter(writers...), - DotDirectory: dotDir, + ContextCancel: cancel, + RefreshInterval: c.flagRefreshInterval, + RecheckSupportInterval: c.flagRecheckSupportInterval, + MaxSearchStaleness: c.flagMaxSearchStaleness, + DatabaseUrl: c.flagDatabaseUrl, + StoreDebug: c.flagStoreDebug, + LogLevel: c.flagLogLevel, + LogFormat: c.flagLogFormat, + LogWriter: io.MultiWriter(writers...), + DotDirectory: dotDir, } srv, err := daemon.New(ctx, cfg) @@ -307,6 +308,10 @@ func (c *StartCommand) makeBackground(ctx context.Context, dotDir string) (bool, env := os.Environ() env = append(env, fmt.Sprintf("%s=%s", backgroundEnvName, backgroundEnvVal)) args := []string{"daemon", "start"} + args = append(args, "-refresh-interval", c.flagRefreshInterval.String()) + args = append(args, "-max-search-staleness", c.flagMaxSearchStaleness.String()) + args = append(args, "-max-search-refresh-timeout", c.flagMaxSearchRefreshTimeout.String()) + args = append(args, "-recheck-support-interval", c.flagRecheckSupportInterval.String()) if c.flagLogLevel != "" { args = append(args, "-log-level", c.flagLogLevel) } @@ -316,18 +321,6 @@ func (c *StartCommand) makeBackground(ctx context.Context, dotDir string) (bool, if c.flagDatabaseUrl != "" { args = append(args, "-database-url", c.flagDatabaseUrl) } - if c.flagRefreshInterval > 0 && c.flagRefreshInterval != daemon.DefaultRefreshInterval { - args = append(args, "-refresh-interval", c.flagRefreshInterval.String()) - } - if c.flagFullFetchInterval > 0 && c.flagFullFetchInterval != daemon.DefaultFullFetchInterval { - args = append(args, "-full-fetch-interval", c.flagFullFetchInterval.String()) - } - if c.flagMaxSearchStaleness > 0 && c.flagMaxSearchStaleness != daemon.DefaultSearchStaleness { - args = append(args, "-max-search-staleness", c.flagMaxSearchStaleness.String()) - } - if c.flagMaxSearchRefreshTimeout > 0 && c.flagMaxSearchRefreshTimeout != daemon.DefaultSearchRefreshTimeout { - args = append(args, "-max-search-refresh-timeout", c.flagMaxSearchRefreshTimeout.String()) - } if c.flagStoreDebug { args = append(args, "-store-debug") } diff --git a/internal/clientcache/cmd/daemon/status.go b/internal/clientcache/cmd/daemon/status.go index c02bf0a63..349a68110 100644 --- a/internal/clientcache/cmd/daemon/status.go +++ b/internal/clientcache/cmd/daemon/status.go @@ -168,8 +168,12 @@ func printUsersTable(us []daemon.UserStatus) []string { ret = append(ret, " User:") nonAttributeMap := map[string]any{ "Id": u.Id, - "Address": u.Address, + "Address": u.BoundaryInstance.Address, "AuthToken Count": len(u.AuthTokens), + "Search Support": u.BoundaryInstance.CacheSupport, + } + if u.BoundaryInstance.LastSupportCheck > 0 { + nonAttributeMap["Since Search Support Check"] = u.BoundaryInstance.LastSupportCheck.Round(time.Second) } maxLength := base.MaxAttributesLength(nonAttributeMap, nil, nil) ret = append(ret, @@ -196,7 +200,7 @@ func printUsersTable(us []daemon.UserStatus) []string { } if r.RefreshToken != nil { nonAttributeMap["Since Last Refresh"] = r.RefreshToken.LastUsed.Round(time.Second) - nonAttributeMap["Since Full Fetch"] = r.RefreshToken.Age.Round(time.Second) + nonAttributeMap["Since Initial Fetch"] = r.RefreshToken.Age.Round(time.Second) } if r.LastError != nil { nonAttributeMap["Since Last Error"] = r.LastError.LastReturned.Round(time.Second) diff --git a/internal/clientcache/cmd/search/search_test.go b/internal/clientcache/cmd/search/search_test.go index d30d16031..2536bdc1c 100644 --- a/internal/clientcache/cmd/search/search_test.go +++ b/internal/clientcache/cmd/search/search_test.go @@ -22,7 +22,7 @@ import ( type testCommander struct { t *testing.T - at *authtokens.AuthToken + at map[string]*authtokens.AuthToken } func (r *testCommander) Client(opt ...base.Option) (*api.Client, error) { @@ -32,7 +32,7 @@ func (r *testCommander) Client(opt ...base.Option) (*api.Client, error) { } func (r *testCommander) ReadTokenFromKeyring(k, a string) *authtokens.AuthToken { - return r.at + return r.at[a] } func TestSearch(t *testing.T) { @@ -43,10 +43,19 @@ func TestSearch(t *testing.T) { Token: "at_1_token", ExpirationTime: time.Now().Add(time.Minute), } - cmd := &testCommander{t: t, at: at} + unsupportedAt := &authtokens.AuthToken{ + Id: "at_2", + UserId: "user_2", + Token: "at_2_token", + ExpirationTime: time.Now().Add(time.Minute), + } + cmd := &testCommander{t: t, at: map[string]*authtokens.AuthToken{"tokenname": at, "unsupported": unsupportedAt}} boundaryTokenReaderFn := func(ctx context.Context, addr, authToken string) (*authtokens.AuthToken, error) { - if authToken == at.Token { + switch authToken { + case at.Token: return at, nil + case unsupportedAt.Token: + return unsupportedAt, nil } return nil, errors.New("test not found error") } @@ -61,6 +70,7 @@ func TestSearch(t *testing.T) { // Give the store some time to get initialized time.Sleep(100 * time.Millisecond) srv.AddKeyringToken(t, "address", "keyringtype", "tokenname", at.Id, boundaryTokenReaderFn) + srv.AddKeyringToken(t, "address", "keyringtype", "unsupported", unsupportedAt.Id, boundaryTokenReaderFn) errorCases := []struct { name string @@ -142,7 +152,21 @@ func TestSearch(t *testing.T) { assert.EqualValues(t, r, &daemon.SearchResult{}) }) - srv.AddResources(t, cmd.at, []*targets.Target{ + t.Run("unsupported boundary instance", func(t *testing.T) { + srv.AddUnsupportedCachingData(t, unsupportedAt, boundaryTokenReaderFn) + resp, r, apiErr, err := search(ctx, srv.BaseDotDir(), filterBy{ + authTokenId: unsupportedAt.Id, + resource: "targets", + }) + assert.NoError(t, err) + require.NotNil(t, apiErr) + assert.NotNil(t, resp) + assert.Nil(t, r) + + assert.Contains(t, apiErr.Message, "doesn't support search") + }) + + srv.AddResources(t, at, []*targets.Target{ {Id: "ttcp_1234567890", Name: "name1", Description: "description1"}, {Id: "ttcp_0987654321", Name: "name2", Description: "description2"}, }, []*sessions.Session{ @@ -151,6 +175,17 @@ func TestSearch(t *testing.T) { }, boundaryTokenReaderFn) t.Run("target response from list", func(t *testing.T) { + resp, r, apiErr, err := search(ctx, srv.BaseDotDir(), filterBy{ + authTokenId: at.Id, + resource: "targets", + }) + require.NoError(t, err) + assert.Nil(t, apiErr) + assert.NotNil(t, resp) + assert.NotNil(t, r) + assert.Len(t, r.Targets, 2) + }) + t.Run("search for unsupported controller", func(t *testing.T) { resp, r, apiErr, err := search(ctx, srv.BaseDotDir(), filterBy{ authTokenId: at.Id, resource: "targets", diff --git a/internal/clientcache/internal/cache/refresh.go b/internal/clientcache/internal/cache/refresh.go index 0a6b2b0c3..e19f087a9 100644 --- a/internal/clientcache/internal/cache/refresh.go +++ b/internal/clientcache/internal/cache/refresh.go @@ -6,6 +6,7 @@ package cache import ( "context" stderrors "errors" + "fmt" "time" "github.com/hashicorp/boundary/api" @@ -114,32 +115,13 @@ func (r *RefreshService) refreshableUsers(ctx context.Context, in []*user) ([]*u const op = "cache.(RefreshService).refreshableUsers" var ret []*user for _, u := range in { - rt, err := r.repo.listRefreshTokens(ctx, u) + cs, err := r.repo.cacheSupportState(ctx, u) if err != nil { return nil, errors.Wrap(ctx, err, op) } - if len(rt) > 0 { + if cs.supported != NotSupportedCacheSupport { 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 } @@ -267,24 +249,24 @@ func (r *RefreshService) Refresh(ctx context.Context, opt ...Option) error { } if err := r.repo.refreshTargets(ctx, u, tokens, opt...); err != nil { - retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op)) + retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("for user id %s", u.Id)))) } if err := r.repo.refreshSessions(ctx, u, tokens, opt...); err != nil { - retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op)) + retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("for user id %s", u.Id)))) } } 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" +// RecheckCachingSupport is used for users of boundary instances which do not +// support refresh tokens. For these users a call to RecheckCachingSupport +// retrieves resources for that user to see if the controller now returns a +// refresh token. Users who do not have any resources nor refresh tokens still +// are Refreshed. If caching moves from unsupported to supported this method +// will cache the results automatically. +func (r *RefreshService) RecheckCachingSupport(ctx context.Context, opt ...Option) error { + const op = "cache.(RefreshService).RecheckCachingSupport" if err := r.repo.cleanExpiredOrOrphanedAuthTokens(ctx); err != nil { return errors.Wrap(ctx, err, op) } @@ -318,11 +300,19 @@ func (r *RefreshService) FullFetch(ctx context.Context, opt ...Option) error { 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.checkCachingTargets(ctx, u, tokens, opt...); err != nil { + if err == ErrRefreshNotSupported { + // This is expected so no need to propogate the error up + continue + } + retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("for user id %s", u.Id)))) } - if err := r.repo.fullFetchSessions(ctx, u, tokens, opt...); err != nil { - retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op)) + if err := r.repo.checkCachingSessions(ctx, u, tokens, opt...); err != nil { + if err == ErrRefreshNotSupported { + // This is expected so no need to propogate the error up + continue + } + retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("for user id %s", u.Id)))) } } diff --git a/internal/clientcache/internal/cache/refresh_test.go b/internal/clientcache/internal/cache/refresh_test.go index 871bb8b2c..44e4c242d 100644 --- a/internal/clientcache/internal/cache/refresh_test.go +++ b/internal/clientcache/internal/cache/refresh_test.go @@ -52,11 +52,11 @@ func testStaticResourceRetrievalFunc[T any](t *testing.T, ret [][]T, removed [][ } } -// testFullFetchRetrievalFunc simulates a controller that doesn't support refresh +// testNoRefreshRetrievalFunc simulates a controller that doesn't support refresh // since it does not return any refresh token. -func testFullFetchRetrievalFunc[T any](t *testing.T, ret []T) func(context.Context, string, string, RefreshTokenValue) ([]T, []string, RefreshTokenValue, error) { +func testNoRefreshRetrievalFunc[T any](t *testing.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 + return nil, nil, "", ErrRefreshNotSupported } } @@ -460,45 +460,52 @@ func TestRefreshForSearch(t *testing.T) { require.NoError(t, err) r, err := NewRepository(ctx, s, &sync.Map{}, mapBasedAuthTokenKeyringLookup(atMap), sliceBasedAuthTokenBoundaryReader(boundaryAuthTokens)) require.NoError(t, err) - rs, err := NewRefreshService(ctx, r, time.Millisecond, 0) + rs, err := NewRefreshService(ctx, r, 0, 0) 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"), } // Get the first set of resources, but no refresh tokens - assert.NoError(t, rs.Refresh(ctx, - WithSessionRetrievalFunc(testFullFetchRetrievalFunc[*sessions.Session](t, nil)), - WithTargetRetrievalFunc(testFullFetchRetrievalFunc[*targets.Target](t, retTargets[:2])))) + err = rs.Refresh(ctx, + WithSessionRetrievalFunc(testNoRefreshRetrievalFunc[*sessions.Session](t)), + WithTargetRetrievalFunc(testNoRefreshRetrievalFunc[*targets.Target](t))) + assert.ErrorContains(t, err, ErrRefreshNotSupported.Error()) got, err := r.ListTargets(ctx, at.Id) assert.NoError(t, err) - assert.ElementsMatch(t, retTargets[:2], got) + assert.Empty(t, got) - // Let 2 milliseconds pass so the items are stale enough - time.Sleep(2 * time.Millisecond) + // Now that we know that this user doesn't support refresh tokens, they + // wont be refreshed any more, and we wont see the error when refreshing + // any more. + err = rs.Refresh(ctx, + WithSessionRetrievalFunc(testNoRefreshRetrievalFunc[*sessions.Session](t)), + WithTargetRetrievalFunc(testNoRefreshRetrievalFunc[*targets.Target](t))) + assert.Nil(t, err) + + err = rs.RecheckCachingSupport(ctx, + WithSessionRetrievalFunc(testNoRefreshRetrievalFunc[*sessions.Session](t)), + WithTargetRetrievalFunc(testNoRefreshRetrievalFunc[*targets.Target](t))) + assert.Nil(t, err) - assert.NoError(t, rs.RefreshForSearch(ctx, at.Id, Targets, - WithSessionRetrievalFunc(testFullFetchRetrievalFunc[*sessions.Session](t, nil)), - WithTargetRetrievalFunc(testFullFetchRetrievalFunc[*targets.Target](t, retTargets)))) got, err = r.ListTargets(ctx, at.Id) assert.NoError(t, err) - // still no change in targets since they weren't refreshed with that search - // due to having no refresh token. - assert.ElementsMatch(t, retTargets[:2], got) + assert.Empty(t, got) + + // Now simulate the controller updating to support refresh tokens and + // the resources starting to be cached. + err = rs.RecheckCachingSupport(ctx, + WithSessionRetrievalFunc(testStaticResourceRetrievalFunc[*sessions.Session](t, nil, nil)), + WithTargetRetrievalFunc(testStaticResourceRetrievalFunc[*targets.Target](t, [][]*targets.Target{retTargets}, [][]string{{}}))) + assert.Nil(t, err, err) - // Only a full fetch will get all the resources - 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) + assert.Len(t, got, 2) }) t.Run("sessions refreshed for searching", func(t *testing.T) { @@ -766,7 +773,7 @@ func TestRefresh(t *testing.T) { }) } -func TestFullFetch(t *testing.T) { +func TestRecheckCachingSupport(t *testing.T) { ctx := context.Background() boundaryAddr := "address" @@ -783,7 +790,7 @@ func TestFullFetch(t *testing.T) { atMap[ringToken{"k", "t"}] = at - t.Run("set targets", func(t *testing.T) { + t.Run("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)) @@ -792,41 +799,32 @@ func TestFullFetch(t *testing.T) { 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)))) + assert.NoError(t, rs.RecheckCachingSupport(ctx, + WithSessionRetrievalFunc(testNoRefreshRetrievalFunc[*sessions.Session](t)), + WithTargetRetrievalFunc(testNoRefreshRetrievalFunc[*targets.Target](t)))) 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])))) + err = rs.Refresh(ctx, + WithSessionRetrievalFunc(testNoRefreshRetrievalFunc[*sessions.Session](t)), + WithTargetRetrievalFunc(testNoRefreshRetrievalFunc[*targets.Target](t))) + assert.ErrorIs(t, err, ErrRefreshNotSupported) got, err = r.ListTargets(ctx, at.Id) assert.NoError(t, err) - assert.ElementsMatch(t, got, retTargets[:2]) + assert.Empty(t, got) // 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) + assert.NoError(t, rs.RecheckCachingSupport(ctx, + WithSessionRetrievalFunc(testNoRefreshRetrievalFunc[*sessions.Session](t)), + WithTargetRetrievalFunc(testNoRefreshRetrievalFunc[*targets.Target](t)))) }) - t.Run("set sessions", func(t *testing.T) { + t.Run("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)) @@ -835,33 +833,29 @@ func TestFullFetch(t *testing.T) { 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)))) + assert.NoError(t, rs.RecheckCachingSupport(ctx, + WithTargetRetrievalFunc(testNoRefreshRetrievalFunc[*targets.Target](t)), + WithSessionRetrievalFunc(testNoRefreshRetrievalFunc[*sessions.Session](t)))) 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]) + err = rs.Refresh(ctx, + WithTargetRetrievalFunc(testNoRefreshRetrievalFunc[*targets.Target](t)), + WithSessionRetrievalFunc(testNoRefreshRetrievalFunc[*sessions.Session](t))) + assert.ErrorIs(t, err, ErrRefreshNotSupported) - 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) + assert.Empty(t, got) + + assert.NoError(t, rs.RecheckCachingSupport(ctx, + WithTargetRetrievalFunc(testNoRefreshRetrievalFunc[*targets.Target](t)), + WithSessionRetrievalFunc(testNoRefreshRetrievalFunc[*sessions.Session](t)))) + got, err = r.ListSessions(ctx, at.Id) + assert.NoError(t, err) + assert.Empty(t, got) }) t.Run("error propogates up", func(t *testing.T) { @@ -873,13 +867,14 @@ func TestFullFetch(t *testing.T) { 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")})))) + err = rs.Refresh(ctx, + WithTargetRetrievalFunc(testNoRefreshRetrievalFunc[*targets.Target](t)), + WithSessionRetrievalFunc(testNoRefreshRetrievalFunc[*sessions.Session](t))) + assert.ErrorIs(t, err, ErrRefreshNotSupported) innerErr := errors.New("test error") - err = rs.FullFetch(ctx, - WithSessionRetrievalFunc(testFullFetchRetrievalFunc[*sessions.Session](t, nil)), + err = rs.RecheckCachingSupport(ctx, + WithSessionRetrievalFunc(testNoRefreshRetrievalFunc[*sessions.Session](t)), 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) @@ -887,9 +882,9 @@ func TestFullFetch(t *testing.T) { })) 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) { + err = rs.RecheckCachingSupport(ctx, + WithSessionRetrievalFunc(testNoRefreshRetrievalFunc[*sessions.Session](t)), + 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 @@ -906,9 +901,10 @@ func TestFullFetch(t *testing.T) { 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")})))) + err = rs.Refresh(ctx, + WithTargetRetrievalFunc(testNoRefreshRetrievalFunc[*targets.Target](t)), + WithSessionRetrievalFunc(testNoRefreshRetrievalFunc[*sessions.Session](t))) + assert.ErrorIs(t, err, ErrRefreshNotSupported) // 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 @@ -923,9 +919,10 @@ func TestFullFetch(t *testing.T) { require.NoError(t, err) assert.Len(t, us, 1) - rs.FullFetch(ctx, - WithSessionRetrievalFunc(testFullFetchRetrievalFunc[*sessions.Session](t, nil)), - WithTargetRetrievalFunc(testFullFetchRetrievalFunc[*targets.Target](t, nil))) + err = rs.RecheckCachingSupport(ctx, + WithSessionRetrievalFunc(testNoRefreshRetrievalFunc[*sessions.Session](t)), + WithTargetRetrievalFunc(testNoRefreshRetrievalFunc[*targets.Target](t))) + assert.NoError(t, err) ps, err = r.listTokens(ctx, u) require.NoError(t, err) diff --git a/internal/clientcache/internal/cache/repository_refresh_token.go b/internal/clientcache/internal/cache/repository_refresh_token.go index f44b0a90b..b80ce9b74 100644 --- a/internal/clientcache/internal/cache/repository_refresh_token.go +++ b/internal/clientcache/internal/cache/repository_refresh_token.go @@ -6,6 +6,7 @@ package cache import ( "context" "database/sql" + stderrors "errors" "time" "github.com/hashicorp/boundary/internal/db" @@ -13,10 +14,70 @@ import ( "github.com/hashicorp/boundary/internal/util" ) +// ErrRefreshNotSupported is returned whenever it is determined a boundary +// instance does not support refresh tokens. +var ErrRefreshNotSupported = stderrors.New("refresh tokens are not supported for this controller") + // RefreshTokenValue is the the type for the actual refresh token value handled // by the client cache. type RefreshTokenValue string +// sentinelNoRefreshToken is the value set in the refresh token table when +// Boundary returns resources but no refresh tokens. The presence of this value +// indicates that the boundary instance queried does not support refresh tokens. +const sentinelNoRefreshToken RefreshTokenValue = "__no_refresh_token_supported__" + +// CacheSupport is an enum that identifies if a boundary instance can be supported +// by the client daemon cache. +type CacheSupport string + +const ( + UnknownCacheSupport CacheSupport = "Unknown" + SupportedCacheSupport CacheSupport = "Supported" + NotSupportedCacheSupport CacheSupport = "Not Supported" +) + +// cacheSupportCheckedState contains the state of a cache support check. +type cacheSupportCheckedState struct { + supported CacheSupport + lastChecked *time.Time +} + +// cacheSupportState returns the cacheSupportCheckedState for the provided user +// which says if caching is supported for that user and when the last time it +// was checked. +func (r *Repository) cacheSupportState(ctx context.Context, u *user) (*cacheSupportCheckedState, error) { + const op = "cache.(Repository).cacheSupportState" + 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) + } + if len(ret) == 0 { + return &cacheSupportCheckedState{supported: UnknownCacheSupport}, nil + } + + lastChecked := time.Time{} + supported := UnknownCacheSupport + for _, rt := range ret { + if rt.UpdateTime.After(lastChecked) { + lastChecked = rt.UpdateTime + } + switch { + case supported == NotSupportedCacheSupport || rt.RefreshToken == sentinelNoRefreshToken: + supported = NotSupportedCacheSupport + default: + supported = SupportedCacheSupport + } + } + return &cacheSupportCheckedState{supported: supported, lastChecked: &lastChecked}, nil +} + // lookupRefreshToken returns the last known valid refresh token or an empty // string if one is unkonwn. No error is returned if no valid refresh token is // found. @@ -54,7 +115,8 @@ func (r *Repository) listRefreshTokens(ctx context.Context, u *user) ([]*refresh 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 { + if err := r.rw.SearchWhere(ctx, &ret, "user_id = @user_id and refresh_token != @sentinel", + []any{sql.Named("user_id", u.Id), sql.Named("sentinel", sentinelNoRefreshToken)}); err != nil { return nil, errors.Wrap(ctx, err, op) } return ret, nil @@ -72,22 +134,40 @@ func (r *Repository) deleteRefreshToken(ctx context.Context, u *user, rType reso return errors.New(ctx, errors.InvalidParameter, op, "resource type is invalid") } - _, err := r.rw.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(r db.Reader, w db.Writer) error { - rt := &refreshToken{ - UserId: u.Id, - ResourceType: rType, - } - n, err := w.Delete(ctx, rt) - if err != nil { - return errors.Wrap(ctx, err, op) - } - if n > 1 { - return errors.New(ctx, errors.MultipleRecords, op, "attempted to delete a single resource but multiple were resource deletions were attempted") - } - return nil + _, err := r.rw.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(_ db.Reader, w db.Writer) error { + return deleteRefreshToken(ctx, w, u, rType) }) if err != nil { - return err + return errors.Wrap(ctx, err, op) + } + return nil +} + +// deleteRefreshToken deletes the refresh token for the provided user and resource type +func deleteRefreshToken(ctx context.Context, w db.Writer, u *user, rType resourceType) error { + const op = "cache.deleteRefreshToken" + 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") + case u.Id == "": + return errors.New(ctx, errors.InvalidParameter, op, "user id is empty") + case !rType.valid(): + return errors.New(ctx, errors.InvalidParameter, op, "resource type is invalid") + } + rt := &refreshToken{ + UserId: u.Id, + ResourceType: rType, + } + n, err := w.Delete(ctx, rt) + if err != nil { + return errors.Wrap(ctx, err, op) + } + if n > 1 { + return errors.New(ctx, errors.MultipleRecords, op, "attempted to delete a single resource but multiple were resource deletions were attempted") } return nil diff --git a/internal/clientcache/internal/cache/repository_refresh_token_test.go b/internal/clientcache/internal/cache/repository_refresh_token_test.go index bf740cb66..c84331677 100644 --- a/internal/clientcache/internal/cache/repository_refresh_token_test.go +++ b/internal/clientcache/internal/cache/repository_refresh_token_test.go @@ -55,6 +55,20 @@ func TestListRefreshToken(t *testing.T) { assert.NoError(t, err) assert.Len(t, got, 1) }) + t.Run("sentinel values", func(t *testing.T) { + known := &user{Id: "sentinelValues", Address: "addr"} + require.NoError(t, r.rw.Create(ctx, known)) + + token := sentinelNoRefreshToken + 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.Empty(t, got) + }) 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)) @@ -72,6 +86,76 @@ func TestListRefreshToken(t *testing.T) { }) } +func TestCacheSupportState(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("unknown", func(t *testing.T) { + known := &user{Id: "unknown", Address: "addr"} + require.NoError(t, r.rw.Create(ctx, known)) + + got, err := r.listRefreshTokens(ctx, known) + assert.NoError(t, err) + assert.Empty(t, got) + + cs, err := r.cacheSupportState(ctx, known) + assert.NoError(t, err) + assert.Equal(t, cs.supported, UnknownCacheSupport) + assert.Nil(t, cs.lastChecked) + }) + + t.Run("unsupported", func(t *testing.T) { + known := &user{Id: "unsupported", Address: "addr"} + require.NoError(t, r.rw.Create(ctx, known)) + + now := time.Now().Truncate(time.Millisecond) + token := sentinelNoRefreshToken + 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.Empty(t, got) + + cs, err := r.cacheSupportState(ctx, known) + assert.NoError(t, err) + assert.Equal(t, cs.supported, NotSupportedCacheSupport) + assert.NotNil(t, cs.lastChecked) + assert.LessOrEqual(t, now, *cs.lastChecked) + }) + + t.Run("Supported", func(t *testing.T) { + known := &user{Id: "supported", Address: "addr"} + require.NoError(t, r.rw.Create(ctx, known)) + + now := time.Now().Truncate(time.Millisecond) + 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) + + cs, err := r.cacheSupportState(ctx, known) + assert.NoError(t, err) + assert.Equal(t, cs.supported, SupportedCacheSupport) + assert.NotNil(t, cs.lastChecked) + assert.LessOrEqual(t, now, *cs.lastChecked) + }) +} + 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 b2f348a88..04fbb411d 100644 --- a/internal/clientcache/internal/cache/repository_sessions.go +++ b/internal/clientcache/internal/cache/repository_sessions.go @@ -40,6 +40,9 @@ func defaultSessionFunc(ctx context.Context, addr, authTok string, refreshTok Re } return nil, nil, "", errors.Wrap(ctx, err, op) } + if l.ResponseType == "" { + return nil, nil, "", ErrRefreshNotSupported + } return l.Items, l.RemovedIds, RefreshTokenValue(l.ListToken), nil } @@ -78,6 +81,7 @@ func (r *Repository) refreshSessions(ctx context.Context, u *user, tokens map[Au var gotResponse bool var resp []*sessions.Session var newRefreshToken RefreshTokenValue + var unsupportedCacheRequest bool var removedIds []string var retErr error for at, t := range tokens { @@ -91,8 +95,12 @@ func (r *Repository) refreshSessions(ctx context.Context, u *user, tokens map[Au resp, removedIds, 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 + if err == ErrRefreshNotSupported { + unsupportedCacheRequest = true + } else { + retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op, errors.WithMsg("for token %q", at.Id))) + continue + } } gotResponse = true break @@ -121,26 +129,37 @@ func (r *Repository) refreshSessions(ctx context.Context, u *user, tokens map[Au 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 + switch { + case unsupportedCacheRequest: + if err := upsertRefreshToken(ctx, w, u, resourceType, sentinelNoRefreshToken); err != nil { + return err + } + case newRefreshToken != "": + if err := upsertSessions(ctx, w, u, resp); err != nil { + return err + } + if err := upsertRefreshToken(ctx, w, u, resourceType, newRefreshToken); err != nil { + return err + } + default: + // controller supports caching, but doesn't have any resources } return nil }) if err != nil { return errors.Wrap(ctx, err, op) } + if unsupportedCacheRequest { + return ErrRefreshNotSupported + } return nil } -// fullFetchSessions fetches all sessions for the provided user and sets the +// checkCachingSessions 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" +func (r *Repository) checkCachingSessions(ctx context.Context, u *user, tokens map[AuthToken]string, opt ...Option) error { + const op = "cache.(Repository).checkSessionsForSearchability" switch { case util.IsNil(u): return errors.New(ctx, errors.InvalidParameter, op, "user is nil") @@ -163,12 +182,17 @@ func (r *Repository) fullFetchSessions(ctx context.Context, u *user, tokens map[ var gotResponse bool var resp []*sessions.Session var newRefreshToken RefreshTokenValue + var unsupportedCacheRequest bool 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 + if err == ErrRefreshNotSupported { + unsupportedCacheRequest = true + } else { + retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op, errors.WithMsg("for token %q", at.Id))) + continue + } } gotResponse = true break @@ -183,24 +207,38 @@ func (r *Repository) fullFetchSessions(ctx context.Context, u *user, tokens map[ 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 fk_user_id = @fk_user_id", - []any{sql.Named("fk_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 + _, err = r.rw.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(reader db.Reader, w db.Writer) error { + switch { + case unsupportedCacheRequest: + if err := upsertRefreshToken(ctx, w, u, resourceType, sentinelNoRefreshToken); err != nil { + return err + } + case newRefreshToken != "": + if _, err := w.Exec(ctx, "delete from session where fk_user_id = @fk_user_id", + []any{sql.Named("fk_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 + } + default: + // This is no longer flagged as not supported, but we dont have a + // refresh token so clear out any refresh token we have stored. + if err := deleteRefreshToken(ctx, w, u, resourceType); err != nil { + return err + } } return nil }) if err != nil { return errors.Wrap(ctx, err, op) } + if unsupportedCacheRequest { + return ErrRefreshNotSupported + } return nil } diff --git a/internal/clientcache/internal/cache/repository_targets.go b/internal/clientcache/internal/cache/repository_targets.go index 5cad4e59b..e042f5d53 100644 --- a/internal/clientcache/internal/cache/repository_targets.go +++ b/internal/clientcache/internal/cache/repository_targets.go @@ -40,6 +40,9 @@ func defaultTargetFunc(ctx context.Context, addr, authTok string, refreshTok Ref } return nil, nil, "", errors.Wrap(ctx, err, op) } + if l.ResponseType == "" { + return nil, nil, "", ErrRefreshNotSupported + } return l.Items, l.RemovedIds, RefreshTokenValue(l.ListToken), nil } @@ -77,21 +80,26 @@ func (r *Repository) refreshTargets(ctx context.Context, u *user, tokens map[Aut var gotResponse bool var resp []*targets.Target var removedIds []string - var newRefreshTokenVal RefreshTokenValue + var newRefreshToken RefreshTokenValue + var unsupportedCacheRequest bool var retErr error for at, t := range tokens { - resp, removedIds, newRefreshTokenVal, err = opts.withTargetRetrievalFunc(ctx, u.Address, t, oldRefreshTokenVal) + resp, removedIds, newRefreshToken, err = opts.withTargetRetrievalFunc(ctx, u.Address, t, oldRefreshTokenVal) if api.ErrInvalidListToken.Is(err) { if err := r.deleteRefreshToken(ctx, u, resourceType); err != nil { return errors.Wrap(ctx, err, op) } // try again without the refresh token oldRefreshToken = nil - resp, removedIds, newRefreshTokenVal, err = opts.withTargetRetrievalFunc(ctx, u.Address, t, "") + resp, removedIds, 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 + if err == ErrRefreshNotSupported { + unsupportedCacheRequest = true + } else { + retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op, errors.WithMsg("for token %q", at.Id))) + continue + } } gotResponse = true break @@ -106,9 +114,9 @@ func (r *Repository) refreshTargets(ctx context.Context, u *user, tokens map[Aut } 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 { + _, err = r.rw.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(_ db.Reader, w db.Writer) error { switch { - case oldRefreshToken == nil: + case oldRefreshToken == nil || unsupportedCacheRequest: if _, err := w.Exec(ctx, "delete from target where fk_user_id = @fk_user_id", []any{sql.Named("fk_user_id", u.Id)}); err != nil { return err @@ -119,28 +127,40 @@ func (r *Repository) refreshTargets(ctx context.Context, u *user, tokens map[Aut return err } } - - if err := upsertTargets(ctx, w, u, resp); err != nil { - return err - } - if newRefreshTokenVal != "" || oldRefreshToken != nil { - if err := upsertRefreshToken(ctx, w, u, resourceType, newRefreshTokenVal); err != nil { + switch { + case unsupportedCacheRequest: + if err := upsertRefreshToken(ctx, w, u, resourceType, sentinelNoRefreshToken); err != nil { return err } + case newRefreshToken != "": + if err := upsertTargets(ctx, w, u, resp); err != nil { + return err + } + if err := upsertRefreshToken(ctx, w, u, resourceType, newRefreshToken); err != nil { + return err + } + default: + // controller supports caching, but doesn't have any resources } return nil }) if err != nil { return errors.Wrap(ctx, err, op) } + if unsupportedCacheRequest { + return ErrRefreshNotSupported + } 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" +// checkCachingTargets fetches all targets for the provided user. If the +// response has at least one target and a refresh token, it makes the targets +// cachable and stores the refresh token. If there is no refresh token in the +// response it marks this user as unable to cache the data. If no data and no +// refresh token is stored it is unknown if the targets are cachable, the user +// is not marked as unknown. +func (r *Repository) checkCachingTargets(ctx context.Context, u *user, tokens map[AuthToken]string, opt ...Option) error { + const op = "cache.(Repository).checkCachingTargets" switch { case util.IsNil(u): return errors.New(ctx, errors.InvalidParameter, op, "user is nil") @@ -161,12 +181,17 @@ func (r *Repository) fullFetchTargets(ctx context.Context, u *user, tokens map[A var gotResponse bool var resp []*targets.Target var newRefreshToken RefreshTokenValue + var unsupportedCacheRequest bool 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 + if err == ErrRefreshNotSupported { + unsupportedCacheRequest = true + } else { + retErr = stderrors.Join(retErr, errors.Wrap(ctx, err, op, errors.WithMsg("for token %q", at.Id))) + continue + } } gotResponse = true break @@ -180,26 +205,42 @@ func (r *Repository) fullFetchTargets(ctx context.Context, u *user, tokens map[A 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 fk_user_id = @fk_user_id", - []any{sql.Named("fk_user_id", u.Id)}); err != nil { - return err - } - - if err := upsertTargets(ctx, w, u, resp); err != nil { - return err - } - if newRefreshToken != "" { + _, err = r.rw.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(reader db.Reader, w db.Writer) error { + switch { + case unsupportedCacheRequest: + // Since we know the controller doesn't support caching, we mark the + // user as unable to cache the data. + if err := upsertRefreshToken(ctx, w, u, resourceType, sentinelNoRefreshToken); err != nil { + return err + } + case newRefreshToken != "": + // Now that there is a refresh token, the data can be cached, so + // cache it and store the refresh token for future refreshes. + if _, err := w.Exec(ctx, "delete from target where fk_user_id = @fk_user_id", + []any{sql.Named("fk_user_id", u.Id)}); err != nil { + return err + } + if err := upsertTargets(ctx, w, u, resp); err != nil { + return err + } if err := upsertRefreshToken(ctx, w, u, resourceType, newRefreshToken); err != nil { return err } + default: + // We know the controller supports caching, but doesn't have a + // refresh token so clear out any refresh token we have for this resource. + if err := deleteRefreshToken(ctx, w, u, resourceType); err != nil { + return err + } } return nil }) if err != nil { return errors.Wrap(ctx, err, op) } + if unsupportedCacheRequest { + return ErrRefreshNotSupported + } return nil } diff --git a/internal/clientcache/internal/cache/search.go b/internal/clientcache/internal/cache/search.go index 3d8bf47eb..b96ae931a 100644 --- a/internal/clientcache/internal/cache/search.go +++ b/internal/clientcache/internal/cache/search.go @@ -63,24 +63,17 @@ type SearchResult struct { // cache. type SearchService struct { searchableResources map[SearchableResource]resourceSearcher + repo *Repository } -// QueryAndLister defines the methods needed for listing an querying the -// resources supported by the boundary frontend. -type QueryAndLister interface { - QueryTargets(context.Context, string, string) ([]*targets.Target, error) - ListTargets(context.Context, string) ([]*targets.Target, error) - QuerySessions(context.Context, string, string) ([]*sessions.Session, error) - ListSessions(context.Context, string) ([]*sessions.Session, error) -} - -func NewSearchService(ctx context.Context, repo QueryAndLister) (*SearchService, error) { +func NewSearchService(ctx context.Context, repo *Repository) (*SearchService, error) { const op = "cache.NewSearchService" switch { case util.IsNil(repo): return nil, errors.New(ctx, errors.InvalidParameter, op, "repo is nil") } return &SearchService{ + repo: repo, searchableResources: map[SearchableResource]resourceSearcher{ Targets: &resourceSearchFns[*targets.Target]{ list: repo.ListTargets, @@ -100,6 +93,30 @@ func NewSearchService(ctx context.Context, repo QueryAndLister) (*SearchService, }, nil } +// Supported returns true if the provided search is supported for the provided +// auth token. +func (s *SearchService) Supported(ctx context.Context, t *AuthToken) (bool, error) { + const op = "cache.(SearchService).Supported" + switch { + case util.IsNil(t): + return false, errors.New(ctx, errors.InvalidParameter, op, "auth token is nil") + case t.UserId == "": + return false, errors.New(ctx, errors.InvalidParameter, op, "auth token's user id is empty") + } + u, err := s.repo.lookupUser(ctx, t.UserId) + if err != nil { + return false, errors.Wrap(ctx, err, op) + } + if u == nil { + return false, errors.New(ctx, errors.NotFound, op, "user not found for auth token") + } + cs, err := s.repo.cacheSupportState(ctx, u) + if err != nil { + return false, errors.Wrap(ctx, err, op) + } + return cs.supported != NotSupportedCacheSupport, nil +} + // Search returns a SearchResult based on the provided SearchParams. If the // SearchParams doesn't have a valid searchable resource or an auth token id // an error is returned. If the auth token id is unrecognized or is associated diff --git a/internal/clientcache/internal/cache/search_test.go b/internal/clientcache/internal/cache/search_test.go index d4a83963c..34343d4ab 100644 --- a/internal/clientcache/internal/cache/search_test.go +++ b/internal/clientcache/internal/cache/search_test.go @@ -7,6 +7,7 @@ import ( "context" "sync" "testing" + "time" "github.com/hashicorp/boundary/api/sessions" "github.com/hashicorp/boundary/api/targets" @@ -103,6 +104,90 @@ func TestSearch_Errors(t *testing.T) { } } +func TestSupported(t *testing.T) { + ctx := context.Background() + s, err := cachedb.Open(ctx) + require.NoError(t, err) + + r, err := NewRepository(ctx, s, &sync.Map{}, + mapBasedAuthTokenKeyringLookup(nil), + sliceBasedAuthTokenBoundaryReader(nil)) + assert.NoError(t, err) + + ss, err := NewSearchService(ctx, r) + require.NoError(t, err) + require.NotNil(t, ss) + + t.Run("supported", func(t *testing.T) { + at := &AuthToken{ + Id: "supported", + UserId: "supported", + } + { + rw := db.New(s) + u := &user{Id: at.UserId, Address: "address"} + require.NoError(t, rw.Create(ctx, u)) + require.NoError(t, rw.Create(ctx, at)) + + refTok := &refreshToken{ + UserId: at.Id, + ResourceType: targetResourceType, + RefreshToken: "sometherefreshtoken", + UpdateTime: time.Now(), + CreateTime: time.Now(), + } + require.NoError(t, rw.Create(ctx, refTok)) + } + got, err := ss.Supported(ctx, at) + assert.NoError(t, err) + assert.True(t, got) + }) + + t.Run("unsupported", func(t *testing.T) { + at := &AuthToken{ + Id: "unsupported", + UserId: "unsupported", + } + { + rw := db.New(s) + u := &user{Id: at.UserId, Address: "address"} + require.NoError(t, rw.Create(ctx, u)) + require.NoError(t, rw.Create(ctx, at)) + + refTok := &refreshToken{ + UserId: at.Id, + ResourceType: targetResourceType, + RefreshToken: sentinelNoRefreshToken, + UpdateTime: time.Now(), + CreateTime: time.Now(), + } + require.NoError(t, rw.Create(ctx, refTok)) + } + got, err := ss.Supported(ctx, at) + assert.NoError(t, err) + assert.False(t, got) + }) + + t.Run("unkonwn", func(t *testing.T) { + at := &AuthToken{ + Id: "unknown", + UserId: "unknown", + } + { + rw := db.New(s) + u := &user{Id: at.UserId, Address: "address"} + require.NoError(t, rw.Create(ctx, u)) + require.NoError(t, rw.Create(ctx, at)) + } + got, err := ss.Supported(ctx, at) + assert.NoError(t, err) + // No refresh token means we don't know if the boundary instance + // supports refresh tokens. Since we likely don't have any resources + // to search for anyways, default to saying it is supported. + assert.True(t, got) + }) +} + func TestSearch(t *testing.T) { ctx := context.Background() s, err := cachedb.Open(ctx) diff --git a/internal/clientcache/internal/cache/status.go b/internal/clientcache/internal/cache/status.go index 880233f5c..b42383219 100644 --- a/internal/clientcache/internal/cache/status.go +++ b/internal/clientcache/internal/cache/status.go @@ -41,12 +41,18 @@ type AuthTokenStatus struct { KeyringlessReferences int } +type BoundaryStatus struct { + // The boundary address for this user + Address string + CachingSupported CacheSupport + LastSupportCheck time.Duration +} + // UserStatus contains the status of a specific user tracked by the cache type UserStatus struct { // The Id of the user this status is for - Id string - // The boundary address for this user - Address string + Id string + BoundaryStatus BoundaryStatus // The auth tokens used by this user to authenticate with the boundary instance AuthTokens []AuthTokenStatus // The resources tracked by the cache for this user @@ -84,29 +90,39 @@ func (s *StatusService) Status(ctx context.Context) (*Status, error) { } for _, u := range users { us := UserStatus{ - Id: u.Id, - Address: u.Address, + Id: u.Id, } - { - toks, err := s.repo.listTokens(ctx, u) + cacheSupported, err := s.repo.cacheSupportState(ctx, u) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + + us.BoundaryStatus = BoundaryStatus{ + Address: u.Address, + CachingSupported: cacheSupported.supported, + } + if cacheSupported.lastChecked != nil && !cacheSupported.lastChecked.IsZero() { + us.BoundaryStatus.LastSupportCheck = time.Since(*cacheSupported.lastChecked) + } + + toks, err := s.repo.listTokens(ctx, u) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + for _, t := range toks { + ts := &AuthTokenStatus{ + Id: t.Id, + } + kt, err := s.repo.listKeyringTokens(ctx, t) if err != nil { return nil, errors.Wrap(ctx, err, op) } - for _, t := range toks { - ts := &AuthTokenStatus{ - Id: t.Id, - } - kt, err := s.repo.listKeyringTokens(ctx, t) - if err != nil { - return nil, errors.Wrap(ctx, err, op) - } - ts.KeyringReferences = len(kt) - if _, ok := s.repo.idToKeyringlessAuthToken.Load(t.Id); ok { - ts.KeyringlessReferences = 1 - } - us.AuthTokens = append(us.AuthTokens, *ts) + ts.KeyringReferences = len(kt) + if _, ok := s.repo.idToKeyringlessAuthToken.Load(t.Id); ok { + ts.KeyringlessReferences = 1 } + us.AuthTokens = append(us.AuthTokens, *ts) } for _, rt := range []resourceType{targetResourceType, sessionResourceType} { @@ -213,6 +229,9 @@ func (s *StatusService) refreshTokenStatus(ctx context.Context, u *user, rt reso } return nil, errors.Wrap(ctx, err, op) } + if refTok.RefreshToken == sentinelNoRefreshToken { + return nil, nil + } ret := &RefreshTokenStatus{ LastUsed: time.Since(refTok.UpdateTime), diff --git a/internal/clientcache/internal/cache/status_test.go b/internal/clientcache/internal/cache/status_test.go index be88e9e7c..cf5039fe1 100644 --- a/internal/clientcache/internal/cache/status_test.go +++ b/internal/clientcache/internal/cache/status_test.go @@ -103,8 +103,11 @@ func TestStatus(t *testing.T) { assert.Equal(t, &Status{ Users: []UserStatus{ { - Id: u1.Id, - Address: u1.Address, + Id: u1.Id, + BoundaryStatus: BoundaryStatus{ + Address: u1.Address, + CachingSupported: UnknownCacheSupport, + }, AuthTokens: []AuthTokenStatus{ { Id: at1a.Id, @@ -129,8 +132,11 @@ func TestStatus(t *testing.T) { }, }, { - Id: u2.Id, - Address: u2.Address, + Id: u2.Id, + BoundaryStatus: BoundaryStatus{ + Address: u2.Address, + CachingSupported: UnknownCacheSupport, + }, AuthTokens: []AuthTokenStatus{ { Id: at2.Id, @@ -252,3 +258,72 @@ func Map[T, U any](in []T, f func(T) U) []U { } return ret } + +func TestStatus_unsupported(t *testing.T) { + ctx := context.Background() + s, err := cachedb.Open(ctx) + require.NoError(t, err) + + boundaryAddr := "address" + u1 := &user{Id: "u1", Address: boundaryAddr} + at1 := &authtokens.AuthToken{ + Id: "at_1a", + Token: "at_1a_token", + UserId: u1.Id, + ExpirationTime: time.Now().Add(time.Minute), + } + + boundaryAuthTokens := []*authtokens.AuthToken{at1} + atMap := map[ringToken]*authtokens.AuthToken{ + {k: "default", t: "default"}: at1, + } + r, err := NewRepository(ctx, s, &sync.Map{}, mapBasedAuthTokenKeyringLookup(atMap), sliceBasedAuthTokenBoundaryReader(boundaryAuthTokens)) + require.NoError(t, err) + + ss, err := NewStatusService(ctx, r) + require.NoError(t, err) + + require.NoError(t, r.AddKeyringToken(ctx, boundaryAddr, KeyringToken{ + KeyringType: "default", + TokenName: "default", + AuthTokenId: at1.Id, + })) + + err = r.refreshTargets(ctx, u1, map[AuthToken]string{{Id: "id"}: "something"}, + WithTargetRetrievalFunc(testNoRefreshRetrievalFunc[*targets.Target](t))) + require.ErrorIs(t, err, ErrRefreshNotSupported) + + got, err := ss.Status(ctx) + assert.NoError(t, err) + + assert.Len(t, got.Users, 1) + assert.Greater(t, got.Users[0].BoundaryStatus.LastSupportCheck, time.Duration(0)) + assert.LessOrEqual(t, got.Users[0].BoundaryStatus.LastSupportCheck, time.Second) + got.Users[0].BoundaryStatus.LastSupportCheck = 0 + + assert.Equal(t, got.Users, []UserStatus{ + { + Id: u1.Id, + BoundaryStatus: BoundaryStatus{ + Address: u1.Address, + CachingSupported: NotSupportedCacheSupport, + }, + AuthTokens: []AuthTokenStatus{ + { + Id: at1.Id, + KeyringReferences: 1, + }, + }, + Resources: []ResourceStatus{ + { + Name: string(targetResourceType), + Count: 0, + }, + { + Name: string(sessionResourceType), + Count: 0, + }, + }, + }, + }) +} diff --git a/internal/clientcache/internal/daemon/options.go b/internal/clientcache/internal/daemon/options.go index 2f667ddeb..f8bc67b3b 100644 --- a/internal/clientcache/internal/daemon/options.go +++ b/internal/clientcache/internal/daemon/options.go @@ -14,7 +14,7 @@ import ( type options struct { withDebug bool withRefreshInterval time.Duration - withFullFetchInterval time.Duration + withRecheckSupportInterval time.Duration testWithIntervalRandomizationFactor float64 testWithIntervalRandomizationFactorSet bool withBoundaryTokenReaderFunc cache.BoundaryTokenReaderFn @@ -57,13 +57,13 @@ func withRefreshInterval(_ context.Context, d time.Duration) Option { } } -// withFullFetchInterval provides an optional full fetch interval. -func withFullFetchInterval(_ context.Context, d time.Duration) Option { +// withRecheckSupportInterval provides an optional full fetch interval. +func withRecheckSupportInterval(_ 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) + return fmt.Errorf("provided recheck support interval %q must be positive", d) } - o.withFullFetchInterval = d + o.withRecheckSupportInterval = d return nil } } diff --git a/internal/clientcache/internal/daemon/options_test.go b/internal/clientcache/internal/daemon/options_test.go index 49e3c2b25..4abe13b26 100644 --- a/internal/clientcache/internal/daemon/options_test.go +++ b/internal/clientcache/internal/daemon/options_test.go @@ -44,15 +44,15 @@ func Test_GetOpts(t *testing.T) { testOpts := getDefaultOptions() assert.Equal(t, opts, testOpts) }) - t.Run("withFullFetchInterval", func(t *testing.T) { - opts, err := getOpts(withFullFetchInterval(ctx, time.Second)) + t.Run("withRecheckSupportInterval", func(t *testing.T) { + opts, err := getOpts(withRecheckSupportInterval(ctx, time.Second)) require.NoError(t, err) testOpts := getDefaultOptions() - testOpts.withFullFetchInterval = time.Second + testOpts.withRecheckSupportInterval = time.Second assert.Equal(t, opts, testOpts) }) - t.Run("withFullFetchInterval_negative", func(t *testing.T) { - opts, err := getOpts(withFullFetchInterval(ctx, -time.Second)) + t.Run("withRecheckSupportInterval_negative", func(t *testing.T) { + opts, err := getOpts(withRecheckSupportInterval(ctx, -time.Second)) require.ErrorContains(t, err, "must be positive") testOpts := getDefaultOptions() assert.Equal(t, opts, testOpts) diff --git a/internal/clientcache/internal/daemon/search_handler.go b/internal/clientcache/internal/daemon/search_handler.go index 70b451675..f4727a779 100644 --- a/internal/clientcache/internal/daemon/search_handler.go +++ b/internal/clientcache/internal/daemon/search_handler.go @@ -10,6 +10,7 @@ import ( "net/http" "strconv" + "github.com/hashicorp/boundary/api" "github.com/hashicorp/boundary/api/sessions" "github.com/hashicorp/boundary/api/targets" "github.com/hashicorp/boundary/internal/clientcache/internal/cache" @@ -70,10 +71,22 @@ func newSearchHandlerFunc(ctx context.Context, repo *cache.Repository, refreshSe return } + supported, err := s.Supported(ctx, t) + if err != nil { + event.WriteError(ctx, op, err, event.WithInfoMsg("when checking if search is supported for the provided auth token", "auth_token_id", authTokenId)) + writeError(w, err.Error(), http.StatusInternalServerError) + return + } + if !supported { + writeUnsupportedError(w) + return + } + var opts []cache.Option if b, err := strconv.ParseBool(r.URL.Query().Get(forceRefreshKey)); err == nil && b { opts = append(opts, cache.WithIgnoreSearchStaleness(true)) } + // Refresh the resources for the provided user, if possible. This is best // effort, so if there is any problem refreshing, we just log the error // and move on to handling the search request. @@ -123,3 +136,17 @@ func toApiResult(sr *cache.SearchResult) *SearchResult { Sessions: sr.Sessions, } } + +var errSearchNotSupported = &api.Error{ + Kind: "Unsupported Search Request", + Message: "The request is for a Boundary instance that doesn't support search. The Boundary instance must support refresh tokens for search to be supported.", +} + +func writeUnsupportedError(w http.ResponseWriter) { + b, err := json.Marshal(errSearchNotSupported) + if err != nil { + http.Error(w, fmt.Sprintf("Unable to marshal error %v into api error format: %s", errSearchNotSupported, err.Error()), http.StatusInternalServerError) + return + } + http.Error(w, string(b), http.StatusBadRequest) +} diff --git a/internal/clientcache/internal/daemon/server.go b/internal/clientcache/internal/daemon/server.go index 292d0f6af..ef1bb94f7 100644 --- a/internal/clientcache/internal/daemon/server.go +++ b/internal/clientcache/internal/daemon/server.go @@ -69,15 +69,15 @@ type CacheServer struct { } type Config struct { - ContextCancel context.CancelFunc - RefreshInterval time.Duration - FullFetchInterval time.Duration - DatabaseUrl string - StoreDebug bool - LogLevel string - LogFormat string - LogWriter io.Writer - DotDirectory string + ContextCancel context.CancelFunc + RefreshInterval time.Duration + RecheckSupportInterval time.Duration + DatabaseUrl string + StoreDebug bool + LogLevel string + LogFormat string + LogWriter io.Writer + DotDirectory string // The amount of time since the last refresh that must have passed for a // search query to trigger an inline refresh. MaxSearchStaleness time.Duration @@ -97,8 +97,8 @@ func (sc *Config) validate(ctx context.Context) error { return errors.New(ctx, errors.InvalidParameter, op, "missing dot directory") case sc.RefreshInterval < 0: return errors.New(ctx, errors.InvalidParameter, op, "negative refresh interval") - case sc.FullFetchInterval < 0: - return errors.New(ctx, errors.InvalidParameter, op, "negative full fetch interval") + case sc.RecheckSupportInterval < 0: + return errors.New(ctx, errors.InvalidParameter, op, "negative recheck support interval") case sc.MaxSearchStaleness < 0: return errors.New(ctx, errors.InvalidParameter, op, "negative max search staleness") } @@ -236,7 +236,7 @@ func (s *CacheServer) Serve(ctx context.Context, cmd Commander, opt ...Option) e s.infoKeys = append(s.infoKeys, "Max Search Refresh Timeout") s.info["Refresh Interval"] = DefaultRefreshInterval.String() s.infoKeys = append(s.infoKeys, "Refresh Interval") - s.info["Full Fetch Interval"] = DefaultFullFetchInterval.String() + s.info["Full Fetch Interval"] = DefaultRecheckSupportInterval.String() s.infoKeys = append(s.infoKeys, "Full Fetch Interval") ticOptions := []Option{} @@ -244,9 +244,9 @@ func (s *CacheServer) Serve(ctx context.Context, cmd Commander, opt ...Option) e s.info["Refresh Interval"] = s.conf.RefreshInterval.String() ticOptions = append(ticOptions, withRefreshInterval(ctx, s.conf.RefreshInterval)) } - if s.conf.FullFetchInterval > 0 { - s.info["Full Fetch Interval"] = s.conf.FullFetchInterval.String() - ticOptions = append(ticOptions, withFullFetchInterval(ctx, s.conf.FullFetchInterval)) + if s.conf.RecheckSupportInterval > 0 { + s.info["Recheck Support Interval"] = s.conf.RecheckSupportInterval.String() + ticOptions = append(ticOptions, withRecheckSupportInterval(ctx, s.conf.RecheckSupportInterval)) } s.printInfo(ctx) @@ -275,7 +275,7 @@ func (s *CacheServer) Serve(ctx context.Context, cmd Commander, opt ...Option) e }() go func() { defer tickerWg.Done() - tic.startFullFetch(tickingCtx) + tic.startRecheckCachingSupport(tickingCtx) }() mux := http.NewServeMux() diff --git a/internal/clientcache/internal/daemon/status_handler.go b/internal/clientcache/internal/daemon/status_handler.go index 47ccc5e63..052665003 100644 --- a/internal/clientcache/internal/daemon/status_handler.go +++ b/internal/clientcache/internal/daemon/status_handler.go @@ -42,12 +42,22 @@ type AuthTokenStatus struct { KeyringlessReferences int `json:"keyringless_references,omitempty"` } +// Information about the boundary controller the data is being cached from. +type BoundaryStatus struct { + // The boundary address for this user + Address string `json:"address,omitempty"` + // Whether the controller responses are supported by the cache + CacheSupport string `json:"cache_support,omitempty"` + // How long ago the Boundary instance was checked for cache compatability + LastSupportCheck time.Duration `json:"last_support_check,omitempty"` +} + // UserStatus contains the status of a specific user tracked by the cache type UserStatus struct { // The Id of the user this status is for Id string `json:"id,omitempty"` - // The boundary address for this user - Address string `json:"address,omitempty"` + // The boundary instance this user connects to + BoundaryInstance BoundaryStatus `json:"boundary_instance,omitempty"` // The auth tokens used by this user to authenticate with the boundary instance AuthTokens []AuthTokenStatus `json:"auth_tokens,omitempty"` // The resources tracked by the cache for this user @@ -116,8 +126,12 @@ func toApiStatus(in *cache.Status, started time.Time, socketAddr string) *Status for _, inU := range in.Users { outU := UserStatus{ - Id: inU.Id, - Address: inU.Address, + Id: inU.Id, + BoundaryInstance: BoundaryStatus{ + Address: inU.BoundaryStatus.Address, + CacheSupport: string(inU.BoundaryStatus.CachingSupported), + LastSupportCheck: inU.BoundaryStatus.LastSupportCheck, + }, } for _, inAt := range inU.AuthTokens { outU.AuthTokens = append(outU.AuthTokens, AuthTokenStatus{ diff --git a/internal/clientcache/internal/daemon/testing.go b/internal/clientcache/internal/daemon/testing.go index 789fb9f24..649bde1b6 100644 --- a/internal/clientcache/internal/daemon/testing.go +++ b/internal/clientcache/internal/daemon/testing.go @@ -33,12 +33,12 @@ func NewTestServer(t *testing.T, cmd Commander, opt ...Option) *TestServer { dotDir := t.TempDir() cfg := &Config{ - ContextCancel: cancel, - RefreshInterval: DefaultRefreshInterval, - FullFetchInterval: DefaultFullFetchInterval, - StoreDebug: opts.withDebug, - LogWriter: io.Discard, - DotDirectory: dotDir, + ContextCancel: cancel, + RefreshInterval: DefaultRefreshInterval, + RecheckSupportInterval: DefaultRecheckSupportInterval, + StoreDebug: opts.withDebug, + LogWriter: io.Discard, + DotDirectory: dotDir, } s, err := New(ctx, cfg) @@ -93,15 +93,45 @@ func (s *TestServer) AddResources(t *testing.T, p *authtokens.AuthToken, tars [] if tok != p.Token { return nil, nil, "", nil } - return tars, nil, "", nil + return tars, nil, "addedtargets", nil } sessFn := func(ctx context.Context, _, tok string, _ cache.RefreshTokenValue) ([]*sessions.Session, []string, cache.RefreshTokenValue, error) { if tok != p.Token { return nil, nil, "", nil } - return sess, nil, "", nil + return sess, nil, "addedsessions", nil } rs, err := cache.NewRefreshService(ctx, r, 0, 0) require.NoError(t, err) require.NoError(t, rs.Refresh(ctx, cache.WithTargetRetrievalFunc(tarFn), cache.WithSessionRetrievalFunc(sessFn))) } + +// AddUnsupportedCachingData provides data in a way that simulates it coming from +// a boundary instance that does not support refresh tokens. Since refresh tokens +// are required for caching, this has the effect of the data not being cached and +// the user being identified as not supported in the cache. +func (s *TestServer) AddUnsupportedCachingData(t *testing.T, p *authtokens.AuthToken, atReadFn cache.BoundaryTokenReaderFn) { + t.Helper() + ctx := context.Background() + r, err := cache.NewRepository(ctx, s.CacheServer.store, &sync.Map{}, s.cmd.ReadTokenFromKeyring, atReadFn) + require.NoError(t, err) + + tarFn := func(ctx context.Context, _, tok string, _ cache.RefreshTokenValue) ([]*targets.Target, []string, cache.RefreshTokenValue, error) { + if tok != p.Token { + return nil, nil, "", nil + } + return []*targets.Target{ + {Id: "ttcp_unsupported", Name: "unsupported", Description: "not supported"}, + }, nil, "", cache.ErrRefreshNotSupported + } + sessFn := func(ctx context.Context, _, tok string, _ cache.RefreshTokenValue) ([]*sessions.Session, []string, cache.RefreshTokenValue, error) { + if tok != p.Token { + return nil, nil, "", nil + } + return []*sessions.Session{}, nil, "", cache.ErrRefreshNotSupported + } + rs, err := cache.NewRefreshService(ctx, r, 0, 0) + require.NoError(t, err) + err = rs.Refresh(ctx, cache.WithTargetRetrievalFunc(tarFn), cache.WithSessionRetrievalFunc(sessFn)) + require.ErrorContains(t, err, "not supported for this controller") +} diff --git a/internal/clientcache/internal/daemon/ticker.go b/internal/clientcache/internal/daemon/ticker.go index e88809663..005eb7118 100644 --- a/internal/clientcache/internal/daemon/ticker.go +++ b/internal/clientcache/internal/daemon/ticker.go @@ -18,31 +18,32 @@ const ( // DefaultRefreshInterval is the default interval that resources that come // from controllers that do support refresh tokens are retrieved and // refreshed. - DefaultRefreshInterval = 60 * time.Second - // DefaultFullFetchInterval is the default interval that resources that come - // from controllers that do not support refresh tokens are fetched. - DefaultFullFetchInterval = 60 * 5 * time.Second + DefaultRefreshInterval = 5 * time.Minute + // DefaultRecheckSupportInterval is the default interval that a boundary + // instance which is not supported in the cache is rechecked to see if it + // has become supported. Support is based on if it supports refresh tokens. + DefaultRecheckSupportInterval = 1 * time.Hour defaultRandomizationFactor float64 = 0.2 ) type refreshService interface { Refresh(context.Context, ...cache.Option) error - FullFetch(context.Context, ...cache.Option) error + RecheckCachingSupport(context.Context, ...cache.Option) error } type refreshTicker struct { - tickerCtx context.Context - refreshInterval time.Duration - fullFetchInterval time.Duration - randomizationFactor float64 - rand *rand.Rand - refreshChan chan struct{} - refresher refreshService + tickerCtx context.Context + refreshInterval time.Duration + recheckSupportInterval time.Duration + randomizationFactor float64 + rand *rand.Rand + refreshChan chan struct{} + refresher refreshService } // newRefreshTicker creates a new refresh ticker. Accepted options are -// withRefreshInterval, and withFullFetchInterval. +// withRefreshInterval, and withRecheckSupportInterval. // 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 @@ -65,9 +66,9 @@ func newRefreshTicker(ctx context.Context, refresh refreshService, opt ...Option refreshInterval = opts.withRefreshInterval } - fullFetchInterval := DefaultFullFetchInterval - if opts.withFullFetchInterval != 0 { - fullFetchInterval = opts.withFullFetchInterval + recheckSupportInterval := DefaultRecheckSupportInterval + if opts.withRecheckSupportInterval != 0 { + recheckSupportInterval = opts.withRecheckSupportInterval } randomizationFactor := defaultRandomizationFactor @@ -76,11 +77,11 @@ func newRefreshTicker(ctx context.Context, refresh refreshService, opt ...Option } return &refreshTicker{ - refreshInterval: refreshInterval, - fullFetchInterval: fullFetchInterval, - randomizationFactor: randomizationFactor, - rand: rand.New(rand.NewSource(time.Now().UnixNano())), - refresher: refresh, + refreshInterval: refreshInterval, + recheckSupportInterval: recheckSupportInterval, + 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 @@ -100,8 +101,11 @@ func (rt *refreshTicker) refresh() { } } -func (rt *refreshTicker) startFullFetch(ctx context.Context) { - const op = "daemon.(refreshTicker).startFullFetch" +// startRecheckCachingSupport repeatedly calls RecheckCachingSupport which +// checks all users that are marked as not supported by the cache to see if +// the controller version has been updated to support refresh tokens/caching. +func (rt *refreshTicker) startRecheckCachingSupport(ctx context.Context) { + const op = "daemon.(refreshTicker).startRecheckCachingSupport" timer := time.NewTimer(0) for { // full fetch doesn't listen to the refreshChan since completely new @@ -112,10 +116,10 @@ func (rt *refreshTicker) startFullFetch(ctx context.Context) { return case <-timer.C: } - if err := rt.refresher.FullFetch(ctx); err != nil { + if err := rt.refresher.RecheckCachingSupport(ctx); err != nil { event.WriteError(rt.tickerCtx, op, err) } - timer.Reset(rt.nextIntervalWithRandomness(rt.fullFetchInterval)) + timer.Reset(rt.nextIntervalWithRandomness(rt.recheckSupportInterval)) } } diff --git a/internal/clientcache/internal/daemon/ticker_test.go b/internal/clientcache/internal/daemon/ticker_test.go index cf688b385..a769eb5a9 100644 --- a/internal/clientcache/internal/daemon/ticker_test.go +++ b/internal/clientcache/internal/daemon/ticker_test.go @@ -47,7 +47,7 @@ func TestTickerRefresh(t *testing.T) { } } -func TestTickerFullFetch(t *testing.T) { +func TestTickerRecheckCachingSupport(t *testing.T) { ctx := context.Background() refresher := &fakeRefresher{make(chan struct{})} @@ -56,7 +56,7 @@ func TestTickerFullFetch(t *testing.T) { require.NoError(t, err) tickerCtx, tickerCancel := context.WithCancel(ctx) - go rt.startFullFetch(tickerCtx) + go rt.startRecheckCachingSupport(tickerCtx) t.Cleanup(func() { tickerCancel() }) @@ -106,7 +106,7 @@ func (r *fakeRefresher) Refresh(context.Context, ...cache.Option) error { return nil } -func (r *fakeRefresher) FullFetch(context.Context, ...cache.Option) error { +func (r *fakeRefresher) RecheckCachingSupport(context.Context, ...cache.Option) error { r.called <- struct{}{} return nil }