Merge pull request #555 from caohe/validate-pod-resources-0.3

cherry pick of #541: feat(reporter): validate specific resources in pod resources response

Author: caohe

.github/workflows/ci.yml

@@ -86,7 +86,9 @@ jobs:
           fi
       - name: Upload coverage to Codecov
         if: ${{ (env.NEED_TO_CHECK == 'true') || (github.event_name != 'pull_request') }}
-        uses: codecov/codecov-action@v3
+        uses: codecov/codecov-action@v4
+        env:
+          CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
         with:
           flags: unittest
           file: coverage.txt

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

@@ -17,6 +17,7 @@ limitations under the License.
 package reporter
 
 import (
+	v1 "k8s.io/api/core/v1"
 	cliflag "k8s.io/component-base/cli/flag"
 	pluginapi "k8s.io/kubelet/pkg/apis/resourceplugin/v1alpha1"
 
@@ -28,6 +29,7 @@ type KubeletPluginOptions struct {
 	KubeletResourcePluginPaths  []string
 	EnableReportTopologyPolicy  bool
 	ResourceNameToZoneTypeMap   map[string]string
+	NeedValidationResources     []string
 }
 
 func NewKubeletPluginOptions() *KubeletPluginOptions {
@@ -40,6 +42,10 @@ func NewKubeletPluginOptions() *KubeletPluginOptions {
 		},
 		EnableReportTopologyPolicy: false,
 		ResourceNameToZoneTypeMap:  make(map[string]string),
+		NeedValidationResources: []string{
+			string(v1.ResourceCPU),
+			string(v1.ResourceMemory),
+		},
 	}
 }
 
@@ -54,13 +60,16 @@ func (o *KubeletPluginOptions) AddFlags(fss *cliflag.NamedFlagSets) {
 		"whether to report topology policy")
 	fs.StringToStringVar(&o.ResourceNameToZoneTypeMap, "resource-name-to-zone-type-map", o.ResourceNameToZoneTypeMap,
 		"a map that stores the mapping relationship between resource names to zone types in KCNR (e.g. nvidia.com/gpu=GPU,...)")
+	fs.StringSliceVar(&o.NeedValidationResources, "need-validation-resources", o.NeedValidationResources,
+		"resources need to be validated")
 }
 
 func (o *KubeletPluginOptions) ApplyTo(c *reporter.KubeletPluginConfiguration) error {
 	c.PodResourcesServerEndpoints = o.PodResourcesServerEndpoints
 	c.KubeletResourcePluginPaths = o.KubeletResourcePluginPaths
 	c.EnableReportTopologyPolicy = o.EnableReportTopologyPolicy
 	c.ResourceNameToZoneTypeMap = o.ResourceNameToZoneTypeMap
+	c.NeedValidationResources = o.NeedValidationResources
 
 	return nil
 }

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

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

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

@@ -86,12 +86,16 @@ type topologyAdapterImpl struct {
 
 	// resourceNameToZoneTypeMap is a map that stores the mapping relationship between resource names to zone types for device zones
 	resourceNameToZoneTypeMap map[string]string
+
+	// needValidationResources is the resources needed to be validated
+	needValidationResources []string
 }
 
 // NewPodResourcesServerTopologyAdapter creates a topology adapter which uses pod resources server
 func NewPodResourcesServerTopologyAdapter(metaServer *metaserver.MetaServer, endpoints []string,
 	kubeletResourcePluginPaths []string, resourceNameToZoneTypeMap map[string]string, skipDeviceNames sets.String,
-	numaInfoGetter NumaInfoGetter, podResourcesFilter PodResourcesFilter, getClientFunc podresources.GetClientFunc) (Adapter, error) {
+	numaInfoGetter NumaInfoGetter, podResourcesFilter PodResourcesFilter, getClientFunc podresources.GetClientFunc,
+	needValidationResources []string) (Adapter, error) {
 	numaInfo, err := numaInfoGetter()
 	if err != nil {
 		return nil, fmt.Errorf("failed to get numa info: %s", err)
@@ -116,6 +120,7 @@ func NewPodResourcesServerTopologyAdapter(metaServer *metaserver.MetaServer, end
 		getClientFunc:              getClientFunc,
 		podResourcesFilter:         podResourcesFilter,
 		resourceNameToZoneTypeMap:  resourceNameToZoneTypeMap,
+		needValidationResources:    needValidationResources,
 	}, nil
 }
 
@@ -149,7 +154,7 @@ func (p *topologyAdapterImpl) GetTopologyZones(parentCtx context.Context) ([]*no
 	}
 
 	// validate pod Resources server response to make sure report topology status is correct
-	if err = validatePodResourcesServerResponse(allocatableResources, listPodResourcesResponse); err != nil {
+	if err = p.validatePodResourcesServerResponse(allocatableResources, listPodResourcesResponse); err != nil {
 		return nil, errors.Wrap(err, "validate pod Resources server response failed")
 	}
 
@@ -266,14 +271,21 @@ func (p *topologyAdapterImpl) Run(ctx context.Context, handler func()) error {
 
 // validatePodResourcesServerResponse validate pod resources server response, if the resource is empty,
 // maybe the kubelet or qrm plugin is restarting
-func validatePodResourcesServerResponse(allocatableResourcesResponse *podresv1.AllocatableResourcesResponse,
-	listPodResourcesResponse *podresv1.ListPodResourcesResponse) error {
-	if allocatableResourcesResponse == nil {
-		return fmt.Errorf("allocatable Resources response is nil")
-	}
+func (p *topologyAdapterImpl) validatePodResourcesServerResponse(allocatableResourcesResponse *podresv1.
+	AllocatableResourcesResponse, listPodResourcesResponse *podresv1.ListPodResourcesResponse) error {
+	if len(p.needValidationResources) > 0 {
+		if allocatableResourcesResponse == nil {
+			return fmt.Errorf("allocatable resources response is nil")
+		}
 
-	if len(allocatableResourcesResponse.Resources) == 0 {
-		return fmt.Errorf("allocatable topology aware Resources is empty")
+		allocResSet := sets.NewString()
+		for _, res := range allocatableResourcesResponse.Resources {
+			allocResSet.Insert(res.ResourceName)
+		}
+
+		if !allocResSet.HasAll(p.needValidationResources...) {
+			return fmt.Errorf("allocatable resources response doen't contain all the resources that need to be validated")
+		}
 	}
 
 	if listPodResourcesResponse == nil {

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

@@ -1663,7 +1663,8 @@ func Test_podResourcesServerTopologyAdapterImpl_GetTopologyZones(t *testing.T) {
 						PodFetcher: &pod.PodFetcherStub{PodList: tt.fields.podList},
 					},
 				},
-				numaSocketZoneNodeMap: tt.fields.numaSocketZoneNodeMap,
+				numaSocketZoneNodeMap:   tt.fields.numaSocketZoneNodeMap,
+				needValidationResources: []string{"cpu", "memory"},
 			}
 			got, err := p.GetTopologyZones(context.TODO())
 			if (err != nil) != tt.wantErr {
@@ -1735,7 +1736,7 @@ func Test_podResourcesServerTopologyAdapterImpl_Run(t *testing.T) {
 	notifier := make(chan struct{}, 1)
 	p, _ := NewPodResourcesServerTopologyAdapter(testMetaServer,
 		endpoints, kubeletResourcePluginPath, nil,
-		nil, getNumaInfo, nil, podresources.GetV1Client)
+		nil, getNumaInfo, nil, podresources.GetV1Client, []string{"cpu", "memory"})
 	err = p.Run(ctx, func() {})
 	assert.NoError(t, err)
 

pkg/config/agent/reporter/kubelet_plugin.go

@@ -21,6 +21,7 @@ type KubeletPluginConfiguration struct {
 	KubeletResourcePluginPaths  []string
 	EnableReportTopologyPolicy  bool
 	ResourceNameToZoneTypeMap   map[string]string
+	NeedValidationResources     []string
 }
 
 func NewKubeletPluginConfiguration() *KubeletPluginConfiguration {