remove lb from port pool status if lb is existed lb and been remove

manually from existedLbIds

Signed-off-by: roc <[email protected]>

Author: imroc

internal/controller/clbbinding.go

@@ -237,24 +237,37 @@ func (r *CLBBindingReconciler[T]) createListener(ctx context.Context, bd clbbind
 // 对账单个监听器
 func (r *CLBBindingReconciler[T]) ensureListener(ctx context.Context, bd clbbinding.CLBBinding, binding *networkingv1alpha1.PortBindingStatus) (*networkingv1alpha1.PortBindingStatus, error) {
 	log.FromContext(ctx).V(10).Info("ensureListener", "binding", binding)
+	// 如果 lb 已被移除，且当前还未绑定成功，则移除该端口绑定，等待重新分配端口（配错 lb导致一直无法绑定成功，更正后，可以触发重新分配以便能够成功绑定）
+	if bd.GetStatus().State != networkingv1alpha1.CLBBindingStateBound && !portpool.Allocator.IsLbExists(binding.Pool, portpool.NewLBKeyFromBinding(binding)) {
+		r.Recorder.Eventf(bd.GetObject(), corev1.EventTypeNormal, "PortBindingRemoved", "lb %q not exists, remove port binding (lbPort:%s protocol:%s)", binding.LoadbalancerId, binding.LoadbalancerPort, binding.Protocol)
+		return nil, nil
+	}
+
+	removeReason := ""
+	removeMsg := ""
+
 	// 确保关联的 pool 无误
 	pool := portpool.Allocator.GetPool(binding.Pool)
-	if pool == nil { // 端口池不存在，移除 binding 并并记录事件
-		r.Recorder.Event(bd.GetObject(), corev1.EventTypeWarning, "PortPoolDeleted", fmt.Sprintf("port pool has been deleted (%s/%s/%d)", binding.Pool, binding.LoadbalancerId, binding.LoadbalancerPort))
+	if pool == nil {
+		// 如果端口池被删，且当前还未绑定成功，则移除该端口绑定
+		if bd.GetStatus().State != networkingv1alpha1.CLBBindingStateBound {
+			removeReason = "PortPoolDeleted"
+			removeMsg = "port pool has been deleted"
+		}
+	} else {
+		// 如果 lb 已被移除，且当前还未绑定成功，则移除该端口绑定。等待重新分配端口（配错 lb导致一直无法绑定成功，更正后，可以触发重新分配以便能够成功绑定）
+		if bd.GetStatus().State != networkingv1alpha1.CLBBindingStateBound && !pool.IsLbExists(portpool.NewLBKeyFromBinding(binding)) {
+			removeReason = "CLBDeleted"
+			removeMsg = "clb has been removed"
+		}
+	}
+	if removeMsg != "" {
+		r.Recorder.Eventf(bd.GetObject(), corev1.EventTypeWarning, removeReason, "%s (%s/%s/%d/%s)", removeMsg, binding.Pool, binding.LoadbalancerId, binding.LoadbalancerPort, binding.Protocol)
 		// 确保监听器被清理
 		if err := r.cleanupPortBinding(ctx, binding); err != nil {
 			return binding, errors.WithStack(err)
 		}
 		return nil, nil
-	} else {
-		if !pool.IsLbExists(portpool.NewLBKeyFromBinding(binding)) { // lb 被删除，移除 binding 并记录到事件
-			r.Recorder.Event(bd.GetObject(), corev1.EventTypeWarning, "CLBDeleted", fmt.Sprintf("clb has been deleted (%s/%s/%d)", binding.Pool, binding.LoadbalancerId, binding.LoadbalancerPort))
-			// 确保监听器被清理
-			if err := r.cleanupPortBinding(ctx, binding); err != nil {
-				return binding, errors.WithStack(err)
-			}
-			return nil, nil
-		}
 	}
 
 	// 没有监听器 ID，尝试直接新建（不调查接口，加速大规模场景扩容速度）

internal/controller/clbportpool_controller.go

@@ -171,6 +171,11 @@ func (r *CLBPortPoolReconciler) ensureLbStatus(ctx context.Context, pool *networ
 	insufficientPorts := true
 	autoCreatedLbNum := uint16(0)
 
+	existedLbIds := make(map[string]struct{})
+	for _, lbId := range pool.Spec.ExsistedLoadBalancerIDs {
+		existedLbIds[lbId] = struct{}{}
+	}
+
 	// 构造当前 lb status 列表
 	for _, lbStatus := range pool.Status.LoadbalancerStatuses {
 		lbId := lbStatus.LoadbalancerID
@@ -187,6 +192,13 @@ func (r *CLBPortPoolReconciler) ensureLbStatus(ctx context.Context, pool *networ
 			}
 			if util.GetValue(lbStatus.AutoCreated) {
 				autoCreatedLbNum++
+			} else { // 已有的 lb，如果被手动移除，可能是 lb 有误（比如错误的 vpc），也将其从status和分配器中移除
+				if _, exists := existedLbIds[lbId]; !exists {
+					if portpool.Allocator.RemoveLB(pool.Name, portpool.NewLBKey(lbId, pool.GetRegion())) {
+						r.Recorder.Eventf(pool, corev1.EventTypeNormal, "RemoveLoadBalancer", "remove existed clb %s from pool", lbId)
+					}
+					continue
+				}
 			}
 			allocatableLBs = append(allocatableLBs, lbKey)
 		} else {

internal/portpool/allocator.go

@@ -117,6 +117,20 @@ func (pa *PortAllocator) Release(pool string, lbKey LBKey, port ProtocolPort) bo
 	return false
 }
 
+func (pa *PortAllocator) IsLbExists(pool string, lbKey LBKey) bool {
+	if pp := pa.GetPool(pool); pp != nil {
+		return pp.IsLbExists(lbKey)
+	}
+	return false
+}
+
+func (pa *PortAllocator) RemoveLB(pool string, lbKey LBKey) bool {
+	if pp := pa.GetPool(pool); pp != nil {
+		return pp.RemoveLB(lbKey)
+	}
+	return false
+}
+
 var Allocator = NewPortAllocator()
 
 func (pa *PortAllocator) MarkAllocated(poolName string, lbKey LBKey, port uint16, endPort *uint16, protocol string) {

internal/portpool/portpool.go

@@ -153,6 +153,17 @@ func (pp *PortPool) AllocatePort(ctx context.Context, quota int64, ports ...Prot
 	return nil, quotaExceeded
 }
 
+func (pp *PortPool) RemoveLB(lbKey LBKey) bool {
+	pp.mu.Lock()
+	defer pp.mu.Unlock()
+	_, exists := pp.cache[lbKey]
+	if !exists {
+		return false
+	}
+	delete(pp.cache, lbKey)
+	return true
+}
+
 // 释放已分配的端口
 func (pp *PortPool) ReleasePort(lbKey LBKey, port ProtocolPort) bool {
 	pp.mu.Lock()

pkg/util/retry.go

@@ -4,16 +4,8 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/util/retry"
-	ctrl "sigs.k8s.io/controller-runtime"
 )
 
-func RequeueIfConflict(err error) (ctrl.Result, error) {
-	if apierrors.IsConflict(err) {
-		return ctrl.Result{Requeue: true}, nil
-	}
-	return ctrl.Result{}, err
-}
-
 func RetryIfPossible(fn func() error) error {
 	return RetryOnErrors(
 		retry.DefaultBackoff,