feat: add node name filter for pod owner cache to track local pods only (#4881)

Signed-off-by: Cyclinder Kuo <[email protected]>

Author: cyclinder

charts/spiderpool/templates/daemonset.yaml

@@ -200,6 +200,10 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.namespace
+            - name: SPIDERPOOL_NODE_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: spec.nodeName
             - name: SPIDERPOOL_LOG_LEVEL
               value: {{ .Values.spiderpoolAgent.debug.logLevel | quote }}
             - name: SPIDERPOOL_ENABLED_METRIC

cmd/spiderpool-agent/cmd/config.go

@@ -57,6 +57,7 @@ var envInfo = []envConf{
 	{"SPIDERPOOL_ENABLED_DEBUG_METRIC", "false", false, nil, &agentContext.Cfg.EnableDebugLevelMetric, nil},
 	{"SPIDERPOOL_POD_NAMESPACE", "", true, &agentContext.Cfg.AgentPodNamespace, nil, nil},
 	{"SPIDERPOOL_POD_NAME", "", true, &agentContext.Cfg.AgentPodName, nil, nil},
+	{"SPIDERPOOL_NODE_NAME", "", true, &agentContext.Cfg.NodeName, nil, nil},
 	{"SPIDERPOOL_HEALTH_PORT", "5710", true, &agentContext.Cfg.HttpPort, nil, nil},
 	{"SPIDERPOOL_METRIC_HTTP_PORT", "5711", true, &agentContext.Cfg.MetricHttpPort, nil, nil},
 	{"SPIDERPOOL_GOPS_LISTEN_PORT", "5712", false, &agentContext.Cfg.GopsListenPort, nil, nil},
@@ -84,6 +85,7 @@ type Config struct {
 	EnableDebugLevelMetric               bool
 	AgentPodNamespace                    string
 	AgentPodName                         string
+	NodeName                             string
 	EnableReleaseConflictIPsForStateless bool
 
 	HttpPort         string

cmd/spiderpool-agent/cmd/metrics_server.go

@@ -8,6 +8,8 @@ import (
 	"fmt"
 	"net/http"
 
+	"go.uber.org/zap"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/informers"
 
 	"github.com/spidernet-io/spiderpool/pkg/constant"
@@ -29,10 +31,22 @@ func initAgentMetricsServer(ctx context.Context) {
 	var cache podownercache.CacheInterface
 	// nolint is used to disable the golint warning for the following line.
 	if agentContext.Cfg.EnableRDMAMetric { //nolint:golint
-		logger.Info("enable rdma metric exporter")
-		informerFactory := informers.NewSharedInformerFactory(agentContext.ClientSet, 0)
+		logger.Info("enable rdma metric exporter",
+			zap.String("nodeName", agentContext.Cfg.NodeName))
+
+		// Create informer factory with field selector to only watch pods on this node
+		informerFactory := informers.NewSharedInformerFactoryWithOptions(
+			agentContext.ClientSet,
+			0,
+			informers.WithTweakListOptions(func(options *metav1.ListOptions) {
+				// Filter pods by node name
+				options.FieldSelector = fmt.Sprintf("spec.nodeName=%s", agentContext.Cfg.NodeName)
+			}),
+		)
+
 		podInformer := informerFactory.Core().V1().Pods().Informer()
 		informerFactory.Start(ctx.Done())
+		informerFactory.WaitForCacheSync(ctx.Done())
 
 		cache, err = podownercache.New(ctx, podInformer, agentContext.CRDManager.GetClient())
 		if err != nil {

pkg/constant/k8s.go

@@ -158,6 +158,7 @@ const ClusterDefaultInterfaceName = "eth0"
 const (
 	MultusDefaultNetAnnot        = "v1.multus-cni.io/default-network"
 	MultusNetworkAttachmentAnnot = "k8s.v1.cni.cncf.io/networks"
+	MultusNetworkStatus          = "k8s.v1.cni.cncf.io/network-status"
 	ResourceNameAnnot            = "k8s.v1.cni.cncf.io/resourceName"
 	ResourceNameOvsCniValue      = "ovs-cni.network.kubevirt.io"
 )

pkg/podownercache/pod_owner_cache.go

@@ -5,6 +5,9 @@ package podownercache
 
 import (
 	"context"
+	"strings"
+
+	"github.com/spidernet-io/spiderpool/pkg/constant"
 	"github.com/spidernet-io/spiderpool/pkg/lock"
 	"github.com/spidernet-io/spiderpool/pkg/logutils"
 	"go.uber.org/zap"
@@ -24,6 +27,8 @@ type PodOwnerCache struct {
 	cacheLock lock.RWMutex
 	pods      map[types.NamespacedName]Pod
 	ipToPod   map[string]types.NamespacedName
+	// Cache for final owner references to reduce API calls, using pod NamespacedName as key
+	ownerCache map[types.NamespacedName]*OwnerInfo
 }
 
 type Pod struct {
@@ -50,11 +55,12 @@ func New(ctx context.Context, podInformer cache.SharedIndexInformer, apiReader c
 	logger.Info("create PodOwnerCache informer")
 
 	res := &PodOwnerCache{
-		ctx:       ctx,
-		apiReader: apiReader,
-		cacheLock: lock.RWMutex{},
-		pods:      make(map[types.NamespacedName]Pod),
-		ipToPod:   make(map[string]types.NamespacedName),
+		ctx:        ctx,
+		apiReader:  apiReader,
+		cacheLock:  lock.RWMutex{},
+		pods:       make(map[types.NamespacedName]Pod),
+		ipToPod:    make(map[string]types.NamespacedName),
+		ownerCache: make(map[types.NamespacedName]*OwnerInfo),
 	}
 
 	_, err := podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
@@ -70,11 +76,20 @@ func New(ctx context.Context, podInformer cache.SharedIndexInformer, apiReader c
 	return res, nil
 }
 
-func (s *PodOwnerCache) onPodAdd(obj interface{}) {
+func (s *PodOwnerCache) onPodAdd(obj any) {
 	if pod, ok := obj.(*corev1.Pod); ok {
 		if pod.Spec.HostNetwork {
 			return
 		}
+
+		if pod.Annotations[constant.MultusNetworkStatus] == "" {
+			return
+		}
+
+		if !strings.Contains(pod.Annotations[constant.MultusNetworkStatus], "rdma-device") {
+			return
+		}
+
 		if len(pod.Status.PodIPs) > 0 {
 			ips := make([]string, 0, len(pod.Status.PodIPs))
 			for _, p := range pod.Status.PodIPs {
@@ -123,6 +138,23 @@ func (s *PodOwnerCache) onPodDel(obj interface{}) {
 func (s *PodOwnerCache) getFinalOwner(obj metav1.Object) (*OwnerInfo, error) {
 	var finalOwner *OwnerInfo
 
+	// Create pod NamespacedName as the cache key
+	podKey := types.NamespacedName{
+		Namespace: obj.GetNamespace(),
+		Name:      obj.GetName(),
+	}
+
+	// Check if we already have a cached final owner for this pod
+	s.cacheLock.RLock()
+	cachedOwner, exists := s.ownerCache[podKey]
+	s.cacheLock.RUnlock()
+
+	if exists {
+		// If we found a cached result, return it immediately
+		logger.Sugar().Debugf("Using cached owner for pod %s/%s", obj.GetNamespace(), obj.GetName())
+		return cachedOwner, nil
+	}
+
 	for {
 		ownerRefs := obj.GetOwnerReferences()
 		if len(ownerRefs) == 0 {
@@ -131,6 +163,8 @@ func (s *PodOwnerCache) getFinalOwner(obj metav1.Object) (*OwnerInfo, error) {
 
 		// Assuming the first owner reference
 		ownerRef := ownerRefs[0]
+
+		// If not in cache, create the owner info
 		finalOwner = &OwnerInfo{
 			APIVersion: ownerRef.APIVersion,
 			Kind:       ownerRef.Kind,
@@ -148,12 +182,12 @@ func (s *PodOwnerCache) getFinalOwner(obj metav1.Object) (*OwnerInfo, error) {
 			Name:      ownerRef.Name,
 		}, ownerObj)
 		if err != nil {
-			if errors.IsForbidden(err) {
-				logger.Sugar().Debugf("forbidden to get owner of pod %s/%s", obj.GetNamespace(), obj.GetName())
-				return nil, nil
-			}
-			if errors.IsNotFound(err) {
-				logger.Sugar().Debugf("owner not found for pod %s/%s", obj.GetNamespace(), obj.GetName())
+			if errors.IsForbidden(err) || errors.IsNotFound(err) {
+				logger.Sugar().Debugf("%v for pod %s/%s", err, obj.GetNamespace(), obj.GetName())
+				// Cache the negative result
+				s.cacheLock.Lock()
+				s.ownerCache[podKey] = nil
+				s.cacheLock.Unlock()
 				return nil, nil
 			}
 			return nil, err
@@ -163,6 +197,19 @@ func (s *PodOwnerCache) getFinalOwner(obj metav1.Object) (*OwnerInfo, error) {
 		obj = ownerObj
 	}
 
+	// Cache the final owner (or nil if no owner found)
+	s.cacheLock.Lock()
+	s.ownerCache[podKey] = finalOwner
+	s.cacheLock.Unlock()
+
+	if finalOwner != nil {
+		logger.Sugar().Debugf("Cached final owner %s/%s of kind %s for pod %s/%s",
+			finalOwner.Namespace, finalOwner.Name, finalOwner.Kind,
+			podKey.Namespace, podKey.Name)
+	} else {
+		logger.Sugar().Debugf("No owner found for pod %s/%s", podKey.Namespace, podKey.Name)
+	}
+
 	return finalOwner, nil
 }
 

pkg/podownercache/pod_owner_cache_test.go

@@ -16,6 +16,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	kruntime "k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	types "k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes/fake"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -42,6 +43,12 @@ func TestPodOwnerCache(t *testing.T) {
 					Name:       "test-rs",
 				},
 			},
+			Annotations: map[string]string{
+				"k8s.v1.cni.cncf.io/network-status": `[{
+					"name": "rdma-device",
+					"interface": "eth0"
+				}]`,
+			},
 		},
 		Status: corev1.PodStatus{
 			PodIPs: []corev1.PodIP{
@@ -56,6 +63,12 @@ func TestPodOwnerCache(t *testing.T) {
 		ObjectMeta: v1.ObjectMeta{
 			Name:      "test-pod-2",
 			Namespace: "test-ns",
+			Annotations: map[string]string{
+				"k8s.v1.cni.cncf.io/network-status": `[{
+					"name": "rdma-device",
+					"interface": "eth0"
+				}]`,
+			},
 		},
 		Status: corev1.PodStatus{
 			PodIPs: []corev1.PodIP{
@@ -197,8 +210,9 @@ func TestGetFinalOwnerForbidden(t *testing.T) {
 
 	mockClient := &MockForbiddenClient{}
 	cache := &PodOwnerCache{
-		ctx:       context.Background(),
-		apiReader: mockClient,
+		ctx:        context.Background(),
+		apiReader:  mockClient,
+		ownerCache: map[types.NamespacedName]*OwnerInfo{},
 	}
 
 	// Create a mock pod object
@@ -242,8 +256,9 @@ func TestGetFinalOwnerNotFound(t *testing.T) {
 
 	mockClient := &MockNotFoundClient{}
 	cache := &PodOwnerCache{
-		ctx:       context.Background(),
-		apiReader: mockClient,
+		ctx:        context.Background(),
+		apiReader:  mockClient,
+		ownerCache: map[types.NamespacedName]*OwnerInfo{},
 	}
 
 	// Create a mock pod object