Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[fswatcher] File Reconciler w/ Resync Interval #149

Merged
merged 2 commits into from
Sep 16, 2024
Merged

[fswatcher] File Reconciler w/ Resync Interval #149

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
60b8dc2
[fswatcher] File Reconciler w/ Resync Interval
onelapahead Sep 10, 2024
16921a8
fix lint
onelapahead Sep 10, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 33 additions & 4 deletions pkg/fswatcher/fswatcher.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// Copyright © 2023 Kaleido, Inc.
// Copyright © 2024 Kaleido, Inc.
//
// SPDX-License-Identifier: Apache-2.0
//
Expand All @@ -20,6 +20,7 @@ import (
"context"
"os"
"path"
"time"

"github.com/fsnotify/fsnotify"
"github.com/hyperledger/firefly-common/pkg/fftypes"
Expand All @@ -35,9 +36,19 @@ import (
// - Only fires if the data in the file is different to the last notification
// - Does not reload the config - that's the caller's responsibility
func Watch(ctx context.Context, fullFilePath string, onChange, onClose func()) error {
return sync(ctx, fullFilePath, onChange, onClose, nil, nil)
}

// Reconcile behaves the same as Watch, except it allows for running the onSync func on a provided
// interval. The default re-sync internal is 1m.
func Reconcile(ctx context.Context, fullFilePath string, onChange, onClose, onSync func(), resyncInterval *time.Duration) error {
return sync(ctx, fullFilePath, onChange, onClose, onSync, resyncInterval)
}

func sync(ctx context.Context, fullFilePath string, onChange, onClose, onSync func(), resyncInterval *time.Duration) error {
filePath := path.Dir(fullFilePath)
fileName := path.Base(fullFilePath)
log.L(ctx).Debugf("Starting file listener for '%s' in directory '%s'", fileName, filePath)
log.L(ctx).Debugf("Starting file reconciler for '%s' in directory '%s'", fileName, filePath)

watcher, err := fsnotify.NewWatcher()
if err == nil {
Expand All @@ -46,7 +57,7 @@ func Watch(ctx context.Context, fullFilePath string, onChange, onClose func()) e
if onClose != nil {
onClose()
}
}, watcher.Events, watcher.Errors)
}, onSync, resyncInterval, watcher.Events, watcher.Errors)
err = watcher.Add(filePath)
}
if err != nil {
Expand All @@ -56,9 +67,18 @@ func Watch(ctx context.Context, fullFilePath string, onChange, onClose func()) e
return nil
}

func fsListenerLoop(ctx context.Context, fullFilePath string, onChange, onClose func(), events chan fsnotify.Event, errors chan error) {
func fsListenerLoop(ctx context.Context, fullFilePath string, onChange, onClose, onSync func(), resyncInterval *time.Duration, events chan fsnotify.Event, errors chan error) {
defer onClose()

timeout := resyncInterval
if timeout == nil {
timeout = func() *time.Duration {
defaultTimeout := time.Minute
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

An alternative to this approach that we use a few places in FireFly codebases, is to wrap the context.
Generally it's a good idea for all for { select .... loops to include <-ctx.Done() so they will end if the context is terminated. This loop doesn't have that - which isn't caused by you, but does seem like an omission.

var cancelLoopCtx context.CancelFunc
tidyUpLoopCtx := func() {
  if cancelLoopCtx != nil {
       cancelLoopCtx();
   }
   cancelLoopCtx = nil;
}
for {
   loopCtx := ctx
   if resyncInterval != nil {
     loopCtx, cancelLoopCtx = context.WithTimeout(ctx, *resyncInterval)
  }
  select {
  ...
  case <-loopCtx.Done():
     // either we're exiting, or the timer popped
     select {
     case <-ctx.Done():
         tidyUpLoopCtx();
         log.L(ctx).Infof("exiting");
     default: // resync timer popped
     }
     tidyUpLoopCtx();
     // do the actual stuff.
  }

}

Copy link

@dwertent dwertent Sep 12, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I’d like to suggest a different approach here:
Using time.Ticker. The implementation would look like this:

ticker := time.NewTicker(time.Minute)
for {
	select {
	case <-ticker.C():
              // perform action
         case <-ctx.Done():
              ticker.Stop()
        }
}

(You should call the ticker.Stop() only if you want to reconcile the FS before exiting)

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How does either of these solutions handle the situation where there is another case statement which in this case is waiting for events that the file has been renamed, written too etc... This also acts as a "sync" I think the current proposed code has an issue that timer is actually not reset in this case if we want it to be.

In the first approach, you would just cancel the context when handling this case and go around the loop again with the timer reset. In the second approach, there is a ticker.Reset(d duration) so both approaches work.

Copy link
Contributor Author

@onelapahead onelapahead Sep 15, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks Enrique - yes it's really important we continue to sync the file over time in the case of Reconcile. The reason for these proposed alternatives is that the case <-time.After I've proposed is essentially optional in the Watch usecase. And so we'd occasionally do unnecessary CPU work when the timer pops, and go back around the loop to wait for the FS events, errors, or ctx to finish.

In a frequently changing FS, that could incur a performance hit. However, I don't believe this Watch is optimized for such rapid changes in the first place. A user would likely prefer to implement their own watcher in that case.

For the main case of a file which may changes on the scale of seconds or more, I don't think an extra case for this timeout would be detrimental, and I think the simplicity/readability of the code is preferred to a wrapper context/nested select statement which could have new bugs in it as Enrique pointed out we have to reset the loopCtx, etc.

So if its alright with the maintainers, I'd prefer to keep it as is.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So if its alright with the maintainers, I'd prefer to keep it as is.

ok - so just marking where we're landing with this PR:

  • This code will wake up every minute, for all existing installations. This is logically like a time.Ticker(), but we're using a time.After().
  • The code will wake up and just go straight back to sleep if no reconciler is configured

All the comments are around code style opinions, and (as per my original comment) I'm not laying down on the tracks for one particular style. So happy to merge.

return &defaultTimeout
}()
}
log.L(ctx).Debugf("re-sync interval set to '%s'", *timeout)

var lastHash *fftypes.Bytes32
for {
select {
Expand All @@ -83,6 +103,15 @@ func fsListenerLoop(ctx context.Context, fullFilePath string, onChange, onClose
lastHash = dataHash
}
}
case <-time.After(*timeout):
if onSync != nil {
data, err := os.ReadFile(fullFilePath)
if err == nil {
dataHash := fftypes.HashString(string(data))
log.L(ctx).Infof("Config file re-sync. Event=Resync Name=%s Size=%d Hash=%s", fullFilePath, len(data), dataHash)
onSync()
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm confused to see the onSync() function only being called if we hit the sync timeout but not when change is detected for example. I'm struggling to see the usefulness of an onSync() that takes zero arguments of when it synced to the calling function over an onChange that actually tells the consume that the file has changed.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But thats the thing - onChange also takes in no args, it expects the consumer to actually read the file themselves and do what they want with it.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, we can have a follow up issue to actually get the contents of the file back

}
case err, ok := <-errors:
if ok {
log.L(ctx).Errorf("FSEvent error: %s", err)
Expand Down
20 changes: 14 additions & 6 deletions pkg/fswatcher/fswatcher_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// Copyright © 2022 Kaleido, Inc.
// Copyright © 2024 Kaleido, Inc.
//
// SPDX-License-Identifier: Apache-2.0
//
Expand All @@ -21,14 +21,15 @@ import (
"fmt"
"os"
"testing"
"time"

"github.com/fsnotify/fsnotify"
"github.com/sirupsen/logrus"
"github.com/spf13/viper"
"github.com/stretchr/testify/assert"
)

func TestFileListenerE2E(t *testing.T) {
func TestFileReconcilerE2E(t *testing.T) {

logrus.SetLevel(logrus.DebugLevel)
tmpDir := t.TempDir()
Expand All @@ -47,14 +48,20 @@ func TestFileListenerE2E(t *testing.T) {
// Start listener on config file
fsListenerDone := make(chan struct{})
fsListenerFired := make(chan bool)
reSyncFired := make(chan bool)
reSyncInterval := 1 * time.Second
ctx, cancelCtx := context.WithCancel(context.Background())
err := Watch(ctx, filePath, func() {
err := Reconcile(ctx, filePath, func() {
err := viper.ReadInConfig()
assert.NoError(t, err)
fsListenerFired <- true
}, func() {
close(fsListenerDone)
})
}, func() {
err := viper.ReadInConfig()
assert.NoError(t, err)
reSyncFired <- true
}, &reSyncInterval)
assert.NoError(t, err)

// Delete and rename in another file
Expand All @@ -63,6 +70,7 @@ func TestFileListenerE2E(t *testing.T) {
os.Rename(fmt.Sprintf("%s/another.yaml", tmpDir), fmt.Sprintf("%s/test.yaml", tmpDir))
<-fsListenerFired
assert.Equal(t, "two", viper.Get("ut_conf"))
<-reSyncFired

defer func() {
cancelCtx()
Expand All @@ -74,7 +82,7 @@ func TestFileListenerE2E(t *testing.T) {

}

func TestFileListenerFail(t *testing.T) {
func TestFileWatcherFail(t *testing.T) {

logrus.SetLevel(logrus.DebugLevel)
tmpDir := t.TempDir()
Expand All @@ -95,7 +103,7 @@ func TestFileListenerLogError(t *testing.T) {
defer cancelCtx()
errors := make(chan error)
fsListenerDone := make(chan struct{})
go fsListenerLoop(ctx, "somefile", func() {}, func() { close(fsListenerDone) }, make(chan fsnotify.Event), errors)
go fsListenerLoop(ctx, "somefile", func() {}, func() { close(fsListenerDone) }, nil, nil, make(chan fsnotify.Event), errors)

errors <- fmt.Errorf("pop")
cancelCtx()
Expand Down
Loading