feat: add nodeselector annotation to limit LB pool members

Author: mweibel

README.md

@@ -429,6 +429,7 @@ Changes to the following annotations causes pools to be recreated and cause an e
 - `k8s.cloudscale.ch/loadbalancer-pool-algorithm`
 - `k8s.cloudscale.ch/loadbalancer-pool-protocol`
 - `k8s.cloudscale.ch/loadbalancer-listener-allowed-subnets`
+- `k8s.cloudscale.ch/loadbalancer-node-selector`
 
 Additionally, changes to `spec.externalTrafficPolicy` have the same effect.
 

examples/nginx-hello-nodeselector.yml

@@ -0,0 +1,57 @@
+# Deploys the docker.io/nginxdemos/hello:plain-text container and creates a
+# loadbalancer service with a node-selector annotation for it:
+#
+#     export KUBECONFIG=path/to/kubeconfig
+#     kubectl apply -f nginx-hello.yml
+#
+# Wait for `kubectl describe service hello` to show "Loadbalancer Ensured",
+# then use the IP address found under "LoadBalancer Ingress" to connect to the
+# service.
+#
+# You can also use the following shortcut:
+#
+#     curl http://$(kubectl get service hello -o jsonpath='{.status.loadBalancer.ingress[0].ip}')
+#
+# If you follow the nginx log, you will see that nginx sees a cluster internal
+# IP address as source of requests:
+#
+#     kubectl logs -l "app=hello"
+#
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: hello
+spec:
+  replicas: 2
+  selector:
+    matchLabels:
+      app: hello
+  template:
+    metadata:
+      labels:
+        app: hello
+    spec:
+      containers:
+      - name: hello
+        image: docker.io/nginxdemos/hello:plain-text
+      nodeSelector:
+        kubernetes.io/hostname: k8test-worker-2
+---
+apiVersion: v1
+kind: Service
+metadata:
+  labels:
+    app: hello
+  annotations:
+    k8s.cloudscale.ch/loadbalancer-node-selector: "kubernetes.io/hostname=k8test-worker-2"
+  name: hello
+spec:
+  ports:
+  - port: 80
+    protocol: TCP
+    targetPort: 80
+    name: http
+  selector:
+    app: hello
+  type: LoadBalancer

pkg/cloudscale_ccm/loadbalancer.go

@@ -11,6 +11,7 @@ import (
 	"github.com/cloudscale-ch/cloudscale-go-sdk/v6"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
 	"k8s.io/utils/ptr"
@@ -208,7 +209,7 @@ const (
 	// connections timing out while the monitor is updated.
 	LoadBalancerHealthMonitorTimeoutS = "k8s.cloudscale.ch/loadbalancer-health-monitor-timeout-s"
 
-	// LoadBalancerHealthMonitorDownThreshold is the number of the checks that
+	// LoadBalancerHealthMonitorUpThreshold is the number of the checks that
 	// need to succeed before a pool member is considered up. Defaults to 2.
 	LoadBalancerHealthMonitorUpThreshold = "k8s.cloudscale.ch/loadbalancer-health-monitor-up-threshold"
 
@@ -278,7 +279,7 @@ const (
 	// Changing this annotation on an established service is considered safe.
 	LoadBalancerListenerTimeoutMemberDataMS = "k8s.cloudscale.ch/loadbalancer-timeout-member-data-ms"
 
-	// LoadBalancerSubnetLimit is a JSON list of subnet UUIDs that the
+	// LoadBalancerListenerAllowedSubnets is a JSON list of subnet UUIDs that the
 	// loadbalancer should use. By default, all subnets of a node are used:
 	//
 	// * `[]` means that anyone is allowed to connect (default).
@@ -291,6 +292,10 @@ const (
 	// This is an advanced feature, useful if you have nodes that are in
 	// multiple private subnets.
 	LoadBalancerListenerAllowedSubnets = "k8s.cloudscale.ch/loadbalancer-listener-allowed-subnets"
+
+	// LoadBalancerNodeSelector can be set to restrict which nodes are added to the LB pool.
+	// It accepts a standard Kubernetes label selector string.
+	LoadBalancerNodeSelector = "k8s.cloudscale.ch/loadbalancer-node-selector"
 )
 
 type loadbalancer struct {
@@ -387,6 +392,11 @@ func (l *loadbalancer) EnsureLoadBalancer(
 		return nil, err
 	}
 
+	nodes, err := filterNodesBySelector(serviceInfo, nodes)
+	if err != nil {
+		return nil, err
+	}
+
 	// Refuse to do anything if there are no nodes
 	if len(nodes) == 0 {
 		return nil, errors.New(
@@ -396,7 +406,7 @@ func (l *loadbalancer) EnsureLoadBalancer(
 	}
 
 	// Reconcile
-	err := reconcileLbState(ctx, l.lbs.client, func() (*lbState, error) {
+	err = reconcileLbState(ctx, l.lbs.client, func() (*lbState, error) {
 		// Get the desired state from Kubernetes
 		servers, err := l.srv.mapNodes(ctx, nodes).All()
 		if err != nil {
@@ -442,6 +452,28 @@ func (l *loadbalancer) EnsureLoadBalancer(
 	return result, nil
 }
 
+func filterNodesBySelector(
+	serviceInfo *serviceInfo,
+	nodes []*v1.Node,
+) ([]*v1.Node, error) {
+	selector := labels.Everything()
+	if v := serviceInfo.annotation(LoadBalancerNodeSelector); v != "" {
+		var err error
+		selector, err = labels.Parse(v)
+		if err != nil {
+			return nil, fmt.Errorf("unable to parse selector: %w", err)
+		}
+	}
+	selectedNodes := make([]*v1.Node, 0, len(nodes))
+	for _, node := range nodes {
+		if selector.Matches(labels.Set(node.Labels)) {
+			selectedNodes = append(selectedNodes, node)
+		}
+	}
+
+	return selectedNodes, nil
+}
+
 // UpdateLoadBalancer updates hosts under the specified load balancer.
 // Implementations must treat the *v1.Service and *v1.Node
 // parameters as read-only and not modify them.
@@ -461,6 +493,19 @@ func (l *loadbalancer) UpdateLoadBalancer(
 		return err
 	}
 
+	nodes, err := filterNodesBySelector(serviceInfo, nodes)
+	if err != nil {
+		return err
+	}
+
+	// Refuse to do anything if there are no nodes
+	if len(nodes) == 0 {
+		return errors.New(
+			"no valid nodes for service found, please verify there is " +
+				"at least one that allows load balancers",
+		)
+	}
+
 	// Reconcile
 	return reconcileLbState(ctx, l.lbs.client, func() (*lbState, error) {
 		// Get the desired state from Kubernetes

pkg/cloudscale_ccm/service_info.go

@@ -118,6 +118,8 @@ func (s serviceInfo) annotation(key string) string {
 		return s.annotationOrDefault(key, "50000")
 	case LoadBalancerListenerAllowedSubnets:
 		return s.annotationOrDefault(key, "[]")
+	case LoadBalancerNodeSelector:
+		return s.annotationOrDefault(key, "")
 	default:
 		return s.annotationOrElse(key, func() string {
 			klog.Warning("unknown annotation:", key)