pharnexbase
/
kratos
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix paladin filewatch

Browse Source
pull/93/head
Tony 6 years ago
parent fcba425d39
commit 0968936285
3 changed files with 80 additions and 102 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 26
      pkg/conf/paladin/default.go
  2. 149
      pkg/conf/paladin/file.go
  3. 7
      pkg/conf/paladin/map.go

26
pkg/conf/paladin/default.go
Unescape View File

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

149
pkg/conf/paladin/file.go
Unescape View File

@ -2,6 +2,7 @@ package paladin
import (
"context"
"errors"
"fmt"
"io/ioutil"
"log"
@ -22,95 +23,40 @@ var _ Client = &file{}
// file is file config client.
type file struct {
values *Map
rawVal map[string]*Value
baseDir string
values *Map
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
done chan 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)
if err != nil {
return nil, err
}
if len(paths) == 0 {
return nil, fmt.Errorf("empty config path")
return nil, errors.New("empty config path")
}
rawVal, err := loadValuesFromPaths(paths)
raws, err := loadValuesFromPaths(paths)
if err != nil {
return nil, err
}
valMap := &Map{}
valMap.Store(rawVal)
values := new(Map)
values.Store(raws)
fc := &file{
values: valMap,
rawVal: rawVal,
baseDir: base,
values: values,
watchChs: make(map[string][]chan Event),
base: base,
done: make(chan struct{}, 1),
done: make(chan struct{}, 1),
}
fc.wg.Add(1)
go fc.daemon()
return fc, nil
}
@ -147,22 +93,21 @@ 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)
log.Printf("paladin: create file watcher fail! reload function will lose efficacy error: %s", 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)
if err = fswatcher.Add(f.baseDir); err != nil {
log.Printf("paladin: create fsnotify for base path %s fail %s, reload function will lose efficacy", f.baseDir, err)
return
}
log.Printf("start watch filepath: %s", f.base)
log.Printf("paladin: start watch config: %s", f.baseDir)
for event := range fswatcher.Events {
switch event.Op {
// use vim edit config will trigger rename
case fsnotify.Write, fsnotify.Create:
switch {
case event.Op&fsnotify.Write == fsnotify.Write, event.Op&fsnotify.Create == fsnotify.Create:
f.reloadFile(event.Name)
case fsnotify.Chmod:
default:
log.Printf("unsupport event %s ingored", event)
log.Printf("paladin: unsupport event %s ingored", event)
}
}
}
@ -172,23 +117,65 @@ func (f *file) reloadFile(name string) {
// 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(name)
if err != nil {
log.Printf("load file %s error: %s, skipped", name, err)
log.Printf("paladin: load file: %s error: %s, skipped", name, err)
return
}
f.rawVal[key] = val
f.values.Store(f.rawVal)
raws := f.values.Load()
raws[name] = value
f.values.Store(raws)
f.mx.Lock()
chs := f.watchChs[key]
f.mx.Unlock()
for _, ch := range chs {
select {
case ch <- Event{Event: EventUpdate, Value: val.raw}:
case ch <- Event{Event: EventUpdate, Value: value.raw}:
default:
log.Printf("event channel full discard file %s update event", name)
log.Printf("paladin: event channel full discard file %s update event", name)
}
}
log.Printf("paladin: reload config: %s notify: %d\n", name, len(chs))
}
func readAllPaths(base string) ([]string, 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)
}
return paths, nil
}
func loadValuesFromPaths(paths []string) (map[string]*Value, error) {
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
}

7
pkg/conf/paladin/map.go
Unescape View File

@ -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.

Write Preview
Loading…
Cancel
Save