Refresh specific resources before searching when they are stale (#4046)
validate versions for all handlers. Refresh specific user before searching Pass flags on to the background process Allow the search refresh timeout to be configured
This commit is contained in:
1 parent
b3d268eae5
commit
cdc6fb0973
19 files changed
+543
-100
No files matched your search
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
+99
-2
@@ -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
|
||||
|
||||
+172
-9
@@ -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}))
|
||||
|
||||
|
||||
+6
-3
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
|
||||
+14
-9
@@ -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))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)))
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user