Feat: Add support for stickyness policy (#88)
diff --git a/README.md b/README.md
index cf936d2..8314a37 100644
--- a/README.md
+++ b/README.md
@@ -56,6 +56,11 @@
The access token needs to be able to fetch VM information and deploy load balancers in the project or domain where the nodes reside.
+Stickiness policies also need `listLBStickinessPolicies`, `createLBStickinessPolicy` and
+`deleteLBStickinessPolicy`. The account that CloudStack's Kubernetes service creates for clusters in
+projects cannot call them, so stickiness does not work there yet. Services without stickiness are
+not affected.
+
To create the secret, use the following command:
```bash
kubectl -n kube-system create secret generic cloudstack-secret --from-file=cloud-config
@@ -180,6 +185,119 @@
Setting it to an empty value (`""`) sends an empty CIDR list to CloudStack — it does not block all
traffic.
+#### `service.beta.kubernetes.io/cloudstack-load-balancer-stickiness-method-name`
+
+**Type:** String
+
+**Default:** Not set (no stickiness policy)
+
+**Description:** Adds a CloudStack **LB stickiness policy** to the service's load balancer rules,
+so the load balancer sends a client to the same node each time. The VirtualRouter, which VPC
+routers also use, supports:
+
+| Method | Keeps a client by | Ports |
+| --- | --- | --- |
+| `LbCookie` | A cookie the load balancer adds | Ports with `appProtocol: http` |
+| `SourceBased` | The client's IP address | All TCP ports |
+
+`LbCookie` switches its ports to HTTP mode, which breaks TLS and any other traffic that is not
+plain HTTP. So it only goes on ports that set
+[`appProtocol: http`](https://kubernetes.io/docs/concepts/services-networking/service/#application-protocol),
+and a service with no such port fails to sync. The ingress-nginx Helm chart sets `appProtocol` on
+its ports by default, so its HTTP port gets the cookie and its HTTPS port does not.
+
+`LbCookie` cannot work on HTTPS ports, because the load balancer passes TLS through and never sees
+the HTTP headers. Use `SourceBased` there, or the cookie affinity of the ingress controller or
+Gateway that terminates TLS.
+
+`AppCookie` is not supported. CloudStack sets it up with HAProxy's `appsession` option, which
+HAProxy removed in version 1.6.
+
+Method and parameter names are checked against what the network supports before anything is
+created, so a typo fails the sync without changing CloudStack. Method names are not case sensitive.
+
+Stickiness keeps a client on the same **node**, not the same pod. kube-proxy on that node can still
+send the request to a pod on another node. Set `externalTrafficPolicy: Local` to keep requests on
+the node's own pods. The router health-checks each node, so nodes without a local pod get no
+traffic.
+
+**Use Case:** Applications that keep a client's state in memory, such as sessions, and need the
+same client to reach the same node.
+
+**Example:**
+```yaml
+apiVersion: v1
+kind: Service
+metadata:
+ name: my-ingress
+ annotations:
+ service.beta.kubernetes.io/cloudstack-load-balancer-stickiness-method-name: "LbCookie"
+ service.beta.kubernetes.io/cloudstack-load-balancer-stickiness-method-param: "cookie-name=SERVERID,nocache=true"
+spec:
+ type: LoadBalancer
+ externalTrafficPolicy: Local
+ ports:
+ - name: http
+ port: 80
+ appProtocol: http
+ - name: https
+ port: 443
+ appProtocol: https
+```
+
+**Notes:**
+- Changing the method or parameters replaces the policy. With `SourceBased`, clients may move to
+ another node. With `LbCookie`, they keep their node unless the cookie name changes.
+- If CloudStack rejects a change, the current policy stays. If the router fails to apply it, the
+ rule can be left without a policy until the service is synced again.
+- Removing the annotation deletes the policies the controller created. They have the description
+ `Managed by the CloudStack Kubernetes Provider`. Policies added by hand are kept, except on ports
+ the annotations select, where the controller replaces them, and on rules the controller
+ recreates, for example after a node port change.
+- A new service whose policy CloudStack rejects keeps its public IP and rules, but stays
+ `<pending>` until the annotation is fixed.
+- Changes apply the next time the service is synced, which happens when the service changes or the
+ controller restarts.
+
+#### `service.beta.kubernetes.io/cloudstack-load-balancer-stickiness-method-param`
+
+**Type:** String (comma-separated `key=value` list)
+
+**Default:** Not set (no parameters)
+
+**Description:** Parameters for the method in
+`service.beta.kubernetes.io/cloudstack-load-balancer-stickiness-method-name`. Without a method,
+this annotation does nothing. The VirtualRouter supports:
+
+| Method | Parameter | Value |
+| --- | --- | --- |
+| `LbCookie` | `cookie-name` | Name of the cookie. CloudStack picks one if it is not set |
+| `LbCookie` | `mode` | `insert` (default), `rewrite` or `prefix` |
+| `LbCookie` | `domain` | Domain the cookie is set for |
+| `LbCookie` | `nocache`, `indirect`, `postonly` | Flags: `true` turns one on, `false` leaves it off |
+| `SourceBased` | `tablesize` | Size of the client table: a number with `k`, `m` or `g`, such as `200k` |
+| `SourceBased` | `expire` | How long an idle client is remembered: a number with `d`, `h`, `m` or `s`, such as `30m` |
+
+Parameter names are not case sensitive. Flags accept the values of Go's `strconv.ParseBool`, such
+as `true`, `false`, `1` and `0`.
+
+**Format:** Comma-separated `key=value` pairs. Spaces around entries and empty entries are ignored.
+CloudStack copies the values into the router's HAProxy config, so the sync fails on:
+
+- an entry without a value, such as `nocache=` (write `nocache=true` instead)
+- a space or one of `= & " ' \ #` inside a key or value
+- a key that is set twice
+- an unknown parameter, an `LbCookie` mode other than the three above, or a flag that is not
+ `true` or `false`
+
+CloudStack itself rejects `tablesize` and `expire` values without a valid suffix.
+
+**Example:**
+```yaml
+ service.beta.kubernetes.io/cloudstack-load-balancer-stickiness-method-name: "SourceBased"
+ service.beta.kubernetes.io/cloudstack-load-balancer-stickiness-method-param: "tablesize=200k,expire=30m"
+```
+
#### `service.beta.kubernetes.io/cloudstack-load-balancer-ip-associated-by-controller`
**Type:** Boolean (`"true"` or `"false"`)
@@ -269,6 +387,13 @@
Any other value makes the service fail to sync with `unsupported load balancer affinity`. Other
CloudStack algorithms, such as `leastconn`, cannot currently be selected.
+The algorithm is separate from stickiness. `spec.sessionAffinity: ClientIP` picks the algorithm,
+while the
+[`stickiness-method-name`](#servicebetakubernetesiocloudstack-load-balancer-stickiness-method-name)
+and
+[`stickiness-method-param`](#servicebetakubernetesiocloudstack-load-balancer-stickiness-method-param)
+annotations add a stickiness policy. The two can be used together.
+
### VPC Networks
VPC networks are supported. VPC networks normally do not offer the Firewall service, so the
@@ -385,6 +510,19 @@
The image architecture does not match the node. Releases up to v1.1.0 were published for amd64 only;
use a newer release, which ships multi-architecture images including arm64.
+### A TLS or TCP port stopped working after enabling `LbCookie` stickiness
+
+`LbCookie` switches its ports to HTTP mode, which breaks traffic that is not plain HTTP. It only
+goes on ports with `appProtocol: http`. Set that only on plain HTTP ports, or use `SourceBased`,
+which works on any TCP port.
+
+### A service's events show a stickiness error
+
+The controller or CloudStack rejected the stickiness annotations, and the event says why.
+`Failed to create stickiness policy` with no reason means the router could not apply the policy.
+If CloudStack rejects a change, an existing service keeps its current policy. A new service keeps
+its public IP and rules, but stays `<pending>` until the annotation is fixed.
+
## Migration Guide
There are several notable differences to the old Kubernetes CloudStack cloud provider that need to be taken into
@@ -405,6 +543,11 @@
rule on the first reconcile, matching it by public IP, IP protocol and public port, and renames it
to the current scheme. Existing rules therefore do not have to be removed before migrating.
+The old provider did not manage stickiness, so its rules may have policies added by hand. The
+controller keeps them. It only deletes policies it created, which have the description
+`Managed by the CloudStack Kubernetes Provider`. On ports the stickiness annotations select, the
+controller's policy replaces a hand-made one.
+
:warning: **Rules of a Service that is recreated during the migration are not adopted.**
Rule names derive from the Service UID, so a Service recreated with a new UID leaves its old rules
diff --git a/cloudstack_loadbalancer.go b/cloudstack_loadbalancer.go
index 481c114..753322d 100644
--- a/cloudstack_loadbalancer.go
+++ b/cloudstack_loadbalancer.go
@@ -21,11 +21,16 @@
import (
"context"
+ "encoding/json"
+ "errors"
"fmt"
+ "maps"
"net"
+ "slices"
"sort"
"strconv"
"strings"
+ "unicode"
"github.com/apache/cloudstack-go/v2/cloudstack"
"github.com/blang/semver/v4"
@@ -57,6 +62,14 @@
// associated the IP address. This annotation is set by the controller when it associates
// an unallocated IP, and is used to determine if the IP should be disassociated on deletion.
ServiceAnnotationLoadBalancerIPAssociatedByController = "service.beta.kubernetes.io/cloudstack-load-balancer-ip-associated-by-controller" //nolint:gosec
+
+ // ServiceAnnotationLoadBalancerStickinessMethodName sets the stickiness method, LbCookie or
+ // SourceBased, for the Service's load balancer rules.
+ ServiceAnnotationLoadBalancerStickinessMethodName = "service.beta.kubernetes.io/cloudstack-load-balancer-stickiness-method-name"
+
+ // ServiceAnnotationLoadBalancerStickinessMethodParam sets the method's parameters as a
+ // comma-separated list of key=value pairs.
+ ServiceAnnotationLoadBalancerStickinessMethodParam = "service.beta.kubernetes.io/cloudstack-load-balancer-stickiness-method-param"
)
// cidrListUpdateVersion is the first CloudStack release whose updateLoadBalancerRule API
@@ -137,6 +150,11 @@
return nil, fmt.Errorf("requested load balancer with no ports")
}
+ stickiness, err := parseStickinessSpec(service)
+ if err != nil {
+ return nil, err
+ }
+
// Get the load balancer details and existing rules.
lb, err := cs.getLoadBalancer(service)
if err != nil {
@@ -163,6 +181,17 @@
return nil, err
}
+ // Read the network before an IP is associated, so a stickiness method the network does not
+ // support fails before anything is created.
+ network, _, err := lb.Network.GetNetworkByID(lb.networkID, cloudstack.WithProject(lb.projectID))
+ if err != nil {
+ return nil, err
+ }
+ if err := stickiness.normalise(supportedStickinessMethods(network)); err != nil {
+ return nil, err
+ }
+
+ rulesApplied := false
if !lb.hasLoadBalancerIP() {
// Create or retrieve the load balancer IP.
if err := lb.getLoadBalancerIP(service.Spec.LoadBalancerIP); err != nil {
@@ -170,8 +199,9 @@
}
if lb.ipAddr != "" && lb.ipAddr != service.Spec.LoadBalancerIP {
+ // Once the rules exist on the new IP, keep it, so the next sync can retry against them.
defer func(lb *loadBalancer) {
- if err != nil {
+ if err != nil && !rulesApplied {
if err := lb.releaseLoadBalancerIP(); err != nil {
klog.Errorf(err.Error())
}
@@ -196,11 +226,6 @@
return nil, err
}
- network, _, err := lb.Network.GetNetworkByID(lb.networkID, cloudstack.WithProject(lb.projectID))
- if err != nil {
- return nil, err
- }
-
blocking, rest := lb.partitionObsoleteRules(desired)
// Obsolete rules holding a public port that a new rule needs have to go first, or
@@ -209,13 +234,18 @@
return nil, err
}
- if err := lb.applyLoadBalancerRules(desired, service, network, cs.version); err != nil {
+ rules, err := lb.applyLoadBalancerRules(desired, service, network, cs.version)
+ if err != nil {
return nil, err
}
+ rulesApplied = true
// Everything else is removed only once the desired rules are in place, so a failure here
- // can never leave the service without the rules it does need.
- if err := lb.pruneRules(rest, desired, network); err != nil {
+ // can never leave the service without the rules it does need. Stickiness is applied last,
+ // even if the prune fails.
+ pruneErr := lb.pruneRules(rest, desired, network)
+ stickinessErr := lb.applyStickinessPolicies(desired, rules, stickiness)
+ if err := errors.Join(pruneErr, stickinessErr); err != nil {
return nil, err
}
@@ -275,7 +305,7 @@
}
if len(remove) > 0 {
- klog.V(4).Infof("Removing old hosts (%v) from load balancer rule: %v", assign, lbRule.Name)
+ klog.V(4).Infof("Removing old hosts (%v) from load balancer rule: %v", remove, lbRule.Name)
if err := lb.removeHostsFromRule(lbRule, remove); err != nil {
return err
}
@@ -1008,15 +1038,20 @@
// ensureLoadBalancerRule brings the load balancer rule of one desired service port in line and
// returns it: an up-to-date rule is left alone, an outdated one is updated in place, and a
// missing one is created. A rule that cannot be updated is deleted immediately before its
-// replacement is created, which keeps the port unserved for as short a time as possible.
+// replacement is created, which keeps the port unserved for as short a time as possible. An
+// adopted rule that has no hosts gets them assigned.
func (lb *loadBalancer) ensureLoadBalancerRule(d desiredLBRule, service *corev1.Service, version semver.Version) (*cloudstack.LoadBalancerRule, error) {
switch d.change {
case ruleUpToDate:
klog.V(4).Infof("Load balancer rule %v is up-to-date", d.name)
- return d.existing, nil
+ return d.existing, lb.repairEmptyRule(d.existing)
case ruleNeedsUpdate:
klog.V(4).Infof("Updating load balancer rule: %v", d.name)
- return d.existing, lb.updateLoadBalancerRule(d.existing, d.name, d.protocol, service, version)
+ if err := lb.updateLoadBalancerRule(d.existing, d.name, d.protocol, service, version); err != nil {
+ return nil, err
+ }
+
+ return d.existing, lb.repairEmptyRule(d.existing)
case ruleNeedsRecreate:
klog.V(4).Infof("Deleting load balancer rule %v so it can be created again", d.existing.Name)
if err := lb.deleteLoadBalancerRule(d.existing); err != nil {
@@ -1038,28 +1073,449 @@
return lbRule, nil
}
+// repairEmptyRule assigns the hosts to an adopted rule that has none, which happens when a create
+// fails before assigning them. Rules that have hosts are left to UpdateLoadBalancer, because node
+// sync runs at the same time as this sync and may have newer node data.
+func (lb *loadBalancer) repairEmptyRule(lbRule *cloudstack.LoadBalancerRule) error {
+ p := lb.LoadBalancer.NewListLoadBalancerRuleInstancesParams(lbRule.Id)
+ instances, err := lb.LoadBalancer.ListLoadBalancerRuleInstances(p)
+ if err != nil {
+ return fmt.Errorf("error retrieving associated instances: %v", err)
+ }
+ if instances.Count > 0 {
+ return nil
+ }
+
+ klog.V(4).Infof("Assigning hosts (%v) to load balancer rule %v, which has none", lb.hostIDs, lbRule.Name)
+ return lb.assignHostsToRule(lbRule, lb.hostIDs)
+}
+
// applyLoadBalancerRules creates or updates the load balancer rule of every desired service
-// port and reconciles the firewall or network ACL rules it needs.
-func (lb *loadBalancer) applyLoadBalancerRules(desired []desiredLBRule, service *corev1.Service, network *cloudstack.Network, version semver.Version) error {
+// port and reconciles the firewall or network ACL rules it needs. It returns the rule of each
+// desired port, in the same order.
+func (lb *loadBalancer) applyLoadBalancerRules(desired []desiredLBRule, service *corev1.Service, network *cloudstack.Network, version semver.Version) ([]*cloudstack.LoadBalancerRule, error) {
+ rules := make([]*cloudstack.LoadBalancerRule, 0, len(desired))
for _, d := range desired {
lbRule, err := lb.ensureLoadBalancerRule(d, service, version)
if err != nil {
- return err
+ return nil, err
}
+ rules = append(rules, lbRule)
if isFirewallSupported(network.Service) {
klog.V(4).Infof("Creating firewall rules for load balancer rule: %v (%v:%v:%v)", d.name, d.protocol, lbRule.Publicip, d.port.Port)
if _, err := lb.updateFirewallRule(lbRule.Publicipid, int(d.port.Port), d.protocol, service.Spec.LoadBalancerSourceRanges); err != nil {
- return err
+ return nil, err
}
} else if isNetworkACLSupported(network.Service) {
klog.V(4).Infof("Creating ACL rules for load balancer rule: %v (%v:%v:%v)", d.name, d.protocol, lbRule.Publicip, d.port.Port)
if _, err := lb.updateNetworkACL(int(d.port.Port), d.protocol, network.Id); err != nil {
- return err
+ return nil, err
}
}
}
+ return rules, nil
+}
+
+const (
+ // stickinessPolicyDescription marks the policies the controller creates. Only policies with
+ // this description are deleted, so policies added by hand are kept.
+ stickinessPolicyDescription = "Managed by the CloudStack Kubernetes Provider"
+
+ lbCookieMethod = "LbCookie"
+ appCookieMethod = "AppCookie"
+
+ // stickinessParamForbidden lists the characters a parameter may not contain. CloudStack splits
+ // stored parameters on "=" and "&", and copies them unquoted into the HAProxy config, where
+ // quotes, backslashes and "#" have a special meaning.
+ stickinessParamForbidden = "=&\"'\\#"
+)
+
+// lbCookieModes are the cookie modes HAProxy accepts. CloudStack does not check the mode.
+var lbCookieModes = []string{"insert", "rewrite", "prefix"}
+
+// stickinessSpec is the stickiness policy a Service asks for. ports holds the selected TCP ports.
+type stickinessSpec struct {
+ method string
+ params map[string]string
+ ports map[int32]bool
+}
+
+// stickinessMethod is a method in a network's SupportedStickinessMethods capability.
+type stickinessMethod struct {
+ Name string `json:"methodname"`
+ Params []stickinessMethodParam `json:"paramlist"`
+}
+
+// stickinessMethodParam is a parameter of a stickiness method.
+type stickinessMethodParam struct {
+ Name string `json:"paramname"`
+ Required bool `json:"required"`
+ IsFlag bool `json:"isflag"`
+}
+
+// parseStickinessSpec reads the stickiness annotations. It returns nil when the Service asks for
+// no stickiness. It makes no CloudStack calls, so bad annotations fail before anything changes.
+func parseStickinessSpec(service *corev1.Service) (*stickinessSpec, error) {
+ method := strings.TrimSpace(getStringFromServiceAnnotation(service, ServiceAnnotationLoadBalancerStickinessMethodName, ""))
+ if method == "" {
+ return nil, nil
+ }
+ if strings.EqualFold(method, appCookieMethod) {
+ return nil, fmt.Errorf("unsupported stickiness method %s in annotation %s: HAProxy removed the appsession option it needs", method, ServiceAnnotationLoadBalancerStickinessMethodName)
+ }
+
+ params, err := parseStickinessParams(getStringFromServiceAnnotation(service, ServiceAnnotationLoadBalancerStickinessMethodParam, ""))
+ if err != nil {
+ return nil, err
+ }
+ ports, err := selectStickinessPorts(service, method)
+ if err != nil {
+ return nil, err
+ }
+
+ return &stickinessSpec{method: method, params: params, ports: ports}, nil
+}
+
+// parseStickinessParams parses the key=value list of the method-param annotation. Empty values,
+// spaces and the characters in stickinessParamForbidden are rejected, because CloudStack would
+// read them back differently or write them into a broken HAProxy config.
+func parseStickinessParams(value string) (map[string]string, error) {
+ params := make(map[string]string)
+ seen := make(map[string]bool)
+ for _, entry := range strings.Split(value, ",") {
+ entry = strings.TrimSpace(entry)
+ if entry == "" {
+ continue
+ }
+ key, val, found := strings.Cut(entry, "=")
+ key, val = strings.TrimSpace(key), strings.TrimSpace(val)
+ switch {
+ case !found || key == "":
+ return nil, stickinessParamError(entry, "expected key=value")
+ case val == "":
+ return nil, stickinessParamError(entry, fmt.Sprintf("missing value, write flags as %s=true", key))
+ case !safeStickinessText(key) || !safeStickinessText(val):
+ return nil, stickinessParamError(entry, `spaces and the characters = & " ' \ # are not allowed`)
+ }
+
+ lowered := strings.ToLower(key)
+ if seen[lowered] {
+ return nil, stickinessParamError(entry, key+" is set twice")
+ }
+ seen[lowered] = true
+ params[key] = val
+ }
+
+ return params, nil
+}
+
+// stickinessParamError reports an invalid entry of the method-param annotation.
+func stickinessParamError(entry, reason string) error {
+ return fmt.Errorf("invalid stickiness parameter %q in annotation %s: %s", entry, ServiceAnnotationLoadBalancerStickinessMethodParam, reason)
+}
+
+// safeStickinessText reports whether s has no spaces and no forbidden characters.
+func safeStickinessText(s string) bool {
+ return !strings.ContainsAny(s, stickinessParamForbidden) && strings.IndexFunc(s, unicode.IsSpace) < 0
+}
+
+// selectStickinessPorts returns the TCP ports that get the policy. LbCookie switches its ports to
+// HTTP mode, which breaks TLS and other non-HTTP traffic, so it only goes on ports with
+// appProtocol: http. Other methods go on every TCP port.
+func selectStickinessPorts(service *corev1.Service, method string) (map[int32]bool, error) {
+ tcpPorts := make(map[int32]bool)
+ httpPorts := make(map[int32]bool)
+ for _, port := range service.Spec.Ports {
+ if port.Protocol != corev1.ProtocolTCP {
+ continue
+ }
+ tcpPorts[port.Port] = true
+ if port.AppProtocol != nil && strings.EqualFold(*port.AppProtocol, "http") {
+ httpPorts[port.Port] = true
+ }
+ }
+
+ if !strings.EqualFold(method, lbCookieMethod) {
+ return tcpPorts, nil
+ }
+ if len(httpPorts) == 0 {
+ return nil, fmt.Errorf("stickiness method %s only works on plain HTTP ports: set appProtocol: http on them", method)
+ }
+
+ return httpPorts, nil
+}
+
+// wants reports whether a Service port should get the policy.
+func (s *stickinessSpec) wants(port corev1.ServicePort) bool {
+ return s != nil && port.Protocol == corev1.ProtocolTCP && s.ports[port.Port]
+}
+
+// matches reports whether a policy is the controller's and matches the spec.
+func (s *stickinessSpec) matches(policy cloudstack.LBStickinessPolicyStickinesspolicy) bool {
+ return policy.Description == stickinessPolicyDescription &&
+ strings.EqualFold(policy.Methodname, s.method) &&
+ maps.Equal(policy.Params, s.params)
+}
+
+// supportedStickinessMethods returns the stickiness methods the network's load balancer
+// supports, or nil if the network does not list them.
+func supportedStickinessMethods(network *cloudstack.Network) []stickinessMethod {
+ for _, svc := range network.Service {
+ if svc.Name != "Lb" {
+ continue
+ }
+ for _, capability := range svc.Capability {
+ if capability.Name != "SupportedStickinessMethods" {
+ continue
+ }
+ var methods []stickinessMethod
+ if err := json.Unmarshal([]byte(capability.Value), &methods); err != nil {
+ klog.Warningf("Cannot read the stickiness methods of network %v: %v", network.Id, err)
+ return nil
+ }
+ return methods
+ }
+ }
+
+ return nil
+}
+
+// normalise checks the spec against the methods the network supports, as CloudStack does, and
+// also checks the LbCookie mode, which CloudStack does not. It writes names and flags the way
+// CloudStack stores them, so a policy read back compares equal. It does nothing if the network
+// does not list its methods.
+func (s *stickinessSpec) normalise(methods []stickinessMethod) error {
+ if s == nil || len(methods) == 0 {
+ return nil
+ }
+ method := findStickinessMethod(methods, s.method)
+ if method == nil {
+ return fmt.Errorf("stickiness method %q in annotation %s is not supported on this network; supported methods: %s", s.method, ServiceAnnotationLoadBalancerStickinessMethodName, stickinessMethodNames(methods))
+ }
+
+ params := make(map[string]string, len(s.params))
+ for key, value := range s.params {
+ param := findStickinessParam(method.Params, key)
+ if param == nil {
+ return fmt.Errorf("stickiness parameter %q in annotation %s is not supported by %s; supported parameters: %s", key, ServiceAnnotationLoadBalancerStickinessMethodParam, method.Name, stickinessParamNames(method.Params))
+ }
+ normalised, send, err := normaliseStickinessValue(method.Name, *param, value)
+ if err != nil {
+ return err
+ }
+ if send {
+ params[param.Name] = normalised
+ }
+ }
+ for _, param := range method.Params {
+ if _, set := params[param.Name]; param.Required && !set {
+ return fmt.Errorf("stickiness method %s needs parameter %s in annotation %s", method.Name, param.Name, ServiceAnnotationLoadBalancerStickinessMethodParam)
+ }
+ }
+
+ s.method, s.params = method.Name, params
+ return nil
+}
+
+// normaliseStickinessValue returns the value to send for a parameter, and whether to send it.
+// HAProxy turns a flag on whenever it is set, whatever the value, so flags are parsed as booleans
+// and sent as true or left out, like the CloudStack UI does.
+func normaliseStickinessValue(method string, param stickinessMethodParam, value string) (string, bool, error) {
+ if param.IsFlag {
+ on, err := strconv.ParseBool(value)
+ if err != nil {
+ return "", false, fmt.Errorf("stickiness flag %s in annotation %s must be true or false, not %q", param.Name, ServiceAnnotationLoadBalancerStickinessMethodParam, value)
+ }
+ return "true", on, nil
+ }
+ if strings.EqualFold(method, lbCookieMethod) && param.Name == "mode" {
+ mode := strings.ToLower(value)
+ if !slices.Contains(lbCookieModes, mode) {
+ return "", false, fmt.Errorf("stickiness parameter mode in annotation %s must be one of %s, not %q", ServiceAnnotationLoadBalancerStickinessMethodParam, strings.Join(lbCookieModes, ", "), value)
+ }
+ return mode, true, nil
+ }
+
+ return value, true, nil
+}
+
+// findStickinessMethod returns the method with the given name, ignoring case like CloudStack.
+func findStickinessMethod(methods []stickinessMethod, name string) *stickinessMethod {
+ for i := range methods {
+ if strings.EqualFold(methods[i].Name, name) {
+ return &methods[i]
+ }
+ }
+
+ return nil
+}
+
+// findStickinessParam returns the parameter with the given name, ignoring case.
+func findStickinessParam(params []stickinessMethodParam, name string) *stickinessMethodParam {
+ for i := range params {
+ if strings.EqualFold(params[i].Name, name) {
+ return ¶ms[i]
+ }
+ }
+
+ return nil
+}
+
+// stickinessMethodNames lists the supported methods for an error message. AppCookie is left out
+// because the controller rejects it.
+func stickinessMethodNames(methods []stickinessMethod) string {
+ names := make([]string, 0, len(methods))
+ for _, method := range methods {
+ if !strings.EqualFold(method.Name, appCookieMethod) {
+ names = append(names, method.Name)
+ }
+ }
+
+ return strings.Join(names, ", ")
+}
+
+// stickinessParamNames lists a method's parameters for an error message.
+func stickinessParamNames(params []stickinessMethodParam) string {
+ names := make([]string, 0, len(params))
+ for _, param := range params {
+ names = append(names, param.Name)
+ }
+
+ return strings.Join(names, ", ")
+}
+
+// applyStickinessPolicies updates the stickiness policy of every desired rule, once the rules are
+// in place. It tries every rule and returns all errors together.
+func (lb *loadBalancer) applyStickinessPolicies(desired []desiredLBRule, rules []*cloudstack.LoadBalancerRule, spec *stickinessSpec) error {
+ var errs []error
+ for i, d := range desired {
+ if err := lb.applyStickinessPolicy(d, rules[i], spec); err != nil {
+ errs = append(errs, fmt.Errorf("load balancer rule %v: %w", d.name, err))
+ }
+ }
+
+ return errors.Join(errs...)
+}
+
+// applyStickinessPolicy updates the stickiness policy of one rule. A rule created in this sync has
+// no policy yet, so it is not looked up. If the account may not call the stickiness APIs and the
+// rule needs no policy, nothing is done. The account CloudStack's Kubernetes service creates for
+// clusters in projects is one such account.
+func (lb *loadBalancer) applyStickinessPolicy(d desiredLBRule, lbRule *cloudstack.LoadBalancerRule, spec *stickinessSpec) error {
+ want := spec.wants(d.port)
+ var live []cloudstack.LBStickinessPolicyStickinesspolicy
+ if !d.createsRule() {
+ var err error
+ live, err = lb.liveStickinessPolicies(lbRule.Id)
+ switch {
+ case err != nil && isNotAllowed(err) && !want:
+ klog.V(4).Infof("Skipping the stickiness policies of load balancer rule %v: %v", d.name, err)
+ return nil
+ case err != nil && isNotAllowed(err):
+ return fmt.Errorf("%w; stickiness needs the listLBStickinessPolicies, createLBStickinessPolicy and deleteLBStickinessPolicy APIs", err)
+ case err != nil:
+ return err
+ }
+ }
+
+ create, stale, err := stickinessChanges(live, spec, want)
+ if err != nil {
+ return err
+ }
+ for _, id := range stale {
+ if err := lb.deleteStickinessPolicy(d.name, id); err != nil {
+ return err
+ }
+ }
+ if create {
+ return lb.createStickinessPolicy(d.name, lbRule.Id, spec)
+ }
+
+ return nil
+}
+
+// isNotAllowed reports whether CloudStack refused a call because the account's role does not
+// allow that API. CloudStack answers with error 432, which the SDK only keeps in the message.
+func isNotAllowed(err error) bool {
+ return strings.Contains(err.Error(), "CloudStack API error 432 ")
+}
+
+// stickinessChanges decides what to do with a rule's live policies. A selected rule should end
+// with one policy, the controller's, matching the spec. Creating it is enough, as CloudStack then
+// revokes the rule's other policies. CloudStack refuses the create while two policies are live,
+// and deleting one first could leave the rule without the current policy if the create then
+// fails, so that case is an error and nothing changes. A rule that is not selected loses the
+// controller's policies and keeps the rest.
+func stickinessChanges(live []cloudstack.LBStickinessPolicyStickinesspolicy, spec *stickinessSpec, want bool) (bool, []string, error) {
+ if !want {
+ var owned []string
+ for _, policy := range live {
+ if policy.Description == stickinessPolicyDescription {
+ owned = append(owned, policy.Id)
+ }
+ }
+ return false, owned, nil
+ }
+
+ switch len(live) {
+ case 0:
+ return true, nil, nil
+ case 1:
+ return !spec.matches(live[0]), nil, nil
+ default:
+ return false, nil, fmt.Errorf("the rule has %d stickiness policies, and CloudStack adds a new one only when there is at most one: remove the extra ones", len(live))
+ }
+}
+
+// liveStickinessPolicies lists the policies of a rule, skipping revoked ones CloudStack has not
+// removed yet. Listing by rule ID returns one entry, so no paging is needed, and the API takes no
+// listall or project ID.
+func (lb *loadBalancer) liveStickinessPolicies(lbRuleID string) ([]cloudstack.LBStickinessPolicyStickinesspolicy, error) {
+ p := lb.LoadBalancer.NewListLBStickinessPoliciesParams()
+ p.SetLbruleid(lbRuleID)
+ l, err := lb.LoadBalancer.ListLBStickinessPolicies(p)
+ if err != nil {
+ return nil, fmt.Errorf("error listing stickiness policies: %w", err)
+ }
+
+ var live []cloudstack.LBStickinessPolicyStickinesspolicy
+ for _, wrapper := range l.LBStickinessPolicies {
+ for _, policy := range wrapper.Stickinesspolicy {
+ if !strings.EqualFold(policy.State, "Revoked") {
+ live = append(live, policy)
+ }
+ }
+ }
+
+ return live, nil
+}
+
+// createStickinessPolicy creates the spec's policy on a rule, marked as the controller's.
+// CloudStack revokes the rule's other policies when it applies the new one.
+func (lb *loadBalancer) createStickinessPolicy(lbRuleName string, lbRuleID string, spec *stickinessSpec) error {
+ klog.V(4).Infof("Creating stickiness policy %v %v on load balancer rule %v", spec.method, spec.params, lbRuleName)
+ p := lb.LoadBalancer.NewCreateLBStickinessPolicyParams(lbRuleID, spec.method, lbRuleName)
+ p.SetDescription(stickinessPolicyDescription)
+ p.SetParam(spec.params)
+
+ if _, err := lb.LoadBalancer.CreateLBStickinessPolicy(p); err != nil {
+ return fmt.Errorf("error creating stickiness policy: %v", err)
+ }
+
+ return nil
+}
+
+// deleteStickinessPolicy deletes a stickiness policy.
+func (lb *loadBalancer) deleteStickinessPolicy(lbRuleName string, stickinessPolicyID string) error {
+ klog.V(4).Infof("Deleting stickiness policy %v from load balancer rule %v", stickinessPolicyID, lbRuleName)
+ p := lb.LoadBalancer.NewDeleteLBStickinessPolicyParams(stickinessPolicyID)
+
+ if _, err := lb.LoadBalancer.DeleteLBStickinessPolicy(p); err != nil {
+ return fmt.Errorf("error deleting stickiness policy %v: %v", stickinessPolicyID, err)
+ }
+
return nil
}
diff --git a/cloudstack_loadbalancer_test.go b/cloudstack_loadbalancer_test.go
index 7933fcc..24e8a14 100644
--- a/cloudstack_loadbalancer_test.go
+++ b/cloudstack_loadbalancer_test.go
@@ -22,6 +22,7 @@
import (
"context"
"fmt"
+ "maps"
"reflect"
"slices"
"sort"
@@ -4619,8 +4620,14 @@
vm *cloudstack.MockVirtualMachineServiceIface
network *cloudstack.MockNetworkServiceIface
firewall *cloudstack.MockFirewallServiceIface
+ address *cloudstack.MockAddressServiceIface
service *corev1.Service
nodes []*corev1.Node
+
+ instances []*cloudstack.VirtualMachine
+ policies map[string][]cloudstack.LBStickinessPolicyStickinesspolicy
+ policyErr error
+ policyLookups []string
}
func newEnsureLBTestEnv(ctrl *gomock.Controller, annotations map[string]string, ports []corev1.ServicePort) *ensureLBTestEnv {
@@ -4629,6 +4636,7 @@
vm: cloudstack.NewMockVirtualMachineServiceIface(ctrl),
network: cloudstack.NewMockNetworkServiceIface(ctrl),
firewall: cloudstack.NewMockFirewallServiceIface(ctrl),
+ address: cloudstack.NewMockAddressServiceIface(ctrl),
}
e.cs = &CSCloud{
@@ -4637,6 +4645,7 @@
VirtualMachine: e.vm,
Network: e.network,
Firewall: e.firewall,
+ Address: e.address,
},
version: semver.Version{Major: 4, Minor: 22, Patch: 0},
}
@@ -4657,9 +4666,40 @@
e.nodes = []*corev1.Node{{ObjectMeta: metav1.ObjectMeta{Name: "node-1"}}}
+ // A sync counts the hosts of each adopted rule and lists its stickiness policies. By default
+ // the rule already has the node and no policies, so tests that change neither work as before.
+ e.instances = []*cloudstack.VirtualMachine{{Id: "vm-1"}}
+ e.policies = map[string][]cloudstack.LBStickinessPolicyStickinesspolicy{}
+ e.lb.EXPECT().NewListLoadBalancerRuleInstancesParams(gomock.Any()).DoAndReturn(func(string) *cloudstack.ListLoadBalancerRuleInstancesParams {
+ return &cloudstack.ListLoadBalancerRuleInstancesParams{}
+ }).AnyTimes()
+ e.lb.EXPECT().ListLoadBalancerRuleInstances(gomock.Any()).DoAndReturn(e.listRuleInstances).AnyTimes()
+ e.lb.EXPECT().NewListLBStickinessPoliciesParams().DoAndReturn(func() *cloudstack.ListLBStickinessPoliciesParams {
+ return &cloudstack.ListLBStickinessPoliciesParams{}
+ }).AnyTimes()
+ e.lb.EXPECT().ListLBStickinessPolicies(gomock.Any()).DoAndReturn(e.listStickinessPolicies).AnyTimes()
+
return e
}
+// listRuleInstances answers a rule instance lookup with e.instances.
+func (e *ensureLBTestEnv) listRuleInstances(*cloudstack.ListLoadBalancerRuleInstancesParams) (*cloudstack.ListLoadBalancerRuleInstancesResponse, error) {
+ return &cloudstack.ListLoadBalancerRuleInstancesResponse{Count: len(e.instances), LoadBalancerRuleInstances: e.instances}, nil
+}
+
+// listStickinessPolicies answers a policy lookup from e.policies, one entry per rule like
+// CloudStack, and records which rule was looked up.
+func (e *ensureLBTestEnv) listStickinessPolicies(p *cloudstack.ListLBStickinessPoliciesParams) (*cloudstack.ListLBStickinessPoliciesResponse, error) {
+ ruleID, _ := p.GetLbruleid()
+ e.policyLookups = append(e.policyLookups, ruleID)
+ if e.policyErr != nil {
+ return nil, e.policyErr
+ }
+ wrapper := &cloudstack.LBStickinessPolicy{Lbruleid: ruleID, Stickinesspolicy: e.policies[ruleID]}
+
+ return &cloudstack.ListLBStickinessPoliciesResponse{Count: 1, LBStickinessPolicies: []*cloudstack.LBStickinessPolicy{wrapper}}, nil
+}
+
// expectHosts registers the node lookup every run performs before resolving rules.
func (e *ensureLBTestEnv) expectHosts() {
e.vm.EXPECT().NewListVirtualMachinesParams().Return(&cloudstack.ListVirtualMachinesParams{})
@@ -4673,11 +4713,31 @@
// expectHostsAndNetwork registers the host and network lookups every run performs.
func (e *ensureLBTestEnv) expectHostsAndNetwork() {
- e.expectHosts()
- e.network.EXPECT().GetNetworkByID("net-1", gomock.Any()).Return(&cloudstack.Network{
+ e.expectHostsAndNetworkWith(&cloudstack.Network{
Id: "net-1",
Service: []cloudstack.NetworkServiceInternal{{Name: "Firewall"}},
- }, 1, nil)
+ })
+}
+
+// expectHostsAndNetworkWith registers the host lookup and answers the network lookup with network.
+func (e *ensureLBTestEnv) expectHostsAndNetworkWith(network *cloudstack.Network) {
+ e.expectHosts()
+ e.network.EXPECT().GetNetworkByID("net-1", gomock.Any()).Return(network, 1, nil)
+}
+
+// expectRules answers the run's load balancer rule listing with rules.
+func (e *ensureLBTestEnv) expectRules(rules ...*cloudstack.LoadBalancerRule) {
+ e.lb.EXPECT().NewListLoadBalancerRulesParams().Return(&cloudstack.ListLoadBalancerRulesParams{})
+ e.lb.EXPECT().ListLoadBalancerRules(gomock.Any()).Return(&cloudstack.ListLoadBalancerRulesResponse{
+ Count: len(rules),
+ LoadBalancerRules: rules,
+ }, nil)
+}
+
+// expectFirewallListings answers each of the run's firewall listings, one per port, with resp.
+func (e *ensureLBTestEnv) expectFirewallListings(resp *cloudstack.ListFirewallRulesResponse, times int) {
+ e.firewall.EXPECT().NewListFirewallRulesParams().Return(&cloudstack.ListFirewallRulesParams{}).Times(times)
+ e.firewall.EXPECT().ListFirewallRules(gomock.Any()).Return(resp, nil).Times(times)
}
func TestEnsureLoadBalancer(t *testing.T) {
@@ -4900,7 +4960,7 @@
movedPort80 := corev1.ServicePort{Port: 80, NodePort: 30001, Protocol: corev1.ProtocolTCP}
sctpPort := corev1.ServicePort{Port: 9000, NodePort: 30900, Protocol: corev1.ProtocolSCTP}
env := newEnsureLBTestEnv(ctrl, nil, []corev1.ServicePort{movedPort80, sctpPort})
- env.expectHosts()
+ env.expectHostsAndNetwork()
// No delete expectation: the existing port 80 rule must survive.
gomock.InOrder(
@@ -5671,3 +5731,793 @@
}
})
}
+
+// virtualRouterStickinessCapability is the SupportedStickinessMethods capability of a VirtualRouter
+// network, copied from the 4.22.1 simulator. It has not changed since 4.16.
+const virtualRouterStickinessCapability = `[{"methodname":"LbCookie","paramlist":[{"paramname":"cookie-name","required":false,"isflag":false,"description":" "},{"paramname":"mode","required":false,"isflag":false,"description":" "},{"paramname":"nocache","required":false,"isflag":true,"description":" "},{"paramname":"indirect","required":false,"isflag":true,"description":" "},{"paramname":"postonly","required":false,"isflag":true,"description":" "},{"paramname":"domain","required":false,"isflag":false,"description":" "}],"description":"This is loadbalancer cookie based stickiness method."},{"methodname":"AppCookie","paramlist":[{"paramname":"cookie-name","required":false,"isflag":false,"description":" "},{"paramname":"length","required":false,"isflag":false,"description":" "},{"paramname":"holdtime","required":false,"isflag":false,"description":" "},{"paramname":"request-learn","required":false,"isflag":true,"description":" "},{"paramname":"prefix","required":false,"isflag":true,"description":" "},{"paramname":"mode","required":false,"isflag":false,"description":" "}],"description":"This is App session based sticky method."},{"methodname":"SourceBased","paramlist":[{"paramname":"tablesize","required":false,"isflag":false,"description":" "},{"paramname":"expire","required":false,"isflag":false,"description":" "}],"description":"This is source based Stickiness method, it can be used for any type of protocol."}]`
+
+// virtualRouterNetwork returns a network that lists the VirtualRouter's stickiness methods.
+func virtualRouterNetwork() *cloudstack.Network {
+ return &cloudstack.Network{
+ Id: "net-1",
+ Service: []cloudstack.NetworkServiceInternal{
+ {Name: "Firewall"},
+ {Name: "Lb", Capability: []cloudstack.NetworkServiceInternalCapability{
+ {Name: "SupportedStickinessMethods", Value: virtualRouterStickinessCapability},
+ }},
+ },
+ }
+}
+
+// stickinessService builds a Service with the given ports and stickiness annotations. An empty
+// value leaves the annotation out.
+func stickinessService(method, params string, servicePorts ...corev1.ServicePort) *corev1.Service {
+ annotations := map[string]string{}
+ for key, value := range map[string]string{
+ ServiceAnnotationLoadBalancerStickinessMethodName: method,
+ ServiceAnnotationLoadBalancerStickinessMethodParam: params,
+ } {
+ if value != "" {
+ annotations[key] = value
+ }
+ }
+
+ return &corev1.Service{
+ ObjectMeta: metav1.ObjectMeta{Name: "test-service", Namespace: "default", Annotations: annotations},
+ Spec: corev1.ServiceSpec{Ports: servicePorts},
+ }
+}
+
+func TestParseStickinessSpec(t *testing.T) {
+ http := corev1.ServicePort{Name: "http", Port: 80, Protocol: corev1.ProtocolTCP}
+ https := corev1.ServicePort{Name: "https", Port: 443, Protocol: corev1.ProtocolTCP}
+ dns := corev1.ServicePort{Name: "dns", Port: 53, Protocol: corev1.ProtocolUDP}
+ declaredHTTP := withAppProtocol(http, "http")
+ declaredHTTPS := withAppProtocol(https, "https")
+
+ tests := []struct {
+ name string
+ service *corev1.Service
+ wantErr string
+ wantNil bool
+ wantParams map[string]string
+ wantPorts []int32
+ }{
+ {name: "no method means no stickiness", service: stickinessService("", "cookie-name=X", http), wantNil: true},
+ {name: "LbCookie goes only on ports declaring appProtocol http", service: stickinessService("LbCookie", "", declaredHTTP, declaredHTTPS, dns), wantParams: map[string]string{}, wantPorts: []int32{80}},
+ {name: "appProtocol is matched ignoring case", service: stickinessService("LbCookie", "", withAppProtocol(http, "HTTP"), https), wantParams: map[string]string{}, wantPorts: []int32{80}},
+ {name: "LbCookie needs appProtocol http even on a single TCP port", service: stickinessService("LbCookie", "", http, dns), wantErr: "appProtocol: http"},
+ {name: "LbCookie on several undeclared TCP ports needs appProtocol", service: stickinessService("LbCookie", "", http, https), wantErr: "appProtocol: http"},
+ {name: "LbCookie on a port declaring https is rejected", service: stickinessService("LbCookie", "", declaredHTTPS), wantErr: "appProtocol: http"},
+ {name: "SourceBased goes on every TCP port whatever appProtocol says", service: stickinessService("SourceBased", "", declaredHTTP, declaredHTTPS, dns), wantParams: map[string]string{}, wantPorts: []int32{80, 443}},
+ {name: "AppCookie is rejected", service: stickinessService("appcookie", "", http), wantErr: "appsession"},
+ {name: "params are trimmed and empty entries dropped", service: stickinessService("LbCookie", ", cookie-name = SERVERID ,nocache=true,", declaredHTTP), wantParams: map[string]string{"cookie-name": "SERVERID", "nocache": "true"}, wantPorts: []int32{80}},
+ {name: "an empty value is rejected with a hint for flags", service: stickinessService("LbCookie", "nocache=", declaredHTTP), wantErr: "nocache=true"},
+ {name: "an entry without = is rejected", service: stickinessService("LbCookie", "nocache", declaredHTTP), wantErr: "expected key=value"},
+ {name: "= in a value is rejected", service: stickinessService("LbCookie", "cookie-name=a=b", declaredHTTP), wantErr: "are not allowed"},
+ {name: "& in a value is rejected", service: stickinessService("LbCookie", "domain=x&y", declaredHTTP), wantErr: "are not allowed"},
+ {name: "a quote in a value is rejected", service: stickinessService("LbCookie", `cookie-name=SERVER"ID`, declaredHTTP), wantErr: "are not allowed"},
+ {name: "# in a value is rejected", service: stickinessService("LbCookie", "domain=a#b", declaredHTTP), wantErr: "are not allowed"},
+ {name: "a backslash in a value is rejected", service: stickinessService("LbCookie", `cookie-name=a\b`, declaredHTTP), wantErr: "are not allowed"},
+ {name: "whitespace in a value is rejected", service: stickinessService("LbCookie", "cookie-name=my cookie", declaredHTTP), wantErr: "are not allowed"},
+ {name: "whitespace in a key is rejected", service: stickinessService("LbCookie", "cookie name=x", declaredHTTP), wantErr: "are not allowed"},
+ {name: "a key set twice is rejected whatever its case", service: stickinessService("LbCookie", "mode=insert,MODE=rewrite", declaredHTTP), wantErr: "set twice"},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ spec, err := parseStickinessSpec(tt.service)
+ if tt.wantErr != "" {
+ if err == nil || !strings.Contains(err.Error(), tt.wantErr) || !strings.Contains(err.Error(), "stickiness") {
+ t.Fatalf("error = %v, want it to mention %q and stickiness", err, tt.wantErr)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if tt.wantNil {
+ if spec != nil {
+ t.Fatalf("spec = %+v, want nil", spec)
+ }
+ return
+ }
+ if !reflect.DeepEqual(spec.params, tt.wantParams) {
+ t.Errorf("params = %v, want %v", spec.params, tt.wantParams)
+ }
+ gotPorts := slices.Sorted(maps.Keys(spec.ports))
+ if !slices.Equal(gotPorts, tt.wantPorts) {
+ t.Errorf("ports = %v, want %v", gotPorts, tt.wantPorts)
+ }
+ })
+ }
+}
+
+func TestStickinessSpecWants(t *testing.T) {
+ spec := &stickinessSpec{method: "LbCookie", ports: map[int32]bool{80: true}}
+ tests := []struct {
+ name string
+ spec *stickinessSpec
+ port corev1.ServicePort
+ want bool
+ }{
+ {name: "a selected TCP port", spec: spec, port: corev1.ServicePort{Port: 80, Protocol: corev1.ProtocolTCP}, want: true},
+ {name: "an unselected TCP port", spec: spec, port: corev1.ServicePort{Port: 443, Protocol: corev1.ProtocolTCP}},
+ {name: "a UDP port with a selected number", spec: spec, port: corev1.ServicePort{Port: 80, Protocol: corev1.ProtocolUDP}},
+ {name: "no spec", port: corev1.ServicePort{Port: 80, Protocol: corev1.ProtocolTCP}},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ if got := tt.spec.wants(tt.port); got != tt.want {
+ t.Errorf("wants = %v, want %v", got, tt.want)
+ }
+ })
+ }
+}
+
+func TestSupportedStickinessMethods(t *testing.T) {
+ if got := supportedStickinessMethods(virtualRouterNetwork()); len(got) != 3 || got[0].Name != "LbCookie" || !got[0].Params[2].IsFlag {
+ t.Errorf("methods = %+v, want the VirtualRouter's three with nocache as a flag", got)
+ }
+ if got := supportedStickinessMethods(&cloudstack.Network{Service: []cloudstack.NetworkServiceInternal{{Name: "Firewall"}}}); got != nil {
+ t.Errorf("methods = %+v, want nil for a network without the Lb service", got)
+ }
+ unreadable := &cloudstack.Network{Service: []cloudstack.NetworkServiceInternal{{Name: "Lb", Capability: []cloudstack.NetworkServiceInternalCapability{
+ {Name: "SupportedStickinessMethods", Value: "not json"},
+ }}}}
+ if got := supportedStickinessMethods(unreadable); got != nil {
+ t.Errorf("methods = %+v, want nil for an unreadable capability", got)
+ }
+}
+
+func TestStickinessSpecNormalise(t *testing.T) {
+ methods := supportedStickinessMethods(virtualRouterNetwork())
+ tests := []struct {
+ name string
+ method string
+ params map[string]string
+ methods []stickinessMethod
+ wantMethod string
+ wantParams map[string]string
+ wantErr string
+ }{
+ {name: "method and param names take the advertised spelling", method: "lbcookie", params: map[string]string{"Cookie-Name": "SERVERID"}, methods: methods, wantMethod: "LbCookie", wantParams: map[string]string{"cookie-name": "SERVERID"}},
+ {name: "a true flag is sent as true", method: "LbCookie", params: map[string]string{"nocache": "1"}, methods: methods, wantMethod: "LbCookie", wantParams: map[string]string{"nocache": "true"}},
+ {name: "a false flag is left out", method: "LbCookie", params: map[string]string{"nocache": "false", "indirect": "True"}, methods: methods, wantMethod: "LbCookie", wantParams: map[string]string{"indirect": "true"}},
+ {name: "a flag that is not a boolean is rejected", method: "LbCookie", params: map[string]string{"nocache": "maybe"}, methods: methods, wantErr: "true or false"},
+ {name: "the cookie mode is lower-cased", method: "LbCookie", params: map[string]string{"mode": "INSERT"}, methods: methods, wantMethod: "LbCookie", wantParams: map[string]string{"mode": "insert"}},
+ {name: "an unknown cookie mode is rejected", method: "LbCookie", params: map[string]string{"mode": "bogus"}, methods: methods, wantErr: "insert, rewrite, prefix"},
+ {name: "an unknown method lists the supported ones", method: "NoSuchMethod", methods: methods, wantErr: "supported methods: LbCookie, SourceBased"},
+ {name: "an unknown param lists the method's", method: "SourceBased", params: map[string]string{"cookie-name": "x"}, methods: methods, wantErr: "tablesize, expire"},
+ {name: "a missing required param is rejected", method: "Custom", methods: []stickinessMethod{{Name: "Custom", Params: []stickinessMethodParam{{Name: "key", Required: true}}}}, wantErr: "needs parameter key"},
+ {name: "without advertised methods nothing changes", method: "lbcookie", params: map[string]string{"nocache": "false"}, wantMethod: "lbcookie", wantParams: map[string]string{"nocache": "false"}},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ spec := &stickinessSpec{method: tt.method, params: tt.params}
+ if spec.params == nil {
+ spec.params = map[string]string{}
+ }
+ err := spec.normalise(tt.methods)
+ if tt.wantErr != "" {
+ if err == nil || !strings.Contains(err.Error(), tt.wantErr) || !strings.Contains(err.Error(), "stickiness") {
+ t.Fatalf("error = %v, want it to mention %q and stickiness", err, tt.wantErr)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if spec.method != tt.wantMethod || !maps.Equal(spec.params, tt.wantParams) {
+ t.Errorf("spec = %s %v, want %s %v", spec.method, spec.params, tt.wantMethod, tt.wantParams)
+ }
+ })
+ }
+
+ var none *stickinessSpec
+ if err := none.normalise(methods); err != nil {
+ t.Errorf("normalising no spec: %v", err)
+ }
+}
+
+func TestStickinessChanges(t *testing.T) {
+ spec := &stickinessSpec{method: "LbCookie", params: map[string]string{"cookie-name": "SERVERID"}}
+ ours := func(id string, params map[string]string) cloudstack.LBStickinessPolicyStickinesspolicy {
+ return cloudstack.LBStickinessPolicyStickinesspolicy{Id: id, Methodname: "LbCookie", Params: params, Description: stickinessPolicyDescription}
+ }
+ foreign := func(id string) cloudstack.LBStickinessPolicyStickinesspolicy {
+ return cloudstack.LBStickinessPolicyStickinesspolicy{Id: id, Methodname: "LbCookie", Params: map[string]string{"cookie-name": "SERVERID"}}
+ }
+ matching := ours("ours", map[string]string{"cookie-name": "SERVERID"})
+
+ tests := []struct {
+ name string
+ live []cloudstack.LBStickinessPolicyStickinesspolicy
+ want bool
+ wantCreate bool
+ wantDeletes []string
+ wantErr string
+ }{
+ {name: "a selected rule without a policy gets one", want: true, wantCreate: true},
+ {name: "a selected rule with our matching policy is left alone", live: append([]cloudstack.LBStickinessPolicyStickinesspolicy{}, matching), want: true},
+ {name: "a changed policy of ours is replaced by one create", live: append([]cloudstack.LBStickinessPolicyStickinesspolicy{}, ours("ours", map[string]string{"cookie-name": "OLD"})), want: true, wantCreate: true},
+ {name: "an identical hand-made policy on a selected rule is replaced", live: append([]cloudstack.LBStickinessPolicyStickinesspolicy{}, foreign("manual")), want: true, wantCreate: true},
+ {name: "ours and a hand-made policy are reported and nothing is deleted", live: append([]cloudstack.LBStickinessPolicyStickinesspolicy{}, matching, foreign("manual")), want: true, wantErr: "remove the extra ones"},
+ {name: "two hand-made policies are reported and nothing is deleted", live: append([]cloudstack.LBStickinessPolicyStickinesspolicy{}, foreign("a"), foreign("b")), want: true, wantErr: "remove the extra ones"},
+ {name: "an unselected rule loses only our policies", live: append([]cloudstack.LBStickinessPolicyStickinesspolicy{}, matching, foreign("manual"))},
+ {name: "an unselected rule without policies is left alone"},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ wantSpec := spec
+ if !tt.want {
+ wantSpec = nil
+ }
+ create, deletes, err := stickinessChanges(tt.live, wantSpec, tt.want)
+ if tt.wantErr != "" {
+ if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
+ t.Fatalf("error = %v, want it to mention %q", err, tt.wantErr)
+ }
+ if create || len(deletes) > 0 {
+ t.Errorf("create, deletes = %v, %v; want no changes with the error", create, deletes)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ wantDeletes := tt.wantDeletes
+ if !tt.want && len(tt.live) > 0 {
+ wantDeletes = []string{"ours"}
+ }
+ if create != tt.wantCreate || !slices.Equal(deletes, wantDeletes) {
+ t.Errorf("create, deletes = %v, %v; want %v, %v", create, deletes, tt.wantCreate, wantDeletes)
+ }
+ })
+ }
+}
+
+func TestLiveStickinessPolicies(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ t.Cleanup(ctrl.Finish)
+ mockLB := cloudstack.NewMockLoadBalancerServiceIface(ctrl)
+ params := &cloudstack.ListLBStickinessPoliciesParams{}
+ mockLB.EXPECT().NewListLBStickinessPoliciesParams().Return(params)
+ mockLB.EXPECT().ListLBStickinessPolicies(params).Return(&cloudstack.ListLBStickinessPoliciesResponse{
+ Count: 1,
+ LBStickinessPolicies: []*cloudstack.LBStickinessPolicy{{Stickinesspolicy: []cloudstack.LBStickinessPolicyStickinesspolicy{
+ {Id: "revoked", State: "Revoked"},
+ {Id: "live"},
+ }}},
+ }, nil)
+
+ lb := &loadBalancer{CloudStackClient: &cloudstack.CloudStackClient{LoadBalancer: mockLB}}
+ live, err := lb.liveStickinessPolicies("rule-1")
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if len(live) != 1 || live[0].Id != "live" {
+ t.Errorf("live = %+v, want only the live policy", live)
+ }
+ if ruleID, _ := params.GetLbruleid(); ruleID != "rule-1" {
+ t.Errorf("lbruleid = %q, want rule-1", ruleID)
+ }
+}
+
+// stickinessTestRule returns an up-to-date TCP rule of the "atestuid" load balancer for a port.
+func stickinessTestRule(id string, port corev1.ServicePort) *cloudstack.LoadBalancerRule {
+ return &cloudstack.LoadBalancerRule{
+ Id: id,
+ Name: fmt.Sprintf("atestuid-tcp-%d", port.Port),
+ Publicip: "10.0.0.1",
+ Publicipid: "ip-1",
+ Publicport: fmt.Sprint(port.Port),
+ Privateport: fmt.Sprint(port.NodePort),
+ Cidrlist: defaultAllowedCIDR,
+ Algorithm: "roundrobin",
+ Protocol: "tcp",
+ Networkid: "net-1",
+ }
+}
+
+// firewallRulesFor returns firewall rules that already open the given ports to everyone.
+func firewallRulesFor(ports ...corev1.ServicePort) *cloudstack.ListFirewallRulesResponse {
+ resp := &cloudstack.ListFirewallRulesResponse{}
+ for _, port := range ports {
+ resp.FirewallRules = append(resp.FirewallRules, &cloudstack.FirewallRule{
+ Id: fmt.Sprintf("fw-%d", port.Port),
+ Protocol: "tcp",
+ Startport: int(port.Port),
+ Endport: int(port.Port),
+ Cidrlist: defaultAllowedCIDR,
+ })
+ }
+ resp.Count = len(resp.FirewallRules)
+
+ return resp
+}
+
+// stickinessAnnotations returns the stickiness annotations that are not empty.
+func stickinessAnnotations(method, params string) map[string]string {
+ return stickinessService(method, params).Annotations
+}
+
+// withAppProtocol returns the port declaring the given application protocol.
+func withAppProtocol(port corev1.ServicePort, appProtocol string) corev1.ServicePort {
+ port.AppProtocol = &appProtocol
+ return port
+}
+
+func TestEnsureLoadBalancerStickiness(t *testing.T) {
+ http := withAppProtocol(corev1.ServicePort{Name: "http", Port: 80, NodePort: 30000, Protocol: corev1.ProtocolTCP}, "http")
+ alt := corev1.ServicePort{Name: "alt", Port: 8080, NodePort: 30080, Protocol: corev1.ProtocolTCP}
+ cookie := map[string]string{"cookie-name": "SERVERID"}
+ ours := func(params map[string]string) cloudstack.LBStickinessPolicyStickinesspolicy {
+ return cloudstack.LBStickinessPolicyStickinesspolicy{Id: "policy-ours", Name: "atestuid-tcp-80", Methodname: "LbCookie", Params: params, Description: stickinessPolicyDescription}
+ }
+ manual := func(id string) cloudstack.LBStickinessPolicyStickinesspolicy {
+ return cloudstack.LBStickinessPolicyStickinesspolicy{Id: id, Name: "manual", Methodname: "LbCookie", Params: cookie}
+ }
+ newEnv := func(t *testing.T, annotations map[string]string, ports ...corev1.ServicePort) *ensureLBTestEnv {
+ ctrl := gomock.NewController(t)
+ t.Cleanup(ctrl.Finish)
+ return newEnsureLBTestEnv(ctrl, annotations, ports)
+ }
+ ensure := func(env *ensureLBTestEnv) error {
+ _, err := env.cs.EnsureLoadBalancer(context.TODO(), "test-cluster", env.service, env.nodes)
+ return err
+ }
+ expectCreate := func(env *ensureLBTestEnv, ruleID, method, ruleName string, createErr error) *cloudstack.CreateLBStickinessPolicyParams {
+ params := &cloudstack.CreateLBStickinessPolicyParams{}
+ env.lb.EXPECT().NewCreateLBStickinessPolicyParams(ruleID, method, ruleName).Return(params)
+ env.lb.EXPECT().CreateLBStickinessPolicy(params).Return(&cloudstack.CreateLBStickinessPolicyResponse{}, createErr)
+ return params
+ }
+ expectDelete := func(env *ensureLBTestEnv, policyID string) {
+ params := &cloudstack.DeleteLBStickinessPolicyParams{}
+ env.lb.EXPECT().NewDeleteLBStickinessPolicyParams(policyID).Return(params)
+ env.lb.EXPECT().DeleteLBStickinessPolicy(params).Return(&cloudstack.DeleteLBStickinessPolicyResponse{}, nil)
+ }
+ sentParams := func(t *testing.T, params *cloudstack.CreateLBStickinessPolicyParams) map[string]string {
+ t.Helper()
+ if description, _ := params.GetDescription(); description != stickinessPolicyDescription {
+ t.Errorf("policy description = %q, want the controller's marker", description)
+ }
+ sent, _ := params.GetParam()
+ return sent
+ }
+
+ t.Run("a policy is created once the rule and its firewall are in place", func(t *testing.T) {
+ env := newEnv(t, stickinessAnnotations("lbcookie", "Cookie-Name=SERVERID, nocache=1"), http)
+ env.expectHostsAndNetworkWith(virtualRouterNetwork())
+ env.expectRules(stickinessTestRule("rule-1", http))
+ create := &cloudstack.CreateLBStickinessPolicyParams{}
+ gomock.InOrder(
+ env.firewall.EXPECT().NewListFirewallRulesParams().Return(&cloudstack.ListFirewallRulesParams{}),
+ env.firewall.EXPECT().ListFirewallRules(gomock.Any()).Return(firewallRulesFor(http), nil),
+ env.lb.EXPECT().NewCreateLBStickinessPolicyParams("rule-1", "LbCookie", "atestuid-tcp-80").Return(create),
+ env.lb.EXPECT().CreateLBStickinessPolicy(create).Return(&cloudstack.CreateLBStickinessPolicyResponse{}, nil),
+ )
+
+ if err := ensure(env); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if sent := sentParams(t, create); !maps.Equal(sent, map[string]string{"cookie-name": "SERVERID", "nocache": "true"}) {
+ t.Errorf("sent params = %v, want the normalised cookie-name and nocache=true", sent)
+ }
+ if !slices.Equal(env.policyLookups, []string{"rule-1"}) {
+ t.Errorf("policy lookups = %v, want one for rule-1", env.policyLookups)
+ }
+ })
+
+ t.Run("the controller's matching policy is left alone", func(t *testing.T) {
+ env := newEnv(t, stickinessAnnotations("LbCookie", "cookie-name=SERVERID"), http)
+ env.expectHostsAndNetworkWith(virtualRouterNetwork())
+ env.expectRules(stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 1)
+ env.policies["rule-1"] = []cloudstack.LBStickinessPolicyStickinesspolicy{ours(cookie)}
+
+ if err := ensure(env); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ })
+
+ t.Run("a changed parameter is swapped by one create, with no delete", func(t *testing.T) {
+ env := newEnv(t, stickinessAnnotations("LbCookie", "cookie-name=SERVERID"), http)
+ env.expectHostsAndNetworkWith(virtualRouterNetwork())
+ env.expectRules(stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 1)
+ env.policies["rule-1"] = []cloudstack.LBStickinessPolicyStickinesspolicy{ours(map[string]string{"cookie-name": "OLD"})}
+ create := expectCreate(env, "rule-1", "LbCookie", "atestuid-tcp-80", nil)
+
+ if err := ensure(env); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if sent := sentParams(t, create); !maps.Equal(sent, cookie) {
+ t.Errorf("sent params = %v, want %v", sent, cookie)
+ }
+ })
+
+ t.Run("a hand-made policy on a selected port is replaced by the controller's", func(t *testing.T) {
+ env := newEnv(t, stickinessAnnotations("LbCookie", "cookie-name=SERVERID"), http)
+ env.expectHostsAndNetworkWith(virtualRouterNetwork())
+ env.expectRules(stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 1)
+ env.policies["rule-1"] = []cloudstack.LBStickinessPolicyStickinesspolicy{manual("policy-manual")}
+ expectCreate(env, "rule-1", "LbCookie", "atestuid-tcp-80", nil)
+
+ if err := ensure(env); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ })
+
+ t.Run("a hand-made policy on an unannotated Service is kept", func(t *testing.T) {
+ env := newEnv(t, nil, http)
+ env.expectHostsAndNetwork()
+ env.expectRules(stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 1)
+ env.policies["rule-1"] = []cloudstack.LBStickinessPolicyStickinesspolicy{manual("policy-manual")}
+
+ if err := ensure(env); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ })
+
+ t.Run("removing the annotation deletes only the controller's policy", func(t *testing.T) {
+ env := newEnv(t, nil, http)
+ env.expectHostsAndNetwork()
+ env.expectRules(stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 1)
+ env.policies["rule-1"] = []cloudstack.LBStickinessPolicyStickinesspolicy{ours(cookie), manual("policy-manual")}
+ expectDelete(env, "policy-ours")
+
+ if err := ensure(env); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ })
+
+ t.Run("LbCookie follows appProtocol http and leaves other ports", func(t *testing.T) {
+ env := newEnv(t, stickinessAnnotations("LbCookie", "cookie-name=SERVERID"), withAppProtocol(http, "https"), withAppProtocol(alt, "http"))
+ env.expectHostsAndNetworkWith(virtualRouterNetwork())
+ env.expectRules(stickinessTestRule("rule-1", http), stickinessTestRule("rule-2", alt))
+ env.expectFirewallListings(firewallRulesFor(http, alt), 2)
+ env.policies["rule-1"] = []cloudstack.LBStickinessPolicyStickinesspolicy{ours(cookie)}
+ expectDelete(env, "policy-ours")
+ expectCreate(env, "rule-2", "LbCookie", "atestuid-tcp-8080", nil)
+
+ if err := ensure(env); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ })
+
+ t.Run("LbCookie on ports declaring no appProtocol fails before any CloudStack call", func(t *testing.T) {
+ undeclared := http
+ undeclared.AppProtocol = nil
+ env := newEnv(t, stickinessAnnotations("LbCookie", ""), undeclared, alt)
+
+ if err := ensure(env); err == nil || !strings.Contains(err.Error(), "appProtocol: http") {
+ t.Fatalf("error = %v, want it to ask for appProtocol: http", err)
+ }
+ })
+
+ t.Run("an invalid parameter fails before any CloudStack call", func(t *testing.T) {
+ env := newEnv(t, stickinessAnnotations("LbCookie", "cookie-name="), http)
+
+ if err := ensure(env); err == nil || !strings.Contains(err.Error(), "missing value") {
+ t.Fatalf("error = %v, want the empty value rejected", err)
+ }
+ })
+
+ t.Run("a method the network does not offer fails before any write", func(t *testing.T) {
+ env := newEnv(t, stickinessAnnotations("NoSuchMethod", ""), http)
+ env.expectHostsAndNetworkWith(virtualRouterNetwork())
+ env.expectRules(stickinessTestRule("rule-1", http))
+
+ if err := ensure(env); err == nil || !strings.Contains(err.Error(), "supported methods: LbCookie, SourceBased") {
+ t.Fatalf("error = %v, want it to list the supported methods", err)
+ }
+ })
+
+ t.Run("a false flag is not sent", func(t *testing.T) {
+ env := newEnv(t, stickinessAnnotations("LbCookie", "cookie-name=SERVERID,nocache=false"), http)
+ env.expectHostsAndNetworkWith(virtualRouterNetwork())
+ env.expectRules(stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 1)
+ create := expectCreate(env, "rule-1", "LbCookie", "atestuid-tcp-80", nil)
+
+ if err := ensure(env); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if sent := sentParams(t, create); !maps.Equal(sent, cookie) {
+ t.Errorf("sent params = %v, want %v without nocache", sent, cookie)
+ }
+ })
+
+ t.Run("without the capability parameters pass through unchanged", func(t *testing.T) {
+ env := newEnv(t, stickinessAnnotations("LbCookie", "nocache=false"), http)
+ env.expectHostsAndNetwork()
+ env.expectRules(stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 1)
+ create := expectCreate(env, "rule-1", "LbCookie", "atestuid-tcp-80", nil)
+
+ if err := ensure(env); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if sent := sentParams(t, create); !maps.Equal(sent, map[string]string{"nocache": "false"}) {
+ t.Errorf("sent params = %v, want nocache=false passed through", sent)
+ }
+ })
+
+ t.Run("a rejected policy keeps the rule and its firewall opening", func(t *testing.T) {
+ env := newEnv(t, stickinessAnnotations("SourceBased", "tablesize=abc"), http)
+ env.expectHostsAndNetworkWith(virtualRouterNetwork())
+ env.expectRules(stickinessTestRule("rule-1", http))
+ create := &cloudstack.CreateLBStickinessPolicyParams{}
+ gomock.InOrder(
+ env.firewall.EXPECT().NewListFirewallRulesParams().Return(&cloudstack.ListFirewallRulesParams{}),
+ env.firewall.EXPECT().ListFirewallRules(gomock.Any()).Return(firewallRulesFor(http), nil),
+ env.lb.EXPECT().NewCreateLBStickinessPolicyParams("rule-1", "SourceBased", "atestuid-tcp-80").Return(create),
+ env.lb.EXPECT().CreateLBStickinessPolicy(create).Return(nil, fmt.Errorf("CloudStack API error 431: tablesize is not in size format")),
+ )
+
+ err := ensure(env)
+ if err == nil || !strings.Contains(err.Error(), "error creating stickiness policy") || !strings.Contains(err.Error(), "tablesize") {
+ t.Fatalf("error = %v, want CloudStack's rejection of the policy", err)
+ }
+ })
+
+ t.Run("every port's failure is reported together", func(t *testing.T) {
+ env := newEnv(t, stickinessAnnotations("SourceBased", "tablesize=200k"), http, alt)
+ env.expectHostsAndNetworkWith(virtualRouterNetwork())
+ env.expectRules(stickinessTestRule("rule-1", http), stickinessTestRule("rule-2", alt))
+ env.expectFirewallListings(firewallRulesFor(http, alt), 2)
+ expectCreate(env, "rule-1", "SourceBased", "atestuid-tcp-80", fmt.Errorf("router A unavailable"))
+ expectCreate(env, "rule-2", "SourceBased", "atestuid-tcp-8080", fmt.Errorf("router B unavailable"))
+
+ err := ensure(env)
+ if err == nil || !strings.Contains(err.Error(), "router A unavailable") || !strings.Contains(err.Error(), "router B unavailable") {
+ t.Fatalf("error = %v, want both ports' failures", err)
+ }
+ })
+
+ t.Run("a recreated rule gets its policy without a lookup", func(t *testing.T) {
+ moved := http
+ moved.NodePort = 30001
+ env := newEnv(t, stickinessAnnotations("LbCookie", "cookie-name=SERVERID"), moved)
+ env.expectHostsAndNetworkWith(virtualRouterNetwork())
+ env.expectRules(stickinessTestRule("rule-1", http))
+ create := &cloudstack.CreateLBStickinessPolicyParams{}
+ gomock.InOrder(
+ env.lb.EXPECT().NewDeleteLoadBalancerRuleParams("rule-1").Return(&cloudstack.DeleteLoadBalancerRuleParams{}),
+ env.lb.EXPECT().DeleteLoadBalancerRule(gomock.Any()).Return(&cloudstack.DeleteLoadBalancerRuleResponse{}, nil),
+ env.lb.EXPECT().NewCreateLoadBalancerRuleParams("roundrobin", "atestuid-tcp-80", 30001, 80).Return(&cloudstack.CreateLoadBalancerRuleParams{}),
+ env.lb.EXPECT().CreateLoadBalancerRule(gomock.Any()).Return(&cloudstack.CreateLoadBalancerRuleResponse{
+ Id: "rule-new", Name: "atestuid-tcp-80", Publicip: "10.0.0.1", Publicipid: "ip-1", Protocol: "tcp",
+ }, nil),
+ env.lb.EXPECT().NewAssignToLoadBalancerRuleParams("rule-new").Return(&cloudstack.AssignToLoadBalancerRuleParams{}),
+ env.lb.EXPECT().AssignToLoadBalancerRule(gomock.Any()).Return(&cloudstack.AssignToLoadBalancerRuleResponse{}, nil),
+ env.firewall.EXPECT().NewListFirewallRulesParams().Return(&cloudstack.ListFirewallRulesParams{}),
+ env.firewall.EXPECT().ListFirewallRules(gomock.Any()).Return(firewallRulesFor(http), nil),
+ env.lb.EXPECT().NewCreateLBStickinessPolicyParams("rule-new", "LbCookie", "atestuid-tcp-80").Return(create),
+ env.lb.EXPECT().CreateLBStickinessPolicy(create).Return(&cloudstack.CreateLBStickinessPolicyResponse{}, nil),
+ )
+
+ if err := ensure(env); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if len(env.policyLookups) != 0 {
+ t.Errorf("policy lookups = %v, want none for a rule created in this run", env.policyLookups)
+ }
+ })
+
+ t.Run("a proxy protocol toggle keeps the renamed rule's policy", func(t *testing.T) {
+ annotations := stickinessAnnotations("LbCookie", "cookie-name=SERVERID")
+ annotations[ServiceAnnotationLoadBalancerProxyProtocol] = "true"
+ env := newEnv(t, annotations, http)
+ env.expectHostsAndNetworkWith(virtualRouterNetwork())
+ env.expectRules(stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 1)
+ env.policies["rule-1"] = []cloudstack.LBStickinessPolicyStickinesspolicy{ours(cookie)}
+ update := &cloudstack.UpdateLoadBalancerRuleParams{}
+ env.lb.EXPECT().NewUpdateLoadBalancerRuleParams("rule-1").Return(update)
+ env.lb.EXPECT().UpdateLoadBalancerRule(update).Return(&cloudstack.UpdateLoadBalancerRuleResponse{}, nil)
+
+ if err := ensure(env); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if name, _ := update.GetName(); name != "atestuid-tcp-proxy-80" {
+ t.Errorf("updated name = %q, want atestuid-tcp-proxy-80", name)
+ }
+ })
+
+ t.Run("a prune failure still runs the stickiness phase", func(t *testing.T) {
+ env := newEnv(t, stickinessAnnotations("LbCookie", "cookie-name=SERVERID"), http)
+ env.expectHostsAndNetworkWith(virtualRouterNetwork())
+ obsolete := stickinessTestRule("rule-obsolete", alt)
+ env.expectRules(obsolete, stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 2)
+ env.lb.EXPECT().NewDeleteLoadBalancerRuleParams("rule-obsolete").Return(&cloudstack.DeleteLoadBalancerRuleParams{})
+ env.lb.EXPECT().DeleteLoadBalancerRule(gomock.Any()).Return(nil, fmt.Errorf("delete rule API error"))
+ expectCreate(env, "rule-1", "LbCookie", "atestuid-tcp-80", nil)
+
+ err := ensure(env)
+ if err == nil || !strings.Contains(err.Error(), "delete rule API error") {
+ t.Fatalf("error = %v, want the prune failure reported", err)
+ }
+ })
+
+ t.Run("a rule with two policies is reported and keeps both", func(t *testing.T) {
+ for name, live := range map[string][]cloudstack.LBStickinessPolicyStickinesspolicy{
+ "ours and a hand-made one": {ours(map[string]string{"cookie-name": "OLD"}), manual("b")},
+ "two hand-made ones": {manual("a"), manual("b")},
+ } {
+ t.Run(name, func(t *testing.T) {
+ env := newEnv(t, stickinessAnnotations("LbCookie", "cookie-name=SERVERID"), http)
+ env.expectHostsAndNetworkWith(virtualRouterNetwork())
+ env.expectRules(stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 1)
+ env.policies["rule-1"] = live
+
+ if err := ensure(env); err == nil || !strings.Contains(err.Error(), "remove the extra ones") {
+ t.Fatalf("error = %v, want the hint to remove the extra policies", err)
+ }
+ })
+ }
+ })
+
+ notAllowed := fmt.Errorf("CloudStack API error 432 (CSExceptionErrorCode: 9999): The API [listLBStickinessPolicies] does not exist or is not available for the account")
+
+ t.Run("an account without the stickiness APIs still syncs a Service without stickiness", func(t *testing.T) {
+ env := newEnv(t, nil, http)
+ env.expectHostsAndNetwork()
+ env.expectRules(stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 1)
+ env.policyErr = notAllowed
+
+ if err := ensure(env); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ })
+
+ t.Run("an account without the stickiness APIs gets a clear error when stickiness is set", func(t *testing.T) {
+ env := newEnv(t, stickinessAnnotations("LbCookie", "cookie-name=SERVERID"), http)
+ env.expectHostsAndNetworkWith(virtualRouterNetwork())
+ env.expectRules(stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 1)
+ env.policyErr = notAllowed
+
+ if err := ensure(env); err == nil || !strings.Contains(err.Error(), "stickiness needs the listLBStickinessPolicies") {
+ t.Fatalf("error = %v, want it to name the missing APIs", err)
+ }
+ })
+
+ t.Run("other policy lookup errors still fail the sync", func(t *testing.T) {
+ env := newEnv(t, nil, http)
+ env.expectHostsAndNetwork()
+ env.expectRules(stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 1)
+ env.policyErr = fmt.Errorf("CloudStack API error 530: internal error")
+
+ if err := ensure(env); err == nil || !strings.Contains(err.Error(), "internal error") {
+ t.Fatalf("error = %v, want the lookup failure", err)
+ }
+ })
+
+ t.Run("a revoked policy is ignored", func(t *testing.T) {
+ env := newEnv(t, stickinessAnnotations("LbCookie", "cookie-name=SERVERID"), http)
+ env.expectHostsAndNetworkWith(virtualRouterNetwork())
+ env.expectRules(stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 1)
+ revoked := ours(cookie)
+ revoked.State = "Revoked"
+ env.policies["rule-1"] = []cloudstack.LBStickinessPolicyStickinesspolicy{revoked}
+ expectCreate(env, "rule-1", "LbCookie", "atestuid-tcp-80", nil)
+
+ if err := ensure(env); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ })
+}
+
+func TestEnsureLoadBalancerRepairsEmptyRules(t *testing.T) {
+ http := corev1.ServicePort{Name: "http", Port: 80, NodePort: 30000, Protocol: corev1.ProtocolTCP}
+
+ t.Run("an adopted rule with no hosts gets them", func(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ t.Cleanup(ctrl.Finish)
+ env := newEnsureLBTestEnv(ctrl, nil, []corev1.ServicePort{http})
+ env.expectHostsAndNetwork()
+ env.expectRules(stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 1)
+ env.instances = nil
+ assign := &cloudstack.AssignToLoadBalancerRuleParams{}
+ env.lb.EXPECT().NewAssignToLoadBalancerRuleParams("rule-1").Return(assign)
+ env.lb.EXPECT().AssignToLoadBalancerRule(assign).Return(&cloudstack.AssignToLoadBalancerRuleResponse{}, nil)
+
+ if _, err := env.cs.EnsureLoadBalancer(context.TODO(), "test-cluster", env.service, env.nodes); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if ids, _ := assign.GetVirtualmachineids(); !slices.Equal(ids, []string{"vm-1"}) {
+ t.Errorf("assigned VMs = %v, want [vm-1]", ids)
+ }
+ })
+
+ t.Run("an adopted rule with other hosts is left to node sync", func(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ t.Cleanup(ctrl.Finish)
+ env := newEnsureLBTestEnv(ctrl, nil, []corev1.ServicePort{http})
+ env.expectHostsAndNetwork()
+ env.expectRules(stickinessTestRule("rule-1", http))
+ env.expectFirewallListings(firewallRulesFor(http), 1)
+ env.instances = []*cloudstack.VirtualMachine{{Id: "vm-2"}}
+
+ if _, err := env.cs.EnsureLoadBalancer(context.TODO(), "test-cluster", env.service, env.nodes); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ })
+}
+
+func TestEnsureLoadBalancerFreshIP(t *testing.T) {
+ http := corev1.ServicePort{Name: "http", Port: 80, NodePort: 30000, Protocol: corev1.ProtocolTCP}
+ freshService := func(t *testing.T, annotations map[string]string, createRuleErr error) *ensureLBTestEnv {
+ ctrl := gomock.NewController(t)
+ t.Cleanup(ctrl.Finish)
+ env := newEnsureLBTestEnv(ctrl, annotations, []corev1.ServicePort{http})
+ env.expectHosts()
+ env.expectRules()
+ env.network.EXPECT().GetNetworkByID("net-1", gomock.Any()).Return(virtualRouterNetwork(), 1, nil).Times(2)
+ env.address.EXPECT().NewAssociateIpAddressParams().Return(&cloudstack.AssociateIpAddressParams{})
+ env.address.EXPECT().AssociateIpAddress(gomock.Any()).Return(&cloudstack.AssociateIpAddressResponse{Id: "ip-new", Ipaddress: "10.0.0.9"}, nil)
+ env.lb.EXPECT().NewCreateLoadBalancerRuleParams("roundrobin", "atestuid-tcp-80", 30000, 80).Return(&cloudstack.CreateLoadBalancerRuleParams{})
+ if createRuleErr != nil {
+ env.lb.EXPECT().CreateLoadBalancerRule(gomock.Any()).Return(nil, createRuleErr)
+ return env
+ }
+ env.lb.EXPECT().CreateLoadBalancerRule(gomock.Any()).Return(&cloudstack.CreateLoadBalancerRuleResponse{
+ Id: "rule-new", Name: "atestuid-tcp-80", Publicip: "10.0.0.9", Publicipid: "ip-new", Protocol: "tcp",
+ }, nil)
+ env.lb.EXPECT().NewAssignToLoadBalancerRuleParams("rule-new").Return(&cloudstack.AssignToLoadBalancerRuleParams{})
+ env.lb.EXPECT().AssignToLoadBalancerRule(gomock.Any()).Return(&cloudstack.AssignToLoadBalancerRuleResponse{}, nil)
+ env.expectFirewallListings(&cloudstack.ListFirewallRulesResponse{}, 1)
+ env.firewall.EXPECT().NewCreateFirewallRuleParams("ip-new", "tcp").Return(&cloudstack.CreateFirewallRuleParams{})
+ env.firewall.EXPECT().CreateFirewallRule(gomock.Any()).Return(&cloudstack.CreateFirewallRuleResponse{}, nil)
+ return env
+ }
+
+ t.Run("an unsupported method fails before an IP is associated", func(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ t.Cleanup(ctrl.Finish)
+ env := newEnsureLBTestEnv(ctrl, stickinessAnnotations("NoSuchMethod", ""), []corev1.ServicePort{http})
+ env.expectHosts()
+ env.expectRules()
+ env.network.EXPECT().GetNetworkByID("net-1", gomock.Any()).Return(virtualRouterNetwork(), 1, nil)
+
+ _, err := env.cs.EnsureLoadBalancer(context.TODO(), "test-cluster", env.service, env.nodes)
+ if err == nil || !strings.Contains(err.Error(), "not supported on this network") {
+ t.Fatalf("error = %v, want the unsupported method rejected", err)
+ }
+ })
+
+ t.Run("a rejected policy keeps the new IP and its rules", func(t *testing.T) {
+ env := freshService(t, stickinessAnnotations("SourceBased", "tablesize=abc"), nil)
+ env.lb.EXPECT().NewCreateLBStickinessPolicyParams("rule-new", "SourceBased", "atestuid-tcp-80").Return(&cloudstack.CreateLBStickinessPolicyParams{})
+ env.lb.EXPECT().CreateLBStickinessPolicy(gomock.Any()).Return(nil, fmt.Errorf("tablesize is not in size format"))
+
+ _, err := env.cs.EnsureLoadBalancer(context.TODO(), "test-cluster", env.service, env.nodes)
+ if err == nil || !strings.Contains(err.Error(), "stickiness") {
+ t.Fatalf("error = %v, want the policy rejection", err)
+ }
+ })
+
+ t.Run("a failure before the rules exist still releases the new IP", func(t *testing.T) {
+ env := freshService(t, nil, fmt.Errorf("create rule API error"))
+ release := &cloudstack.DisassociateIpAddressParams{}
+ env.address.EXPECT().NewDisassociateIpAddressParams("ip-new").Return(release)
+ env.address.EXPECT().DisassociateIpAddress(release).Return(&cloudstack.DisassociateIpAddressResponse{}, nil)
+
+ _, err := env.cs.EnsureLoadBalancer(context.TODO(), "test-cluster", env.service, env.nodes)
+ if err == nil || !strings.Contains(err.Error(), "create rule API error") {
+ t.Fatalf("error = %v, want the rule creation failure", err)
+ }
+ })
+}
diff --git a/docs/development.md b/docs/development.md
index 28757dd..9727f2c 100644
--- a/docs/development.md
+++ b/docs/development.md
@@ -250,12 +250,14 @@
detect it by looking up its own pod, which cannot work when running as a host
process.
-The script then loads the image into kind, applies the repository's
-[`deployment.yaml`](../deployment.yaml) and patches it for testing: the local
-image with `imagePullPolicy: Never`, `--leader-elect=false` (single replica,
-faster startup), `--v=4` for useful logs, and higher CPU limits — the stock
-manifest's `limits.cpu: 50m` throttles informer startup badly on shared CI
-runners.
+The script then loads the image into kind and applies the repository's
+[`deployment.yaml`](../deployment.yaml), changed for testing before it is
+applied: the local image with `imagePullPolicy: Never`, `--leader-elect=false`
+(single replica, faster startup), `--v=4` for useful logs, and higher CPU
+limits — the stock manifest's `limits.cpu: 50m` throttles informer startup
+badly on shared CI runners. Applying the stock manifest first would start the
+released image, which can keep running for its 30s grace period and reconcile
+the test Services next to the build under test.
Finally it waits for every node to lose the uninitialized taint.
diff --git a/go.mod b/go.mod
index 24e177a..be0b8af 100644
--- a/go.mod
+++ b/go.mod
@@ -3,13 +3,14 @@
go 1.23.0
require (
- github.com/apache/cloudstack-go/v2 v2.19.0
+ github.com/apache/cloudstack-go/v2 v2.19.1
github.com/blang/semver/v4 v4.0.0
github.com/spf13/pflag v1.0.5
go.uber.org/mock v0.5.0
gopkg.in/gcfg.v1 v1.2.3
k8s.io/api v0.24.17
k8s.io/apimachinery v0.24.17
+ k8s.io/client-go v0.24.17
k8s.io/cloud-provider v0.24.17
k8s.io/component-base v0.24.17
k8s.io/klog/v2 v2.80.1
@@ -95,7 +96,6 @@
gopkg.in/yaml.v2 v2.4.0 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
k8s.io/apiserver v0.24.17 // indirect
- k8s.io/client-go v0.24.17 // indirect
k8s.io/component-helpers v0.24.17 // indirect
k8s.io/controller-manager v0.24.17 // indirect
k8s.io/kube-openapi v0.0.0-20220328201542-3ee0da9b0b42 // indirect
diff --git a/go.sum b/go.sum
index 85fd3e0..423e8ea 100644
--- a/go.sum
+++ b/go.sum
@@ -52,8 +52,8 @@
github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
-github.com/apache/cloudstack-go/v2 v2.19.0 h1:YHLw770MmgiqXx6NRFYw2Nr7DpnylLhLG2KYNCftgnc=
-github.com/apache/cloudstack-go/v2 v2.19.0/go.mod h1:p/YBUwIEkQN6CQxFhw8Ff0wzf1MY0qRRRuGYNbcb1F8=
+github.com/apache/cloudstack-go/v2 v2.19.1 h1:1K5O4NZpdWzOZUN6XuaVNsdX+QnoFRc5VE/oc1nUckQ=
+github.com/apache/cloudstack-go/v2 v2.19.1/go.mod h1:p/YBUwIEkQN6CQxFhw8Ff0wzf1MY0qRRRuGYNbcb1F8=
github.com/asaskevich/govalidator v0.0.0-20190424111038-f61b66f89f4a/go.mod h1:lB+ZfQJz7igIIfQNfa7Ml4HSf2uFQQRzpGGRXenZAgY=
github.com/benbjohnson/clock v1.0.3/go.mod h1:bGMdMPoPVvcYyt1gHDf4J2KE153Yf9BuiUKYMaxlTDM=
github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
diff --git a/hack/e2e/40-ccm-deploy.sh b/hack/e2e/40-ccm-deploy.sh
index adaa27c..7dc080f 100755
--- a/hack/e2e/40-ccm-deploy.sh
+++ b/hack/e2e/40-ccm-deploy.sh
@@ -68,19 +68,22 @@
--from-file=cloud-config="${E2E_OUT}/cloud-config" \
--dry-run=client -o yaml | kubectl apply -f -
-kubectl apply -f "${REPO_ROOT}/deployment.yaml"
-# Adjust the stock manifest for e2e: local image, no leader election (single
-# replica, faster startup), verbose logs, and enough CPU that informer startup
-# is not throttled on shared runners.
-kubectl -n kube-system patch deployment cloud-controller-manager --type=json -p '[
- {"op":"replace","path":"/spec/template/spec/containers/0/image","value":"'"$CCM_IMAGE"'"},
- {"op":"replace","path":"/spec/template/spec/containers/0/imagePullPolicy","value":"Never"},
- {"op":"replace","path":"/spec/template/spec/containers/0/args","value":[
- "--cloud-provider=external-cloudstack","--cloud-config=/config/cloud-config",
- "--leader-elect=false","--v=4"]},
- {"op":"replace","path":"/spec/template/spec/containers/0/resources","value":{
- "requests":{"cpu":"100m","memory":"128Mi"},"limits":{"cpu":"1","memory":"512Mi"}}}
-]'
+# Adjust the stock manifest for e2e before applying it: local image, no leader
+# election (single replica, faster startup), verbose logs, and enough CPU that
+# informer startup is not throttled on shared runners. Applying it unchanged
+# first would start the released image, which can keep running for its whole
+# 30s grace period and reconcile the test Services next to the build under test.
+kubectl create --dry-run=client -o json -f "${REPO_ROOT}/deployment.yaml" |
+ jq -s --arg image "$CCM_IMAGE" '{apiVersion: "v1", kind: "List", items: map(
+ if .kind != "Deployment" then . else
+ .spec.template.spec.containers[0] |= (
+ .image = $image
+ | .imagePullPolicy = "Never"
+ | .args = ["--cloud-provider=external-cloudstack", "--cloud-config=/config/cloud-config",
+ "--leader-elect=false", "--v=4"]
+ | .resources = {requests: {cpu: "100m", memory: "128Mi"}, limits: {cpu: "1", memory: "512Mi"}})
+ end)}' |
+ kubectl apply -f -
kubectl -n kube-system rollout status deployment/cloud-controller-manager --timeout=180s
diff --git a/test/e2e/annotations_test.go b/test/e2e/annotations_test.go
index f551348..05f04f3 100644
--- a/test/e2e/annotations_test.go
+++ b/test/e2e/annotations_test.go
@@ -24,6 +24,7 @@
import (
"context"
"fmt"
+ "maps"
"strings"
"testing"
@@ -33,10 +34,12 @@
)
const (
- annotationSourceCidrs = "service.beta.kubernetes.io/cloudstack-load-balancer-source-cidrs"
- annotationHostname = "service.beta.kubernetes.io/cloudstack-load-balancer-hostname"
- annotationIPAssociated = "service.beta.kubernetes.io/cloudstack-load-balancer-ip-associated-by-controller" //nolint:gosec
- annotationProxyProtocol = "service.beta.kubernetes.io/cloudstack-load-balancer-proxy-protocol"
+ annotationSourceCidrs = "service.beta.kubernetes.io/cloudstack-load-balancer-source-cidrs"
+ annotationHostname = "service.beta.kubernetes.io/cloudstack-load-balancer-hostname"
+ annotationIPAssociated = "service.beta.kubernetes.io/cloudstack-load-balancer-ip-associated-by-controller" //nolint:gosec
+ annotationProxyProtocol = "service.beta.kubernetes.io/cloudstack-load-balancer-proxy-protocol"
+ annotationStickinessMethod = "service.beta.kubernetes.io/cloudstack-load-balancer-stickiness-method-name"
+ annotationStickinessParams = "service.beta.kubernetes.io/cloudstack-load-balancer-stickiness-method-param"
)
func TestAnnot_SourceCIDRs(t *testing.T) {
@@ -134,6 +137,166 @@
})
}
+// LbCookie follows appProtocol: http, is kept by a sync that changes nothing, is replaced when a
+// parameter changes and is removed with the annotations. The rules stay throughout.
+func TestAnnot_Stickiness(t *testing.T) {
+ f := NewFramework(t)
+ http := "http"
+ svc := f.CreateLBService(func(s *corev1.Service) {
+ s.Annotations = map[string]string{
+ annotationStickinessMethod: "LbCookie",
+ annotationStickinessParams: "cookie-name=SERVERID,nocache=true",
+ }
+ s.Spec.Ports = []corev1.ServicePort{
+ {Name: "http", Port: 80, Protocol: corev1.ProtocolTCP, AppProtocol: &http},
+ {Name: "alt", Port: 8080, Protocol: corev1.ProtocolTCP},
+ }
+ })
+ lbName := defaultLoadBalancerName(svc)
+ f.WaitForIngressIP(svc)
+ rules := ruleIDsByPort(f.WaitForLBRules(lbName, 2))
+ cookie := map[string]string{"cookie-name": "SERVERID", "nocache": "true"}
+
+ policy := f.WaitForStickinessPolicy(rules["80"], "LbCookie", cookie)
+ if policy.Description != stickinessPolicyMarker {
+ t.Errorf("policy description = %q, want the controller's marker", policy.Description)
+ }
+ f.WaitForNoStickinessPolicy(rules["8080"])
+
+ // Recreating an unchanged policy on every sync would reset affinity each time.
+ f.ResyncAndWait(svc)
+ if again := f.WaitForStickinessPolicy(rules["80"], "LbCookie", cookie); again.Id != policy.Id {
+ t.Errorf("policy replaced %s -> %s by a reconcile that changed nothing", policy.Id, again.Id)
+ }
+
+ f.UpdateService(svc, func(s *corev1.Service) {
+ s.Spec.Ports[0].AppProtocol = nil
+ s.Spec.Ports[1].AppProtocol = &http
+ })
+ moved := f.WaitForStickinessPolicy(rules["8080"], "LbCookie", cookie)
+ f.WaitForNoStickinessPolicy(rules["80"])
+
+ f.UpdateService(svc, func(s *corev1.Service) {
+ s.Annotations[annotationStickinessParams] = "cookie-name=JSESSIONID"
+ })
+ if replaced := f.WaitForStickinessPolicy(rules["8080"], "LbCookie", map[string]string{"cookie-name": "JSESSIONID"}); replaced.Id == moved.Id {
+ t.Errorf("policy %s was kept across a parameter change, want it replaced", moved.Id)
+ }
+
+ f.UpdateService(svc, func(s *corev1.Service) {
+ delete(s.Annotations, annotationStickinessMethod)
+ delete(s.Annotations, annotationStickinessParams)
+ })
+ f.WaitForNoStickinessPolicy(rules["8080"])
+ if after := ruleIDsByPort(f.WaitForLBRules(lbName, 2)); !maps.Equal(after, rules) {
+ t.Errorf("rules changed %v -> %v, want them kept", rules, after)
+ }
+}
+
+// A policy added by hand stays on a Service that does not ask for stickiness.
+func TestAnnot_StickinessForeignPolicy(t *testing.T) {
+ f := NewFramework(t)
+ svc := f.CreateLBService(nil)
+ f.WaitForIngressIP(svc)
+ rules := f.WaitForLBRules(defaultLoadBalancerName(svc), 1)
+ manual := f.CreateStickinessPolicy(rules[0].Id, "manual", "SourceBased", map[string]string{"tablesize": "200k"})
+
+ f.ResyncAndWait(svc)
+ live, err := f.StickinessPolicies(rules[0].Id)
+ if err != nil {
+ t.Fatalf("listing stickiness policies: %v", err)
+ }
+ if len(live) != 1 || live[0].Id != manual {
+ t.Errorf("live policies = %+v, want the hand-made %s kept", live, manual)
+ }
+}
+
+// A stickiness change that cannot be applied fails the sync and leaves the current policies alone,
+// whether the controller or CloudStack rejects it. Each case uses its own Service, because failing
+// syncs keep recording events and Kubernetes drops events for an object that records too many.
+func TestAnnot_StickinessRejected(t *testing.T) {
+ cases := []struct {
+ name string
+ method string
+ params string
+ undeclared bool
+ wantErr string
+ }{
+ {name: "a value CloudStack rejects", method: "SourceBased", params: "tablesize=abc", wantErr: "tablesize"},
+ {name: "a method the network does not offer", method: "NoSuchMethod", wantErr: "not supported on this network"},
+ {name: "an LbCookie mode HAProxy does not know", method: "LbCookie", params: "mode=bogus", wantErr: "insert, rewrite, prefix"},
+ {name: "an empty parameter value", method: "LbCookie", params: "cookie-name=", wantErr: "missing value"},
+ {name: "a character HAProxy would misread", method: "LbCookie", params: "cookie-name=a#b", wantErr: "are not allowed"},
+ {name: "LbCookie on a port without appProtocol http", method: "LbCookie", undeclared: true, wantErr: "appProtocol: http"},
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ f := NewFramework(t)
+ http := "http"
+ svc := f.CreateLBService(func(s *corev1.Service) {
+ s.Annotations = map[string]string{
+ annotationStickinessMethod: "SourceBased",
+ annotationStickinessParams: "tablesize=200k",
+ }
+ if !tc.undeclared {
+ s.Spec.Ports[0].AppProtocol = &http
+ }
+ })
+ policies := map[string]string{}
+ for _, rule := range f.WaitForLBRules(defaultLoadBalancerName(svc), 1) {
+ policies[rule.Id] = f.WaitForStickinessPolicy(rule.Id, "SourceBased", map[string]string{"tablesize": "200k"}).Id
+ }
+
+ f.UpdateService(svc, func(s *corev1.Service) {
+ s.Annotations[annotationStickinessMethod] = tc.method
+ s.Annotations[annotationStickinessParams] = tc.params
+ })
+ f.WaitForSyncFailure(svc, tc.wantErr)
+
+ for ruleID, policyID := range policies {
+ live, err := f.StickinessPolicies(ruleID)
+ if err != nil {
+ t.Fatalf("listing stickiness policies of rule %s: %v", ruleID, err)
+ }
+ if len(live) != 1 || live[0].Id != policyID {
+ t.Errorf("rule %s policies = %+v, want %s untouched", ruleID, live, policyID)
+ }
+ }
+ })
+ }
+}
+
+// A new Service whose policy CloudStack rejects keeps its IP and rule and stays pending. Once the
+// annotation is fixed, it publishes the same IP.
+func TestAnnot_StickinessNewServiceRejected(t *testing.T) {
+ f := NewFramework(t)
+ svc := f.CreateLBService(func(s *corev1.Service) {
+ s.Annotations = map[string]string{
+ annotationStickinessMethod: "SourceBased",
+ annotationStickinessParams: "tablesize=abc",
+ }
+ })
+ lbName := defaultLoadBalancerName(svc)
+ f.WaitForSyncFailure(svc, "tablesize")
+ rules := f.WaitForLBRules(lbName, 1)
+
+ current, err := f.K8s.CoreV1().Services(svc.Namespace).Get(context.Background(), svc.Name, metav1.GetOptions{})
+ if err != nil {
+ t.Fatalf("getting service: %v", err)
+ }
+ if len(current.Status.LoadBalancer.Ingress) != 0 {
+ t.Errorf("ingress = %v, want none while the sync fails", current.Status.LoadBalancer.Ingress)
+ }
+
+ f.UpdateService(svc, func(s *corev1.Service) {
+ s.Annotations[annotationStickinessParams] = "tablesize=200k"
+ })
+ if ingress := f.WaitForIngressIP(svc); ingress.IP != rules[0].Publicip {
+ t.Errorf("ingress IP = %q, want the IP the failed sync kept, %q", ingress.IP, rules[0].Publicip)
+ }
+ f.WaitForStickinessPolicy(rules[0].Id, "SourceBased", map[string]string{"tablesize": "200k"})
+}
+
func TestAnnot_ExplicitLoadBalancerIP(t *testing.T) {
f := NewFramework(t)
diff --git a/test/e2e/framework.go b/test/e2e/framework.go
index dbf7ecb..bed2cea 100644
--- a/test/e2e/framework.go
+++ b/test/e2e/framework.go
@@ -39,6 +39,7 @@
"crypto/rand"
"encoding/hex"
"fmt"
+ "maps"
"os"
"strconv"
"strings"
@@ -57,6 +58,13 @@
const (
lbSyncTimeout = 3 * time.Minute
lbSyncInterval = 3 * time.Second
+
+ // resyncAnnotation is changed by ResyncAndWait to make the service controller sync a Service
+ // again. The CCM ignores it.
+ resyncAnnotation = "e2e.cloudstack.apache.org/resync"
+
+ // stickinessPolicyMarker is the description the CCM gives the stickiness policies it creates.
+ stickinessPolicyMarker = "Managed by the CloudStack Kubernetes Provider"
)
// Framework bundles the clients and helpers shared by all e2e tests.
@@ -364,6 +372,163 @@
return rules
}
+// StickinessPolicies lists the stickiness policies of a rule, skipping revoked ones CloudStack
+// has not removed yet.
+func (f *Framework) StickinessPolicies(ruleID string) ([]cloudstack.LBStickinessPolicyStickinesspolicy, error) {
+ p := f.CS.LoadBalancer.NewListLBStickinessPoliciesParams()
+ p.SetLbruleid(ruleID)
+ resp, err := f.CS.LoadBalancer.ListLBStickinessPolicies(p)
+ if err != nil {
+ return nil, err
+ }
+ var live []cloudstack.LBStickinessPolicyStickinesspolicy
+ for _, wrapper := range resp.LBStickinessPolicies {
+ for _, policy := range wrapper.Stickinesspolicy {
+ if !strings.EqualFold(policy.State, "Revoked") {
+ live = append(live, policy)
+ }
+ }
+ }
+ return live, nil
+}
+
+// WaitForStickinessPolicy waits until the rule has exactly one policy, with the given method and
+// parameters, and returns it.
+func (f *Framework) WaitForStickinessPolicy(ruleID, method string, params map[string]string) cloudstack.LBStickinessPolicyStickinesspolicy {
+ f.T.Helper()
+ var policy cloudstack.LBStickinessPolicyStickinesspolicy
+ f.Eventually(lbSyncTimeout, lbSyncInterval,
+ fmt.Sprintf("rule %s to carry one %s stickiness policy with params %v", ruleID, method, params),
+ func() (bool, error) {
+ live, err := f.StickinessPolicies(ruleID)
+ if err != nil {
+ return false, err
+ }
+ if len(live) != 1 {
+ return false, fmt.Errorf("rule has %d live policies", len(live))
+ }
+ if !strings.EqualFold(live[0].Methodname, method) || !maps.Equal(live[0].Params, params) {
+ return false, fmt.Errorf("policy is %s %v", live[0].Methodname, live[0].Params)
+ }
+ policy = live[0]
+ return true, nil
+ })
+ return policy
+}
+
+// WaitForNoStickinessPolicy waits until the rule has no stickiness policy.
+func (f *Framework) WaitForNoStickinessPolicy(ruleID string) {
+ f.T.Helper()
+ f.Eventually(lbSyncTimeout, lbSyncInterval, fmt.Sprintf("rule %s to carry no stickiness policy", ruleID),
+ func() (bool, error) {
+ live, err := f.StickinessPolicies(ruleID)
+ if err != nil {
+ return false, err
+ }
+ if len(live) != 0 {
+ return false, fmt.Errorf("rule has %d live policies", len(live))
+ }
+ return true, nil
+ })
+}
+
+// CreateStickinessPolicy adds a policy to a rule through the CloudStack API, the way an operator
+// would by hand, and returns its ID.
+func (f *Framework) CreateStickinessPolicy(ruleID, name, method string, params map[string]string) string {
+ f.T.Helper()
+ p := f.CS.LoadBalancer.NewCreateLBStickinessPolicyParams(ruleID, method, name)
+ p.SetParam(params)
+ resp, err := f.CS.LoadBalancer.CreateLBStickinessPolicy(p)
+ if err != nil {
+ f.T.Fatalf("creating stickiness policy %s on rule %s: %v", name, ruleID, err)
+ }
+ if len(resp.Stickinesspolicy) == 0 {
+ f.T.Fatalf("creating stickiness policy %s on rule %s returned no policy", name, ruleID)
+ }
+ return resp.Stickinesspolicy[0].Id
+}
+
+// ResyncAndWait makes the service controller sync svc again, by changing an annotation the CCM
+// ignores, and waits for the sync to succeed. EnsureLoadBalancer only runs when a Service changes
+// or the controller restarts, so tests need this to check what a sync leaves alone.
+func (f *Framework) ResyncAndWait(svc *corev1.Service) {
+ f.T.Helper()
+ before, err := f.ensuredCount(svc)
+ if err != nil {
+ f.T.Fatalf("counting reconciles of service %s/%s: %v", svc.Namespace, svc.Name, err)
+ }
+ f.UpdateService(svc, func(s *corev1.Service) {
+ if s.Annotations == nil {
+ s.Annotations = map[string]string{}
+ }
+ s.Annotations[resyncAnnotation] = strconv.FormatInt(time.Now().UnixNano(), 10)
+ })
+ f.Eventually(lbSyncTimeout, lbSyncInterval, fmt.Sprintf("service %s/%s to be reconciled again", svc.Namespace, svc.Name),
+ func() (bool, error) {
+ after, err := f.ensuredCount(svc)
+ return after > before, err
+ })
+}
+
+// ensuredCount counts the successful syncs recorded for svc, including repeats that Kubernetes
+// folded into one event.
+func (f *Framework) ensuredCount(svc *corev1.Service) (int32, error) {
+ events, err := f.K8s.CoreV1().Events(svc.Namespace).List(context.Background(), metav1.ListOptions{
+ FieldSelector: "involvedObject.name=" + svc.Name + ",reason=EnsuredLoadBalancer",
+ })
+ if err != nil {
+ return 0, err
+ }
+ var total int32
+ for _, event := range events.Items {
+ count := max(event.Count, 1)
+ if event.Series != nil {
+ count = max(count, event.Series.Count)
+ }
+ total += count
+ }
+ return total, nil
+}
+
+// WaitForSyncFailure waits until a failed sync of svc is reported with a message containing substr.
+func (f *Framework) WaitForSyncFailure(svc *corev1.Service, substr string) {
+ f.T.Helper()
+ f.Eventually(lbSyncTimeout, lbSyncInterval, fmt.Sprintf("a failed sync of service %s/%s mentioning %q", svc.Namespace, svc.Name, substr),
+ func() (bool, error) {
+ events, err := f.K8s.CoreV1().Events(svc.Namespace).List(context.Background(), metav1.ListOptions{
+ FieldSelector: "involvedObject.name=" + svc.Name + ",reason=SyncLoadBalancerFailed",
+ })
+ if err != nil {
+ return false, err
+ }
+ for _, event := range events.Items {
+ if strings.Contains(event.Message, substr) {
+ return true, nil
+ }
+ }
+ return false, nil
+ })
+}
+
+// ruleIDsByPort maps the public port of each load balancer rule to the rule's id.
+func ruleIDsByPort(rules []*cloudstack.LoadBalancerRule) map[string]string {
+ ids := make(map[string]string, len(rules))
+ for _, rule := range rules {
+ ids[rule.Publicport] = rule.Id
+ }
+ return ids
+}
+
+// RuleInstanceCount returns how many VMs are assigned to a load balancer rule.
+func (f *Framework) RuleInstanceCount(ruleID string) (int, error) {
+ p := f.CS.LoadBalancer.NewListLoadBalancerRuleInstancesParams(ruleID)
+ resp, err := f.CS.LoadBalancer.ListLoadBalancerRuleInstances(p)
+ if err != nil {
+ return 0, err
+ }
+ return len(resp.LoadBalancerRuleInstances), nil
+}
+
// FirewallRules lists the firewall rules on a public IP.
func (f *Framework) FirewallRules(publicIPID string) ([]*cloudstack.FirewallRule, error) {
p := f.CS.Firewall.NewListFirewallRulesParams()
diff --git a/test/e2e/loadbalancer_test.go b/test/e2e/loadbalancer_test.go
index b4918b1..5c5d018 100644
--- a/test/e2e/loadbalancer_test.go
+++ b/test/e2e/loadbalancer_test.go
@@ -161,6 +161,54 @@
})
}
+// A rule left without hosts, for example because assigning them failed after
+// the rule was created, gets them back on the next sync instead of waiting for
+// a node change.
+func TestLB_HostsRestoredOnResync(t *testing.T) {
+ f := NewFramework(t)
+ svc := f.CreateLBService(nil)
+ lbName := defaultLoadBalancerName(svc)
+
+ f.WaitForIngressIP(svc)
+ rules := f.WaitForLBRules(lbName, 1)
+ var assigned []string
+ f.Eventually(lbSyncTimeout, lbSyncInterval, "hosts to be assigned to the rule",
+ func() (bool, error) {
+ p := f.CS.LoadBalancer.NewListLoadBalancerRuleInstancesParams(rules[0].Id)
+ resp, err := f.CS.LoadBalancer.ListLoadBalancerRuleInstances(p)
+ if err != nil {
+ return false, err
+ }
+ assigned = assigned[:0]
+ for _, inst := range resp.LoadBalancerRuleInstances {
+ assigned = append(assigned, inst.Id)
+ }
+ return len(assigned) > 0, nil
+ })
+
+ remove := f.CS.LoadBalancer.NewRemoveFromLoadBalancerRuleParams(rules[0].Id)
+ remove.SetVirtualmachineids(assigned)
+ if _, err := f.CS.LoadBalancer.RemoveFromLoadBalancerRule(remove); err != nil {
+ t.Fatalf("removing hosts from rule %s: %v", rules[0].Id, err)
+ }
+ if got, err := f.RuleInstanceCount(rules[0].Id); err != nil || got != 0 {
+ t.Fatalf("rule still has %d hosts after removal (err %v)", got, err)
+ }
+
+ forceReconcile(f, svc, lbName)
+ f.Eventually(lbSyncTimeout, lbSyncInterval, "the sync to restore the rule's hosts",
+ func() (bool, error) {
+ got, err := f.RuleInstanceCount(rules[0].Id)
+ if err != nil {
+ return false, err
+ }
+ if got != len(assigned) {
+ return false, fmt.Errorf("rule has %d hosts, want %d", got, len(assigned))
+ }
+ return true, nil
+ })
+}
+
func TestLB_PortChange(t *testing.T) {
f := NewFramework(t)
svc := f.CreateLBService(nil)
