Merge pull request #438 from bilibili/common/update-paladin

update paladin

pkg/conf/paladin/file.go

@@ -2,80 +2,122 @@ package paladin
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"io/ioutil"
 	"log"
 	"os"
 	"path"
 	"path/filepath"
+	"strings"
 	"sync"
 	"time"
 
 	"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
-	wmu      sync.RWMutex
-	notify   *fsnotify.Watcher
-	watchers map[*watcher]struct{}
+	values *Map
+	rawVal map[string]*Value
+
+	watchChs map[string][]chan Event
+	mx       sync.Mutex
+	wg       sync.WaitGroup
+
+	base string
+	done chan struct{}
+}
+
+func isHiddenFile(name string) bool {
+	// TODO: support windows.
+	return strings.HasPrefix(filepath.Base(name), ".")
+}
+
+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() && !isHiddenFile(file.Name()) {
+				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
 }
 
 // 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)
-	raws, err := loadValues(base)
+
+	paths, err := readAllPaths(base)
 	if err != nil {
 		return nil, err
 	}
-	notify, err := fsnotify.NewWatcher()
+	if len(paths) == 0 {
+		return nil, fmt.Errorf("empty config path")
+	}
+
+	rawVal, err := loadValuesFromPaths(paths)
 	if err != nil {
 		return nil, err
 	}
-	values := new(Map)
-	values.Store(raws)
-	f := &file{
-		values:   values,
-		notify:   notify,
-		watchers: make(map[*watcher]struct{}),
+
+	valMap := &Map{}
+	valMap.Store(rawVal)
+	fc := &file{
+		values:   valMap,
+		rawVal:   rawVal,
+		watchChs: make(map[string][]chan Event),
+
+		base: base,
+		done: make(chan struct{}, 1),
 	}
-	go f.watchproc(base)
-	return f, nil
+
+	fc.wg.Add(1)
+	go fc.daemon()
+
+	return fc, nil
 }
 
 // Get return value by key.
@@ -88,109 +130,74 @@ func (f *file) GetAll() *Map {
 	return f.values
 }
 
-// WatchEvent watch with the specified keys.
+// WatchEvent watch multi key.
 func (f *file) WatchEvent(ctx context.Context, keys ...string) <-chan Event {
-	w := newWatcher(keys)
-	f.wmu.Lock()
-	f.watchers[w] = struct{}{}
-	f.wmu.Unlock()
-	return w.C
+	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
 }
 
 // Close close watcher.
 func (f *file) Close() error {
-	if err := f.notify.Close(); err != nil {
-		return err
-	}
-	f.wmu.RLock()
-	for w := range f.watchers {
-		close(w.C)
-	}
-	f.wmu.RUnlock()
+	f.done <- struct{}{}
+	f.wg.Wait()
 	return nil
 }
 
 // file config daemon to watch file modification
-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)
+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)
 		return
 	}
-	log.Printf("paladin: start watch config: %s", base)
-	for event := range f.notify.Events {
+	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 {
 		// use vim edit config will trigger rename
-		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)
-			}
+		case fsnotify.Write, fsnotify.Create:
+			f.reloadFile(event.Name)
+		case fsnotify.Chmod:
 		default:
-			log.Printf("paladin: unsupport event %s ingored", event)
+			log.Printf("unsupport event %s ingored", event)
 		}
 	}
 }
 
-func (f *file) reloadFile(fpath string) (err error) {
+func (f *file) reloadFile(name string) {
+	if isHiddenFile(name) {
+		return
+	}
 	// 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)
-	value, err := loadValue(fpath)
+	key := filepath.Base(name)
+	val, err := loadValue(name)
 	if err != nil {
+		log.Printf("load file %s error: %s, skipped", name, err)
 		return
 	}
-	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
-}
+	f.rawVal[key] = val
+	f.values.Store(f.rawVal)
 
-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() && (file.Mode()&os.ModeSymlink) != os.ModeSymlink {
-				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
-}
+	f.mx.Lock()
+	chs := f.watchChs[key]
+	f.mx.Unlock()
 
-func loadValue(name string) (*Value, error) {
-	data, err := ioutil.ReadFile(name)
-	if err != nil {
-		return nil, err
+	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)
+		}
 	}
-	content := string(data)
-	return &Value{val: content, raw: content}, nil
 }

pkg/conf/paladin/file_test.go

@@ -82,9 +82,8 @@ func TestFileEvent(t *testing.T) {
 	cli, err := NewFile(path)
 	assert.Nil(t, err)
 	assert.NotNil(t, cli)
-	time.Sleep(time.Millisecond * 100)
 	ch := cli.WatchEvent(context.Background(), "test.toml", "abc.toml")
-	time.Sleep(time.Millisecond * 100)
+	time.Sleep(time.Millisecond)
 	ioutil.WriteFile(path+"test.toml", []byte(`hello`), 0644)
 	timeout := time.NewTimer(time.Second)
 	select {
@@ -94,4 +93,39 @@ func TestFileEvent(t *testing.T) {
 		assert.Equal(t, EventUpdate, ev.Event)
 		assert.Equal(t, "hello", ev.Value)
 	}
+	ioutil.WriteFile(path+"abc.toml", []byte(`test`), 0644)
+	select {
+	case <-timeout.C:
+		t.Fatalf("run test timeout")
+	case ev := <-ch:
+		assert.Equal(t, EventUpdate, ev.Event)
+		assert.Equal(t, "test", ev.Value)
+	}
+	content1, _ := cli.Get("test.toml").String()
+	assert.Equal(t, "hello", content1)
+	content2, _ := cli.Get("abc.toml").String()
+	assert.Equal(t, "test", content2)
+}
+
+func TestHiddenFile(t *testing.T) {
+	path := "/tmp/test_hidden_event/"
+	assert.Nil(t, os.MkdirAll(path, 0700))
+	assert.Nil(t, ioutil.WriteFile(path+"test.toml", []byte(`hello`), 0644))
+	assert.Nil(t, ioutil.WriteFile(path+".abc.toml", []byte(`
+		text = "hello"	
+		number = 100
+	`), 0644))
+	// test client
+	// test client
+	cli, err := NewFile(path)
+	assert.Nil(t, err)
+	assert.NotNil(t, cli)
+	cli.WatchEvent(context.Background(), "test.toml")
+	time.Sleep(time.Millisecond)
+	ioutil.WriteFile(path+".abc.toml", []byte(`hello`), 0644)
+	time.Sleep(time.Second)
+	content1, _ := cli.Get("test.toml").String()
+	assert.Equal(t, "hello", content1)
+	_, err = cli.Get(".abc.toml").String()
+	assert.NotNil(t, err)
 }