Merge pull request #1054 from PiotrProkop/use-new-nrt-api

Advertise TopologyManger policy and scope as Attributes in NRT api v1alpha2

cmd/nfd-topology-updater/main.go

@@ -28,7 +28,6 @@ import (
 
 	topology "sigs.k8s.io/node-feature-discovery/pkg/nfd-topology-updater"
 	"sigs.k8s.io/node-feature-discovery/pkg/resourcemonitor"
-	"sigs.k8s.io/node-feature-discovery/pkg/topologypolicy"
 	"sigs.k8s.io/node-feature-discovery/pkg/utils"
 	"sigs.k8s.io/node-feature-discovery/pkg/utils/hostpath"
 	"sigs.k8s.io/node-feature-discovery/pkg/utils/kubeconf"
@@ -88,11 +87,8 @@ func main() {
 		klog.Exitf("unsupported URI scheme: %v", u.Scheme)
 	}
 
-	tmPolicy := string(topologypolicy.DetectTopologyPolicy(klConfig.TopologyManagerPolicy, klConfig.TopologyManagerScope))
-	klog.Infof("detected kubelet Topology Manager policy %q", tmPolicy)
-
 	// Get new TopologyUpdater instance
-	instance := topology.NewTopologyUpdater(*args, *resourcemonitorArgs, tmPolicy)
+	instance := topology.NewTopologyUpdater(*args, *resourcemonitorArgs, klConfig.TopologyManagerPolicy, klConfig.TopologyManagerScope)
 
 	if err = instance.Run(); err != nil {
 		klog.Exit(err)

pkg/nfd-topology-updater/nfd-topology-updater.go

@@ -32,11 +32,19 @@ import (
 	"sigs.k8s.io/node-feature-discovery/pkg/apihelper"
 	"sigs.k8s.io/node-feature-discovery/pkg/podres"
 	"sigs.k8s.io/node-feature-discovery/pkg/resourcemonitor"
+	"sigs.k8s.io/node-feature-discovery/pkg/topologypolicy"
 	"sigs.k8s.io/node-feature-discovery/pkg/utils"
 	"sigs.k8s.io/node-feature-discovery/pkg/version"
 	"sigs.k8s.io/yaml"
 )
 
+const (
+	// TopologyManagerPolicyAttributeName represents an attribute which defines Topology Manager Policy
+	TopologyManagerPolicyAttributeName = "topologyManagerPolicy"
+	// TopologyManagerScopeAttributeName represents an attribute which defines Topology Manager Policy Scope
+	TopologyManagerScopeAttributeName = "topologyManagerScope"
+)
+
 // Args are the command line arguments
 type Args struct {
 	NoPublish      bool
@@ -60,10 +68,21 @@ type NfdTopologyUpdater interface {
 type staticNodeInfo struct {
 	nodeName string
 	tmPolicy string
+	tmScope  string
+}
+
+func newStaticNodeInfo(policy, scope string) staticNodeInfo {
+	nodeName := utils.NodeName()
+	klog.InfoS("detected kubelet Topology Manager configuration", "policy", policy, "scope", scope, "nodeName", nodeName)
+	return staticNodeInfo{
+		nodeName: nodeName,
+		tmPolicy: policy,
+		tmScope:  scope,
+	}
 }
 
 type nfdTopologyUpdater struct {
-	nodeInfo            *staticNodeInfo
+	nodeInfo            staticNodeInfo
 	args                Args
 	apihelper           apihelper.APIHelpers
 	resourcemonitorArgs resourcemonitor.Args
@@ -73,16 +92,13 @@ type nfdTopologyUpdater struct {
 }
 
 // NewTopologyUpdater creates a new NfdTopologyUpdater instance.
-func NewTopologyUpdater(args Args, resourcemonitorArgs resourcemonitor.Args, policy string) NfdTopologyUpdater {
+func NewTopologyUpdater(args Args, resourcemonitorArgs resourcemonitor.Args, policy, scope string) NfdTopologyUpdater {
 	nfd := &nfdTopologyUpdater{
 		args:                args,
 		resourcemonitorArgs: resourcemonitorArgs,
-		nodeInfo: &staticNodeInfo{
-			nodeName: utils.NodeName(),
-			tmPolicy: policy,
-		},
-		stop:   make(chan struct{}, 1),
-		config: &NFDConfig{},
+		nodeInfo:            newStaticNodeInfo(policy, scope),
+		stop:                make(chan struct{}, 1),
+		config:              &NFDConfig{},
 	}
 	if args.ConfigFile != "" {
 		nfd.configFilePath = filepath.Clean(args.ConfigFile)
@@ -185,7 +201,8 @@ func (w *nfdTopologyUpdater) updateNodeResourceTopology(zoneInfo v1alpha2.ZoneLi
 				Name: w.nodeInfo.nodeName,
 			},
 			Zones:            zoneInfo,
-			TopologyPolicies: []string{w.nodeInfo.tmPolicy},
+			TopologyPolicies: []string{string(topologypolicy.DetectTopologyPolicy(w.nodeInfo.tmPolicy, w.nodeInfo.tmScope))},
+			Attributes:       createTopologyAttributes(w.nodeInfo.tmPolicy, w.nodeInfo.tmScope),
 		}
 
 		_, err := cli.TopologyV1alpha2().NodeResourceTopologies().Create(context.TODO(), &nrtNew, metav1.CreateOptions{})
@@ -231,3 +248,16 @@ func (w *nfdTopologyUpdater) configure() error {
 	klog.Infof("configuration file %q parsed:\n %v", w.configFilePath, w.config)
 	return nil
 }
+
+func createTopologyAttributes(policy string, scope string) v1alpha2.AttributeList {
+	return v1alpha2.AttributeList{
+		{
+			Name:  TopologyManagerPolicyAttributeName,
+			Value: policy,
+		},
+		{
+			Name:  TopologyManagerScopeAttributeName,
+			Value: scope,
+		},
+	}
+}

pkg/utils/kubeconf/kubelet_config_file.go

@@ -17,23 +17,40 @@ limitations under the License.
 package kubeconf
 
 import (
-	"os"
-
-	"github.com/ghodss/yaml"
+	"fmt"
 
 	kubeletconfigv1beta1 "k8s.io/kubelet/config/v1beta1"
+	kubeletconfigscheme "k8s.io/kubernetes/pkg/kubelet/apis/config/scheme"
+	"k8s.io/kubernetes/pkg/kubelet/kubeletconfig/configfiles"
+	utilfs "k8s.io/kubernetes/pkg/util/filesystem"
 )
 
 // GetKubeletConfigFromLocalFile returns KubeletConfiguration loaded from the node local config
+// based on https://github.com/kubernetes/kubernetes/blob/master/cmd/kubelet/app/server.go#L337
+// it fills empty fields with default values
 func GetKubeletConfigFromLocalFile(kubeletConfigPath string) (*kubeletconfigv1beta1.KubeletConfiguration, error) {
-	kubeletBytes, err := os.ReadFile(kubeletConfigPath)
+	const errFmt = "failed to load Kubelet config file %s, error %w"
+
+	loader, err := configfiles.NewFsLoader(&utilfs.DefaultFs{}, kubeletConfigPath)
+	if err != nil {
+		return nil, fmt.Errorf(errFmt, kubeletConfigPath, err)
+	}
+
+	kc, err := loader.Load()
+	if err != nil {
+		return nil, fmt.Errorf(errFmt, kubeletConfigPath, err)
+	}
+
+	scheme, _, err := kubeletconfigscheme.NewSchemeAndCodecs()
 	if err != nil {
 		return nil, err
 	}
 
 	kubeletConfig := &kubeletconfigv1beta1.KubeletConfiguration{}
-	if err := yaml.Unmarshal(kubeletBytes, kubeletConfig); err != nil {
+	err = scheme.Convert(kc, kubeletConfig, nil)
+	if err != nil {
 		return nil, err
 	}
+
 	return kubeletConfig, nil
 }

test/e2e/utils/noderesourcetopology.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"reflect"
 	"runtime"
 	"strings"
 	"time"
@@ -28,6 +29,7 @@ import (
 	"github.com/k8stopologyawareschedwg/noderesourcetopology-api/pkg/apis/topology/v1alpha2"
 	topologyclientset "github.com/k8stopologyawareschedwg/noderesourcetopology-api/pkg/generated/clientset/versioned"
 	"github.com/onsi/gomega"
+	nfdtopologyupdater "sigs.k8s.io/node-feature-discovery/pkg/nfd-topology-updater"
 	"sigs.k8s.io/node-feature-discovery/pkg/topologypolicy"
 
 	corev1 "k8s.io/api/core/v1"
@@ -198,6 +200,24 @@ func IsValidNodeTopology(nodeTopology *v1alpha2.NodeResourceTopology, kubeletCon
 		return false
 	}
 
+	expectedPolicyAttribute := v1alpha2.AttributeInfo{
+		Name:  nfdtopologyupdater.TopologyManagerPolicyAttributeName,
+		Value: kubeletConfig.TopologyManagerPolicy,
+	}
+	if !containsAttribute(nodeTopology.Attributes, expectedPolicyAttribute) {
+		framework.Logf("topology policy attributes don't have correct topologyManagerPolicy attribute expected %v attributeList %v", expectedPolicyAttribute, nodeTopology.Attributes)
+		return false
+	}
+
+	expectedScopeAttribute := v1alpha2.AttributeInfo{
+		Name:  nfdtopologyupdater.TopologyManagerScopeAttributeName,
+		Value: kubeletConfig.TopologyManagerScope,
+	}
+	if !containsAttribute(nodeTopology.Attributes, expectedScopeAttribute) {
+		framework.Logf("topology policy attributes don't have correct topologyManagerScope attribute expected %v attributeList %v", expectedScopeAttribute, nodeTopology.Attributes)
+		return false
+	}
+
 	if nodeTopology.Zones == nil || len(nodeTopology.Zones) == 0 {
 		framework.Logf("failed to get topology zones from the node topology resource")
 		return false
@@ -259,3 +279,12 @@ func isValidResourceList(zoneName string, resources v1alpha2.ResourceInfoList) b
 	}
 	return foundCpu
 }
+
+func containsAttribute(attributes v1alpha2.AttributeList, attribute v1alpha2.AttributeInfo) bool {
+	for _, attr := range attributes {
+		if reflect.DeepEqual(attr, attribute) {
+			return true
+		}
+	}
+	return false
+}