nrt: overreserve: minimize the pod listings

In #557
we replaced the custom indexer implementation with the simpler
lister implementation.
The change works, but we now do unnecessary work relisting for each
node which is processed, which is wasteful: even though client-go
is optimized, let's avoid useless work, because we can relist
the pods once per resync attempt.

Signed-off-by: Francesco Romani <[email protected]>

Author: ffromani

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")
+}