feat: Use fsnotify to detect model/policy files change in casbin plugin

plugins/go.mod

@@ -27,6 +27,7 @@ require (
 	github.com/casbin/casbin/v2 v2.88.0
 	github.com/coreos/go-oidc/v3 v3.10.0
 	github.com/envoyproxy/envoy v1.29.4
+	github.com/fsnotify/fsnotify v1.7.0
 	github.com/google/cel-go v0.20.1
 	github.com/google/uuid v1.6.0
 	github.com/gorilla/securecookie v1.1.2

plugins/go.sum

@@ -60,6 +60,8 @@ github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8
 github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
 github.com/foxcpp/go-mockdns v1.1.0 h1:jI0rD8M0wuYAxL7r/ynTrCQQq0BVqfB99Vgk7DlmewI=
 github.com/foxcpp/go-mockdns v1.1.0/go.mod h1:IhLeSFGed3mJIAXPH2aiRQB+kqz7oqu8ld2qVbOu7Wk=
+github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA=
+github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM=
 github.com/go-ini/ini v1.67.0 h1:z6ZrTEZqSWOTyH2FlglNbNgARyHG8oLW9gMELqKr06A=
 github.com/go-ini/ini v1.67.0/go.mod h1:ByCAeIL28uOIIG0E3PJtZPDL8WnHpFKFOtgjp+3Ies8=
 github.com/go-jose/go-jose/v4 v4.0.1 h1:QVEPDE3OluqXBQZDcnNvQrInro2h0e4eqNbnZSWqS6U=

plugins/pkg/file/fs.go

@@ -15,11 +15,13 @@
 package file
 
 import (
+	"errors"
+	"fmt"
 	"os"
 	"sync"
 	"time"
 
-	"github.com/jellydator/ttlcache/v3"
+	"github.com/fsnotify/fsnotify"
 
 	"mosn.io/htnn/api/pkg/log"
 )
@@ -48,65 +50,87 @@
 	f.lock.Unlock()
 }
 
-type fs struct {
-	cache *ttlcache.Cache[string, os.FileInfo]
+type Fsnotify struct {
+	Watcher *fsnotify.Watcher
 }
 
-func newFS(ttl time.Duration) *fs {
-	loader := ttlcache.LoaderFunc[string, os.FileInfo](
-		func(c *ttlcache.Cache[string, os.FileInfo], key string) *ttlcache.Item[string, os.FileInfo] {
-			info, err := os.Stat(key)
-			if err != nil {
-				logger.Error(err, "reload file info to cache", "file", key)
-				return nil
-			}
-			item := c.Set(key, info, ttlcache.DefaultTTL)
-			logger.Info("update file mtime", "file", key, "mtime", item.Value().ModTime())
-			return item
-		},
-	)
-	cache := ttlcache.New(
-		ttlcache.WithTTL[string, os.FileInfo](ttl),
-		ttlcache.WithLoader[string, os.FileInfo](loader),
-	)
-	go cache.Start()
-
-	return &fs{
-		cache: cache,
+func newFsnotify() (fs *Fsnotify) {
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		logger.Error(err, "create watcher failed")
+		return
 	}
+	fs = &Fsnotify{
+		Watcher: watcher,
+	}
+	return
 }
 
 var (
-	// TODO: rewrite it to use inotify
-	defaultFs = newFS(10 * time.Second)
+	defaultFsnotify = newFsnotify()
 )
 
-func IsChanged(files ...*File) bool {
+func Update(onChange func(), files ...*File) (err error) {
+	err = WatchFiles(onChange, files...)
+	return
+}
+
+func WatchFiles(onChange func(), files ...*File) (err error) {
+	if len(files) < 1 {
+		err = errors.New("must specify at least one file to watch")
+		logger.Error(err, "")
+		return
+	}
+
+	watcher := newFsnotify().Watcher
+	if err != nil {
+
+		logger.Error(err, "failed to create watcher")
+		return
+	}
+
+	// Add files to watcher.
 	for _, file := range files {
-		changed := defaultFs.isChanged(file)
-		if changed {
-			return true
-		}
+		go defaultFsnotify.watchFiles(onChange, watcher, file)
 	}
-	return false
+
+	return
 }
 
-func (f *fs) isChanged(file *File) bool {
-	item := f.cache.Get(file.Name)
-	if item == nil {
-		// As a protection, failed to fetch the real file means file not changed
-		return false
+func (f *Fsnotify) watchFiles(onChange func(), w *fsnotify.Watcher, files *File) {
+	defer func(w *fsnotify.Watcher) {
+		err := w.Close()
+		if err != nil {
+			logger.Error(err, "failed to close fsnotify watcher")
+		}
+	}(w)
+	err := w.Add(files.Name)
+	if err != nil {
+		logger.Error(err, "add file to watcher failed")
+	}
+	for {
+		select {
+		case event, ok := <-w.Events:
+			if !ok {
+				return
+			}
+			logger.Info(fmt.Sprintf("event: %v", event))
+			onChange()
+			return
+		case err, ok := <-w.Errors:
+			if !ok {
+				return
+			}
+			logger.Error(err, "error watching files")
+		}
 	}
-
-	return file.Mtime().Before(item.Value().ModTime())
 }
 
-func (f *fs) Stat(path string) (*File, error) {
+func (f *Fsnotify) Stat(path string) (*File, error) {
 	info, err := os.Stat(path)
 	if err != nil {
 		return nil, err
 	}
-	f.cache.Set(path, info, ttlcache.DefaultTTL)
 
 	return &File{
 		Name:  path,
@@ -115,24 +139,5 @@
 }
 
 func Stat(path string) (*File, error) {
-	return defaultFs.Stat(path)
-}
-
-func Update(files ...*File) bool {
-	for _, file := range files {
-		if !defaultFs.update(file) {
-			return false
-		}
-	}
-	return true
-}
-
-func (f *fs) update(file *File) bool {
-	item := f.cache.Get(file.Name)
-	if item == nil {
-		return false
-	}
-
-	file.SetMtime(item.Value().ModTime())
-	return true
+	return defaultFsnotify.Stat(path)
 }

plugins/pkg/file/fs_test.go

@@ -16,28 +16,45 @@ package file
 
 import (
 	"os"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 )
 
-func TestFileMtimeDetection(t *testing.T) {
-	defaultFs = newFS(2000 * time.Millisecond)
+func TestFileIsChanged(t *testing.T) {
+	var mu sync.Mutex
+	i := 1
+	tmpfile, _ := os.CreateTemp("./", "example")
+	defer func(name string) {
+		err := os.Remove(name)
+		if err != nil {
+			t.Logf("%v", err)
+		}
+	}(tmpfile.Name())
 
-	tmpfile, _ := os.CreateTemp("", "example")
-	defer os.Remove(tmpfile.Name()) // clean up
-
-	f, err := Stat(tmpfile.Name())
-	assert.Nil(t, err)
-	assert.False(t, IsChanged(f))
-	time.Sleep(1000 * time.Millisecond)
+	file := &File{Name: tmpfile.Name()}
+	_ = WatchFiles(func() {
+		mu.Lock()
+		i = 2
+		mu.Unlock()
+	}, file)
+	time.Sleep(1 * time.Millisecond)
 	tmpfile.Write([]byte("bls"))
-	tmpfile.Close()
-	assert.False(t, IsChanged(f))
+	tmpfile.Sync()
+	mu.Lock()
+	assert.Equal(t, 2, i)
+	mu.Unlock()
 
-	time.Sleep(2500 * time.Millisecond)
-	assert.True(t, IsChanged(f))
-	assert.True(t, Update(f))
-	assert.False(t, IsChanged(f))
+	_ = WatchFiles(func() {
+		mu.Lock()
+		i = 1
+		mu.Unlock()
+	}, file)
+	time.Sleep(1 * time.Millisecond)
+	tmpfile.Sync()
+	mu.Lock()
+	assert.Equal(t, 2, i)
+	mu.Unlock()
 }

plugins/plugins/casbin/filter.go

@@ -35,13 +35,9 @@
 	config    *config
 }
 
-func (f *filter) DecodeHeaders(headers api.RequestHeaderMap, endStream bool) api.ResultAction {
+func (f *filter) reloadEnforcer() {
 	conf := f.config
-	role, _ := headers.Get(conf.Token.Name) // role can be ""
-	url := headers.Url()
-
-	policyChanged := file.IsChanged(conf.modelFile, conf.policyFile)
-	if policyChanged && !conf.updating.Load() {
+	if !conf.updating.Load() {
 		conf.updating.Store(true)
 		api.LogWarnf("policy %s or model %s changed, reload enforcer", conf.policyFile.Name, conf.modelFile.Name)
 
@@ -58,11 +54,28 @@
 				conf.enforcer = e
 				conf.lock.Unlock()
 
-				file.Update(conf.modelFile, conf.policyFile)
+				err = file.Update(func() {
+					f.reloadEnforcer()
+				}, conf.modelFile, conf.policyFile)
+				if err != nil {
+					api.LogErrorf("failed to update Enforcer: %v", err)
+				}
 				api.LogWarnf("policy %s or model %s changed, enforcer reloaded", conf.policyFile.Name, conf.modelFile.Name)
 			}
 		}()
 	}
+}
+func (f *filter) DecodeHeaders(headers api.RequestHeaderMap, endStream bool) api.ResultAction {
+	conf := f.config
+	role, _ := headers.Get(conf.Token.Name) // role can be ""
+	url := headers.Url()
+
+	err := file.WatchFiles(f.reloadEnforcer, conf.modelFile, conf.policyFile)
+
+	if err != nil {
+		api.LogErrorf("failed to watch files: %v", err)
+		return &api.LocalResponse{Code: 500}
+	}
 
 	conf.lock.RLock()
 	ok, err := f.config.enforcer.Enforce(role, url.Path, headers.Method())