Skip to content

Commit 76d673a

Browse files
beeme1mrbeeme1mr
authored
feat: support yaml in blob, file, and http syncs (#1522)
Signed-off-by: Michael Beemer <beeme1mr@users.noreply.github.com>
1 parent 7a06567 commit 76d673a

13 files changed

+485
-290
lines changed

‎core/go.sum

+2-131
Large diffs are not rendered by default.

‎core/pkg/sync/blob/blob_sync.go

+15-4
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,16 @@
11
package blob
22

33
import (
4-
"bytes"
54
"context"
65
"errors"
76
"fmt"
7+
"io"
8+
"path/filepath"
89
"time"
910

1011
"github.com/open-feature/flagd/core/pkg/logger"
1112
"github.com/open-feature/flagd/core/pkg/sync"
13+
"github.com/open-feature/flagd/core/pkg/utils"
1214
"gocloud.dev/blob"
1315
_ "gocloud.dev/blob/azureblob" // needed to initialize Azure Blob Storage driver
1416
_ "gocloud.dev/blob/gcsblob" // needed to initialize GCS driver
@@ -126,11 +128,20 @@ func (hs *Sync) fetchObjectModificationTime(ctx context.Context, bucket *blob.Bu
126128
}
127129

128130
func (hs *Sync) fetchObject(ctx context.Context, bucket *blob.Bucket) (string, error) {
129-
buf := bytes.NewBuffer(nil)
130-
err := bucket.Download(ctx, hs.Object, buf, nil)
131+
r, err := bucket.NewReader(ctx, hs.Object, nil)
132+
if err != nil {
133+
return "", fmt.Errorf("error opening reader for object %s/%s: %w", hs.Bucket, hs.Object, err)
134+
}
135+
defer r.Close()
136+
137+
data, err := io.ReadAll(r)
131138
if err != nil {
132139
return "", fmt.Errorf("error downloading object %s/%s: %w", hs.Bucket, hs.Object, err)
133140
}
134141

135-
return buf.String(), nil
142+
json, err := utils.ConvertToJSON(data, filepath.Ext(hs.Object), r.ContentType())
143+
if err != nil {
144+
return "", fmt.Errorf("error converting blob data to json: %w", err)
145+
}
146+
return json, nil
136147
}

‎core/pkg/sync/blob/blob_sync_test.go

+73-48
Original file line numberDiff line numberDiff line change
@@ -12,57 +12,77 @@ import (
1212
"go.uber.org/mock/gomock"
1313
)
1414

15-
const (
16-
scheme = "xyz"
17-
bucket = "b"
18-
object = "o"
19-
)
20-
21-
func TestSync(t *testing.T) {
22-
ctrl := gomock.NewController(t)
23-
mockCron := synctesting.NewMockCron(ctrl)
24-
mockCron.EXPECT().AddFunc(gomock.Any(), gomock.Any()).DoAndReturn(func(spec string, cmd func()) error {
25-
return nil
26-
})
27-
mockCron.EXPECT().Start().Times(1)
28-
29-
blobSync := &Sync{
30-
Bucket: scheme + "://" + bucket,
31-
Object: object,
32-
Cron: mockCron,
33-
Logger: logger.NewLogger(nil, false),
15+
func TestBlobSync(t *testing.T) {
16+
tests := map[string]struct {
17+
scheme string
18+
bucket string
19+
object string
20+
content string
21+
convertedContent string
22+
}{
23+
"json file type": {
24+
scheme: "xyz",
25+
bucket: "b",
26+
object: "flags.json",
27+
content: "{\"flags\":{}}",
28+
convertedContent: "{\"flags\":{}}",
29+
},
30+
"yaml file type": {
31+
scheme: "xyz",
32+
bucket: "b",
33+
object: "flags.yaml",
34+
content: "flags: []",
35+
convertedContent: "{\"flags\":[]}",
36+
},
3437
}
35-
blobMock := NewMockBlob(scheme, func() *Sync {
36-
return blobSync
37-
})
38-
blobSync.BlobURLMux = blobMock.URLMux()
39-
40-
ctx := context.Background()
41-
dataSyncChan := make(chan sync.DataSync, 1)
42-
43-
config := "my-config"
44-
blobMock.AddObject(object, config)
4538

46-
go func() {
47-
err := blobSync.Sync(ctx, dataSyncChan)
48-
if err != nil {
49-
log.Fatalf("Error start sync: %s", err.Error())
50-
return
51-
}
52-
}()
53-
54-
data := <-dataSyncChan // initial sync
55-
if data.FlagData != config {
56-
t.Errorf("expected content: %s, but received content: %s", config, data.FlagData)
39+
for name, tt := range tests {
40+
t.Run(name, func(t *testing.T) {
41+
ctrl := gomock.NewController(t)
42+
mockCron := synctesting.NewMockCron(ctrl)
43+
mockCron.EXPECT().AddFunc(gomock.Any(), gomock.Any()).DoAndReturn(func(spec string, cmd func()) error {
44+
return nil
45+
})
46+
mockCron.EXPECT().Start().Times(1)
47+
48+
blobSync := &Sync{
49+
Bucket: tt.scheme + "://" + tt.bucket,
50+
Object: tt.object,
51+
Cron: mockCron,
52+
Logger: logger.NewLogger(nil, false),
53+
}
54+
blobMock := NewMockBlob(tt.scheme, func() *Sync {
55+
return blobSync
56+
})
57+
blobSync.BlobURLMux = blobMock.URLMux()
58+
59+
ctx := context.Background()
60+
dataSyncChan := make(chan sync.DataSync, 1)
61+
62+
blobMock.AddObject(tt.object, tt.content)
63+
64+
go func() {
65+
err := blobSync.Sync(ctx, dataSyncChan)
66+
if err != nil {
67+
log.Fatalf("Error start sync: %s", err.Error())
68+
return
69+
}
70+
}()
71+
72+
data := <-dataSyncChan // initial sync
73+
if data.FlagData != tt.convertedContent {
74+
t.Errorf("expected content: %s, but received content: %s", tt.convertedContent, data.FlagData)
75+
}
76+
tickWithConfigChange(t, mockCron, dataSyncChan, blobMock, tt.object, tt.convertedContent)
77+
tickWithoutConfigChange(t, mockCron, dataSyncChan)
78+
tickWithConfigChange(t, mockCron, dataSyncChan, blobMock, tt.object, tt.convertedContent)
79+
tickWithoutConfigChange(t, mockCron, dataSyncChan)
80+
tickWithoutConfigChange(t, mockCron, dataSyncChan)
81+
})
5782
}
58-
tickWithConfigChange(t, mockCron, dataSyncChan, blobMock, "new config")
59-
tickWithoutConfigChange(t, mockCron, dataSyncChan)
60-
tickWithConfigChange(t, mockCron, dataSyncChan, blobMock, "new config 2")
61-
tickWithoutConfigChange(t, mockCron, dataSyncChan)
62-
tickWithoutConfigChange(t, mockCron, dataSyncChan)
6383
}
6484

65-
func tickWithConfigChange(t *testing.T, mockCron *synctesting.MockCron, dataSyncChan chan sync.DataSync, blobMock *MockBlob, newConfig string) {
85+
func tickWithConfigChange(t *testing.T, mockCron *synctesting.MockCron, dataSyncChan chan sync.DataSync, blobMock *MockBlob, object string, newConfig string) {
6686
time.Sleep(1 * time.Millisecond) // sleep so the new file has different modification date
6787
blobMock.AddObject(object, newConfig)
6888
mockCron.Tick()
@@ -73,7 +93,7 @@ func tickWithConfigChange(t *testing.T, mockCron *synctesting.MockCron, dataSync
7393
t.Errorf("expected content: %s, but received content: %s", newConfig, data.FlagData)
7494
}
7595
} else {
76-
t.Errorf("data channel unexpecdly closed")
96+
t.Errorf("data channel unexpectedly closed")
7797
}
7898
default:
7999
t.Errorf("data channel has no expected update")
@@ -87,13 +107,18 @@ func tickWithoutConfigChange(t *testing.T, mockCron *synctesting.MockCron, dataS
87107
if ok {
88108
t.Errorf("unexpected update: %s", data.FlagData)
89109
} else {
90-
t.Errorf("data channel unexpecdly closed")
110+
t.Errorf("data channel unexpectedly closed")
91111
}
92112
default:
93113
}
94114
}
95115

96116
func TestReSync(t *testing.T) {
117+
const (
118+
scheme = "xyz"
119+
bucket = "b"
120+
object = "flags.json"
121+
)
97122
ctrl := gomock.NewController(t)
98123
mockCron := synctesting.NewMockCron(ctrl)
99124

‎core/pkg/sync/file/filepath_sync.go

+14-36
Original file line numberDiff line numberDiff line change
@@ -2,17 +2,17 @@ package file
22

33
import (
44
"context"
5-
"encoding/json"
65
"errors"
76
"fmt"
7+
"io"
88
"os"
9-
"strings"
9+
"path/filepath"
1010
msync "sync"
1111

1212
"github.com/fsnotify/fsnotify"
1313
"github.com/open-feature/flagd/core/pkg/logger"
1414
"github.com/open-feature/flagd/core/pkg/sync"
15-
"gopkg.in/yaml.v3"
15+
"github.com/open-feature/flagd/core/pkg/utils"
1616
)
1717

1818
const (
@@ -32,8 +32,6 @@ type Watcher interface {
3232
type Sync struct {
3333
URI string
3434
Logger *logger.Logger
35-
// FileType indicates the file type e.g., json, yaml/yml etc.,
36-
fileType string
3735
// watchType indicates how to watch the file FSNOTIFY|FILEINFO
3836
watchType string
3937
watcher Watcher
@@ -176,42 +174,22 @@ func (fs *Sync) fetch(_ context.Context) (string, error) {
176174
if fs.URI == "" {
177175
return "", errors.New("no filepath string set")
178176
}
179-
if fs.fileType == "" {
180-
uriSplit := strings.Split(fs.URI, ".")
181-
fs.fileType = uriSplit[len(uriSplit)-1]
182-
}
183-
rawFile, err := os.ReadFile(fs.URI)
184-
if err != nil {
185-
return "", fmt.Errorf("error reading file %s: %w", fs.URI, err)
186-
}
187-
188-
switch fs.fileType {
189-
case "yaml", "yml":
190-
return yamlToJSON(rawFile)
191-
case "json":
192-
return string(rawFile), nil
193-
default:
194-
return "", fmt.Errorf("filepath extension for URI: '%s' is not supported", fs.URI)
195-
}
196-
}
197177

198-
// yamlToJSON is a generic helper function to convert
199-
// yaml to json
200-
func yamlToJSON(rawFile []byte) (string, error) {
201-
if len(rawFile) == 0 {
202-
return "", nil
178+
file, err := os.Open(fs.URI)
179+
if err != nil {
180+
return "", fmt.Errorf("error opening file %s: %w", fs.URI, err)
203181
}
182+
defer file.Close()
204183

205-
var ms map[string]interface{}
206-
// yaml.Unmarshal unmarshals to map[interface]interface{}
207-
if err := yaml.Unmarshal(rawFile, &ms); err != nil {
208-
return "", fmt.Errorf("unmarshal yaml: %w", err)
184+
data, err := io.ReadAll(file)
185+
if err != nil {
186+
return "", fmt.Errorf("error reading file %s: %w", fs.URI, err)
209187
}
210188

211-
r, err := json.Marshal(ms)
189+
// File extension is used to determine the content type, so media type is unnecessary
190+
json, err := utils.ConvertToJSON(data, filepath.Ext(fs.URI), "")
212191
if err != nil {
213-
return "", fmt.Errorf("convert yaml to json: %w", err)
192+
return "", fmt.Errorf("error converting file content to json: %w", err)
214193
}
215-
216-
return string(r), err
194+
return json, nil
217195
}

‎core/pkg/sync/file/filepath_sync_test.go

+3-31
Original file line numberDiff line numberDiff line change
@@ -190,7 +190,7 @@ func TestSimpleSync(t *testing.T) {
190190

191191
func TestFilePathSync_Fetch(t *testing.T) {
192192
successDirName := t.TempDir()
193-
falureDirName := t.TempDir()
193+
failureDirName := t.TempDir()
194194
tests := map[string]struct {
195195
fpSync Sync
196196
handleResponse func(t *testing.T, fetched string, err error)
@@ -213,9 +213,9 @@ func TestFilePathSync_Fetch(t *testing.T) {
213213
},
214214
},
215215
"not found": {
216-
fetchDirName: falureDirName,
216+
fetchDirName: failureDirName,
217217
fpSync: Sync{
218-
URI: fmt.Sprintf("%s/%s", falureDirName, "not_found"),
218+
URI: fmt.Sprintf("%s/%s", failureDirName, "not_found"),
219219
Logger: logger.NewLogger(nil, false),
220220
},
221221
handleResponse: func(t *testing.T, fetched string, err error) {
@@ -309,31 +309,3 @@ func writeToFile(t *testing.T, fetchDirName, fileContents string) {
309309
t.Fatal(err)
310310
}
311311
}
312-
313-
func TestFilePathSync_yamlToJSON(t *testing.T) {
314-
tests := map[string]struct {
315-
input []byte
316-
handleResponse func(t *testing.T, output string, err error)
317-
}{
318-
"empty": {
319-
input: []byte(""),
320-
handleResponse: func(t *testing.T, output string, err error) {
321-
if err != nil {
322-
t.Fatalf("expect no err, got err = %v", err)
323-
}
324-
325-
if output != "" {
326-
t.Fatalf("expect output = '', got output = '%v'", output)
327-
}
328-
},
329-
},
330-
}
331-
332-
for name, tt := range tests {
333-
t.Run(name, func(t *testing.T) {
334-
output, err := yamlToJSON(tt.input)
335-
336-
tt.handleResponse(t, output, err)
337-
})
338-
}
339-
}

0 commit comments

Comments
 (0)