blob: 753322dc84a69ebef9a2433a5e315d125d9ff34b [file]
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package cloudstack
import (
"context"
"encoding/json"
"errors"
"fmt"
"maps"
"net"
"slices"
"sort"
"strconv"
"strings"
"unicode"
"github.com/apache/cloudstack-go/v2/cloudstack"
"github.com/blang/semver/v4"
"k8s.io/klog/v2"
corev1 "k8s.io/api/core/v1"
cloudprovider "k8s.io/cloud-provider"
)
const (
// defaultAllowedCIDR is the network range that is allowed on the firewall
// by default when no explicit CIDR list is given on a LoadBalancer.
defaultAllowedCIDR = "0.0.0.0/0"
// ServiceAnnotationLoadBalancerProxyProtocol is the annotation used on the
// service to enable the proxy protocol on a CloudStack load balancer.
// Note that this protocol only applies to TCP service ports and
// CloudStack >= 4.6 is required for it to work.
ServiceAnnotationLoadBalancerProxyProtocol = "service.beta.kubernetes.io/cloudstack-load-balancer-proxy-protocol"
ServiceAnnotationLoadBalancerLoadbalancerHostname = "service.beta.kubernetes.io/cloudstack-load-balancer-hostname"
// ServiceAnnotationLoadBalancerSourceCidrs is the annotation used on the
// service to specify the source CIDR list for a CloudStack load balancer.
// The CIDR list is a comma-separated list of CIDR ranges (e.g., "10.0.0.0/8,192.168.1.0/24").
// If not specified, the default is to allow all sources ("0.0.0.0/0").
ServiceAnnotationLoadBalancerSourceCidrs = "service.beta.kubernetes.io/cloudstack-load-balancer-source-cidrs"
// ServiceAnnotationLoadBalancerIPAssociatedByController indicates that the controller
// 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
// accepts a cidrlist. Below it, a changed source CIDR list can only be applied by deleting
// the rule and creating it again.
var cidrListUpdateVersion = semver.Version{Major: 4, Minor: 22, Patch: 0}
type loadBalancer struct {
*cloudstack.CloudStackClient
name string
algorithm string
hostIDs []string
ipAddr string
ipAddrID string
networkID string
projectID string
rules map[string]*cloudstack.LoadBalancerRule
duplicateRules []*cloudstack.LoadBalancerRule
networks map[string]*cloudstack.Network
ipAssociatedByController bool
}
// ruleChange is what applying a desired service port does to its load balancer rule.
type ruleChange int
const (
ruleMissing ruleChange = iota // no rule exists, so one is created
ruleUpToDate // the existing rule is left alone
ruleNeedsUpdate // the existing rule is updated in place
ruleNeedsRecreate // the existing rule is deleted and created again
)
// desiredLBRule describes the load balancer rule a service port should be represented by,
// together with the existing CloudStack rule it resolved to (if any) and what applying it does.
type desiredLBRule struct {
name string
port corev1.ServicePort
protocol LoadBalancerProtocol
existing *cloudstack.LoadBalancerRule // nil when change is ruleMissing
change ruleChange
}
// createsRule reports whether applying this port creates a load balancer rule, and so needs
// its public port free of any other rule first.
func (d desiredLBRule) createsRule() bool {
return d.change == ruleMissing || d.change == ruleNeedsRecreate
}
// GetLoadBalancer returns whether the specified load balancer exists, and if so, what its status is.
func (cs *CSCloud) GetLoadBalancer(ctx context.Context, clusterName string, service *corev1.Service) (*corev1.LoadBalancerStatus, bool, error) {
klog.V(4).Infof("GetLoadBalancer(%v, %v, %v)", clusterName, service.Namespace, service.Name)
// Get the load balancer details and existing rules.
lb, err := cs.getLoadBalancer(service)
if err != nil {
return nil, false, err
}
// If we don't have any rules, the load balancer does not exist.
if len(lb.rules) == 0 {
return nil, false, nil
}
klog.V(4).Infof("Found a load balancer associated with IP %v", lb.ipAddr)
status := &corev1.LoadBalancerStatus{}
status.Ingress = append(status.Ingress, corev1.LoadBalancerIngress{IP: lb.ipAddr})
return status, true, nil
}
// EnsureLoadBalancer creates a new load balancer, or updates the existing one. Returns the status of the balancer.
func (cs *CSCloud) EnsureLoadBalancer(ctx context.Context, clusterName string, service *corev1.Service, nodes []*corev1.Node) (status *corev1.LoadBalancerStatus, err error) {
klog.V(4).Infof("EnsureLoadBalancer(%v, %v, %v, %v, %v, %v)", clusterName, service.Namespace, service.Name, service.Spec.LoadBalancerIP, service.Spec.Ports, nodes)
if len(service.Spec.Ports) == 0 {
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 {
return nil, err
}
if err := lb.deleteDuplicateRules(); err != nil {
return nil, err
}
// Set the load balancer algorithm.
switch service.Spec.SessionAffinity {
case corev1.ServiceAffinityNone:
lb.algorithm = "roundrobin"
case corev1.ServiceAffinityClientIP:
lb.algorithm = "source"
default:
return nil, fmt.Errorf("unsupported load balancer affinity: %v", service.Spec.SessionAffinity)
}
// Verify that all the hosts belong to the same network, and retrieve their ID's.
lb.hostIDs, lb.networkID, err = cs.verifyHosts(nodes)
if err != nil {
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 {
return nil, err
}
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 && !rulesApplied {
if err := lb.releaseLoadBalancerIP(); err != nil {
klog.Errorf(err.Error())
}
}
}(lb)
}
// If the controller associated the IP and matches the service spec, set the annotation to persist this information.
if lb.ipAssociatedByController && lb.ipAddr == service.Spec.LoadBalancerIP {
if err := cs.setServiceAnnotation(ctx, service, ServiceAnnotationLoadBalancerIPAssociatedByController, "true"); err != nil {
// Log the error but don't fail - the annotation is helpful but not critical
klog.Warningf("Failed to set annotation on service %s/%s: %v", service.Namespace, service.Name, err)
}
}
}
klog.V(4).Infof("Load balancer %v is associated with IP %v", lb.name, lb.ipAddr)
// Resolve every service port to the rule that should represent it.
desired, err := lb.resolveLoadBalancerRules(service, cs.version)
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
// CloudStack rejects the create as a port conflict.
if err := lb.pruneRules(blocking, desired, network); err != nil {
return nil, err
}
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. 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
}
status = &corev1.LoadBalancerStatus{}
// If hostname is explicitly set using service annotation
// Workaround for https://github.com/kubernetes/kubernetes/issues/66607
if hostname := getStringFromServiceAnnotation(service, ServiceAnnotationLoadBalancerLoadbalancerHostname, ""); hostname != "" {
status.Ingress = []corev1.LoadBalancerIngress{{Hostname: hostname}}
return status, nil
}
// Default to IP
status.Ingress = []corev1.LoadBalancerIngress{{IP: lb.ipAddr}}
return status, nil
}
// UpdateLoadBalancer updates hosts under the specified load balancer.
func (cs *CSCloud) UpdateLoadBalancer(ctx context.Context, clusterName string, service *corev1.Service, nodes []*corev1.Node) error {
klog.V(4).Infof("UpdateLoadBalancer(%v, %v, %v, %v)", clusterName, service.Namespace, service.Name, nodes)
// Get the load balancer details and existing rules.
lb, err := cs.getLoadBalancer(service)
if err != nil {
return err
}
// Verify that all the hosts belong to the same network, and retrieve their ID's.
lb.hostIDs, _, err = cs.verifyHosts(nodes)
if err != nil {
return err
}
for _, lbRule := range lb.rules {
p := lb.LoadBalancer.NewListLoadBalancerRuleInstancesParams(lbRule.Id)
// Retrieve all VMs currently associated to this load balancer rule. There
// is one per load balanced node, so this grows with the cluster.
instances, err := listAll(p, func() (int, []*cloudstack.VirtualMachine, error) {
l, err := lb.LoadBalancer.ListLoadBalancerRuleInstances(p)
if err != nil {
return 0, nil, err
}
return l.Count, l.LoadBalancerRuleInstances, nil
})
if err != nil {
return fmt.Errorf("error retrieving associated instances: %v", err)
}
assign, remove := symmetricDifference(lb.hostIDs, instances)
if len(assign) > 0 {
klog.V(4).Infof("Assigning new hosts (%v) to load balancer rule: %v", assign, lbRule.Name)
if err := lb.assignHostsToRule(lbRule, assign); err != nil {
return err
}
}
if len(remove) > 0 {
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
}
}
}
return nil
}
func isFirewallSupported(services []cloudstack.NetworkServiceInternal) bool {
for _, svc := range services {
if svc.Name == "Firewall" {
return true
}
}
return false
}
func isNetworkACLSupported(services []cloudstack.NetworkServiceInternal) bool {
for _, svc := range services {
if svc.Name == "NetworkACL" {
return true
}
}
return false
}
// EnsureLoadBalancerDeleted deletes the specified load balancer if it exists, returning
// nil if the load balancer specified either didn't exist or was successfully deleted.
func (cs *CSCloud) EnsureLoadBalancerDeleted(ctx context.Context, clusterName string, service *corev1.Service) error {
klog.V(4).Infof("EnsureLoadBalancerDeleted(%v, %v, %v)", clusterName, service.Namespace, service.Name)
// Get the load balancer details and existing rules.
lb, err := cs.getLoadBalancer(service)
if err != nil {
return err
}
// Reported only once this service's own resources are gone, so a retry sees the
// leftover duplicate as an ordinary rule and deletes it through the path above.
sweepErr := lb.deleteDuplicateRules()
if sweepErr != nil {
klog.Errorf("Error removing duplicate load balancer rules for %v/%v: %v", service.Namespace, service.Name, sweepErr)
}
for _, lbRule := range lb.rules {
klog.V(4).Infof("Deleting firewall rules / Network ACLs for load balancer: %v", lbRule.Name)
protocol := ProtocolFromLoadBalancer(lbRule.Protocol)
if protocol == LoadBalancerProtocolInvalid {
klog.Errorf("Error parsing protocol: %v", lbRule.Protocol)
} else {
port, err := strconv.ParseInt(lbRule.Publicport, 10, 32)
if err != nil {
klog.Errorf("Error parsing port: %v", err)
} else {
networkId, err := cs.getNetworkIDFromIPAddress(lb.ipAddrID)
if err != nil {
return err
}
network, count, err := lb.Network.GetNetworkByID(networkId, cloudstack.WithProject(lb.projectID))
if err != nil {
if count == 0 {
klog.Errorf("No network found with ID: %v", networkId)
return err
}
return err
}
if network.Vpcid == "" {
_, err = lb.deleteFirewallRule(lbRule.Publicipid, int(port), protocol)
if err != nil {
klog.Errorf("Error deleting firewall rule: %v", err)
}
} else {
klog.V(4).Infof("Deleting network ACLs for %v - %v", int(port), protocol)
_, err = lb.deleteNetworkACLRule(int(port), protocol, networkId)
if err != nil {
klog.Errorf("Error deleting Network ACL rule: %v", err)
}
}
}
klog.V(4).Infof("Deleting load balancer rule: %v", lbRule.Name)
if err := lb.deleteLoadBalancerRule(lbRule); err != nil {
return err
}
}
}
if lb.ipAddr != "" {
// If the IP was allocated by the controller (not specified in service spec), release it.
if lb.ipAddr != service.Spec.LoadBalancerIP {
klog.V(4).Infof("Releasing load balancer IP: %v", lb.ipAddr)
if err := lb.releaseLoadBalancerIP(); err != nil {
return err
}
} else {
// If the IP was specified in service spec, check if it was associated by the controller.
// First, check if there's an annotation indicating the controller associated it.
// If not, check if there are any other load balancer rules using this IP.
shouldDisassociate := getBoolFromServiceAnnotation(service, ServiceAnnotationLoadBalancerIPAssociatedByController, false)
if shouldDisassociate {
// Annotation is set, so check if there are any other load balancer rules using this IP.
// Since we've already deleted all rules for this service, any remaining rules must belong
// to other services. If no other rules exist, it's safe to disassociate the IP.
ip, count, err := lb.Address.GetPublicIpAddressByID(lb.ipAddrID, cloudstack.WithProject(lb.projectID))
if err != nil {
klog.Errorf("Error retrieving IP address %v for disassociation check: %v", lb.ipAddr, err)
shouldDisassociate = false
} else if count > 0 && ip.Allocated != "" {
p := lb.LoadBalancer.NewListLoadBalancerRulesParams()
p.SetPublicipid(lb.ipAddrID)
p.SetListall(true)
if lb.projectID != "" {
p.SetProjectid(lb.projectID)
}
otherRules, err := lb.LoadBalancer.ListLoadBalancerRules(p)
if err != nil {
klog.Errorf("Error checking for other load balancer rules using IP %v: %v", lb.ipAddr, err)
shouldDisassociate = false
} else if otherRules.Count > 0 {
// Other load balancer rules are using this IP (other services are using it),
// so don't disassociate.
shouldDisassociate = false
}
}
}
if shouldDisassociate {
klog.V(4).Infof("Disassociating IP %v that was associated by the controller", lb.ipAddr)
if err := lb.releaseLoadBalancerIP(); err != nil {
return err
}
}
}
}
return sweepErr
}
// GetLoadBalancerName retrieves the name of the LoadBalancer.
func (cs *CSCloud) GetLoadBalancerName(ctx context.Context, clusterName string, service *corev1.Service) string {
return cloudprovider.DefaultLoadBalancerName(service)
}
// getLoadBalancer retrieves the IP address and ID and all the existing rules it can find.
func (cs *CSCloud) getLoadBalancer(service *corev1.Service) (*loadBalancer, error) {
lb := &loadBalancer{
CloudStackClient: cs.client,
name: cs.GetLoadBalancerName(context.TODO(), "", service),
projectID: cs.projectID,
rules: make(map[string]*cloudstack.LoadBalancerRule),
}
p := cs.client.LoadBalancer.NewListLoadBalancerRulesParams()
p.SetKeyword(lb.name)
p.SetListall(true)
if cs.projectID != "" {
p.SetProjectid(cs.projectID)
}
// The keyword is matched as a substring server side, so this can return more
// rules than just this service's and has to be paged through.
lbRules, err := listAll(p, func() (int, []*cloudstack.LoadBalancerRule, error) {
l, err := cs.client.LoadBalancer.ListLoadBalancerRules(p)
if err != nil {
return 0, nil, err
}
return l.Count, l.LoadBalancerRules, nil
})
if err != nil {
return nil, fmt.Errorf("error retrieving load balancer rules: %v", err)
}
lbRules = dedupeByID(lbRules, func(rule *cloudstack.LoadBalancerRule) string { return rule.Id })
// Keeping the rule on the address the Service is already published on stops a
// duplicate sweep from deleting the rule that clients and DNS are pointing at. The
// same address is the one the rules are reconciled towards when they span several.
// Without one, the first-listed rule wins here as it does in the sweep.
preferredIP := service.Spec.LoadBalancerIP
if preferredIP == "" && len(service.Status.LoadBalancer.Ingress) > 0 {
preferredIP = service.Status.LoadBalancer.Ingress[0].IP
}
for _, lbRule := range lbRules {
if existing, seen := lb.rules[lbRule.Name]; seen {
duplicate := lbRule
if lbRule.Publicip == preferredIP && existing.Publicip != preferredIP {
duplicate = existing
}
klog.Warningf("Duplicate load balancer rule %v for service %v/%v, removing %v on %v", lbRule.Name, service.Namespace, service.Name, duplicate.Id, duplicate.Publicip)
lb.duplicateRules = append(lb.duplicateRules, duplicate)
if duplicate == lbRule {
continue
}
}
lb.rules[lbRule.Name] = lbRule
if lb.ipAddr != "" && lb.ipAddr != lbRule.Publicip {
klog.Warningf("Load balancer for service %v/%v has rules associated with different IP's: %v, %v", service.Namespace, service.Name, lb.ipAddr, lbRule.Publicip)
}
if lb.ipAddr == "" || (lbRule.Publicip == preferredIP && lb.ipAddr != preferredIP) {
lb.ipAddr = lbRule.Publicip
lb.ipAddrID = lbRule.Publicipid
}
}
klog.V(4).Infof("Load balancer %v contains %d rule(s)", lb.name, len(lb.rules))
return lb, nil
}
// Get network ID from Public IP Address
// Every failure returns an error: GetNetworkByID does not reject an empty ID but
// matches an unfiltered network list, so ("", nil) would resolve to any network.
func (cs *CSCloud) getNetworkIDFromIPAddress(publicIpId string) (string, error) {
ip, count, err := cs.client.Address.GetPublicIpAddressByID(publicIpId, cloudstack.WithProject(cs.projectID))
if err != nil {
klog.Errorf("Failed to fetch the public IP for id: %v", publicIpId)
return "", err
}
if count == 0 {
return "", fmt.Errorf("no public IP address found with ID %v", publicIpId)
}
if ip.Associatednetworkid != "" {
network, _, netErr := cs.client.Network.GetNetworkByID(ip.Associatednetworkid, cloudstack.WithProject(cs.projectID))
if netErr != nil {
klog.Errorf("Failed to fetch the network for id: %v", ip.Associatednetworkid)
return "", netErr
}
return network.Id, nil
}
return "", fmt.Errorf("public IP address %v is not associated with a network", publicIpId)
}
// verifyHosts verifies if all hosts belong to the same network, and returns the host ID's and network ID.
func (cs *CSCloud) verifyHosts(nodes []*corev1.Node) ([]string, string, error) {
hostNames := map[string]bool{}
for _, node := range nodes {
// node.Name can be an FQDN as well, and CloudStack VM names aren't
// To match, we need to Split the domain part off here, if present
hostNames[strings.Split(strings.ToLower(node.Name), ".")[0]] = true
}
p := cs.client.VirtualMachine.NewListVirtualMachinesParams()
p.SetListall(true)
p.SetDetails([]string{"min", "nics"})
if cs.projectID != "" {
p.SetProjectid(cs.projectID)
}
vms, err := listAll(p, func() (int, []*cloudstack.VirtualMachine, error) {
l, err := cs.client.VirtualMachine.ListVirtualMachines(p)
if err != nil {
return 0, nil, err
}
return l.Count, l.VirtualMachines, nil
})
if err != nil {
return nil, "", fmt.Errorf("error retrieving list of hosts: %v", err)
}
var hostIDs []string
var networkID string
seen := map[string]bool{} // used to check whether the changing set of VMs contains one we had already seen in another page.
// Check if the virtual machine is in the hosts slice, then add the corresponding ID.
for _, vm := range vms {
if !hostNames[strings.ToLower(vm.Name)] || seen[vm.Id] {
continue
}
seen[vm.Id] = true
if networkID != "" && networkID != vm.Nic[0].Networkid {
return nil, "", fmt.Errorf("found hosts that belong to different networks")
}
networkID = vm.Nic[0].Networkid
hostIDs = append(hostIDs, vm.Id)
}
if len(hostIDs) == 0 || len(networkID) == 0 {
return nil, "", fmt.Errorf("none of the hosts matched the list of VMs retrieved from CS API")
}
return hostIDs, networkID, nil
}
// hasLoadBalancerIP returns true if we have a load balancer address and ID.
func (lb *loadBalancer) hasLoadBalancerIP() bool {
return lb.ipAddr != "" && lb.ipAddrID != ""
}
// getLoadBalancerIP retrieves an existing IP or associates a new IP.
func (lb *loadBalancer) getLoadBalancerIP(loadBalancerIP string) error {
if loadBalancerIP != "" {
return lb.getPublicIPAddress(loadBalancerIP)
}
return lb.associatePublicIPAddress()
}
// getPublicIPAddressID retrieves the ID of the given IP, and sets the address and it's ID.
func (lb *loadBalancer) getPublicIPAddress(loadBalancerIP string) error {
klog.V(4).Infof("Retrieve load balancer IP details: %v", loadBalancerIP)
p := lb.Address.NewListPublicIpAddressesParams()
p.SetIpaddress(loadBalancerIP)
p.SetAllocatedonly(false)
p.SetListall(true)
if lb.projectID != "" {
p.SetProjectid(lb.projectID)
}
l, err := lb.Address.ListPublicIpAddresses(p)
if err != nil {
return fmt.Errorf("error retrieving IP address: %v", err)
}
if l.Count != 1 {
return fmt.Errorf("could not find IP address %v. Found %d addresses", loadBalancerIP, l.Count)
}
lb.ipAddr = l.PublicIpAddresses[0].Ipaddress
lb.ipAddrID = l.PublicIpAddresses[0].Id
// If the IP is not allocated, associate it.
if l.PublicIpAddresses[0].Allocated == "" {
return lb.associatePublicIPAddress()
}
return nil
}
// associatePublicIPAddress associates a new IP and sets the address and it's ID.
func (lb *loadBalancer) associatePublicIPAddress() error {
klog.V(4).Infof("Allocate new IP for load balancer: %v", lb.name)
// If a network belongs to a VPC, the IP address needs to be associated with
// the VPC instead of with the network.
network, count, err := lb.Network.GetNetworkByID(lb.networkID, cloudstack.WithProject(lb.projectID))
if err != nil {
if count == 0 {
return fmt.Errorf("could not find network %v", lb.networkID)
}
return fmt.Errorf("error retrieving network: %v", err)
}
p := lb.Address.NewAssociateIpAddressParams()
if network.Vpcid != "" {
p.SetVpcid(network.Vpcid)
} else {
p.SetNetworkid(lb.networkID)
}
if lb.projectID != "" {
p.SetProjectid(lb.projectID)
}
if lb.ipAddr != "" {
p.SetIpaddress(lb.ipAddr)
}
// Associate a new IP address
r, err := lb.Address.AssociateIpAddress(p)
if err != nil {
return fmt.Errorf("error associating new IP address: %v", err)
}
lb.ipAddr = r.Ipaddress
lb.ipAddrID = r.Id
lb.ipAssociatedByController = true
return nil
}
// releasePublicIPAddress releases an associated IP.
func (lb *loadBalancer) releaseLoadBalancerIP() error {
p := lb.Address.NewDisassociateIpAddressParams(lb.ipAddrID)
if _, err := lb.Address.DisassociateIpAddress(p); err != nil {
return fmt.Errorf("error releasing load balancer IP %v: %v", lb.ipAddr, err)
}
return nil
}
func (lb *loadBalancer) getCIDRList(service *corev1.Service) ([]string, error) {
sourceCIDRs := getStringFromServiceAnnotation(service, ServiceAnnotationLoadBalancerSourceCidrs, defaultAllowedCIDR)
var cidrList []string
if sourceCIDRs != "" {
cidrList = strings.Split(sourceCIDRs, ",")
for i, cidr := range cidrList {
cidr = strings.TrimSpace(cidr)
if _, _, err := net.ParseCIDR(cidr); err != nil {
return nil, fmt.Errorf("invalid CIDR %s in annotation %s: %w", cidr, ServiceAnnotationLoadBalancerSourceCidrs, err)
}
cidrList[i] = cidr
}
}
return cidrList, nil
}
// splitCIDRList splits the CIDR list of an existing CloudStack rule into its entries.
// CloudStack has reported these both comma and space separated, and a CIDR can contain
// neither character, so treat both as separators.
func splitCIDRList(cidrList string) []string {
return strings.FieldsFunc(cidrList, func(r rune) bool {
return r == ',' || r == ' '
})
}
// resolveLoadBalancerRules maps every service port to the load balancer rule that should
// represent it, claiming each match as it goes so that what remains in lb.rules is exactly
// the obsolete set and no rule can be claimed twice. It changes nothing in CloudStack: a rule
// that has to be recreated is only deleted when it is applied, so an error on a later port, or
// anywhere before the apply phase, leaves the existing rule serving.
func (lb *loadBalancer) resolveLoadBalancerRules(service *corev1.Service, version semver.Version) ([]desiredLBRule, error) {
desired := make([]desiredLBRule, 0, len(service.Spec.Ports))
for _, port := range service.Spec.Ports {
// Construct the protocol name first, we need it a few times
protocol := ProtocolFromServicePort(port, service)
if protocol == LoadBalancerProtocolInvalid {
return nil, fmt.Errorf("unsupported load balancer protocol: %v", port.Protocol)
}
// All ports have their own load balancer rule, so add the port to lbName to keep the names unique.
lbRuleName := fmt.Sprintf("%s-%s-%d", lb.name, protocol, port.Port)
lbRule := lb.findLoadBalancerRule(lbRuleName, port, protocol)
change, err := lb.checkLoadBalancerRule(lbRule, lbRuleName, port, protocol, service, version)
if err != nil {
return nil, err
}
if lbRule != nil {
// Claim by the rule's actual name: after a protocol change it still carries the old one.
delete(lb.rules, lbRule.Name)
}
desired = append(desired, desiredLBRule{
name: lbRuleName,
port: port,
protocol: protocol,
existing: lbRule,
change: change,
})
}
return desired, nil
}
// findLoadBalancerRule locates the existing CloudStack rule for a desired service port. It
// prefers an exact name match, then falls back to matching on the tuple. That fallback is what
// lets a protocol change (tcp <-> tcp-proxy) update the existing rule instead of creating a
// conflicting one.
//
// Only rules on the IP being reconciled towards are eligible; a rule on any other IP is left
// for the prune pass, which also cleans up the firewall rules it leaves behind.
func (lb *loadBalancer) findLoadBalancerRule(lbRuleName string, port corev1.ServicePort, protocol LoadBalancerProtocol) *cloudstack.LoadBalancerRule {
if lbRule, ok := lb.rules[lbRuleName]; ok && lbRule.Publicipid == lb.ipAddrID {
return lbRule
}
publicPort := strconv.Itoa(int(port.Port))
var names []string
for name, lbRule := range lb.rules {
if lbRule.Publicipid == lb.ipAddrID &&
ProtocolFromLoadBalancer(lbRule.Protocol).IPProtocol() == protocol.IPProtocol() &&
lbRule.Publicport == publicPort {
names = append(names, name)
}
}
if len(names) == 0 {
return nil
}
// Map iteration order is randomized; sort so the pick is deterministic.
sort.Strings(names)
if len(names) > 1 {
klog.Warningf("Multiple load balancer rules match %s port %s: %v; using %v", protocol.IPProtocol(), publicPort, names, names[0])
}
return lb.rules[names[0]]
}
// portProtocol is the tuple CloudStack refuses to place two load balancer rules on, and that
// firewall and network ACL rules are keyed on. IPProtocol maps both tcp and tcp-proxy to
// "tcp", so a tcp and a tcp-proxy rule on one port share a tuple, and one firewall/ACL rule.
type portProtocol struct {
ipProtocol string
publicPort int32
}
// obsoleteRule is a rule no desired service port claimed, with its tuple already parsed.
type obsoleteRule struct {
rule *cloudstack.LoadBalancerRule
protocol LoadBalancerProtocol
tuple portProtocol
}
// partitionObsoleteRules splits the rules left in lb.rules — those no desired port claimed —
// into the ones holding a tuple that a rule still to be created needs, and the rest.
func (lb *loadBalancer) partitionObsoleteRules(desired []desiredLBRule) (blocking, rest []obsoleteRule) {
// CloudStack refuses two load balancer rules with overlapping public port ranges on one
// IP whatever their protocols, so the port alone decides what blocks a create. Note this
// is deliberately coarser than the firewall/ACL claim, which is per protocol because
// firewall rules are.
neededPorts := make(map[int32]bool)
for _, d := range desired {
if d.createsRule() {
neededPorts[d.port.Port] = true
}
}
// Iterate in name order so the prune sequence is reproducible.
names := make([]string, 0, len(lb.rules))
for name := range lb.rules {
names = append(names, name)
}
sort.Strings(names)
for _, name := range names {
lbRule := lb.rules[name]
port, err := strconv.ParseInt(lbRule.Publicport, 10, 32)
if err != nil {
klog.Warningf("Skipping obsolete load balancer rule %v (%v on public IP %v) with invalid public port %q: %v", lbRule.Name, lbRule.Id, lbRule.Publicipid, lbRule.Publicport, err)
continue
}
// Conflicts are per public IP, so only a rule on the IP being reconciled towards can
// block a create.
blocksACreate := lbRule.Publicipid == lb.ipAddrID && neededPorts[int32(port)]
// A protocol the provider cannot interpret leaves its firewall or ACL rule
// unresolvable, so such a rule is normally left alone. One holding a port a create
// needs still has to go, or CloudStack rejects that create as a port conflict.
protocol := ProtocolFromLoadBalancer(lbRule.Protocol)
if protocol == LoadBalancerProtocolInvalid && !blocksACreate {
klog.Warningf("Skipping obsolete load balancer rule %v (%v on public IP %v) with unknown protocol %q", lbRule.Name, lbRule.Id, lbRule.Publicipid, lbRule.Protocol)
continue
}
obsolete := obsoleteRule{
rule: lbRule,
protocol: protocol,
tuple: portProtocol{protocol.IPProtocol(), int32(port)},
}
if blocksACreate {
blocking = append(blocking, obsolete)
} else {
rest = append(rest, obsolete)
}
}
return blocking, rest
}
// ruleNetworkID is the network whose ACL rules an existing load balancer rule was opened in, or
// "" when that network cannot be established. CloudStack omits the network on rules of some
// network types; such a rule is known to belong to the network being reconciled towards only
// when it sits on the public IP being reconciled towards.
func (lb *loadBalancer) ruleNetworkID(lbRule *cloudstack.LoadBalancerRule) string {
if lbRule.Networkid != "" {
return lbRule.Networkid
}
if lbRule.Publicipid == lb.ipAddrID {
return lb.networkID
}
return ""
}
// claimedTuples are the tuples the desired service ports still need, and whose firewall or
// network ACL rules therefore have to survive a prune.
func claimedTuples(desired []desiredLBRule) map[portProtocol]bool {
claimed := make(map[portProtocol]bool, len(desired))
for _, d := range desired {
claimed[portProtocol{d.protocol.IPProtocol(), d.port.Port}] = true
}
return claimed
}
// pruneFirewallRule deletes the firewall rule admitting traffic to an obsolete load balancer
// rule. Firewall rules belong to a single public IP, so a claim only covers a rule on the IP
// the service is being reconciled towards.
func (lb *loadBalancer) pruneFirewallRule(o obsoleteRule, claimed map[portProtocol]bool) error {
lbRule, port := o.rule, int(o.tuple.publicPort)
if claimed[o.tuple] && lbRule.Publicipid == lb.ipAddrID {
klog.V(4).Infof("Keeping firewall rules of obsolete load balancer rule %v (%v:%v:%v): still claimed by a service port", lbRule.Name, o.protocol, lbRule.Publicip, port)
return nil
}
klog.V(4).Infof("Deleting firewall rules associated with load balancer rule: %v (%v:%v:%v)", lbRule.Name, o.protocol, lbRule.Publicip, port)
_, err := lb.deleteFirewallRule(lbRule.Publicipid, port, o.protocol)
return err
}
// pruneNetworkACLRule deletes the network ACL rule admitting traffic to an obsolete load
// balancer rule, in the network that rule belongs to. ACL rules belong to a network rather than
// an IP, so a claim only covers a rule in the network the service is being reconciled towards.
func (lb *loadBalancer) pruneNetworkACLRule(o obsoleteRule, claimed map[portProtocol]bool, networkID string) error {
lbRule, port := o.rule, int(o.tuple.publicPort)
if claimed[o.tuple] && networkID == lb.networkID {
klog.V(4).Infof("Keeping Network ACL rules of obsolete load balancer rule %v (%v:%v:%v): still claimed by a service port", lbRule.Name, o.protocol, networkID, port)
return nil
}
klog.V(4).Infof("Deleting Network ACL rules associated with load balancer rule: %v (%v:%v:%v)", lbRule.Name, o.protocol, networkID, port)
_, err := lb.deleteNetworkACLRule(port, o.protocol, networkID)
return err
}
// rememberNetwork records a network already fetched, so resolving the network of a rule in it
// costs no further call.
func (lb *loadBalancer) rememberNetwork(networkID string, network *cloudstack.Network) {
if lb.networks == nil {
lb.networks = make(map[string]*cloudstack.Network)
}
lb.networks[networkID] = network
}
// networkByID is the network with the given ID, or nil when CloudStack no longer has it. An
// empty ID is nil rather than a lookup, which GetNetworkByID would answer with an arbitrary
// network from an unfiltered list. GetNetworkByID reports not-found as an error with a count of
// 0, so the count has to be checked before the error.
func (lb *loadBalancer) networkByID(networkID string) (*cloudstack.Network, error) {
if networkID == "" {
return nil, nil
}
if network, ok := lb.networks[networkID]; ok {
return network, nil
}
network, count, err := lb.Network.GetNetworkByID(networkID, cloudstack.WithProject(lb.projectID))
switch {
case count == 0:
network = nil
case err != nil:
return nil, fmt.Errorf("error fetching network %v: %v", networkID, err)
}
lb.rememberNetwork(networkID, network)
return network, nil
}
// ruleNetwork is the network an existing load balancer rule was created in, or nil when that
// network cannot be established, either because CloudStack reported no network for the rule or
// because the network has since been deleted.
func (lb *loadBalancer) ruleNetwork(lbRule *cloudstack.LoadBalancerRule) (*cloudstack.Network, error) {
return lb.networkByID(lb.ruleNetworkID(lbRule))
}
// pruneRuleOpening deletes the firewall or network ACL rule admitting traffic to an obsolete
// load balancer rule, unless a desired service port still claims that same opening. Which of
// the two a rule has follows the network that rule belongs to, not the one being reconciled
// towards, so a rule left behind in a network of the other kind does not keep its opening.
//
// A rule with an uninterpretable protocol keeps its opening, which cannot be identified without
// one. A rule whose network cannot be established still has its firewall rule deleted, that
// being scoped to the rule's own public IP, while any ACL rule is left in place.
func (lb *loadBalancer) pruneRuleOpening(o obsoleteRule, claimed map[portProtocol]bool) error {
if o.protocol == LoadBalancerProtocolInvalid {
klog.Warningf("Leaving the firewall or Network ACL rule of obsolete load balancer rule %v in place: unknown protocol %v", o.rule.Name, o.rule.Protocol)
return nil
}
network, err := lb.ruleNetwork(o.rule)
if err != nil {
return err
}
switch {
case network == nil:
klog.Warningf("Cannot establish the network of obsolete load balancer rule %v; leaving any Network ACL rule of it in place", o.rule.Name)
return lb.pruneFirewallRule(o, claimed)
case isFirewallSupported(network.Service):
return lb.pruneFirewallRule(o, claimed)
case isNetworkACLSupported(network.Service):
return lb.pruneNetworkACLRule(o, claimed, network.Id)
}
return nil
}
// pruneRules deletes the given obsolete rules along with their firewall or network ACL rules.
// A firewall/ACL rule is kept when a desired port still claims the same tuple, since the two
// load balancer rules share it and pruning would strip the survivor of its opening.
//
// The network being reconciled towards is taken as already fetched, so only a rule belonging to
// some other network costs a lookup of its own.
//
// A rule that fails to delete is reported but does not stop the others being pruned.
func (lb *loadBalancer) pruneRules(obsolete []obsoleteRule, desired []desiredLBRule, network *cloudstack.Network) error {
lb.rememberNetwork(network.Id, network)
claimed := claimedTuples(desired)
var firstErr error
recordErr := func(err error) {
klog.Errorf("Error pruning obsolete load balancer rule: %v", err)
if firstErr == nil {
firstErr = err
}
}
for _, o := range obsolete {
if err := lb.pruneRuleOpening(o, claimed); err != nil {
recordErr(err)
continue
}
klog.V(4).Infof("Deleting obsolete load balancer rule: %v", o.rule.Name)
if err := lb.deleteLoadBalancerRule(o.rule); err != nil {
recordErr(err)
}
}
return firstErr
}
// 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. 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, lb.repairEmptyRule(d.existing)
case ruleNeedsUpdate:
klog.V(4).Infof("Updating load balancer rule: %v", d.name)
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 {
return nil, err
}
}
klog.V(4).Infof("Creating load balancer rule: %v", d.name)
lbRule, err := lb.createLoadBalancerRule(d.name, d.port, d.protocol, service)
if err != nil {
return nil, err
}
klog.V(4).Infof("Assigning hosts (%v) to load balancer rule: %v", lb.hostIDs, d.name)
if err := lb.assignHostsToRule(lbRule, lb.hostIDs); err != nil {
return nil, err
}
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. 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 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 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 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 &params[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
}
// effectiveSourceCIDRs is the source CIDR list a load balancer rule enforces. An empty list
// allows every source, since the virtual router's HAProxy configuration adds no source filter
// for it. Rules created by the in-tree provider, and by this one before it sent a CIDR list,
// report an empty list, and must compare equal to an unrestricted Service rather than be
// recreated on every CloudStack release that cannot update the list in place.
func effectiveSourceCIDRs(cidrs []string) []string {
if len(cidrs) == 0 {
return []string{defaultAllowedCIDR}
}
return cidrs
}
// checkLoadBalancerRule decides what applying a service port does to the given existing rule
// (nil if none was found): nothing, an update call, or deleting and creating it again. It makes
// no CloudStack call.
func (lb *loadBalancer) checkLoadBalancerRule(lbRule *cloudstack.LoadBalancerRule, lbRuleName string, port corev1.ServicePort, protocol LoadBalancerProtocol, service *corev1.Service, version semver.Version) (ruleChange, error) {
if lbRule == nil {
return ruleMissing, nil
}
cidrList, err := lb.getCIDRList(service)
if err != nil {
return ruleMissing, err
}
// Check if basic properties match (IP and ports). If not, we need to recreate the rule.
basicPropsMatch := lbRule.Publicip == lb.ipAddr &&
lbRule.Privateport == strconv.Itoa(int(port.NodePort)) &&
lbRule.Publicport == strconv.Itoa(int(port.Port))
cidrListChanged := !compareStringSlice(effectiveSourceCIDRs(cidrList), effectiveSourceCIDRs(splitCIDRList(lbRule.Cidrlist)))
updateProto := lbRule.Protocol != protocol.CSProtocol()
// A CIDR change on an older CloudStack can only be applied by recreating the rule.
if !basicPropsMatch || (cidrListChanged && version.LT(cidrListUpdateVersion)) {
return ruleNeedsRecreate, nil
}
updateAlgo := lbRule.Algorithm != lb.algorithm
// The name encodes the protocol, so a rule matched across a protocol change needs renaming.
updateName := lbRule.Name != lbRuleName
if updateAlgo || updateProto || updateName || cidrListChanged {
return ruleNeedsUpdate, nil
}
return ruleUpToDate, nil
}
// updateLoadBalancerRule updates a load balancer rule.
func (lb *loadBalancer) updateLoadBalancerRule(lbRule *cloudstack.LoadBalancerRule, lbRuleName string, protocol LoadBalancerProtocol, service *corev1.Service, version semver.Version) error {
p := lb.LoadBalancer.NewUpdateLoadBalancerRuleParams(lbRule.Id)
p.SetAlgorithm(lb.algorithm)
p.SetProtocol(protocol.CSProtocol())
p.SetName(lbRuleName)
// Only send the CIDR list where the API accepts it; checkLoadBalancerRule recreates the
// rule instead on older releases, so a change can never be silently dropped here.
if version.GTE(cidrListUpdateVersion) {
cidrList, err := lb.getCIDRList(service)
if err != nil {
return err
}
p.SetCidrlist(cidrList)
}
_, err := lb.LoadBalancer.UpdateLoadBalancerRule(p)
return err
}
// createLoadBalancerRule creates a new load balancer rule and returns it's ID.
func (lb *loadBalancer) createLoadBalancerRule(lbRuleName string, port corev1.ServicePort, protocol LoadBalancerProtocol, service *corev1.Service) (*cloudstack.LoadBalancerRule, error) {
p := lb.LoadBalancer.NewCreateLoadBalancerRuleParams(
lb.algorithm,
lbRuleName,
int(port.NodePort),
int(port.Port),
)
p.SetNetworkid(lb.networkID)
p.SetPublicipid(lb.ipAddrID)
p.SetProtocol(protocol.CSProtocol())
// Do not open the firewall implicitly, we always create explicit firewall rules
p.SetOpenfirewall(false)
// Read the source CIDR annotation
cidrList, err := lb.getCIDRList(service)
if err != nil {
return nil, err
}
// Set the CIDR list in the parameters
p.SetCidrlist(cidrList)
// Create a new load balancer rule.
r, err := lb.LoadBalancer.CreateLoadBalancerRule(p)
if err != nil {
return nil, fmt.Errorf("error creating load balancer rule %v: %v", lbRuleName, err)
}
lbRule := &cloudstack.LoadBalancerRule{
Id: r.Id,
Algorithm: r.Algorithm,
Cidrlist: r.Cidrlist,
Name: r.Name,
Networkid: r.Networkid,
Privateport: r.Privateport,
Publicport: r.Publicport,
Publicip: r.Publicip,
Publicipid: r.Publicipid,
Protocol: r.Protocol,
}
return lbRule, nil
}
// deleteLoadBalancerRule deletes a load balancer rule.
func (lb *loadBalancer) deleteLoadBalancerRule(lbRule *cloudstack.LoadBalancerRule) error {
p := lb.LoadBalancer.NewDeleteLoadBalancerRuleParams(lbRule.Id)
if _, err := lb.LoadBalancer.DeleteLoadBalancerRule(p); err != nil {
return fmt.Errorf("error deleting load balancer rule %v: %v", lbRule.Name, err)
}
// A duplicate shares its name with the rule being kept, which owns the map entry.
if kept, ok := lb.rules[lbRule.Name]; ok && kept.Id == lbRule.Id {
delete(lb.rules, lbRule.Name)
}
return nil
}
// deleteDuplicateRules removes rules that collided by name with the one being
// managed. A duplicate sits on its own public IP, since CloudStack rejects a
// second rule on the same IP and port, so its firewall rule and IP go with it;
// network ACL rules are shared per tier and port with the kept rule and stay.
func (lb *loadBalancer) deleteDuplicateRules() error {
for _, lbRule := range lb.duplicateRules {
klog.V(4).Infof("Deleting duplicate load balancer rule: %v (%v)", lbRule.Name, lbRule.Id)
if err := lb.deleteDuplicateRule(lbRule); err != nil {
return err
}
}
lb.duplicateRules = nil
return nil
}
// deleteDuplicateRule tears down one duplicate: its firewall rule, the rule
// itself, and its public IP once no other rule uses that IP.
func (lb *loadBalancer) deleteDuplicateRule(lbRule *cloudstack.LoadBalancerRule) error {
port, err := strconv.Atoi(lbRule.Publicport)
protocol := ProtocolFromLoadBalancer(lbRule.Protocol)
if err != nil || protocol == LoadBalancerProtocolInvalid {
klog.Warningf("Leaving duplicate rule %v (%v) in place: unusable public port %q or protocol %q", lbRule.Name, lbRule.Id, lbRule.Publicport, lbRule.Protocol)
return nil
}
if _, err := lb.deleteFirewallRule(lbRule.Publicipid, port, protocol); err != nil {
return err
}
if err := lb.deleteLoadBalancerRule(lbRule); err != nil {
return err
}
if lbRule.Publicipid == lb.ipAddrID {
return nil
}
inUse, err := lb.publicIPHasRules(lbRule.Publicipid)
if err != nil || inUse {
return err
}
p := lb.Address.NewDisassociateIpAddressParams(lbRule.Publicipid)
if _, err := lb.Address.DisassociateIpAddress(p); err != nil {
return fmt.Errorf("error releasing public IP %v: %v", lbRule.Publicipid, err)
}
return nil
}
// publicIPHasRules reports whether any load balancer rule still uses the IP.
func (lb *loadBalancer) publicIPHasRules(publicIPID string) (bool, error) {
p := lb.LoadBalancer.NewListLoadBalancerRulesParams()
p.SetPublicipid(publicIPID)
p.SetListall(true)
if lb.projectID != "" {
p.SetProjectid(lb.projectID)
}
rules, err := lb.LoadBalancer.ListLoadBalancerRules(p)
if err != nil {
return false, fmt.Errorf("error listing load balancer rules on IP %v: %v", publicIPID, err)
}
return rules.Count > 0, nil
}
// assignHostsToRule assigns hosts to a load balancer rule.
func (lb *loadBalancer) assignHostsToRule(lbRule *cloudstack.LoadBalancerRule, hostIDs []string) error {
p := lb.LoadBalancer.NewAssignToLoadBalancerRuleParams(lbRule.Id)
p.SetVirtualmachineids(hostIDs)
if _, err := lb.LoadBalancer.AssignToLoadBalancerRule(p); err != nil {
return fmt.Errorf("error assigning hosts to load balancer rule %v: %v", lbRule.Name, err)
}
return nil
}
// removeHostsFromRule removes hosts from a load balancer rule.
func (lb *loadBalancer) removeHostsFromRule(lbRule *cloudstack.LoadBalancerRule, hostIDs []string) error {
p := lb.LoadBalancer.NewRemoveFromLoadBalancerRuleParams(lbRule.Id)
p.SetVirtualmachineids(hostIDs)
if _, err := lb.LoadBalancer.RemoveFromLoadBalancerRule(p); err != nil {
return fmt.Errorf("error removing hosts from load balancer rule %v: %v", lbRule.Name, err)
}
return nil
}
// symmetricDifference returns the symmetric difference between the old (existing) and new (wanted) host ID's.
func symmetricDifference(hostIDs []string, lbInstances []*cloudstack.VirtualMachine) ([]string, []string) {
new := make(map[string]bool)
for _, hostID := range hostIDs {
new[hostID] = true
}
// Paging over the instances of a rule can return the same instance twice. A
// duplicate would otherwise be dropped from new on its first occurrence and
// then added to remove on its second, so the same host would be both kept
// and removed.
seen := make(map[string]bool)
var remove []string
for _, instance := range lbInstances {
if seen[instance.Id] {
continue
}
seen[instance.Id] = true
if new[instance.Id] {
delete(new, instance.Id)
continue
}
remove = append(remove, instance.Id)
}
var assign []string
for hostID := range new {
assign = append(assign, hostID)
}
return assign, remove
}
// compareStringSlice compares two unsorted slices of strings without sorting them first.
//
// The slices are equal if and only if both contain the same number of every unique element.
//
// Thanks to: https://stackoverflow.com/a/36000696
func compareStringSlice(x, y []string) bool {
if len(x) != len(y) {
return false
}
// create a map of string -> int
diff := make(map[string]int, len(x))
for _, _x := range x {
// 0 value for int is 0, so just increment a counter for the string
diff[_x]++
}
for _, _y := range y {
// If the string _y is not in diff bail out early
if _, ok := diff[_y]; !ok {
return false
}
diff[_y] -= 1
if diff[_y] == 0 {
delete(diff, _y)
}
}
return len(diff) == 0
}
func ruleToString(rule *cloudstack.FirewallRule) string {
ls := &strings.Builder{}
if rule == nil {
ls.WriteString("nil")
} else {
switch rule.Protocol {
case "tcp":
fallthrough
case "udp":
fmt.Fprintf(ls, "{[%s] -> %s:[%d-%d] (%s)}", rule.Cidrlist, rule.Ipaddress, rule.Startport, rule.Endport, rule.Protocol)
case "icmp":
fmt.Fprintf(ls, "{[%s] -> %s [%d,%d] (%s)}", rule.Cidrlist, rule.Ipaddress, rule.Icmptype, rule.Icmpcode, rule.Protocol)
default:
fmt.Fprintf(ls, "{[%s] -> %s (%s)}", rule.Cidrlist, rule.Ipaddress, rule.Protocol)
}
}
return ls.String()
}
func rulesToString(rules []*cloudstack.FirewallRule) string {
ls := &strings.Builder{}
first := true
for _, rule := range rules {
if first {
first = false
} else {
ls.WriteString(", ")
}
ls.WriteString(ruleToString(rule))
}
return ls.String()
}
func rulesMapToString(rules map[*cloudstack.FirewallRule]bool) string {
ls := &strings.Builder{}
first := true
for rule := range rules {
if first {
first = false
} else {
ls.WriteString(", ")
}
ls.WriteString(ruleToString(rule))
}
return ls.String()
}
// listFirewallRules retrieves all firewall rules associated with a public IP.
//
// Rules are deduplicated by ID: paging over a set that is changing underneath us
// can return the same rule on more than one page, and each page decodes into its
// own struct, so a repeat arrives as a second pointer to an equal rule. Callers
// key their bookkeeping on the pointer, which would treat the two as unrelated
// rules and delete one of them.
func (lb *loadBalancer) listFirewallRules(publicIpId string) ([]*cloudstack.FirewallRule, error) {
p := lb.Firewall.NewListFirewallRulesParams()
p.SetIpaddressid(publicIpId)
p.SetListall(true)
if lb.projectID != "" {
p.SetProjectid(lb.projectID)
}
klog.V(4).Infof("Listing firewall rules for %v", p)
rules, err := listAll(p, func() (int, []*cloudstack.FirewallRule, error) {
r, err := lb.Firewall.ListFirewallRules(p)
if err != nil {
return 0, nil, err
}
return r.Count, r.FirewallRules, nil
})
if err != nil {
return nil, fmt.Errorf("error fetching firewall rules for public IP %v: %v", publicIpId, err)
}
return dedupeByID(rules, func(rule *cloudstack.FirewallRule) string { return rule.Id }), nil
}
// updateFirewallRule creates a firewall rule for a load balancer rule
//
// If the rule list is empty, all internet (IPv4: 0.0.0.0/0) is opened for the
// load balancer's port+protocol implicitly.
//
// Returns true if the firewall rule was created or updated
func (lb *loadBalancer) updateFirewallRule(publicIpId string, publicPort int, protocol LoadBalancerProtocol, allowedIPs []string) (bool, error) {
if len(allowedIPs) == 0 {
allowedIPs = []string{defaultAllowedCIDR}
}
firewallRules, err := lb.listFirewallRules(publicIpId)
if err != nil {
return false, err
}
klog.V(4).Infof("All firewall rules for %v: %v", lb.ipAddr, rulesToString(firewallRules))
// find all rules that have a matching proto+port
// a map may or may not be faster, but is a bit easier to understand
filtered := make(map[*cloudstack.FirewallRule]bool)
for _, rule := range firewallRules {
if rule.Protocol == protocol.IPProtocol() && rule.Startport == publicPort && rule.Endport == publicPort {
filtered[rule] = true
}
}
klog.V(4).Infof("Matching rules for %v: %v", lb.ipAddr, rulesMapToString(filtered))
// determine if we already have a rule with matching cidrs
var match *cloudstack.FirewallRule
for rule := range filtered {
cidrlist := splitCIDRList(rule.Cidrlist)
if compareStringSlice(cidrlist, allowedIPs) {
klog.V(4).Infof("Found identical rule: %v", rule)
match = rule
break
}
}
if match != nil {
// no need to create a new rule - but prevent deletion of the matching rule
delete(filtered, match)
}
// delete all other rules that didn't match the CIDR list
// do this first to prevent CS rule conflict errors
klog.V(4).Infof("Firewall rules to be deleted for %v: %v", lb.ipAddr, rulesMapToString(filtered))
for rule := range filtered {
p := lb.Firewall.NewDeleteFirewallRuleParams(rule.Id)
_, err = lb.Firewall.DeleteFirewallRule(p)
if err != nil {
// report the error, but keep on deleting the other rules
klog.Errorf("Error deleting old firewall rule %v: %v", rule.Id, err)
}
}
// create new rule if necessary
if match == nil {
// no rule found, create a new one
p := lb.Firewall.NewCreateFirewallRuleParams(publicIpId, protocol.IPProtocol())
p.SetCidrlist(allowedIPs)
p.SetStartport(publicPort)
p.SetEndport(publicPort)
_, err = lb.Firewall.CreateFirewallRule(p)
if err != nil {
// return immediately if we can't create the new rule
return false, fmt.Errorf("error creating new firewall rule for public IP %v, proto %v, port %v, allowed %v: %v", publicIpId, protocol, publicPort, allowedIPs, err)
}
}
// return true (because we changed something), but also the last error if deleting one old rule failed
return true, err
}
func (lb *loadBalancer) updateNetworkACL(publicPort int, protocol LoadBalancerProtocol, networkId string) (bool, error) {
network, _, err := lb.Network.GetNetworkByID(networkId, cloudstack.WithProject(lb.projectID))
if err != nil {
return false, fmt.Errorf("error fetching Network with ID: %v, due to: %s", networkId, err)
}
networkAclList, count, err := lb.NetworkACL.GetNetworkACLListByID(network.Aclid, cloudstack.WithProject(lb.projectID))
if err != nil {
return false, fmt.Errorf("error fetching Network ACL List with ID: %v, due to: %s", network.Aclid, err)
}
if count == 0 {
return false, fmt.Errorf("failed to find network ACL List with id: %v", network.Aclid)
}
if networkAclList.Name == "default_allow" || networkAclList.Name == "default_deny" {
klog.Infof("Network is using a default network ACL. Cannot add ACL rules to default ACLs")
return true, err
}
networkAclParams := lb.NetworkACL.NewListNetworkACLsParams()
networkAclParams.SetAclid(network.Aclid)
networkAclParams.SetNetworkid(networkId)
networkAclParams.SetListall(true)
if lb.projectID != "" {
networkAclParams.SetProjectid(lb.projectID)
}
networkAcls, err := listAll(networkAclParams, func() (int, []*cloudstack.NetworkACL, error) {
networkAclResponse, err := lb.NetworkACL.ListNetworkACLs(networkAclParams)
if err != nil {
return 0, nil, err
}
return networkAclResponse.Count, networkAclResponse.NetworkACLs, nil
})
if err != nil {
return false, fmt.Errorf("error fetching Network ACL with ID: %v for network with id: %v, due to: %s", network.Aclid, networkId, err)
}
// find all network ACL rules that have a matching proto+port
// a map may or may not be faster, but is a bit easier to understand
filtered := make(map[*cloudstack.NetworkACL]bool)
for _, netAclRule := range networkAcls {
if netAclRule.Protocol == protocol.IPProtocol() && netAclRule.Startport == strconv.Itoa(publicPort) && netAclRule.Endport == strconv.Itoa(publicPort) {
filtered[netAclRule] = true
}
}
if len(filtered) > 0 {
klog.V(4).Infof("Network ACL rule for port %v and protocol %v already exists. No need to added a duplicate rule", publicPort, protocol)
return true, err
}
// create ACL rule
// ACL rules only know tcp/udp/icmp, so tcp-proxy maps to tcp. This also matches the
// filter above, which would otherwise never find the rule again.
acl := lb.NetworkACL.NewCreateNetworkACLParams(protocol.IPProtocol())
acl.SetAclid(network.Aclid)
acl.SetAction("Allow")
acl.SetCidrlist([]string{"0.0.0.0/0"})
acl.SetStartport(publicPort)
acl.SetEndport(publicPort)
acl.SetNetworkid(networkId)
acl.SetTraffictype("Ingress")
_, err = lb.NetworkACL.CreateNetworkACL(acl)
if err != nil {
return false, fmt.Errorf("error creating Network ACL for port: %v, due to: %s", publicPort, err)
}
return true, err
}
// deleteFirewallRule deletes the firewall rule associated with the ip:port:protocol combo
//
// returns true when corresponding rules were deleted
func (lb *loadBalancer) deleteFirewallRule(publicIpId string, publicPort int, protocol LoadBalancerProtocol) (bool, error) {
firewallRules, err := lb.listFirewallRules(publicIpId)
if err != nil {
return false, err
}
// filter by proto:port
filtered := make([]*cloudstack.FirewallRule, 0, 1)
for _, rule := range firewallRules {
if rule.Protocol == protocol.IPProtocol() && rule.Startport == publicPort && rule.Endport == publicPort {
filtered = append(filtered, rule)
}
}
// delete all rules
deleted := false
for _, rule := range filtered {
p := lb.Firewall.NewDeleteFirewallRuleParams(rule.Id)
_, err = lb.Firewall.DeleteFirewallRule(p)
if err != nil {
klog.Errorf("Error deleting old firewall rule %v: %v", rule.Id, err)
} else {
deleted = true
}
}
return deleted, err
}
// Delete Network ACLs deletes the Network ACL rule associated with the ip:port:protocol combo
func (lb *loadBalancer) deleteNetworkACLRule(publicPort int, protocol LoadBalancerProtocol, networkID string) (bool, error) {
p := lb.NetworkACL.NewListNetworkACLsParams()
p.SetListall(true)
p.SetNetworkid(networkID)
if lb.projectID != "" {
p.SetProjectid(lb.projectID)
}
networkAcls, err := listAll(p, func() (int, []*cloudstack.NetworkACL, error) {
r, err := lb.NetworkACL.ListNetworkACLs(p)
if err != nil {
return 0, nil, err
}
return r.Count, r.NetworkACLs, nil
})
if err != nil {
return false, fmt.Errorf("error fetching Network ACL rules Network ID %v: %v", networkID, err)
}
// filter by proto:port
filtered := make([]*cloudstack.NetworkACL, 0, 1)
for _, rule := range networkAcls {
if rule.Protocol == protocol.IPProtocol() && rule.Startport == strconv.Itoa(publicPort) && rule.Endport == strconv.Itoa(publicPort) {
filtered = append(filtered, rule)
}
}
// delete first filtered rules
if len(filtered) == 0 {
klog.V(4).Infof("No ACL rules found matching protocol: %v and port: %v", protocol, publicPort)
return true, nil
}
deleted := false
ruleToBeDeleted := filtered[0]
deleteAclParams := lb.NetworkACL.NewDeleteNetworkACLParams(ruleToBeDeleted.Id)
_, err = lb.NetworkACL.DeleteNetworkACL(deleteAclParams)
if err != nil {
klog.Errorf("Error deleting old Network ACL rule %v: %v", ruleToBeDeleted.Id, err)
} else {
deleted = true
}
return deleted, err
}
// getStringFromServiceAnnotation searches a given v1.Service for a specific annotationKey and either returns the annotation's value or a specified defaultSetting
func getStringFromServiceAnnotation(service *corev1.Service, annotationKey string, defaultSetting string) string {
klog.V(4).Infof("getStringFromServiceAnnotation(%s/%s, %v, %v)", service.Namespace, service.Name, annotationKey, defaultSetting)
if annotationValue, ok := service.Annotations[annotationKey]; ok {
//if there is an annotation for this setting, set the "setting" var to it
// annotationValue can be empty, it is working as designed
// it makes possible for instance provisioning loadbalancer without floatingip
klog.V(4).Infof("Found a Service Annotation: %v = %v", annotationKey, annotationValue)
return annotationValue
}
//if there is no annotation, set "settings" var to the value from cloud config
if defaultSetting != "" {
klog.V(4).Infof("Could not find a Service Annotation; falling back on cloud-config setting: %v = %v", annotationKey, defaultSetting)
}
return defaultSetting
}
// getBoolFromServiceAnnotation searches a given v1.Service for a specific annotationKey and either returns the annotation's boolean value or a specified defaultSetting
func getBoolFromServiceAnnotation(service *corev1.Service, annotationKey string, defaultSetting bool) bool {
klog.V(4).Infof("getBoolFromServiceAnnotation(%s/%s, %v, %v)", service.Namespace, service.Name, annotationKey, defaultSetting)
if annotationValue, ok := service.Annotations[annotationKey]; ok {
returnValue := false
switch annotationValue {
case "true":
returnValue = true
case "false":
returnValue = false
default:
returnValue = defaultSetting
}
klog.V(4).Infof("Found a Service Annotation: %v = %v", annotationKey, returnValue)
return returnValue
}
klog.V(4).Infof("Could not find a Service Annotation; falling back to default setting: %v = %v", annotationKey, defaultSetting)
return defaultSetting
}