Merge pull request #93 from bilibili/fix-paladin-filewatch

fix paladin filewatch

pkg/conf/paladin/default.go

@@ -3,15 +3,12 @@ package paladin
 import (
 	"context"
 	"flag"
-
-	"github.com/bilibili/kratos/pkg/log"
 )
 
 var (
 	// DefaultClient default client.
 	DefaultClient Client
 	confPath      string
-	vars          = make(map[string][]Setter) // NOTE: no thread safe
 )
 
 func init() {
@@ -23,27 +20,12 @@ func Init() (err error) {
 	if confPath != "" {
 		DefaultClient, err = NewFile(confPath)
 	} else {
-		// TODO: config service
+		// TODO: Get the configuration from the remote service
 		panic("Please specify a file or dir name by -conf flag.")
-		return
 	}
 	if err != nil {
 		return
 	}
-	go func() {
-		for event := range DefaultClient.WatchEvent(context.Background()) {
-			if event.Event != EventUpdate && event.Event != EventAdd {
-				continue
-			}
-			if sets, ok := vars[event.Key]; ok {
-				for _, s := range sets {
-					if err := s.Set(event.Value); err != nil {
-						log.Error("paladin: vars:%v event:%v error(%v)", s, event, err)
-					}
-				}
-			}
-		}
-	}()
 	return
 }
 
@@ -57,7 +39,11 @@ func Watch(key string, s Setter) error {
 	if err := s.Set(str); err != nil {
 		return err
 	}
-	vars[key] = append(vars[key], s)
+	go func() {
+		for event := range WatchEvent(context.Background(), key) {
+			s.Set(event.Value)
+		}
+	}()
 	return nil
 }
 

pkg/conf/paladin/file.go

@@ -2,6 +2,7 @@ package paladin
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"log"
@@ -14,104 +15,67 @@ import (
 	"github.com/fsnotify/fsnotify"
 )
 
-const (
-	defaultChSize = 10
-)
-
 var _ Client = &file{}
 
+type watcher struct {
+	keys []string
+	C    chan Event
+}
+
+func newWatcher(keys []string) *watcher {
+	return &watcher{keys: keys, C: make(chan Event, 5)}
+}
+
+func (w *watcher) HasKey(key string) bool {
+	if len(w.keys) == 0 {
+		return true
+	}
+	for _, k := range w.keys {
+		if keyNamed(k) == key {
+			return true
+		}
+	}
+	return false
+}
+
+func (w *watcher) Handle(event Event) {
+	select {
+	case w.C <- event:
+	default:
+		log.Printf("paladin: event channel full discard file %s update event", event.Key)
+	}
+}
+
 // file is file config client.
 type file struct {
-	values *Map
-	rawVal map[string]*Value
-
-	watchChs map[string][]chan Event
-	mx       sync.Mutex
-	wg       sync.WaitGroup
-
-	base string
-	done chan struct{}
-}
-
-func readAllPaths(base string) ([]string, error) {
-	fi, err := os.Stat(base)
-	if err != nil {
-		return nil, fmt.Errorf("check local config file fail! error: %s", err)
-	}
-	// dirs or file to paths
-	var paths []string
-	if fi.IsDir() {
-		files, err := ioutil.ReadDir(base)
-		if err != nil {
-			return nil, fmt.Errorf("read dir %s error: %s", base, err)
-		}
-		for _, file := range files {
-			if !file.IsDir() {
-				paths = append(paths, path.Join(base, file.Name()))
-			}
-		}
-	} else {
-		paths = append(paths, base)
-	}
-	return paths, nil
-}
-
-func loadValuesFromPaths(paths []string) (map[string]*Value, error) {
-	// laod config file to values
-	var err error
-	values := make(map[string]*Value, len(paths))
-	for _, fpath := range paths {
-		if values[path.Base(fpath)], err = loadValue(fpath); err != nil {
-			return nil, err
-		}
-	}
-	return values, nil
-}
-
-func loadValue(fpath string) (*Value, error) {
-	data, err := ioutil.ReadFile(fpath)
-	if err != nil {
-		return nil, err
-	}
-	content := string(data)
-	return &Value{val: content, raw: content}, nil
+	values   *Map
+	wmu      sync.RWMutex
+	notify   *fsnotify.Watcher
+	watchers map[*watcher]struct{}
 }
 
 // NewFile new a config file client.
 // conf = /data/conf/app/
 // conf = /data/conf/app/xxx.toml
 func NewFile(base string) (Client, error) {
-	// paltform slash
 	base = filepath.FromSlash(base)
-
-	paths, err := readAllPaths(base)
+	raws, err := loadValues(base)
 	if err != nil {
 		return nil, err
 	}
-	if len(paths) == 0 {
-		return nil, fmt.Errorf("empty config path")
-	}
-
-	rawVal, err := loadValuesFromPaths(paths)
+	notify, err := fsnotify.NewWatcher()
 	if err != nil {
 		return nil, err
 	}
-
-	valMap := &Map{}
-	valMap.Store(rawVal)
-	fc := &file{
-		values:   valMap,
-		rawVal:   rawVal,
-		watchChs: make(map[string][]chan Event),
-
-		base: base,
-		done: make(chan struct{}, 1),
+	values := new(Map)
+	values.Store(raws)
+	f := &file{
+		values:   values,
+		notify:   notify,
+		watchers: make(map[*watcher]struct{}),
 	}
-
-	fc.wg.Add(1)
-	go fc.daemon()
-
-	return fc, nil
+	go f.watchproc(base)
+	return f, nil
 }
 
 // Get return value by key.
@@ -124,71 +88,109 @@ func (f *file) GetAll() *Map {
 	return f.values
 }
 
-// WatchEvent watch multi key.
+// WatchEvent watch with the specified keys.
 func (f *file) WatchEvent(ctx context.Context, keys ...string) <-chan Event {
-	f.mx.Lock()
-	defer f.mx.Unlock()
-	ch := make(chan Event, defaultChSize)
-	for _, key := range keys {
-		f.watchChs[key] = append(f.watchChs[key], ch)
-	}
-	return ch
+	w := newWatcher(keys)
+	f.wmu.Lock()
+	f.watchers[w] = struct{}{}
+	f.wmu.Unlock()
+	return w.C
 }
 
 // Close close watcher.
 func (f *file) Close() error {
-	f.done <- struct{}{}
-	f.wg.Wait()
+	if err := f.notify.Close(); err != nil {
+		return err
+	}
+	f.wmu.RLock()
+	for w := range f.watchers {
+		close(w.C)
+	}
+	f.wmu.RUnlock()
 	return nil
 }
 
 // file config daemon to watch file modification
-func (f *file) daemon() {
-	defer f.wg.Done()
-	fswatcher, err := fsnotify.NewWatcher()
-	if err != nil {
-		log.Printf("create file watcher fail! reload function will lose efficacy error: %s", err)
+func (f *file) watchproc(base string) {
+	if err := f.notify.Add(base); err != nil {
+		log.Printf("paladin: create fsnotify for base path %s fail %s, reload function will lose efficacy", base, err)
 		return
 	}
-	if err = fswatcher.Add(f.base); err != nil {
-		log.Printf("create fsnotify for base path %s fail %s, reload function will lose efficacy", f.base, err)
-		return
-	}
-	log.Printf("start watch filepath: %s", f.base)
-	for event := range fswatcher.Events {
-		switch event.Op {
+	log.Printf("paladin: start watch config: %s", base)
+	for event := range f.notify.Events {
 		// use vim edit config will trigger rename
-		case fsnotify.Write, fsnotify.Create:
-			f.reloadFile(event.Name)
-		case fsnotify.Chmod:
+		switch {
+		case event.Op&fsnotify.Write == fsnotify.Write, event.Op&fsnotify.Create == fsnotify.Create:
+			if err := f.reloadFile(event.Name); err != nil {
+				log.Printf("paladin: load file: %s error: %s, skipped", event.Name, err)
+			}
 		default:
-			log.Printf("unsupport event %s ingored", event)
+			log.Printf("paladin: unsupport event %s ingored", event)
 		}
 	}
 }
 
-func (f *file) reloadFile(name string) {
+func (f *file) reloadFile(fpath string) (err error) {
 	// NOTE: in some case immediately read file content after receive event
 	// will get old content, sleep 100ms make sure get correct content.
 	time.Sleep(100 * time.Millisecond)
-	key := filepath.Base(name)
-	val, err := loadValue(name)
+	value, err := loadValue(fpath)
 	if err != nil {
-		log.Printf("load file %s error: %s, skipped", name, err)
 		return
 	}
-	f.rawVal[key] = val
-	f.values.Store(f.rawVal)
-
-	f.mx.Lock()
-	chs := f.watchChs[key]
-	f.mx.Unlock()
-
-	for _, ch := range chs {
-		select {
-		case ch <- Event{Event: EventUpdate, Value: val.raw}:
-		default:
-			log.Printf("event channel full discard file %s update event", name)
+	key := keyNamed(path.Base(fpath))
+	raws := f.values.Load()
+	raws[key] = value
+	f.values.Store(raws)
+	f.wmu.RLock()
+	n := 0
+	for w := range f.watchers {
+		if w.HasKey(key) {
+			n++
+			w.Handle(Event{Event: EventUpdate, Key: key, Value: value.raw})
 		}
 	}
+	f.wmu.RUnlock()
+	log.Printf("paladin: reload config: %s events: %d\n", key, n)
+	return
+}
+
+func loadValues(base string) (map[string]*Value, error) {
+	fi, err := os.Stat(base)
+	if err != nil {
+		return nil, fmt.Errorf("paladin: check local config file fail! error: %s", err)
+	}
+	var paths []string
+	if fi.IsDir() {
+		files, err := ioutil.ReadDir(base)
+		if err != nil {
+			return nil, fmt.Errorf("paladin: read dir %s error: %s", base, err)
+		}
+		for _, file := range files {
+			if !file.IsDir() {
+				paths = append(paths, path.Join(base, file.Name()))
+			}
+		}
+	} else {
+		paths = append(paths, base)
+	}
+	if len(paths) == 0 {
+		return nil, errors.New("empty config path")
+	}
+	values := make(map[string]*Value, len(paths))
+	for _, fpath := range paths {
+		if values[path.Base(fpath)], err = loadValue(fpath); err != nil {
+			return nil, err
+		}
+	}
+	return values, nil
+}
+
+func loadValue(name string) (*Value, error) {
+	data, err := ioutil.ReadFile(name)
+	if err != nil {
+		return nil, err
+	}
+	content := string(data)
+	return &Value{val: content, raw: content}, nil
 }

pkg/conf/paladin/map.go

@@ -26,7 +26,12 @@ func (m *Map) Store(values map[string]*Value) {
 
 // Load returns the value set by the most recent Store.
 func (m *Map) Load() map[string]*Value {
-	return m.values.Load().(map[string]*Value)
+	src := m.values.Load().(map[string]*Value)
+	dst := make(map[string]*Value, len(src))
+	for k, v := range src {
+		dst[k] = v
+	}
+	return dst
 }
 
 // Exist check if values map exist a key.