From f82c01833be1af6436e70227482697df006a2e79 Mon Sep 17 00:00:00 2001 From: Lucas Manuel Rodriguez Date: Mon, 7 Feb 2022 17:14:23 -0300 Subject: [PATCH] Change webhooks db lock duration to one hour (#3589) * Change webhooks db lock duration to one hour * Reload interval and check time left * Change interval to one hour to match the db lock * Add missing continue in case of failure * Fix lint-go --- changes/issue-3584-fix-webhooks-lock-duration | 1 + cmd/fleet/serve.go | 41 ++++-- cmd/fleet/serve_test.go | 119 +++++++++++++++++- 3 files changed, 149 insertions(+), 12 deletions(-) create mode 100644 changes/issue-3584-fix-webhooks-lock-duration diff --git a/changes/issue-3584-fix-webhooks-lock-duration b/changes/issue-3584-fix-webhooks-lock-duration new file mode 100644 index 0000000000..00f57defaa --- /dev/null +++ b/changes/issue-3584-fix-webhooks-lock-duration @@ -0,0 +1 @@ +* Amend webhook db lock duration to be always one hour. diff --git a/cmd/fleet/serve.go b/cmd/fleet/serve.go index 706bde17b2..f4b4794e6c 100644 --- a/cmd/fleet/serve.go +++ b/cmd/fleet/serve.go @@ -565,7 +565,7 @@ func runCrons(ds fleet.Datastore, task *async.Task, logger kitlog.Logger, config go cronCleanups(ctx, ds, kitlog.With(logger, "cron", "cleanups"), ourIdentifier, license) go cronVulnerabilities( ctx, ds, kitlog.With(logger, "cron", "vulnerabilities"), ourIdentifier, config) - go cronWebhooks(ctx, ds, kitlog.With(logger, "cron", "webhooks"), ourIdentifier, failingPoliciesSet) + go cronWebhooks(ctx, ds, kitlog.With(logger, "cron", "webhooks"), ourIdentifier, failingPoliciesSet, 1*time.Hour) return cancelBackground } @@ -744,7 +744,14 @@ func checkVulnerabilities(ctx context.Context, ds fleet.Datastore, logger kitlog return recentVulns } -func cronWebhooks(ctx context.Context, ds fleet.Datastore, logger kitlog.Logger, identifier string, failingPoliciesSet fleet.FailingPolicySet) { +func cronWebhooks( + ctx context.Context, + ds fleet.Datastore, + logger kitlog.Logger, + identifier string, + failingPoliciesSet fleet.FailingPolicySet, + intervalReload time.Duration, +) { appConfig, err := ds.AppConfig(ctx) if err != nil { level.Error(logger).Log("config", "couldn't read app config", "err", err) @@ -754,6 +761,7 @@ func cronWebhooks(ctx context.Context, ds fleet.Datastore, logger kitlog.Logger, interval := appConfig.WebhookSettings.Interval.ValueOr(24 * time.Hour) level.Debug(logger).Log("interval", interval.String()) ticker := time.NewTicker(interval) + start := time.Now() for { level.Debug(logger).Log("waiting", "on ticker") select { @@ -762,21 +770,32 @@ func cronWebhooks(ctx context.Context, ds fleet.Datastore, logger kitlog.Logger, case <-ctx.Done(): level.Debug(logger).Log("exit", "done with cron.") return + case <-time.After(intervalReload): + // Reload interval and check if it has been reduced. + appConfig, err := ds.AppConfig(ctx) + if err != nil { + level.Error(logger).Log("config", "couldn't read app config", "err", err) + continue + } + if currInterval := appConfig.WebhookSettings.Interval.ValueOr(24 * time.Hour); time.Since(start) < currInterval { + continue + } } // Reread app config to be able to read latest data used by the webhook - // and update any intervals for next run. + // and update the ticker for the next run. appConfig, err = ds.AppConfig(ctx) if err != nil { level.Error(logger).Log("config", "couldn't read app config", "err", err) sentry.CaptureException(err) } else { - interval = appConfig.WebhookSettings.Interval.ValueOr(24 * time.Hour) - ticker.Reset(interval) + ticker.Reset(appConfig.WebhookSettings.Interval.ValueOr(24 * time.Hour)) + start = time.Now() } - maybeTriggerHostStatus(ctx, ds, logger, identifier, appConfig, interval) - maybeTriggerGlobalFailingPoliciesWebhook(ctx, ds, logger, identifier, appConfig, interval, failingPoliciesSet) + // We set the db lock durations to match the intervalReload. + maybeTriggerHostStatus(ctx, ds, logger, identifier, appConfig, intervalReload) + maybeTriggerGlobalFailingPoliciesWebhook(ctx, ds, logger, identifier, appConfig, intervalReload, failingPoliciesSet) level.Debug(logger).Log("loop", "done") } @@ -788,9 +807,9 @@ func maybeTriggerHostStatus( logger kitlog.Logger, identifier string, appConfig *fleet.AppConfig, - interval time.Duration, + lockDuration time.Duration, ) { - if locked, err := ds.Lock(ctx, lockKeyWebhooksHostStatus, identifier, interval); err != nil || !locked { + if locked, err := ds.Lock(ctx, lockKeyWebhooksHostStatus, identifier, lockDuration); err != nil || !locked { level.Debug(logger).Log("leader-host-status", "Not the leader. Skipping...") return } @@ -809,10 +828,10 @@ func maybeTriggerGlobalFailingPoliciesWebhook( logger kitlog.Logger, identifier string, appConfig *fleet.AppConfig, - interval time.Duration, + lockDuration time.Duration, failingPoliciesSet fleet.FailingPolicySet, ) { - if locked, err := ds.Lock(ctx, lockKeyWebhooksFailingPolicies, identifier, interval); err != nil || !locked { + if locked, err := ds.Lock(ctx, lockKeyWebhooksFailingPolicies, identifier, lockDuration); err != nil || !locked { level.Debug(logger).Log("leader-failing-policies", "Not the leader. Skipping...") return } diff --git a/cmd/fleet/serve_test.go b/cmd/fleet/serve_test.go index e2251d0b7e..abf8c7c5ee 100644 --- a/cmd/fleet/serve_test.go +++ b/cmd/fleet/serve_test.go @@ -8,6 +8,7 @@ import ( "net/http/httptest" "os" "path" + "sync" "sync/atomic" "testing" "time" @@ -165,7 +166,7 @@ func TestCronWebhooks(t *testing.T) { defer cancelFunc() failingPoliciesSet := service.NewMemFailingPolicySet() - go cronWebhooks(ctx, ds, kitlog.With(kitlog.NewNopLogger(), "cron", "webhooks"), "1234", failingPoliciesSet) + go cronWebhooks(ctx, ds, kitlog.With(kitlog.NewNopLogger(), "cron", "webhooks"), "1234", failingPoliciesSet, 5*time.Minute) <-calledOnce time.Sleep(1 * time.Second) @@ -318,3 +319,119 @@ func TestCronVulnerabilitiesSkipCreationIfStatic(t *testing.T) { require.NoDirExists(t, vulnPath) } + +// TestCronWebhooksLockDuration tests that the Lock method is being called +// for the current webhook crons and that their duration is always one hour (see #3584). +func TestCronWebhooksLockDuration(t *testing.T) { + ds := new(mock.Store) + + ds.AppConfigFunc = func(ctx context.Context) (*fleet.AppConfig, error) { + return &fleet.AppConfig{ + WebhookSettings: fleet.WebhookSettings{ + Interval: fleet.Duration{Duration: 1 * time.Second}, + }, + }, nil + } + hostStatus := make(chan struct{}) + hostStatusClosed := false + failingPolicies := make(chan struct{}) + failingPoliciesClosed := false + unknownName := false + ds.LockFunc = func(ctx context.Context, name string, owner string, expiration time.Duration) (bool, error) { + if expiration != 1*time.Hour { + return false, nil + } + switch name { + case lockKeyWebhooksHostStatus: + if !hostStatusClosed { + close(hostStatus) + hostStatusClosed = true + } + case lockKeyWebhooksFailingPolicies: + if !failingPoliciesClosed { + close(failingPolicies) + failingPoliciesClosed = true + } + default: + unknownName = true + } + return true, nil + } + + ctx, cancelFunc := context.WithCancel(context.Background()) + defer cancelFunc() + + go cronWebhooks(ctx, ds, kitlog.NewNopLogger(), "1234", service.NewMemFailingPolicySet(), 1*time.Hour) + + select { + case <-failingPolicies: + case <-time.After(5 * time.Second): + t.Error("failing policies timeout") + } + select { + case <-hostStatus: + case <-time.After(5 * time.Second): + t.Error("host status timeout") + } + require.False(t, unknownName) +} + +func TestCronWebhooksIntervalChange(t *testing.T) { + ds := new(mock.Store) + + interval := struct { + sync.Mutex + value time.Duration + }{ + value: 5 * time.Hour, + } + configLoaded := make(chan struct{}, 1) + + ds.AppConfigFunc = func(ctx context.Context) (*fleet.AppConfig, error) { + select { + case configLoaded <- struct{}{}: + default: + // OK + } + + interval.Lock() + defer interval.Unlock() + + return &fleet.AppConfig{ + WebhookSettings: fleet.WebhookSettings{ + Interval: fleet.Duration{Duration: interval.value}, + }, + }, nil + } + + lockCalled := make(chan struct{}, 1) + ds.LockFunc = func(ctx context.Context, name string, owner string, expiration time.Duration) (bool, error) { + select { + case lockCalled <- struct{}{}: + default: + // OK + } + return true, nil + } + + ctx, cancelFunc := context.WithCancel(context.Background()) + defer cancelFunc() + + go cronWebhooks(ctx, ds, kitlog.NewNopLogger(), "1234", service.NewMemFailingPolicySet(), 200*time.Millisecond) + + select { + case <-configLoaded: + case <-time.After(5 * time.Second): + t.Fatal("timeout: initial config load") + } + + interval.Lock() + interval.value = 1 * time.Second + interval.Unlock() + + select { + case <-lockCalled: + case <-time.After(5 * time.Second): + t.Fatal("timeout: interval change did not trigger lock call") + } +}