feat: add intervalSeed source option (#1945)

* adds an `intervalSeed` source option
  * enables configuration of deterministic offset to polling
  * disabled by default (no offset, wall-clock polling is maintained
* improves sync source logging consistency

---

Fixes: #1944

The HTTP sync and blob syncs use robfig/cron with a simple
interval-based cron expression (e.g. */5 * * * *). This means all flagd
instances configured with the same polling interval will poll at exactly
the same wall-clock times, which can cause thundering herd issues
against the flag source server; this adds a feature to address this
issue in large deployments.

---------

Signed-off-by: Todd Baert <todd.baert@dynatrace.com>

Author: toddbaert

core/go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/open-feature/flagd-schemas v0.2.13
 	github.com/open-feature/open-feature-operator/api v0.2.47
 	github.com/prometheus/client_golang v1.23.2
-	github.com/robfig/cron v1.2.0
+	github.com/robfig/cron/v3 v3.0.1
 	github.com/santhosh-tekuri/jsonschema/v6 v6.0.2
 	github.com/stretchr/testify v1.11.1
 	github.com/twmb/murmur3 v1.1.8

core/go.sum

@@ -279,8 +279,8 @@ github.com/prometheus/procfs v0.20.1 h1:XwbrGOIplXW/AU3YhIhLODXMJYyC1isLFfYCsTEy
 github.com/prometheus/procfs v0.20.1/go.mod h1:o9EMBZGRyvDrSPH1RqdxhojkuXstoe4UlK79eF5TGGo=
 github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E=
 github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw=
-github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ=
-github.com/robfig/cron v1.2.0/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfmt2k=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
 github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 h1:KRzFb2m7YtdldCEkzs6KqmJw4nqEVZGK7IN2kJkjTuQ=

core/pkg/sync/blob/blob_sync.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/open-feature/flagd/core/pkg/logger"
 	"github.com/open-feature/flagd/core/pkg/sync"
+	"github.com/open-feature/flagd/core/pkg/sync/internal/polling"
 	"github.com/open-feature/flagd/core/pkg/utils"
 	"gocloud.dev/blob"
 	_ "gocloud.dev/blob/azureblob" // needed to initialize Azure Blob Storage driver
@@ -21,20 +22,13 @@ type Sync struct {
 	Bucket      string
 	Object      string
 	BlobURLMux  *blob.URLMux
-	Cron        Cron
+	Poller      polling.Poller
 	Logger      *logger.Logger
 	Interval    uint32
 	ready       bool
 	lastUpdated time.Time
 }
 
-// Cron defines the behaviour required of a cron
-type Cron interface {
-	AddFunc(spec string, cmd func()) error
-	Start()
-	Stop()
-}
-
 func (hs *Sync) Init(_ context.Context) error {
 	if hs.Bucket == "" {
 		return errors.New("no bucket string set")
@@ -50,24 +44,26 @@ func (hs *Sync) IsReady() bool {
 }
 
 func (hs *Sync) Sync(ctx context.Context, dataSync chan<- sync.DataSync) error {
-	hs.Logger.Info(fmt.Sprintf("starting sync from %s/%s with interval %ds", hs.Bucket, hs.Object, hs.Interval))
-	_ = hs.Cron.AddFunc(fmt.Sprintf("*/%d * * * *", hs.Interval), func() {
-		err := hs.sync(ctx, dataSync, false)
-		if err != nil {
-			hs.Logger.Warn(fmt.Sprintf("sync failed: %v", err))
-		}
-	})
+	hs.Logger.Info(fmt.Sprintf("starting sync from %s/%s (interval: %ds)", hs.Bucket, hs.Object, hs.Interval))
+
 	// Initial fetch
-	hs.Logger.Debug(fmt.Sprintf("initial sync of the %s/%s", hs.Bucket, hs.Object))
+	hs.Logger.Debug(fmt.Sprintf("initial fetch from %s/%s", hs.Bucket, hs.Object))
 	err := hs.sync(ctx, dataSync, false)
 	if err != nil {
 		return err
 	}
 
 	hs.ready = true
-	hs.Cron.Start()
-	<-ctx.Done()
-	hs.Cron.Stop()
+
+	hs.Logger.Debug(fmt.Sprintf("polling %s/%s every %ds (offset: %ds)",
+		hs.Bucket, hs.Object, hs.Interval, hs.Poller.Offset()))
+
+	hs.Poller.Start(ctx, func() {
+		err := hs.sync(ctx, dataSync, false)
+		if err != nil {
+			hs.Logger.Warn(fmt.Sprintf("sync failed: %v", err))
+		}
+	})
 
 	return nil
 }

core/pkg/sync/blob/blob_sync_test.go

@@ -9,7 +9,6 @@ import (
 	"github.com/open-feature/flagd/core/pkg/logger"
 	"github.com/open-feature/flagd/core/pkg/sync"
 	synctesting "github.com/open-feature/flagd/core/pkg/sync/testing"
-	"go.uber.org/mock/gomock"
 )
 
 func TestBlobSync(t *testing.T) {
@@ -38,17 +37,12 @@ func TestBlobSync(t *testing.T) {
 
 	for name, tt := range tests {
 		t.Run(name, func(t *testing.T) {
-			ctrl := gomock.NewController(t)
-			mockCron := synctesting.NewMockCron(ctrl)
-			mockCron.EXPECT().AddFunc(gomock.Any(), gomock.Any()).DoAndReturn(func(spec string, cmd func()) error {
-				return nil
-			})
-			mockCron.EXPECT().Start().Times(1)
+			mockPoller := synctesting.NewMockPoller()
 
 			blobSync := &Sync{
 				Bucket: tt.scheme + "://" + tt.bucket,
 				Object: tt.object,
-				Cron:   mockCron,
+				Poller: mockPoller,
 				Logger: logger.NewLogger(nil, false),
 			}
 			blobMock := NewMockBlob(tt.scheme, func() *Sync {
@@ -73,19 +67,19 @@ func TestBlobSync(t *testing.T) {
 			if data.FlagData != tt.convertedContent {
 				t.Errorf("expected content: %s, but received content: %s", tt.convertedContent, data.FlagData)
 			}
-			tickWithConfigChange(t, mockCron, dataSyncChan, blobMock, tt.object, tt.convertedContent)
-			tickWithoutConfigChange(t, mockCron, dataSyncChan)
-			tickWithConfigChange(t, mockCron, dataSyncChan, blobMock, tt.object, tt.convertedContent)
-			tickWithoutConfigChange(t, mockCron, dataSyncChan)
-			tickWithoutConfigChange(t, mockCron, dataSyncChan)
+			tickWithConfigChange(t, mockPoller, dataSyncChan, blobMock, tt.object, tt.convertedContent)
+			tickWithoutConfigChange(t, mockPoller, dataSyncChan)
+			tickWithConfigChange(t, mockPoller, dataSyncChan, blobMock, tt.object, tt.convertedContent)
+			tickWithoutConfigChange(t, mockPoller, dataSyncChan)
+			tickWithoutConfigChange(t, mockPoller, dataSyncChan)
 		})
 	}
 }
 
-func tickWithConfigChange(t *testing.T, mockCron *synctesting.MockCron, dataSyncChan chan sync.DataSync, blobMock *MockBlob, object string, newConfig string) {
+func tickWithConfigChange(t *testing.T, mockPoller *synctesting.MockPoller, dataSyncChan chan sync.DataSync, blobMock *MockBlob, object string, newConfig string) {
 	time.Sleep(1 * time.Millisecond) // sleep so the new file has different modification date
 	blobMock.AddObject(object, newConfig)
-	mockCron.Tick()
+	mockPoller.Tick()
 	select {
 	case data, ok := <-dataSyncChan:
 		if ok {
@@ -100,8 +94,8 @@ func tickWithConfigChange(t *testing.T, mockCron *synctesting.MockCron, dataSync
 	}
 }
 
-func tickWithoutConfigChange(t *testing.T, mockCron *synctesting.MockCron, dataSyncChan chan sync.DataSync) {
-	mockCron.Tick()
+func tickWithoutConfigChange(t *testing.T, mockPoller *synctesting.MockPoller, dataSyncChan chan sync.DataSync) {
+	mockPoller.Tick()
 	select {
 	case data, ok := <-dataSyncChan:
 		if ok {
@@ -119,13 +113,12 @@ func TestReSync(t *testing.T) {
 		bucket = "b"
 		object = "flags.json"
 	)
-	ctrl := gomock.NewController(t)
-	mockCron := synctesting.NewMockCron(ctrl)
+	mockPoller := synctesting.NewMockPoller()
 
 	blobSync := &Sync{
 		Bucket: scheme + "://" + bucket,
 		Object: object,
-		Cron:   mockCron,
+		Poller: mockPoller,
 		Logger: logger.NewLogger(nil, false),
 	}
 	blobMock := NewMockBlob(scheme, func() *Sync {

core/pkg/sync/builder/syncbuilder.go

@@ -12,8 +12,8 @@ import (
 	"github.com/open-feature/flagd/core/pkg/sync/grpc"
 	"github.com/open-feature/flagd/core/pkg/sync/grpc/credentials"
 	httpSync "github.com/open-feature/flagd/core/pkg/sync/http"
+	"github.com/open-feature/flagd/core/pkg/sync/internal/polling"
 	"github.com/open-feature/flagd/core/pkg/sync/kubernetes"
-	"github.com/robfig/cron"
 	"go.uber.org/zap"
 	"gocloud.dev/blob"
 	"k8s.io/client-go/dynamic"
@@ -110,19 +110,19 @@ func (sb *SyncBuilder) syncFromConfig(sourceConfig sync.SourceConfig, logger *lo
 		return sb.newK8s(sourceConfig.URI, logger)
 	case syncProviderHTTP:
 		logger.Debug(fmt.Sprintf("using remote sync-provider for: %s", sourceConfig.URI))
-		return httpSync.NewHTTP(sourceConfig, logger), nil
+		return sb.newHTTP(sourceConfig, logger)
 	case syncProviderGrpc:
 		logger.Debug(fmt.Sprintf("using grpc sync-provider for: %s", sourceConfig.URI))
 		return sb.newGRPC(sourceConfig, logger), nil
 	case syncProviderGcs:
 		logger.Debug(fmt.Sprintf("using blob sync-provider with gcs driver for: %s", sourceConfig.URI))
-		return sb.newGcs(sourceConfig, logger), nil
+		return sb.newGcs(sourceConfig, logger)
 	case syncProviderAzblob:
 		logger.Debug(fmt.Sprintf("using blob sync-provider with azblob driver for: %s", sourceConfig.URI))
 		return sb.newAzblob(sourceConfig, logger)
 	case syncProviderS3:
 		logger.Debug(fmt.Sprintf("using blob sync-provider with s3 driver for: %s", sourceConfig.URI))
-		return sb.newS3(sourceConfig, logger), nil
+		return sb.newS3(sourceConfig, logger)
 
 	default:
 		return nil, fmt.Errorf("invalid sync provider: %s, must be one of with "+
@@ -201,17 +201,38 @@ func (sb *SyncBuilder) newGRPC(config sync.SourceConfig, logger *logger.Logger)
 	}
 }
 
-func (sb *SyncBuilder) newGcs(config sync.SourceConfig, logger *logger.Logger) *blobSync.Sync {
+// defaultInterval returns the configured interval or 5 seconds if unset,
+// along with a validated poller.
+func newPoller(config sync.SourceConfig) (uint32, *polling.CronPoller, error) {
+	var interval uint32 = 5
+	if config.Interval != 0 {
+		interval = config.Interval
+	}
+	poller, err := polling.NewCronPoller(interval, config.IntervalSeed)
+	if err != nil {
+		return 0, nil, err
+	}
+	return interval, poller, nil
+}
+
+func (sb *SyncBuilder) newHTTP(config sync.SourceConfig, logger *logger.Logger) (*httpSync.Sync, error) {
+	interval, poller, err := newPoller(config)
+	if err != nil {
+		return nil, fmt.Errorf("invalid http sync configuration: %w", err)
+	}
+	return httpSync.NewHTTP(config, logger, poller, interval), nil
+}
+
+func (sb *SyncBuilder) newGcs(config sync.SourceConfig, logger *logger.Logger) (*blobSync.Sync, error) {
 	// Extract bucket uri and object name from the full URI:
 	// gs://bucket/path/to/object results in gs://bucket/ as bucketUri and
 	// path/to/object as an object name.
 	bucketURI := regGcs.FindString(config.URI)
 	objectName := regGcs.ReplaceAllString(config.URI, "")
 
-	// Defaults to 5 seconds if interval is not set.
-	var interval uint32 = 5
-	if config.Interval != 0 {
-		interval = config.Interval
+	interval, poller, err := newPoller(config)
+	if err != nil {
+		return nil, fmt.Errorf("invalid gcs sync configuration: %w", err)
 	}
 
 	return &blobSync.Sync{
@@ -225,8 +246,8 @@ func (sb *SyncBuilder) newGcs(config sync.SourceConfig, logger *logger.Logger) *
 			zap.String("sync", "gcs"),
 		),
 		Interval: interval,
-		Cron:     cron.New(),
-	}
+		Poller:   poller,
+	}, nil
 }
 
 func (sb *SyncBuilder) newAzblob(config sync.SourceConfig, logger *logger.Logger) (*blobSync.Sync, error) {
@@ -245,10 +266,9 @@ func (sb *SyncBuilder) newAzblob(config sync.SourceConfig, logger *logger.Logger
 	bucketURI := regAzblob.FindString(config.URI)
 	objectName := regAzblob.ReplaceAllString(config.URI, "")
 
-	// Defaults to 5 seconds if interval is not set.
-	var interval uint32 = 5
-	if config.Interval != 0 {
-		interval = config.Interval
+	interval, poller, err := newPoller(config)
+	if err != nil {
+		return nil, fmt.Errorf("invalid azblob sync configuration: %w", err)
 	}
 
 	return &blobSync.Sync{
@@ -262,21 +282,20 @@ func (sb *SyncBuilder) newAzblob(config sync.SourceConfig, logger *logger.Logger
 			zap.String("sync", "azblob"),
 		),
 		Interval: interval,
-		Cron:     cron.New(),
+		Poller:   poller,
 	}, nil
 }
 
-func (sb *SyncBuilder) newS3(config sync.SourceConfig, logger *logger.Logger) *blobSync.Sync {
+func (sb *SyncBuilder) newS3(config sync.SourceConfig, logger *logger.Logger) (*blobSync.Sync, error) {
 	// Extract bucket uri and object name from the full URI:
 	// s3://bucket/path/to/object results in s3://bucket/ as bucketUri and
 	// path/to/object as an object name.
 	bucketURI := regS3.FindString(config.URI)
 	objectName := regS3.ReplaceAllString(config.URI, "")
 
-	// Defaults to 5 seconds if interval is not set.
-	var interval uint32 = 5
-	if config.Interval != 0 {
-		interval = config.Interval
+	interval, poller, err := newPoller(config)
+	if err != nil {
+		return nil, fmt.Errorf("invalid s3 sync configuration: %w", err)
 	}
 
 	return &blobSync.Sync{
@@ -290,8 +309,8 @@ func (sb *SyncBuilder) newS3(config sync.SourceConfig, logger *logger.Logger) *b
 			zap.String("sync", "s3"),
 		),
 		Interval: interval,
-		Cron:     cron.New(),
-	}
+		Poller:   poller,
+	}, nil
 }
 
 type IK8sClientBuilder interface {

core/pkg/sync/builder/syncbuilder_test.go

@@ -338,10 +338,11 @@ func Test_GcsConfig(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			gcsSync := NewSyncBuilder().newGcs(sync.SourceConfig{
+			gcsSync, err := NewSyncBuilder().newGcs(sync.SourceConfig{
 				URI:      tt.uri,
 				Interval: tt.interval,
 			}, lg)
+			require.NoError(t, err)
 			require.Equal(t, tt.expectedBucket, gcsSync.Bucket)
 			require.Equal(t, tt.expectedObject, gcsSync.Object)
 			require.Equal(t, int(tt.expectedInterval), int(gcsSync.Interval))
@@ -479,10 +480,11 @@ func Test_S3Config(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			s3Sync := NewSyncBuilder().newS3(sync.SourceConfig{
+			s3Sync, err := NewSyncBuilder().newS3(sync.SourceConfig{
 				URI:      tt.uri,
 				Interval: tt.interval,
 			}, lg)
+			require.NoError(t, err)
 			require.Equal(t, tt.expectedBucket, s3Sync.Bucket)
 			require.Equal(t, tt.expectedObject, s3Sync.Object)
 			require.Equal(t, int(tt.expectedInterval), int(s3Sync.Interval))

core/pkg/sync/builder/utils_test.go

@@ -40,8 +40,8 @@ func TestParseSource(t *testing.T) {
 					Provider: syncProviderFile,
 				},
 				{
-					URI:         "http://test.com",
-					Provider:    syncProviderHTTP,
+					URI:        "http://test.com",
+					Provider:   syncProviderHTTP,
 					AuthHeader: "Bearer :)",
 				},
 				{
@@ -71,7 +71,7 @@ func TestParseSource(t *testing.T) {
 				{"uri":"config/samples/example_flags.json","provider":"file"},
 				{"uri":"https://secure-remote","provider":"http","authHeader":"Bearer bearer-dji34ld2l"},
 				{"uri":"https://secure-remote","provider":"http","authHeader":"Basic dXNlcjpwYXNz"},
-				{"uri":"http://site.com","provider":"http","interval":77 },
+				{"uri":"http://site.com","provider":"http","interval":77,"intervalSeed":"my-pod-123"},
 				{"uri":"default/my-flag-config","provider":"kubernetes"},
 				{"uri":"grpc-source:8080","provider":"grpc"},
 				{"uri":"my-flag-source:8080","provider":"grpc", "tls":true, "certPath": "/certs/ca.cert", "providerID": "flagd-weatherapp-sidecar", "selector": "source=database,app=weatherapp"}
@@ -93,9 +93,10 @@ func TestParseSource(t *testing.T) {
 					AuthHeader: "Basic dXNlcjpwYXNz",
 				},
 				{
-					URI:      "http://site.com",
-					Provider: syncProviderHTTP,
-					Interval: 77,
+					URI:          "http://site.com",
+					Provider:     syncProviderHTTP,
+					Interval:     77,
+					IntervalSeed: "my-pod-123",
 				},
 				{
 					URI:      "default/my-flag-config",

core/pkg/sync/file/filepath_sync.go

@@ -94,9 +94,11 @@ func (fs *Sync) setReady(val bool) {
 //nolint:funlen
 func (fs *Sync) Sync(ctx context.Context, dataSync chan<- sync.DataSync) error {
 	defer fs.watcher.Close()
+	fs.Logger.Info(fmt.Sprintf("starting sync from %s", fs.URI))
+	fs.Logger.Debug(fmt.Sprintf("initial fetch from %s", fs.URI))
 	fs.sendDataSync(ctx, dataSync)
 	fs.setReady(true)
-	fs.Logger.Info(fmt.Sprintf("watching filepath: %s", fs.URI))
+	fs.Logger.Debug(fmt.Sprintf("watching %s for changes", fs.URI))
 	for {
 		select {
 		case event, ok := <-fs.watcher.Events():