Do not cache or allow searching for old controllers (#4136)
* Do not cache or allow searching for old controllers * Identify resourceless responses as coming from supported or unsupported controllers * Do not report refresh token status info for unsupported controllers
This commit is contained in:
1 parent
d94a25bbd2
commit
e429e91bc4
21 files changed
+855
-322
No files matched your search
@@ -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")
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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",
|
||||
|
||||
+25
-35
@@ -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))))
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
+76
-79
@@ -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)
|
||||
|
||||
+95
-15
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
+63
-25
@@ -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
|
||||
}
|
||||
|
||||
|
||||
+72
-31
@@ -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
|
||||
}
|
||||
|
||||
|
||||
+27
-10
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
+39
-20
@@ -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),
|
||||
|
||||
+79
-4
@@ -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,
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
@@ -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{
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
Reference in new issue
Block a user