podinfo/pkg/fscache/fscache.go

package fscache

import (
	"errors"
	"io/ioutil"
	"log"
	"path/filepath"
	"strings"
	"sync"

	"github.com/fsnotify/fsnotify"
)

type Watcher struct {
	dir       string
	fswatcher *fsnotify.Watcher
	Cache     *sync.Map
}

// NewWatch creates a directory watcher and
// updates the cache when any file changes in that dir
func NewWatch(dir string) (*Watcher, error) {
	if len(dir) < 1 {
		return nil, errors.New("directory is empty")
	}

	fw, err := fsnotify.NewWatcher()
	if err != nil {
		return nil, err
	}

	w := &Watcher{
		dir:       dir,
		fswatcher: fw,
		Cache:     new(sync.Map),
	}

	log.Printf("fscache start watcher for %s", w.dir)
	err = w.fswatcher.Add(w.dir)
	if err != nil {
		return nil, err
	}

	// initial read
	err = w.updateCache()
	if err != nil {
		return nil, err
	}

	return w, nil
}

// Watch watches for when kubelet updates the volume mount content
func (w *Watcher) Watch() {
	go func() {
		for {
			select {
			// it can take up to a 2 minutes for kubelet to recreate the ..data symlink
			case event := <-w.fswatcher.Events:
				if event.Op&fsnotify.Create == fsnotify.Create {
					if filepath.Base(event.Name) == "..data" {
						err := w.updateCache()
						if err != nil {
							log.Printf("fscache update error %v", err)
						} else {
							log.Printf("fscache reload %s", w.dir)
						}
					}
				}
			case err := <-w.fswatcher.Errors:
				log.Printf("fswatcher %s error %v", w.dir, err)
			}
		}
	}()
}

// updateCache reads files content and loads them into the cache
func (w *Watcher) updateCache() error {
	fileMap := make(map[string]string)
	files, err := ioutil.ReadDir(w.dir)
	if err != nil {
		return err
	}

	// read files ignoring symlinks and sub directories
	for _, file := range files {
		name := filepath.Base(file.Name())
		if !file.IsDir() && !strings.Contains(name, "..") {
			b, err := ioutil.ReadFile(filepath.Join(w.dir, file.Name()))
			if err != nil {
				return err
			}
			fileMap[name] = string(b)
		}
	}

	// remove deleted files from cache
	w.Cache.Range(func(key interface{}, value interface{}) bool {
		_, ok := fileMap[key.(string)]
		if !ok {
			w.Cache.Delete(key)
		}
		return true
	})

	// sync cache
	for k, v := range fileMap {
		w.Cache.Store(k, v)
	}

	return nil
}