-
Notifications
You must be signed in to change notification settings - Fork 71
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
8 changed files
with
434 additions
and
62 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,144 @@ | ||
| package blob | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "errors" | ||
| "fmt" | ||
| "io" | ||
| "time" | ||
|
|
||
| "github.com/open-feature/flagd/core/pkg/logger" | ||
| "github.com/open-feature/flagd/core/pkg/sync" | ||
| "gocloud.dev/blob" | ||
| //nolint:gosec | ||
| ) | ||
|
|
||
| type Sync struct { | ||
| Bucket string | ||
| Object string | ||
| BlobURLMux *blob.URLMux | ||
| Cron Cron | ||
| 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(ctx context.Context) error { | ||
| return nil | ||
| } | ||
|
|
||
| func (hs *Sync) IsReady() bool { | ||
| return hs.ready | ||
| } | ||
|
|
||
| func (hs *Sync) Sync(ctx context.Context, dataSync chan<- sync.DataSync) error { | ||
| hs.Logger.Info(fmt.Sprintf("starting sync from %s/%s with interval %d", hs.Bucket, hs.Object, hs.Interval)) | ||
| // Initial fetch | ||
| hs.Logger.Debug(fmt.Sprintf("initial sync of the %s/%s", hs.Bucket, hs.Object)) | ||
| err := hs.ReSync(ctx, dataSync) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| hs.ready = true | ||
|
|
||
| hs.Logger.Debug(fmt.Sprintf("polling %s/%s every %d seconds", hs.Bucket, hs.Object, hs.Interval)) | ||
| _ = hs.Cron.AddFunc(fmt.Sprintf("*/%d * * * *", hs.Interval), func() { | ||
| hs.Logger.Debug(fmt.Sprintf("fetching configuration from %s/%s", hs.Bucket, hs.Object)) | ||
| bucket, err := hs.getBucket(ctx) | ||
| if err != nil { | ||
| hs.Logger.Warn(fmt.Sprintf("couldn't get bucket: %v", err)) | ||
| return | ||
| } | ||
| defer bucket.Close() | ||
| updated, err := hs.fetchObjectModificationTime(ctx, bucket) | ||
| if err != nil { | ||
| hs.Logger.Warn(fmt.Sprintf("couldn't get object attributes: %v", err)) | ||
| return | ||
| } | ||
| if hs.lastUpdated == updated { | ||
| hs.Logger.Debug("configuration hasn't changed, skipping fetching full object") | ||
| return | ||
| } | ||
| msg, err := hs.fetchObject(ctx, bucket) | ||
| if err != nil { | ||
| hs.Logger.Warn(fmt.Sprintf("couldn't get object: %v", err)) | ||
| return | ||
| } | ||
| hs.Logger.Info(fmt.Sprintf("configuration updated: %s", msg)) | ||
| dataSync <- sync.DataSync{FlagData: msg, Source: hs.Bucket + "/" + hs.Object, Type: sync.ALL} | ||
| hs.lastUpdated = updated | ||
| }) | ||
|
|
||
| hs.Cron.Start() | ||
|
|
||
| <-ctx.Done() | ||
| hs.Cron.Stop() | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (hs *Sync) ReSync(ctx context.Context, dataSync chan<- sync.DataSync) error { | ||
| bucket, err := hs.getBucket(ctx) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| defer bucket.Close() | ||
| updated, err := hs.fetchObjectModificationTime(ctx, bucket) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| msg, err := hs.fetchObject(ctx, bucket) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| hs.Logger.Info(fmt.Sprintf("configuration updated: %s", msg)) | ||
| dataSync <- sync.DataSync{FlagData: msg, Source: hs.Bucket + "/" + hs.Object, Type: sync.ALL} | ||
| hs.lastUpdated = updated | ||
| return nil | ||
| } | ||
|
|
||
| func (hs *Sync) getBucket(ctx context.Context) (*blob.Bucket, error) { | ||
| if hs.Bucket == "" { | ||
| return nil, errors.New("no bucket string set") | ||
| } | ||
| return hs.BlobURLMux.OpenBucket(ctx, hs.Bucket) | ||
| } | ||
|
|
||
| func (hs *Sync) fetchObjectModificationTime(ctx context.Context, bucket *blob.Bucket) (time.Time, error) { | ||
| if hs.Object == "" { | ||
| return time.Time{}, errors.New("no object string set") | ||
| } | ||
| attrs, err := bucket.Attributes(ctx, hs.Object) | ||
| if err != nil { | ||
| return time.Time{}, fmt.Errorf("error fetching attributes for object %s/%s: %w", hs.Bucket, hs.Object, err) | ||
| } | ||
| return attrs.ModTime, nil | ||
| } | ||
|
|
||
| func (hs *Sync) fetchObject(ctx context.Context, bucket *blob.Bucket) (string, error) { | ||
| if hs.Object == "" { | ||
| return "", errors.New("no object string set") | ||
| } | ||
| r, err := bucket.NewReader(ctx, hs.Object, nil) | ||
| if err != nil { | ||
| return "", fmt.Errorf("error creating reader for object %s/%s: %w", hs.Bucket, hs.Object, err) | ||
| } | ||
| defer r.Close() | ||
|
|
||
| buf := bytes.NewBuffer(nil) | ||
| _, err = io.Copy(buf, r) | ||
| if err != nil { | ||
| return "", fmt.Errorf("error reading object %s/%s: %w", hs.Bucket, hs.Object, err) | ||
| } | ||
|
|
||
| return string(buf.Bytes()), nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,127 @@ | ||
| package blob | ||
|
|
||
| import ( | ||
| "context" | ||
| "log" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "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" | ||
| ) | ||
|
|
||
| const ( | ||
| scheme = "xyz" | ||
| bucket = "b" | ||
| object = "o" | ||
| ) | ||
|
|
||
| func TestSync(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) | ||
|
|
||
| blobSync := &Sync{ | ||
| Bucket: scheme + "://" + bucket, | ||
| Object: object, | ||
| Cron: mockCron, | ||
| Logger: logger.NewLogger(nil, false), | ||
| } | ||
| blobMock := NewMockBlob(scheme, func() *Sync { | ||
| return blobSync | ||
| }) | ||
| blobSync.BlobURLMux = blobMock.URLMux() | ||
|
|
||
| ctx := context.Background() | ||
| dataSyncChan := make(chan sync.DataSync, 1) | ||
|
|
||
| config := "my-config" | ||
| blobMock.AddObject(object, config) | ||
|
|
||
| go func() { | ||
| err := blobSync.Sync(ctx, dataSyncChan) | ||
| if err != nil { | ||
| log.Fatalf("Error start sync: %s", err.Error()) | ||
| return | ||
| } | ||
| }() | ||
|
|
||
| data := <-dataSyncChan // initial sync | ||
| if data.FlagData != config { | ||
| t.Errorf("expected content: %s, but received content: %s", config, data.FlagData) | ||
| } | ||
| tickWithConfigChange(t, mockCron, dataSyncChan, blobMock, "new config") | ||
| tickWithoutConfigChange(t, mockCron, dataSyncChan) | ||
| tickWithConfigChange(t, mockCron, dataSyncChan, blobMock, "new config 2") | ||
| tickWithoutConfigChange(t, mockCron, dataSyncChan) | ||
| tickWithoutConfigChange(t, mockCron, dataSyncChan) | ||
| } | ||
|
|
||
| func tickWithConfigChange(t *testing.T, mockCron *synctesting.MockCron, dataSyncChan chan sync.DataSync, blobMock *MockBlob, newConfig string) { | ||
| time.Sleep(time.Millisecond) // sleep so the new file has different modification date | ||
| blobMock.AddObject(object, newConfig) | ||
| mockCron.Tick() | ||
| select { | ||
| case data, ok := <-dataSyncChan: | ||
| if ok { | ||
| if data.FlagData != newConfig { | ||
| t.Errorf("expected content: %s, but received content: %s", newConfig, data.FlagData) | ||
| } | ||
| } else { | ||
| t.Errorf("data channel unexpecdly closed") | ||
| } | ||
| default: | ||
| t.Errorf("data channel has no expected update") | ||
| } | ||
| } | ||
|
|
||
| func tickWithoutConfigChange(t *testing.T, mockCron *synctesting.MockCron, dataSyncChan chan sync.DataSync) { | ||
| mockCron.Tick() | ||
| select { | ||
| case data, ok := <-dataSyncChan: | ||
| if ok { | ||
| t.Errorf("unexpected update: %s", data.FlagData) | ||
| } else { | ||
| t.Errorf("data channel unexpecdly closed") | ||
| } | ||
| default: | ||
| } | ||
| } | ||
|
|
||
| func TestReSync(t *testing.T) { | ||
| ctrl := gomock.NewController(t) | ||
| mockCron := synctesting.NewMockCron(ctrl) | ||
|
|
||
| blobSync := &Sync{ | ||
| Bucket: scheme + "://" + bucket, | ||
| Object: object, | ||
| Cron: mockCron, | ||
| Logger: logger.NewLogger(nil, false), | ||
| } | ||
| blobMock := NewMockBlob(scheme, func() *Sync { | ||
| return blobSync | ||
| }) | ||
| blobSync.BlobURLMux = blobMock.URLMux() | ||
|
|
||
| ctx := context.Background() | ||
| dataSyncChan := make(chan sync.DataSync, 1) | ||
|
|
||
| config := "my-config" | ||
| blobMock.AddObject(object, config) | ||
|
|
||
| err := blobSync.ReSync(ctx, dataSyncChan) | ||
| if err != nil { | ||
| log.Fatalf("Error start sync: %s", err.Error()) | ||
| return | ||
| } | ||
|
|
||
| data := <-dataSyncChan | ||
| if data.FlagData != config { | ||
| t.Errorf("expected content: %s, but received content: %s", config, data.FlagData) | ||
| } | ||
| } |
Oops, something went wrong.