Revert to client-side filtering strategy while keeping status.phase indexer

Co-authored-by: rbtr <[email protected]>

Author: Copilot

cns/ipampool/v2/adapter.go

@@ -43,11 +43,4 @@ func PodIPDemandListener(ch chan<- int) func([]v1.Pod) {
 	}
 }
 
-// PodIPDemandListenerNoFilter counts all pods without client-side filtering.
-// This is used when server-side filtering has already excluded terminal pods.
-func PodIPDemandListenerNoFilter(ch chan<- int) func([]v1.Pod) {
-	return func(pods []v1.Pod) {
-		// All pods are already filtered server-side to exclude terminal phases
-		ch <- len(pods)
-	}
-}
+

cns/ipampool/v2/adapter_test.go

@@ -99,62 +99,4 @@ func TestPodIPDemandListener(t *testing.T) {
 	}
 }
 
-func TestPodIPDemandListenerNoFilter(t *testing.T) {
-	tests := []struct {
-		name     string
-		pods     []v1.Pod
-		expected int
-	}{
-		{
-			name:     "empty pod list",
-			pods:     []v1.Pod{},
-			expected: 0,
-		},
-		{
-			name: "single pod",
-			pods: []v1.Pod{
-				{
-					ObjectMeta: metav1.ObjectMeta{Name: "pod1"},
-					Status:     v1.PodStatus{Phase: v1.PodRunning},
-				},
-			},
-			expected: 1,
-		},
-		{
-			name: "multiple pods - counts all since filtering is done server-side",
-			pods: []v1.Pod{
-				{
-					ObjectMeta: metav1.ObjectMeta{Name: "pod1"},
-					Status:     v1.PodStatus{Phase: v1.PodRunning},
-				},
-				{
-					ObjectMeta: metav1.ObjectMeta{Name: "pod2"},
-					Status:     v1.PodStatus{Phase: v1.PodPending},
-				},
-				{
-					ObjectMeta: metav1.ObjectMeta{Name: "pod3"},
-					Status:     v1.PodStatus{Phase: v1.PodUnknown},
-				},
-			},
-			expected: 3, // All pods counted since server-side filtering already excluded terminal pods
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			ch := make(chan int, 1)
-			listener := PodIPDemandListenerNoFilter(ch)
 
-			listener(tt.pods)
-
-			select {
-			case result := <-ch:
-				if result != tt.expected {
-					t.Errorf("expected %d, got %d", tt.expected, result)
-				}
-			default:
-				t.Error("expected value in channel")
-			}
-		})
-	}
-}

cns/kubecontroller/pod/reconciler.go

@@ -7,7 +7,6 @@ import (
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
 	v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/fields"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/event"
@@ -81,53 +80,7 @@ func (p *watcher) NewNotifierFunc(listOpts *client.ListOptions, limiter limiter,
 	}
 }
 
-// NewPodIPDemandNotifierFunc returns a reconcile.Func optimized for IP demand calculation
-// that uses server-side filtering to exclude terminal Pods (Succeeded/Failed) from the results.
-// This avoids client-side iteration over all pods by using multiple targeted queries.
-func (p *watcher) NewPodIPDemandNotifierFunc(baseListOpts *client.ListOptions, limiter limiter, listeners ...func([]v1.Pod)) reconcile.Func {
-	p.z.Info("adding pod IP demand notifier for listeners", zap.Int("listeners", len(listeners)))
-	return func(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-		if !limiter.Allow() {
-			// rate limit exceeded, requeue
-			p.z.Info("rate limit exceeded")
-			return ctrl.Result{Requeue: true}, nil
-		}
-
-		// Create field selectors for active pod phases (exclude Succeeded and Failed)
-		activePhases := []v1.PodPhase{v1.PodRunning, v1.PodPending, v1.PodUnknown}
-		var allActivePods []v1.Pod
-
-		for _, phase := range activePhases {
-			// Clone base list options and add phase filter
-			listOpts := &client.ListOptions{}
-			if baseListOpts.FieldSelector != nil {
-				// Combine existing field selector with phase filter
-				phaseSelector := fields.SelectorFromSet(fields.Set{"status.phase": string(phase)})
-				combinedSelector := fields.AndSelectors(baseListOpts.FieldSelector, phaseSelector)
-				listOpts.FieldSelector = combinedSelector
-			} else {
-				listOpts.FieldSelector = fields.SelectorFromSet(fields.Set{"status.phase": string(phase)})
-			}
-			
-			// Copy other options from base
-			listOpts.LabelSelector = baseListOpts.LabelSelector
-			listOpts.Namespace = baseListOpts.Namespace
-			listOpts.Limit = baseListOpts.Limit
-			listOpts.Continue = baseListOpts.Continue
 
-			podList := &v1.PodList{}
-			if err := p.cli.List(ctx, podList, listOpts); err != nil {
-				return ctrl.Result{}, errors.Wrapf(err, "failed to list pods with phase %s", phase)
-			}
-			allActivePods = append(allActivePods, podList.Items...)
-		}
-
-		for _, l := range listeners {
-			l(allActivePods)
-		}
-		return ctrl.Result{}, nil
-	}
-}
 
 var hostNetworkIndexer = client.IndexerFunc(func(o client.Object) []string {
 	pod, ok := o.(*v1.Pod)

cns/service/main.go

@@ -1576,16 +1576,10 @@ func InitializeCRDState(ctx context.Context, httpRestService cns.HTTPService, cn
 	if cnsconfig.WatchPods {
 		pw := podctrl.New(z)
 		if cnsconfig.EnableIPAMv2 {
-			// Create field selector to filter only podsubnet pods
-			fieldSet := fields.Set{
-				"spec.hostNetwork": "false",
-			}
-			hostNetworkSelector := fields.SelectorFromSet(fieldSet)
-			baseListOpts := &client.ListOptions{FieldSelector: hostNetworkSelector}
+			hostNetworkListOpt := &client.ListOptions{FieldSelector: fields.SelectorFromSet(fields.Set{"spec.hostNetwork": "false"})} // filter only podsubnet pods
 			// don't relist pods more than every 500ms
 			limit := rate.NewLimiter(rate.Every(500*time.Millisecond), 1) //nolint:gomnd // clearly 500ms
-			// Use specialized notifier that filters terminal pods server-side
-			pw.With(pw.NewPodIPDemandNotifierFunc(baseListOpts, limit, ipampoolv2.PodIPDemandListenerNoFilter(ipDemandCh)))
+			pw.With(pw.NewNotifierFunc(hostNetworkListOpt, limit, ipampoolv2.PodIPDemandListener(ipDemandCh)))
 		}
 		if err := pw.SetupWithManager(ctx, manager); err != nil {
 			return errors.Wrapf(err, "failed to setup pod watcher with manager")