TopologyManager Scope support

This policy accounts resources per pod level basis.

Signed-off-by: Alexey Perevalov <[email protected]>

Author: AlexeyPerevalov

pkg/apis/config/types.go

@@ -51,6 +51,8 @@ const (
 
 	// to preserve consistency keep it in pkg/apis/core/types.go"
 	SingleNUMANodeTopologyManagerPolicy TopologyManagerPolicy = "SingleNUMANode"
+
+	PodTopologyScope TopologyManagerPolicy = "PodTopologyScope"
 )
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/noderesourcetopology/match.go

@@ -29,17 +29,16 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
 	"k8s.io/klog/v2"
-
 	v1qos "k8s.io/kubernetes/pkg/apis/core/v1/helper/qos"
 	bm "k8s.io/kubernetes/pkg/kubelet/cm/topologymanager/bitmask"
 	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
+	apiconfig "sigs.k8s.io/scheduler-plugins/pkg/apis/config"
+
 	topologyv1alpha1 "github.com/k8stopologyawareschedwg/noderesourcetopology-api/pkg/apis/topology/v1alpha1"
 	topoclientset "github.com/k8stopologyawareschedwg/noderesourcetopology-api/pkg/generated/clientset/versioned"
 	topoinformerexternal "github.com/k8stopologyawareschedwg/noderesourcetopology-api/pkg/generated/informers/externalversions"
 	topologyinformers "github.com/k8stopologyawareschedwg/noderesourcetopology-api/pkg/generated/informers/externalversions"
 	topoinformerv1alpha1 "github.com/k8stopologyawareschedwg/noderesourcetopology-api/pkg/generated/informers/externalversions/topology/v1alpha1"
-
-	apiconfig "sigs.k8s.io/scheduler-plugins/pkg/apis/config"
 )
 
 const (
@@ -69,6 +68,14 @@ type NodeResourceTopologyMatch struct {
 	topologyPolicyHandlers PolicyHandlerMap
 }
 
+type SingleNUMANodeHandler struct {
+	match *NodeResourceTopologyMatch
+}
+
+type PodLevelResourceHandler struct {
+	match *NodeResourceTopologyMatch
+}
+
 type NUMANode struct {
 	NUMAID    int
 	Resources v1.ResourceList
@@ -82,39 +89,11 @@ func (tm *NodeResourceTopologyMatch) Name() string {
 }
 
 func filter(containers []v1.Container, nodes NUMANodeList, qos v1.PodQOSClass) *framework.Status {
-	if qos == v1.PodQOSBestEffort {
-		return nil
-	}
-
-	zeroQuantity := resource.MustParse("0")
 	for _, container := range containers {
 		bitmask := bm.NewEmptyBitMask()
 		bitmask.Fill()
-		for resource, quantity := range container.Resources.Requests {
-			resourceBitmask := bm.NewEmptyBitMask()
-			for _, numaNode := range nodes {
-				numaQuantity, ok := numaNode.Resources[resource]
-				// if can't find requested resource on the node - skip (don't set it as available NUMA node)
-				// if unfound resource has 0 quantity probably this numa node can be considered
-				if !ok && quantity.Cmp(zeroQuantity) != 0 {
-					continue
-				}
-				// Check for the following:
-				// 1. set numa node as possible node if resource is memory or Hugepages (until memory manager will not be merged and
-				// memory will not be provided in CRD
-				// 2. set numa node as possible node if resource is cpu and it's not guaranteed QoS, since cpu will flow
-				// 3. set numa node as possible node if zero quantity for non existing resource was requested (TODO check topology manaager behaviour)
-				// 4. otherwise check amount of resources
-				if resource == v1.ResourceMemory ||
-					strings.HasPrefix(string(resource), string(v1.ResourceHugePagesPrefix)) ||
-					resource == v1.ResourceCPU && qos != v1.PodQOSGuaranteed ||
-					quantity.Cmp(zeroQuantity) == 0 ||
-					numaQuantity.Cmp(quantity) >= 0 {
-					resourceBitmask.Add(numaNode.NUMAID)
-				}
-			}
-			bitmask.And(resourceBitmask)
-		}
+
+		checkResourcesForNUMANodes(bitmask, nodes, container.Resources.Requests, qos)
 		if bitmask.IsEmpty() {
 			// definitly we can't align container, so we can't align a pod
 			return framework.NewStatus(framework.Unschedulable, fmt.Sprintf("Can't align container: %s", container.Name))
@@ -147,14 +126,76 @@ func extractResources(zone topologyv1alpha1.Zone) v1.ResourceList {
 	return res
 }
 
-func (tm NodeResourceTopologyMatch) PolicyFilter(pod *v1.Pod, zones topologyv1alpha1.ZoneList) *framework.Status {
+func (sh SingleNUMANodeHandler) PolicyFilter(pod *v1.Pod, zones topologyv1alpha1.ZoneList) *framework.Status {
 	containers := []v1.Container{}
 	containers = append(pod.Spec.InitContainers, pod.Spec.Containers...)
 
-	tm.nodeTopologyGuard.RLock()
-	defer tm.nodeTopologyGuard.RUnlock()
+	sh.match.nodeTopologyGuard.RLock()
+	defer sh.match.nodeTopologyGuard.RUnlock()
 	// prepare NUMANodes list from zoneMap
 
+	nodes := createNUMANodeList(zones)
+	return filter(containers, nodes, v1qos.GetPodQOS(pod))
+}
+
+func checkResourcesForNUMANodes(bitmask bm.BitMask, nodes NUMANodeList, resources v1.ResourceList, qos v1.PodQOSClass) {
+	zeroQuantity := resource.MustParse("0")
+	for resource, quantity := range resources {
+		resourceBitmask := bm.NewEmptyBitMask()
+		for _, numaNode := range nodes {
+			numaQuantity, ok := numaNode.Resources[resource]
+			// if can't find requested resource on the node - skip (don't set it as available NUMA node)
+			// if unfound resource has 0 quantity probably this numa node can be considered
+			if !ok && quantity.Cmp(zeroQuantity) != 0 {
+				continue
+			}
+			// Check for the following:
+			// 1. set numa node as possible node if resource is memory or Hugepages (until memory manager will not be merged and
+			// memory will not be provided in CRD
+			// 2. set numa node as possible node if resource is cpu and it's not guaranteed QoS, since cpu will flow
+			// 3. set numa node as possible node if zero quantity for non existing resource was requested (TODO check topology manaager behaviour)
+			// 4. otherwise check amount of resources
+			if resource == v1.ResourceMemory ||
+				strings.HasPrefix(string(resource), string(v1.ResourceHugePagesPrefix)) ||
+				resource == v1.ResourceCPU && qos != v1.PodQOSGuaranteed ||
+				quantity.Cmp(zeroQuantity) == 0 ||
+				numaQuantity.Cmp(quantity) >= 0 {
+				resourceBitmask.Add(numaNode.NUMAID)
+			}
+		}
+		bitmask.And(resourceBitmask)
+	}
+}
+
+func (ph PodLevelResourceHandler) PolicyFilter(pod *v1.Pod, zones topologyv1alpha1.ZoneList) *framework.Status {
+	containers := []v1.Container{}
+	containers = append(pod.Spec.InitContainers, pod.Spec.Containers...)
+
+	resources := make(v1.ResourceList)
+
+	for _, container := range containers {
+		for resource, quantity := range container.Resources.Requests {
+			if quan, ok := resources[resource]; ok {
+				quantity.Add(quan)
+			}
+			resources[resource] = quantity
+		}
+	}
+
+	nodes := createNUMANodeList(zones)
+	bitmask := bm.NewEmptyBitMask()
+	bitmask.Fill()
+	checkResourcesForNUMANodes(bitmask, nodes, resources, v1qos.GetPodQOS(pod))
+
+	if bitmask.IsEmpty() {
+		// definitly we can't align container, so we can't align a pod
+		return framework.NewStatus(framework.Unschedulable, fmt.Sprintf("Can't align pod: %s", pod.Name))
+	}
+	return nil
+
+}
+
+func createNUMANodeList(zones topologyv1alpha1.ZoneList) NUMANodeList {
 	nodes := make(NUMANodeList, 0)
 	for _, zone := range zones {
 		if zone.Type == "Node" {
@@ -164,14 +205,18 @@ func (tm NodeResourceTopologyMatch) PolicyFilter(pod *v1.Pod, zones topologyv1al
 			nodes = append(nodes, NUMANode{NUMAID: numaID, Resources: resources})
 		}
 	}
-	return filter(containers, nodes, v1qos.GetPodQOS(pod))
+	return nodes
 }
 
 // Filter Now only single-numa-node supported
 func (tm *NodeResourceTopologyMatch) Filter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
 	if nodeInfo.Node() == nil {
 		return framework.NewStatus(framework.Error, fmt.Sprintf("Node is nil %s", nodeInfo.Node().Name))
 	}
+	if v1qos.GetPodQOS(pod) == v1.PodQOSBestEffort {
+		return nil
+	}
+
 	nodeName := nodeInfo.Node().Name
 
 	topologyPolicies := getTopologyPolicies(tm.nodeTopologies, nodeName)
@@ -305,7 +350,8 @@ func NewNodeResourceTopologyMatch(args runtime.Object, handle framework.Framewor
 	topologyMatch.handle = handle
 
 	topologyMatch.topologyPolicyHandlers = make(PolicyHandlerMap)
-	topologyMatch.topologyPolicyHandlers[apiconfig.SingleNUMANodeTopologyManagerPolicy] = topologyMatch
+	topologyMatch.topologyPolicyHandlers[apiconfig.SingleNUMANodeTopologyManagerPolicy] = SingleNUMANodeHandler{match: topologyMatch}
+	topologyMatch.topologyPolicyHandlers[apiconfig.PodTopologyScope] = PodLevelResourceHandler{match: topologyMatch}
 	topologyMatch.nodeTopologies = nodeTopologyMap{}
 
 	return topologyMatch, nil

pkg/noderesourcetopology/match_test.go

@@ -28,6 +28,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	intstr "k8s.io/apimachinery/pkg/util/intstr"
 	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
+	"k8s.io/klog/v2"
 
 	apiconfig "sigs.k8s.io/scheduler-plugins/pkg/apis/config"
 )
@@ -40,11 +41,45 @@ func makePodByResourceList(resources *v1.ResourceList) *v1.Pod {
 	return &v1.Pod{Spec: v1.PodSpec{Containers: []v1.Container{{
 		Resources: v1.ResourceRequirements{
 			Requests: *resources,
+			Limits: *resources,
 		},
 	}},
 	}}
 }
 
+func makeResourceListFromZones(zones topologyv1alpha1.ZoneList) v1.ResourceList {
+	result := make(v1.ResourceList)
+	for _, zone := range zones {
+		for _, resInfo := range zone.Resources {
+			resQuantity, err := resource.ParseQuantity(resInfo.Allocatable.String())
+			if err != nil {
+				klog.Errorf("Failed to parse %s", resInfo.Allocatable.String())
+				continue
+			}
+			if quantity, ok := result[v1.ResourceName(resInfo.Name)]; ok {
+				resQuantity.Add(quantity)
+			}
+			result[v1.ResourceName(resInfo.Name)] = resQuantity
+		}
+	}
+	return result
+}
+
+
+func makePodByResourceListWithManyContainers(resources *v1.ResourceList, containerCount int) *v1.Pod {
+	containers := []v1.Container{}
+
+	for i := 0; i < containerCount; i++ {
+		containers = append(containers, v1.Container{
+			Resources: v1.ResourceRequirements{
+				Requests: *resources,
+				Limits: *resources,
+			},
+		})
+	}
+	return &v1.Pod{Spec: v1.PodSpec{Containers: containers },}
+}
+
 func TestTopologyRequests(t *testing.T) {
 	nodes := nodeTopologyMap{}
 	nodes["node1"] = topologyv1alpha1.NodeResourceTopology{
@@ -92,7 +127,7 @@ func TestTopologyRequests(t *testing.T) {
 	}
 
 	nodes["node2"] = topologyv1alpha1.NodeResourceTopology{
-		ObjectMeta:       metav1.ObjectMeta{Name: "node1"},
+		ObjectMeta:       metav1.ObjectMeta{Name: "node2"},
 		TopologyPolicies: []string{string(apiconfig.SingleNUMANodeTopologyManagerPolicy)},
 		Zones: topologyv1alpha1.ZoneList{
 			topologyv1alpha1.Zone{
@@ -134,12 +169,52 @@ func TestTopologyRequests(t *testing.T) {
 			},
 		},
 	}
-	node1Resources := v1.ResourceList{
-		v1.ResourceCPU:    *resource.NewQuantity(12, resource.DecimalSI),
-		v1.ResourceMemory: resource.MustParse("16Gi"),
-		v1.ResourcePods:   *resource.NewQuantity(20, resource.DecimalSI),
-		nicResourceName:   *resource.NewQuantity(14, resource.DecimalSI),
+
+
+	nodes["node3"] = topologyv1alpha1.NodeResourceTopology{
+		ObjectMeta:       metav1.ObjectMeta{Name: "node3"},
+		TopologyPolicies: []string{string(apiconfig.PodTopologyScope)},
+		Zones: topologyv1alpha1.ZoneList{
+			topologyv1alpha1.Zone{
+				Name: "node-0",
+				Type: "Node",
+				Resources: topologyv1alpha1.ResourceInfoList{
+					topologyv1alpha1.ResourceInfo{
+						Name:        "cpu",
+						Capacity:    intstr.Parse("20"),
+						Allocatable: intstr.Parse("2"),
+					}, topologyv1alpha1.ResourceInfo{
+						Name:        "memory",
+						Capacity:    intstr.Parse("8Gi"),
+						Allocatable: intstr.Parse("4Gi"),
+					}, topologyv1alpha1.ResourceInfo{
+						Name:        nicResourceName,
+						Capacity:    intstr.Parse("30"),
+						Allocatable: intstr.Parse("5"),
+					},
+				},
+			}, topologyv1alpha1.Zone{
+				Name: "node-1",
+				Type: "Node",
+				Resources: topologyv1alpha1.ResourceInfoList{
+					topologyv1alpha1.ResourceInfo{
+						Name:        "cpu",
+						Capacity:    intstr.Parse("30"),
+						Allocatable: intstr.Parse("4"),
+					}, topologyv1alpha1.ResourceInfo{
+						Name:        "memory",
+						Capacity:    intstr.Parse("8Gi"),
+						Allocatable: intstr.Parse("4Gi"),
+					}, topologyv1alpha1.ResourceInfo{
+						Name:        nicResourceName,
+						Capacity:    intstr.Parse("30"),
+						Allocatable: intstr.Parse("2"),
+					},
+				},
+			},
+		},
 	}
+	node1Resources := makeResourceListFromZones(nodes["node1"].Zones)
 	node1 := v1.Node{
 		ObjectMeta: metav1.ObjectMeta{Name: "node1"},
 		Status: v1.NodeStatus{
@@ -148,18 +223,21 @@ func TestTopologyRequests(t *testing.T) {
 		},
 	}
 
-	node2Resources := v1.ResourceList{
-		v1.ResourceCPU:    *resource.NewQuantity(6, resource.DecimalSI),
-		v1.ResourceMemory: resource.MustParse("8Gi"),
-		v1.ResourcePods:   *resource.NewQuantity(20, resource.DecimalSI),
-		nicResourceName:   *resource.NewQuantity(7, resource.DecimalSI),
-	}
-	node2 := v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node1"}, Status: v1.NodeStatus{
+	node2Resources := makeResourceListFromZones(nodes["node2"].Zones)
+	node2 := v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node2"}, Status: v1.NodeStatus{
 		Capacity:    node2Resources,
 		Allocatable: node2Resources,
 	},
 	}
 
+	node3Resources := makeResourceListFromZones(nodes["node3"].Zones)
+
+	node3 := v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node3"}, Status: v1.NodeStatus{
+		Capacity:    node3Resources,
+		Allocatable: node3Resources,
+	},
+	}
+
 	// Test different QoS Guaranteed/Burstable/BestEffort
 	topologyTests := []struct {
 		pod            *v1.Pod
@@ -230,23 +308,47 @@ func TestTopologyRequests(t *testing.T) {
 			nodeTopologies: &nodes,
 			name:           "Guaranteed QoS, pod fit",
 			node:           node1,
-			wantStatus:     nil, //topology_match has to skip request of 0 resources
+			wantStatus:     nil,
+		},
+		{
+			pod: makePodByResourceListWithManyContainers(&v1.ResourceList{
+				v1.ResourceCPU:             *resource.NewQuantity(3, resource.DecimalSI),
+				v1.ResourceMemory:          resource.MustParse("1Gi"),
+				notExistingNICResourceName: *resource.NewQuantity(0, resource.DecimalSI)}, 3),
+			nodeTopologies: &nodes,
+			name:           "Guaranteed QoS Topology Scope, pod doesn't fit",
+			node:           node3,
+			wantStatus:     framework.NewStatus(framework.Unschedulable, "Can't align pod: "),
+		},
+		{
+			pod: makePodByResourceListWithManyContainers(&v1.ResourceList{
+				v1.ResourceCPU:             *resource.NewQuantity(1, resource.DecimalSI),
+				v1.ResourceMemory:          resource.MustParse("1Gi"),
+				notExistingNICResourceName: *resource.NewQuantity(0, resource.DecimalSI)}, 3),
+			nodeTopologies: &nodes,
+			name:           "Guaranteed QoS Topology Scope, pod fit",
+			node:           node3,
+			wantStatus:     nil,
 		},
 	}
 
 	nodeInfo := framework.NewNodeInfo()
 	for _, test := range topologyTests {
 		t.Run(test.name, func(t *testing.T) {
 			tm := NodeResourceTopologyMatch{}
-			tm.nodeTopologies = nodes
+			tm.nodeTopologies = *test.nodeTopologies
 			tm.topologyPolicyHandlers = make(PolicyHandlerMap)
-			tm.topologyPolicyHandlers[apiconfig.SingleNUMANodeTopologyManagerPolicy] = tm
+			tm.topologyPolicyHandlers[apiconfig.SingleNUMANodeTopologyManagerPolicy] = SingleNUMANodeHandler{match: &tm}
+			tm.topologyPolicyHandlers[apiconfig.PodTopologyScope] = PodLevelResourceHandler{match: &tm}
 			nodeInfo.SetNode(&test.node)
 			test.pod.Spec.Containers[0].Name = containerName
-			test.pod.Spec.Containers[0].Resources.Limits = test.pod.Spec.Containers[0].Resources.Requests
+			// this was done to make pod's QoS Guaranted
+			//test.pod.Spec.Containers[0].Resources.Limits = test.pod.Spec.Containers[0].Resources.Requests
 			gotStatus := tm.Filter(context.Background(), framework.NewCycleState(), test.pod, nodeInfo)
+
+			fmt.Printf("test.Name: %v; status: %v\n", test.name, gotStatus)
 			if !reflect.DeepEqual(gotStatus, test.wantStatus) {
-				t.Errorf("status does not match: %v, want: %v", gotStatus, test.wantStatus)
+				t.Errorf("status does not match: %v, want: %v\n", gotStatus, test.wantStatus)
 			}
 		})
 	}