Merge pull request kubewharf#692 from luomingmeng/dev/topology-adapte…

…r-support-watch-resource-file-config

topology adapter support configure custom kubelet resource plugin state file

cmd/katalyst-agent/app/options/reporter/kubelet_plugin.go

@@ -22,15 +22,17 @@ import (
 	pluginapi "k8s.io/kubelet/pkg/apis/resourceplugin/v1alpha1"
 
 	"github.com/kubewharf/katalyst-core/pkg/config/agent/reporter"
+	"github.com/kubewharf/katalyst-core/pkg/consts"
 )
 
 type KubeletPluginOptions struct {
-	PodResourcesServerEndpoints []string
-	KubeletResourcePluginPaths  []string
-	EnableReportTopologyPolicy  bool
-	ResourceNameToZoneTypeMap   map[string]string
-	NeedValidationResources     []string
-	EnablePodResourcesFilter    bool
+	PodResourcesServerEndpoints    []string
+	KubeletResourcePluginPaths     []string
+	KubeletResourcePluginStateFile string
+	EnableReportTopologyPolicy     bool
+	ResourceNameToZoneTypeMap      map[string]string
+	NeedValidationResources        []string
+	EnablePodResourcesFilter       bool
 }
 
 func NewKubeletPluginOptions() *KubeletPluginOptions {
@@ -41,8 +43,9 @@ func NewKubeletPluginOptions() *KubeletPluginOptions {
 		KubeletResourcePluginPaths: []string{
 			pluginapi.ResourcePluginPath,
 		},
-		EnableReportTopologyPolicy: false,
-		ResourceNameToZoneTypeMap:  make(map[string]string),
+		KubeletResourcePluginStateFile: consts.KubeletQoSResourceManagerCheckpoint,
+		EnableReportTopologyPolicy:     false,
+		ResourceNameToZoneTypeMap:      make(map[string]string),
 		NeedValidationResources: []string{
 			string(v1.ResourceCPU),
 			string(v1.ResourceMemory),
@@ -58,6 +61,8 @@ func (o *KubeletPluginOptions) AddFlags(fss *cliflag.NamedFlagSets) {
 		"the endpoint of pod resource api server")
 	fs.StringSliceVar(&o.KubeletResourcePluginPaths, "kubelet-resource-plugin-path", o.KubeletResourcePluginPaths,
 		"the path of kubelet resource plugin")
+	fs.StringVar(&o.KubeletResourcePluginStateFile, "kubelet-resource-plugin-state-file", o.KubeletResourcePluginStateFile,
+		"the state file of kubelet resource plugin")
 	fs.BoolVar(&o.EnableReportTopologyPolicy, "enable-report-topology-policy", o.EnableReportTopologyPolicy,
 		"whether to report topology policy")
 	fs.StringToStringVar(&o.ResourceNameToZoneTypeMap, "resource-name-to-zone-type-map", o.ResourceNameToZoneTypeMap,
@@ -71,6 +76,7 @@ func (o *KubeletPluginOptions) AddFlags(fss *cliflag.NamedFlagSets) {
 func (o *KubeletPluginOptions) ApplyTo(c *reporter.KubeletPluginConfiguration) error {
 	c.PodResourcesServerEndpoints = o.PodResourcesServerEndpoints
 	c.KubeletResourcePluginPaths = o.KubeletResourcePluginPaths
+	c.KubeletResourcePluginStateFile = o.KubeletResourcePluginStateFile
 	c.EnableReportTopologyPolicy = o.EnableReportTopologyPolicy
 	c.ResourceNameToZoneTypeMap = o.ResourceNameToZoneTypeMap
 	c.NeedValidationResources = o.NeedValidationResources

pkg/agent/resourcemanager/fetcher/kubelet/kubeletplugin.go

@@ -97,8 +97,8 @@ func NewKubeletReporterPlugin(emitter metrics.MetricEmitter, metaServer *metaser
 	}
 
 	topologyStatusAdapter, err := topology.NewPodResourcesServerTopologyAdapter(metaServer, conf.QoSConfiguration,
-		conf.PodResourcesServerEndpoints, conf.KubeletResourcePluginPaths, conf.ResourceNameToZoneTypeMap,
-		nil, p.getNumaInfo, podResourcesFilter, podresources.GetV1Client,
+		conf.PodResourcesServerEndpoints, conf.KubeletResourcePluginPaths, conf.KubeletResourcePluginStateFile,
+		conf.ResourceNameToZoneTypeMap, nil, p.getNumaInfo, podResourcesFilter, podresources.GetV1Client,
 		conf.NeedValidationResources)
 	if err != nil {
 		return nil, err

pkg/agent/resourcemanager/fetcher/kubelet/topology/topology_adapter.go

@@ -40,7 +40,6 @@ import (
 	apiconsts "github.com/kubewharf/katalyst-api/pkg/consts"
 	"github.com/kubewharf/katalyst-api/pkg/utils"
 	"github.com/kubewharf/katalyst-core/pkg/config/generic"
-	"github.com/kubewharf/katalyst-core/pkg/consts"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver"
 	metaserverpod "github.com/kubewharf/katalyst-core/pkg/metaserver/agent/pod"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/spd"
@@ -88,6 +87,9 @@ type topologyAdapterImpl struct {
 	// kubeletResourcePluginPaths is the path of kubelet resource plugin
 	kubeletResourcePluginPaths []string
 
+	// kubeletResourcePluginStateFile is the path of kubelet resource plugin checkpoint file
+	kubeletResourcePluginStateFile string
+
 	// resourceNameToZoneTypeMap is a map that stores the mapping relationship between resource names to zone types for device zones
 	resourceNameToZoneTypeMap map[string]string
 
@@ -97,7 +99,7 @@ type topologyAdapterImpl struct {
 
 // NewPodResourcesServerTopologyAdapter creates a topology adapter which uses pod resources server
 func NewPodResourcesServerTopologyAdapter(metaServer *metaserver.MetaServer, qosConf *generic.QoSConfiguration,
-	endpoints []string, kubeletResourcePluginPaths []string, resourceNameToZoneTypeMap map[string]string,
+	endpoints []string, kubeletResourcePluginPaths []string, kubeletResourcePluginStateFile string, resourceNameToZoneTypeMap map[string]string,
 	skipDeviceNames sets.String, numaInfoGetter NumaInfoGetter, podResourcesFilter PodResourcesFilter,
 	getClientFunc podresources.GetClientFunc, needValidationResources []string,
 ) (Adapter, error) {
@@ -117,16 +119,17 @@ func NewPodResourcesServerTopologyAdapter(metaServer *metaserver.MetaServer, qos
 
 	numaSocketZoneNodeMap := util.GenerateNumaSocketZone(numaInfo)
 	return &topologyAdapterImpl{
-		endpoints:                  endpoints,
-		kubeletResourcePluginPaths: kubeletResourcePluginPaths,
-		qosConf:                    qosConf,
-		metaServer:                 metaServer,
-		numaSocketZoneNodeMap:      numaSocketZoneNodeMap,
-		skipDeviceNames:            skipDeviceNames,
-		getClientFunc:              getClientFunc,
-		podResourcesFilter:         podResourcesFilter,
-		resourceNameToZoneTypeMap:  resourceNameToZoneTypeMap,
-		needValidationResources:    needValidationResources,
+		endpoints:                      endpoints,
+		kubeletResourcePluginPaths:     kubeletResourcePluginPaths,
+		kubeletResourcePluginStateFile: kubeletResourcePluginStateFile,
+		qosConf:                        qosConf,
+		metaServer:                     metaServer,
+		numaSocketZoneNodeMap:          numaSocketZoneNodeMap,
+		skipDeviceNames:                skipDeviceNames,
+		getClientFunc:                  getClientFunc,
+		podResourcesFilter:             podResourcesFilter,
+		resourceNameToZoneTypeMap:      resourceNameToZoneTypeMap,
+		needValidationResources:        needValidationResources,
 	}, nil
 }
 
@@ -243,7 +246,7 @@ func (p *topologyAdapterImpl) Run(ctx context.Context, handler func()) error {
 		ctx.Done(),
 		general.FileWatcherInfo{
 			Path:     p.kubeletResourcePluginPaths,
-			Filename: consts.KubeletQoSResourceManagerCheckpoint,
+			Filename: p.kubeletResourcePluginStateFile,
 			Op:       fsnotify.Create,
 		},
 	)

pkg/agent/resourcemanager/fetcher/kubelet/topology/topology_adapter_test.go

@@ -2969,7 +2969,7 @@ func Test_podResourcesServerTopologyAdapterImpl_Run(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.TODO())
 	notifier := make(chan struct{}, 1)
 	p, _ := NewPodResourcesServerTopologyAdapter(testMetaServer, generic.NewQoSConfiguration(),
-		endpoints, kubeletResourcePluginPath, nil,
+		endpoints, kubeletResourcePluginPath, pkgconsts.KubeletQoSResourceManagerCheckpoint, nil,
 		nil, getNumaInfo, nil, podresources.GetV1Client, []string{"cpu", "memory"})
 	err = p.Run(ctx, func() {})
 	assert.NoError(t, err)

pkg/config/agent/reporter/kubelet_plugin.go

@@ -17,12 +17,13 @@ limitations under the License.
 package reporter
 
 type KubeletPluginConfiguration struct {
-	PodResourcesServerEndpoints []string
-	KubeletResourcePluginPaths  []string
-	EnableReportTopologyPolicy  bool
-	ResourceNameToZoneTypeMap   map[string]string
-	NeedValidationResources     []string
-	EnablePodResourcesFilter    bool
+	PodResourcesServerEndpoints    []string
+	KubeletResourcePluginPaths     []string
+	KubeletResourcePluginStateFile string
+	EnableReportTopologyPolicy     bool
+	ResourceNameToZoneTypeMap      map[string]string
+	NeedValidationResources        []string
+	EnablePodResourcesFilter       bool
 }
 
 func NewKubeletPluginConfiguration() *KubeletPluginConfiguration {