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

fix: statefulset duplicate IP issue

pkg/gcmanager/scanAll_IPPool.go

@@ -11,7 +11,9 @@ import (
 	"time"
 
 	"go.uber.org/zap"
+	appsv1 "k8s.io/api/apps/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/tools/cache"
 
 	"github.com/spidernet-io/spiderpool/pkg/constant"
@@ -190,15 +192,23 @@ func (s *SpiderGC) executeScanAll(ctx context.Context) {
 				} else {
 					// case: The pod in IPPool's ip-allocationDetail is also exist in k8s, but the IPPool IP corresponding allocation pod UID is different with pod UID
 					if string(podYaml.UID) != poolIPAllocation.PodUID {
-						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"),
+							zap.String("podYamlUID", string(podYaml.UID)), zap.String("podYamlNode", podYaml.Spec.NodeName))
+
+						ownerReference := metav1.GetControllerOf(podYaml)
+						if ownerReference != nil {
+							if s.gcConfig.EnableStatefulSet && ownerReference.APIVersion == appsv1.SchemeGroupVersion.String() && ownerReference.Kind == constant.KindStatefulSet {
+								wrappedLog.Sugar().Warnf("StatefulSet Pod %s/%s is restarting", podYaml.Namespace, podYaml.Name)
+								continue
+							}
+						}
+
 						// 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)
 						if err != nil {

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"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -102,7 +103,7 @@ func (i *ipam) retrieveStsIPAllocation(ctx context.Context, nic string, pod *cor
 
 	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")
@@ -123,6 +124,11 @@ func (i *ipam) retrieveStsIPAllocation(ctx context.Context, nic string, pod *cor
 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()
@@ -142,7 +148,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

@@ -33,7 +33,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) (*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 {
@@ -270,7 +270,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
@@ -294,6 +294,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

@@ -475,7 +475,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())
 			})
 
@@ -487,7 +487,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())
 			})
 
@@ -504,7 +504,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))
 			})
 
@@ -521,7 +521,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.ErrRetriesExhausted))
 			})
 
@@ -536,7 +536,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