feature: support read args from KubeletConfiguration (#470)

Signed-off-by: Yue Zhang <huaihuan.zy@alibaba-inc.com>

go.mod

@@ -138,6 +138,7 @@ require (
 	k8s.io/csi-translation-lib v0.22.6 // indirect
 	k8s.io/gengo v0.0.0-20201214224949-b6c5ce23f027 // indirect
 	k8s.io/kube-openapi v0.0.0-20211109043538-20434351676c // indirect
+	k8s.io/kubelet v0.0.0 // indirect
 	k8s.io/mount-utils v0.22.6 // indirect
 	sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.27 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.1 // indirect

go.sum

@@ -1163,6 +1163,7 @@ k8s.io/kube-proxy v0.22.6/go.mod h1:xLxEZ3sHyz11XaRyxqI4Z4F3I/Wtt+Jlep8w5yxQPAY=
 k8s.io/kube-scheduler v0.22.6 h1:Ei9deaOE7n5mTg5k5OFNz8vjIbMLKd3PCxs9cv4tx0A=
 k8s.io/kube-scheduler v0.22.6/go.mod h1:DcHj6ixvb0M1PvWFbg133a1pz/vv7OSCgZUDU/UUhlU=
 k8s.io/kubectl v0.22.6/go.mod h1:9ktAgMwUsd2w12Yhj/xhMZhNna1t9rfExJg9j9jCIYk=
+k8s.io/kubelet v0.22.6 h1:RYr3K4CANyAOi51dNUZg0fa6aXgHDADITwKEhlsa78s=
 k8s.io/kubelet v0.22.6/go.mod h1:/nSfVw7oYzpmLn8Ua2q2Zix09Fq5gpDGnNqTbab9wts=
 k8s.io/kubernetes v1.22.6 h1:OPKNO4FElcN6wHc3N3P6uW3P1oHvzNxu+HJ8vGQtBzM=
 k8s.io/kubernetes v1.22.6/go.mod h1:l2ikQCpfvsMAXgL7FDtzgn/AVdjt4XGUYHMXn2vuzYI=

pkg/koordlet/statesinformer/states_noderesourcetopology.go

@@ -29,8 +29,11 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/util/retry"
 	"k8s.io/klog/v2"
+	kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config"
 	"k8s.io/kubernetes/pkg/kubelet/cm/cpumanager/state"
 	"k8s.io/kubernetes/pkg/kubelet/cm/cpuset"
+	"k8s.io/kubernetes/pkg/kubelet/kubeletconfig/configfiles"
+	utilfs "k8s.io/kubernetes/pkg/util/filesystem"
 
 	"github.com/koordinator-sh/koordinator/apis/extension"
 	"github.com/koordinator-sh/koordinator/pkg/koordlet/metriccache"
@@ -156,15 +159,42 @@ func (s *statesInformer) reportNodeTopology() {
 	if err != nil {
 		return
 	}
-	// TODO: support https://kubernetes.io/docs/reference/config-api/kubelet-config.v1beta1/
+
+	// default policy is none
+	cpuManagerPolicy := extension.KubeletCPUManagerPolicy{
+		Policy: "none",
+	}
+
+	filePath, err := system.GuessConfigFilePathFromKubelet()
+	if err != nil {
+		klog.Error("failed to read config file path from kubelet args")
+	}
+	if filePath != "" {
+		kubeletConfig, err := loadConfigFile(filePath)
+		if err != nil {
+			klog.Errorf("failed to load kubelet config, err: %v", err)
+		} else {
+			if len(kubeletConfig.CPUManagerPolicyOptions) != 0 {
+				cpuManagerPolicy.Options = kubeletConfig.CPUManagerPolicyOptions
+			}
+			if kubeletConfig.CPUManagerPolicy != "" {
+				cpuManagerPolicy.Policy = kubeletConfig.CPUManagerPolicy
+			}
+		}
+	}
+
 	cpuPolicy, stateFilePath, cpuManagerOpt, err := system.GuessCPUManagerOptFromKubelet()
 	if err != nil {
 		klog.Errorf("failed to guess kubelet cpu manager opt, err: %v", err)
 	}
-	cpuManagerPolicy := extension.KubeletCPUManagerPolicy{
-		Policy:  cpuPolicy,
-		Options: cpuManagerOpt,
+	// use args first
+	if cpuPolicy != "" {
+		cpuManagerPolicy.Policy = cpuPolicy
+	}
+	if len(cpuManagerOpt) != 0 {
+		cpuManagerPolicy.Options = cpuManagerOpt
 	}
+
 	cpuManagerPolicyJson, err := json.Marshal(cpuManagerPolicy)
 	if err != nil {
 		klog.Errorf("failed to marshal cpu manager policy, err: %v", err)
@@ -311,3 +341,16 @@ func (s *statesInformer) getNodeTopo() *v1alpha1.NodeResourceTopology {
 	defer s.nodeTopoMutex.RUnlock()
 	return s.nodeTopology.DeepCopy()
 }
+
+func loadConfigFile(kubeletConfigFile string) (*kubeletconfig.KubeletConfiguration, error) {
+	const errFmt = "failed to load Kubelet config file %s, error %v"
+	loader, err := configfiles.NewFsLoader(&utilfs.DefaultFs{}, kubeletConfigFile)
+	if err != nil {
+		return nil, fmt.Errorf(errFmt, kubeletConfigFile, err)
+	}
+	kc, err := loader.Load()
+	if err != nil {
+		return nil, fmt.Errorf(errFmt, kubeletConfigFile, err)
+	}
+	return kc, err
+}

pkg/util/system/common_linux.go

@@ -151,3 +151,16 @@ func execCmdOnHostFn(cmds []string) ([]byte, int, error) {
 		return out, 0, nil
 	}
 }
+
+// return working dir of process
+func WorkingDirOf(pid int) (string, error) {
+	var errB bytes.Buffer
+	command := exec.Command("pwdx", fmt.Sprintf("%d", pid))
+	command.Stderr = &errB
+	if out, err := command.Output(); err != nil {
+		return "", fmt.Errorf("pwdx command('%d') failed: %v, stderr: %s", pid, err, errB.String())
+	} else {
+		tokens := strings.Split(string(out), ":")
+		return strings.TrimSpace(tokens[1]), nil
+	}
+}

pkg/util/system/common_linux_test.go

@@ -61,3 +61,16 @@ func Test_PidOf(t *testing.T) {
 		assert.Error(t, err)
 	})
 }
+
+func Test_WorkingDirOf(t *testing.T) {
+	t.Run("testing process wd args should match", func(t *testing.T) {
+		wd, err := WorkingDirOf(os.Getpid())
+		assert.Empty(t, err)
+		expectedWd, _ := os.Getwd()
+		assert.Equal(t, wd, expectedWd)
+	})
+	t.Run("fake process should fail", func(t *testing.T) {
+		_, err := WorkingDirOf(1909043242)
+		assert.NotEmpty(t, err)
+	})
+}

pkg/util/system/common_unsupported.go

@@ -38,3 +38,7 @@ var ExecCmdOnHost = execCmdOnHostFn
 func execCmdOnHostFn(cmds []string) ([]byte, int, error) {
 	return nil, -1, fmt.Errorf("only support linux")
 }
+
+func WorkingDirOf(pid int) (string, error) {
+	return "", fmt.Errorf("only support linux")
+}

pkg/util/system/cpu_manager_linux.go

@@ -22,6 +22,7 @@ package system
 import (
 	"fmt"
 	"path"
+	"path/filepath"
 
 	"github.com/spf13/pflag"
 	cliflag "k8s.io/component-base/cli/flag"
@@ -45,10 +46,46 @@ func GuessCPUManagerOptFromKubelet() (string, string, map[string]string, error)
 	fs := pflag.NewFlagSet("GuessTest", pflag.ContinueOnError)
 	fs.ParseErrorsWhitelist.UnknownFlags = true
 	fs.StringVar(&argsRootDir, "root-dir", "/var/lib/kubelet", "")
-	fs.StringVar(&argsCpuPolicy, "cpu-manager-policy", "none", "")
+	fs.StringVar(&argsCpuPolicy, "cpu-manager-policy", "", "")
 	fs.Var(cliflag.NewMapStringStringNoSplit(&argsCpuManagerOpt), "cpu-manager-policy-options", "")
 	if err := fs.Parse(kubeletArgs[1:]); err != nil {
 		return "", "", nil, fmt.Errorf("failed to parse kubelet's args, kubelet version may not support: %v", err)
 	}
 	return argsCpuPolicy, path.Join(argsRootDir, "cpu_manager_state"), argsCpuManagerOpt, nil
 }
+
+// return kubelet config file path
+func GuessConfigFilePathFromKubelet() (string, error) {
+	pids, err := PidOf(Conf.ProcRootDir, "kubelet")
+	if err != nil || len(pids) == 0 {
+		return "", fmt.Errorf("failed to find kubelet's pid, kubelet may stop: %v", err)
+	}
+	kubeletPid := pids[0]
+
+	kubeletArgs, err := ProcCmdLine(Conf.ProcRootDir, kubeletPid)
+	if err != nil || len(kubeletArgs) <= 1 {
+		return "", fmt.Errorf("failed to get kubelet's args: %v", err)
+	}
+	var argsFilePath string
+	fs := pflag.NewFlagSet("GuessTest", pflag.ContinueOnError)
+	fs.ParseErrorsWhitelist.UnknownFlags = true
+	fs.StringVar(&argsFilePath, "config", argsFilePath, "")
+	if err := fs.Parse(kubeletArgs[1:]); err != nil {
+		return "", fmt.Errorf("failed to parse kubelet's args, kubelet version may not support: %v", err)
+	}
+	if argsFilePath == "" {
+		return "", nil
+	}
+	if FileExists(argsFilePath) {
+		return argsFilePath, nil
+	}
+	wd, err := WorkingDirOf(kubeletPid)
+	if err != nil {
+		return "", err
+	}
+	absPath := filepath.Join(wd, argsFilePath)
+	if FileExists(absPath) {
+		return absPath, nil
+	}
+	return "", fmt.Errorf("failed to get kubelet config file")
+}

pkg/util/system/cpu_manager_unsupported.go

@@ -22,3 +22,7 @@ package system
 func GuessCPUManagerOptFromKubelet() (string, string, map[string]string, error) {
 	return "", "", nil, nil
 }
+
+func GuessConfigFilePathFromKubelet() (string, error) {
+	return "", nil
+}