Update sync interval seconds field. (#1698)
* Update sync interval seconds field. * If the sync interval was updated, schedule the next sync job wakeup immediately. * Calculate the next sync time for the scheduler should wake up if we only updated the sync interval.
This commit is contained in:
4 files changed
+69
-1
No files matched your search
@@ -242,6 +242,7 @@ func (r *Repository) UpdateSet(ctx context.Context, scopeId string, s *HostSet,
|
||||
// Clone the set so that we can set fields.
|
||||
newSet := currentSet.clone()
|
||||
var updateAttributes bool
|
||||
var updateSyncInterval bool
|
||||
var dbMask, nullFields []string
|
||||
const (
|
||||
endpointOpNoop = "endpointOpNoop"
|
||||
@@ -263,6 +264,14 @@ func (r *Repository) UpdateSet(ctx context.Context, scopeId string, s *HostSet,
|
||||
case strings.EqualFold("description", f) && s.Description != "":
|
||||
dbMask = append(dbMask, "description")
|
||||
newSet.Description = s.Description
|
||||
case strings.EqualFold("SyncIntervalSeconds", f) && s.SyncIntervalSeconds == 0:
|
||||
nullFields = append(nullFields, "SyncIntervalSeconds")
|
||||
newSet.SyncIntervalSeconds = s.SyncIntervalSeconds
|
||||
updateSyncInterval = true
|
||||
case strings.EqualFold("SyncIntervalSeconds", f) && s.SyncIntervalSeconds != 0:
|
||||
dbMask = append(dbMask, "SyncIntervalSeconds")
|
||||
newSet.SyncIntervalSeconds = s.SyncIntervalSeconds
|
||||
updateSyncInterval = true
|
||||
case strings.EqualFold("PreferredEndpoints", f) && len(s.PreferredEndpoints) == 0:
|
||||
endpointOp = endpointOpDelete
|
||||
newSet.PreferredEndpoints = s.PreferredEndpoints
|
||||
@@ -514,9 +523,16 @@ func (r *Repository) UpdateSet(ctx context.Context, scopeId string, s *HostSet,
|
||||
numUpdated = 1
|
||||
}
|
||||
|
||||
if updateAttributes {
|
||||
switch {
|
||||
case updateAttributes:
|
||||
// Request a host sync since we have updated attributes.
|
||||
_ = r.scheduler.UpdateJobNextRunInAtLeast(ctx, setSyncJobName, 0)
|
||||
case updateSyncInterval:
|
||||
tilNextSync := time.Until(returnedSet.LastSyncTime.AsTime().Add(time.Duration(returnedSet.SyncIntervalSeconds) * time.Second))
|
||||
if tilNextSync < 0 {
|
||||
tilNextSync = 0
|
||||
}
|
||||
_ = r.scheduler.UpdateJobNextRunInAtLeast(ctx, setSyncJobName, tilNextSync)
|
||||
}
|
||||
|
||||
return returnedSet, hosts, plg, numUpdated, nil
|
||||
|
||||
@@ -476,6 +476,13 @@ func TestRepository_UpdateSet(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
changeSyncInterval := func(s int32) changeHostSetFunc {
|
||||
return func(c *HostSet) *HostSet {
|
||||
c.SyncIntervalSeconds = s
|
||||
return c
|
||||
}
|
||||
}
|
||||
|
||||
changePreferredEndpoints := func(s []string) changeHostSetFunc {
|
||||
return func(c *HostSet) *HostSet {
|
||||
c.PreferredEndpoints = s
|
||||
@@ -546,6 +553,13 @@ func TestRepository_UpdateSet(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
checkSyncInterval := func(want int32) checkHostSetFunc {
|
||||
return func(t *testing.T, got *HostSet) {
|
||||
t.Helper()
|
||||
assert.Equal(t, want, got.SyncIntervalSeconds, "checkSyncInterval")
|
||||
}
|
||||
}
|
||||
|
||||
checkNeedSync := func(want bool) checkHostSetFunc {
|
||||
return func(t *testing.T, got *HostSet) {
|
||||
t.Helper()
|
||||
@@ -889,6 +903,23 @@ func TestRepository_UpdateSet(t *testing.T) {
|
||||
checkVerifySetOplog(oplog.OpType_OP_TYPE_UPDATE),
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "set sync interval",
|
||||
startingSet: setupBareHostSet,
|
||||
changeFuncs: []changeHostSetFunc{changeSyncInterval(42)},
|
||||
fieldMask: []string{"SyncIntervalSeconds"},
|
||||
wantCheckPluginReqFuncs: []checkPluginReqFunc{
|
||||
checkUpdateSetRequestPersistedSecrets(map[string]interface{}{
|
||||
"one": "two",
|
||||
}),
|
||||
},
|
||||
wantCheckSetFuncs: []checkHostSetFunc{
|
||||
checkVersion(2),
|
||||
checkNeedSync(true),
|
||||
checkSyncInterval(42),
|
||||
checkVerifySetOplog(oplog.OpType_OP_TYPE_UPDATE),
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "add preferred endpoints",
|
||||
startingSet: setupBareHostSet,
|
||||
|
||||
@@ -1521,6 +1521,11 @@ func TestUpdate_Plugin(t *testing.T) {
|
||||
c.PreferredEndpoints = i
|
||||
}
|
||||
}
|
||||
updateSyncInterval := func(i *wrapperspb.Int32Value) updateFn {
|
||||
return func(c *pb.HostSet) {
|
||||
c.SyncIntervalSeconds = i
|
||||
}
|
||||
}
|
||||
updateAttrs := func(i *structpb.Struct) updateFn {
|
||||
return func(c *pb.HostSet) {
|
||||
c.Attributes = i
|
||||
@@ -1576,6 +1581,17 @@ func TestUpdate_Plugin(t *testing.T) {
|
||||
assert.Equal(t, in.PreferredEndpoints, []string{"dns:new"})
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Update sync interval",
|
||||
masks: []string{"sync_interval_seconds"},
|
||||
changes: []updateFn{
|
||||
clearReadOnlyFields(),
|
||||
updateSyncInterval(wrapperspb.Int32(42)),
|
||||
},
|
||||
check: func(t *testing.T, in *pb.HostSet) {
|
||||
assert.Equal(t, in.SyncIntervalSeconds, wrapperspb.Int32(42))
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Don't update preferred_endpoints",
|
||||
masks: []string{"name"},
|
||||
|
||||
@@ -237,6 +237,11 @@ func TestCrud(t *testing.T) {
|
||||
assert.Equal(h.Item.Attributes, map[string]interface{}{"key": "val"})
|
||||
assert.Equal(h.Item.PreferredEndpoints, []string{"dns:update"})
|
||||
|
||||
h, err = hClient.Update(tc.Context(), h.Item.Id, h.Item.Version, hostsets.WithSyncIntervalSeconds(42))
|
||||
require.NoError(err)
|
||||
require.NotNil(h)
|
||||
assert.Equal(int32(42), h.Item.SyncIntervalSeconds)
|
||||
|
||||
_, err = hClient.Delete(tc.Context(), h.Item.Id)
|
||||
assert.NoError(err)
|
||||
_, err = hClient.Delete(tc.Context(), h.Item.Id)
|
||||
|
||||
Reference in new issue
Block a user