Expose swap behavior via label

Signed-off-by: Feruzjon Muyassarov <[email protected]>

Author: fmuyassarov

cmd/nfd-worker/main.go

@@ -19,7 +19,9 @@ package main
 import (
 	"flag"
 	"fmt"
+	"net"
 	"os"
+	"strings"
 
 	"k8s.io/klog/v2"
 
@@ -32,7 +34,8 @@ import (
 
 const (
 	// ProgramName is the canonical name of this program
-	ProgramName = "nfd-worker"
+	ProgramName       = "nfd-worker"
+	kubeletSecurePort = 10250
 )
 
 func main() {
@@ -82,6 +85,20 @@ func parseArgs(flags *flag.FlagSet, osArgs ...string) *worker.Args {
 		os.Exit(2)
 	}
 
+	if len(args.KubeletConfigURI) == 0 {
+		nodeAddress := os.Getenv("NODE_ADDRESS")
+		if len(nodeAddress) == 0 {
+			_, _ = fmt.Fprintf(flags.Output(), "unable to determine the default kubelet config endpoint 'https://${NODE_ADDRESS}:%d/configz' due to empty NODE_ADDRESS environment, "+
+				"please either define the NODE_ADDRESS environment variable or specify endpoint with the -kubelet-config-uri flag\n", kubeletSecurePort)
+			os.Exit(1)
+		}
+		if isIPv6(nodeAddress) {
+			// With IPv6 we need to wrap the IP address in brackets as we append :port below
+			nodeAddress = "[" + nodeAddress + "]"
+		}
+		args.KubeletConfigURI = fmt.Sprintf("https://%s:%d/configz", nodeAddress, kubeletSecurePort)
+	}
+
 	// Handle overrides
 	flags.Visit(func(f *flag.Flag) {
 		switch f.Name {
@@ -106,6 +123,10 @@ func initFlags(flagset *flag.FlagSet) (*worker.Args, *worker.ConfigOverrideArgs)
 		"Config file to use.")
 	flagset.StringVar(&args.Kubeconfig, "kubeconfig", "",
 		"Kubeconfig to use")
+	flagset.StringVar(&args.KubeletConfigURI, "kubelet-config-uri", "",
+		"Kubelet config URI path. Default to kubelet configz endpoint.")
+	flagset.StringVar(&args.APIAuthTokenFile, "api-auth-token-file", "/var/run/secrets/kubernetes.io/serviceaccount/token",
+		"API auth token file path. It is used to request kubelet configz endpoint, only takes effect when kubelet-config-uri is https. Default to /var/run/secrets/kubernetes.io/serviceaccount/token.")
 	flagset.BoolVar(&args.Oneshot, "oneshot", false,
 		"Do not publish feature labels")
 	flagset.IntVar(&args.Port, "port", 8080,
@@ -134,3 +155,8 @@ func initFlags(flagset *flag.FlagSet) (*worker.Args, *worker.ConfigOverrideArgs)
 
 	return args, overrides
 }
+
+func isIPv6(addr string) bool {
+	ip := net.ParseIP(addr)
+	return ip != nil && strings.Count(ip.String(), ":") >= 2
+}

deployment/base/rbac/worker-role.yaml

@@ -1,5 +1,5 @@
 apiVersion: rbac.authorization.k8s.io/v1
-kind: Role
+kind: ClusterRole
 metadata:
   name: nfd-worker
 rules:
@@ -14,7 +14,8 @@ rules:
   - delete
 - apiGroups:
   - ""
-  resources:
-  - pods
-  verbs:
-  - get
+  resources: ["pods"]
+  verbs: ["get"]
+- apiGroups: [""]
+  resources: ["nodes/proxy"]
+  verbs: ["get"]

deployment/base/rbac/worker-rolebinding.yaml

@@ -1,10 +1,10 @@
 apiVersion: rbac.authorization.k8s.io/v1
-kind: RoleBinding
+kind: ClusterRoleBinding
 metadata:
   name: nfd-worker
 roleRef:
   apiGroup: rbac.authorization.k8s.io
-  kind: Role
+  kind: ClusterRole
   name: nfd-worker
 subjects:
 - kind: ServiceAccount

deployment/components/common/env.yaml

@@ -13,3 +13,11 @@
       valueFrom:
         fieldRef:
           fieldPath: metadata.uid
+    - name: NODE_ADDRESS
+      valueFrom:
+        fieldRef:
+          fieldPath: status.hostIP
+    - name: POD_NAMESPACE
+      valueFrom:
+        fieldRef:
+          fieldPath: metadata.namespace

deployment/components/worker-config/kustomization.yaml

@@ -5,3 +5,9 @@ configMapGenerator:
 - files:
   - nfd-worker.conf=nfd-worker.conf.example
   name: nfd-worker-conf
+
+# patches:
+# - path: worker-securitycontext.yaml
+#   target:
+#     labelSelector: app=nfd-worker
+#     name: nfd-worker

pkg/nfd-worker/nfd-worker.go

@@ -20,6 +20,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"net/url"
 	"os"
 	"path/filepath"
 	"regexp"
@@ -38,8 +39,10 @@ import (
 	"k8s.io/apimachinery/pkg/util/validation"
 	k8sclient "k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
+	kubeletconfigv1beta1 "k8s.io/kubelet/config/v1beta1"
 	"k8s.io/utils/ptr"
 	klogutils "sigs.k8s.io/node-feature-discovery/pkg/utils/klog"
+	"sigs.k8s.io/node-feature-discovery/pkg/utils/kubeconf"
 	"sigs.k8s.io/yaml"
 
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
@@ -57,6 +60,7 @@ import (
 	_ "sigs.k8s.io/node-feature-discovery/source/kernel"
 	_ "sigs.k8s.io/node-feature-discovery/source/local"
 	_ "sigs.k8s.io/node-feature-discovery/source/memory"
+	memory "sigs.k8s.io/node-feature-discovery/source/memory"
 	_ "sigs.k8s.io/node-feature-discovery/source/network"
 	_ "sigs.k8s.io/node-feature-discovery/source/pci"
 	_ "sigs.k8s.io/node-feature-discovery/source/storage"
@@ -94,13 +98,16 @@ type Labels map[string]string
 
 // Args are the command line arguments of NfdWorker.
 type Args struct {
-	ConfigFile  string
-	Klog        map[string]*utils.KlogFlagVal
-	Kubeconfig  string
-	Oneshot     bool
-	Options     string
-	Port        int
-	NoOwnerRefs bool
+	ConfigFile        string
+	Klog              map[string]*utils.KlogFlagVal
+	Kubeconfig        string
+	Oneshot           bool
+	Options           string
+	Port              int
+	NoOwnerRefs       bool
+	KubeletConfigPath string
+	KubeletConfigURI  string
+	APIAuthTokenFile  string
 
 	Overrides ConfigOverrideArgs
 }
@@ -124,6 +131,7 @@ type nfdWorker struct {
 	featureSources      []source.FeatureSource
 	labelSources        []source.LabelSource
 	ownerReference      []metav1.OwnerReference
+	kubeletConfigFunc   func() (*kubeletconfigv1beta1.KubeletConfiguration, error)
 }
 
 // This ticker can represent infinite and normal intervals.
@@ -169,12 +177,25 @@ func NewNfdWorker(opts ...NfdWorkerOption) (NfdWorker, error) {
 		stop:                make(chan struct{}),
 	}
 
+	if nfd.args.ConfigFile != "" {
+		nfd.configFilePath = filepath.Clean(nfd.args.ConfigFile)
+	}
+
 	for _, o := range opts {
 		o.apply(nfd)
 	}
 
-	if nfd.args.ConfigFile != "" {
-		nfd.configFilePath = filepath.Clean(nfd.args.ConfigFile)
+	kubeletConfigFunc, err := getKubeletConfigFunc(nfd.args.KubeletConfigURI, nfd.args.APIAuthTokenFile)
+	if err != nil {
+		return nil, err
+	}
+
+	nfd = &nfdWorker{
+		kubeletConfigFunc: kubeletConfigFunc,
+	}
+
+	for _, o := range opts {
+		o.apply(nfd)
 	}
 
 	// k8sClient might've been set via opts by tests
@@ -239,6 +260,8 @@ func (w *nfdWorker) runFeatureDiscovery() error {
 	}
 	// Get the set of feature labels.
 	labels := createFeatureLabels(w.labelSources, w.config.Core.LabelWhiteList.Regexp)
+	// Append a label with app=nfd
+	labels["app"] = "nfd"
 
 	// Update the node with the feature labels.
 	if !w.config.Core.NoPublish {
@@ -255,9 +278,10 @@ func (w *nfdWorker) setOwnerReference() error {
 	if !w.config.Core.NoOwnerRefs {
 		// Get pod owner reference
 		podName := os.Getenv("POD_NAME")
+		podNamespace := os.Getenv("POD_NAMESPACE")
 		// Add pod owner reference if it exists
 		if podName != "" {
-			if selfPod, err := w.k8sClient.CoreV1().Pods(w.kubernetesNamespace).Get(context.TODO(), podName, metav1.GetOptions{}); err != nil {
+			if selfPod, err := w.k8sClient.CoreV1().Pods(podNamespace).Get(context.TODO(), podName, metav1.GetOptions{}); err != nil {
 				klog.ErrorS(err, "failed to get self pod, cannot inherit ownerReference for NodeFeature")
 				return err
 			} else {
@@ -312,6 +336,12 @@ func (w *nfdWorker) Run() error {
 	httpMux.Handle("/metrics", promhttp.HandlerFor(promRegistry, promhttp.HandlerOpts{}))
 	registerVersion(version.Get())
 
+	klConfig, err := w.kubeletConfigFunc()
+	if err != nil {
+		return err
+	}
+	memory.SetSwapMode(klConfig.MemorySwap.SwapBehavior)
+
 	err = w.runFeatureDiscovery()
 	if err != nil {
 		return err
@@ -464,6 +494,7 @@ func (w *nfdWorker) configureCore(c coreConfig) error {
 func (w *nfdWorker) configure(filepath string, overrides string) error {
 	// Create a new default config
 	c := newDefaultConfig()
+
 	confSources := source.GetAllConfigurableSources()
 	c.Sources = make(map[string]source.Config, len(confSources))
 	for _, s := range confSources {
@@ -624,7 +655,7 @@ func (m *nfdWorker) updateNodeFeatureObject(labels Labels) error {
 		return err
 	}
 	nodename := utils.NodeName()
-	namespace := m.kubernetesNamespace
+	namespace := os.Getenv("POD_NAMESPACE")
 
 	features := source.GetAllFeatures()
 
@@ -720,3 +751,38 @@ func (c *sourcesConfig) UnmarshalJSON(data []byte) error {
 
 	return nil
 }
+
+func getKubeletConfigFunc(uri, apiAuthTokenFile string) (func() (*kubeletconfigv1beta1.KubeletConfiguration, error), error) {
+	u, err := url.ParseRequestURI(uri)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse -kubelet-config-uri: %w", err)
+	}
+
+	// init kubelet API client
+	var klConfig *kubeletconfigv1beta1.KubeletConfiguration
+	switch u.Scheme {
+	case "file":
+		return func() (*kubeletconfigv1beta1.KubeletConfiguration, error) {
+			klConfig, err = kubeconf.GetKubeletConfigFromLocalFile(u.Path)
+			if err != nil {
+				return nil, fmt.Errorf("failed to read kubelet config: %w", err)
+			}
+			return klConfig, err
+		}, nil
+	case "https":
+		restConfig, err := kubeconf.InsecureConfig(u.String(), apiAuthTokenFile)
+		if err != nil {
+			return nil, fmt.Errorf("failed to initialize rest config for kubelet config uri: %w", err)
+		}
+
+		return func() (*kubeletconfigv1beta1.KubeletConfiguration, error) {
+			klConfig, err = kubeconf.GetKubeletConfiguration(restConfig)
+			if err != nil {
+				return nil, fmt.Errorf("failed to get kubelet config from configz endpoint: %w", err)
+			}
+			return klConfig, nil
+		}, nil
+	}
+
+	return nil, fmt.Errorf("unsupported URI scheme: %v", u.Scheme)
+}

pkg/resourcemonitor/podresourcesscanner.go

@@ -134,7 +134,7 @@ func (resMon *PodResourcesScanner) Scan() (ScanResponse, error) {
 		if err != nil {
 			klog.ErrorS(err, "failed to calculate fingerprint")
 		} else {
-			klog.InfoS("podFingerprint calculated", "status", status.Repr())
+			klog.V(2).InfoS("podFingerprint calculated", "status", status.Repr())
 
 			retVal.Attributes = append(retVal.Attributes, v1alpha2.AttributeInfo{
 				Name:  podfingerprint.Attribute,
@@ -145,7 +145,7 @@ func (resMon *PodResourcesScanner) Scan() (ScanResponse, error) {
 	var podResData []PodResources
 
 	for _, podResource := range respPodResources {
-		klog.InfoS("scanning pod", "podName", podResource.GetName())
+		klog.V(2).InfoS("scanning pod", "podName", podResource.GetName())
 		hasDevice := hasDevice(podResource)
 		isWatchable, isIntegralGuaranteed, err := resMon.isWatchable(podResource.GetNamespace(), podResource.GetName(), hasDevice)
 		if err != nil {
@@ -228,7 +228,7 @@ func hasDevice(podResource *podresourcesapi.PodResources) bool {
 			return true
 		}
 	}
-	klog.InfoS("pod doesn't have devices", "podName", podResource.GetName())
+	klog.V(2).InfoS("pod doesn't have devices", "podName", podResource.GetName())
 	return false
 }
 

source/memory/memory.go

@@ -56,11 +56,17 @@ type memorySource struct {
 
 // Singleton source instance
 var (
-	src memorySource
-	_   source.FeatureSource = &src
-	_   source.LabelSource   = &src
+	src                 memorySource
+	_                   source.FeatureSource = &src
+	_                   source.LabelSource   = &src
+	defaultSwapBehavior                      = "NoSwap"
+	swapBehavior        string
 )
 
+func SetSwapMode(behavior string) {
+	swapBehavior = behavior
+}
+
 // Name returns an identifier string for this feature source.
 func (s *memorySource) Name() string { return Name }
 
@@ -80,6 +86,7 @@ func (s *memorySource) GetLabels() (source.FeatureLabels, error) {
 	// Swap
 	if isSwap, ok := features.Attributes[SwapFeature].Elements["enabled"]; ok && isSwap == "true" {
 		labels["swap"] = true
+		labels["swap.behavior"] = features.Attributes[SwapFeature].Elements["behavior"]
 	}
 
 	// NVDIMM
@@ -106,12 +113,18 @@ func (s *memorySource) Discover() error {
 	} else {
 		s.features.Attributes[NumaFeature] = nfdv1alpha1.AttributeFeatureSet{Elements: numa}
 	}
-
-	// Detect Swap
+	// Detect Swap and Swap Behavior
 	if swap, err := detectSwap(); err != nil {
 		klog.ErrorS(err, "failed to detect Swap nodes")
 	} else {
 		s.features.Attributes[SwapFeature] = nfdv1alpha1.AttributeFeatureSet{Elements: swap}
+		if err != nil {
+			klog.V(3).ErrorS(err, "failed to detect swap behavior; kubelet swapBehavior configuration may be missing or misconfigured")
+		} else if swapBehavior == "" {
+			swap["behavior"] = defaultSwapBehavior
+		} else {
+			swap["behavior"] = swapBehavior
+		}
 	}
 
 	// Detect NVDIMM