Discharge7443/fleet
1
0
Fork 0
Code Issues Pull Requests Actions 5 Packages Projects Releases Wiki Activity

Change webhooks db lock duration to one hour (#3589)

Browse Source 
* 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
This commit is contained in:
Lucas Manuel Rodriguez
2022-02-07 17:14:23 -03:00
committed by GitHub
parent b9e2ca6790
commit f82c01833b
3 changed files with 149 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
changes/issue-3584-fix-webhooks-lock-duration
+1
View File
@@ -0,0 +1 @@
* Amend webhook db lock duration to be always one hour.
cmd/fleet/serve.go
+30 -11
View File
@@ -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
}
cmd/fleet/serve_test.go
+118 -1
View File
@@ -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")
}
}