nfd-worker: Watch features.d changes

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

Author: ozhuraki

pkg/nfd-worker/nfd-worker-internal_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package nfdworker
 
 import (
+	"context"
 	"os"
 	"regexp"
 	"strings"
@@ -102,7 +103,7 @@ func TestConfigParse(t *testing.T) {
 		overrides := `{"core": {"labelSources": ["fake"],"noPublish": true},"sources": {"cpu": {"cpuid": {"attributeBlacklist": ["foo","bar"]}}}}`
 
 		Convey("and no core cmdline flags have been specified", func() {
-			So(worker.configure("non-existing-file", overrides), ShouldBeNil)
+			So(worker.configure(context.Background(), "non-existing-file", overrides), ShouldBeNil)
 
 			Convey("core overrides should be in effect", func() {
 				So(worker.config.Core.LabelSources, ShouldResemble, []string{"fake"})
@@ -114,7 +115,7 @@ func TestConfigParse(t *testing.T) {
 			worker.args = Args{Overrides: ConfigOverrideArgs{
 				LabelSources:   &utils.StringSliceVal{"cpu", "kernel", "pci"},
 				FeatureSources: &utils.StringSliceVal{"cpu"}}}
-			So(worker.configure("non-existing-file", overrides), ShouldBeNil)
+			So(worker.configure(context.Background(), "non-existing-file", overrides), ShouldBeNil)
 
 			Convey("core cmdline flags should be in effect instead overrides", func() {
 				So(worker.config.Core.LabelSources, ShouldResemble, []string{"cpu", "kernel", "pci"})
@@ -150,7 +151,7 @@ sources:
 
 		Convey("and a proper config file is specified", func() {
 			worker.args = Args{Overrides: ConfigOverrideArgs{LabelSources: &utils.StringSliceVal{"cpu", "kernel", "pci"}}}
-			So(worker.configure(f.Name(), ""), ShouldBeNil)
+			So(worker.configure(context.Background(), f.Name(), ""), ShouldBeNil)
 
 			Convey("specified configuration should take effect", func() {
 				// Verify core config
@@ -172,7 +173,7 @@ sources:
 		Convey("and a proper config file and overrides are given", func() {
 			worker.args = Args{Overrides: ConfigOverrideArgs{FeatureSources: &utils.StringSliceVal{"cpu"}}}
 			overrides := `{"core": {"labelSources": ["fake"],"noPublish": true},"sources": {"pci": {"deviceClassWhitelist": ["03"]}}}`
-			So(worker.configure(f.Name(), overrides), ShouldBeNil)
+			So(worker.configure(context.Background(), f.Name(), overrides), ShouldBeNil)
 
 			Convey("overrides should take precedence over the config file", func() {
 				// Verify core config
@@ -205,7 +206,7 @@ func TestNewNfdWorker(t *testing.T) {
 				So(err, ShouldBeNil)
 			})
 			worker := w.(*nfdWorker)
-			So(worker.configure("", ""), ShouldBeNil)
+			So(worker.configure(context.Background(), "", ""), ShouldBeNil)
 			Convey("all sources should be enabled and the whitelist regexp should be empty", func() {
 				So(len(worker.featureSources), ShouldEqual, len(source.GetAllFeatureSources())-1)
 				So(len(worker.labelSources), ShouldEqual, len(source.GetAllLabelSources())-1)
@@ -223,7 +224,7 @@ func TestNewNfdWorker(t *testing.T) {
 				So(err, ShouldBeNil)
 			})
 			worker := w.(*nfdWorker)
-			So(worker.configure("", ""), ShouldBeNil)
+			So(worker.configure(context.Background(), "", ""), ShouldBeNil)
 			Convey("proper sources should be enabled", func() {
 				So(len(worker.featureSources), ShouldEqual, 1)
 				So(worker.featureSources[0].Name(), ShouldEqual, "cpu")

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
@@ -167,6 +168,7 @@ func NewNfdWorker(opts ...NfdWorkerOption) (NfdWorker, error) {
 		config:              &NFDConfig{},
 		kubernetesNamespace: utils.GetKubernetesNamespace(),
 		stop:                make(chan struct{}),
+		sourceEvent:         make(chan *source.FeatureSource),
 	}
 
 	for _, o := range opts {
@@ -220,6 +222,19 @@ func (i *infiniteTicker) Reset(d time.Duration) {
 	}
 }
 
+// Publish labels.
+func (w *nfdWorker) publishNodeFeatureObject() 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 +252,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.publishNodeFeatureObject(); err != nil {
+		return err
 	}
 
 	return nil
@@ -293,8 +305,11 @@ func (w *nfdWorker) setOwnerReference() error {
 func (w *nfdWorker) Run() error {
 	klog.InfoS("Node Feature Discovery Worker", "version", version.Get(), "nodeName", utils.NodeName(), "namespace", w.kubernetesNamespace)
 
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
 	// Read configuration file
-	err := w.configure(w.configFilePath, w.args.Options)
+	err := w.configure(ctx, w.configFilePath, w.args.Options)
 	if err != nil {
 		return err
 	}
@@ -341,6 +356,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.publishNodeFeatureObject(); err != nil {
+				return err
+			}
+
 		case <-w.stop:
 			klog.InfoS("shutting down nfd-worker")
 			return nil
@@ -361,7 +385,7 @@ func (c *coreConfig) sanitize() {
 	}
 }
 
-func (w *nfdWorker) configureCore(c coreConfig) error {
+func (w *nfdWorker) configureCore(ctx context.Context, c coreConfig) error {
 	// Handle klog
 	err := klogutils.MergeKlogConfiguration(w.args.Klog, c.Klog)
 	if err != nil {
@@ -438,6 +462,15 @@ func (w *nfdWorker) configureCore(c coreConfig) error {
 		return w.labelSources[i].Name() < w.labelSources[j].Name()
 	})
 
+	eventSources := source.GetAllEventSources()
+	for _, s := range eventSources {
+		if ok := featureSources[s.Name()]; ok != nil {
+			if err := s.SetNotifyChannel(ctx, w.sourceEvent); err != nil {
+				klog.ErrorS(err, "failed to set notify channel for event source", "source", s.Name())
+			}
+		}
+	}
+
 	if klogV := klog.V(1); klogV.Enabled() {
 		n := make([]string, len(w.featureSources))
 		for i, s := range w.featureSources {
@@ -461,7 +494,7 @@ func (w *nfdWorker) configureCore(c coreConfig) error {
 }
 
 // Parse configuration options
-func (w *nfdWorker) configure(filepath string, overrides string) error {
+func (w *nfdWorker) configure(ctx context.Context, filepath string, overrides string) error {
 	// Create a new default config
 	c := newDefaultConfig()
 	confSources := source.GetAllConfigurableSources()
@@ -516,7 +549,7 @@ func (w *nfdWorker) configure(filepath string, overrides string) error {
 
 	w.config = c
 
-	if err := w.configureCore(c.Core); err != nil {
+	if err := w.configureCore(ctx, c.Core); err != nil {
 		return err
 	}
 

source/local/local.go

@@ -18,6 +18,7 @@ package local
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -26,6 +27,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 +67,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 +90,7 @@ var (
 	_   source.FeatureSource      = &src
 	_   source.LabelSource        = &src
 	_   source.ConfigurableSource = &src
+	_   source.EventSource        = &src
 )
 
 // Name method of the LabelSource interface
@@ -318,6 +322,63 @@ func getFileContent(fileName string) ([][]byte, error) {
 	return lines, nil
 }
 
+func (s *localSource) runNotifier(ctx context.Context, ch chan *source.FeatureSource) {
+	rateLimit := time.NewTicker(time.Second)
+	defer rateLimit.Stop()
+	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.C:
+			limit = false
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// SetNotifyChannel method of the EventSource Interface
+func (s *localSource) SetNotifyChannel(ctx context.Context, ch chan *source.FeatureSource) error {
+	info, err := os.Stat(featureFilesDir)
+	if err != nil {
+		if !os.IsNotExist(err) {
+			return err
+		}
+	}
+
+	if info != nil && info.IsDir() {
+		if s.fsWatcher == nil {
+			watcher, err := fsnotify.NewWatcher()
+			if err != nil {
+				return err
+			}
+			err = watcher.Add(featureFilesDir)
+			if err != nil {
+				errWatcher := watcher.Close()
+				if errWatcher != nil {
+					klog.ErrorS(errWatcher, "failed to close fsnotify watcher")
+				}
+				return fmt.Errorf("unable to access %v: %w", featureFilesDir, err)
+			}
+			s.fsWatcher = watcher
+		}
+		go s.runNotifier(ctx, ch)
+	}
+
+	return nil
+}
+
 func init() {
 	source.Register(&src)
 }

source/source.go

@@ -19,6 +19,7 @@ package source
 //go:generate mockery --name=LabelSource --inpackage
 
 import (
+	"context"
 	"fmt"
 
 	nfdv1alpha1 "sigs.k8s.io/node-feature-discovery/api/nfd/v1alpha1"
@@ -77,6 +78,16 @@ type SupplementalSource interface {
 	DisableByDefault() bool
 }
 
+// EventSource is an interface for a source that can send events
+type EventSource interface {
+	FeatureSource
+
+	// SetNotifyChannel sets the notification channel used to send updates about feature changes.
+	// The provided channel will receive a notification (a pointer to the FeatureSource) whenever
+	// the source detects new or updated features, typically after a successful Discover operation.
+	SetNotifyChannel(ctx context.Context, ch chan *FeatureSource) error
+}
+
 // FeatureLabelValue represents the value of one feature label
 type FeatureLabelValue interface{}
 
@@ -155,6 +166,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 {