Add a service controller to handle loadbalancerclass case

Signed-off-by: lubronzhan <[email protected]>

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:

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)
 			}

pkg/controller/service.go pkg/provider/loadbalancerclass.go

@@ -1,13 +1,16 @@
-package controller
+package provider
 
 import (
+	"context"
 	"fmt"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
@@ -20,46 +23,56 @@ const (
 	controllerName = "service-with-loadbalancerclass-controller"
 )
 
-type Controller struct {
+type loadbalancerClassServiceController struct {
+	kubeClient          kubernetes.Interface
 	serviceInformer     cache.SharedIndexInformer
 	serviceLister       corelisters.ServiceLister
 	serviceListerSynced cache.InformerSynced
 
 	recorder  record.EventRecorder
 	workqueue workqueue.RateLimitingInterface
+
+	cmName      string
+	cmNamespace string
 }
 
-func NewController(
+func newLoadbalancerClassServiceController(
 	sharedInformer informers.SharedInformerFactory,
-) *Controller {
+	kubeClient kubernetes.Interface,
+	cmName, cmNamespace string,
+) *loadbalancerClassServiceController {
 	eventBroadcaster := record.NewBroadcaster()
 	eventBroadcaster.StartLogging(klog.Infof)
 	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerName})
 	serviceInformer := sharedInformer.Core().V1().Services().Informer()
-	c := &Controller{
+	c := &loadbalancerClassServiceController{
 		serviceInformer:     serviceInformer,
 		serviceLister:       sharedInformer.Core().V1().Services().Lister(),
 		serviceListerSynced: serviceInformer.HasSynced,
+		kubeClient:          kubeClient,
 
 		recorder:  recorder,
 		workqueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "Nodes"),
+
+		cmName:      cmName,
+		cmNamespace: cmNamespace,
 	}
 
-	serviceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+	_, _ = serviceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(cur interface{}) {
-			node := cur.(*corev1.Node).DeepCopy()
-			c.enqueueService(node)
+			s := cur.(*corev1.Node).DeepCopy()
+			c.enqueueService(s)
 		},
 		UpdateFunc: func(old interface{}, new interface{}) {
-			c.enqueueService(old)
+			c.enqueueService(new)
 		},
 		// DeleteFunc: ,
 	})
 
 	return c
 }
 
-func (c *Controller) enqueueService(obj interface{}) {
+func (c *loadbalancerClassServiceController) enqueueService(obj interface{}) {
 	var key string
 	var err error
 	if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
@@ -70,17 +83,17 @@ func (c *Controller) enqueueService(obj interface{}) {
 }
 
 // Run starts the worker to process node updates
-func (c *Controller) Run(stopCh <-chan struct{}) {
+func (c *loadbalancerClassServiceController) Run(stopCh <-chan struct{}) {
 	defer utilruntime.HandleCrash()
 	defer c.workqueue.ShutDown()
 
 	klog.V(4).Info("Waiting cache to be synced.")
 
-	if !cache.WaitForNamedCacheSync("node", stopCh, c.serviceListerSynced) {
+	if !cache.WaitForNamedCacheSync("service", stopCh, c.serviceListerSynced) {
 		return
 	}
 
-	klog.V(4).Info("Starting node workers.")
+	klog.V(4).Info("Starting service workers for loadbalancerclass.")
 	go wait.Until(c.runWorker, time.Second, stopCh)
 
 	<-stopCh
@@ -89,14 +102,14 @@ func (c *Controller) Run(stopCh <-chan struct{}) {
 // runWorker is a long-running function that will continually call the
 // processNextWorkItem function in order to read and process a message on the
 // workqueue.
-func (c *Controller) runWorker() {
+func (c *loadbalancerClassServiceController) runWorker() {
 	for c.processNextWorkItem() {
 	}
 }
 
 // processNextWorkItem will read a single work item off the workqueue and
 // attempt to process it, by calling the syncHandler.
-func (c *Controller) processNextWorkItem() bool {
+func (c *loadbalancerClassServiceController) processNextWorkItem() bool {
 	obj, shutdown := c.workqueue.Get()
 	if shutdown {
 		return false
@@ -138,7 +151,7 @@ func (c *Controller) processNextWorkItem() bool {
 // syncService will sync the Service with the given key if it has had its expectations fulfilled,
 // meaning it did not expect to see any more of its pods created or deleted. This function is not meant to be
 // invoked concurrently with the same key.
-func (c *Controller) syncService(key string) error {
+func (c *loadbalancerClassServiceController) syncService(key string) error {
 	startTime := time.Now()
 	defer func() {
 		klog.V(4).Infof("Finished syncing service %q (%v)", key, time.Since(startTime))
@@ -154,15 +167,22 @@ func (c *Controller) syncService(key string) error {
 	switch {
 	case err != nil:
 		utilruntime.HandleError(fmt.Errorf("unable to retrieve service %v from store: %v", key, err))
+	case loadbalancerClassMatch(svc):
+		klog.V(4).Infof("Skip reoconciling service %s/%s, since loadbalancerClass doesn't match", svc.Namespace, svc.Name)
 	default:
 		err = c.processServiceCreateOrUpdate(svc)
 	}
 
 	return err
 }
 
-func (c *Controller) processServiceCreateOrUpdate(svc *corev1.Service) error {
+func (c *loadbalancerClassServiceController) processServiceCreateOrUpdate(svc *corev1.Service) error {
 	// ctx := context.Background()
+	_, err := syncLoadBalancer(context.Background(), c.kubeClient, svc, c.cmName, c.cmNamespace)
+	c.recorder.Eventf(svc, v1.EventTypeWarning, "syncLoadBalancer", "Error syncing load balancer: %v", err)
+	return err
+}
 
-	return nil
+func loadbalancerClassMatch(svc *corev1.Service) bool {
+	return svc != nil && svc.Spec.LoadBalancerClass != nil && *svc.Spec.LoadBalancerClass == LoadbalancerClass
 }