Merge pull request #2539 from Icarus9913/fix/wk/duplicate-ip

optimize codes to verify IP update

pkg/gcmanager/scanAll_IPPool.go

@@ -207,15 +207,13 @@ func (s *SpiderGC) executeScanAll(ctx context.Context) {
 						if podmanager.IsStaticIPPod(s.gcConfig.EnableStatefulSet, s.gcConfig.EnableKubevirtStaticIP, podYaml) {
 							scanAllLogger.Sugar().Debugf("Static IP Pod just restarts, keep the static IP '%s' from the IPPool", poolIP)
 						} else {
-							wrappedLog := scanAllLogger.With(zap.String("gc-reason", "IPPoolAllocation pod UID is different with Endpoint pod UID"))
+							wrappedLog := scanAllLogger.With(zap.String("gc-reason", "IPPoolAllocation pod UID is different with pod UID"))
 							// we are afraid that no one removes the old same name Endpoint finalizer
-							err := s.releaseSingleIPAndRemoveWEPFinalizer(ctx, pool.Name, poolIP, poolIPAllocation)
+							err := s.releaseSingleIPAndRemoveWEPFinalizer(logutils.IntoContext(ctx, wrappedLog), pool.Name, poolIP, poolIPAllocation)
 							if nil != err {
 								wrappedLog.Sugar().Errorf("failed to release ip '%s', error: '%v'", poolIP, err)
 								continue
 							}
-
-							wrappedLog.Sugar().Infof("release ip '%s' successfully!", poolIP)
 						}
 					} else {
 						endpoint, err := s.wepMgr.GetEndpointByName(ctx, podYaml.Namespace, podYaml.Name, constant.UseCache)

pkg/ipam/allocate.go

@@ -16,6 +16,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/utils/strings/slices"
 	kubevirtv1 "kubevirt.io/api/core/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -108,7 +109,7 @@ func (i *ipam) retrieveStaticIPAllocation(ctx context.Context, nic string, pod *
 
 	logger.Info("Concurrently refresh IP records of IPPools")
 	if err := i.reallocateIPPoolIPRecords(ctx, string(pod.UID), endpoint); err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to reallocate IPPool IP records, error: %w", err)
 	}
 
 	logger.Info("Refresh the current IP allocation of the Endpoint")
@@ -129,6 +130,11 @@ func (i *ipam) retrieveStaticIPAllocation(ctx context.Context, nic string, pod *
 func (i *ipam) reallocateIPPoolIPRecords(ctx context.Context, uid string, endpoint *spiderpoolv2beta1.SpiderEndpoint) error {
 	logger := logutils.FromContext(ctx)
 
+	namespaceKey, err := cache.MetaNamespaceKeyFunc(endpoint)
+	if nil != err {
+		return fmt.Errorf("failed to parse object %+v meta key", endpoint)
+	}
+
 	pius := convert.GroupIPAllocationDetails(uid, endpoint.Status.Current.IPs)
 	tickets := pius.Pools()
 	timeRecorder := metric.NewTimeRecorder()
@@ -148,7 +154,7 @@ func (i *ipam) reallocateIPPoolIPRecords(ctx context.Context, uid string, endpoi
 		go func(poolName string, ipAndUIDs []types.IPAndUID) {
 			defer wg.Done()
 
-			if err := i.ipPoolManager.UpdateAllocatedIPs(ctx, poolName, ipAndUIDs); err != nil {
+			if err := i.ipPoolManager.UpdateAllocatedIPs(ctx, poolName, namespaceKey, ipAndUIDs); err != nil {
 				logger.Warn(err.Error())
 				errCh <- err
 				return

pkg/ippoolmanager/ippool_manager.go

@@ -34,7 +34,7 @@ type IPPoolManager interface {
 	ListIPPools(ctx context.Context, cached bool, opts ...client.ListOption) (*spiderpoolv2beta1.SpiderIPPoolList, error)
 	AllocateIP(ctx context.Context, poolName, nic string, pod *corev1.Pod, podController types.PodTopController) (*models.IPConfig, error)
 	ReleaseIP(ctx context.Context, poolName string, ipAndUIDs []types.IPAndUID) error
-	UpdateAllocatedIPs(ctx context.Context, poolName string, ipAndCIDs []types.IPAndUID) error
+	UpdateAllocatedIPs(ctx context.Context, poolName, namespacedName string, ipAndCIDs []types.IPAndUID) error
 }
 
 type ipPoolManager struct {
@@ -291,7 +291,7 @@ func (im *ipPoolManager) ReleaseIP(ctx context.Context, poolName string, ipAndUI
 	return nil
 }
 
-func (im *ipPoolManager) UpdateAllocatedIPs(ctx context.Context, poolName string, ipAndUIDs []types.IPAndUID) error {
+func (im *ipPoolManager) UpdateAllocatedIPs(ctx context.Context, poolName, namespacedName string, ipAndUIDs []types.IPAndUID) error {
 	logger := logutils.FromContext(ctx)
 
 	backoff := retry.DefaultRetry
@@ -315,6 +315,10 @@ func (im *ipPoolManager) UpdateAllocatedIPs(ctx context.Context, poolName string
 		recreate := false
 		for _, iu := range ipAndUIDs {
 			if record, ok := allocatedRecords[iu.IP]; ok {
+				if record.NamespacedName != namespacedName {
+					return fmt.Errorf("failed to update allocated IP because of data broken: IPPool %s IP %s allocation detail %v mistach namespacedName %s",
+						poolName, iu.IP, record, namespacedName)
+				}
 				if record.PodUID != iu.UID {
 					record.PodUID = iu.UID
 					allocatedRecords[iu.IP] = record

pkg/ippoolmanager/ippool_manager_test.go

@@ -577,7 +577,7 @@ var _ = Describe("IPPoolManager", Label("ippool_manager_test"), func() {
 			})
 
 			It("updates the allocated IP record from non-existent IPPool", func() {
-				err := ipPoolManager.UpdateAllocatedIPs(ctx, ipPoolName, []spiderpooltypes.IPAndUID{})
+				err := ipPoolManager.UpdateAllocatedIPs(ctx, ipPoolName, "default/pod", []spiderpooltypes.IPAndUID{})
 				Expect(apierrors.IsNotFound(err)).To(BeTrue())
 			})
 
@@ -589,7 +589,7 @@ var _ = Describe("IPPoolManager", Label("ippool_manager_test"), func() {
 				err = tracker.Add(ipPoolT)
 				Expect(err).NotTo(HaveOccurred())
 
-				err = ipPoolManager.UpdateAllocatedIPs(ctx, ipPoolName, []spiderpooltypes.IPAndUID{{IP: ip, UID: uid}})
+				err = ipPoolManager.UpdateAllocatedIPs(ctx, ipPoolName, "default/pod", []spiderpooltypes.IPAndUID{{IP: ip, UID: uid}})
 				Expect(err).NotTo(HaveOccurred())
 			})
 
@@ -606,7 +606,7 @@ var _ = Describe("IPPoolManager", Label("ippool_manager_test"), func() {
 				err = tracker.Add(ipPoolT)
 				Expect(err).NotTo(HaveOccurred())
 
-				err = ipPoolManager.UpdateAllocatedIPs(ctx, ipPoolName, []spiderpooltypes.IPAndUID{{IP: ip, UID: string(uuid.NewUUID())}})
+				err = ipPoolManager.UpdateAllocatedIPs(ctx, ipPoolName, "default/pod", []spiderpooltypes.IPAndUID{{IP: ip, UID: string(uuid.NewUUID())}})
 				Expect(err).To(MatchError(constant.ErrUnknown))
 			})
 
@@ -623,10 +623,27 @@ var _ = Describe("IPPoolManager", Label("ippool_manager_test"), func() {
 				err = tracker.Add(ipPoolT)
 				Expect(err).NotTo(HaveOccurred())
 
-				err = ipPoolManager.UpdateAllocatedIPs(ctx, ipPoolName, []spiderpooltypes.IPAndUID{{IP: ip, UID: string(uuid.NewUUID())}})
+				err = ipPoolManager.UpdateAllocatedIPs(ctx, ipPoolName, "default/pod", []spiderpooltypes.IPAndUID{{IP: ip, UID: string(uuid.NewUUID())}})
 				Expect(err).To(MatchError(constant.ErrRetriesExhausted))
 			})
 
+			It("failed to update IPPool due to data broken", func() {
+				patches := gomonkey.ApplyMethodReturn(fakeClient.Status(), "Update", constant.ErrUnknown)
+				defer patches.Reset()
+
+				data, err := convert.MarshalIPPoolAllocatedIPs(records)
+				Expect(err).NotTo(HaveOccurred())
+
+				ipPoolT.Status.AllocatedIPs = data
+				err = fakeClient.Create(ctx, ipPoolT)
+				Expect(err).NotTo(HaveOccurred())
+				err = tracker.Add(ipPoolT)
+				Expect(err).NotTo(HaveOccurred())
+
+				err = ipPoolManager.UpdateAllocatedIPs(ctx, ipPoolName, "default/abc", []spiderpooltypes.IPAndUID{{IP: ip, UID: string(uuid.NewUUID())}})
+				Expect(err).To(HaveOccurred())
+			})
+
 			It("updates the allocated IP record", func() {
 				data, err := convert.MarshalIPPoolAllocatedIPs(records)
 				Expect(err).NotTo(HaveOccurred())
@@ -638,7 +655,7 @@ var _ = Describe("IPPoolManager", Label("ippool_manager_test"), func() {
 				Expect(err).NotTo(HaveOccurred())
 
 				newUID := string(uuid.NewUUID())
-				err = ipPoolManager.UpdateAllocatedIPs(ctx, ipPoolName, []spiderpooltypes.IPAndUID{{IP: ip, UID: newUID}})
+				err = ipPoolManager.UpdateAllocatedIPs(ctx, ipPoolName, "default/pod", []spiderpooltypes.IPAndUID{{IP: ip, UID: newUID}})
 				Expect(err).NotTo(HaveOccurred())
 
 				var ipPool spiderpoolv2beta1.SpiderIPPool