This is an automated email from the ASF dual-hosted git repository.
vishesh92 pushed a commit to branch main
in repository
https://gitbox.apache.org/repos/asf/cloudstack-kubernetes-provider.git
The following commit(s) were added to refs/heads/main by this push:
new 4170ad5c Add pagination to cloudstack api calls (#102)
4170ad5c is described below
commit 4170ad5cc1ead23532acb88b0c5a74c8c56c583d
Author: Vishesh <[email protected]>
AuthorDate: Mon Sep 21 17:50:17 2026 +0530
Add pagination to cloudstack api calls (#102)
---
cloudstack_loadbalancer.go | 146 +++++++++++-----
cloudstack_loadbalancer_test.go | 356 ++++++++++++++++++++++++++++++++++++++++
pagination.go | 118 +++++++++++++
pagination_test.go | 305 ++++++++++++++++++++++++++++++++++
4 files changed, 888 insertions(+), 37 deletions(-)
diff --git a/cloudstack_loadbalancer.go b/cloudstack_loadbalancer.go
index 69d16f51..899d3186 100644
--- a/cloudstack_loadbalancer.go
+++ b/cloudstack_loadbalancer.go
@@ -281,13 +281,21 @@ func (cs *CSCloud) UpdateLoadBalancer(ctx
context.Context, clusterName string, s
for _, lbRule := range lb.rules {
p :=
lb.LoadBalancer.NewListLoadBalancerRuleInstancesParams(lbRule.Id)
- // Retrieve all VMs currently associated to this load balancer
rule.
- l, err := lb.LoadBalancer.ListLoadBalancerRuleInstances(p)
+ // Retrieve all VMs currently associated to this load balancer
rule. There
+ // is one per load balanced node, so this grows with the
cluster.
+ instances, err := listAll(p, func() (int,
[]*cloudstack.VirtualMachine, error) {
+ l, err :=
lb.LoadBalancer.ListLoadBalancerRuleInstances(p)
+ if err != nil {
+ return 0, nil, err
+ }
+
+ return l.Count, l.LoadBalancerRuleInstances, nil
+ })
if err != nil {
return fmt.Errorf("error retrieving associated
instances: %v", err)
}
- assign, remove := symmetricDifference(lb.hostIDs,
l.LoadBalancerRuleInstances)
+ assign, remove := symmetricDifference(lb.hostIDs, instances)
if len(assign) > 0 {
klog.V(4).Infof("Assigning new hosts (%v) to load
balancer rule: %v", assign, lbRule.Name)
@@ -460,11 +468,22 @@ func (cs *CSCloud) getLoadBalancer(service
*corev1.Service) (*loadBalancer, erro
p.SetProjectid(cs.projectID)
}
- l, err := cs.client.LoadBalancer.ListLoadBalancerRules(p)
+ // The keyword is matched as a substring server side, so this can
return more
+ // rules than just this service's and has to be paged through.
+ lbRules, err := listAll(p, func() (int, []*cloudstack.LoadBalancerRule,
error) {
+ l, err := cs.client.LoadBalancer.ListLoadBalancerRules(p)
+ if err != nil {
+ return 0, nil, err
+ }
+
+ return l.Count, l.LoadBalancerRules, nil
+ })
if err != nil {
return nil, fmt.Errorf("error retrieving load balancer rules:
%v", err)
}
+ lbRules = dedupeByID(lbRules, func(rule *cloudstack.LoadBalancerRule)
string { return rule.Id })
+
// Keeping the rule on the address the Service is already published on
stops a
// duplicate sweep from deleting the rule that clients and DNS are
pointing at.
preferredIP := service.Spec.LoadBalancerIP
@@ -472,7 +491,7 @@ func (cs *CSCloud) getLoadBalancer(service *corev1.Service)
(*loadBalancer, erro
preferredIP = service.Status.LoadBalancer.Ingress[0].IP
}
- for _, lbRule := range l.LoadBalancerRules {
+ for _, lbRule := range lbRules {
if existing, seen := lb.rules[lbRule.Name]; seen {
duplicate := lbRule
if lbRule.Publicip == preferredIP && existing.Publicip
!= preferredIP {
@@ -539,24 +558,35 @@ func (cs *CSCloud) verifyHosts(nodes []*corev1.Node)
([]string, string, error) {
p.SetProjectid(cs.projectID)
}
- l, err := cs.client.VirtualMachine.ListVirtualMachines(p)
+ vms, err := listAll(p, func() (int, []*cloudstack.VirtualMachine,
error) {
+ l, err := cs.client.VirtualMachine.ListVirtualMachines(p)
+ if err != nil {
+ return 0, nil, err
+ }
+
+ return l.Count, l.VirtualMachines, nil
+ })
if err != nil {
return nil, "", fmt.Errorf("error retrieving list of hosts:
%v", err)
}
var hostIDs []string
var networkID string
+ seen := map[string]bool{} // used to check whether the changing set of
VMs contains one we had already seen in another page.
// Check if the virtual machine is in the hosts slice, then add the
corresponding ID.
- for _, vm := range l.VirtualMachines {
- if hostNames[strings.ToLower(vm.Name)] {
- if networkID != "" && networkID != vm.Nic[0].Networkid {
- return nil, "", fmt.Errorf("found hosts that
belong to different networks")
- }
+ for _, vm := range vms {
+ if !hostNames[strings.ToLower(vm.Name)] || seen[vm.Id] {
+ continue
+ }
+ seen[vm.Id] = true
- networkID = vm.Nic[0].Networkid
- hostIDs = append(hostIDs, vm.Id)
+ if networkID != "" && networkID != vm.Nic[0].Networkid {
+ return nil, "", fmt.Errorf("found hosts that belong to
different networks")
}
+
+ networkID = vm.Nic[0].Networkid
+ hostIDs = append(hostIDs, vm.Id)
}
if len(hostIDs) == 0 || len(networkID) == 0 {
@@ -903,8 +933,19 @@ func symmetricDifference(hostIDs []string, lbInstances
[]*cloudstack.VirtualMach
new[hostID] = true
}
+ // Paging over the instances of a rule can return the same instance
twice. A
+ // duplicate would otherwise be dropped from new on its first
occurrence and
+ // then added to remove on its second, so the same host would be both
kept
+ // and removed.
+ seen := make(map[string]bool)
+
var remove []string
for _, instance := range lbInstances {
+ if seen[instance.Id] {
+ continue
+ }
+ seen[instance.Id] = true
+
if new[instance.Id] {
delete(new, instance.Id)
continue
@@ -996,6 +1037,37 @@ func rulesMapToString(rules
map[*cloudstack.FirewallRule]bool) string {
return ls.String()
}
+// listFirewallRules retrieves all firewall rules associated with a public IP.
+//
+// Rules are deduplicated by ID: paging over a set that is changing underneath
us
+// can return the same rule on more than one page, and each page decodes into
its
+// own struct, so a repeat arrives as a second pointer to an equal rule.
Callers
+// key their bookkeeping on the pointer, which would treat the two as unrelated
+// rules and delete one of them.
+func (lb *loadBalancer) listFirewallRules(publicIpId string)
([]*cloudstack.FirewallRule, error) {
+ p := lb.Firewall.NewListFirewallRulesParams()
+ p.SetIpaddressid(publicIpId)
+ p.SetListall(true)
+ if lb.projectID != "" {
+ p.SetProjectid(lb.projectID)
+ }
+
+ klog.V(4).Infof("Listing firewall rules for %v", p)
+ rules, err := listAll(p, func() (int, []*cloudstack.FirewallRule,
error) {
+ r, err := lb.Firewall.ListFirewallRules(p)
+ if err != nil {
+ return 0, nil, err
+ }
+
+ return r.Count, r.FirewallRules, nil
+ })
+ if err != nil {
+ return nil, fmt.Errorf("error fetching firewall rules for
public IP %v: %v", publicIpId, err)
+ }
+
+ return dedupeByID(rules, func(rule *cloudstack.FirewallRule) string {
return rule.Id }), nil
+}
+
// updateFirewallRule creates a firewall rule for a load balancer rule
//
// If the rule list is empty, all internet (IPv4: 0.0.0.0/0) is opened for the
@@ -1007,23 +1079,16 @@ func (lb *loadBalancer) updateFirewallRule(publicIpId
string, publicPort int, pr
allowedIPs = []string{defaultAllowedCIDR}
}
- p := lb.Firewall.NewListFirewallRulesParams()
- p.SetIpaddressid(publicIpId)
- p.SetListall(true)
- if lb.projectID != "" {
- p.SetProjectid(lb.projectID)
- }
- klog.V(4).Infof("Listing firewall rules for %v", p)
- r, err := lb.Firewall.ListFirewallRules(p)
+ firewallRules, err := lb.listFirewallRules(publicIpId)
if err != nil {
- return false, fmt.Errorf("error fetching firewall rules for
public IP %v: %v", publicIpId, err)
+ return false, err
}
- klog.V(4).Infof("All firewall rules for %v: %v", lb.ipAddr,
rulesToString(r.FirewallRules))
+ klog.V(4).Infof("All firewall rules for %v: %v", lb.ipAddr,
rulesToString(firewallRules))
// find all rules that have a matching proto+port
// a map may or may not be faster, but is a bit easier to understand
filtered := make(map[*cloudstack.FirewallRule]bool)
- for _, rule := range r.FirewallRules {
+ for _, rule := range firewallRules {
if rule.Protocol == protocol.IPProtocol() && rule.Startport ==
publicPort && rule.Endport == publicPort {
filtered[rule] = true
}
@@ -1104,8 +1169,14 @@ func (lb *loadBalancer) updateNetworkACL(publicPort int,
protocol LoadBalancerPr
networkAclParams.SetProjectid(lb.projectID)
}
- networkAclResponse, err :=
lb.NetworkACL.ListNetworkACLs(networkAclParams)
+ networkAcls, err := listAll(networkAclParams, func() (int,
[]*cloudstack.NetworkACL, error) {
+ networkAclResponse, err :=
lb.NetworkACL.ListNetworkACLs(networkAclParams)
+ if err != nil {
+ return 0, nil, err
+ }
+ return networkAclResponse.Count,
networkAclResponse.NetworkACLs, nil
+ })
if err != nil {
return false, fmt.Errorf("error fetching Network ACL with ID:
%v for network with id: %v, due to: %s", network.Aclid, networkId, err)
}
@@ -1113,7 +1184,7 @@ func (lb *loadBalancer) updateNetworkACL(publicPort int,
protocol LoadBalancerPr
// find all network ACL rules that have a matching proto+port
// a map may or may not be faster, but is a bit easier to understand
filtered := make(map[*cloudstack.NetworkACL]bool)
- for _, netAclRule := range networkAclResponse.NetworkACLs {
+ for _, netAclRule := range networkAcls {
if netAclRule.Protocol == protocol.IPProtocol() &&
netAclRule.Startport == strconv.Itoa(publicPort) && netAclRule.Endport ==
strconv.Itoa(publicPort) {
filtered[netAclRule] = true
}
@@ -1145,20 +1216,14 @@ func (lb *loadBalancer) updateNetworkACL(publicPort
int, protocol LoadBalancerPr
//
// returns true when corresponding rules were deleted
func (lb *loadBalancer) deleteFirewallRule(publicIpId string, publicPort int,
protocol LoadBalancerProtocol) (bool, error) {
- p := lb.Firewall.NewListFirewallRulesParams()
- p.SetIpaddressid(publicIpId)
- p.SetListall(true)
- if lb.projectID != "" {
- p.SetProjectid(lb.projectID)
- }
- r, err := lb.Firewall.ListFirewallRules(p)
+ firewallRules, err := lb.listFirewallRules(publicIpId)
if err != nil {
- return false, fmt.Errorf("error fetching firewall rules for
public IP %v: %v", publicIpId, err)
+ return false, err
}
// filter by proto:port
filtered := make([]*cloudstack.FirewallRule, 0, 1)
- for _, rule := range r.FirewallRules {
+ for _, rule := range firewallRules {
if rule.Protocol == protocol.IPProtocol() && rule.Startport ==
publicPort && rule.Endport == publicPort {
filtered = append(filtered, rule)
}
@@ -1188,14 +1253,21 @@ func (lb *loadBalancer) deleteNetworkACLRule(publicPort
int, protocol LoadBalanc
p.SetProjectid(lb.projectID)
}
- r, err := lb.NetworkACL.ListNetworkACLs(p)
+ networkAcls, err := listAll(p, func() (int, []*cloudstack.NetworkACL,
error) {
+ r, err := lb.NetworkACL.ListNetworkACLs(p)
+ if err != nil {
+ return 0, nil, err
+ }
+
+ return r.Count, r.NetworkACLs, nil
+ })
if err != nil {
return false, fmt.Errorf("error fetching Network ACL rules
Network ID %v: %v", networkID, err)
}
// filter by proto:port
filtered := make([]*cloudstack.NetworkACL, 0, 1)
- for _, rule := range r.NetworkACLs {
+ for _, rule := range networkAcls {
if rule.Protocol == protocol.IPProtocol() && rule.Startport ==
strconv.Itoa(publicPort) && rule.Endport == strconv.Itoa(publicPort) {
filtered = append(filtered, rule)
}
diff --git a/cloudstack_loadbalancer_test.go b/cloudstack_loadbalancer_test.go
index 39b229d5..933f7685 100644
--- a/cloudstack_loadbalancer_test.go
+++ b/cloudstack_loadbalancer_test.go
@@ -23,6 +23,7 @@ import (
"context"
"fmt"
"reflect"
+ "slices"
"sort"
"strings"
"testing"
@@ -182,6 +183,30 @@ func TestSymmetricDifference(t *testing.T) {
wantAssign: []string{"host3"},
wantRemove: []string{"host2"},
},
+ {
+ // Paging over a changing result set can return the
same instance on
+ // two pages. A wanted host must not end up in remove
because of it.
+ name: "duplicate instance of a wanted host",
+ hostIDs: []string{"host1", "host2"},
+ lbInstances: []*cloudstack.VirtualMachine{
+ {Id: "host1"},
+ {Id: "host2"},
+ {Id: "host1"},
+ },
+ wantAssign: nil,
+ wantRemove: nil,
+ },
+ {
+ name: "duplicate instance of an unwanted host",
+ hostIDs: []string{"host1"},
+ lbInstances: []*cloudstack.VirtualMachine{
+ {Id: "host1"},
+ {Id: "host2"},
+ {Id: "host2"},
+ },
+ wantAssign: nil,
+ wantRemove: []string{"host2"},
+ },
{
name: "add one host",
hostIDs: []string{"host1", "host2", "host3"},
@@ -2651,6 +2676,74 @@ func TestUpdateFirewallRule(t *testing.T) {
})
}
+func TestListFirewallRulesDeduplicates(t *testing.T) {
+ // Each page decodes into its own structs, so a rule returned on two
pages
+ // arrives as two pointers to an equal rule. updateFirewallRule keys its
+ // bookkeeping on the pointer: it would keep one copy as the CIDR match
and
+ // delete the other by ID, removing the very rule it had just matched
and
+ // creating nothing in its place.
+ ctrl := gomock.NewController(t)
+ t.Cleanup(ctrl.Finish)
+
+ rule := func() *cloudstack.FirewallRule {
+ return &cloudstack.FirewallRule{
+ Id: "fw-keep", Protocol: "tcp", Startport: 80, Endport:
80,
+ Cidrlist: defaultAllowedCIDR,
+ }
+ }
+
+ mockFirewall := cloudstack.NewMockFirewallServiceIface(ctrl)
+ mockFirewall.EXPECT().NewListFirewallRulesParams().
+ Return(&cloudstack.ListFirewallRulesParams{})
+
+ // Count of 2 with one rule per page makes listAll page, and the same
rule
+ // comes back both times. It must be built per call: a real second page
is
+ // decoded into its own struct, so the repeat is a distinct pointer.
+ mockFirewall.EXPECT().ListFirewallRules(gomock.Any()).Times(2).
+ DoAndReturn(func(p *cloudstack.ListFirewallRulesParams)
(*cloudstack.ListFirewallRulesResponse, error) {
+ return &cloudstack.ListFirewallRulesResponse{
+ Count: 2,
+ FirewallRules:
[]*cloudstack.FirewallRule{rule()},
+ }, nil
+ })
+
+ var deleted []string
+
mockFirewall.EXPECT().NewDeleteFirewallRuleParams(gomock.Any()).AnyTimes().
+ DoAndReturn(func(id string)
*cloudstack.DeleteFirewallRuleParams {
+ deleted = append(deleted, id)
+ return &cloudstack.DeleteFirewallRuleParams{}
+ })
+ mockFirewall.EXPECT().DeleteFirewallRule(gomock.Any()).AnyTimes().
+ Return(&cloudstack.DeleteFirewallRuleResponse{}, nil)
+
+ created := 0
+ mockFirewall.EXPECT().NewCreateFirewallRuleParams(gomock.Any(),
gomock.Any()).AnyTimes().
+ DoAndReturn(func(ip, proto string)
*cloudstack.CreateFirewallRuleParams {
+ created++
+ return &cloudstack.CreateFirewallRuleParams{}
+ })
+ mockFirewall.EXPECT().CreateFirewallRule(gomock.Any()).AnyTimes().
+ Return(&cloudstack.CreateFirewallRuleResponse{}, nil)
+
+ lb := &loadBalancer{
+ CloudStackClient: &cloudstack.CloudStackClient{Firewall:
mockFirewall},
+ ipAddr: "203.0.113.1",
+ }
+
+ // The existing rule already allows exactly what is wanted, so nothing
should
+ // be deleted and nothing created.
+ if _, err := lb.updateFirewallRule("ip-123", 80,
LoadBalancerProtocolTCP, []string{defaultAllowedCIDR}); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ if len(deleted) > 0 {
+ t.Errorf("deleted %v, but that rule already matched the wanted
CIDR", deleted)
+ }
+ if created > 0 {
+ t.Errorf("created %d replacement rules, want 0", created)
+ }
+}
+
func TestDeleteFirewallRule(t *testing.T) {
t.Run("delete matching rule", func(t *testing.T) {
ctrl := gomock.NewController(t)
@@ -3452,6 +3545,49 @@ func TestGetLoadBalancer(t *testing.T) {
})
}
+func TestGetLoadBalancerDeduplicatesPagedRules(t *testing.T) {
+ // A rule returned on two pages arrives as two pointers to an equal
rule.
+ // getLoadBalancer would see the name already in lb.rules, call it a
+ // duplicate, and hand it to the sweep in EnsureLoadBalancer - deleting
the
+ // only rule the Service is published on. The IP preference cannot save
it,
+ // because both copies carry the same address.
+ ctrl := gomock.NewController(t)
+ t.Cleanup(ctrl.Finish)
+
+ mockLB := cloudstack.NewMockLoadBalancerServiceIface(ctrl)
+ mockLB.EXPECT().NewListLoadBalancerRulesParams().
+ Return(&cloudstack.ListLoadBalancerRulesParams{})
+ mockLB.EXPECT().ListLoadBalancerRules(gomock.Any()).Times(2).
+ DoAndReturn(func(p *cloudstack.ListLoadBalancerRulesParams)
(*cloudstack.ListLoadBalancerRulesResponse, error) {
+ return &cloudstack.ListLoadBalancerRulesResponse{
+ Count: 2,
+ LoadBalancerRules:
[]*cloudstack.LoadBalancerRule{
+ {Id: "rule-1", Name: "a-svc-TCP-80",
Publicip: "1.2.3.4", Publicipid: "ip-1"},
+ },
+ }, nil
+ })
+
+ cs := &CSCloud{client: &cloudstack.CloudStackClient{LoadBalancer:
mockLB}}
+ service := &corev1.Service{
+ ObjectMeta: metav1.ObjectMeta{Name: "svc", Namespace:
"default", UID: "abc123"},
+ }
+
+ lb, err := cs.getLoadBalancer(service)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ if len(lb.duplicateRules) != 0 {
+ t.Errorf("rule %v was repeated across pages, not duplicated; it
must not be swept", lb.duplicateRules)
+ }
+ if len(lb.rules) != 1 {
+ t.Errorf("rules = %d, want 1", len(lb.rules))
+ }
+ if lb.ipAddr != "1.2.3.4" {
+ t.Errorf("ipAddr = %q, want %q", lb.ipAddr, "1.2.3.4")
+ }
+}
+
// A failed sweep is reported so the service controller retries, but only after
// this service's own rules and IP are gone, so the retry sees the leftover
// duplicate as an ordinary rule instead of blocking deletion for ever.
@@ -3992,3 +4128,223 @@ func TestVerifyHosts(t *testing.T) {
}
})
}
+
+// pagedRequest is the read side of the paging parameters that every
+// cloudstack-go List*Params exposes.
+type pagedRequest interface {
+ GetPage() (int, bool)
+ GetPagesize() (int, bool)
+}
+
+// pageOf returns the window of items a request asks for, mirroring how
+// CloudStack serves a list: a request carrying no paging parameters comes back
+// truncated at pageSize, and later pages are served by offset. It also asserts
+// the paging contract CloudStack enforces.
+func pageOf[T any](t *testing.T, p pagedRequest, items []T, pageSize int) []T {
+ t.Helper()
+
+ page, paged := p.GetPage()
+ size, sized := p.GetPagesize()
+
+ if paged != sized {
+ t.Errorf("page and pagesize must be sent together, got page set
= %v, pagesize set = %v", paged, sized)
+ }
+
+ switch {
+ case !paged:
+ page, size = 1, pageSize
+ case page < 2:
+ t.Errorf("page = %d, want >= 2 (CloudStack rejects page 0)",
page)
+ case size != pageSize:
+ t.Errorf("pagesize = %d, want %d", size, pageSize)
+ }
+
+ start := (page - 1) * size
+ if start >= len(items) {
+ return nil
+ }
+
+ end := start + size
+ if end > len(items) {
+ end = len(items)
+ }
+
+ return items[start:end]
+}
+
+// nodesNamed builds the node list a cloudprovider call receives.
+func nodesNamed(names ...string) []*corev1.Node {
+ nodes := make([]*corev1.Node, 0, len(names))
+ for _, name := range names {
+ nodes = append(nodes, &corev1.Node{ObjectMeta:
metav1.ObjectMeta{Name: name}})
+ }
+ return nodes
+}
+
+func TestVerifyHostsPagination(t *testing.T) {
+ // CloudStack truncates list responses at default.page.size while still
+ // reporting the full total in count. This is the regression from issue
#99:
+ // with more VMs in the account than fit in one page, the nodes beyond
the
+ // first page were invisible and the load balancer was never created.
+ const pageSize = 500
+
+ // Only the last VM is a cluster node, so it lands on the final page.
+ vms := make([]*cloudstack.VirtualMachine, 750)
+ for i := range vms {
+ vms[i] = &cloudstack.VirtualMachine{
+ Id: fmt.Sprintf("vm-%d", i),
+ Name: fmt.Sprintf("other-%d", i),
+ Nic: []cloudstack.Nic{{Networkid: "net-123"}},
+ }
+ }
+ vms[len(vms)-1].Id = "vm-node-1"
+ vms[len(vms)-1].Name = "node-1"
+
+ t.Run("collects hosts from every page", func(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ t.Cleanup(ctrl.Finish)
+
+ mockVM := cloudstack.NewMockVirtualMachineServiceIface(ctrl)
+ // Params are built once and reused across pages.
+ mockVM.EXPECT().NewListVirtualMachinesParams().
+ Return(&cloudstack.ListVirtualMachinesParams{})
+ mockVM.EXPECT().ListVirtualMachines(gomock.Any()).Times(2).
+ DoAndReturn(func(p
*cloudstack.ListVirtualMachinesParams)
(*cloudstack.ListVirtualMachinesResponse, error) {
+ return &cloudstack.ListVirtualMachinesResponse{
+ Count: len(vms),
+ VirtualMachines: pageOf(t, p, vms,
pageSize),
+ }, nil
+ })
+
+ cs := &CSCloud{client:
&cloudstack.CloudStackClient{VirtualMachine: mockVM}}
+
+ hostIDs, networkID, err := cs.verifyHosts(nodesNamed("node-1"))
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if !reflect.DeepEqual(hostIDs, []string{"vm-node-1"}) {
+ t.Errorf("hostIDs = %v, want %v", hostIDs,
[]string{"vm-node-1"})
+ }
+ if networkID != "net-123" {
+ t.Errorf("networkID = %q, want %q", networkID,
"net-123")
+ }
+ })
+
+ t.Run("deduplicates hosts repeated across pages", func(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ t.Cleanup(ctrl.Finish)
+
+ node := &cloudstack.VirtualMachine{
+ Id: "vm-node-1",
+ Name: "node-1",
+ Nic: []cloudstack.Nic{{Networkid: "net-123"}},
+ }
+
+ // A VM is removed between the requests, shifting the offset so
the node
+ // comes back on both pages.
+ pages := [][]*cloudstack.VirtualMachine{{node, node}, {node}}
+
+ mockVM := cloudstack.NewMockVirtualMachineServiceIface(ctrl)
+ mockVM.EXPECT().NewListVirtualMachinesParams().
+ Return(&cloudstack.ListVirtualMachinesParams{})
+
mockVM.EXPECT().ListVirtualMachines(gomock.Any()).Times(len(pages)).
+ DoAndReturn(func(p
*cloudstack.ListVirtualMachinesParams)
(*cloudstack.ListVirtualMachinesResponse, error) {
+ page := pages[0]
+ pages = pages[1:]
+ return
&cloudstack.ListVirtualMachinesResponse{Count: 3, VirtualMachines: page}, nil
+ })
+
+ cs := &CSCloud{client:
&cloudstack.CloudStackClient{VirtualMachine: mockVM}}
+
+ hostIDs, _, err := cs.verifyHosts(nodesNamed("node-1"))
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if !reflect.DeepEqual(hostIDs, []string{"vm-node-1"}) {
+ t.Errorf("hostIDs = %v, want %v", hostIDs,
[]string{"vm-node-1"})
+ }
+ })
+}
+
+func TestUpdateLoadBalancerPagination(t *testing.T) {
+ // Instances of a load balancer rule are one per load balanced node, so
on a
+ // large cluster the un-paged response was truncated and the stale
nodes on
+ // later pages were never removed from the rule.
+ const pageSize = 500
+
+ instances := make([]*cloudstack.VirtualMachine, 600)
+ for i := range instances {
+ instances[i] = &cloudstack.VirtualMachine{Id:
fmt.Sprintf("vm-%d", i)}
+ }
+
+ ctrl := gomock.NewController(t)
+ t.Cleanup(ctrl.Finish)
+
+ mockVM := cloudstack.NewMockVirtualMachineServiceIface(ctrl)
+ mockLB := cloudstack.NewMockLoadBalancerServiceIface(ctrl)
+
+ // getLoadBalancer: one rule, fits in a single page.
+ mockLB.EXPECT().NewListLoadBalancerRulesParams().
+ Return(&cloudstack.ListLoadBalancerRulesParams{})
+ mockLB.EXPECT().ListLoadBalancerRules(gomock.Any()).
+ Return(&cloudstack.ListLoadBalancerRulesResponse{
+ Count: 1,
+ LoadBalancerRules: []*cloudstack.LoadBalancerRule{
+ {Id: "rule-1", Name: "rule-1", Publicip:
"1.2.3.4", Publicipid: "ip-1"},
+ },
+ }, nil)
+
+ // verifyHosts: the cluster is down to a single node.
+ mockVM.EXPECT().NewListVirtualMachinesParams().
+ Return(&cloudstack.ListVirtualMachinesParams{})
+ mockVM.EXPECT().ListVirtualMachines(gomock.Any()).
+ Return(&cloudstack.ListVirtualMachinesResponse{
+ Count: 1,
+ VirtualMachines: []*cloudstack.VirtualMachine{
+ {Id: "vm-0", Name: "node-0", Nic:
[]cloudstack.Nic{{Networkid: "net-123"}}},
+ },
+ }, nil)
+
+ // The rule's members arrive a page at a time.
+ mockLB.EXPECT().NewListLoadBalancerRuleInstancesParams("rule-1").
+ Return(&cloudstack.ListLoadBalancerRuleInstancesParams{})
+ mockLB.EXPECT().ListLoadBalancerRuleInstances(gomock.Any()).Times(2).
+ DoAndReturn(func(p
*cloudstack.ListLoadBalancerRuleInstancesParams)
(*cloudstack.ListLoadBalancerRuleInstancesResponse, error) {
+ return
&cloudstack.ListLoadBalancerRuleInstancesResponse{
+ Count: len(instances),
+ LoadBalancerRuleInstances: pageOf(t, p,
instances, pageSize),
+ }, nil
+ })
+
+ var removed []string
+ mockLB.EXPECT().NewRemoveFromLoadBalancerRuleParams("rule-1").
+ Return(&cloudstack.RemoveFromLoadBalancerRuleParams{})
+ mockLB.EXPECT().RemoveFromLoadBalancerRule(gomock.Any()).
+ DoAndReturn(func(p
*cloudstack.RemoveFromLoadBalancerRuleParams)
(*cloudstack.RemoveFromLoadBalancerRuleResponse, error) {
+ removed, _ = p.GetVirtualmachineids()
+ return
&cloudstack.RemoveFromLoadBalancerRuleResponse{}, nil
+ })
+
+ cs := &CSCloud{
+ client: &cloudstack.CloudStackClient{VirtualMachine: mockVM,
LoadBalancer: mockLB},
+ }
+ service := &corev1.Service{
+ ObjectMeta: metav1.ObjectMeta{Name: "test-svc", Namespace:
"default", UID: "abc123"},
+ }
+
+ if err := cs.UpdateLoadBalancer(context.TODO(), "cluster", service,
nodesNamed("node-0")); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ // Every instance except vm-0 must be removed, including those that only
+ // appeared on the second page.
+ if len(removed) != len(instances)-1 {
+ t.Fatalf("removed %d hosts, want %d", len(removed),
len(instances)-1)
+ }
+ if slices.Contains(removed, "vm-0") {
+ t.Errorf("vm-0 is still a node but was removed from the rule")
+ }
+ if !slices.Contains(removed, "vm-599") {
+ t.Errorf("vm-599 is on the second page and should have been
removed, got %v", removed)
+ }
+}
diff --git a/pagination.go b/pagination.go
new file mode 100644
index 00000000..64e19a8c
--- /dev/null
+++ b/pagination.go
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package cloudstack
+
+import "fmt"
+
+// maxListPages bounds a single walk. Because the total is re-read on every
page,
+// a result set that grows as fast as it is consumed would otherwise keep the
+// walk going indefinitely. It is a backstop, not a limit expected to be
reached:
+// at CloudStack's default page size it allows half a million records.
+const maxListPages = 1000
+
+// pageableParams is the paging surface that every cloudstack-go List*Params
+// type exposes.
+type pageableParams interface {
+ SetPage(int)
+ SetPagesize(int)
+}
+
+// listAll makes further requests to fetch the remaining items if the count is
higher
+// than the number of items returned. It fails rather than returning a partial
+// list, so callers never reconcile against one.
+//
+// CloudStack offers no cursor, only page and pagesize, so a record removed
from
+// an earlier page shifts the rest down and one on a page boundary can be
missed.
+// Repeats, the other half of that, are dropped by dedupeByID at the call
sites.
+func listAll[T any](p pageableParams, list func() (count int, items []T, err
error)) ([]T, error) {
+ count, items, err := list()
+ if err != nil {
+ return nil, err
+ }
+
+ // Nothing was truncated, or there is nothing to page through.
+ if len(items) >= count || len(items) == 0 {
+ return items, nil
+ }
+
+ // The server just demonstrated how many records it will return at a
time,
+ // which is the one page size it is guaranteed to accept.
+ pageSize := len(items)
+ collected := items
+
+ for page := 2; len(collected) < count; page++ {
+ if page > maxListPages {
+ return nil, fmt.Errorf("gave up paging after %d pages
holding %d of %d records", maxListPages, len(collected), count)
+ }
+
+ p.SetPage(page)
+ p.SetPagesize(pageSize)
+
+ pageCount, items, err := list()
+ if err != nil {
+ return nil, err
+ }
+
+ // Records may be added while we are paging, which pushes the
total up.
+ // Track the highest the server has reported so growth cannot
cut the
+ // walk short; taking the highest rather than the latest also
keeps a
+ // shrinking total from ending the walk before the pages say so.
+ if pageCount > count {
+ count = pageCount
+ }
+
+ // Records may equally have been removed since the first
request, so
+ // trust the pages rather than the count and stop as soon as
one runs
+ // short.
+ if len(items) == 0 {
+ break
+ }
+
+ collected = append(collected, items...)
+
+ if len(items) < pageSize {
+ break
+ }
+ }
+
+ return collected, nil
+}
+
+// dedupeByID drops repeats from a walked list, keeping the first of each ID.
+//
+// A list that changes while it is being walked can return the same record on
+// more than one page, and every page decodes into its own structs, so a repeat
+// arrives as a second pointer to an equal record. Callers that key bookkeeping
+// on the pointer would otherwise treat the two as unrelated records.
+func dedupeByID[T any](items []T, id func(T) string) []T {
+ unique := make([]T, 0, len(items))
+ seen := make(map[string]bool, len(items))
+
+ for _, item := range items {
+ key := id(item)
+ if seen[key] {
+ continue
+ }
+ seen[key] = true
+ unique = append(unique, item)
+ }
+
+ return unique
+}
diff --git a/pagination_test.go b/pagination_test.go
new file mode 100644
index 00000000..aabb6ceb
--- /dev/null
+++ b/pagination_test.go
@@ -0,0 +1,305 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package cloudstack
+
+import (
+ "errors"
+ "reflect"
+ "testing"
+)
+
+// pagedServer doubles as the params object and the list API, faking a
CloudStack
+// server that truncates a request carrying no page at default.page.size.
+type pagedServer struct {
+ items []int
+ // pageSize is the server's default.page.size.
+ pageSize int
+ // count overrides the reported total when non-zero, to model a server
whose
+ // count disagrees with the records it actually hands out.
+ count int
+
+ // afterEach runs once a response has been built, to model a result set
that
+ // changes while it is being walked.
+ afterEach func(s *pagedServer)
+
+ // Paging parameters as set by listAll. Zero means "not sent".
+ page int
+ reqSize int
+
+ // requests records the (page, pagesize) pair of every call.
+ requests [][2]int
+}
+
+func (s *pagedServer) SetPage(v int) { s.page = v }
+func (s *pagedServer) SetPagesize(v int) { s.reqSize = v }
+
+func (s *pagedServer) list() (int, []int, error) {
+ s.requests = append(s.requests, [2]int{s.page, s.reqSize})
+
+ total := s.count
+ if total == 0 {
+ total = len(s.items)
+ }
+
+ // No page requested: the server applies default.page.size from page
one.
+ page, size := s.page, s.reqSize
+ if page == 0 {
+ page, size = 1, s.pageSize
+ }
+
+ defer func() {
+ if s.afterEach != nil {
+ s.afterEach(s)
+ }
+ }()
+
+ start := (page - 1) * size
+ if start >= len(s.items) {
+ return total, nil, nil
+ }
+
+ end := start + size
+ if end > len(s.items) {
+ end = len(s.items)
+ }
+
+ return total, s.items[start:end], nil
+}
+
+// grow appends n further records, as if they had been created elsewhere while
+// the walk was in progress.
+func (s *pagedServer) grow(n int) {
+ for i := 0; i < n; i++ {
+ s.items = append(s.items, len(s.items))
+ }
+}
+
+func seq(n int) []int {
+ items := make([]int, n)
+ for i := range items {
+ items[i] = i
+ }
+ return items
+}
+
+func TestListAll(t *testing.T) {
+ tests := []struct {
+ name string
+ server pagedServer
+ want []int
+ wantRequests [][2]int
+ }{
+ {
+ name: "fits in one page",
+ server: pagedServer{items: seq(3), pageSize: 500},
+ want: seq(3),
+ wantRequests: [][2]int{{0, 0}},
+ },
+ {
+ name: "empty result",
+ server: pagedServer{items: nil, pageSize: 500},
+ want: nil,
+ wantRequests: [][2]int{{0, 0}},
+ },
+ {
+ name: "exactly one full page",
+ server: pagedServer{items: seq(500), pageSize:
500},
+ want: seq(500),
+ wantRequests: [][2]int{{0, 0}},
+ },
+ {
+ name: "truncated, partial second page",
+ server: pagedServer{items: seq(750), pageSize:
500},
+ want: seq(750),
+ wantRequests: [][2]int{{0, 0}, {2, 500}},
+ },
+ {
+ name: "truncated, exact page multiple",
+ server: pagedServer{items: seq(1000), pageSize:
500},
+ want: seq(1000),
+ wantRequests: [][2]int{{0, 0}, {2, 500}},
+ },
+ {
+ name: "several pages",
+ server: pagedServer{items: seq(12), pageSize: 5},
+ want: seq(12),
+ wantRequests: [][2]int{{0, 0}, {2, 5}, {3, 5}},
+ },
+ {
+ name: "a page size of one still terminates",
+ server: pagedServer{items: seq(3), pageSize: 1},
+ want: seq(3),
+ wantRequests: [][2]int{{0, 0}, {2, 1}, {3, 1}},
+ },
+ {
+ // Records removed between requests: the walk stops on
the short page
+ // rather than spinning until the stale count is
reached.
+ name: "count overstates what the server
returns",
+ server: pagedServer{items: seq(7), pageSize: 5,
count: 100},
+ want: seq(7),
+ wantRequests: [][2]int{{0, 0}, {2, 5}},
+ },
+ {
+ // Every page is full but the count is never reached,
so termination
+ // has to come from the first empty page.
+ name: "count overstates on an exact page
boundary",
+ server: pagedServer{items: seq(10), pageSize: 5,
count: 100},
+ want: seq(10),
+ wantRequests: [][2]int{{0, 0}, {2, 5}, {3, 5}},
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ server := tt.server
+
+ got, err := listAll(&server, server.list)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if !reflect.DeepEqual(got, tt.want) {
+ t.Errorf("items = %v, want %v", got, tt.want)
+ }
+ if !reflect.DeepEqual(server.requests, tt.wantRequests)
{
+ t.Errorf("requests = %v, want %v",
server.requests, tt.wantRequests)
+ }
+ })
+ }
+}
+
+func TestListAllFollowsAGrowingResultSet(t *testing.T) {
+ // The count reported on the first page goes stale as soon as records
are
+ // added, so a walk that trusts only that first total stops early. Here
five
+ // records appear after the first request: a walk pinned to the original
+ // total of 10 would return 10 of the 15 that exist.
+ server := &pagedServer{items: seq(10), pageSize: 5}
+ server.afterEach = func(s *pagedServer) {
+ if len(s.requests) == 1 {
+ s.grow(5)
+ }
+ }
+
+ got, err := listAll(server, server.list)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if len(got) != 15 {
+ t.Errorf("collected %d records, want 15 - the walk stopped at a
stale total", len(got))
+ }
+ if !reflect.DeepEqual(got, seq(15)) {
+ t.Errorf("items = %v, want %v", got, seq(15))
+ }
+}
+
+func TestListAllFailsOnAGrowingResultSetRunningAway(t *testing.T) {
+ // Re-reading the total on every page means a set that grows exactly as
fast
+ // as it is consumed would never satisfy the loop condition. The page
cap
+ // stops the walk, and it has to fail rather than hand back what it
holds:
+ // callers decide whether rules and hosts exist from these lists, so a
+ // partial one would have them delete or recreate live resources.
+ server := &pagedServer{items: seq(10), pageSize: 5}
+ server.afterEach = func(s *pagedServer) { s.grow(5) }
+
+ got, err := listAll(server, server.list)
+ if err == nil {
+ t.Fatalf("collected %d records with no error, want a failure at
the cap", len(got))
+ }
+ if got != nil {
+ t.Errorf("items = %v, want nil so a partial list cannot be
used", got)
+ }
+ if len(server.requests) != maxListPages {
+ t.Errorf("made %d requests, want the walk capped at %d",
len(server.requests), maxListPages)
+ }
+}
+
+func TestListAllNeverSendsPageZero(t *testing.T) {
+ // CloudStack rejects a page parameter that is merely present, so
page=0 is
+ // an error rather than a way of asking for the first page. The walk
starts
+ // at 2, which makes that unrepresentable.
+ server := &pagedServer{items: seq(150), pageSize: 50}
+
+ if _, err := listAll(server, server.list); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ if len(server.requests) < 2 {
+ t.Fatalf("expected the result to be paged, got %v",
server.requests)
+ }
+ if first := server.requests[0]; first != [2]int{0, 0} {
+ t.Errorf("first request = %v, want no paging parameters at
all", first)
+ }
+ for _, request := range server.requests[1:] {
+ if request[0] < 2 {
+ t.Errorf("request %v used page %d, want >= 2", request,
request[0])
+ }
+ }
+}
+
+func TestListAllPagesWithTheServersOwnPageSize(t *testing.T) {
+ // pagesize may not exceed default.page.size, so the walk has to reuse
the
+ // length the server itself returned rather than a fixed value.
+ server := &pagedServer{items: seq(150), pageSize: 50}
+
+ if _, err := listAll(server, server.list); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ for _, request := range server.requests[1:] {
+ if request[1] != 50 {
+ t.Errorf("request %v used pagesize %d, want 50",
request, request[1])
+ }
+ }
+}
+
+func TestListAllError(t *testing.T) {
+ wantErr := errors.New("boom")
+ server := &pagedServer{}
+
+ t.Run("on the first request", func(t *testing.T) {
+ got, err := listAll(server, func() (int, []int, error) {
+ return 0, nil, wantErr
+ })
+ if !errors.Is(err, wantErr) {
+ t.Errorf("err = %v, want %v", err, wantErr)
+ }
+ if got != nil {
+ t.Errorf("items = %v, want nil", got)
+ }
+ })
+
+ t.Run("on a later page", func(t *testing.T) {
+ calls := 0
+ got, err := listAll(server, func() (int, []int, error) {
+ calls++
+ if calls == 1 {
+ return 750, seq(500), nil
+ }
+ return 0, nil, wantErr
+ })
+ if !errors.Is(err, wantErr) {
+ t.Errorf("err = %v, want %v", err, wantErr)
+ }
+ // A partial result is worse than no result: callers of these
lists treat
+ // a missing record as "does not exist" and create or delete
accordingly.
+ if got != nil {
+ t.Errorf("items = %v, want nil", got)
+ }
+ })
+}
