Merge pull request #580 from ffromani/nrt-minimize-list

nrt: overreserve: minimize the pod listings

Author: k8s-ci-robot

pkg/noderesourcetopology/cache/overreserve.go

@@ -191,6 +191,13 @@ func (ov *OverReserve) Resync() {
 		return
 	}
 
+	// node -> pod identifier (namespace, name)
+	nodeToObjsMap, err := makeNodeToNamespacedNamesMap(ov.podLister, logID)
+	if err != nil {
+		klog.ErrorS(err, "cannot find the mapping between running pods and nodes")
+		return
+	}
+
 	klog.V(6).InfoS("nrtcache: resync NodeTopology cache starting", "logID", logID)
 	defer klog.V(6).InfoS("nrtcache: resync NodeTopology cache complete", "logID", logID)
 
@@ -206,6 +213,13 @@ func (ov *OverReserve) Resync() {
 			continue
 		}
 
+		objs, ok := nodeToObjsMap[nodeName]
+		if !ok {
+			// this really should never happen
+			klog.V(3).InfoS("nrtcache: cannot find any pod for node", "logID", logID, "node", nodeName)
+			continue
+		}
+
 		pfpExpected := podFingerprintForNodeTopology(nrtCandidate)
 		if pfpExpected == "" {
 			klog.V(3).InfoS("nrtcache: missing NodeTopology podset fingerprint data", "logID", logID, "node", nodeName)
@@ -214,7 +228,7 @@ func (ov *OverReserve) Resync() {
 
 		klog.V(6).InfoS("nrtcache: trying to resync NodeTopology", "logID", logID, "node", nodeName, "fingerprint", pfpExpected)
 
-		err = checkPodFingerprintForNode(logID, ov.podLister, nodeName, pfpExpected)
+		err = checkPodFingerprintForNode(logID, objs, nodeName, pfpExpected)
 		if errors.Is(err, podfingerprint.ErrSignatureMismatch) {
 			// can happen, not critical
 			klog.V(5).InfoS("nrtcache: NodeTopology podset fingerprint mismatch", "logID", logID, "node", nodeName)

pkg/noderesourcetopology/cache/overreserve_test.go

@@ -18,7 +18,6 @@ package cache
 
 import (
 	"encoding/json"
-	"fmt"
 	"reflect"
 	"sort"
 	"testing"
@@ -31,7 +30,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
 	podlisterv1 "k8s.io/client-go/listers/core/v1"
 )
 
@@ -657,39 +655,6 @@ func dumpNRT(nrtObj *topologyv1alpha2.NodeResourceTopology) string {
 	return string(nrtJson)
 }
 
-type fakePodLister struct {
-	pods []*corev1.Pod
-	err  error
-}
-
-type fakePodNamespaceLister struct {
-	parent    *fakePodLister
-	namespace string
-}
-
-func (fpl *fakePodLister) AddPod(pod *corev1.Pod) {
-	fpl.pods = append(fpl.pods, pod)
-}
-
-func (fpl *fakePodLister) List(selector labels.Selector) ([]*corev1.Pod, error) {
-	return fpl.pods, fpl.err
-}
-
-func (fpl *fakePodLister) Pods(namespace string) podlisterv1.PodNamespaceLister {
-	return &fakePodNamespaceLister{
-		parent:    fpl,
-		namespace: namespace,
-	}
-}
-
-func (fpnl *fakePodNamespaceLister) List(selector labels.Selector) ([]*corev1.Pod, error) {
-	return nil, fmt.Errorf("not yet implemented")
-}
-
-func (fpnl *fakePodNamespaceLister) Get(name string) (*corev1.Pod, error) {
-	return nil, fmt.Errorf("not yet implemented")
-}
-
 func MakeTopologyResInfo(name, capacity, available string) topologyv1alpha2.ResourceInfo {
 	return topologyv1alpha2.ResourceInfo{
 		Name:      name,

pkg/noderesourcetopology/cache/store.go

@@ -211,12 +211,7 @@ func podFingerprintForNodeTopology(nrt *topologyv1alpha2.NodeResourceTopology) s
 // checkPodFingerprintForNode verifies if the given pods fingeprint (usually from NRT update) matches the
 // computed one using the stored data about pods running on nodes. Returns nil on success, or an error
 // describing the failure
-func checkPodFingerprintForNode(logID string, podLister podlisterv1.PodLister, nodeName, pfpExpected string) error {
-	objs, err := getPodNamespacedNamesByNode(podLister, logID, nodeName)
-	if err != nil {
-		return err
-	}
-
+func checkPodFingerprintForNode(logID string, objs []types.NamespacedName, nodeName, pfpExpected string) error {
 	st := podfingerprint.MakeStatus(nodeName)
 	pfp := podfingerprint.NewTracingFingerprint(len(objs), &st)
 	for _, obj := range objs {
@@ -230,24 +225,23 @@ func checkPodFingerprintForNode(logID string, podLister podlisterv1.PodLister, n
 	return pfp.Check(pfpExpected)
 }
 
-func getPodNamespacedNamesByNode(podLister podlisterv1.PodLister, logID, nodeName string) ([]types.NamespacedName, error) {
+func makeNodeToNamespacedNamesMap(podLister podlisterv1.PodLister, logID string) (map[string][]types.NamespacedName, error) {
+	nodeToObjsMap := make(map[string][]types.NamespacedName)
 	pods, err := podLister.List(labels.Everything())
 	if err != nil {
-		return []types.NamespacedName{}, err
+		return nodeToObjsMap, err
 	}
-	objs := make([]types.NamespacedName, 0, len(pods))
 	for _, pod := range pods {
-		if pod.Spec.NodeName != nodeName {
-			continue
-		}
 		if pod.Status.Phase != corev1.PodRunning {
 			// we are interested only about nodes which consume resources
 			continue
 		}
-		objs = append(objs, types.NamespacedName{
+		nodeObjs := nodeToObjsMap[pod.Spec.NodeName]
+		nodeObjs = append(nodeObjs, types.NamespacedName{
 			Namespace: pod.Namespace,
 			Name:      pod.Name,
 		})
+		nodeToObjsMap[pod.Spec.NodeName] = nodeObjs
 	}
-	return objs, nil
+	return nodeToObjsMap, nil
 }

pkg/noderesourcetopology/cache/store_test.go

@@ -17,14 +17,20 @@ limitations under the License.
 package cache
 
 import (
+	"errors"
+	"fmt"
 	"reflect"
 	"sort"
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
 	topologyv1alpha2 "github.com/k8stopologyawareschedwg/noderesourcetopology-api/pkg/apis/topology/v1alpha2"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	podlisterv1 "k8s.io/client-go/listers/core/v1"
 
 	"github.com/k8stopologyawareschedwg/podfingerprint"
 )
@@ -451,6 +457,157 @@ func TestResourceStoreUpdate(t *testing.T) {
 	}
 }
 
+func TestMakeNodeToNamespacedNamesMap(t *testing.T) {
+	tcases := []struct {
+		description string
+		pods        []*corev1.Pod
+		err         error
+		expected    map[string][]types.NamespacedName
+		expectedErr error
+	}{
+		{
+			description: "empty pod list",
+			expected:    make(map[string][]types.NamespacedName),
+		},
+		{
+			description: "single pod NOT running",
+			pods: []*corev1.Pod{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "namespace1",
+						Name:      "pod1",
+					},
+					Spec: corev1.PodSpec{
+						NodeName: "node1",
+					},
+				},
+			},
+			expected: make(map[string][]types.NamespacedName),
+		},
+		{
+			description: "single pod running",
+			pods: []*corev1.Pod{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "namespace1",
+						Name:      "pod1",
+					},
+					Spec: corev1.PodSpec{
+						NodeName: "node1",
+					},
+					Status: corev1.PodStatus{
+						Phase: corev1.PodRunning,
+					},
+				},
+			},
+			expected: map[string][]types.NamespacedName{
+				"node1": {
+					{
+						Namespace: "namespace1",
+						Name:      "pod1",
+					},
+				},
+			},
+		},
+		{
+			description: "few pods, single node running",
+			pods: []*corev1.Pod{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "namespace1",
+						Name:      "pod1",
+					},
+					Spec: corev1.PodSpec{
+						NodeName: "node1",
+					},
+					Status: corev1.PodStatus{
+						Phase: corev1.PodRunning,
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "namespace2",
+						Name:      "pod2",
+					},
+					Spec: corev1.PodSpec{
+						NodeName: "node1",
+					},
+					Status: corev1.PodStatus{
+						Phase: corev1.PodRunning,
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "namespace2",
+						Name:      "pod3",
+					},
+					Spec: corev1.PodSpec{
+						NodeName: "node1",
+					},
+					Status: corev1.PodStatus{
+						Phase: corev1.PodRunning,
+					},
+				},
+			},
+			expected: map[string][]types.NamespacedName{
+				"node1": {
+					{
+						Namespace: "namespace1",
+						Name:      "pod1",
+					},
+					{
+						Namespace: "namespace2",
+						Name:      "pod2",
+					},
+					{
+						Namespace: "namespace2",
+						Name:      "pod3",
+					},
+				},
+			},
+		},
+	}
+
+	for _, tcase := range tcases {
+		t.Run(tcase.description, func(t *testing.T) {
+			podLister := &fakePodLister{
+				pods: tcase.pods,
+				err:  tcase.err,
+			}
+			got, err := makeNodeToNamespacedNamesMap(podLister, tcase.description)
+			if err != tcase.expectedErr {
+				t.Errorf("error mismatch: got %v expected %v", err, tcase.expectedErr)
+			}
+			if diff := cmp.Diff(got, tcase.expected); diff != "" {
+				t.Errorf("unexpected result: %v", diff)
+			}
+		})
+	}
+}
+
+func TestCheckPodFingerprintForNode(t *testing.T) {
+	tcases := []struct {
+		description string
+		objs        []types.NamespacedName
+		pfp         string
+		expectedErr error
+	}{
+		{
+			description: "nil objs",
+			expectedErr: podfingerprint.ErrMalformed,
+		},
+	}
+
+	for _, tcase := range tcases {
+		t.Run(tcase.description, func(t *testing.T) {
+			gotErr := checkPodFingerprintForNode("testing", tcase.objs, "test-node", tcase.pfp)
+			if !errors.Is(gotErr, tcase.expectedErr) {
+				t.Errorf("got error %v expected %v", gotErr, tcase.expectedErr)
+			}
+		})
+	}
+}
+
 func findResourceInfo(rinfos []topologyv1alpha2.ResourceInfo, name string) *topologyv1alpha2.ResourceInfo {
 	for idx := 0; idx < len(rinfos); idx++ {
 		if rinfos[idx].Name == name {
@@ -459,3 +616,36 @@ func findResourceInfo(rinfos []topologyv1alpha2.ResourceInfo, name string) *topo
 	}
 	return nil
 }
+
+type fakePodLister struct {
+	pods []*corev1.Pod
+	err  error
+}
+
+type fakePodNamespaceLister struct {
+	parent    *fakePodLister
+	namespace string
+}
+
+func (fpl *fakePodLister) AddPod(pod *corev1.Pod) {
+	fpl.pods = append(fpl.pods, pod)
+}
+
+func (fpl *fakePodLister) List(selector labels.Selector) ([]*corev1.Pod, error) {
+	return fpl.pods, fpl.err
+}
+
+func (fpl *fakePodLister) Pods(namespace string) podlisterv1.PodNamespaceLister {
+	return &fakePodNamespaceLister{
+		parent:    fpl,
+		namespace: namespace,
+	}
+}
+
+func (fpnl *fakePodNamespaceLister) List(selector labels.Selector) ([]*corev1.Pod, error) {
+	return nil, fmt.Errorf("not yet implemented")
+}
+
+func (fpnl *fakePodNamespaceLister) Get(name string) (*corev1.Pod, error) {
+	return nil, fmt.Errorf("not yet implemented")
+}