From 18088e75f1a3074ca55a27769ebe54cf57fa466e Mon Sep 17 00:00:00 2001
From: Talor Itzhak <titzhak@redhat.com>
Date: Wed, 6 Oct 2021 17:37:11 +0300
Subject: [PATCH] Get Topology Manager policy from configz-endpoint

This patch allows nfd-master to pull the config from the Kubernetes configz endpoint.
This update introduces --obtain-kubelet-config=<option> as it provides the option to obtain Kubelet config from configz-endpoint in addition to the already existing way of obtaining kubelet-config-file.

Co-authored-by: Killian Muldoon <killian.muldoon@intel.com>
Signed-off-by: Killian Muldoon <killian.muldoon@intel.com>
Signed-off-by: Talor Itzhak <titzhak@redhat.com>
---
 cmd/nfd-topology-updater/main.go      | 32 +++++++++++--------
 cmd/nfd-topology-updater/main_test.go | 10 ++++++
 go.mod                                |  1 +
 pkg/apihelper/apihelpers.go           |  6 +++-
 pkg/apihelper/k8shelpers.go           | 46 ++++++++++++++++++++++++++-
 pkg/apihelper/mock_APIHelpers.go      | 24 ++++++++++++++
 pkg/nfd-master/nfd-master.go          | 15 +++++++++
 pkg/resourcemonitor/types.go          |  9 +++---
 8 files changed, 123 insertions(+), 20 deletions(-)

diff --git a/cmd/nfd-topology-updater/main.go b/cmd/nfd-topology-updater/main.go
index ede3bbca20..20caa4eeaf 100644
--- a/cmd/nfd-topology-updater/main.go
+++ b/cmd/nfd-topology-updater/main.go
@@ -38,6 +38,12 @@ const (
 	ProgramName = "nfd-topology-updater"
 )
 
+// Options to obtain Kubelet config file
+const (
+	kubeletFile = "kubelet-config-file"
+	configz     = "configz-endpoint"
+)
+
 func main() {
 	flags := flag.NewFlagSet(ProgramName, flag.ExitOnError)
 
@@ -58,12 +64,15 @@ func main() {
 	// Plug klog into grpc logging infrastructure
 	utils.ConfigureGrpcKlog()
 
-	klConfig, err := kubeconf.GetKubeletConfigFromLocalFile(resourcemonitorArgs.KubeletConfigFile)
-	if err != nil {
-		klog.Fatalf("error reading kubelet config: %v", err)
-	}
-	tmPolicy := string(topologypolicy.DetectTopologyPolicy(klConfig.TopologyManagerPolicy, klConfig.TopologyManagerScope))
-	klog.Infof("detected kubelet Topology Manager policy %q", tmPolicy)
+	var tmPolicy string
+	if resourcemonitorArgs.KubeletConfigObtainOpt == kubeletFile {
+		klConfig, err := kubeconf.GetKubeletConfigFromLocalFile(resourcemonitorArgs.KubeletConfigFile)
+		if err != nil {
+			klog.Fatalf("error reading kubelet config: %v", err)
+		}
+		tmPolicy = string(topologypolicy.DetectTopologyPolicy(klConfig.TopologyManagerPolicy, klConfig.TopologyManagerScope))
+		klog.Infof("detected kubelet Topology Manager policy %q", tmPolicy)
+	} // Otherwise get TopologyManagerPolicy from configz-endpoint
 
 	// Get new TopologyUpdater instance
 	instance, err := topology.NewTopologyUpdater(*args, *resourcemonitorArgs, tmPolicy)
@@ -109,6 +118,7 @@ func initFlags(flagset *flag.FlagSet) (*topology.Args, *resourcemonitor.Args) {
 		"Time to sleep between CR updates. Non-positive value implies no CR updatation (i.e. infinite sleep). [Default: 60s]")
 	flagset.StringVar(&resourcemonitorArgs.Namespace, "watch-namespace", "*",
 		"Namespace to watch pods (for testing/debugging purpose). Use * for all namespaces.")
+	flagset.StringVar(&resourcemonitorArgs.KubeletConfigObtainOpt, "obtain-kubelet-config", kubeletFile, "Options to obtain Kubelet config file. [Possible arguments: kubelet-config-file/configz-endpoint]")
 	flagset.StringVar(&resourcemonitorArgs.KubeletConfigFile, "kubelet-config-file", source.VarDir.Path("lib/kubelet/config.yaml"),
 		"Kubelet config file path.")
 	flagset.StringVar(&resourcemonitorArgs.PodResourceSocketPath, "podresources-socket", source.VarDir.Path("lib/kubelet/pod-resources/kubelet.sock"),
@@ -118,13 +128,7 @@ func initFlags(flagset *flag.FlagSet) (*topology.Args, *resourcemonitor.Args) {
 	flagset.StringVar(&args.ServerNameOverride, "server-name-override", "",
 		"Hostname expected from server certificate, useful in testing")
 
-	initKlogFlags(flagset)
+	klog.InitFlags(flagset)
 
 	return args, resourcemonitorArgs
-}
-
-func initKlogFlags(flagset *flag.FlagSet) {
-	flags := flag.NewFlagSet("klog flags", flag.ExitOnError)
-	//flags.SetOutput(ioutil.Discard)
-	klog.InitFlags(flags)
-}
+}
\ No newline at end of file
diff --git a/cmd/nfd-topology-updater/main_test.go b/cmd/nfd-topology-updater/main_test.go
index 5a226d824f..80b8d38556 100644
--- a/cmd/nfd-topology-updater/main_test.go
+++ b/cmd/nfd-topology-updater/main_test.go
@@ -36,6 +36,7 @@ func TestArgsParse(t *testing.T) {
 				So(args.Oneshot, ShouldBeTrue)
 				So(finderArgs.SleepInterval, ShouldEqual, 60*time.Second)
 				So(finderArgs.KubeletConfigFile, ShouldEqual, "/var/lib/kubelet/config.yaml")
+				So(finderArgs.KubeletConfigObtainOpt, ShouldEqual, kubeletFile)
 				So(finderArgs.PodResourceSocketPath, ShouldEqual, "/var/lib/kubelet/pod-resources/kubelet.sock")
 			})
 		})
@@ -100,5 +101,14 @@ func TestArgsParse(t *testing.T) {
 				So(finderArgs.PodResourceSocketPath, ShouldEqual, "/path/testkubelet.sock")
 			})
 		})
+
+		Convey("When configz-endpoint specified", func() {
+			_, finderArgs := parseArgs(flags,
+				"--obtain-kubelet-config=configz-endpoint")
+
+			Convey("obtain-kubelet-config option should be configz-endpoint", func() {
+				So(finderArgs.KubeletConfigObtainOpt, ShouldEqual, configz)
+			})
+		})
 	})
 }
diff --git a/go.mod b/go.mod
index e4b2ce5949..c1648a7a0a 100644
--- a/go.mod
+++ b/go.mod
@@ -12,6 +12,7 @@ require (
 	github.com/klauspost/cpuid/v2 v2.0.9
 	github.com/onsi/ginkgo v1.14.0
 	github.com/onsi/gomega v1.10.1
+	github.com/pkg/errors v0.9.1
 	github.com/smartystreets/assertions v1.2.0
 	github.com/smartystreets/goconvey v1.6.4
 	github.com/stretchr/testify v1.7.0
diff --git a/pkg/apihelper/apihelpers.go b/pkg/apihelper/apihelpers.go
index d3fef4c46b..22b72380ba 100644
--- a/pkg/apihelper/apihelpers.go
+++ b/pkg/apihelper/apihelpers.go
@@ -22,6 +22,7 @@ import (
 	topologyclientset "github.com/k8stopologyawareschedwg/noderesourcetopology-api/pkg/generated/clientset/versioned"
 	api "k8s.io/api/core/v1"
 	k8sclient "k8s.io/client-go/kubernetes"
+	kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config"
 )
 
 // APIHelpers represents a set of API helpers for Kubernetes
@@ -47,6 +48,9 @@ type APIHelpers interface {
 	// GetTopologyClient returns a topologyclientset
 	GetTopologyClient() (*topologyclientset.Clientset, error)
 
-	// GetPod returns the Kubernetes pod in a namepace with a name.
+	// GetPod returns the Kubernetes pod in a namespace with a name.
 	GetPod(*k8sclient.Clientset, string, string) (*api.Pod, error)
+
+	// GetKubeletConfig returns node's kubelet config file
+	GetKubeletConfig(c *k8sclient.Clientset, nodeName string) (*kubeletconfig.KubeletConfiguration, error)
 }
diff --git a/pkg/apihelper/k8shelpers.go b/pkg/apihelper/k8shelpers.go
index ef6888cd9c..70b8525ea0 100644
--- a/pkg/apihelper/k8shelpers.go
+++ b/pkg/apihelper/k8shelpers.go
@@ -20,13 +20,18 @@ import (
 	"context"
 	"encoding/json"
 
-	topologyclientset "github.com/k8stopologyawareschedwg/noderesourcetopology-api/pkg/generated/clientset/versioned"
 	api "k8s.io/api/core/v1"
 	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	k8sclient "k8s.io/client-go/kubernetes"
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
+	kubeletconfigv1beta1 "k8s.io/kubelet/config/v1beta1"
+	kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config"
+	kubeletconfigscheme "k8s.io/kubernetes/pkg/kubelet/apis/config/scheme"
+
+	topologyclientset "github.com/k8stopologyawareschedwg/noderesourcetopology-api/pkg/generated/clientset/versioned"
+	"github.com/pkg/errors"
 )
 
 // Implements APIHelpers
@@ -132,3 +137,42 @@ func (h K8sHelpers) GetPod(cli *k8sclient.Clientset, namespace string, podName s
 
 	return pod, nil
 }
+
+func (h K8sHelpers) GetKubeletConfig(c *k8sclient.Clientset, nodeName string) (*kubeletconfig.KubeletConfiguration, error) {
+	//request here is the same as: curl https://$APISERVER:6443/api/v1/nodes/$NODE_NAME/proxy/configz
+	res := c.CoreV1().RESTClient().Get().Resource("nodes").Name(nodeName).SubResource("proxy").Suffix("configz").Do(context.TODO())
+	return decodeConfigz(&res)
+}
+
+// Decodes the rest result from /configz and returns a kubeletconfig.KubeletConfiguration (internal type).
+func decodeConfigz(res *restclient.Result) (*kubeletconfig.KubeletConfiguration, error) {
+	// This hack because /configz reports the following structure:
+	// {"kubeletconfig": {the JSON representation of kubeletconfigv1beta1.KubeletConfiguration}}
+	type configzWrapper struct {
+		ComponentConfig kubeletconfigv1beta1.KubeletConfiguration `json:"kubeletconfig"`
+	}
+
+	configz := configzWrapper{}
+	kubeCfg := kubeletconfig.KubeletConfiguration{}
+
+	contentsBytes, err := res.Raw()
+	if err != nil {
+		return nil, err
+	}
+
+	err = json.Unmarshal(contentsBytes, &configz)
+	if err != nil {
+		return nil, errors.Wrap(err, string(contentsBytes))
+	}
+
+	scheme, _, err := kubeletconfigscheme.NewSchemeAndCodecs()
+	if err != nil {
+		return nil, err
+	}
+	err = scheme.Convert(&configz.ComponentConfig, &kubeCfg, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	return &kubeCfg, nil
+}
diff --git a/pkg/apihelper/mock_APIHelpers.go b/pkg/apihelper/mock_APIHelpers.go
index 98087ba643..80707b1dc0 100644
--- a/pkg/apihelper/mock_APIHelpers.go
+++ b/pkg/apihelper/mock_APIHelpers.go
@@ -5,6 +5,7 @@ package apihelper
 import (
 	mock "github.com/stretchr/testify/mock"
 	kubernetes "k8s.io/client-go/kubernetes"
+	kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config"
 
 	v1 "k8s.io/api/core/v1"
 
@@ -39,6 +40,29 @@ func (_m *MockAPIHelpers) GetClient() (*kubernetes.Clientset, error) {
 	return r0, r1
 }
 
+// GetKubeletConfig provides a mock function with given fields: c, nodeName
+func (_m *MockAPIHelpers) GetKubeletConfig(c *kubernetes.Clientset, nodeName string) (*kubeletconfig.KubeletConfiguration, error) {
+	ret := _m.Called(c, nodeName)
+
+	var r0 *kubeletconfig.KubeletConfiguration
+	if rf, ok := ret.Get(0).(func(*kubernetes.Clientset, string) *kubeletconfig.KubeletConfiguration); ok {
+		r0 = rf(c, nodeName)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*kubeletconfig.KubeletConfiguration)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(*kubernetes.Clientset, string) error); ok {
+		r1 = rf(c, nodeName)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetNode provides a mock function with given fields: _a0, _a1
 func (_m *MockAPIHelpers) GetNode(_a0 *kubernetes.Clientset, _a1 string) (*v1.Node, error) {
 	ret := _m.Called(_a0, _a1)
diff --git a/pkg/nfd-master/nfd-master.go b/pkg/nfd-master/nfd-master.go
index 1a681323fb..ba11b7f78c 100644
--- a/pkg/nfd-master/nfd-master.go
+++ b/pkg/nfd-master/nfd-master.go
@@ -446,6 +446,21 @@ func (m *nfdMaster) UpdateNodeTopology(c context.Context, r *topologypb.NodeTopo
 	} else {
 		klog.Infof("received CR updation request for node %q", r.NodeName)
 	}
+	if len(r.TopologyPolicies[0]) == 0 {
+		klog.Warningf("Using configz-endpoint in order to get Kubelet configuration, consider to be unstable")
+		cli, err := m.apihelper.GetClient()
+		if err != nil {
+			klog.Errorf("%v", err.Error())
+			return &topologypb.NodeTopologyResponse{}, err
+		}
+		kc, err := m.apihelper.GetKubeletConfig(cli, r.NodeName)
+		if err != nil {
+			klog.Errorf("failed to get Kubelet config: %v", err.Error())
+			return &topologypb.NodeTopologyResponse{}, err
+		}
+		r.TopologyPolicies[0] = kc.TopologyManagerPolicy
+		klog.Infof("detected topology policy: %q", kc.TopologyManagerPolicy)
+	}
 	if !m.args.NoPublish {
 		err := m.updateCR(r.NodeName, r.TopologyPolicies, r.Zones, m.args.NRTNamespace)
 		if err != nil {
diff --git a/pkg/resourcemonitor/types.go b/pkg/resourcemonitor/types.go
index 4389643329..fba536bb0c 100644
--- a/pkg/resourcemonitor/types.go
+++ b/pkg/resourcemonitor/types.go
@@ -26,10 +26,11 @@ import (
 
 // Args stores commandline arguments used for resource monitoring
 type Args struct {
-	PodResourceSocketPath string
-	SleepInterval         time.Duration
-	Namespace             string
-	KubeletConfigFile     string
+	PodResourceSocketPath  string
+	SleepInterval          time.Duration
+	Namespace              string
+	KubeletConfigFile      string
+	KubeletConfigObtainOpt string
 }
 
 // ResourceInfo stores information of resources and their corresponding IDs obtained from PodResource API