diff --git a/internal/filedata/document.go b/internal/filedata/document.go new file mode 100644 index 00000000..4c4f9a34 --- /dev/null +++ b/internal/filedata/document.go @@ -0,0 +1,71 @@ +// Package filedata contains the file reading, parsing, and merging logic shared by the +// components that load flag and segment data from local files. +package filedata + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "unicode" + + "github.com/launchdarkly/go-sdk-common/v3/ldvalue" + "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldbuilders" + "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldmodel" + + "gopkg.in/ghodss/yaml.v1" +) + +// Document is the parsed form of a single data file. A document may contain full flag +// definitions, simplified flag-key-to-value entries, and segment definitions. +type Document struct { + Flags *map[string]ldmodel.FeatureFlag + FlagValues *map[string]ldvalue.Value + Segments *map[string]ldmodel.Segment +} + +// ReadFile reads and parses a single data file, which may be in JSON or YAML format. +func ReadFile(path string) (Document, error) { + var data Document + var rawData []byte + var err error + if rawData, err = os.ReadFile(path); err != nil { //nolint:gosec // G304: ok to read file into variable + return data, fmt.Errorf("unable to read file: %s", err) + } + if detectJSON(rawData) { + err = json.Unmarshal(rawData, &data) + } else { + err = yaml.Unmarshal(rawData, &data) + } + if err != nil { + err = fmt.Errorf("error parsing file: %s", err) + } + return data, err +} + +func detectJSON(rawData []byte) bool { + // A valid JSON file for our purposes must be an object, i.e. it must start with '{' + return strings.HasPrefix(strings.TrimLeftFunc(string(rawData), unicode.IsSpace), "{") +} + +// AbsFilePaths converts each of the given paths to an absolute path. +func AbsFilePaths(paths []string) ([]string, error) { + absPaths := make([]string, 0) + for _, p := range paths { + absPath, err := filepath.Abs(p) + if err != nil { + // COVERAGE: there's no reliable cross-platform way to simulate an invalid path in unit tests + return nil, fmt.Errorf("unable to determine absolute path for '%s'", p) + } + absPaths = append(absPaths, absPath) + } + return absPaths, nil +} + +// MakeFlagWithValue expands a flag-key-to-value entry into a full flag definition that +// returns the given value for every context. +func MakeFlagWithValue(key string, v interface{}) *ldmodel.FeatureFlag { + flag := ldbuilders.NewFlagBuilder(key).SingleVariation(ldvalue.CopyArbitraryValue(v)).Build() + return &flag +} diff --git a/internal/filedata/filedata_test.go b/internal/filedata/filedata_test.go new file mode 100644 index 00000000..01126c55 --- /dev/null +++ b/internal/filedata/filedata_test.go @@ -0,0 +1,158 @@ +package filedata + +import ( + "os" + "path/filepath" + "testing" + + "github.com/launchdarkly/go-sdk-common/v3/ldvalue" + "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldbuilders" + "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldmodel" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func writeTempFile(t *testing.T, content string) string { + t.Helper() + f, err := os.CreateTemp(t.TempDir(), "filedata-test") + require.NoError(t, err) + _, err = f.WriteString(content) + require.NoError(t, err) + require.NoError(t, f.Close()) + return f.Name() +} + +func TestReadFileJSON(t *testing.T) { + path := writeTempFile(t, `{"flagValues": {"my-flag": true}, "segments": {"my-segment": {"key": "my-segment", "version": 3}}}`) + doc, err := ReadFile(path) + require.NoError(t, err) + require.NotNil(t, doc.FlagValues) + assert.Equal(t, ldvalue.Bool(true), (*doc.FlagValues)["my-flag"]) + require.NotNil(t, doc.Segments) + assert.Equal(t, 3, (*doc.Segments)["my-segment"].Version) + assert.Nil(t, doc.Flags) +} + +func TestReadFileYAML(t *testing.T) { + path := writeTempFile(t, "flagValues:\n my-flag: yes\n") + doc, err := ReadFile(path) + require.NoError(t, err) + require.NotNil(t, doc.FlagValues) + assert.Equal(t, ldvalue.Bool(true), (*doc.FlagValues)["my-flag"]) +} + +func TestReadFileErrors(t *testing.T) { + _, err := ReadFile(filepath.Join(t.TempDir(), "nonexistent")) + assert.ErrorContains(t, err, "unable to read file") + + path := writeTempFile(t, `{"flagValues"`) + _, err = ReadFile(path) + assert.ErrorContains(t, err, "error parsing file") + + path = writeTempFile(t, "\t: not yaml") + _, err = ReadFile(path) + assert.ErrorContains(t, err, "error parsing file") +} + +func TestAbsFilePaths(t *testing.T) { + abs, err := AbsFilePaths([]string{"relative/path", string(filepath.Separator) + "already-absolute"}) + require.NoError(t, err) + require.Len(t, abs, 2) + for _, p := range abs { + assert.True(t, filepath.IsAbs(p), "expected absolute path, got %s", p) + } +} + +func TestMakeFlagWithValue(t *testing.T) { + flag := MakeFlagWithValue("my-flag", "on") + assert.Equal(t, "my-flag", flag.Key) + require.Len(t, flag.Variations, 1) + assert.Equal(t, ldvalue.String("on"), flag.Variations[0]) + require.NotNil(t, flag.OffVariation) +} + +func docWithFlag(flag ldmodel.FeatureFlag) Document { + flags := map[string]ldmodel.FeatureFlag{flag.Key: flag} + return Document{Flags: &flags} +} + +func docWithFlagValue(key string, value ldvalue.Value) Document { + values := map[string]ldvalue.Value{key: value} + return Document{FlagValues: &values} +} + +func docWithSegment(segment ldmodel.Segment) Document { + segments := map[string]ldmodel.Segment{segment.Key: segment} + return Document{Segments: &segments} +} + +func TestMergeCombinesDocuments(t *testing.T) { + flag1 := ldbuilders.NewFlagBuilder("flag1").Version(2).Build() + segment1 := ldbuilders.NewSegmentBuilder("segment1").Version(4).Build() + + result, err := Merge(DuplicateKeysFail, + docWithFlag(flag1), docWithFlagValue("flag2", ldvalue.Bool(true)), docWithSegment(segment1)) + require.NoError(t, err) + + require.Len(t, result.Flags, 2) + assert.Equal(t, "flag1", result.Flags[0].Key) + assert.Equal(t, 2, result.Flags[0].Item.Version) + require.IsType(t, &ldmodel.FeatureFlag{}, result.Flags[0].Item.Item) + assert.Equal(t, "flag2", result.Flags[1].Key) + expanded := result.Flags[1].Item.Item.(*ldmodel.FeatureFlag) + require.Len(t, expanded.Variations, 1) + assert.Equal(t, ldvalue.Bool(true), expanded.Variations[0]) + + require.Len(t, result.Segments, 1) + assert.Equal(t, "segment1", result.Segments[0].Key) + assert.Equal(t, 4, result.Segments[0].Item.Version) +} + +func TestMergeDuplicateKeys(t *testing.T) { + flagA := ldbuilders.NewFlagBuilder("flag1").Version(1).Build() + flagB := ldbuilders.NewFlagBuilder("flag1").Version(2).Build() + + t.Run("fail", func(t *testing.T) { + _, err := Merge(DuplicateKeysFail, docWithFlag(flagA), docWithFlag(flagB)) + assert.ErrorContains(t, err, "flag 'flag1' is specified by multiple files") + }) + + t.Run("unrecognized handling behaves as fail", func(t *testing.T) { + _, err := Merge(DuplicateKeysHandling("bogus"), docWithFlag(flagA), docWithFlag(flagB)) + assert.Error(t, err) + }) + + t.Run("ignore all but first", func(t *testing.T) { + result, err := Merge(DuplicateKeysIgnoreAllButFirst, docWithFlag(flagA), docWithFlag(flagB)) + require.NoError(t, err) + require.Len(t, result.Flags, 1) + assert.Equal(t, 1, result.Flags[0].Item.Version) + }) + + t.Run("full flag and flag value collide", func(t *testing.T) { + _, err := Merge(DuplicateKeysFail, docWithFlag(flagA), docWithFlagValue("flag1", ldvalue.Bool(true))) + assert.Error(t, err) + }) + + t.Run("duplicate segments", func(t *testing.T) { + segment := ldbuilders.NewSegmentBuilder("segment1").Build() + _, err := Merge(DuplicateKeysFail, docWithSegment(segment), docWithSegment(segment)) + assert.ErrorContains(t, err, "segment 'segment1' is specified by multiple files") + }) +} + +func TestMergePreservesDocumentOrder(t *testing.T) { + docs := make([]Document, 0, 5) + expectedKeys := []string{"flag-a", "flag-b", "flag-c", "flag-d", "flag-e"} + for _, key := range expectedKeys { + docs = append(docs, docWithFlag(ldbuilders.NewFlagBuilder(key).Build())) + } + result, err := Merge(DuplicateKeysFail, docs...) + require.NoError(t, err) + keys := make([]string, 0, len(result.Flags)) + for _, item := range result.Flags { + keys = append(keys, item.Key) + } + assert.Equal(t, expectedKeys, keys) +} diff --git a/internal/filedata/merge.go b/internal/filedata/merge.go new file mode 100644 index 00000000..cd340200 --- /dev/null +++ b/internal/filedata/merge.go @@ -0,0 +1,100 @@ +package filedata + +import ( + "fmt" + + "github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes" +) + +// DuplicateKeysHandling determines what happens when the same flag or segment key appears +// in more than one document. +// +// The values match the public option types in the packages that expose this behavior, so +// those types can be converted to this one directly. +type DuplicateKeysHandling string + +const ( + // DuplicateKeysFail means a duplicated key causes the merge to fail. + DuplicateKeysFail DuplicateKeysHandling = "fail" + // DuplicateKeysIgnoreAllButFirst means only the first occurrence of a duplicated key is + // used, in the order the documents were given. + DuplicateKeysIgnoreAllButFirst DuplicateKeysHandling = "ignore" +) + +// MergeResult holds the merged items from one or more documents. Item values are +// *ldmodel.FeatureFlag or *ldmodel.Segment. +// +// Ordering is deterministic at document granularity only: all of one document's items +// precede the next document's, matching the order the documents were given, but the +// relative order of items within a single document is unspecified. Consumers key items by +// their Key and must not rely on within-document ordering. +type MergeResult struct { + Flags []ldstoretypes.KeyedItemDescriptor + Segments []ldstoretypes.KeyedItemDescriptor +} + +type itemCategory string + +const ( + flagCategory itemCategory = "flag" + segmentCategory itemCategory = "segment" +) + +// Merge combines the items of the given documents, expanding flag-value entries into full +// flag definitions and applying the given duplicate-key handling. An unrecognized +// DuplicateKeysHandling value behaves as DuplicateKeysFail. +func Merge(duplicateKeysHandling DuplicateKeysHandling, docs ...Document) (MergeResult, error) { + var result MergeResult + seenKeys := map[itemCategory]map[string]bool{ + flagCategory: {}, + segmentCategory: {}, + } + + insert := func( + items *[]ldstoretypes.KeyedItemDescriptor, + category itemCategory, + key string, + data ldstoretypes.ItemDescriptor, + ) error { + if seenKeys[category][key] { + switch duplicateKeysHandling { + case DuplicateKeysIgnoreAllButFirst: + return nil + default: + return fmt.Errorf("%s '%s' is specified by multiple files", category, key) + } + } + *items = append(*items, ldstoretypes.KeyedItemDescriptor{Key: key, Item: data}) + seenKeys[category][key] = true + return nil + } + + for _, d := range docs { + if d.Flags != nil { + for key, f := range *d.Flags { + data := ldstoretypes.ItemDescriptor{Version: f.Version, Item: &f} + if err := insert(&result.Flags, flagCategory, key, data); err != nil { + return MergeResult{}, err + } + } + } + if d.FlagValues != nil { + for key, value := range *d.FlagValues { + flag := MakeFlagWithValue(key, value) + data := ldstoretypes.ItemDescriptor{Version: flag.Version, Item: flag} + if err := insert(&result.Flags, flagCategory, key, data); err != nil { + return MergeResult{}, err + } + } + } + if d.Segments != nil { + for key, s := range *d.Segments { + data := ldstoretypes.ItemDescriptor{Version: s.Version, Item: &s} + if err := insert(&result.Segments, segmentCategory, key, data); err != nil { + return MergeResult{}, err + } + } + } + } + return result, nil +} diff --git a/ldfiledata/file_data_source_impl.go b/ldfiledata/file_data_source_impl.go index 5f277548..466471a4 100644 --- a/ldfiledata/file_data_source_impl.go +++ b/ldfiledata/file_data_source_impl.go @@ -1,25 +1,15 @@ package ldfiledata import ( - "encoding/json" - "fmt" - "os" - "path/filepath" - "strings" "sync" "time" - "unicode" "github.com/launchdarkly/go-sdk-common/v3/ldlog" - "github.com/launchdarkly/go-sdk-common/v3/ldvalue" - "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldbuilders" - "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldmodel" "github.com/launchdarkly/go-server-sdk/v7/interfaces" "github.com/launchdarkly/go-server-sdk/v7/internal/datakinds" + "github.com/launchdarkly/go-server-sdk/v7/internal/filedata" "github.com/launchdarkly/go-server-sdk/v7/subsystems" "github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes" - - "gopkg.in/ghodss/yaml.v1" ) type fileDataSource struct { @@ -32,7 +22,12 @@ type fileDataSource struct { readyCh chan<- struct{} readyOnce sync.Once closeOnce sync.Once - closeReloaderCh chan struct{} + // closeReloaderCh is created up front rather than when the reloader starts, so that + // Close never races with Start assigning it. + closeReloaderCh chan struct{} + // reloaderStarted means reload calls may now come from the reloader, which is worth a + // log line; the initial load is not. + reloaderStarted bool } func newFileDataSourceImpl( @@ -42,7 +37,7 @@ func newFileDataSourceImpl( duplicateKeysHandling DuplicateKeysHandling, reloaderFactory ReloaderFactory, ) (subsystems.DataSource, error) { - abs, err := absFilePaths(filePaths) + abs, err := filedata.AbsFilePaths(filePaths) if err != nil { // COVERAGE: there's no reliable cross-platform way to simulate an invalid path in unit tests return nil, err @@ -54,6 +49,7 @@ func newFileDataSourceImpl( duplicateKeysHandling: duplicateKeysHandling, reloaderFactory: reloaderFactory, loggers: context.GetLogging().Loggers, + closeReloaderCh: make(chan struct{}), } fs.loggers.SetPrefix("FileDataSource:") return fs, nil @@ -76,7 +72,7 @@ func (fs *fileDataSource) Start(closeWhenReady chan<- struct{}) { // If there is a reloader, and if we haven't yet successfully loaded data, then the // readiness signal will happen the first time we do get valid data (in reload). - fs.closeReloaderCh = make(chan struct{}) + fs.reloaderStarted = true err := fs.reloaderFactory(fs.absFilePaths, fs.loggers, fs.reload, fs.closeReloaderCh) if err != nil { fs.loggers.Errorf("Unable to start reloader: %s\n", err) @@ -87,14 +83,14 @@ func (fs *fileDataSource) Start(closeWhenReady chan<- struct{}) { // and update the feature flag state. If any file cannot be loaded or parsed, the flag state will not // be modified. func (fs *fileDataSource) reload() { - if fs.closeReloaderCh != nil { + if fs.reloaderStarted { fs.loggers.Info("Reloading flag data after detecting a change") } - filesData := make([]fileData, 0) + docs := make([]filedata.Document, 0) for _, path := range fs.absFilePaths { - data, err := readFile(path) + doc, err := filedata.ReadFile(path) if err == nil { - filesData = append(filesData, data) + docs = append(docs, doc) } else { fs.loggers.Errorf("Unable to load flags: %s [%s]", err, path) fs.dataSourceUpdates.UpdateStatus(interfaces.DataSourceStateInterrupted, @@ -106,8 +102,12 @@ func (fs *fileDataSource) reload() { return } } - storeData, err := mergeFileData(fs.duplicateKeysHandling, filesData...) + merged, err := filedata.Merge(filedata.DuplicateKeysHandling(fs.duplicateKeysHandling), docs...) if err == nil { + storeData := []ldstoretypes.Collection{ + {Kind: datakinds.Features, Items: merged.Flags}, + {Kind: datakinds.Segments, Items: merged.Segments}, + } if fs.dataSourceUpdates.Init(storeData) { fs.signalStartComplete(true) fs.dataSourceUpdates.UpdateStatus(interfaces.DataSourceStateValid, interfaces.DataSourceErrorInfo{}) @@ -134,126 +134,10 @@ func (fs *fileDataSource) signalStartComplete(succeeded bool) { }) } -func absFilePaths(paths []string) ([]string, error) { - absPaths := make([]string, 0) - for _, p := range paths { - absPath, err := filepath.Abs(p) - if err != nil { - // COVERAGE: there's no reliable cross-platform way to simulate an invalid path in unit tests - return nil, fmt.Errorf("unable to determine absolute path for '%s'", p) - } - absPaths = append(absPaths, absPath) - } - return absPaths, nil -} - -type fileData struct { - Flags *map[string]ldmodel.FeatureFlag - FlagValues *map[string]ldvalue.Value - Segments *map[string]ldmodel.Segment -} - -func insertData( - all map[ldstoretypes.DataKind]map[string]ldstoretypes.ItemDescriptor, - kind ldstoretypes.DataKind, - key string, - data ldstoretypes.ItemDescriptor, - duplicateKeysHandling DuplicateKeysHandling, -) error { - if _, exists := all[kind][key]; exists { - switch duplicateKeysHandling { - case DuplicateKeysIgnoreAllButFirst: - return nil - default: - return fmt.Errorf("%s '%s' is specified by multiple files", kind, key) - } - } - all[kind][key] = data - return nil -} - -func readFile(path string) (fileData, error) { - var data fileData - var rawData []byte - var err error - if rawData, err = os.ReadFile(path); err != nil { //nolint:gosec // G304: ok to read file into variable - return data, fmt.Errorf("unable to read file: %s", err) - } - if detectJSON(rawData) { - err = json.Unmarshal(rawData, &data) - } else { - err = yaml.Unmarshal(rawData, &data) - } - if err != nil { - err = fmt.Errorf("error parsing file: %s", err) - } - return data, err -} - -func detectJSON(rawData []byte) bool { - // A valid JSON file for our purposes must be an object, i.e. it must start with '{' - return strings.HasPrefix(strings.TrimLeftFunc(string(rawData), unicode.IsSpace), "{") -} - -func mergeFileData( - duplicateKeysHandling DuplicateKeysHandling, - allFileData ...fileData, -) ([]ldstoretypes.Collection, error) { - all := map[ldstoretypes.DataKind]map[string]ldstoretypes.ItemDescriptor{ - datakinds.Features: {}, - datakinds.Segments: {}, - } - for _, d := range allFileData { - if d.Flags != nil { - for key, f := range *d.Flags { - ff := f - data := ldstoretypes.ItemDescriptor{Version: f.Version, Item: &ff} - if err := insertData(all, datakinds.Features, key, data, duplicateKeysHandling); err != nil { - return nil, err - } - } - } - if d.FlagValues != nil { - for key, value := range *d.FlagValues { - flag := makeFlagWithValue(key, value) - data := ldstoretypes.ItemDescriptor{Version: flag.Version, Item: flag} - if err := insertData(all, datakinds.Features, key, data, duplicateKeysHandling); err != nil { - return nil, err - } - } - } - if d.Segments != nil { - for key, s := range *d.Segments { - ss := s - data := ldstoretypes.ItemDescriptor{Version: s.Version, Item: &ss} - if err := insertData(all, datakinds.Segments, key, data, duplicateKeysHandling); err != nil { - return nil, err - } - } - } - } - ret := []ldstoretypes.Collection{} - for kind, itemsMap := range all { - items := make([]ldstoretypes.KeyedItemDescriptor, 0, len(itemsMap)) - for k, v := range itemsMap { - items = append(items, ldstoretypes.KeyedItemDescriptor{Key: k, Item: v}) - } - ret = append(ret, ldstoretypes.Collection{Kind: kind, Items: items}) - } - return ret, nil -} - -func makeFlagWithValue(key string, v interface{}) *ldmodel.FeatureFlag { - flag := ldbuilders.NewFlagBuilder(key).SingleVariation(ldvalue.CopyArbitraryValue(v)).Build() - return &flag -} - // Close is called automatically when the client is closed. func (fs *fileDataSource) Close() (err error) { fs.closeOnce.Do(func() { - if fs.closeReloaderCh != nil { - close(fs.closeReloaderCh) - } + close(fs.closeReloaderCh) }) return nil } diff --git a/ldfiledata/file_data_source_test.go b/ldfiledata/file_data_source_test.go index 04ca4553..b0553b41 100644 --- a/ldfiledata/file_data_source_test.go +++ b/ldfiledata/file_data_source_test.go @@ -4,6 +4,7 @@ import ( "errors" "os" "testing" + "time" "github.com/launchdarkly/go-server-sdk/v7/internal/sharedtest/mocks" @@ -253,3 +254,29 @@ func requireSegment(t *testing.T, store subsystems.DataStore, key string) *ldmod require.NotNil(t, item.Item) return item.Item.(*ldmodel.Segment) } + +func TestCloseStopsReloader(t *testing.T) { + th.WithTempFileData([]byte(`{"flags": {"my-flag": {"on": true}}}`), func(filename string) { + reloaderCloseCh := make(chan (<-chan struct{}), 1) + f := func(paths []string, loggers ldlog.Loggers, reload func(), closeCh <-chan struct{}) error { + reloaderCloseCh <- closeCh + return nil + } + + factory := DataSource().FilePaths(filename).Reloader(f) + withFileDataSourceTestParams(factory, func(p fileDataSourceTestParams) { + p.waitForStart() + + closeCh := <-reloaderCloseCh + assert.NoError(t, p.dataSource.Close()) + + // The channel given to the reloader must be closed, so that whatever goroutines the + // reloader started can terminate. + select { + case <-closeCh: + case <-time.After(time.Second): + assert.Fail(t, "reloader close channel was not closed by Close()") + } + }) + }) +} diff --git a/ldfiledatav2/file_data_source_impl.go b/ldfiledatav2/file_data_source_impl.go index fb177d55..daec3a50 100644 --- a/ldfiledatav2/file_data_source_impl.go +++ b/ldfiledatav2/file_data_source_impl.go @@ -2,42 +2,38 @@ package ldfiledatav2 import ( "context" - "encoding/json" "errors" "fmt" - "os" - "path/filepath" - "strings" - "sync" "sync/atomic" "time" - "unicode" "github.com/launchdarkly/go-sdk-common/v3/ldlog" - "github.com/launchdarkly/go-sdk-common/v3/ldvalue" - "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldbuilders" - "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldmodel" "github.com/launchdarkly/go-server-sdk/v7/interfaces" "github.com/launchdarkly/go-server-sdk/v7/internal" + "github.com/launchdarkly/go-server-sdk/v7/internal/filedata" "github.com/launchdarkly/go-server-sdk/v7/subsystems" "github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoreimpl" "github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes" - - "gopkg.in/ghodss/yaml.v1" ) type fileDataSource struct { changeSetBroadcaster *internal.Broadcaster[subsystems.ChangeSet] statusBroadcaster *internal.Broadcaster[interfaces.DataSynchronizerStatus] // NOTE: this is not really used anymore because file data sources at this - // moment will not report a selector. - version int + // moment will not report a selector. It is atomic because loads can happen + // concurrently from Fetch and from the reloader. + version atomic.Int64 absFilePaths []string duplicateKeysHandling DuplicateKeysHandling reloaderFactory ReloaderFactory loggers ldlog.Loggers - closeReloaderCh chan struct{} + // closeReloaderCh is created up front rather than when the reloader starts, so that + // Close never races with Sync assigning it. + closeReloaderCh chan struct{} + // reloaderStarted means reload calls may now come from the reloader, which is worth a + // log line; the initial load is not. + reloaderStarted bool closed atomic.Bool quit chan struct{} @@ -49,7 +45,7 @@ func newFileDataSourceImpl( duplicateKeysHandling DuplicateKeysHandling, reloaderFactory ReloaderFactory, ) (subsystems.DataSynchronizer, error) { - abs, err := absFilePaths(filePaths) + abs, err := filedata.AbsFilePaths(filePaths) if err != nil { // COVERAGE: there's no reliable cross-platform way to simulate an invalid path in unit tests return nil, err @@ -62,6 +58,7 @@ func newFileDataSourceImpl( duplicateKeysHandling: duplicateKeysHandling, reloaderFactory: reloaderFactory, loggers: context.GetLogging().Loggers, + closeReloaderCh: make(chan struct{}), quit: make(chan struct{}), } fs.loggers.SetPrefix("FileDataSource:") @@ -91,6 +88,7 @@ func (fs *fileDataSource) Sync(ds subsystems.DataSelector) <-chan subsystems.Dat fs.reload() if fs.reloaderFactory != nil { + fs.reloaderStarted = true err := fs.reloaderFactory(fs.absFilePaths, fs.loggers, fs.reload, fs.closeReloaderCh) if err != nil { fs.loggers.Errorf("Unable to start reloader: %s\n", err) @@ -136,233 +134,99 @@ func (fs *fileDataSource) Sync(ds subsystems.DataSelector) <-chan subsystems.Dat } func (fs *fileDataSource) Fetch(ds subsystems.DataSelector, ctx context.Context) (*subsystems.Basis, bool, error) { - changeSetChan := fs.changeSetBroadcaster.AddListener() - statusChan := fs.statusBroadcaster.AddListener() - - changeset := subsystems.NewChangeSetBuilder().NoChanges() - - var err error - basis := &subsystems.Basis{ - ChangeSet: *changeset, - Persist: false, + changeSet, err := fs.load() + if err != nil { + return nil, false, err } - - var wg sync.WaitGroup - wg.Add(1) - - go func() { - fs.reload() - - defer wg.Done() - - select { - case changeSet, ok := <-changeSetChan: - if !ok { - return - } - basis.ChangeSet = changeSet - return - case statusChange, ok := <-statusChan: - if !ok { - return - } - - if statusChange.State != interfaces.DataSourceStateValid { - err = errors.New("data source did not receive change set") - return - } - default: - return - } - }() - - wg.Wait() - - return basis, false, err + return &subsystems.Basis{ + ChangeSet: *changeSet, + Persist: false, + }, false, nil } // Reload tells the data source to immediately attempt to reread all of the configured source files // and update the feature flag state. If any file cannot be loaded or parsed, the flag state will not // be modified. func (fs *fileDataSource) reload() { - if fs.closeReloaderCh != nil { + if fs.reloaderStarted { fs.loggers.Info("Reloading flag data after detecting a change") } - filesData := make([]fileData, 0) - for _, path := range fs.absFilePaths { - data, err := readFile(path) - if err == nil { - filesData = append(filesData, data) - } else { - fs.loggers.Errorf("Unable to load flags: %s [%s]", err, path) - fs.statusBroadcaster.Broadcast(interfaces.DataSynchronizerStatus{ - State: interfaces.DataSourceStateInterrupted, - Error: interfaces.DataSourceErrorInfo{ - Kind: interfaces.DataSourceErrorKindUnknown, - StatusCode: 0, - Message: err.Error(), - Time: time.Time{}, - }, - }) - return - } - } - - fs.version++ - changeSet, err := mergeFileData(fs.duplicateKeysHandling, fs.version, filesData...) - + changeSet, err := fs.load() if err == nil { fs.changeSetBroadcaster.Broadcast(*changeSet) } else { + fs.loggers.Errorf("Unable to load flags: %s", err) + errorKind := interfaces.DataSourceErrorKindInvalidData + var readErr *fileReadError + if errors.As(err, &readErr) { + errorKind = interfaces.DataSourceErrorKindUnknown + } fs.statusBroadcaster.Broadcast(interfaces.DataSynchronizerStatus{ State: interfaces.DataSourceStateInterrupted, Error: interfaces.DataSourceErrorInfo{ - Kind: interfaces.DataSourceErrorKindInvalidData, + Kind: errorKind, StatusCode: 0, Message: err.Error(), Time: time.Time{}, }, FallbackToFDv1: false, }) - fs.loggers.Error(err) } } -func absFilePaths(paths []string) ([]string, error) { - absPaths := make([]string, 0) - for _, p := range paths { - absPath, err := filepath.Abs(p) - if err != nil { - // COVERAGE: there's no reliable cross-platform way to simulate an invalid path in unit tests - return nil, fmt.Errorf("unable to determine absolute path for '%s'", p) - } - absPaths = append(absPaths, absPath) - } - return absPaths, nil +// fileReadError distinguishes a failure to read or parse one of the source files from a +// failure to merge their contents. +type fileReadError struct { + err error + path string } -type fileData struct { - Flags *map[string]ldmodel.FeatureFlag - FlagValues *map[string]ldvalue.Value - Segments *map[string]ldmodel.Segment +func (e *fileReadError) Error() string { + return fmt.Sprintf("%s [%s]", e.err, e.path) } -func insertDataIntoCollection( - items *[]ldstoretypes.KeyedItemDescriptor, - seenKeys map[subsystems.ObjectKind]map[string]bool, - objectKind subsystems.ObjectKind, - key string, - data ldstoretypes.ItemDescriptor, - duplicateKeysHandling DuplicateKeysHandling, -) error { - if _, exists := seenKeys[objectKind][key]; exists { - switch duplicateKeysHandling { - case DuplicateKeysIgnoreAllButFirst: - return nil - default: - return fmt.Errorf("%s '%s' is specified by multiple files", objectKind, key) - } - } - - *items = append(*items, ldstoretypes.KeyedItemDescriptor{ - Key: key, - Item: data, - }) - seenKeys[objectKind][key] = true - - return nil +func (e *fileReadError) Unwrap() error { + return e.err } -func readFile(path string) (fileData, error) { - var data fileData - var rawData []byte - var err error - if rawData, err = os.ReadFile(path); err != nil { //nolint:gosec // G304: ok to read file into variable - return data, fmt.Errorf("unable to read file: %s", err) - } - if detectJSON(rawData) { - err = json.Unmarshal(rawData, &data) - } else { - err = yaml.Unmarshal(rawData, &data) +// load synchronously reads and merges all of the configured source files, returning the +// result as a full-transfer change set. +func (fs *fileDataSource) load() (*subsystems.ChangeSet, error) { + docs := make([]filedata.Document, 0) + for _, path := range fs.absFilePaths { + doc, err := filedata.ReadFile(path) + if err != nil { + return nil, &fileReadError{err: err, path: path} + } + docs = append(docs, doc) } + + merged, err := filedata.Merge(filedata.DuplicateKeysHandling(fs.duplicateKeysHandling), docs...) if err != nil { - err = fmt.Errorf("error parsing file: %s", err) + return nil, err } - return data, err -} -func detectJSON(rawData []byte) bool { - // A valid JSON file for our purposes must be an object, i.e. it must start with '{' - return strings.HasPrefix(strings.TrimLeftFunc(string(rawData), unicode.IsSpace), "{") -} - -func mergeFileData( - duplicateKeysHandling DuplicateKeysHandling, - version int, - allFileData ...fileData, -) (*subsystems.ChangeSet, error) { intent := subsystems.ServerIntent{ Payload: subsystems.Payload{ ID: "", - Target: version, + Target: int(fs.version.Add(1)), Code: subsystems.IntentTransferFull, Reason: "payload-missing", }, } - // Build collections directly instead of using ChangeSetBuilder - flagItems := make([]ldstoretypes.KeyedItemDescriptor, 0) - segmentItems := make([]ldstoretypes.KeyedItemDescriptor, 0) - - seenKeys := map[subsystems.ObjectKind]map[string]bool{ - subsystems.FlagKind: {}, - subsystems.SegmentKind: {}, - } - - for _, d := range allFileData { - if d.Flags != nil { - for key, f := range *d.Flags { - data := ldstoretypes.ItemDescriptor{Version: f.Version, Item: &f} - err := insertDataIntoCollection(&flagItems, seenKeys, subsystems.FlagKind, key, data, duplicateKeysHandling) - if err != nil { - return nil, err - } - } - } - if d.FlagValues != nil { - for key, value := range *d.FlagValues { - flag := makeFlagWithValue(key, value) - data := ldstoretypes.ItemDescriptor{Version: flag.Version, Item: flag} - err := insertDataIntoCollection(&flagItems, seenKeys, subsystems.FlagKind, key, data, duplicateKeysHandling) - if err != nil { - return nil, err - } - } - } - if d.Segments != nil { - for key, s := range *d.Segments { - data := ldstoretypes.ItemDescriptor{Version: s.Version, Item: &s} - err := insertDataIntoCollection(&segmentItems, seenKeys, subsystems.SegmentKind, key, data, duplicateKeysHandling) - if err != nil { - return nil, err - } - } - } - } - - // Build collections collections := make([]ldstoretypes.Collection, 0, 2) - if len(flagItems) > 0 { + if len(merged.Flags) > 0 { collections = append(collections, ldstoretypes.Collection{ Kind: ldstoreimpl.Features(), - Items: flagItems, + Items: merged.Flags, }) } - if len(segmentItems) > 0 { + if len(merged.Segments) > 0 { collections = append(collections, ldstoretypes.Collection{ Kind: ldstoreimpl.Segments(), - Items: segmentItems, + Items: merged.Segments, }) } @@ -373,19 +237,11 @@ func mergeFileData( return subsystems.NewChangeSetFromCollections(intent, subsystems.NoSelector(), collections) } -func makeFlagWithValue(key string, v interface{}) *ldmodel.FeatureFlag { - flag := ldbuilders.NewFlagBuilder(key).SingleVariation(ldvalue.CopyArbitraryValue(v)).Build() - return &flag -} - // Close is called automatically when the client is closed. func (fs *fileDataSource) Close() (err error) { if swapped := fs.closed.CompareAndSwap(false, true); swapped { close(fs.quit) - - if fs.closeReloaderCh != nil { - close(fs.closeReloaderCh) - } + close(fs.closeReloaderCh) return nil // already closed } diff --git a/ldfiledatav2/file_data_source_test.go b/ldfiledatav2/file_data_source_test.go index a5d42d09..210944ed 100644 --- a/ldfiledatav2/file_data_source_test.go +++ b/ldfiledatav2/file_data_source_test.go @@ -315,3 +315,31 @@ func TestInitializerReturnsErrorIfFileDoesNotExist(t *testing.T) { assert.Error(t, err) }) } + +func TestCloseStopsReloader(t *testing.T) { + th.WithTempFileData([]byte(`{"flags": {"my-flag": {"on": true}}}`), func(filename string) { + reloaderCloseCh := make(chan (<-chan struct{}), 1) + f := func(paths []string, loggers ldlog.Loggers, reload func(), closeCh <-chan struct{}) error { + reloaderCloseCh <- closeCh + return nil + } + + factory := DataSource().FilePaths(filename).Reloader(f) + sync, err := factory.Build(subsystems.BasicClientContext{}) + assert.NoError(t, err) + + resultChan := sync.Sync(mocks.NewMockDataSelector(subsystems.NoSelector())) + <-resultChan + + closeCh := <-reloaderCloseCh + assert.NoError(t, sync.Close()) + + // The channel given to the reloader must be closed, so that whatever goroutines the + // reloader started can terminate. + select { + case <-closeCh: + case <-time.After(time.Second): + assert.Fail(t, "reloader close channel was not closed by Close()") + } + }) +}