Merge pull request #100 from lubronzhan/topic/lubron/fix-61

Support loadbalancerClass

Author: lubronzhan

README.md

@@ -19,7 +19,8 @@ The `kube-vip-cloud-provider` will only implement the `loadBalancer` functionali
 - Setting the special IP `0.0.0.0` for DHCP workflow.
 - Support single stack IPv6 or IPv4
 - Support for dualstack via the annotation: `kube-vip.io/loadbalancerIPs: 192.168.10.10,2001:db8::1`
-- Support ascending and descending search order by setting search-order=desc
+- Support ascending and descending search order when allocating IP from pool or range by setting search-order=desc
+- Support loadbalancerClass `kube-vip.io/kube-vip-class`
 
 ## Installing the `kube-vip-cloud-provider`
 
@@ -134,6 +135,12 @@ address in each of both IP families for the pool.
 
 Set the CIDR to `0.0.0.0/32`, that will make the controller to give all _LoadBalancers_ the IP `0.0.0.0`.
 
+
+## LoadbalancerClass support
+
+If users only want kube-vip-cloud-provider to allocate ip for specific set of services, they can pass `KUBEVIP_ENABLE_LOADBALANCERCLASS: true` as an environment variable to kube-vip-cloud-provider. kube-vip-cloud-provider will only allocate ip to service with `spec.loadBalancerClass: kube-vip.io/kube-vip-class`.
+
+
 ## Debugging
 
 The logs for the cloud-provider controller can be viewed with the following command:

go.mod

@@ -11,6 +11,8 @@ require (
 	k8s.io/cloud-provider v0.29.1
 	k8s.io/component-base v0.29.1
 	k8s.io/klog v1.0.0
+	k8s.io/klog/v2 v2.110.1
+	k8s.io/utils v0.0.0-20230726121419-3b25d923346b
 )
 
 require (
@@ -97,10 +99,8 @@ require (
 	k8s.io/apiserver v0.29.1 // indirect
 	k8s.io/component-helpers v0.29.1 // indirect
 	k8s.io/controller-manager v0.29.1 // indirect
-	k8s.io/klog/v2 v2.110.1 // indirect
 	k8s.io/kms v0.29.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 // indirect
-	k8s.io/utils v0.0.0-20230726121419-3b25d923346b // indirect
 	sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.28.0 // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect

main.go

@@ -19,6 +19,7 @@ package main
 import (
 	"fmt"
 	"os"
+	"strconv"
 
 	"github.com/kube-vip/kube-vip-cloud-provider/pkg/provider"
 	"github.com/spf13/pflag"
@@ -48,7 +49,21 @@ func main() {
 
 	fss := cliflag.NamedFlagSets{}
 
-	command := app.NewCloudControllerManagerCommand(opts, cloudInitializer, controllerInitializers(), names.CCMControllerAliases(), fss, wait.NeverStop)
+	controllerInitializers := controllerInitializers()
+	lbc := os.Getenv(provider.EnableLoadbalancerClassEnvKey)
+	if len(lbc) > 0 {
+		enableLBClass, err := strconv.ParseBool(lbc)
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "%s value '%s' is invalid, %v\n", provider.EnableLoadbalancerClassEnvKey, lbc, err)
+			os.Exit(1)
+		}
+		if enableLBClass {
+			controllerInitializers = make(map[string]app.ControllerInitFuncConstructor)
+			klog.Infoln("skipping default cloud-provider service controller")
+		}
+	}
+
+	command := app.NewCloudControllerManagerCommand(opts, cloudInitializer, controllerInitializers, names.CCMControllerAliases(), fss, wait.NeverStop)
 
 	command.Flags().BoolVar(&provider.OutSideCluster, "OutSideCluster", false, "Start Controller outside of cluster")
 

pkg/provider/config.go

@@ -5,6 +5,7 @@ import (
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 )
 
 // Services functions - once the service data is taken from the configMap, these functions will interact with the data
@@ -45,12 +46,12 @@ import (
 // 	return
 // }
 
-func (k *kubevipLoadBalancerManager) GetConfigMap(ctx context.Context, cm, nm string) (*v1.ConfigMap, error) {
+func getConfigMap(ctx context.Context, kubeClient kubernetes.Interface, cm, nm string) (*v1.ConfigMap, error) {
 	// Attempt to retrieve the config map
-	return k.kubeClient.CoreV1().ConfigMaps(nm).Get(ctx, cm, metav1.GetOptions{})
+	return kubeClient.CoreV1().ConfigMaps(nm).Get(ctx, cm, metav1.GetOptions{})
 }
 
-func (k *kubevipLoadBalancerManager) CreateConfigMap(ctx context.Context, cm, nm string) (*v1.ConfigMap, error) {
+func createConfigMap(ctx context.Context, kubeClient kubernetes.Interface, cm, nm string) (*v1.ConfigMap, error) {
 	// Create new configuration map in the correct namespace
 	newConfigMap := v1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
@@ -59,7 +60,7 @@ func (k *kubevipLoadBalancerManager) CreateConfigMap(ctx context.Context, cm, nm
 		},
 	}
 	// Return results of configMap create
-	return k.kubeClient.CoreV1().ConfigMaps(nm).Create(ctx, &newConfigMap, metav1.CreateOptions{})
+	return kubeClient.CoreV1().ConfigMaps(nm).Create(ctx, &newConfigMap, metav1.CreateOptions{})
 }
 
 // func (k *kubevipLoadBalancerManager) UpdateConfigMap(ctx context.Context, cm *v1.ConfigMap, s *kubevipServices) (*v1.ConfigMap, error) {

pkg/provider/loadBalancer.go

@@ -44,11 +44,11 @@ func newLoadBalancer(kubeClient kubernetes.Interface, ns, cm string) cloudprovid
 }
 
 func (k *kubevipLoadBalancerManager) EnsureLoadBalancer(ctx context.Context, _ string, service *v1.Service, _ []*v1.Node) (lbs *v1.LoadBalancerStatus, err error) {
-	return k.syncLoadBalancer(ctx, service)
+	return syncLoadBalancer(ctx, k.kubeClient, service, k.cloudConfigMap, k.namespace)
 }
 
 func (k *kubevipLoadBalancerManager) UpdateLoadBalancer(ctx context.Context, _ string, service *v1.Service, _ []*v1.Node) (err error) {
-	_, err = k.syncLoadBalancer(ctx, service)
+	_, err = syncLoadBalancer(ctx, k.kubeClient, service, k.cloudConfigMap, k.namespace)
 	return err
 }
 
@@ -87,7 +87,7 @@ func (k *kubevipLoadBalancerManager) deleteLoadBalancer(ctx context.Context, ser
 // 2b. Get the network configuration for this service (namespace) / (CIDR/Range)
 // 2c. Between the two find a free address
 
-func (k *kubevipLoadBalancerManager) syncLoadBalancer(ctx context.Context, service *v1.Service) (*v1.LoadBalancerStatus, error) {
+func syncLoadBalancer(ctx context.Context, kubeClient kubernetes.Interface, service *v1.Service, cmName, cmNamespace string) (*v1.LoadBalancerStatus, error) {
 	// This function reconciles the load balancer state
 	klog.Infof("syncing service '%s' (%s)", service.Name, service.UID)
 
@@ -97,7 +97,7 @@ func (k *kubevipLoadBalancerManager) syncLoadBalancer(ctx context.Context, servi
 			klog.Warningf("service.Spec.LoadBalancerIP is defined but annotations '%s' is not, assume it's a legacy service, updates its annotations", loadbalancerIPsAnnotations)
 			// assume it's legacy service, need to update the annotation.
 			err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-				recentService, getErr := k.kubeClient.CoreV1().Services(service.Namespace).Get(ctx, service.Name, metav1.GetOptions{})
+				recentService, getErr := kubeClient.CoreV1().Services(service.Namespace).Get(ctx, service.Name, metav1.GetOptions{})
 				if getErr != nil {
 					return getErr
 				}
@@ -109,7 +109,7 @@ func (k *kubevipLoadBalancerManager) syncLoadBalancer(ctx context.Context, servi
 				delete(recentService.Labels, legacyIpamAddressLabelKey)
 
 				// Update the actual service with the annotations
-				_, updateErr := k.kubeClient.CoreV1().Services(recentService.Namespace).Update(ctx, recentService, metav1.UpdateOptions{})
+				_, updateErr := kubeClient.CoreV1().Services(recentService.Namespace).Update(ctx, recentService, metav1.UpdateOptions{})
 				return updateErr
 			})
 			if err != nil {
@@ -125,7 +125,7 @@ func (k *kubevipLoadBalancerManager) syncLoadBalancer(ctx context.Context, servi
 		if service.Labels == nil || service.Labels[implementationLabelKey] != implementationLabelValue {
 			klog.Infof("service '%s/%s' created with pre-defined ip '%s'", service.Namespace, service.Name, v)
 			err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-				recentService, getErr := k.kubeClient.CoreV1().Services(service.Namespace).Get(ctx, service.Name, metav1.GetOptions{})
+				recentService, getErr := kubeClient.CoreV1().Services(service.Namespace).Get(ctx, service.Name, metav1.GetOptions{})
 				if getErr != nil {
 					return getErr
 				}
@@ -135,7 +135,7 @@ func (k *kubevipLoadBalancerManager) syncLoadBalancer(ctx context.Context, servi
 				}
 				recentService.Labels[implementationLabelKey] = implementationLabelValue
 				// Update the actual service with the annotations
-				_, updateErr := k.kubeClient.CoreV1().Services(recentService.Namespace).Update(ctx, recentService, metav1.UpdateOptions{})
+				_, updateErr := kubeClient.CoreV1().Services(recentService.Namespace).Update(ctx, recentService, metav1.UpdateOptions{})
 				return updateErr
 			})
 			if err != nil {
@@ -146,31 +146,31 @@ func (k *kubevipLoadBalancerManager) syncLoadBalancer(ctx context.Context, servi
 	}
 
 	// Get the clound controller configuration map
-	controllerCM, err := k.GetConfigMap(ctx, k.cloudConfigMap, k.namespace)
+	controllerCM, err := getConfigMap(ctx, kubeClient, cmName, cmNamespace)
 	if err != nil {
-		klog.Errorf("Unable to retrieve kube-vip ipam config from configMap [%s] in %s", k.cloudConfigMap, k.namespace)
+		klog.Errorf("Unable to retrieve kube-vip ipam config from configMap [%s] in %s", cmName, cmNamespace)
 		// TODO - determine best course of action, create one if it doesn't exist
-		controllerCM, err = k.CreateConfigMap(ctx, k.cloudConfigMap, k.namespace)
+		controllerCM, err = createConfigMap(ctx, kubeClient, cmName, cmNamespace)
 		if err != nil {
 			return nil, err
 		}
 	}
 
 	// Get ip pool from configmap and determine if it is namespace specific or global
-	pool, global, err := discoverPool(controllerCM, service.Namespace, k.cloudConfigMap)
+	pool, global, err := discoverPool(controllerCM, service.Namespace, cmName)
 	if err != nil {
 		return nil, err
 	}
 
 	// Get all services in this namespace or globally, that have the correct label
 	var svcs *v1.ServiceList
 	if global {
-		svcs, err = k.kubeClient.CoreV1().Services("").List(ctx, metav1.ListOptions{LabelSelector: getKubevipImplementationLabel()})
+		svcs, err = kubeClient.CoreV1().Services("").List(ctx, metav1.ListOptions{LabelSelector: getKubevipImplementationLabel()})
 		if err != nil {
 			return &service.Status.LoadBalancer, err
 		}
 	} else {
-		svcs, err = k.kubeClient.CoreV1().Services(service.Namespace).List(ctx, metav1.ListOptions{LabelSelector: getKubevipImplementationLabel()})
+		svcs, err = kubeClient.CoreV1().Services(service.Namespace).List(ctx, metav1.ListOptions{LabelSelector: getKubevipImplementationLabel()})
 		if err != nil {
 			return &service.Status.LoadBalancer, err
 		}
@@ -201,7 +201,7 @@ func (k *kubevipLoadBalancerManager) syncLoadBalancer(ctx context.Context, servi
 
 	// Update the services with this new address
 	retryErr := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		recentService, getErr := k.kubeClient.CoreV1().Services(service.Namespace).Get(ctx, service.Name, metav1.GetOptions{})
+		recentService, getErr := kubeClient.CoreV1().Services(service.Namespace).Get(ctx, service.Name, metav1.GetOptions{})
 		if getErr != nil {
 			return getErr
 		}
@@ -226,7 +226,7 @@ func (k *kubevipLoadBalancerManager) syncLoadBalancer(ctx context.Context, servi
 		recentService.Spec.LoadBalancerIP = strings.Split(loadBalancerIPs, ",")[0]
 
 		// Update the actual service with the address and the labels
-		_, updateErr := k.kubeClient.CoreV1().Services(recentService.Namespace).Update(ctx, recentService, metav1.UpdateOptions{})
+		_, updateErr := kubeClient.CoreV1().Services(recentService.Namespace).Update(ctx, recentService, metav1.UpdateOptions{})
 		return updateErr
 	})
 	if retryErr != nil {

pkg/provider/loadBalancer_test.go

@@ -888,7 +888,7 @@ func Test_syncLoadBalancer(t *testing.T) {
 				}
 			}
 
-			_, err = mgr.syncLoadBalancer(context.Background(), &tt.originalService) // #nosec G601
+			_, err = syncLoadBalancer(context.Background(), mgr.kubeClient, &tt.originalService, cm, ns) // #nosec G601
 			if err != nil {
 				t.Error(err)
 			}