Refactor the different repo methods into separate files (#2096)
* Refactor the different repo methods into separate files * Move DefaultLiveness to the repository.go file to be shared across workers and controllers.
This commit is contained in:
6 files changed
+387
-347
No files matched your search
@@ -1,15 +1,11 @@
|
||||
package servers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/hashicorp/boundary/globals"
|
||||
"github.com/hashicorp/boundary/internal/db"
|
||||
"github.com/hashicorp/boundary/internal/errors"
|
||||
"github.com/hashicorp/boundary/internal/kms"
|
||||
"github.com/hashicorp/boundary/internal/servers/store"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -50,272 +46,3 @@ func NewRepository(r db.Reader, w db.Writer, kms *kms.Kms) (*Repository, error)
|
||||
kms: kms,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// ListWorkers is a passthrough to listWorkersWithReader that uses the repo's normal reader.
|
||||
func (r *Repository) ListWorkers(ctx context.Context, opt ...Option) ([]*Worker, error) {
|
||||
return r.listWorkersWithReader(ctx, r.reader, opt...)
|
||||
}
|
||||
|
||||
// listWorkersWithReader will return a listing of resources and honor the WithLimit option or the repo
|
||||
// defaultLimit. It accepts a reader, allowing it to be used within a transaction or without.
|
||||
func (r *Repository) listWorkersWithReader(ctx context.Context, reader db.Reader, opt ...Option) ([]*Worker, error) {
|
||||
opts := getOpts(opt...)
|
||||
liveness := opts.withLiveness
|
||||
if liveness == 0 {
|
||||
liveness = DefaultLiveness
|
||||
}
|
||||
|
||||
var where string
|
||||
if liveness > 0 {
|
||||
where = fmt.Sprintf("update_time > now() - interval '%d seconds'", uint32(liveness.Seconds()))
|
||||
}
|
||||
|
||||
var workers []*Worker
|
||||
if err := reader.SearchWhere(
|
||||
ctx,
|
||||
&workers,
|
||||
where,
|
||||
[]interface{}{},
|
||||
db.WithLimit(-1),
|
||||
); err != nil {
|
||||
return nil, errors.Wrap(ctx, err, "workers.listWorkersWithReader")
|
||||
}
|
||||
|
||||
return workers, nil
|
||||
}
|
||||
|
||||
func (r *Repository) ListControllers(ctx context.Context, opt ...Option) ([]*store.Controller, error) {
|
||||
return r.listControllersWithReader(ctx, r.reader, opt...)
|
||||
}
|
||||
|
||||
// listControllersWithReader will return a listing of resources and honor the
|
||||
// WithLimit option or the repo defaultLimit. It accepts a reader, allowing it
|
||||
// to be used within a transaction or without.
|
||||
func (r *Repository) listControllersWithReader(ctx context.Context, reader db.Reader, opt ...Option) ([]*store.Controller, error) {
|
||||
opts := getOpts(opt...)
|
||||
liveness := opts.withLiveness
|
||||
if liveness == 0 {
|
||||
liveness = DefaultLiveness
|
||||
}
|
||||
|
||||
var where string
|
||||
if liveness > 0 {
|
||||
where = fmt.Sprintf("update_time > now() - interval '%d seconds'", uint32(liveness.Seconds()))
|
||||
}
|
||||
|
||||
var controllers []*store.Controller
|
||||
if err := reader.SearchWhere(
|
||||
ctx,
|
||||
&controllers,
|
||||
where,
|
||||
[]interface{}{},
|
||||
db.WithLimit(-1),
|
||||
); err != nil {
|
||||
return nil, errors.Wrap(ctx, err, "workers.listControllersWithReader")
|
||||
}
|
||||
|
||||
return controllers, nil
|
||||
}
|
||||
|
||||
// ListTagsForWorkers returns a map from the worker's id to the list of Tags
|
||||
// that are for that worker. All options are ignored.
|
||||
func (r *Repository) ListTagsForWorkers(ctx context.Context, workerIds []string, _ ...Option) (map[string][]*Tag, error) {
|
||||
const op = "servers.ListTagsForWorkers"
|
||||
var workerTags []*store.WorkerTag
|
||||
if err := r.reader.SearchWhere(
|
||||
ctx,
|
||||
&workerTags,
|
||||
"worker_id in (?)",
|
||||
[]interface{}{workerIds},
|
||||
db.WithLimit(-1),
|
||||
); err != nil {
|
||||
return nil, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("worker IDs %v", workerIds)))
|
||||
}
|
||||
|
||||
ret := make(map[string][]*Tag, len(workerIds))
|
||||
for _, t := range workerTags {
|
||||
ret[t.WorkerId] = append(ret[t.WorkerId], &Tag{Key: t.Key, Value: t.Value})
|
||||
}
|
||||
return ret, nil
|
||||
}
|
||||
|
||||
// UpsertWorker creates a new worker if one with the provided public id doesn't
|
||||
// already exist. If it does, UpsertWorker updates the worker. The
|
||||
// WithUpdateTags option is the only one used. All others are ignored.
|
||||
func (r *Repository) UpsertWorker(ctx context.Context, worker *Worker, opt ...Option) ([]*store.Controller, int, error) {
|
||||
const op = "servers.UpsertWorker"
|
||||
|
||||
switch {
|
||||
case worker == nil:
|
||||
return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "worker is nil")
|
||||
case worker.Address == "":
|
||||
return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "address is empty")
|
||||
case worker.ScopeId == "":
|
||||
return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "scope id is empty")
|
||||
case worker.PublicId == "":
|
||||
return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "public id is empty")
|
||||
}
|
||||
|
||||
opts := getOpts(opt...)
|
||||
|
||||
var rowsUpdated int64
|
||||
var controllers []*store.Controller
|
||||
_, err := r.writer.DoTx(
|
||||
ctx,
|
||||
db.StdRetryCnt,
|
||||
db.ExpBackoff{},
|
||||
func(read db.Reader, w db.Writer) error {
|
||||
var err error
|
||||
onConflict := &db.OnConflict{
|
||||
Target: db.Columns{"public_id"},
|
||||
Action: append(db.SetColumns([]string{"description", "address"}), db.SetColumnValues(map[string]interface{}{"update_time": "now()"})...),
|
||||
}
|
||||
err = w.Create(ctx, worker, db.WithOnConflict(onConflict), db.WithReturnRowsAffected(&rowsUpdated))
|
||||
if err != nil {
|
||||
return errors.Wrap(ctx, err, op+":Upsert")
|
||||
}
|
||||
|
||||
// Fetch current controllers to feed to the workers
|
||||
controllers, err = r.listControllersWithReader(ctx, read)
|
||||
if err != nil {
|
||||
return errors.Wrap(ctx, err, op+":ListController")
|
||||
}
|
||||
|
||||
// If we've been told to update tags, we need to clean out old
|
||||
// ones and add new ones. Within the current transaction, simply
|
||||
// delete all tags for the given worker, then add the new ones
|
||||
// we've been sent.
|
||||
if opts.withUpdateTags {
|
||||
_, err = w.Delete(ctx, &store.WorkerTag{}, db.WithWhere(deleteConfigTagsSql, worker.GetPublicId()))
|
||||
if err != nil {
|
||||
return errors.Wrap(ctx, err, op+":DeleteTags", errors.WithMsg(worker.GetPublicId()))
|
||||
}
|
||||
|
||||
// If tags were cleared out entirely, then we'll have nothing
|
||||
// to do here, e.g., it will result in deletion of all tags.
|
||||
// Otherwise, go through and stage each tuple for insertion
|
||||
// below.
|
||||
if len(worker.Tags) > 0 {
|
||||
tags := make([]interface{}, 0, len(worker.Tags))
|
||||
for _, v := range worker.Tags {
|
||||
if v == nil {
|
||||
return errors.New(ctx, errors.InvalidParameter, op+":RangeTags", fmt.Sprintf("found nil tag value for worker %s", worker.GetPublicId()))
|
||||
}
|
||||
tags = append(tags, &store.WorkerTag{
|
||||
WorkerId: worker.GetPublicId(),
|
||||
Key: v.Key,
|
||||
Value: v.Value,
|
||||
Source: string(ConfigurationTagSource),
|
||||
})
|
||||
}
|
||||
if err = w.CreateItems(ctx, tags); err != nil {
|
||||
return errors.Wrap(ctx, err, op+":CreateTags", errors.WithMsg(worker.GetPublicId()))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
},
|
||||
)
|
||||
if err != nil {
|
||||
return nil, db.NoRowsAffected, err
|
||||
}
|
||||
|
||||
return controllers, int(rowsUpdated), nil
|
||||
}
|
||||
|
||||
func (r *Repository) UpsertController(ctx context.Context, controller *store.Controller) (int, error) {
|
||||
const op = "servers.UpsertController"
|
||||
|
||||
if controller == nil {
|
||||
return db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "controller is nil")
|
||||
}
|
||||
|
||||
var rowsUpdated int64
|
||||
_, err := r.writer.DoTx(
|
||||
ctx,
|
||||
db.StdRetryCnt,
|
||||
db.ExpBackoff{},
|
||||
func(read db.Reader, w db.Writer) error {
|
||||
var err error
|
||||
onConflict := &db.OnConflict{
|
||||
Target: db.Columns{"private_id"},
|
||||
Action: append(db.SetColumns([]string{"description", "address"}), db.SetColumnValues(map[string]interface{}{"update_time": "now()"})...),
|
||||
}
|
||||
err = w.Create(ctx, controller, db.WithOnConflict(onConflict), db.WithReturnRowsAffected(&rowsUpdated))
|
||||
if err != nil {
|
||||
return errors.Wrap(ctx, err, op+":Upsert")
|
||||
}
|
||||
|
||||
return nil
|
||||
},
|
||||
)
|
||||
if err != nil {
|
||||
return db.NoRowsAffected, err
|
||||
}
|
||||
|
||||
return int(rowsUpdated), nil
|
||||
}
|
||||
|
||||
type Nonce struct {
|
||||
Nonce string
|
||||
Purpose string
|
||||
}
|
||||
|
||||
// TableName returns the table name.
|
||||
func (n *Nonce) TableName() string {
|
||||
return "nonce"
|
||||
}
|
||||
|
||||
const (
|
||||
NoncePurposeRecovery = "recovery"
|
||||
NoncePurposeWorkerAuth = "worker-auth"
|
||||
)
|
||||
|
||||
// AddNonce adds a nonce
|
||||
func (r *Repository) AddNonce(ctx context.Context, nonce, purpose string, opt ...Option) error {
|
||||
const op = "servers.AddNonce"
|
||||
if nonce == "" {
|
||||
return errors.New(ctx, errors.InvalidParameter, op, "empty nonce")
|
||||
}
|
||||
switch purpose {
|
||||
case NoncePurposeRecovery, NoncePurposeWorkerAuth:
|
||||
case "":
|
||||
return errors.New(ctx, errors.InvalidParameter, op, "empty nonce purpose")
|
||||
default:
|
||||
return errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("unknown nonce purpose %q", purpose))
|
||||
}
|
||||
if err := r.writer.Create(ctx, &Nonce{
|
||||
Nonce: nonce,
|
||||
Purpose: purpose,
|
||||
}); err != nil {
|
||||
return errors.Wrap(ctx, err, op+":Insertion")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// CleanupNonces removes nonces that no longer need to be stored
|
||||
func (r *Repository) CleanupNonces(ctx context.Context, opt ...Option) (int, error) {
|
||||
// Use the largest validity period out of the various nonces we're looking at
|
||||
maxDuration := globals.RecoveryTokenValidityPeriod
|
||||
if globals.WorkerAuthNonceValidityPeriod > maxDuration {
|
||||
maxDuration = globals.WorkerAuthNonceValidityPeriod
|
||||
}
|
||||
// If something was inserted before 3x the actual validity period, clean it out
|
||||
endTime := time.Now().Add(-3 * maxDuration)
|
||||
|
||||
rows, err := r.writer.Delete(ctx, &Nonce{}, db.WithWhere(deleteWhereCreateTimeSql, endTime))
|
||||
if err != nil {
|
||||
return db.NoRowsAffected, errors.Wrap(ctx, err, "servers.CleanupNonces")
|
||||
}
|
||||
return rows, nil
|
||||
}
|
||||
|
||||
// ListNonces lists nonces. Used only for tests at the moment.
|
||||
func (r *Repository) ListNonces(ctx context.Context, purpose string, opt ...Option) ([]*Nonce, error) {
|
||||
var nonces []*Nonce
|
||||
if err := r.reader.SearchWhere(ctx, &nonces, "purpose = ?", []interface{}{purpose}, db.WithLimit(-1)); err != nil {
|
||||
return nil, errors.Wrap(ctx, err, "servers.ListNonces")
|
||||
}
|
||||
return nonces, nil
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
package servers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/hashicorp/boundary/internal/db"
|
||||
"github.com/hashicorp/boundary/internal/errors"
|
||||
"github.com/hashicorp/boundary/internal/servers/store"
|
||||
)
|
||||
|
||||
func (r *Repository) ListControllers(ctx context.Context, opt ...Option) ([]*store.Controller, error) {
|
||||
return r.listControllersWithReader(ctx, r.reader, opt...)
|
||||
}
|
||||
|
||||
// listControllersWithReader will return a listing of resources and honor the
|
||||
// WithLimit option or the repo defaultLimit. It accepts a reader, allowing it
|
||||
// to be used within a transaction or without.
|
||||
func (r *Repository) listControllersWithReader(ctx context.Context, reader db.Reader, opt ...Option) ([]*store.Controller, error) {
|
||||
opts := getOpts(opt...)
|
||||
liveness := opts.withLiveness
|
||||
if liveness == 0 {
|
||||
liveness = DefaultLiveness
|
||||
}
|
||||
|
||||
var where string
|
||||
if liveness > 0 {
|
||||
where = fmt.Sprintf("update_time > now() - interval '%d seconds'", uint32(liveness.Seconds()))
|
||||
}
|
||||
|
||||
var controllers []*store.Controller
|
||||
if err := reader.SearchWhere(
|
||||
ctx,
|
||||
&controllers,
|
||||
where,
|
||||
[]interface{}{},
|
||||
db.WithLimit(-1),
|
||||
); err != nil {
|
||||
return nil, errors.Wrap(ctx, err, "workers.listControllersWithReader")
|
||||
}
|
||||
|
||||
return controllers, nil
|
||||
}
|
||||
|
||||
func (r *Repository) UpsertController(ctx context.Context, controller *store.Controller) (int, error) {
|
||||
const op = "servers.UpsertController"
|
||||
|
||||
if controller == nil {
|
||||
return db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "controller is nil")
|
||||
}
|
||||
|
||||
var rowsUpdated int64
|
||||
_, err := r.writer.DoTx(
|
||||
ctx,
|
||||
db.StdRetryCnt,
|
||||
db.ExpBackoff{},
|
||||
func(read db.Reader, w db.Writer) error {
|
||||
var err error
|
||||
onConflict := &db.OnConflict{
|
||||
Target: db.Columns{"private_id"},
|
||||
Action: append(db.SetColumns([]string{"description", "address"}), db.SetColumnValues(map[string]interface{}{"update_time": "now()"})...),
|
||||
}
|
||||
err = w.Create(ctx, controller, db.WithOnConflict(onConflict), db.WithReturnRowsAffected(&rowsUpdated))
|
||||
if err != nil {
|
||||
return errors.Wrap(ctx, err, op+":Upsert")
|
||||
}
|
||||
|
||||
return nil
|
||||
},
|
||||
)
|
||||
if err != nil {
|
||||
return db.NoRowsAffected, err
|
||||
}
|
||||
|
||||
return int(rowsUpdated), nil
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
package servers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/hashicorp/boundary/globals"
|
||||
"github.com/hashicorp/boundary/internal/db"
|
||||
"github.com/hashicorp/boundary/internal/errors"
|
||||
)
|
||||
|
||||
type Nonce struct {
|
||||
Nonce string
|
||||
Purpose string
|
||||
}
|
||||
|
||||
// TableName returns the table name.
|
||||
func (n *Nonce) TableName() string {
|
||||
return "nonce"
|
||||
}
|
||||
|
||||
const (
|
||||
NoncePurposeRecovery = "recovery"
|
||||
NoncePurposeWorkerAuth = "worker-auth"
|
||||
)
|
||||
|
||||
// AddNonce adds a nonce
|
||||
func (r *Repository) AddNonce(ctx context.Context, nonce, purpose string, opt ...Option) error {
|
||||
const op = "servers.AddNonce"
|
||||
if nonce == "" {
|
||||
return errors.New(ctx, errors.InvalidParameter, op, "empty nonce")
|
||||
}
|
||||
switch purpose {
|
||||
case NoncePurposeRecovery, NoncePurposeWorkerAuth:
|
||||
case "":
|
||||
return errors.New(ctx, errors.InvalidParameter, op, "empty nonce purpose")
|
||||
default:
|
||||
return errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("unknown nonce purpose %q", purpose))
|
||||
}
|
||||
if err := r.writer.Create(ctx, &Nonce{
|
||||
Nonce: nonce,
|
||||
Purpose: purpose,
|
||||
}); err != nil {
|
||||
return errors.Wrap(ctx, err, op+":Insertion")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// CleanupNonces removes nonces that no longer need to be stored
|
||||
func (r *Repository) CleanupNonces(ctx context.Context, opt ...Option) (int, error) {
|
||||
// Use the largest validity period out of the various nonces we're looking at
|
||||
maxDuration := globals.RecoveryTokenValidityPeriod
|
||||
if globals.WorkerAuthNonceValidityPeriod > maxDuration {
|
||||
maxDuration = globals.WorkerAuthNonceValidityPeriod
|
||||
}
|
||||
// If something was inserted before 3x the actual validity period, clean it out
|
||||
endTime := time.Now().Add(-3 * maxDuration)
|
||||
|
||||
rows, err := r.writer.Delete(ctx, &Nonce{}, db.WithWhere(deleteWhereCreateTimeSql, endTime))
|
||||
if err != nil {
|
||||
return db.NoRowsAffected, errors.Wrap(ctx, err, "servers.CleanupNonces")
|
||||
}
|
||||
return rows, nil
|
||||
}
|
||||
|
||||
// ListNonces lists nonces. Used only for tests at the moment.
|
||||
func (r *Repository) ListNonces(ctx context.Context, purpose string, opt ...Option) ([]*Nonce, error) {
|
||||
var nonces []*Nonce
|
||||
if err := r.reader.SearchWhere(ctx, &nonces, "purpose = ?", []interface{}{purpose}, db.WithLimit(-1)); err != nil {
|
||||
return nil, errors.Wrap(ctx, err, "servers.ListNonces")
|
||||
}
|
||||
return nonces, nil
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
package servers_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/hashicorp/boundary/api/recovery"
|
||||
"github.com/hashicorp/boundary/api/roles"
|
||||
"github.com/hashicorp/boundary/globals"
|
||||
"github.com/hashicorp/boundary/internal/daemon/controller"
|
||||
"github.com/hashicorp/boundary/internal/db"
|
||||
"github.com/hashicorp/boundary/internal/servers"
|
||||
"github.com/hashicorp/boundary/internal/types/scope"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestRecoveryNonces(t *testing.T) {
|
||||
assert, require := assert.New(t), require.New(t)
|
||||
|
||||
// Set these low so that we can not have the test run forever
|
||||
globals.RecoveryTokenValidityPeriod = 10 * time.Second
|
||||
globals.WorkerAuthNonceValidityPeriod = 10 * time.Second
|
||||
controller.NonceCleanupInterval = 20 * time.Second
|
||||
|
||||
wrapper := db.TestWrapper(t)
|
||||
tc := controller.NewTestController(t, &controller.TestControllerOpts{
|
||||
RecoveryKms: wrapper,
|
||||
})
|
||||
defer tc.Shutdown()
|
||||
|
||||
externalWrappers := tc.Kms().GetExternalWrappers(context.Background())
|
||||
externalRecovery := externalWrappers.Recovery()
|
||||
require.Equal(externalRecovery, wrapper)
|
||||
|
||||
client := tc.Client()
|
||||
repo := tc.ServersRepo()
|
||||
|
||||
// First, validate that we can't use the same token twice. Get two tokens,
|
||||
// try to use the first twice (should succeed first time, fail second), then
|
||||
// ensure second token works
|
||||
token1, err := recovery.GenerateRecoveryToken(tc.Context(), wrapper)
|
||||
require.NoError(err)
|
||||
token2, err := recovery.GenerateRecoveryToken(tc.Context(), wrapper)
|
||||
require.NoError(err)
|
||||
|
||||
// Token 1, try 1
|
||||
client.SetToken(token1)
|
||||
roleClient := roles.NewClient(client)
|
||||
_, err = roleClient.Create(tc.Context(), scope.Global.String())
|
||||
require.NoError(err)
|
||||
nonces, err := repo.ListNonces(tc.Context(), servers.NoncePurposeRecovery)
|
||||
require.NoError(err)
|
||||
assert.Len(nonces, 1)
|
||||
|
||||
// Token 1, try 2
|
||||
_, err = roleClient.Create(tc.Context(), scope.Global.String())
|
||||
require.Error(err)
|
||||
nonces, err = repo.ListNonces(tc.Context(), servers.NoncePurposeRecovery)
|
||||
require.NoError(err)
|
||||
assert.Len(nonces, 1)
|
||||
|
||||
// Token 2
|
||||
roleClient.ApiClient().SetToken(token2)
|
||||
_, err = roleClient.Create(tc.Context(), scope.Global.String())
|
||||
require.NoError(err)
|
||||
nonces, err = repo.ListNonces(tc.Context(), servers.NoncePurposeRecovery)
|
||||
require.NoError(err)
|
||||
assert.Len(nonces, 2)
|
||||
|
||||
// Make sure they get cleaned up
|
||||
time.Sleep(2 * controller.NonceCleanupInterval)
|
||||
nonces, err = repo.ListNonces(tc.Context(), servers.NoncePurposeRecovery)
|
||||
require.NoError(err)
|
||||
assert.Len(nonces, 0)
|
||||
|
||||
// And finally, make sure they still can't be used
|
||||
for _, token := range []string{token1, token2} {
|
||||
roleClient.ApiClient().SetToken(token)
|
||||
_, err = roleClient.Create(tc.Context(), scope.Global.String())
|
||||
require.Error(err)
|
||||
nonces, err = repo.ListNonces(tc.Context(), servers.NoncePurposeRecovery)
|
||||
require.NoError(err)
|
||||
assert.Len(nonces, 0)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,150 @@
|
||||
package servers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/hashicorp/boundary/internal/db"
|
||||
"github.com/hashicorp/boundary/internal/errors"
|
||||
"github.com/hashicorp/boundary/internal/servers/store"
|
||||
)
|
||||
|
||||
// ListWorkers is a passthrough to listWorkersWithReader that uses the repo's normal reader.
|
||||
func (r *Repository) ListWorkers(ctx context.Context, opt ...Option) ([]*Worker, error) {
|
||||
return r.listWorkersWithReader(ctx, r.reader, opt...)
|
||||
}
|
||||
|
||||
// listWorkersWithReader will return a listing of resources and honor the WithLimit option or the repo
|
||||
// defaultLimit. It accepts a reader, allowing it to be used within a transaction or without.
|
||||
func (r *Repository) listWorkersWithReader(ctx context.Context, reader db.Reader, opt ...Option) ([]*Worker, error) {
|
||||
opts := getOpts(opt...)
|
||||
liveness := opts.withLiveness
|
||||
if liveness == 0 {
|
||||
liveness = DefaultLiveness
|
||||
}
|
||||
|
||||
var where string
|
||||
if liveness > 0 {
|
||||
where = fmt.Sprintf("update_time > now() - interval '%d seconds'", uint32(liveness.Seconds()))
|
||||
}
|
||||
|
||||
var workers []*Worker
|
||||
if err := reader.SearchWhere(
|
||||
ctx,
|
||||
&workers,
|
||||
where,
|
||||
[]interface{}{},
|
||||
db.WithLimit(-1),
|
||||
); err != nil {
|
||||
return nil, errors.Wrap(ctx, err, "workers.listWorkersWithReader")
|
||||
}
|
||||
|
||||
return workers, nil
|
||||
}
|
||||
|
||||
// ListTagsForWorkers returns a map from the worker's id to the list of Tags
|
||||
// that are for that worker. All options are ignored.
|
||||
func (r *Repository) ListTagsForWorkers(ctx context.Context, workerIds []string, _ ...Option) (map[string][]*Tag, error) {
|
||||
const op = "servers.ListTagsForWorkers"
|
||||
var workerTags []*store.WorkerTag
|
||||
if err := r.reader.SearchWhere(
|
||||
ctx,
|
||||
&workerTags,
|
||||
"worker_id in (?)",
|
||||
[]interface{}{workerIds},
|
||||
db.WithLimit(-1),
|
||||
); err != nil {
|
||||
return nil, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("worker IDs %v", workerIds)))
|
||||
}
|
||||
|
||||
ret := make(map[string][]*Tag, len(workerIds))
|
||||
for _, t := range workerTags {
|
||||
ret[t.WorkerId] = append(ret[t.WorkerId], &Tag{Key: t.Key, Value: t.Value})
|
||||
}
|
||||
return ret, nil
|
||||
}
|
||||
|
||||
// UpsertWorker creates a new worker if one with the provided public id doesn't
|
||||
// already exist. If it does, UpsertWorker updates the worker. The
|
||||
// WithUpdateTags option is the only one used. All others are ignored.
|
||||
func (r *Repository) UpsertWorker(ctx context.Context, worker *Worker, opt ...Option) ([]*store.Controller, int, error) {
|
||||
const op = "servers.UpsertWorker"
|
||||
|
||||
switch {
|
||||
case worker == nil:
|
||||
return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "worker is nil")
|
||||
case worker.Address == "":
|
||||
return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "address is empty")
|
||||
case worker.ScopeId == "":
|
||||
return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "scope id is empty")
|
||||
case worker.PublicId == "":
|
||||
return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "public id is empty")
|
||||
}
|
||||
|
||||
opts := getOpts(opt...)
|
||||
|
||||
var rowsUpdated int64
|
||||
var controllers []*store.Controller
|
||||
_, err := r.writer.DoTx(
|
||||
ctx,
|
||||
db.StdRetryCnt,
|
||||
db.ExpBackoff{},
|
||||
func(read db.Reader, w db.Writer) error {
|
||||
var err error
|
||||
onConflict := &db.OnConflict{
|
||||
Target: db.Columns{"public_id"},
|
||||
Action: append(db.SetColumns([]string{"description", "address"}), db.SetColumnValues(map[string]interface{}{"update_time": "now()"})...),
|
||||
}
|
||||
err = w.Create(ctx, worker, db.WithOnConflict(onConflict), db.WithReturnRowsAffected(&rowsUpdated))
|
||||
if err != nil {
|
||||
return errors.Wrap(ctx, err, op+":Upsert")
|
||||
}
|
||||
|
||||
// Fetch current controllers to feed to the workers
|
||||
controllers, err = r.listControllersWithReader(ctx, read)
|
||||
if err != nil {
|
||||
return errors.Wrap(ctx, err, op+":ListController")
|
||||
}
|
||||
|
||||
// If we've been told to update tags, we need to clean out old
|
||||
// ones and add new ones. Within the current transaction, simply
|
||||
// delete all tags for the given worker, then add the new ones
|
||||
// we've been sent.
|
||||
if opts.withUpdateTags {
|
||||
_, err = w.Delete(ctx, &store.WorkerTag{}, db.WithWhere(deleteConfigTagsSql, worker.GetPublicId()))
|
||||
if err != nil {
|
||||
return errors.Wrap(ctx, err, op+":DeleteTags", errors.WithMsg(worker.GetPublicId()))
|
||||
}
|
||||
|
||||
// If tags were cleared out entirely, then we'll have nothing
|
||||
// to do here, e.g., it will result in deletion of all tags.
|
||||
// Otherwise, go through and stage each tuple for insertion
|
||||
// below.
|
||||
if len(worker.Tags) > 0 {
|
||||
tags := make([]interface{}, 0, len(worker.Tags))
|
||||
for _, v := range worker.Tags {
|
||||
if v == nil {
|
||||
return errors.New(ctx, errors.InvalidParameter, op+":RangeTags", fmt.Sprintf("found nil tag value for worker %s", worker.GetPublicId()))
|
||||
}
|
||||
tags = append(tags, &store.WorkerTag{
|
||||
WorkerId: worker.GetPublicId(),
|
||||
Key: v.Key,
|
||||
Value: v.Value,
|
||||
Source: string(ConfigurationTagSource),
|
||||
})
|
||||
}
|
||||
if err = w.CreateItems(ctx, tags); err != nil {
|
||||
return errors.Wrap(ctx, err, op+":CreateTags", errors.WithMsg(worker.GetPublicId()))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
},
|
||||
)
|
||||
if err != nil {
|
||||
return nil, db.NoRowsAffected, err
|
||||
}
|
||||
|
||||
return controllers, int(rowsUpdated), nil
|
||||
}
|
||||
@@ -5,88 +5,14 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/hashicorp/boundary/api/recovery"
|
||||
"github.com/hashicorp/boundary/api/roles"
|
||||
"github.com/hashicorp/boundary/globals"
|
||||
"github.com/hashicorp/boundary/internal/daemon/controller"
|
||||
"github.com/hashicorp/boundary/internal/db"
|
||||
"github.com/hashicorp/boundary/internal/kms"
|
||||
"github.com/hashicorp/boundary/internal/servers"
|
||||
"github.com/hashicorp/boundary/internal/types/scope"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestRecoveryNonces(t *testing.T) {
|
||||
assert, require := assert.New(t), require.New(t)
|
||||
|
||||
// Set these low so that we can not have the test run forever
|
||||
globals.RecoveryTokenValidityPeriod = 10 * time.Second
|
||||
globals.WorkerAuthNonceValidityPeriod = 10 * time.Second
|
||||
controller.NonceCleanupInterval = 20 * time.Second
|
||||
|
||||
wrapper := db.TestWrapper(t)
|
||||
tc := controller.NewTestController(t, &controller.TestControllerOpts{
|
||||
RecoveryKms: wrapper,
|
||||
})
|
||||
defer tc.Shutdown()
|
||||
|
||||
externalWrappers := tc.Kms().GetExternalWrappers(context.Background())
|
||||
externalRecovery := externalWrappers.Recovery()
|
||||
require.Equal(externalRecovery, wrapper)
|
||||
|
||||
client := tc.Client()
|
||||
repo := tc.ServersRepo()
|
||||
|
||||
// First, validate that we can't use the same token twice. Get two tokens,
|
||||
// try to use the first twice (should succeed first time, fail second), then
|
||||
// ensure second token works
|
||||
token1, err := recovery.GenerateRecoveryToken(tc.Context(), wrapper)
|
||||
require.NoError(err)
|
||||
token2, err := recovery.GenerateRecoveryToken(tc.Context(), wrapper)
|
||||
require.NoError(err)
|
||||
|
||||
// Token 1, try 1
|
||||
client.SetToken(token1)
|
||||
roleClient := roles.NewClient(client)
|
||||
_, err = roleClient.Create(tc.Context(), scope.Global.String())
|
||||
require.NoError(err)
|
||||
nonces, err := repo.ListNonces(tc.Context(), servers.NoncePurposeRecovery)
|
||||
require.NoError(err)
|
||||
assert.Len(nonces, 1)
|
||||
|
||||
// Token 1, try 2
|
||||
_, err = roleClient.Create(tc.Context(), scope.Global.String())
|
||||
require.Error(err)
|
||||
nonces, err = repo.ListNonces(tc.Context(), servers.NoncePurposeRecovery)
|
||||
require.NoError(err)
|
||||
assert.Len(nonces, 1)
|
||||
|
||||
// Token 2
|
||||
roleClient.ApiClient().SetToken(token2)
|
||||
_, err = roleClient.Create(tc.Context(), scope.Global.String())
|
||||
require.NoError(err)
|
||||
nonces, err = repo.ListNonces(tc.Context(), servers.NoncePurposeRecovery)
|
||||
require.NoError(err)
|
||||
assert.Len(nonces, 2)
|
||||
|
||||
// Make sure they get cleaned up
|
||||
time.Sleep(2 * controller.NonceCleanupInterval)
|
||||
nonces, err = repo.ListNonces(tc.Context(), servers.NoncePurposeRecovery)
|
||||
require.NoError(err)
|
||||
assert.Len(nonces, 0)
|
||||
|
||||
// And finally, make sure they still can't be used
|
||||
for _, token := range []string{token1, token2} {
|
||||
roleClient.ApiClient().SetToken(token)
|
||||
_, err = roleClient.Create(tc.Context(), scope.Global.String())
|
||||
require.Error(err)
|
||||
nonces, err = repo.ListNonces(tc.Context(), servers.NoncePurposeRecovery)
|
||||
require.NoError(err)
|
||||
assert.Len(nonces, 0)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTagUpdatingListing(t *testing.T) {
|
||||
require := require.New(t)
|
||||
|
||||
Reference in new issue
Block a user