Skip to content

Commit 59efe1c

Browse files
[occm] add a node selector support for loadbalancer services (kubernetes#2601)
* POC of TargetNodeLabels selector on OpenStack LB * Fix type errors * Update implementation of getKeyValuePropertiesFromServiceAnnotation * gofmt -w -s ./pkg * Polish the code and add documentation --------- Co-authored-by: Ririko Nakamura <[email protected]>
1 parent 309db6d commit 59efe1c

File tree

7 files changed

+277
-8
lines changed

7 files changed

+277
-8
lines changed

docs/openstack-cloud-controller-manager/expose-applications-using-loadbalancer-type-service.md

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -236,6 +236,12 @@ Request Body:
236236
This annotation is automatically added and it contains the floating ip address of the load balancer service.
237237
When using `loadbalancer.openstack.org/hostname` annotation it is the only place to see the real address of the load balancer.
238238

239+
- `loadbalancer.openstack.org/node-selector`
240+
241+
A set of key=value annotations used to filter nodes for targeting by the load balancer. When defined, only nodes that match all the specified key=value annotations will be targeted. If an annotation includes only a key without a value, the filter will check only for the existence of the key on the node. If the value is not set, the `node-selector` value defined in the OCCM configuration is applied.
242+
243+
Example: To filter nodes with the labels `env=production` and `region=default`, set the `loadbalancer.openstack.org/node-selector` annotation to `env=production, region=default`
244+
239245
### Switching between Floating Subnets by using preconfigured Classes
240246

241247
If you have multiple `FloatingIPPools` and/or `FloatingIPSubnets` it might be desirable to offer the user logical meanings for `LoadBalancers` like `internetFacing` or `DMZ` instead of requiring the user to select a dedicated network or subnet ID at the service object level as an annotation.

docs/openstack-cloud-controller-manager/using-openstack-cloud-controller-manager.md

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -207,7 +207,7 @@ Although the openstack-cloud-controller-manager was initially implemented with N
207207
* `ROUND_ROBIN` (default)
208208
* `LEAST_CONNECTIONS`
209209
* `SOURCE_IP`
210-
210+
211211
If `lb-provider` is set to "ovn" the value must be set to `SOURCE_IP_PORT`.
212212
213213
* `lb-provider`
@@ -248,6 +248,23 @@ Although the openstack-cloud-controller-manager was initially implemented with N
248248
* `internal-lb`
249249
Determines whether or not to create an internal load balancer (no floating IP) by default. Default: false.
250250
251+
* `node-selector`
252+
A comma separated list of key=value annotations used to filter nodes for targeting by the load balancer. When defined, only nodes that match all the specified key=value annotations will be targeted. If an annotation includes only a key without a value, the filter will check only for the existence of the key on the node. When node-selector is not set (default value), all nodes will be added as members to a load balancer pool.
253+
254+
Note: This configuration option can be overridden with the `loadbalancer.openstack.org/node-selector` service annotation. Refer to [Exposing applications using services of LoadBalancer type](./expose-applications-using-loadbalancer-type-service.md)
255+
256+
Example: To filter nodes with the labels `env=production` and `region=default`, set the `node-selector` as follows:
257+
258+
```
259+
node-selector="env=production, region=default"
260+
```
261+
262+
Example: To filter nodes that have the key `env` with any value and the key `region` specifically set to `default`, set the `node-selector` as follows:
263+
264+
```
265+
node-selector="env, region=default"
266+
```
267+
251268
* `cascade-delete`
252269
Determines whether or not to perform cascade deletion of load balancers. Default: true.
253270

pkg/openstack/loadbalancer.go

Lines changed: 74 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@ const (
5656
annotationXForwardedFor = "X-Forwarded-For"
5757

5858
ServiceAnnotationLoadBalancerInternal = "service.beta.kubernetes.io/openstack-internal-load-balancer"
59+
ServiceAnnotationLoadBalancerNodeSelector = "loadbalancer.openstack.org/node-selector"
5960
ServiceAnnotationLoadBalancerConnLimit = "loadbalancer.openstack.org/connection-limit"
6061
ServiceAnnotationLoadBalancerFloatingNetworkID = "loadbalancer.openstack.org/floating-network-id"
6162
ServiceAnnotationLoadBalancerFloatingSubnet = "loadbalancer.openstack.org/floating-subnet"
@@ -119,6 +120,7 @@ type serviceConfig struct {
119120
lbMemberSubnetID string
120121
lbPublicNetworkID string
121122
lbPublicSubnetSpec *floatingSubnetSpec
123+
nodeSelectors map[string]string
122124
keepClientIP bool
123125
enableProxyProtocol bool
124126
timeoutClientData int
@@ -405,6 +407,14 @@ func nodeAddressForLB(node *corev1.Node, preferredIPFamily corev1.IPFamily) (str
405407
return "", cpoerrors.ErrNoAddressFound
406408
}
407409

410+
// getKeyValueFromServiceAnnotation converts a comma-separated list of key-value
411+
// pairs from the specified annotation into a map or returns the specified
412+
// defaultSetting if the annotation is empty
413+
func getKeyValueFromServiceAnnotation(service *corev1.Service, annotationKey string, defaultSetting string) map[string]string {
414+
annotationValue := getStringFromServiceAnnotation(service, annotationKey, defaultSetting)
415+
return cpoutil.StringToMap(annotationValue)
416+
}
417+
408418
// getStringFromServiceAnnotation searches a given v1.Service for a specific annotationKey and either returns the annotation's value or a specified defaultSetting
409419
func getStringFromServiceAnnotation(service *corev1.Service, annotationKey string, defaultSetting string) string {
410420
klog.V(4).Infof("getStringFromServiceAnnotation(%s/%s, %v, %v)", service.Namespace, service.Name, annotationKey, defaultSetting)
@@ -1229,6 +1239,16 @@ func (lbaas *LbaasV2) checkServiceUpdate(service *corev1.Service, nodes []*corev
12291239
svcConf.lbID = getStringFromServiceAnnotation(service, ServiceAnnotationLoadBalancerID, "")
12301240
svcConf.supportLBTags = openstackutil.IsOctaviaFeatureSupported(lbaas.lb, openstackutil.OctaviaFeatureTags, lbaas.opts.LBProvider)
12311241

1242+
// Get service node-selector annotations
1243+
svcConf.nodeSelectors = getKeyValueFromServiceAnnotation(service, ServiceAnnotationLoadBalancerNodeSelector, lbaas.opts.NodeSelector)
1244+
for key, value := range svcConf.nodeSelectors {
1245+
if value == "" {
1246+
klog.V(3).InfoS("Target node label %s key is set to LoadBalancer service %s", key, serviceName)
1247+
} else {
1248+
klog.V(3).InfoS("Target node label %s=%s is set to LoadBalancer service %s", key, value, serviceName)
1249+
}
1250+
}
1251+
12321252
// Find subnet ID for creating members
12331253
memberSubnetID, err := lbaas.getMemberSubnetID(service)
12341254
if err != nil {
@@ -1314,6 +1334,16 @@ func (lbaas *LbaasV2) checkService(service *corev1.Service, nodes []*corev1.Node
13141334
svcConf.lbID = getStringFromServiceAnnotation(service, ServiceAnnotationLoadBalancerID, "")
13151335
svcConf.supportLBTags = openstackutil.IsOctaviaFeatureSupported(lbaas.lb, openstackutil.OctaviaFeatureTags, lbaas.opts.LBProvider)
13161336

1337+
// Get service node-selector annotations
1338+
svcConf.nodeSelectors = getKeyValueFromServiceAnnotation(service, ServiceAnnotationLoadBalancerNodeSelector, lbaas.opts.NodeSelector)
1339+
for key, value := range svcConf.nodeSelectors {
1340+
if value == "" {
1341+
klog.V(3).InfoS("Target node label %s key is set to LoadBalancer service %s", key, serviceName)
1342+
} else {
1343+
klog.V(3).InfoS("Target node label %s=%s is set to LoadBalancer service %s", key, value, serviceName)
1344+
}
1345+
}
1346+
13171347
// If in the config file internal-lb=true, user is not allowed to create external service.
13181348
if lbaas.opts.InternalLB {
13191349
if !getBoolFromServiceAnnotation(service, ServiceAnnotationLoadBalancerInternal, false) {
@@ -1602,6 +1632,9 @@ func (lbaas *LbaasV2) ensureOctaviaLoadBalancer(ctx context.Context, clusterName
16021632
return nil, err
16031633
}
16041634

1635+
// apply node-selector to a list of nodes
1636+
filteredNodes := filterNodes(nodes, svcConf.nodeSelectors)
1637+
16051638
// Use more meaningful name for the load balancer but still need to check the legacy name for backward compatibility.
16061639
lbName := lbaas.GetLoadBalancerName(ctx, clusterName, service)
16071640
svcConf.lbName = lbName
@@ -1666,7 +1699,7 @@ func (lbaas *LbaasV2) ensureOctaviaLoadBalancer(ctx context.Context, clusterName
16661699
return nil, fmt.Errorf("error getting loadbalancer for Service %s: %v", serviceName, err)
16671700
}
16681701
klog.InfoS("Creating loadbalancer", "lbName", lbName, "service", klog.KObj(service))
1669-
loadbalancer, err = lbaas.createOctaviaLoadBalancer(lbName, clusterName, service, nodes, svcConf)
1702+
loadbalancer, err = lbaas.createOctaviaLoadBalancer(lbName, clusterName, service, filteredNodes, svcConf)
16701703
if err != nil {
16711704
return nil, fmt.Errorf("error creating loadbalancer %s: %v", lbName, err)
16721705
}
@@ -1712,7 +1745,7 @@ func (lbaas *LbaasV2) ensureOctaviaLoadBalancer(ctx context.Context, clusterName
17121745
return nil, err
17131746
}
17141747

1715-
pool, err := lbaas.ensureOctaviaPool(loadbalancer.ID, cpoutil.Sprintf255(poolFormat, portIndex, lbName), listener, service, port, nodes, svcConf)
1748+
pool, err := lbaas.ensureOctaviaPool(loadbalancer.ID, cpoutil.Sprintf255(poolFormat, portIndex, lbName), listener, service, port, filteredNodes, svcConf)
17161749
if err != nil {
17171750
return nil, err
17181751
}
@@ -1765,7 +1798,7 @@ func (lbaas *LbaasV2) ensureOctaviaLoadBalancer(ctx context.Context, clusterName
17651798
status := lbaas.createLoadBalancerStatus(service, svcConf, addr)
17661799

17671800
if lbaas.opts.ManageSecurityGroups {
1768-
err := lbaas.ensureAndUpdateOctaviaSecurityGroup(clusterName, service, nodes, svcConf)
1801+
err := lbaas.ensureAndUpdateOctaviaSecurityGroup(clusterName, service, filteredNodes, svcConf)
17691802
if err != nil {
17701803
return status, fmt.Errorf("failed when reconciling security groups for LB service %v/%v: %v", service.Namespace, service.Name, err)
17711804
}
@@ -1818,8 +1851,11 @@ func (lbaas *LbaasV2) updateOctaviaLoadBalancer(ctx context.Context, clusterName
18181851
return err
18191852
}
18201853

1854+
// apply node-selector to a list of nodes
1855+
filteredNodes := filterNodes(nodes, svcConf.nodeSelectors)
1856+
18211857
serviceName := fmt.Sprintf("%s/%s", service.Namespace, service.Name)
1822-
klog.V(2).Infof("Updating %d nodes for Service %s in cluster %s", len(nodes), serviceName, clusterName)
1858+
klog.V(2).Infof("Updating %d nodes for Service %s in cluster %s", len(filteredNodes), serviceName, clusterName)
18231859

18241860
// Get load balancer
18251861
var loadbalancer *loadbalancers.LoadBalancer
@@ -1866,7 +1902,7 @@ func (lbaas *LbaasV2) updateOctaviaLoadBalancer(ctx context.Context, clusterName
18661902
return fmt.Errorf("loadbalancer %s does not contain required listener for port %d and protocol %s", loadbalancer.ID, port.Port, port.Protocol)
18671903
}
18681904

1869-
pool, err := lbaas.ensureOctaviaPool(loadbalancer.ID, cpoutil.Sprintf255(poolFormat, portIndex, loadbalancer.Name), &listener, service, port, nodes, svcConf)
1905+
pool, err := lbaas.ensureOctaviaPool(loadbalancer.ID, cpoutil.Sprintf255(poolFormat, portIndex, loadbalancer.Name), &listener, service, port, filteredNodes, svcConf)
18701906
if err != nil {
18711907
return err
18721908
}
@@ -1878,7 +1914,7 @@ func (lbaas *LbaasV2) updateOctaviaLoadBalancer(ctx context.Context, clusterName
18781914
}
18791915

18801916
if lbaas.opts.ManageSecurityGroups {
1881-
err := lbaas.ensureAndUpdateOctaviaSecurityGroup(clusterName, service, nodes, svcConf)
1917+
err := lbaas.ensureAndUpdateOctaviaSecurityGroup(clusterName, service, filteredNodes, svcConf)
18821918
if err != nil {
18831919
return fmt.Errorf("failed to update Security Group for loadbalancer service %s: %v", serviceName, err)
18841920
}
@@ -2184,3 +2220,35 @@ func PreserveGopherError(rawError error) error {
21842220
}
21852221
return rawError
21862222
}
2223+
2224+
// filterNodes uses node labels to filter the nodes that should be targeted by the LB,
2225+
// ensuring that all the labels provided in an annotation are present on the nodes
2226+
func filterNodes(nodes []*corev1.Node, filterLabels map[string]string) []*corev1.Node {
2227+
if len(filterLabels) == 0 {
2228+
return nodes
2229+
}
2230+
2231+
filteredNodes := make([]*corev1.Node, 0, len(nodes))
2232+
for _, node := range nodes {
2233+
if matchNodeLabels(node, filterLabels) {
2234+
filteredNodes = append(filteredNodes, node)
2235+
}
2236+
}
2237+
2238+
return filteredNodes
2239+
}
2240+
2241+
// matchNodeLabels checks if a node has all the labels in filterLabels with matching values
2242+
func matchNodeLabels(node *corev1.Node, filterLabels map[string]string) bool {
2243+
if node == nil || len(node.Labels) == 0 {
2244+
return false
2245+
}
2246+
2247+
for k, v := range filterLabels {
2248+
if nodeLabelValue, ok := node.Labels[k]; !ok || (v != "" && nodeLabelValue != v) {
2249+
return false
2250+
}
2251+
}
2252+
2253+
return true
2254+
}

pkg/openstack/loadbalancer_test.go

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2452,3 +2452,104 @@ func TestBuildListenerCreateOpt(t *testing.T) {
24522452
})
24532453
}
24542454
}
2455+
2456+
func TestFilterNodes(t *testing.T) {
2457+
tests := []struct {
2458+
name string
2459+
nodeLabels map[string]string
2460+
service *corev1.Service
2461+
annotationKey string
2462+
defaultSetting map[string]string
2463+
nodeFiltered bool
2464+
}{
2465+
{
2466+
name: "when no filter is provided, node should be filtered",
2467+
nodeLabels: map[string]string{"k1": "v1"},
2468+
service: &corev1.Service{
2469+
ObjectMeta: v1.ObjectMeta{},
2470+
},
2471+
annotationKey: ServiceAnnotationLoadBalancerNodeSelector,
2472+
defaultSetting: make(map[string]string),
2473+
nodeFiltered: true,
2474+
},
2475+
{
2476+
name: "when all key-value filters match, node should be filtered",
2477+
nodeLabels: map[string]string{"k1": "v1", "k2": "v2"},
2478+
service: &corev1.Service{
2479+
ObjectMeta: v1.ObjectMeta{
2480+
Annotations: map[string]string{ServiceAnnotationLoadBalancerNodeSelector: "k1=v1,k2=v2"},
2481+
},
2482+
},
2483+
annotationKey: ServiceAnnotationLoadBalancerNodeSelector,
2484+
defaultSetting: make(map[string]string),
2485+
nodeFiltered: true,
2486+
},
2487+
{
2488+
name: "when all key-value filters match and a key value contains equals sign, node should be filtered",
2489+
nodeLabels: map[string]string{"k1": "v1", "k2": "v2=true"},
2490+
service: &corev1.Service{
2491+
ObjectMeta: v1.ObjectMeta{
2492+
Annotations: map[string]string{ServiceAnnotationLoadBalancerNodeSelector: "k1=v1,k2=v2=true"},
2493+
},
2494+
},
2495+
annotationKey: ServiceAnnotationLoadBalancerNodeSelector,
2496+
defaultSetting: make(map[string]string),
2497+
nodeFiltered: true,
2498+
},
2499+
{
2500+
name: "when all just-key filter match, node should be filtered",
2501+
nodeLabels: map[string]string{"k1": "v1", "k2": "v2"},
2502+
service: &corev1.Service{
2503+
ObjectMeta: v1.ObjectMeta{
2504+
Annotations: map[string]string{ServiceAnnotationLoadBalancerNodeSelector: "k1,k2"},
2505+
},
2506+
},
2507+
annotationKey: ServiceAnnotationLoadBalancerNodeSelector,
2508+
defaultSetting: make(map[string]string),
2509+
nodeFiltered: true,
2510+
},
2511+
{
2512+
name: "when some filters do not match, node should not be filtered",
2513+
nodeLabels: map[string]string{"k1": "v1"},
2514+
service: &corev1.Service{
2515+
ObjectMeta: v1.ObjectMeta{
2516+
Annotations: map[string]string{ServiceAnnotationLoadBalancerNodeSelector: " k1=v1, k2 "},
2517+
},
2518+
},
2519+
annotationKey: ServiceAnnotationLoadBalancerNodeSelector,
2520+
defaultSetting: make(map[string]string),
2521+
nodeFiltered: false,
2522+
},
2523+
{
2524+
name: "when no filter matches, node should not be filtered",
2525+
nodeLabels: map[string]string{"k1": "v1", "k2": "v2"},
2526+
service: &corev1.Service{
2527+
ObjectMeta: v1.ObjectMeta{
2528+
Annotations: map[string]string{ServiceAnnotationLoadBalancerNodeSelector: "k3=v3"},
2529+
},
2530+
},
2531+
annotationKey: ServiceAnnotationLoadBalancerNodeSelector,
2532+
defaultSetting: make(map[string]string),
2533+
nodeFiltered: false,
2534+
},
2535+
}
2536+
2537+
for _, test := range tests {
2538+
t.Run(test.name, func(t *testing.T) {
2539+
node := &corev1.Node{}
2540+
node.Labels = test.nodeLabels
2541+
2542+
// TODO: add testArgs
2543+
targetNodeLabels := getKeyValueFromServiceAnnotation(test.service, ServiceAnnotationLoadBalancerNodeSelector, "")
2544+
2545+
nodes := []*corev1.Node{node}
2546+
filteredNodes := filterNodes(nodes, targetNodeLabels)
2547+
2548+
if test.nodeFiltered {
2549+
assert.Equal(t, nodes, filteredNodes)
2550+
} else {
2551+
assert.Empty(t, filteredNodes)
2552+
}
2553+
})
2554+
}
2555+
}

pkg/openstack/openstack.go

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -114,7 +114,8 @@ type LoadBalancerOpts struct {
114114
MonitorMaxRetries uint `gcfg:"monitor-max-retries"`
115115
MonitorMaxRetriesDown uint `gcfg:"monitor-max-retries-down"`
116116
ManageSecurityGroups bool `gcfg:"manage-security-groups"`
117-
InternalLB bool `gcfg:"internal-lb"` // default false
117+
InternalLB bool `gcfg:"internal-lb"` // default false
118+
NodeSelector string `gcfg:"node-selector"` // If specified, the loadbalancer members will be assined only from nodes list filtered by node-selector labels
118119
CascadeDelete bool `gcfg:"cascade-delete"`
119120
FlavorID string `gcfg:"flavor-id"`
120121
AvailabilityZone string `gcfg:"availability-zone"`
@@ -222,6 +223,7 @@ func ReadConfig(config io.Reader) (Config, error) {
222223
// Set default values explicitly
223224
cfg.LoadBalancer.Enabled = true
224225
cfg.LoadBalancer.InternalLB = false
226+
cfg.LoadBalancer.NodeSelector = ""
225227
cfg.LoadBalancer.LBProvider = "amphora"
226228
cfg.LoadBalancer.LBMethod = "ROUND_ROBIN"
227229
cfg.LoadBalancer.CreateMonitor = false

pkg/util/util.go

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"context"
55
"encoding/json"
66
"fmt"
7+
"strings"
78
"time"
89

910
"github.com/container-storage-interface/spec/lib/go/csi"
@@ -77,6 +78,31 @@ func Contains(list []string, strToSearch string) bool {
7778
return false
7879
}
7980

81+
// StringToMap converts a string of comma-separated key-values into a map
82+
func StringToMap(str string) map[string]string {
83+
// break up a "key1=val,key2=val2,key3=,key4" string into a list
84+
values := strings.Split(strings.TrimSpace(str), ",")
85+
keyValues := make(map[string]string, len(values))
86+
87+
for _, kv := range values {
88+
kv := strings.SplitN(strings.TrimSpace(kv), "=", 2)
89+
90+
k := kv[0]
91+
if len(kv) == 1 {
92+
if k != "" {
93+
// process "key=" or "key"
94+
keyValues[k] = ""
95+
}
96+
continue
97+
}
98+
99+
// process "key=val" or "key=val=foo"
100+
keyValues[k] = kv[1]
101+
}
102+
103+
return keyValues
104+
}
105+
80106
// RoundUpSize calculates how many allocation units are needed to accommodate
81107
// a volume of given size. E.g. when user wants 1500MiB volume, while AWS EBS
82108
// allocates volumes in gibibyte-sized chunks,

0 commit comments

Comments
 (0)