Improve stale IP recycling in AntreaIPAM controller (#7538)

When a Node is deleted, the CNI DEL command may not be invoked for Pods
running on that Node. As a result, antrea-agent cannot release their IPs,
leaving stale entries in the IPPool. To mitigate this, increase the IP
recycle interval from 10 minutes to 1 minute.

Additionally, add a benchmark test for the IP recycle function and refine
its implementation based on benchmark results to reduce memory allocations
and improve cleanup performance.

Signed-off-by: Lan Luo <[email protected]>

Author: luolanzone

pkg/controller/ipam/antrea_ipam_controller.go

@@ -54,7 +54,7 @@ const (
 	minRetryDelay = 5 * time.Second
 	maxRetryDelay = 300 * time.Second
 
-	garbageCollectionInterval = 10 * time.Minute
+	garbageCollectionInterval = 1 * time.Minute
 
 	// Default number of workers processing an IPPool change.
 	defaultWorkers = 4
@@ -171,32 +171,37 @@ func (c *AntreaIPAMController) enqueueStatefulSetDeleteEvent(obj interface{}) {
 }
 
 // Inspect all IPPools for stale IP Address entries.
-// This may happen if controller was down during StatefulSet delete event.
-// If such entry is found, enqueue cleanup event for this StatefulSet.
-func (c *AntreaIPAMController) cleanupStaleAddresses() {
-	pools, _ := c.ipPoolLister.List(labels.Everything())
-	lister := c.statefulSetInformer.Lister()
-	statefulSets, _ := lister.List(labels.Everything())
-	statefulSetMap := make(map[string]bool)
-
+// This may happen if controller was down during StatefulSet/Pod delete event.
+// If such entry is found, enqueue cleanup event for this StatefulSet/Pod.
+func (c *AntreaIPAMController) cleanUpStaleIPAddresses() {
 	klog.InfoS("Cleanup job for IP Pools started")
 
+	pools, _ := c.ipPoolLister.List(labels.Everything())
+	statefulSets, _ := c.statefulSetInformer.Lister().List(labels.Everything())
+	statefulSetMap := make(map[string]struct{}, len(statefulSets))
+
 	for _, ss := range statefulSets {
 		// Prepare map of existing StatefulSets for quick reference below
-		statefulSetMap[k8s.NamespacedName(ss.Namespace, ss.Name)] = true
+		statefulSetMap[k8s.NamespacedName(ss.Namespace, ss.Name)] = struct{}{}
+	}
+
+	pods, _ := c.podLister.List(labels.Everything())
+	podsMap := make(map[string]struct{}, len(pods))
+	for _, p := range pods {
+		podsMap[k8s.NamespacedName(p.Namespace, p.Name)] = struct{}{}
 	}
 
 	poolsUpdated := 0
 	for _, ipPool := range pools {
 		updateNeeded := false
-		ipPoolCopy := ipPool.DeepCopy()
-		var newList []crdv1b1.IPAddressState
-		for _, address := range ipPoolCopy.Status.IPAddresses {
+		ipAddrs := ipPool.Status.IPAddresses
+		newList := make([]crdv1b1.IPAddressState, 0, len(ipAddrs))
+
+		for _, address := range ipAddrs {
 			// Cleanup reserved addresses
 			if address.Owner.Pod != nil {
-				_, err := c.podLister.Pods(address.Owner.Pod.Namespace).Get(address.Owner.Pod.Name)
-				if err != nil && errors.IsNotFound(err) {
-					klog.InfoS("IPPool contains stale IPAddress for Pod that no longer exists", "IPPool", ipPool.Name, "Namespace", address.Owner.Pod.Namespace, "Pod", address.Owner.Pod.Name)
+				if _, ok := podsMap[k8s.NamespacedName(address.Owner.Pod.Namespace, address.Owner.Pod.Name)]; !ok {
+					klog.V(2).InfoS("IPPool contains stale IP address for Pod that no longer exists", "IPPool", ipPool.Name, "Namespace", address.Owner.Pod.Namespace, "Pod", address.Owner.Pod.Name)
 					address.Owner.Pod = nil
 					if address.Owner.StatefulSet != nil {
 						address.Phase = crdv1b1.IPAddressPhaseReserved
@@ -208,10 +213,9 @@ func (c *AntreaIPAMController) cleanupStaleAddresses() {
 				key := k8s.NamespacedName(address.Owner.StatefulSet.Namespace, address.Owner.StatefulSet.Name)
 				if _, ok := statefulSetMap[key]; !ok {
 					// This entry refers to StatefulSet that no longer exists
-					klog.InfoS("IPPool contains stale IPAddress for StatefulSet that no longer exists", "IPPool", ipPool.Name, "Namespace", address.Owner.StatefulSet.Namespace, "StatefulSet", address.Owner.StatefulSet.Name)
+					klog.V(2).InfoS("IPPool contains stale IP address for StatefulSet that no longer exists", "IPPool", ipPool.Name, "Namespace", address.Owner.StatefulSet.Namespace, "StatefulSet", address.Owner.StatefulSet.Name)
 					address.Owner.StatefulSet = nil
 					updateNeeded = true
-
 				}
 			}
 
@@ -221,6 +225,7 @@ func (c *AntreaIPAMController) cleanupStaleAddresses() {
 		}
 
 		if updateNeeded {
+			ipPoolCopy := ipPool.DeepCopy()
 			ipPoolCopy.Status.IPAddresses = newList
 			_, err := c.crdClient.CrdV1beta1().IPPools().UpdateStatus(context.TODO(), ipPoolCopy, metav1.UpdateOptions{})
 			if err != nil {
@@ -229,7 +234,6 @@ func (c *AntreaIPAMController) cleanupStaleAddresses() {
 			} else {
 				poolsUpdated += 1
 			}
-
 		}
 	}
 
@@ -242,6 +246,8 @@ func (c *AntreaIPAMController) cleanIPPoolForStatefulSet(namespacedName string)
 	klog.InfoS("Processing delete notification", "StatefulSet", namespacedName)
 	ipPools, _ := c.ipPoolInformer.Informer().GetIndexer().ByIndex(statefulSetIndex, namespacedName)
 
+	var retry bool
+	namespace, name := k8s.SplitNamespacedName(namespacedName)
 	for _, item := range ipPools {
 		ipPool := item.(*crdv1b1.IPPool)
 		allocator, err := poolallocator.NewIPPoolAllocator(ipPool.Name, c.crdClient, c.ipPoolLister)
@@ -251,15 +257,18 @@ func (c *AntreaIPAMController) cleanIPPoolForStatefulSet(namespacedName string)
 			continue
 		}
 
-		namespace, name := k8s.SplitNamespacedName(namespacedName)
 		err = allocator.ReleaseStatefulSet(namespace, name)
 		if err != nil {
 			// This can be a transient error - worker will retry
 			klog.ErrorS(err, "Failed to clean IP allocations", "StatefulSet", namespacedName, "IPPool", ipPool.Name)
+			retry = true
 			continue
 		}
 	}
 
+	if retry {
+		return fmt.Errorf("stale IP allocations cleanup failed for at least one IPPool")
+	}
 	return nil
 }
 
@@ -491,7 +500,7 @@ func (c *AntreaIPAMController) Run(stopCh <-chan struct{}) {
 	}
 
 	// Periodic cleanup IP Pools of stale IP addresses
-	go wait.NonSlidingUntil(c.cleanupStaleAddresses, garbageCollectionInterval, stopCh)
+	go wait.NonSlidingUntil(c.cleanUpStaleIPAddresses, garbageCollectionInterval, stopCh)
 
 	go wait.Until(c.statefulSetWorker, time.Second, stopCh)
 

pkg/controller/ipam/antrea_ipam_controller_test.go

@@ -28,8 +28,11 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes/fake"
+	k8stesting "k8s.io/client-go/testing"
+	"k8s.io/client-go/tools/cache"
 
 	crdv1b1 "antrea.io/antrea/pkg/apis/crd/v1beta1"
 	fakecrd "antrea.io/antrea/pkg/client/clientset/versioned/fake"
@@ -334,3 +337,84 @@ func TestAntreaIPAMController_getIPPoolsForStatefulSet(t *testing.T) {
 		})
 	}
 }
+
+// Creates fake IPPools with N addresses each
+func makeFakeIPPools(poolCount, ipsPerPool int) []*crdv1b1.IPPool {
+	var pools []*crdv1b1.IPPool
+	for i := 0; i < poolCount; i++ {
+		pool := &crdv1b1.IPPool{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: fmt.Sprintf("pool-%d", i),
+			},
+			Status: crdv1b1.IPPoolStatus{
+				IPAddresses: []crdv1b1.IPAddressState{},
+			},
+		}
+
+		for j := 0; j < ipsPerPool; j++ {
+			ip := crdv1b1.IPAddressState{
+				IPAddress: fmt.Sprintf("10.0.%d.%d", i, j),
+				Phase:     crdv1b1.IPAddressPhaseAllocated,
+			}
+			// Randomly assign Pod/StatefulSet owner
+			ip.Owner.Pod = &crdv1b1.PodOwner{
+				Namespace:   "default",
+				Name:        fmt.Sprintf("pod-%d-%d", i, j),
+				ContainerID: fmt.Sprintf("container-%d-%d", i, j),
+			}
+			ip.Owner.StatefulSet = &crdv1b1.StatefulSetOwner{
+				Namespace: "default",
+				Name:      fmt.Sprintf("ss-%d", i),
+				Index:     1,
+			}
+			pool.Status.IPAddresses = append(pool.Status.IPAddresses, ip)
+		}
+		pools = append(pools, pool)
+	}
+	return pools
+}
+
+func BenchmarkCleanUpStaleIPAddresses(b *testing.B) {
+	benchmarks := []struct {
+		name       string
+		poolCount  int
+		ipsPerPool int
+	}{
+		// Sample result for small:
+		//   1515	    764513 ns/op	  292771 B/op	    5255 allocs/op
+		{"Small", 10, 50},
+		// Sample result for medium:
+		// 39	  27798675 ns/op	10321643 B/op	  202300 allocs/op
+		{"Medium", 100, 200},
+		// Sample result for large:
+		// 5	 205046442 ns/op	126322694 B/op	 2510655 allocs/op
+		{"Large", 500, 500},
+	}
+
+	for _, bm := range benchmarks {
+		b.Run(bm.name, func(b *testing.B) {
+			stopCh := make(chan struct{})
+			defer close(stopCh)
+
+			c := newFakeAntreaIPAMController(&crdv1b1.IPPool{}, &corev1.Namespace{}, &appsv1.StatefulSet{})
+			watcher := watch.NewFake()
+			c.fakeCRDClient.PrependWatchReactor("ippools", k8stesting.DefaultWatchReactor(watcher, nil))
+			c.informerFactory.Start(stopCh)
+			c.crdInformerFactory.Start(stopCh)
+			c.informerFactory.WaitForCacheSync(stopCh)
+			c.crdInformerFactory.WaitForCacheSync(stopCh)
+
+			pools := makeFakeIPPools(bm.poolCount, bm.ipsPerPool)
+			for _, p := range pools {
+				_, _ = c.fakeCRDClient.CrdV1beta1().IPPools().Create(context.TODO(), p, metav1.CreateOptions{})
+				watcher.Add(p)
+			}
+			cache.WaitForCacheSync(stopCh, c.ipPoolInformer.Informer().HasSynced)
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				c.cleanUpStaleIPAddresses()
+			}
+		})
+	}
+}

test/performance/benchmark.yml

@@ -75,3 +75,6 @@ benchmarks:
     package: "./pkg/apis/controlplane"
   - name: "BenchmarkSymmetricDifferenceString"
     package: "./pkg/util/sets"
+  - name: "BenchmarkCleanUpStaleIPAddresses"
+    package: "./pkg/controller/ipam"
+    threshold: 0.3