support DeleteLoadBalancerListeners (batch delete listeners)

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

Author: imroc

internal/controller/clbbinding.go

@@ -340,13 +340,13 @@ func (r *CLBBindingReconciler[T]) ensurePortBound(ctx context.Context, backend c
 		TargetIP:   backend.GetIP(),
 		TargetPort: int64(binding.Port),
 	}
-	targetToDelete := []clb.Target{}
+	targetToDelete := []*clb.Target{}
 	alreadyAdded := false
 	for _, target := range targets {
 		if *target == backendTarget {
 			alreadyAdded = true
 		} else {
-			targetToDelete = append(targetToDelete, *target)
+			targetToDelete = append(targetToDelete, target)
 			log.FromContext(ctx).V(10).Info("remove unexpected target", "got", target, "expect", backendTarget)
 		}
 	}
@@ -521,7 +521,7 @@ func (r *CLBBindingReconciler[T]) cleanup(ctx context.Context, bd T) (result ctr
 	status := bd.GetStatus()
 	for _, binding := range status.PortBindings {
 		// 解绑 lb
-		if _, err := clb.DeleteListenerByPort(ctx, binding.Region, binding.LoadbalancerId, int64(binding.LoadbalancerPort), binding.Protocol); err != nil {
+		if err := clb.DeleteListenerByIdOrPort(ctx, binding.Region, binding.LoadbalancerId, binding.ListenerId, int64(binding.LoadbalancerPort), binding.Protocol); err != nil {
 			e := errors.Cause(err)
 			if clb.IsLbIdNotFoundError(e) { // lb 不存在，忽略
 				continue

pkg/clb/batch-listener.go

@@ -195,3 +195,55 @@ func startDescribeListenerProccessor(concurrent int) {
 		}
 	})
 }
+
+type DeleteListenerTask struct {
+	Ctx        context.Context
+	Region     string
+	LbId       string
+	ListenerId string
+	Result     chan error
+}
+
+func (t *DeleteListenerTask) GetLbId() string {
+	return t.LbId
+}
+
+func (t *DeleteListenerTask) GetRegion() string {
+	return t.Region
+}
+
+var DeleteListenerChan = make(chan *DeleteListenerTask, 100)
+
+func startDeleteListenerProccessor(concurrent int) {
+	apiName := "DeleteLoadBalancerListeners"
+	StartBatchProccessor(concurrent, apiName, DeleteListenerChan, func(region, lbId string, tasks []*DeleteListenerTask) {
+		startTime := time.Now()
+		defer func() {
+			clbLog.V(10).Info(fmt.Sprintf("batch proccess %s performance", apiName), "cost", time.Since(startTime).String())
+		}()
+		req := clb.NewDeleteLoadBalancerListenersRequest()
+		req.LoadBalancerId = &lbId
+		for _, task := range tasks {
+			req.ListenerIds = append(req.ListenerIds, &task.ListenerId)
+		}
+		client := GetClient(region)
+		resp, err := client.DeleteLoadBalancerListeners(req)
+		LogAPI(nil, apiName, req, resp, err)
+		if err != nil {
+			for _, task := range tasks {
+				task.Result <- err
+			}
+			return
+		}
+		_, err = Wait(context.Background(), region, *resp.Response.RequestId, apiName)
+		if err != nil {
+			for _, task := range tasks {
+				task.Result <- err
+			}
+			return
+		}
+		for _, task := range tasks {
+			task.Result <- nil
+		}
+	})
+}

pkg/clb/batch.go

@@ -19,6 +19,7 @@ func init() {
 	go startDescribeListenerProccessor(concurrency)
 	go startDescribeTargetsProccessor(concurrency)
 	go startDeregisterTargetsProccessor(concurrency)
+	go startDeleteListenerProccessor(concurrency)
 }
 
 const (

pkg/clb/listener.go

@@ -154,6 +154,34 @@ func CreateListener(ctx context.Context, region, lbId string, port, endPort int6
 	return
 }
 
+func DeleteListenerByIdOrPort(ctx context.Context, region, lbId, listenerId string, port int64, protocol string) error {
+	if listenerId == "" { // 没有监听器 ID，走慢路径
+		if _, err := DeleteListenerByPort(ctx, region, lbId, port, protocol); err != nil {
+			return errors.WithStack(err)
+		}
+	}
+	if err := DeleteListenerById(ctx, region, lbId, listenerId); err != nil {
+		return errors.WithStack(err)
+	}
+	return nil
+}
+
+func DeleteListenerById(ctx context.Context, region, lbId, listenerId string) error {
+	task := &DeleteListenerTask{
+		Ctx:        ctx,
+		Region:     region,
+		LbId:       lbId,
+		ListenerId: listenerId,
+		Result:     make(chan error),
+	}
+	DeleteListenerChan <- task
+	err := <-task.Result
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	return nil
+}
+
 func DeleteListenerByPort(ctx context.Context, region, lbId string, port int64, protocol string) (id string, err error) {
 	lis, err := GetListenerByPort(ctx, region, lbId, port, protocol)
 	if err != nil {

pkg/clb/target.go

@@ -131,7 +131,20 @@ func DeregisterAllTargets(ctx context.Context, region, lbId, listenerId string)
 	return nil
 }
 
-func DeregisterTargetsForListenerTryBatch(ctx context.Context, region, lbId, listenerId string, targets ...Target) error {
+func DeregisterTargetsForListenerTryBatch(ctx context.Context, region, lbId, listenerId string, targets ...*Target) error {
+	task := &DeregisterTargetsTask{
+		Ctx:        ctx,
+		Region:     region,
+		LbId:       lbId,
+		ListenerId: listenerId,
+		Targets:    targets,
+		Result:     make(chan error),
+	}
+	DeregisterTargetsChan <- task
+	err := <-task.Result
+	if err != nil {
+		return errors.WithStack(err)
+	}
 	return nil
 }