nfd-worker: Watch features.d changes

Signed-off-by: Oleg Zhurakivskyy <[email protected]>

Author: ozhuraki

pkg/nfd-worker/nfd-worker.go

@@ -120,7 +120,8 @@ type nfdWorker struct {
 	kubernetesNamespace string
 	k8sClient           k8sclient.Interface
 	nfdClient           nfdclient.Interface
-	stop                chan struct{} // channel for signaling stop
+	stop                chan struct{}              // channel for signaling stop
+	sourceEvent         chan *source.FeatureSource // channel for events from sources
 	featureSources      []source.FeatureSource
 	labelSources        []source.LabelSource
 	ownerReference      []metav1.OwnerReference
@@ -220,6 +221,19 @@ func (i *infiniteTicker) Reset(d time.Duration) {
 	}
 }
 
+// Publish labels.
+func (w *nfdWorker) publishLabels() error {
+	// Get the set of feature labels.
+	labels := createFeatureLabels(w.labelSources, w.config.Core.LabelWhiteList.Regexp)
+
+	// Update the node with the feature labels.
+	if !w.config.Core.NoPublish {
+		return w.advertiseFeatures(labels)
+	}
+
+	return nil
+}
+
 // Run feature discovery.
 func (w *nfdWorker) runFeatureDiscovery() error {
 	discoveryStart := time.Now()
@@ -237,12 +251,9 @@ func (w *nfdWorker) runFeatureDiscovery() error {
 	if w.config.Core.SleepInterval.Duration > 0 && discoveryDuration > w.config.Core.SleepInterval.Duration/2 {
 		klog.InfoS("feature discovery sources took over half of sleep interval ", "duration", discoveryDuration, "sleepInterval", w.config.Core.SleepInterval.Duration)
 	}
-	// Get the set of feature labels.
-	labels := createFeatureLabels(w.labelSources, w.config.Core.LabelWhiteList.Regexp)
 
-	// Update the node with the feature labels.
-	if !w.config.Core.NoPublish {
-		return w.advertiseFeatures(labels)
+	if err := w.publishLabels(); err != nil {
+		return err
 	}
 
 	return nil
@@ -341,6 +352,15 @@ func (w *nfdWorker) Run() error {
 				return err
 			}
 
+		case s := <-w.sourceEvent:
+			if err := (*s).Discover(); err != nil {
+				klog.ErrorS(err, "feature discovery failed", "source", (*s).Name())
+				break
+			}
+			if err = w.publishLabels(); err != nil {
+				return err
+			}
+
 		case <-w.stop:
 			klog.InfoS("shutting down nfd-worker")
 			return nil
@@ -525,6 +545,14 @@ func (w *nfdWorker) configure(filepath string, overrides string) error {
 		s.SetConfig(c.Sources[s.Name()])
 	}
 
+	w.sourceEvent = make(chan *source.FeatureSource)
+	eventSources := source.GetAllEventSources()
+	for _, s := range eventSources {
+		if err := s.SetNotifyChannel(w.sourceEvent); err != nil {
+			klog.ErrorS(err, "failed to set notify channel for event source", "source", s.Name())
+		}
+	}
+
 	klog.InfoS("configuration successfully updated", "configuration", w.config)
 	return nil
 }

source/local/local.go

@@ -26,6 +26,7 @@ import (
 
 	"k8s.io/klog/v2"
 
+	"github.com/fsnotify/fsnotify"
 	nfdv1alpha1 "sigs.k8s.io/node-feature-discovery/api/nfd/v1alpha1"
 	"sigs.k8s.io/node-feature-discovery/pkg/utils"
 	"sigs.k8s.io/node-feature-discovery/source"
@@ -65,10 +66,11 @@ var (
 	featureFilesDir = "/etc/kubernetes/node-feature-discovery/features.d/"
 )
 
-// localSource implements the FeatureSource and LabelSource interfaces.
+// localSource implements the FeatureSource, LabelSource, EventSource interfaces.
 type localSource struct {
-	features *nfdv1alpha1.Features
-	config   *Config
+	features  *nfdv1alpha1.Features
+	config    *Config
+	fsWatcher *fsnotify.Watcher
 }
 
 type Config struct {
@@ -87,6 +89,7 @@ var (
 	_   source.FeatureSource      = &src
 	_   source.LabelSource        = &src
 	_   source.ConfigurableSource = &src
+	_   source.EventSource        = &src
 )
 
 // Name method of the LabelSource interface
@@ -318,6 +321,57 @@ func getFileContent(fileName string) ([][]byte, error) {
 	return lines, nil
 }
 
+func (s *localSource) runNotifier(ch chan *source.FeatureSource) {
+	rateLimit := time.After(time.Second)
+	limit := false
+	for {
+		select {
+		case event := <-s.fsWatcher.Events:
+			opAny := fsnotify.Create | fsnotify.Write | fsnotify.Remove | fsnotify.Rename | fsnotify.Chmod
+			if event.Op&opAny != 0 {
+				klog.V(2).InfoS("fsnotify event", "eventName", event.Name, "eventOp", event.Op)
+				if !limit {
+					fs := source.FeatureSource(s)
+					ch <- &fs
+					limit = true
+				}
+			}
+		case err := <-s.fsWatcher.Errors:
+			klog.ErrorS(err, "failed to watch features.d changes")
+		case <-rateLimit:
+			rateLimit = time.After(time.Second)
+			limit = false
+		}
+	}
+}
+
+// SetNotifyChannel method of the EventSource Interface
+func (s *localSource) SetNotifyChannel(ch chan *source.FeatureSource) error {
+	info, err := os.Stat(featureFilesDir)
+	if err != nil {
+		if !os.IsNotExist(err) {
+			return err
+		}
+	}
+
+	if info != nil && info.IsDir() {
+		watcher, err := fsnotify.NewWatcher()
+		if err != nil {
+			return err
+		}
+
+		err = watcher.Add(featureFilesDir)
+		if err != nil {
+			return fmt.Errorf("unable to access %v: %w", featureFilesDir, err)
+		}
+		s.fsWatcher = watcher
+
+		go s.runNotifier(ch)
+	}
+
+	return nil
+}
+
 func init() {
 	source.Register(&src)
 }

source/source.go

@@ -77,6 +77,14 @@ type SupplementalSource interface {
 	DisableByDefault() bool
 }
 
+// EventSource is an interface for a source that can send events
+type EventSource interface {
+	FeatureSource
+
+	// SetNotifyChannel sets the channel
+	SetNotifyChannel(chan *FeatureSource) error
+}
+
 // FeatureLabelValue represents the value of one feature label
 type FeatureLabelValue interface{}
 
@@ -155,6 +163,17 @@ func GetAllConfigurableSources() map[string]ConfigurableSource {
 	return all
 }
 
+// GetAllEventSources returns all registered event sources
+func GetAllEventSources() map[string]EventSource {
+	all := make(map[string]EventSource)
+	for k, v := range sources {
+		if s, ok := v.(EventSource); ok {
+			all[k] = s
+		}
+	}
+	return all
+}
+
 // GetAllFeatures returns a combined set of all features from all feature
 // sources.
 func GetAllFeatures() *nfdv1alpha1.Features {