feat: add 'watcher' interface to file sync (#1365)

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 <[email protected]>
Signed-off-by: Kavindu Dodanduwa <[email protected]>
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
Co-authored-by: Kavindu Dodanduwa <[email protected]>
Co-authored-by: Michael Beemer <[email protected]>

Author: 4 people

core/pkg/sync/builder/syncbuilder.go

@@ -5,7 +5,6 @@ import (
 	"net/http"
 	"os"
 	"regexp"
-	msync "sync"
 	"time"
 
 	"github.com/open-feature/flagd/core/pkg/logger"
@@ -26,6 +25,8 @@ import (
 
 const (
 	syncProviderFile       = "file"
+	syncProviderFsNotify   = "fsnotify"
+	syncProviderFileInfo   = "fileinfo"
 	syncProviderGrpc       = "grpc"
 	syncProviderKubernetes = "kubernetes"
 	syncProviderHTTP       = "http"
@@ -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)
@@ -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) {

core/pkg/sync/file/fileinfo_watcher.go

@@ -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
+}