Skip to content

Commit

Permalink
feat: add 'watcher' interface to file sync (#1365)
Browse files Browse the repository at this point in the history
Implement fsnotify and `os.Stat` based watchers

fixes: #1344

<!-- Please use this template for your pull request. -->
<!-- Please use the sections that you need and delete other sections -->

## This PR
Intent of this PR is to begin a conversation about fixing #1344. The
approach taken is to replace the current use of `fsontify.Watcher` with
a local `Watcher` interface type that describes the `fsnotify.Watcher`
interface.

My original take was to use fsnotify.Watcher directly as an
implementation of local `Watcher`, but fsnotify's Watcher directly
exposes its Error and Event channels, making it impossible to describe
with an interface, so I had to create a small wrapper for
`fsnotify.Watcher` to satisfy the new Watcher interface (this is
fsnotify_watcher.go). From there, we implement the `Watcher` interface
again, this time using `os.Stat` and `fs.FileInfo` (this is
fileinfo_watcher.go).

Then we change the filepath sync code to use an interface to Watcher,
rather than fsnotify.Watcher directly. The new fileinfo watcher plugs
right in, and nothing really needs to change in the sync.

* I have not wired up configs, so the fileinfo watcher has a hard-coded
1-second polling interval, and there is no current means of selecting
between them.
* I've added a couple tests, to demonstrate how unit tests would work in
general (we use a configurable os-stat func in the fileinfo watcher,
which can be mocked for tests)
* I don't have a way of testing this on Windows. I'm vaguely aware
there's an upstream issue in package `fs` that may require some
work-around boilerplate to make this work on windows at the moment.

If yall are favorable to this approach, I'll finish wiring up configs,
and flesh out the tests. I didn't want to go much further without some
buy-in or feedback.

### Related Issues

Fixes #1344 

### Notes
See bullet-points above

### How to test
go test -v ./...

---------

Signed-off-by: Dave Josephsen <dave.josephsen@gmail.com>
Signed-off-by: Kavindu Dodanduwa <kavindudodanduwa@gmail.com>
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
Co-authored-by: Kavindu Dodanduwa <Kavindu-Dodan@users.noreply.github.com>
Co-authored-by: Michael Beemer <beeme1mr@users.noreply.github.com>
  • Loading branch information
4 people authored Aug 22, 2024
1 parent abb5ca3 commit 61fff43
Show file tree
Hide file tree
Showing 6 changed files with 617 additions and 26 deletions.
52 changes: 42 additions & 10 deletions core/pkg/sync/builder/syncbuilder.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@ import (
"net/http"
"os"
"regexp"
msync "sync"
"time"

"github.com/open-feature/flagd/core/pkg/logger"
Expand All @@ -26,6 +25,8 @@ import (

const (
syncProviderFile = "file"
syncProviderFsNotify = "fsnotify"
syncProviderFileInfo = "fileinfo"
syncProviderGrpc = "grpc"
syncProviderKubernetes = "kubernetes"
syncProviderHTTP = "http"
Expand Down Expand Up @@ -91,8 +92,13 @@ func (sb *SyncBuilder) SyncsFromConfig(sourceConfigs []sync.SourceConfig, logger
func (sb *SyncBuilder) syncFromConfig(sourceConfig sync.SourceConfig, logger *logger.Logger) (sync.ISync, error) {
switch sourceConfig.Provider {
case syncProviderFile:
logger.Debug(fmt.Sprintf("using filepath sync-provider for: %q", sourceConfig.URI))
return sb.newFile(sourceConfig.URI, logger), nil
case syncProviderFsNotify:
logger.Debug(fmt.Sprintf("using fsnotify sync-provider for: %q", sourceConfig.URI))
return sb.newFsNotify(sourceConfig.URI, logger), nil
case syncProviderFileInfo:
logger.Debug(fmt.Sprintf("using fileinfo sync-provider for: %q", sourceConfig.URI))
return sb.newFileInfo(sourceConfig.URI, logger), nil
case syncProviderKubernetes:
logger.Debug(fmt.Sprintf("using kubernetes sync-provider for: %s", sourceConfig.URI))
return sb.newK8s(sourceConfig.URI, logger)
Expand All @@ -107,20 +113,46 @@ func (sb *SyncBuilder) syncFromConfig(sourceConfig sync.SourceConfig, logger *lo
return sb.newGcs(sourceConfig, logger), nil

default:
return nil, fmt.Errorf("invalid sync provider: %s, must be one of with '%s', '%s', '%s' or '%s'",
sourceConfig.Provider, syncProviderFile, syncProviderKubernetes, syncProviderHTTP, syncProviderKubernetes)
return nil, fmt.Errorf("invalid sync provider: %s, must be one of with '%s', '%s', '%s', %s', '%s' or '%s'",
sourceConfig.Provider, syncProviderFile, syncProviderFsNotify, syncProviderFileInfo,
syncProviderKubernetes, syncProviderHTTP, syncProviderKubernetes)
}
}

// newFile returns an fsinfo sync if we are in k8s or fileinfo if not
func (sb *SyncBuilder) newFile(uri string, logger *logger.Logger) *file.Sync {
return &file.Sync{
URI: regFile.ReplaceAllString(uri, ""),
Logger: logger.WithFields(
switch os.Getenv("KUBERNETES_SERVICE_HOST") {
case "":
// no k8s service host env; use fileinfo
return sb.newFileInfo(uri, logger)
default:
// default to fsnotify
return sb.newFsNotify(uri, logger)
}
}

// return a new file.Sync that uses fsnotify under the hood
func (sb *SyncBuilder) newFsNotify(uri string, logger *logger.Logger) *file.Sync {
return file.NewFileSync(
regFile.ReplaceAllString(uri, ""),
file.FSNOTIFY,
logger.WithFields(
zap.String("component", "sync"),
zap.String("sync", "filepath"),
zap.String("sync", syncProviderFsNotify),
),
Mux: &msync.RWMutex{},
}
)
}

// return a new file.Sync that uses os.Stat/fs.FileInfo under the hood
func (sb *SyncBuilder) newFileInfo(uri string, logger *logger.Logger) *file.Sync {
return file.NewFileSync(
regFile.ReplaceAllString(uri, ""),
file.FILEINFO,
logger.WithFields(
zap.String("component", "sync"),
zap.String("sync", syncProviderFileInfo),
),
)
}

func (sb *SyncBuilder) newK8s(uri string, logger *logger.Logger) (*kubernetes.Sync, error) {
Expand Down
202 changes: 202 additions & 0 deletions core/pkg/sync/file/fileinfo_watcher.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,202 @@
package file

import (
"context"
"errors"
"fmt"
"io/fs"
"os"
"sync"
"time"

"github.com/fsnotify/fsnotify"
"github.com/open-feature/flagd/core/pkg/logger"
)

// Implements file.Watcher using a timer and os.FileInfo
type fileInfoWatcher struct {
// Event Chan
evChan chan fsnotify.Event
// Errors Chan
erChan chan error
// logger
logger *logger.Logger
// Func to wrap os.Stat (injection point for test helpers)
statFunc func(string) (fs.FileInfo, error)
// thread-safe interface to underlying files we are watching
mu sync.RWMutex
watches map[string]fs.FileInfo // filename -> info
}

// NewFsNotifyWatcher returns a new fsNotifyWatcher
func NewFileInfoWatcher(ctx context.Context, logger *logger.Logger) Watcher {
fiw := &fileInfoWatcher{
evChan: make(chan fsnotify.Event, 32),
erChan: make(chan error, 32),
statFunc: getFileInfo,
logger: logger,
watches: make(map[string]fs.FileInfo),
}
fiw.run(ctx, (1 * time.Second))
return fiw
}

// fileInfoWatcher explicitly implements file.Watcher
var _ Watcher = &fileInfoWatcher{}

// Close calls close on the underlying fsnotify.Watcher
func (f *fileInfoWatcher) Close() error {
// close all channels and exit
close(f.evChan)
close(f.erChan)
return nil
}

// Add calls Add on the underlying fsnotify.Watcher
func (f *fileInfoWatcher) Add(name string) error {
f.mu.Lock()
defer f.mu.Unlock()

// exit early if name already exists
if _, ok := f.watches[name]; ok {
return nil
}

info, err := f.statFunc(name)
if err != nil {
return err
}

f.watches[name] = info

return nil
}

// Remove calls Remove on the underlying fsnotify.Watcher
func (f *fileInfoWatcher) Remove(name string) error {
f.mu.Lock()
defer f.mu.Unlock()

// no need to exit early, deleting non-existent key is a no-op
delete(f.watches, name)

return nil
}

// Watchlist calls watchlist on the underlying fsnotify.Watcher
func (f *fileInfoWatcher) WatchList() []string {
f.mu.RLock()
defer f.mu.RUnlock()
out := []string{}
for name := range f.watches {
n := name
out = append(out, n)
}
return out
}

// Events returns the underlying watcher's Events chan
func (f *fileInfoWatcher) Events() chan fsnotify.Event {
return f.evChan
}

// Errors returns the underlying watcher's Errors chan
func (f *fileInfoWatcher) Errors() chan error {
return f.erChan
}

// run is a blocking function that starts the filewatcher's timer thread
func (f *fileInfoWatcher) run(ctx context.Context, s time.Duration) {
// timer thread
go func() {
// execute update on the configured interval of time
ticker := time.NewTicker(s)
defer ticker.Stop()

for {
select {
case <-ctx.Done():
return
case <-ticker.C:
if err := f.update(); err != nil {
f.erChan <- err
return
}
}
}
}()
}

func (f *fileInfoWatcher) update() error {
f.mu.Lock()
defer f.mu.Unlock()

for path, info := range f.watches {
newInfo, err := f.statFunc(path)
if err != nil {
// if the file isn't there, it must have been removed
// fire off a remove event and remove it from the watches
if errors.Is(err, os.ErrNotExist) {
f.evChan <- fsnotify.Event{
Name: path,
Op: fsnotify.Remove,
}
delete(f.watches, path)
continue
}
return err
}

// if the new stat doesn't match the old stat, figure out what changed
if info != newInfo {
event := f.generateEvent(path, newInfo)
if event != nil {
f.evChan <- *event
}
f.watches[path] = newInfo
}
}
return nil
}

// generateEvent figures out what changed and generates an fsnotify.Event for it. (if we care)
// file removal are handled above in the update() method
func (f *fileInfoWatcher) generateEvent(path string, newInfo fs.FileInfo) *fsnotify.Event {
info := f.watches[path]
switch {
// new mod time is more recent than old mod time, generate a write event
case newInfo.ModTime().After(info.ModTime()):
return &fsnotify.Event{
Name: path,
Op: fsnotify.Write,
}
// the file modes changed, generate a chmod event
case info.Mode() != newInfo.Mode():
return &fsnotify.Event{
Name: path,
Op: fsnotify.Chmod,
}
// nothing changed that we care about
default:
return nil
}
}

// getFileInfo returns the fs.FileInfo for the given path
func getFileInfo(path string) (fs.FileInfo, error) {
f, err := os.Open(path)
if err != nil {
return nil, fmt.Errorf("error from os.Open(%s): %w", path, err)
}

info, err := f.Stat()
if err != nil {
return info, fmt.Errorf("error from fs.Stat(%s): %w", path, err)
}

if err := f.Close(); err != nil {
return info, fmt.Errorf("err from fs.Close(%s): %w", path, err)
}

return info, nil
}
Loading

0 comments on commit 61fff43

Please sign in to comment.