occm: add a new batch pools members update endpoint support

go.mod

@@ -44,6 +44,7 @@ require (
 
 // the below fixes the "go list -m all" execution
 replace (
+	github.com/gophercloud/gophercloud/v2 => github.com/kayrus/gophercloud/v2 v2.9.1-0.20260112125525-25600003eb41
 	k8s.io/cluster-bootstrap => k8s.io/cluster-bootstrap v0.34.2
 	k8s.io/cri-client => k8s.io/cri-client v0.34.2
 	k8s.io/dynamic-resource-allocation => k8s.io/dynamic-resource-allocation v0.34.2

go.sum

@@ -176,8 +176,6 @@ github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/gophercloud/gophercloud/v2 v2.8.0 h1:of2+8tT6+FbEYHfYC8GBu8TXJNsXYSNm9KuvpX7Neqo=
-github.com/gophercloud/gophercloud/v2 v2.8.0/go.mod h1:Ki/ILhYZr/5EPebrPL9Ej+tUg4lqx71/YH2JWVeU+Qk=
 github.com/gophercloud/utils/v2 v2.0.0-20250930154317-576cdf6142a7 h1:y0/17y47lq1dD97qEqbbvjsMF8PzLmqmD839aOePRuc=
 github.com/gophercloud/utils/v2 v2.0.0-20250930154317-576cdf6142a7/go.mod h1:dVCIqYUB0Q8JDbMZaReU6BkAQAS9j3l3Kyc7GuSIztU=
 github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 h1:JeSE6pjso5THxAzdVpqr6/geYxZytqFMBCOtn/ujyeo=
@@ -203,6 +201,8 @@ github.com/joshdk/go-junit v1.0.0 h1:S86cUKIdwBHWwA6xCmFlf3RTLfVXYQfvanM5Uh+K6GE
 github.com/joshdk/go-junit v1.0.0/go.mod h1:TiiV0PqkaNfFXjEiyjWM3XXrhVyCa1K4Zfga6W52ung=
 github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
+github.com/kayrus/gophercloud/v2 v2.9.1-0.20260112125525-25600003eb41 h1:XVoMzOJQNH6tN7hnm48lny1u9A/xzmxm+eBDfckY/kc=
+github.com/kayrus/gophercloud/v2 v2.9.1-0.20260112125525-25600003eb41/go.mod h1:Ki/ILhYZr/5EPebrPL9Ej+tUg4lqx71/YH2JWVeU+Qk=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=

pkg/openstack/loadbalancer.go

@@ -959,14 +959,17 @@ func (lbaas *LbaasV2) ensureOctaviaPool(ctx context.Context, lbID string, name s
 		return nil, err
 	}
 
-	if !curMembers.Equal(newMembers) {
+	if !curMembers.Equal(newMembers) && !lbaas.opts.BatchPoolsMembersUpdate {
 		klog.V(2).Infof("Updating %d members for pool %s", len(members), pool.ID)
 		if err := openstackutil.BatchUpdatePoolMembers(ctx, lbaas.lb, lbID, pool.ID, members); err != nil {
 			return nil, err
 		}
 		klog.V(2).Infof("Successfully updated %d members for pool %s", len(members), pool.ID)
 	}
 
+	// set the updated members to the pool for further batch update
+	pool.Members = batchUpdateMemberOptsToMembers(members)
+
 	return pool, nil
 }
 
@@ -1050,6 +1053,20 @@ func (lbaas *LbaasV2) buildBatchUpdateMemberOpts(ctx context.Context, port corev
 	return members, newMembers, nil
 }
 
+func batchUpdateMemberOptsToMembers(opts []v2pools.BatchUpdateMemberOpts) []v2pools.Member {
+	members := make([]v2pools.Member, len(opts))
+	for i := range opts {
+		members[i] = v2pools.Member{
+			Address:      opts[i].Address,
+			ProtocolPort: opts[i].ProtocolPort,
+			Name:         ptr.Deref(opts[i].Name, ""),
+			SubnetID:     ptr.Deref(opts[i].SubnetID, ""),
+			MonitorPort:  ptr.Deref(opts[i].MonitorPort, 0),
+		}
+	}
+	return members
+}
+
 func (lbaas *LbaasV2) buildCreateMemberOpts(ctx context.Context, port corev1.ServicePort, nodes []*corev1.Node, svcConf *serviceConfig) ([]v2pools.CreateMemberOpts, sets.Set[string], error) {
 	batchUpdateMemberOpts, newMembers, err := lbaas.buildBatchUpdateMemberOpts(ctx, port, nodes, svcConf)
 	if err != nil {
@@ -1783,6 +1800,8 @@ func (lbaas *LbaasV2) ensureOctaviaLoadBalancer(ctx context.Context, clusterName
 			return nil, err
 		}
 
+		// a list of pools to update
+		pools := make([]v2pools.Pool, 0, len(service.Spec.Ports))
 		for portIndex, port := range service.Spec.Ports {
 			listener, err := lbaas.ensureOctaviaListener(ctx, loadbalancer.ID, cpoutil.Sprintf255(listenerFormat, portIndex, lbName), curListenerMapping, port, svcConf)
 			if err != nil {
@@ -1793,6 +1812,7 @@ func (lbaas *LbaasV2) ensureOctaviaLoadBalancer(ctx context.Context, clusterName
 			if err != nil {
 				return nil, err
 			}
+			pools = append(pools, *pool)
 
 			if err := lbaas.ensureOctaviaHealthMonitor(ctx, loadbalancer.ID, cpoutil.Sprintf255(monitorFormat, portIndex, lbName), pool, port, svcConf); err != nil {
 				return nil, err
@@ -1804,6 +1824,16 @@ func (lbaas *LbaasV2) ensureOctaviaLoadBalancer(ctx context.Context, clusterName
 			curListeners = popListener(curListeners, listener.ID)
 		}
 
+		if lbaas.opts.BatchPoolsMembersUpdate {
+			err := openstackutil.BatchUpdatePoolsMembers(ctx, lbaas.lb, loadbalancer.ID, pools)
+			if err != nil {
+				err = PreserveGopherError(err)
+				msg := fmt.Sprintf("Error updating batch pools members for LoadBalancer: %v", err)
+				klog.Errorf(msg, "lbID", loadbalancer.ID)
+				return nil, err
+			}
+		}
+
 		// Deal with the remaining listeners, delete the listener if it was created by this Service previously.
 		if err := lbaas.deleteOctaviaListeners(ctx, loadbalancer.ID, curListeners, isLBOwner, lbName); err != nil {
 			return nil, err
@@ -1935,6 +1965,9 @@ func (lbaas *LbaasV2) updateOctaviaLoadBalancer(ctx context.Context, clusterName
 		lbListeners[key] = l
 	}
 
+	// a list of pools to update
+	pools := make([]v2pools.Pool, 0, len(service.Spec.Ports))
+
 	// Update pool members for each listener.
 	for portIndex, port := range service.Spec.Ports {
 		proto := getListenerProtocol(port.Protocol, svcConf)
@@ -1950,13 +1983,24 @@ func (lbaas *LbaasV2) updateOctaviaLoadBalancer(ctx context.Context, clusterName
 		if err != nil {
 			return err
 		}
+		pools = append(pools, *pool)
 
 		err = lbaas.ensureOctaviaHealthMonitor(ctx, loadbalancer.ID, cpoutil.Sprintf255(monitorFormat, portIndex, loadbalancer.Name), pool, port, svcConf)
 		if err != nil {
 			return err
 		}
 	}
 
+	if lbaas.opts.BatchPoolsMembersUpdate {
+		err := openstackutil.BatchUpdatePoolsMembers(ctx, lbaas.lb, loadbalancer.ID, pools)
+		if err != nil {
+			err = PreserveGopherError(err)
+			msg := fmt.Sprintf("Error updating batch pools members for LoadBalancer: %v", err)
+			klog.Errorf(msg, "lbID", loadbalancer.ID)
+			return err
+		}
+	}
+
 	if lbaas.opts.ManageSecurityGroups {
 		err := lbaas.ensureAndUpdateOctaviaSecurityGroup(ctx, clusterName, service, filteredNodes, svcConf)
 		if err != nil {

pkg/openstack/openstack.go

@@ -120,6 +120,7 @@ type LoadBalancerOpts struct {
 	MaxSharedLB                    int                 `gcfg:"max-shared-lb"`                      //  Number of Services in maximum can share a single load balancer. Default 2
 	ContainerStore                 string              `gcfg:"container-store"`                    // Used to specify the store of the tls-container-ref
 	ProviderRequiresSerialAPICalls bool                `gcfg:"provider-requires-serial-api-calls"` // default false, the provider supports the "bulk update" API call
+	BatchPoolsMembersUpdate        bool                `gcfg:"batch-pools-members-update"`         // default false, the controller will update all pools members in batch when possible
 	// revive:disable:var-naming
 	TlsContainerRef string `gcfg:"default-tls-container-ref"` //  reference to a tls container
 	// revive:enable:var-naming

pkg/util/openstack/loadbalancer.go

@@ -603,6 +603,47 @@ func BatchUpdatePoolMembers(ctx context.Context, client *gophercloud.ServiceClie
 	return nil
 }
 
+// BatchUpdatePoolsMembers
+func BatchUpdatePoolsMembers(ctx context.Context, client *gophercloud.ServiceClient, lbID string, newPools []pools.Pool) error {
+	mc := metrics.NewMetricContext("loadbalancer_pools_members", "update")
+
+	opts := make([]pools.BatchUpdateMemberOpts, 0, len(newPools)*len(newPools[0].Members))
+	for _, p := range newPools {
+		for _, m := range p.Members {
+			var name, subnetID *string
+			if m.Name != "" {
+				name = &m.Name
+			}
+			if m.SubnetID != "" {
+				subnetID = &m.SubnetID
+			}
+			var monitorPort *int
+			if m.MonitorPort > 0 {
+				monitorPort = &m.MonitorPort
+			}
+			opts = append(opts, pools.BatchUpdateMemberOpts{
+				PoolID:       p.ID,
+				Address:      m.Address,
+				ProtocolPort: m.ProtocolPort,
+				Name:         name,
+				SubnetID:     subnetID,
+				MonitorPort:  monitorPort,
+			})
+		}
+	}
+
+	err := pools.BatchUpdatePoolsMembers(ctx, client, opts).ExtractErr()
+	if mc.ObserveRequest(err) != nil {
+		return err
+	}
+
+	if _, err := WaitActiveAndGetLoadBalancer(ctx, client, lbID); err != nil {
+		return fmt.Errorf("failed to wait for load balancer %s ACTIVE after updating pools members: %v", lbID, err)
+	}
+
+	return nil
+}
+
 // GetL7policies retrieves all l7 policies for the given listener.
 func GetL7policies(ctx context.Context, client *gophercloud.ServiceClient, listenerID string) ([]l7policies.L7Policy, error) {
 	var policies []l7policies.L7Policy