diff --git a/internal/clientcache/cmd/daemon/command_wrapper.go b/internal/clientcache/cmd/daemon/command_wrapper.go index cdb372ecb..fda2568c1 100644 --- a/internal/clientcache/cmd/daemon/command_wrapper.go +++ b/internal/clientcache/cmd/daemon/command_wrapper.go @@ -51,6 +51,12 @@ func (w *CommandWrapper) Run(args []string) int { w.cacheEnabledCommand.BaseCommand().Opts = append(w.cacheEnabledCommand.BaseCommand().Opts, base.WithInterceptedToken(&token)) r := w.cacheEnabledCommand.Run(args) + if r != base.CommandSuccess { + // if we were not successful in running our command, do not continue to + // start the daemon and add the token. + return r + } + ctx := context.Background() if w.startDaemon(ctx) { w.addTokenToCache(ctx, token) diff --git a/internal/clientcache/cmd/daemon/start.go b/internal/clientcache/cmd/daemon/start.go index 8cd524e34..f6ff5f844 100644 --- a/internal/clientcache/cmd/daemon/start.go +++ b/internal/clientcache/cmd/daemon/start.go @@ -49,13 +49,15 @@ type server interface { type StartCommand struct { *base.Command - flagRefreshInterval time.Duration - flagFullFetchInterval time.Duration - flagDatabaseUrl string - flagLogLevel string - flagLogFormat string - flagStoreDebug bool - flagBackground bool + flagRefreshInterval time.Duration + flagFullFetchInterval time.Duration + flagMaxSearchStaleness time.Duration + flagMaxSearchRefreshTimeout time.Duration + flagDatabaseUrl string + flagLogLevel string + flagLogFormat string + flagStoreDebug bool + flagBackground bool } func (c *StartCommand) Synopsis() string { @@ -100,29 +102,41 @@ func (c *StartCommand) Flags() *base.FlagSets { Usage: `If set, specifies the URL used to connect to the sqlite database (store) for caching. This can refer to a file on disk (file://) from which a URL will be read; an env var (env://) from which the URL will be read; or a direct database URL.`, }) f.DurationVar(&base.DurationVar{ - Name: "refresh-interval-seconds", + Name: "refresh-interval", Target: &c.flagRefreshInterval, - Usage: `If set, specifies the number of seconds between refresh token supported cache refreshes. Default: 1 minute`, - Default: daemon.DefaultRefreshIntervalSeconds * time.Second, + Usage: `Specifies the interval between refresh token supported cache refreshes. Default: 1 minute`, + Default: daemon.DefaultRefreshInterval, }) f.DurationVar(&base.DurationVar{ - Name: "full-fetch-interval-seconds", + Name: "full-fetch-interval", Target: &c.flagFullFetchInterval, - Usage: `If set, specifies the number of seconds between full cache refresh for boundary instances which do not support refresh tokens. Default: 5 minutes`, - Default: daemon.DefaultFullFetchIntervalSeconds * time.Second, + Usage: `Specifies the interval between full cache refresh for boundary instances which do not support refresh tokens. Default: 5 minutes`, + Default: daemon.DefaultFullFetchInterval, + }) + 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`, + 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`, + Default: daemon.DefaultSearchRefreshTimeout, }) f.BoolVar(&base.BoolVar{ Name: "store-debug", Target: &c.flagStoreDebug, Default: false, - Usage: `Turn on store debugging`, + Usage: `Turn on sqlite query debugging`, Aliases: []string{"d"}, }) f.BoolVar(&base.BoolVar{ Name: "background", Target: &c.flagBackground, Default: false, - Usage: `Turn on store debugging`, + Usage: `Run the client cache daemon in the background`, }) return set @@ -156,7 +170,7 @@ func (c *StartCommand) Run(args []string) int { return base.CommandCliError } - continueRun, writers, cleanup, err := makeBackground(ctx, dotDir, c.flagBackground) + continueRun, writers, cleanup, err := c.makeBackground(ctx, dotDir) defer func() { if cleanup != nil { cleanup() @@ -182,15 +196,16 @@ func (c *StartCommand) Run(args []string) int { writers = append(writers, lf) cfg := &daemon.Config{ - ContextCancel: cancel, - RefreshInterval: c.flagRefreshInterval, - FullFetchInterval: c.flagFullFetchInterval, - DatabaseUrl: c.flagDatabaseUrl, - StoreDebug: c.flagStoreDebug, - LogLevel: c.flagLogLevel, - LogFormat: c.flagLogFormat, - LogWriter: io.MultiWriter(writers...), - DotDirectory: dotDir, + 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, } srv, err := daemon.New(ctx, cfg) @@ -259,7 +274,7 @@ func logFile(ctx context.Context, dotDir string, maxSizeMb int) (io.WriteCloser, return logFile, nil } -func makeBackground(ctx context.Context, dotDir string, runBackgroundFlag bool) (bool, []io.Writer, pidCleanup, error) { +func (c *StartCommand) makeBackground(ctx context.Context, dotDir string) (bool, []io.Writer, pidCleanup, error) { const op = "daemon.makeBackground" writers := []io.Writer{} @@ -270,11 +285,11 @@ func makeBackground(ctx context.Context, dotDir string, runBackgroundFlag bool) return false, writers, noopPidCleanup, fmt.Errorf("Error when checking if the daemon pid is in use: %w.", err) } - if !runBackgroundFlag && os.Getenv(backgroundEnvName) != backgroundEnvVal { + if !c.flagBackground && os.Getenv(backgroundEnvName) != backgroundEnvVal { writers = append(writers, os.Stderr) } - if !runBackgroundFlag || os.Getenv(backgroundEnvName) == backgroundEnvVal { + if !c.flagBackground || os.Getenv(backgroundEnvName) == backgroundEnvVal { // We are either already running in the background or background was // not requested. Write the pid file and continue. cleanup, err := writePidFile(ctx, pidPath) @@ -291,7 +306,32 @@ func makeBackground(ctx context.Context, dotDir string, runBackgroundFlag bool) env := os.Environ() env = append(env, fmt.Sprintf("%s=%s", backgroundEnvName, backgroundEnvVal)) - cmd := exec.Command(absPath, "daemon", "start") + args := []string{"daemon", "start"} + if c.flagLogLevel != "" { + args = append(args, "-log-level", c.flagLogLevel) + } + if c.flagLogFormat != "" { + args = append(args, "-log-format", c.flagLogFormat) + } + 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") + } + cmd := exec.Command(absPath, args...) cmd.Env = env if err = cmd.Start(); err != nil { return false, writers, noopPidCleanup, errors.Wrap(ctx, err, op) diff --git a/internal/clientcache/cmd/daemon/status.go b/internal/clientcache/cmd/daemon/status.go index 3a6fe24fb..c02bf0a63 100644 --- a/internal/clientcache/cmd/daemon/status.go +++ b/internal/clientcache/cmd/daemon/status.go @@ -20,7 +20,7 @@ import ( "github.com/posener/complete" ) -var errDaemonNotRunning = stderr.New("daemon is not running") +var errDaemonNotRunning = stderr.New("The daemon process is not running.") var ( _ cli.Command = (*AddTokenCommand)(nil) diff --git a/internal/clientcache/cmd/daemon/stop.go b/internal/clientcache/cmd/daemon/stop.go index 1b11b0b15..76f2b3131 100644 --- a/internal/clientcache/cmd/daemon/stop.go +++ b/internal/clientcache/cmd/daemon/stop.go @@ -5,7 +5,6 @@ package daemon import ( "context" - stderr "errors" "strings" "github.com/hashicorp/boundary/internal/cmd/base" @@ -60,7 +59,7 @@ func (c *StopCommand) Run(args []string) int { } if err := c.stop(c.Context); err != nil { if errors.Match(errors.T(errors.NotFound), err) { - c.PrintCliError(stderr.New("The deamon process is not running.")) + c.PrintCliError(errDaemonNotRunning) } else { c.PrintCliError(err) } diff --git a/internal/clientcache/cmd/search/search.go b/internal/clientcache/cmd/search/search.go index 8d456d1ff..ec5db74bf 100644 --- a/internal/clientcache/cmd/search/search.go +++ b/internal/clientcache/cmd/search/search.go @@ -32,6 +32,8 @@ var ( "targets", "sessions", } + + errDaemonNotRunning = stderrors.New("The deamon process is not running.") ) type SearchCommand struct { @@ -165,10 +167,11 @@ func (c *SearchCommand) Search(ctx context.Context) (*api.Response, *daemon.Sear func search(ctx context.Context, daemonPath string, fb filterBy, opt ...client.Option) (*api.Response, *daemon.SearchResult, *api.Error, error) { addr, err := daemon.SocketAddress(daemonPath) if err != nil { - return nil, nil, nil, fmt.Errorf("Error when retrieving the socket address: %w", err) + return nil, nil, nil, fmt.Errorf("Error getting socket address: %w", err) } - if _, err := os.Stat(addr.Path); addr.Scheme == "unix" && err == os.ErrNotExist { - return nil, nil, nil, fmt.Errorf("Error when detecting if the domain socket is present: %w.", err) + _, err = os.Stat(addr.Path) + if addr.Scheme == "unix" && err != nil { + return nil, nil, nil, errDaemonNotRunning } c, err := client.New(ctx, addr) if err != nil { diff --git a/internal/clientcache/internal/cache/refresh.go b/internal/clientcache/internal/cache/refresh.go index 517aa3222..5c0cef14d 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" + "time" "github.com/hashicorp/boundary/api" "github.com/hashicorp/boundary/api/authtokens" @@ -16,15 +17,27 @@ import ( type RefreshService struct { repo *Repository + // the amount of time that should have passed since the last refresh for a + // call to RefreshForSearch will cause a refresh request to be sent to a + // boundary controller. + maxSearchStaleness time.Duration + // the amount of time that RefreshForSearch should run before it times out. + maxSearchRefreshTimeout time.Duration } -func NewRefreshService(ctx context.Context, r *Repository) (*RefreshService, error) { +func NewRefreshService(ctx context.Context, r *Repository, maxSearchStaleness, maxSearchRefreshTimeout time.Duration) (*RefreshService, error) { const op = "cache.NewRefreshService" switch { case util.IsNil(r): return nil, errors.New(ctx, errors.InvalidParameter, op, "repository is nil") + case maxSearchStaleness < 0: + return nil, errors.New(ctx, errors.InvalidParameter, op, "max search staleness is negative") } - return &RefreshService{repo: r}, nil + return &RefreshService{ + repo: r, + maxSearchStaleness: maxSearchStaleness, + maxSearchRefreshTimeout: maxSearchRefreshTimeout, + }, nil } // cleanAndPickAuthTokens removes from the cache all auth tokens which are @@ -131,6 +144,90 @@ func (r *RefreshService) refreshableUsers(ctx context.Context, in []*user) ([]*u return ret, nil } +// RefreshForSearch refreshes a specific resource type owned by the user associated +// with the provided auth token id, if it hasn't been refreshed in a certain +// amount of time. If the resources has been updated recently, or if it is known +// that the user's boundary instance doesn't support partial refreshing, this +// method returns without any requests to the boundary controller. +// While the criteria of whether the user will refresh or not is almost the same +// as the criteria used in Refresh, RefreshForSearch will not refresh any +// data if there is not a refresh token stored for the resource. It +// might make sense to change this in the future, but the reasoning used is +// that we should not be making an initial load of all resources while blocking +// a search query, in case we have not yet even attempted to load the resources +// for this user yet. +// Note: Currently, if the context timesout we stop refreshing completely and +// return to the caller. A possible enhancement in the future would be to return +// when the context is Done, but allow the refresh to proceed in the background. +func (r *RefreshService) RefreshForSearch(ctx context.Context, authTokenid string, resourceType SearchableResource, opt ...Option) error { + const op = "cache.(RefreshService).RefreshForSearch" + if r.maxSearchRefreshTimeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, r.maxSearchRefreshTimeout) + defer cancel() + } + at, err := r.repo.LookupToken(ctx, authTokenid) + if err != nil { + return errors.Wrap(ctx, err, op) + } + if at == nil { + return errors.New(ctx, errors.NotFound, op, "auth token not found") + } + u, err := r.repo.lookupUser(ctx, at.UserId) + if err != nil { + return errors.Wrap(ctx, err, op) + } + if u == nil { + return errors.New(ctx, errors.NotFound, op, "user not found") + } + us, err := r.refreshableUsers(ctx, []*user{u}) + if err != nil { + return errors.Wrap(ctx, err, op) + } + + switch len(us) { + case 0: + // The provided user is not refreshable, so do not refresh. + return nil + case 1: + u = us[0] + default: + return errors.New(ctx, errors.Internal, op, "unexpected number of refreshable users") + } + + tokens, err := r.cleanAndPickAuthTokens(ctx, u) + if err != nil { + return errors.Wrap(ctx, err, op) + } + + switch resourceType { + case Targets: + rtv, err := r.repo.lookupRefreshToken(ctx, u, targetResourceType) + if err != nil { + return errors.Wrap(ctx, err, op) + } + if rtv != nil && time.Since(rtv.UpdateTime) > r.maxSearchStaleness { + if err := r.repo.refreshTargets(ctx, u, tokens, opt...); err != nil { + return errors.Wrap(ctx, err, op) + } + } + case Sessions: + rtv, err := r.repo.lookupRefreshToken(ctx, u, sessionResourceType) + if err != nil { + return errors.Wrap(ctx, err, op) + } + if rtv != nil && time.Since(rtv.UpdateTime) > r.maxSearchStaleness { + if err := r.repo.refreshSessions(ctx, u, tokens, opt...); err != nil { + return errors.Wrap(ctx, err, op) + } + } + default: + return errors.New(ctx, errors.InvalidParameter, op, "unrecognized resource type") + } + + return nil +} + // Refresh iterates over all tokens in the cache that are for users who either // have a refresh token or which do not have any resources in the cache yet. It // then attempts to read those user's resources from boundary and updates the diff --git a/internal/clientcache/internal/cache/refresh_test.go b/internal/clientcache/internal/cache/refresh_test.go index 4364c0660..9b8c69038 100644 --- a/internal/clientcache/internal/cache/refresh_test.go +++ b/internal/clientcache/internal/cache/refresh_test.go @@ -52,6 +52,8 @@ func testStaticResourceRetrievalFunc[T any](t *testing.T, ret [][]T, removed [][ } } +// testFullFetchRetrievalFunc 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) { return func(_ context.Context, _, _ string, _ RefreshTokenValue) ([]T, []string, RefreshTokenValue, error) { return ret, nil, "", nil @@ -137,7 +139,7 @@ func TestCleanAndPickTokens(t *testing.T) { mapBasedAuthTokenKeyringLookup(atMap), fakeBoundaryLookupFn) require.NoError(t, err) - rs, err := NewRefreshService(ctx, r) + rs, err := NewRefreshService(ctx, r, 0, 0) require.NoError(t, err) t.Run("unknown user", func(t *testing.T) { @@ -331,6 +333,167 @@ func TestCleanAndPickTokens(t *testing.T) { }) } +func TestRefreshForSearch(t *testing.T) { + ctx := context.Background() + + boundaryAddr := "address" + u := &user{Id: "u1", Address: boundaryAddr} + at := &authtokens.AuthToken{ + Id: "at_1", + Token: "at_1_token", + UserId: u.Id, + ExpirationTime: time.Now().Add(time.Minute), + } + + boundaryAuthTokens := []*authtokens.AuthToken{at} + atMap := make(map[ringToken]*authtokens.AuthToken) + + atMap[ringToken{"k", "t"}] = at + + t.Run("targets refreshed for searching", func(t *testing.T) { + s, err := db.Open(ctx) + require.NoError(t, err) + r, err := NewRepository(ctx, s, &sync.Map{}, mapBasedAuthTokenKeyringLookup(atMap), sliceBasedAuthTokenBoundaryReader(boundaryAuthTokens)) + require.NoError(t, err) + rs, err := NewRefreshService(ctx, r, time.Millisecond, 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"), + } + opts := []Option{ + WithSessionRetrievalFunc(testStaticResourceRetrievalFunc[*sessions.Session](t, nil, nil)), + WithTargetRetrievalFunc(testStaticResourceRetrievalFunc[*targets.Target](t, + [][]*targets.Target{ + retTargets[:3], + retTargets[3:], + }, + [][]string{ + nil, + {retTargets[0].Id, retTargets[1].Id}, + }, + )), + } + assert.NoError(t, rs.RefreshForSearch(ctx, at.Id, Targets, opts...)) + + cachedTargets, err := r.ListTargets(ctx, at.Id) + assert.NoError(t, err) + assert.Empty(t, cachedTargets) + + // Now load up a few resources and a token, and trying again should + // see the RefreshForSearch update more fields. + assert.NoError(t, rs.Refresh(ctx, opts...)) + cachedTargets, err = r.ListTargets(ctx, at.Id) + assert.NoError(t, err) + assert.ElementsMatch(t, retTargets[:3], cachedTargets) + + // Let 2 milliseconds pass so the items are stale enough + time.Sleep(2 * time.Millisecond) + + assert.NoError(t, rs.RefreshForSearch(ctx, at.Id, Targets, opts...)) + cachedTargets, err = r.ListTargets(ctx, at.Id) + assert.NoError(t, err) + assert.ElementsMatch(t, retTargets[2:], cachedTargets) + }) + + t.Run("no refresh token no refresh for search", func(t *testing.T) { + s, err := db.Open(ctx) + require.NoError(t, err) + r, err := NewRepository(ctx, s, &sync.Map{}, mapBasedAuthTokenKeyringLookup(atMap), sliceBasedAuthTokenBoundaryReader(boundaryAuthTokens)) + require.NoError(t, err) + rs, err := NewRefreshService(ctx, r, time.Millisecond, 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])))) + + got, err := r.ListTargets(ctx, at.Id) + assert.NoError(t, err) + assert.ElementsMatch(t, retTargets[:2], got) + + // Let 2 milliseconds pass so the items are stale enough + time.Sleep(2 * time.Millisecond) + + 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) + + // 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) + }) + + t.Run("sessions refreshed for searching", func(t *testing.T) { + s, err := db.Open(ctx) + require.NoError(t, err) + r, err := NewRepository(ctx, s, &sync.Map{}, mapBasedAuthTokenKeyringLookup(atMap), sliceBasedAuthTokenBoundaryReader(boundaryAuthTokens)) + require.NoError(t, err) + rs, err := NewRefreshService(ctx, r, 0, 0) + 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"), + } + opts := []Option{ + WithTargetRetrievalFunc(testStaticResourceRetrievalFunc[*targets.Target](t, nil, nil)), + WithSessionRetrievalFunc(testStaticResourceRetrievalFunc[*sessions.Session](t, + [][]*sessions.Session{ + retSess[:3], + retSess[3:], + }, + [][]string{ + nil, + {retSess[0].Id, retSess[1].Id}, + }, + )), + } + + // First call doesn't sync anything because no sessions were already synced yet + assert.NoError(t, rs.RefreshForSearch(ctx, at.Id, Sessions, opts...)) + cachedSessions, err := r.ListSessions(ctx, at.Id) + assert.NoError(t, err) + assert.Empty(t, cachedSessions) + + assert.NoError(t, rs.Refresh(ctx, opts...)) + cachedSessions, err = r.ListSessions(ctx, at.Id) + assert.NoError(t, err) + assert.ElementsMatch(t, retSess[:3], cachedSessions) + + // Second call removes the first 2 resources from the cache and adds the last + assert.NoError(t, rs.RefreshForSearch(ctx, at.Id, Sessions, opts...)) + cachedSessions, err = r.ListSessions(ctx, at.Id) + assert.NoError(t, err) + assert.ElementsMatch(t, retSess[2:], cachedSessions) + }) +} + func TestRefresh(t *testing.T) { ctx := context.Background() @@ -353,7 +516,7 @@ func TestRefresh(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) + 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})) @@ -394,7 +557,7 @@ func TestRefresh(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) + 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})) @@ -434,7 +597,7 @@ func TestRefresh(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) + 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})) @@ -462,7 +625,7 @@ func TestRefresh(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) + 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})) @@ -516,7 +679,7 @@ func TestFullFetch(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) + 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})) @@ -559,7 +722,7 @@ func TestFullFetch(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) + 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})) @@ -597,7 +760,7 @@ func TestFullFetch(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) + 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})) @@ -630,7 +793,7 @@ func TestFullFetch(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) + 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})) diff --git a/internal/clientcache/internal/cache/repository.go b/internal/clientcache/internal/cache/repository.go index f699280f0..eb5883df5 100644 --- a/internal/clientcache/internal/cache/repository.go +++ b/internal/clientcache/internal/cache/repository.go @@ -26,6 +26,7 @@ type KeyringTokenLookupFn func(keyring string, tokenName string) *authtokens.Aut type BoundaryTokenReaderFn func(ctx context.Context, addr string, authToken string) (*authtokens.AuthToken, error) type Repository struct { + serverCtx context.Context rw *db.Db tokenKeyringFn KeyringTokenLookupFn tokenReadFromBoundaryFn BoundaryTokenReaderFn @@ -33,7 +34,8 @@ type Repository struct { idToKeyringlessAuthToken *sync.Map } -// NewRepository returns a cache repository. The provided Store must be +// NewRepository returns a cache repository. The provided context is stored as +// the server context for purposes like storing boundary request errors. func NewRepository(ctx context.Context, conn *db.DB, idToAuthToken *sync.Map, keyringFn KeyringTokenLookupFn, atReadFn BoundaryTokenReaderFn, opt ...Option) (*Repository, error) { const op = "cache.NewRepository" switch { @@ -47,6 +49,7 @@ func NewRepository(ctx context.Context, conn *db.DB, idToAuthToken *sync.Map, ke return nil, errors.New(ctx, errors.InvalidParameter, op, "missing auth token read function") } return &Repository{ + serverCtx: ctx, rw: db.New(conn), tokenKeyringFn: keyringFn, tokenReadFromBoundaryFn: atReadFn, @@ -110,8 +113,8 @@ func (r *Repository) lookupError(ctx context.Context, u *user, resourceType reso type apiError struct { UserId string `gorm:"primaryKey"` ResourceType resourceType `gorm:"primaryKey"` - Error string - CreateTime time.Time + Error string `gorm:"default:null"` + CreateTime time.Time `gorm:"default:(strftime('%Y-%m-%d %H:%M:%f','now'))"` } func (*apiError) TableName() string { diff --git a/internal/clientcache/internal/cache/repository_refresh_token.go b/internal/clientcache/internal/cache/repository_refresh_token.go index 46c2076ec..f44b0a90b 100644 --- a/internal/clientcache/internal/cache/repository_refresh_token.go +++ b/internal/clientcache/internal/cache/repository_refresh_token.go @@ -11,7 +11,6 @@ import ( "github.com/hashicorp/boundary/internal/db" "github.com/hashicorp/boundary/internal/errors" "github.com/hashicorp/boundary/internal/util" - "github.com/hashicorp/go-dbw" ) // RefreshTokenValue is the the type for the actual refresh token value handled @@ -21,15 +20,15 @@ type RefreshTokenValue string // 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. -func (r *Repository) lookupRefreshToken(ctx context.Context, u *user, resourceType resourceType) (RefreshTokenValue, error) { +func (r *Repository) lookupRefreshToken(ctx context.Context, u *user, resourceType resourceType) (*refreshToken, error) { const op = "cache.(Repsoitory).lookupRefreshToken" switch { case util.IsNil(u): - return "", errors.New(ctx, errors.InvalidParameter, op, "user is nil") + return nil, errors.New(ctx, errors.InvalidParameter, op, "user is nil") case u.Id == "": - return "", errors.New(ctx, errors.InvalidParameter, op, "user id is empty") + return nil, errors.New(ctx, errors.InvalidParameter, op, "user id is empty") case !resourceType.valid(): - return "", errors.New(ctx, errors.InvalidParameter, op, "resource type is invalid") + return nil, errors.New(ctx, errors.InvalidParameter, op, "resource type is invalid") } rt := &refreshToken{ @@ -37,12 +36,12 @@ func (r *Repository) lookupRefreshToken(ctx context.Context, u *user, resourceTy ResourceType: resourceType, } if err := r.rw.LookupById(ctx, rt); err != nil { - if errors.Is(err, dbw.ErrRecordNotFound) { - return "", nil + if errors.IsNotFoundError(err) { + return nil, nil } - return "", errors.Wrap(ctx, err, op) + return nil, errors.Wrap(ctx, err, op) } - return rt.RefreshToken, nil + return rt, nil } // listRefreshTokens returns all refresh tokens associated with a specific user diff --git a/internal/clientcache/internal/cache/repository_refresh_token_test.go b/internal/clientcache/internal/cache/repository_refresh_token_test.go index 24340d9ea..bf740cb66 100644 --- a/internal/clientcache/internal/cache/repository_refresh_token_test.go +++ b/internal/clientcache/internal/cache/repository_refresh_token_test.go @@ -7,6 +7,7 @@ import ( "context" "sync" "testing" + "time" "github.com/hashicorp/boundary/api/authtokens" cachedb "github.com/hashicorp/boundary/internal/clientcache/internal/db" @@ -119,6 +120,7 @@ func TestLookupRefreshToken(t *testing.T) { known := &user{Id: "withrefreshtoken", Address: "addr"} require.NoError(t, r.rw.Create(ctx, known)) + before := time.Now().Truncate(time.Millisecond).UTC() 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 @@ -126,7 +128,8 @@ func TestLookupRefreshToken(t *testing.T) { got, err := r.lookupRefreshToken(ctx, known, targetResourceType) assert.NoError(t, err) - assert.Equal(t, token, got) + assert.Equal(t, token, got.RefreshToken) + assert.GreaterOrEqual(t, got.CreateTime, before) }) } diff --git a/internal/clientcache/internal/cache/repository_sessions.go b/internal/clientcache/internal/cache/repository_sessions.go index 3a523fe06..05645431a 100644 --- a/internal/clientcache/internal/cache/repository_sessions.go +++ b/internal/clientcache/internal/cache/repository_sessions.go @@ -65,10 +65,14 @@ func (r *Repository) refreshSessions(ctx context.Context, u *user, tokens map[Au if opts.withSessionRetrievalFunc == nil { opts.withSessionRetrievalFunc = defaultSessionFunc } + var oldRefreshTokenVal RefreshTokenValue oldRefreshToken, err := r.lookupRefreshToken(ctx, u, resourceType) if err != nil { return errors.Wrap(ctx, err, op) } + if oldRefreshToken != nil { + oldRefreshTokenVal = oldRefreshToken.RefreshToken + } // Find and use a token for retrieving sessions var gotResponse bool @@ -77,13 +81,13 @@ func (r *Repository) refreshSessions(ctx context.Context, u *user, tokens map[Au var removedIds []string var retErr error for at, t := range tokens { - resp, removedIds, newRefreshToken, err = opts.withSessionRetrievalFunc(ctx, u.Address, t, oldRefreshToken) + resp, removedIds, newRefreshToken, err = opts.withSessionRetrievalFunc(ctx, u.Address, t, oldRefreshTokenVal) if api.ErrInvalidRefreshToken.Is(err) { if err := r.deleteRefreshToken(ctx, u, resourceType); err != nil { return errors.Wrap(ctx, err, op) } // try again without the refresh token - oldRefreshToken = "" + oldRefreshToken = nil resp, removedIds, newRefreshToken, err = opts.withSessionRetrievalFunc(ctx, u.Address, t, "") } if err != nil { @@ -95,7 +99,7 @@ func (r *Repository) refreshSessions(ctx context.Context, u *user, tokens map[Au } if retErr != nil { - if saveErr := r.saveError(ctx, u, resourceType, retErr); saveErr != nil { + if saveErr := r.saveError(r.serverCtx, u, resourceType, retErr); saveErr != nil { return stderrors.Join(err, errors.Wrap(ctx, saveErr, op)) } } @@ -106,7 +110,7 @@ func (r *Repository) refreshSessions(ctx context.Context, u *user, tokens map[Au event.WriteSysEvent(ctx, op, fmt.Sprintf("updating %d sessions for user %v", len(resp), u)) _, err = r.rw.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(_ db.Reader, w db.Writer) error { switch { - case oldRefreshToken == "": + case oldRefreshToken == nil: if _, err := w.Exec(ctx, "delete from session where user_id = @user_id", []any{sql.Named("user_id", u.Id)}); err != nil { return err @@ -171,7 +175,7 @@ func (r *Repository) fullFetchSessions(ctx context.Context, u *user, tokens map[ } if retErr != nil { - if saveErr := r.saveError(ctx, u, resourceType, retErr); saveErr != nil { + if saveErr := r.saveError(r.serverCtx, u, resourceType, retErr); saveErr != nil { return stderrors.Join(err, errors.Wrap(ctx, saveErr, op)) } } diff --git a/internal/clientcache/internal/cache/repository_targets.go b/internal/clientcache/internal/cache/repository_targets.go index f4fa8d25d..b46c86658 100644 --- a/internal/clientcache/internal/cache/repository_targets.go +++ b/internal/clientcache/internal/cache/repository_targets.go @@ -63,26 +63,31 @@ func (r *Repository) refreshTargets(ctx context.Context, u *user, tokens map[Aut if opts.withTargetRetrievalFunc == nil { opts.withTargetRetrievalFunc = defaultTargetFunc } + + var oldRefreshTokenVal RefreshTokenValue oldRefreshToken, err := r.lookupRefreshToken(ctx, u, resourceType) if err != nil { return errors.Wrap(ctx, err, op) } + if oldRefreshToken != nil { + oldRefreshTokenVal = oldRefreshToken.RefreshToken + } // Find and use a token for retrieving targets var gotResponse bool var resp []*targets.Target var removedIds []string - var newRefreshToken RefreshTokenValue + var newRefreshTokenVal RefreshTokenValue var retErr error for at, t := range tokens { - resp, removedIds, newRefreshToken, err = opts.withTargetRetrievalFunc(ctx, u.Address, t, oldRefreshToken) + resp, removedIds, newRefreshTokenVal, err = opts.withTargetRetrievalFunc(ctx, u.Address, t, oldRefreshTokenVal) if api.ErrInvalidRefreshToken.Is(err) { if err := r.deleteRefreshToken(ctx, u, resourceType); err != nil { return errors.Wrap(ctx, err, op) } // try again without the refresh token - oldRefreshToken = "" - resp, removedIds, newRefreshToken, err = opts.withTargetRetrievalFunc(ctx, u.Address, t, "") + oldRefreshToken = nil + resp, removedIds, newRefreshTokenVal, 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))) @@ -92,7 +97,7 @@ func (r *Repository) refreshTargets(ctx context.Context, u *user, tokens map[Aut break } if retErr != nil { - if saveErr := r.saveError(ctx, u, resourceType, retErr); saveErr != nil { + if saveErr := r.saveError(r.serverCtx, u, resourceType, retErr); saveErr != nil { return stderrors.Join(err, errors.Wrap(ctx, saveErr, op)) } } @@ -103,7 +108,7 @@ 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 { switch { - case oldRefreshToken == "": + case oldRefreshToken == nil: if _, err := w.Exec(ctx, "delete from target where user_id = @user_id", []any{sql.Named("user_id", u.Id)}); err != nil { return err @@ -118,8 +123,8 @@ func (r *Repository) refreshTargets(ctx context.Context, u *user, tokens map[Aut if err := upsertTargets(ctx, w, u, resp); err != nil { return err } - if newRefreshToken != "" || oldRefreshToken != "" { - if err := upsertRefreshToken(ctx, w, u, resourceType, newRefreshToken); err != nil { + if newRefreshTokenVal != "" || oldRefreshToken != nil { + if err := upsertRefreshToken(ctx, w, u, resourceType, newRefreshTokenVal); err != nil { return err } } @@ -167,7 +172,7 @@ func (r *Repository) fullFetchTargets(ctx context.Context, u *user, tokens map[A break } if retErr != nil { - if saveErr := r.saveError(ctx, u, resourceType, retErr); saveErr != nil { + if saveErr := r.saveError(r.serverCtx, u, resourceType, retErr); saveErr != nil { return stderrors.Join(err, errors.Wrap(ctx, saveErr, op)) } } diff --git a/internal/clientcache/internal/cache/repository_token.go b/internal/clientcache/internal/cache/repository_token.go index aa592390f..6a17b0ef6 100644 --- a/internal/clientcache/internal/cache/repository_token.go +++ b/internal/clientcache/internal/cache/repository_token.go @@ -382,6 +382,23 @@ func cleanExpiredOrOrphanedAuthTokens(ctx context.Context, writer db.Writer, idT return nil } +// lookupUser returns a user if one is present in the repository or nil if not. +func (r *Repository) lookupUser(ctx context.Context, id string) (*user, error) { + const op = "cache.(Repository).lookupUser" + switch { + case id == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "empty id") + } + ret := &user{Id: id} + if err := r.rw.LookupById(ctx, ret); err != nil { + if errors.IsNotFoundError(err) { + return nil, nil + } + return nil, errors.Wrap(ctx, err, op) + } + return ret, nil +} + // listUsers returns all known tokens in the cache func (r *Repository) listUsers(ctx context.Context) ([]*user, error) { const op = "cache.(Repository).listUsers" diff --git a/internal/clientcache/internal/cache/repository_token_test.go b/internal/clientcache/internal/cache/repository_token_test.go index 08ffc3a77..b6fd94e29 100644 --- a/internal/clientcache/internal/cache/repository_token_test.go +++ b/internal/clientcache/internal/cache/repository_token_test.go @@ -610,6 +610,48 @@ func TestRepository_LookupToken(t *testing.T) { }) } +func TestRepository_lookupUpser(t *testing.T) { + ctx := context.Background() + s, err := cachedb.Open(ctx) + require.NoError(t, err) + + addr := "address" + at := &authtokens.AuthToken{ + Id: "at_1", + Token: "at_1_token", + UserId: "u_1", + } + kt := KeyringToken{ + TokenName: "t1", + KeyringType: "k1", + AuthTokenId: at.Id, + } + + boundaryAuthTokens := []*authtokens.AuthToken{at} + atMap := map[ringToken]*authtokens.AuthToken{ + {kt.KeyringType, kt.TokenName}: at, + } + r, err := NewRepository(ctx, s, &sync.Map{}, mapBasedAuthTokenKeyringLookup(atMap), sliceBasedAuthTokenBoundaryReader(boundaryAuthTokens)) + require.NoError(t, err) + assert.NoError(t, r.AddKeyringToken(ctx, addr, kt)) + + t.Run("empty user id", func(t *testing.T) { + u, err := r.lookupUser(ctx, "") + assert.ErrorContains(t, err, "empty id") + assert.Nil(t, u) + }) + t.Run("not found user id", func(t *testing.T) { + u, err := r.lookupUser(ctx, "notfound") + assert.NoError(t, err) + assert.Nil(t, u) + }) + t.Run("found", func(t *testing.T) { + u, err := r.lookupUser(ctx, at.UserId) + assert.NoError(t, err) + assert.Equal(t, &user{Id: at.UserId, Address: addr}, u) + }) +} + func TestRepository_RemoveStaleTokens(t *testing.T) { ctx := context.Background() s, err := cachedb.Open(ctx) diff --git a/internal/clientcache/internal/daemon/search_handler.go b/internal/clientcache/internal/daemon/search_handler.go index 559e1ccee..819c36141 100644 --- a/internal/clientcache/internal/daemon/search_handler.go +++ b/internal/clientcache/internal/daemon/search_handler.go @@ -13,6 +13,7 @@ import ( "github.com/hashicorp/boundary/api/targets" "github.com/hashicorp/boundary/internal/clientcache/internal/cache" "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/event" "github.com/hashicorp/boundary/internal/util" ) @@ -31,11 +32,13 @@ const ( authTokenIdKey = "auth_token_id" ) -func newSearchHandlerFunc(ctx context.Context, repo *cache.Repository) (http.HandlerFunc, error) { +func newSearchHandlerFunc(ctx context.Context, repo *cache.Repository, refreshService *cache.RefreshService) (http.HandlerFunc, error) { const op = "daemon.newSearchHandlerFunc" switch { case util.IsNil(repo): return nil, errors.New(ctx, errors.InvalidParameter, op, "repository is missing") + case util.IsNil(refreshService): + return nil, errors.New(ctx, errors.InvalidParameter, op, "refresh service is missing") } s, err := cache.NewSearchService(ctx, repo) @@ -67,6 +70,14 @@ func newSearchHandlerFunc(ctx context.Context, repo *cache.Repository) (http.Han return } + // 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. + if err := refreshService.RefreshForSearch(ctx, authTokenId, searchableResource); err != nil { + // we don't stop the search, we just log that the inline refresh failed + event.WriteError(ctx, op, err, event.WithInfoMsg("when refreshing the resources inline for search", "auth_token_id", authTokenId, "resource", searchableResource)) + } + query := r.URL.Query().Get(queryKey) filter := r.URL.Query().Get(filterKey) diff --git a/internal/clientcache/internal/daemon/server.go b/internal/clientcache/internal/daemon/server.go index f28488bab..292d0f6af 100644 --- a/internal/clientcache/internal/daemon/server.go +++ b/internal/clientcache/internal/daemon/server.go @@ -31,6 +31,16 @@ import ( "github.com/hashicorp/go-secure-stdlib/parseutil" ) +const ( + // DefaultSearchStaleness is the default amount of time that can pass before + // a search request causes a blocking refresh to happen for the resources + // being searched for. + DefaultSearchStaleness = 60 * time.Second + // DefaultSearchRefreshTimeout is the default amount of time that a refresh + // that is blocking a search request can run before timing out. + DefaultSearchRefreshTimeout = 1 * time.Second +) + // Commander is an interface that provides a way to get an apiClient // and retrieve the keyring and token information used by a command. type Commander interface { @@ -68,6 +78,12 @@ type Config struct { 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 + // The maximum amount of time a refresh should block a search request from + // completeing before it times out. + MaxSearchRefreshTimeout time.Duration } func (sc *Config) validate(ctx context.Context) error { @@ -79,6 +95,12 @@ func (sc *Config) validate(ctx context.Context) error { return errors.New(ctx, errors.InvalidParameter, op, "missing contextCancel") case sc.DotDirectory == "": 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.MaxSearchStaleness < 0: + return errors.New(ctx, errors.InvalidParameter, op, "negative max search staleness") } return nil } @@ -196,8 +218,36 @@ func (s *CacheServer) Serve(ctx context.Context, cmd Commander, opt ...Option) e if s.store, s.storeUrl, err = openStore(ctx, s.conf.DatabaseUrl, s.conf.StoreDebug); err != nil { return errors.Wrap(ctx, err, op) } - s.info["Database URL"] = s.storeUrl - s.infoKeys = append(s.infoKeys, "Database URL") + if s.storeUrl != "" { + s.info["Database URL"] = s.storeUrl + s.infoKeys = append(s.infoKeys, "Database URL") + } + maxSearchRefreshTimeout := DefaultSearchRefreshTimeout + if s.conf.MaxSearchRefreshTimeout > 0 { + maxSearchRefreshTimeout = s.conf.MaxSearchRefreshTimeout + } + maxSearchStaleness := DefaultSearchStaleness + if s.conf.MaxSearchStaleness > 0 { + maxSearchStaleness = s.conf.MaxSearchStaleness + } + s.info["Max Search Staleness"] = maxSearchStaleness.String() + s.infoKeys = append(s.infoKeys, "Max Search Staleness") + s.info["Max Search Refresh Timeout"] = maxSearchRefreshTimeout.String() + 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.infoKeys = append(s.infoKeys, "Full Fetch Interval") + + ticOptions := []Option{} + if s.conf.RefreshInterval > 0 { + 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)) + } s.printInfo(ctx) @@ -206,17 +256,10 @@ func (s *CacheServer) Serve(ctx context.Context, cmd Commander, opt ...Option) e return errors.Wrap(ctx, err, op) } - refreshService, err := cache.NewRefreshService(ctx, repo) + refreshService, err := cache.NewRefreshService(ctx, repo, maxSearchStaleness, maxSearchRefreshTimeout) if err != nil { return errors.Wrap(ctx, err, op) } - ticOptions := []Option{} - if s.conf.RefreshInterval > 0 { - ticOptions = append(ticOptions, withRefreshInterval(ctx, s.conf.RefreshInterval)) - } - if s.conf.FullFetchInterval > 0 { - ticOptions = append(ticOptions, withFullFetchInterval(ctx, s.conf.FullFetchInterval)) - } tic, err := newRefreshTicker(ctx, refreshService, ticOptions...) if err != nil { return errors.Wrap(ctx, err, op) @@ -236,23 +279,23 @@ func (s *CacheServer) Serve(ctx context.Context, cmd Commander, opt ...Option) e }() mux := http.NewServeMux() - searchFn, err := newSearchHandlerFunc(ctx, repo) + searchFn, err := newSearchHandlerFunc(ctx, repo, refreshService) if err != nil { return errors.Wrap(ctx, err, op) } - mux.HandleFunc("/v1/search", searchFn) + mux.Handle("/v1/search", versionEnforcement(searchFn)) statusFn, err := newStatusHandlerFunc(ctx, repo, l.Addr().String()) if err != nil { return errors.Wrap(ctx, err, op) } - mux.HandleFunc("/v1/status", statusFn) + mux.Handle("/v1/status", versionEnforcement(statusFn)) tokenFn, err := newTokenHandlerFunc(ctx, repo, tic) if err != nil { return errors.Wrap(ctx, err, op) } - mux.HandleFunc("/v1/tokens", tokenFn) + mux.Handle("/v1/tokens", versionEnforcement(tokenFn)) stopFn, err := newStopHandlerFunc(ctx, s.conf.ContextCancel) if err != nil { diff --git a/internal/clientcache/internal/daemon/testing.go b/internal/clientcache/internal/daemon/testing.go index 112045d60..789fb9f24 100644 --- a/internal/clientcache/internal/daemon/testing.go +++ b/internal/clientcache/internal/daemon/testing.go @@ -34,8 +34,8 @@ func NewTestServer(t *testing.T, cmd Commander, opt ...Option) *TestServer { cfg := &Config{ ContextCancel: cancel, - RefreshInterval: defaultRefreshInterval, - FullFetchInterval: defaultFullFetchInterval, + RefreshInterval: DefaultRefreshInterval, + FullFetchInterval: DefaultFullFetchInterval, StoreDebug: opts.withDebug, LogWriter: io.Discard, DotDirectory: dotDir, @@ -101,7 +101,7 @@ func (s *TestServer) AddResources(t *testing.T, p *authtokens.AuthToken, tars [] } return sess, nil, "", nil } - rs, err := cache.NewRefreshService(ctx, r) + rs, err := cache.NewRefreshService(ctx, r, 0, 0) require.NoError(t, err) require.NoError(t, rs.Refresh(ctx, cache.WithTargetRetrievalFunc(tarFn), cache.WithSessionRetrievalFunc(sessFn))) } diff --git a/internal/clientcache/internal/daemon/ticker.go b/internal/clientcache/internal/daemon/ticker.go index 94637febe..e88809663 100644 --- a/internal/clientcache/internal/daemon/ticker.go +++ b/internal/clientcache/internal/daemon/ticker.go @@ -15,11 +15,13 @@ import ( ) const ( - DefaultRefreshIntervalSeconds = 60 - defaultRefreshInterval = DefaultRefreshIntervalSeconds * time.Second - - DefaultFullFetchIntervalSeconds = 60 * 5 - defaultFullFetchInterval = DefaultFullFetchIntervalSeconds * time.Second + // 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 defaultRandomizationFactor float64 = 0.2 ) @@ -58,12 +60,12 @@ func newRefreshTicker(ctx context.Context, refresh refreshService, opt ...Option return nil, errors.Wrap(ctx, err, op) } - refreshInterval := defaultRefreshInterval + refreshInterval := DefaultRefreshInterval if opts.withRefreshInterval != 0 { refreshInterval = opts.withRefreshInterval } - fullFetchInterval := defaultFullFetchInterval + fullFetchInterval := DefaultFullFetchInterval if opts.withFullFetchInterval != 0 { fullFetchInterval = opts.withFullFetchInterval } diff --git a/internal/clientcache/internal/daemon/token_handler.go b/internal/clientcache/internal/daemon/token_handler.go index 25d2d6bcb..7f58a0591 100644 --- a/internal/clientcache/internal/daemon/token_handler.go +++ b/internal/clientcache/internal/daemon/token_handler.go @@ -100,7 +100,7 @@ func newTokenHandlerFunc(ctx context.Context, repo *cache.Repository, refresher } } - tok, err := repo.LookupToken(ctx, perReq.AuthTokenId) + oldTok, err := repo.LookupToken(ctx, perReq.AuthTokenId) if err != nil { writeError(w, "error performing auth token lookup", http.StatusInternalServerError) return @@ -124,9 +124,15 @@ func newTokenHandlerFunc(ctx context.Context, repo *cache.Repository, refresher } } + newTok, err := repo.LookupToken(ctx, perReq.AuthTokenId) + if err != nil { + writeError(w, "error performing follow up auth token lookup", http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusNoContent) - if tok == nil { + if oldTok == nil && newTok != nil { refresher.refresh() } }, nil