CyberCoveLLC/threshold
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

internal/host/plugin: ensure UpdateCatalog triggers updates for sets (#1736)

Browse Source 
* internal/host/plugin: ensure UpdateCatalog triggers updates for sets

This ensures that attribute updates to a host catalog triggers an update
to any of its downstream host sets. This ensures that updates to
elements that affect a host set, such as the region a host catalog is
configured for, take effect immediately instead of requiring a wait of
long as 10m after an update.

* Add WithNextRunIn to test sync job

This is to ensure it doesn't run right away when added. This ensures
that we can accurately test *when* the job is set to run immediately.

* Update internal/host/plugin/repository_host_catalog.go

Co-authored-by: Todd <T.Alan.Knight@gmail.com>

* Change TestRepository_UpdateCatalog_SyncSets to check next run time, not wait for run

* Remove explicit version increment and instead rely on implicit increment

* Fix lanuage for rows updated check when flagging host sets for sync

* remove errant non-zero check when looping over the thing we're checking

* remove scheduler start line from TestRepository_UpdateCatalog_SyncSets

Co-authored-by: Todd <T.Alan.Knight@gmail.com>
This commit is contained in:
Chris MarchesiandTodd authored and GitHub committed 2021-11-30 14:37:06 -08:00
1 parent 9b6d7d8105
commit 84febd9983
2 files changed
+164 -1
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/host/plugin/repository_host_catalog.go
+35 -1
View File
@@ -286,12 +286,16 @@ func (r *Repository) UpdateCatalog(ctx context.Context, c *HostCatalog, version
}
}
var needSetSync, runSyncJob bool
if updateAttributes {
dbMask = append(dbMask, "attributes")
newCatalog.Attributes, err = patchstruct.PatchBytes(newCatalog.Attributes, c.Attributes)
if err != nil {
return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithMsg("error in catalog attribute JSON"))
}
// Flag for host sets under this catalog to be synced.
needSetSync = true
}
// Get the plugin for the host catalog - this is to return it back
@@ -454,6 +458,32 @@ func (r *Repository) UpdateCatalog(ctx context.Context, c *HostCatalog, version
recordUpdated = true
}
if needSetSync {
// We also need to mark all host sets in this catalog to be
// synced as well.
setsForCatalog, _, err := r.getSets(ctx, "", returnedCatalog.PublicId)
if err != nil {
return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get sets for host catalog"))
}
for _, set := range setsForCatalog {
newSet := set.clone()
newSet.NeedSync = true
var msg oplog.Message
n, err := w.Update(ctx, newSet, []string{"NeedSync"}, []string{}, db.NewOplogMsg(&msg))
if err != nil {
return errors.Wrap(ctx, err, op, errors.WithMsg("unable to update host set"))
}
if n > 1 {
return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("expected no more than 1 host set to be updated while flagging host set id %q for synchronization, got %d", newSet.PublicId, n))
}
msgs = append(msgs, &msg)
runSyncJob = true
}
}
if len(msgs) != 0 {
metadata := newCatalog.oplog(oplog.OpType_OP_TYPE_UPDATE)
if err := w.WriteOplogEntryWith(ctx, oplogWrapper, ticket, metadata, msgs); err != nil {
@@ -472,8 +502,12 @@ func (r *Repository) UpdateCatalog(ctx context.Context, c *HostCatalog, version
return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("in %s", newCatalog.PublicId)))
}
if runSyncJob {
_ = r.scheduler.UpdateJobNextRunInAtLeast(ctx, setSyncJobName, 0)
}
// Even if we didn't update any records, if we were able to find the record
// with the appropriate version, returning 1 row updated is consistant with
// with the appropriate version, returning 1 row updated is consistent with
// static's update catalog behavior.
numUpdated := 1
internal/host/plugin/repository_host_catalog_test.go
+129
View File
@@ -9,6 +9,7 @@ import (
"github.com/google/go-cmp/cmp"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/db/timestamp"
"github.com/hashicorp/boundary/internal/errors"
"github.com/hashicorp/boundary/internal/host/plugin/store"
"github.com/hashicorp/boundary/internal/iam"
@@ -17,6 +18,7 @@ import (
hostplg "github.com/hashicorp/boundary/internal/plugin/host"
hostplugin "github.com/hashicorp/boundary/internal/plugin/host"
"github.com/hashicorp/boundary/internal/scheduler"
"github.com/hashicorp/boundary/internal/scheduler/job"
plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -1441,6 +1443,133 @@ func TestRepository_DeleteCatalogX(t *testing.T) {
}
}
// Create a test scheduled job that we can use to catch when a run is executed.
// We will use this to track whether or not an update was sent in
// TestRepository_UpdateCatalog_SyncSets.
type testSyncJob struct{}
func (j *testSyncJob) Status() scheduler.JobStatus {
return scheduler.JobStatus{
Completed: 0,
Total: 1,
}
}
func (j *testSyncJob) Run(_ context.Context) error { return nil }
func (j *testSyncJob) NextRunIn() (time.Duration, error) { return setSyncJobRunInterval, nil }
func (j *testSyncJob) Name() string { return setSyncJobName }
func (j *testSyncJob) Description() string { return setSyncJobName }
func TestRepository_UpdateCatalog_SyncSets(t *testing.T) {
ctx := context.Background()
dbConn, _ := db.TestSetup(t, "postgres")
dbRW := db.New(dbConn)
dbWrapper := db.TestWrapper(t)
sched := scheduler.TestScheduler(t, dbConn, dbWrapper)
dbKmsCache := kms.TestKms(t, dbConn, dbWrapper)
_, projectScope := iam.TestScopes(t, iam.TestRepo(t, dbConn, dbWrapper))
testPlugin := hostplg.TestPlugin(t, dbConn, "test")
dummyPluginMap := map[string]plgpb.HostPluginServiceClient{
testPlugin.GetPublicId(): &WrappingPluginClient{Server: &plgpb.UnimplementedHostPluginServiceServer{}},
}
// Set up a test catalog and the secrets for it
testCatalog := TestCatalog(t, dbConn, projectScope.PublicId, testPlugin.GetPublicId())
// Set up another test catalog that will not have sets on it
emptyTestCatalog := TestCatalog(t, dbConn, projectScope.PublicId, testPlugin.GetPublicId())
// Create a couple of host sets.
set1 := TestSet(t, dbConn, dbKmsCache, sched, testCatalog, dummyPluginMap)
set1.LastSyncTime = timestamp.New(time.Now())
set1.NeedSync = false
numSetsUpdated, err := dbRW.Update(ctx, set1, []string{"LastSyncTime", "NeedSync"}, []string{})
require.NoError(t, err)
require.Equal(t, 1, numSetsUpdated)
// Set 2
set2 := TestSet(t, dbConn, dbKmsCache, sched, testCatalog, dummyPluginMap)
set2.LastSyncTime = timestamp.New(time.Now())
set2.NeedSync = false
numSetsUpdated, err = dbRW.Update(ctx, set2, []string{"LastSyncTime", "NeedSync"}, []string{})
require.NoError(t, err)
require.Equal(t, 1, numSetsUpdated)
j := &testSyncJob{}
err = sched.RegisterJob(ctx, j, scheduler.WithNextRunIn(setSyncJobRunInterval))
require.NoError(t, err)
repo, err := NewRepository(dbRW, dbRW, dbKmsCache, sched, dummyPluginMap)
require.NoError(t, err)
require.NotNil(t, repo)
// Load the job repository here so that we can validate run times.
jobRepo, err := job.NewRepository(dbRW, dbRW, dbKmsCache)
require.NoError(t, err)
require.NotNil(t, repo)
// Updating an empty catalog should not trigger an update.
emptyTestCatalog.Attributes = mustMarshal(map[string]interface{}{"foo": "bar"})
var catalogsUpdated int
_, _, catalogsUpdated, err = repo.UpdateCatalog(ctx, emptyTestCatalog, emptyTestCatalog.Version, []string{"attributes"})
require.NoError(t, err)
require.Equal(t, 1, catalogsUpdated)
// Job run should have not been requested
jj, err := jobRepo.LookupJob(ctx, setSyncJobName)
require.NoError(t, err)
require.NotNil(t, jj)
require.GreaterOrEqual(t, time.Until(jj.GetNextScheduledRun().GetTimestamp().AsTime()), time.Second)
// Updating name should not flag sets for sync
testCatalog.Name = "foobar"
testCatalog, _, catalogsUpdated, err = repo.UpdateCatalog(ctx, testCatalog, testCatalog.Version, []string{"name"})
require.NoError(t, err)
require.Equal(t, 1, catalogsUpdated)
gotSet1, _, err := repo.LookupSet(ctx, set1.PublicId)
require.NoError(t, err)
require.False(t, gotSet1.NeedSync)
require.Equal(t, set1.LastSyncTime, gotSet1.LastSyncTime)
require.Equal(t, set1.Version, gotSet1.Version)
gotSet2, _, err := repo.LookupSet(ctx, set2.PublicId)
require.NoError(t, err)
require.False(t, gotSet2.NeedSync)
require.Equal(t, set2.LastSyncTime, gotSet2.LastSyncTime)
require.Equal(t, set2.Version, gotSet2.Version)
// Job run still should have not been requested
jj, err = jobRepo.LookupJob(ctx, setSyncJobName)
require.NoError(t, err)
require.NotNil(t, jj)
require.GreaterOrEqual(t, time.Until(jj.GetNextScheduledRun().GetTimestamp().AsTime()), time.Second)
// Updating attributes should trigger update
testCatalog.Attributes = mustMarshal(map[string]interface{}{"foo": "bar"})
_, _, catalogsUpdated, err = repo.UpdateCatalog(ctx, testCatalog, testCatalog.Version, []string{"attributes"})
require.NoError(t, err)
require.Equal(t, 1, catalogsUpdated)
gotSet1, _, err = repo.LookupSet(ctx, set1.PublicId)
require.NoError(t, err)
require.True(t, gotSet1.NeedSync)
require.Equal(t, set1.Version+1, gotSet1.Version)
gotSet2, _, err = repo.LookupSet(ctx, set2.PublicId)
require.NoError(t, err)
require.True(t, gotSet2.NeedSync)
require.Equal(t, set2.Version+1, gotSet2.Version)
// Job run should have been requested
jj, err = jobRepo.LookupJob(ctx, setSyncJobName)
require.NoError(t, err)
require.NotNil(t, jj)
require.Less(t, time.Until(jj.GetNextScheduledRun().GetTimestamp().AsTime()), time.Second)
}
func assertPluginBasedPublicId(t *testing.T, prefix, actual string) {
t.Helper()
assert.NotEmpty(t, actual)