diff --git a/internal/host/plugin/repository_host_catalog.go b/internal/host/plugin/repository_host_catalog.go index 8f2114c7c..2bf89a2c3 100644 --- a/internal/host/plugin/repository_host_catalog.go +++ b/internal/host/plugin/repository_host_catalog.go @@ -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 diff --git a/internal/host/plugin/repository_host_catalog_test.go b/internal/host/plugin/repository_host_catalog_test.go index c46ca951a..3125a81ac 100644 --- a/internal/host/plugin/repository_host_catalog_test.go +++ b/internal/host/plugin/repository_host_catalog_test.go @@ -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)