pkg/cdi: implement automatic refresh.

Implement automatic refresh and enable it by default. The new
WithAutoRefresh(bool) option can be used to explicitly change
the auto-refreshing behavior of the cache.

When auto-refresh is enabled, CDI Spec directories are monitored
for changes. A cache refresh is internally triggered whenever a
relevant change to a cache file or directory is detected.

Signed-off-by: Krisztian Litkey <krisztian.litkey@intel.com>

Author: klihub

pkg/cdi/cache.go

@@ -22,6 +22,7 @@ import (
 	"strings"
 	"sync"
 
+	"github.com/fsnotify/fsnotify"
 	"github.com/hashicorp/go-multierror"
 	oci "github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/pkg/errors"
@@ -33,30 +34,46 @@ type Option func(*Cache) error
 // Cache stores CDI Specs loaded from Spec directories.
 type Cache struct {
 	sync.Mutex
-	specDirs []string
-	specs    map[string][]*Spec
-	devices  map[string]*Device
-	errors   map[string][]error
+	specDirs  []string
+	specs     map[string][]*Spec
+	devices   map[string]*Device
+	errors    map[string][]error
+	dirErrors map[string]error
+
+	autoRefresh bool
+	watch       *watch
+}
+
+// WithAutoRefresh returns an option to control automatic Cache refresh.
+// By default auto-refresh is enabled, the list of Spec directories are
+// monitored and the Cache is automatically refreshed whenever a change
+// is detected. This option can be used to disable this behavior when a
+// manually refreshed mode is preferable.
+func WithAutoRefresh(autoRefresh bool) Option {
+	return func(c *Cache) error {
+		c.autoRefresh = autoRefresh
+		return nil
+	}
 }
 
 // NewCache creates a new CDI Cache. The cache is populated from a set
 // of CDI Spec directories. These can be specified using a WithSpecDirs
 // option. The default set of directories is exposed in DefaultSpecDirs.
 func NewCache(options ...Option) (*Cache, error) {
-	c := &Cache{}
-
-	if err := c.Configure(options...); err != nil {
-		return nil, err
-	}
-	if len(c.specDirs) == 0 {
-		c.Configure(WithSpecDirs(DefaultSpecDirs...))
+	c := &Cache{
+		autoRefresh: true,
+		watch:       &watch{},
 	}
 
-	return c, c.Refresh()
+	WithSpecDirs(DefaultSpecDirs...)(c)
+	c.Lock()
+	defer c.Unlock()
+
+	return c, c.configure(options...)
 }
 
-// Configure applies options to the cache. Updates the cache if options have
-// changed.
+// Configure applies options to the Cache. Updates and refreshes the
+// Cache if options have changed.
 func (c *Cache) Configure(options ...Option) error {
 	if len(options) == 0 {
 		return nil
@@ -65,17 +82,54 @@ func (c *Cache) Configure(options ...Option) error {
 	c.Lock()
 	defer c.Unlock()
 
+	return c.configure(options...)
+}
+
+// Configure the Cache. Start/stop CDI Spec directory watch, refresh
+// the Cache if necessary.
+func (c *Cache) configure(options ...Option) error {
+	var err error
+
 	for _, o := range options {
-		if err := o(c); err != nil {
+		if err = o(c); err != nil {
 			return errors.Wrapf(err, "failed to apply cache options")
 		}
 	}
 
+	c.dirErrors = make(map[string]error)
+
+	c.watch.stop()
+	if c.autoRefresh {
+		c.watch.setup(c.specDirs, c.dirErrors)
+		c.watch.start(&c.Mutex, c.refresh, c.dirErrors)
+	}
+	c.refresh()
+
 	return nil
 }
 
 // Refresh rescans the CDI Spec directories and refreshes the Cache.
+// In manual refresh mode the cache is always refreshed. In auto-
+// refresh mode the cache is only refreshed if it is out of date.
 func (c *Cache) Refresh() error {
+	c.Lock()
+	defer c.Unlock()
+
+	// force a refresh in manual mode
+	if refreshed, err := c.refreshIfRequired(!c.autoRefresh); refreshed {
+		return err
+	}
+
+	// collect and return cached errors, much like refresh() does it
+	var result error
+	for _, err := range c.errors {
+		result = multierror.Append(result, err...)
+	}
+	return result
+}
+
+// Refresh the Cache by rescanning CDI Spec directories and files.
+func (c *Cache) refresh() error {
 	var (
 		specs      = map[string][]*Spec{}
 		devices    = map[string]*Device{}
@@ -135,9 +189,6 @@ func (c *Cache) Refresh() error {
 		delete(devices, conflict)
 	}
 
-	c.Lock()
-	defer c.Unlock()
-
 	c.specs = specs
 	c.devices = devices
 	c.errors = specErrors
@@ -149,6 +200,17 @@ func (c *Cache) Refresh() error {
 	return nil
 }
 
+// RefreshIfRequired triggers a refresh if necessary.
+func (c *Cache) refreshIfRequired(force bool) (bool, error) {
+	// We need to refresh if
+	// - it's forced by an explicitly call to Refresh() in manual mode
+	// - a missing Spec dir appears (added to watch) in auto-refresh mode
+	if force || (c.autoRefresh && c.watch.update(c.dirErrors)) {
+		return true, c.refresh()
+	}
+	return false, nil
+}
+
 // InjectDevices injects the given qualified devices to an OCI Spec. It
 // returns any unresolvable devices and an error if injection fails for
 // any of the devices.
@@ -162,6 +224,8 @@ func (c *Cache) InjectDevices(ociSpec *oci.Spec, devices ...string) ([]string, e
 	c.Lock()
 	defer c.Unlock()
 
+	c.refreshIfRequired(false)
+
 	edits := &ContainerEdits{}
 	specs := map[*Spec]struct{}{}
 
@@ -195,6 +259,8 @@ func (c *Cache) GetDevice(device string) *Device {
 	c.Lock()
 	defer c.Unlock()
 
+	c.refreshIfRequired(false)
+
 	return c.devices[device]
 }
 
@@ -205,6 +271,8 @@ func (c *Cache) ListDevices() []string {
 	c.Lock()
 	defer c.Unlock()
 
+	c.refreshIfRequired(false)
+
 	for name := range c.devices {
 		devices = append(devices, name)
 	}
@@ -220,6 +288,8 @@ func (c *Cache) ListVendors() []string {
 	c.Lock()
 	defer c.Unlock()
 
+	c.refreshIfRequired(false)
+
 	for vendor := range c.specs {
 		vendors = append(vendors, vendor)
 	}
@@ -238,6 +308,8 @@ func (c *Cache) ListClasses() []string {
 	c.Lock()
 	defer c.Unlock()
 
+	c.refreshIfRequired(false)
+
 	for _, specs := range c.specs {
 		for _, spec := range specs {
 			cmap[spec.GetClass()] = struct{}{}
@@ -256,6 +328,8 @@ func (c *Cache) GetVendorSpecs(vendor string) []*Spec {
 	c.Lock()
 	defer c.Unlock()
 
+	c.refreshIfRequired(false)
+
 	return c.specs[vendor]
 }
 
@@ -268,12 +342,158 @@ func (c *Cache) GetSpecErrors(spec *Spec) []error {
 // GetErrors returns all errors encountered during the last
 // cache refresh.
 func (c *Cache) GetErrors() map[string][]error {
-	return c.errors
+	c.Lock()
+	defer c.Unlock()
+
+	errors := map[string][]error{}
+	for path, errs := range c.errors {
+		errors[path] = errs
+	}
+	for path, err := range c.dirErrors {
+		errors[path] = []error{err}
+	}
+
+	return errors
 }
 
 // GetSpecDirectories returns the CDI Spec directories currently in use.
 func (c *Cache) GetSpecDirectories() []string {
+	c.Lock()
+	defer c.Unlock()
+
 	dirs := make([]string, len(c.specDirs))
 	copy(dirs, c.specDirs)
 	return dirs
 }
+
+// GetSpecDirErrors returns any errors related to configured Spec directories.
+func (c *Cache) GetSpecDirErrors() map[string]error {
+	if c.dirErrors == nil {
+		return nil
+	}
+
+	c.Lock()
+	defer c.Unlock()
+
+	errors := make(map[string]error)
+	for dir, err := range c.dirErrors {
+		errors[dir] = err
+	}
+	return errors
+}
+
+// Our fsnotify helper wrapper.
+type watch struct {
+	watcher *fsnotify.Watcher
+	tracked map[string]bool
+}
+
+// Setup monitoring for the given Spec directories.
+func (w *watch) setup(dirs []string, dirErrors map[string]error) {
+	var (
+		dir string
+		err error
+	)
+	w.tracked = make(map[string]bool)
+	for _, dir = range dirs {
+		w.tracked[dir] = false
+	}
+
+	w.watcher, err = fsnotify.NewWatcher()
+	if err != nil {
+		for _, dir := range dirs {
+			dirErrors[dir] = errors.Wrap(err, "failed to create watcher")
+		}
+		return
+	}
+
+	w.update(dirErrors)
+}
+
+// Start watching Spec directories for relevant changes.
+func (w *watch) start(m *sync.Mutex, refresh func() error, dirErrors map[string]error) {
+	go w.watch(m, refresh, dirErrors)
+}
+
+// Stop watching directories.
+func (w *watch) stop() {
+	if w.watcher == nil {
+		return
+	}
+
+	w.watcher.Close()
+	w.tracked = nil
+}
+
+// Watch Spec directory changes, triggering a refresh if necessary.
+func (w *watch) watch(m *sync.Mutex, refresh func() error, dirErrors map[string]error) {
+	watch := w.watcher
+	if watch == nil {
+		return
+	}
+	for {
+		select {
+		case event, ok := <-watch.Events:
+			if !ok {
+				return
+			}
+
+			if (event.Op & (fsnotify.Rename | fsnotify.Remove | fsnotify.Write)) == 0 {
+				continue
+			}
+			if event.Op == fsnotify.Write {
+				if ext := filepath.Ext(event.Name); ext != ".json" && ext != ".yaml" {
+					continue
+				}
+			}
+
+			m.Lock()
+			if event.Op == fsnotify.Remove && w.tracked[event.Name] {
+				w.update(dirErrors, event.Name)
+			} else {
+				w.update(dirErrors)
+			}
+			refresh()
+			m.Unlock()
+
+		case _, ok := <-watch.Errors:
+			if !ok {
+				return
+			}
+		}
+	}
+}
+
+// Update watch with pending/missing or removed directories.
+func (w *watch) update(dirErrors map[string]error, removed ...string) bool {
+	var (
+		dir    string
+		ok     bool
+		err    error
+		update bool
+	)
+
+	for dir, ok = range w.tracked {
+		if ok {
+			continue
+		}
+
+		err = w.watcher.Add(dir)
+		if err == nil {
+			w.tracked[dir] = true
+			delete(dirErrors, dir)
+			update = true
+		} else {
+			w.tracked[dir] = false
+			dirErrors[dir] = errors.Wrap(err, "failed to monitor for changes")
+		}
+	}
+
+	for _, dir = range removed {
+		w.tracked[dir] = false
+		dirErrors[dir] = errors.New("directory removed")
+		update = true
+	}
+
+	return update
+}