MEDIUM: Add support for endpoint slices

* Use endpointSlices to discover endpoints if the cluster supports them
* Add permissions to the ClusterRole for querying the endpointslices

Relocate HAPRoxySrvs from PortEndpoints to Namespace

With EndpointSlices one object does not include all endpoints to a
service anymore. In order to track changes over all slices the metadata
used for syncing the new addresses and haproxy backend servers are moved
to Namespace level.

Author: petuomin

controller/haproxy/api/api.go

@@ -56,7 +56,7 @@ type HAProxyClient interface {
 	SetServerState(backendName string, serverName string, state string) error
 	ServerGet(serverName, backendNa string) (models.Server, error)
 	SetAuxCfgFile(auxCfgFile string)
-	SyncBackendSrvs(oldEndpoints, newEndpoints *store.PortEndpoints) error
+	SyncBackendSrvs(backend *store.RuntimeBackend, portUpdated bool) error
 	UserListDeleteAll() error
 	UserListExistsByGroup(group string) (bool, error)
 	UserListCreateByGroup(group string, userPasswordMap map[string][]byte) error

controller/haproxy/api/runtime.go

@@ -32,23 +32,20 @@ func (c *clientNative) GetMap(mapFile string) (*models.Map, error) {
 }
 
 // SyncBackendSrvs syncs states and addresses of a backend servers with corresponding endpoints.
-func (c *clientNative) SyncBackendSrvs(oldEndpoints, newEndpoints *store.PortEndpoints) error {
-	if oldEndpoints.BackendName == "" {
+func (c *clientNative) SyncBackendSrvs(backend *store.RuntimeBackend, portUpdated bool) error {
+	if backend.Name == "" {
 		return nil
 	}
-	newEndpoints.HAProxySrvs = oldEndpoints.HAProxySrvs
-	newEndpoints.BackendName = oldEndpoints.BackendName
-	haproxySrvs := newEndpoints.HAProxySrvs
-	newAddresses := newEndpoints.AddrNew
-	portChanged := newEndpoints.Port != oldEndpoints.Port
+	haproxySrvs := backend.HAProxySrvs
+	addresses := backend.Endpoints.Addresses
 	// Disable stale entries from HAProxySrvs
 	// and provide list of Disabled Srvs
 	var disabled []*store.HAProxySrv
 	var errors utils.Errors
 	for i, srv := range haproxySrvs {
-		srv.Modified = portChanged || srv.Modified
-		if _, ok := newAddresses[srv.Address]; ok {
-			delete(newAddresses, srv.Address)
+		srv.Modified = srv.Modified || portUpdated
+		if _, ok := addresses[srv.Address]; ok {
+			delete(addresses, srv.Address)
 		} else {
 			haproxySrvs[i].Address = ""
 			haproxySrvs[i].Modified = true
@@ -57,14 +54,14 @@ func (c *clientNative) SyncBackendSrvs(oldEndpoints, newEndpoints *store.PortEnd
 	}
 
 	// Configure new Addresses in available HAProxySrvs
-	for newAddr := range newAddresses {
+	for newAddr := range addresses {
 		if len(disabled) == 0 {
 			break
 		}
 		disabled[0].Address = newAddr
 		disabled[0].Modified = true
 		disabled = disabled[1:]
-		delete(newAddresses, newAddr)
+		delete(addresses, newAddr)
 	}
 	// Dynamically updates HAProxy backend servers  with HAProxySrvs content
 	var addrErr, stateErr error
@@ -74,15 +71,15 @@ func (c *clientNative) SyncBackendSrvs(oldEndpoints, newEndpoints *store.PortEnd
 		}
 		if srv.Address == "" {
 			// logger.Tracef("server '%s/%s' changed status to %v", newEndpoints.BackendName, srv.Name, "maint")
-			addrErr = c.SetServerAddr(newEndpoints.BackendName, srv.Name, "127.0.0.1", 0)
-			stateErr = c.SetServerState(newEndpoints.BackendName, srv.Name, "maint")
+			addrErr = c.SetServerAddr(backend.Name, srv.Name, "127.0.0.1", 0)
+			stateErr = c.SetServerState(backend.Name, srv.Name, "maint")
 		} else {
 			// logger.Tracef("server '%s/%s' changed status to %v", newEndpoints.BackendName, srv.Name, "ready")
-			addrErr = c.SetServerAddr(newEndpoints.BackendName, srv.Name, srv.Address, int(newEndpoints.Port))
-			stateErr = c.SetServerState(newEndpoints.BackendName, srv.Name, "ready")
+			addrErr = c.SetServerAddr(backend.Name, srv.Name, srv.Address, int(backend.Endpoints.Port))
+			stateErr = c.SetServerState(backend.Name, srv.Name, "ready")
 		}
 		if addrErr != nil || stateErr != nil {
-			newEndpoints.DynUpdateFailed = true
+			backend.DynUpdateFailed = true
 			errors.Add(addrErr)
 			errors.Add(stateErr)
 		}

controller/ingress.go

@@ -100,7 +100,7 @@ func (c *HAProxyController) handleIngressPath(ingress *store.Ingress, host strin
 	}
 	c.Cfg.ActiveBackends[backendName] = struct{}{}
 	// Endpoints
-	endpointsReload := svc.HandleEndpoints(c.Client, c.Store)
+	endpointsReload := svc.HandleHAProxySrvs(c.Client, c.Store)
 	return backendReload || endpointsReload || routeReload, err
 }
 
@@ -143,7 +143,7 @@ func (c *HAProxyController) setDefaultService(ingress *store.Ingress, frontends
 		}
 	}
 	c.Cfg.ActiveBackends[backendName] = struct{}{}
-	endpointsReload := svc.HandleEndpoints(c.Client, c.Store)
+	endpointsReload := svc.HandleHAProxySrvs(c.Client, c.Store)
 	reload = bdReload || ftReload || endpointsReload
 	return reload, err
 }

controller/kubernetes.go

@@ -24,6 +24,8 @@ import (
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/util/wait"
 
+	discoveryv1 "k8s.io/api/discovery/v1"
+	discoveryv1beta1 "k8s.io/api/discovery/v1beta1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
@@ -120,11 +122,12 @@ func (k *K8s) EventsNamespaces(channel chan SyncDataEvent, stop chan struct{}, i
 					status = DELETED
 				}
 				item := &store.Namespace{
-					Name:      data.GetName(),
-					Endpoints: make(map[string]*store.Endpoints),
-					Services:  make(map[string]*store.Service),
-					Ingresses: make(map[string]*store.Ingress),
-					Secret:    make(map[string]*store.Secret),
+					Name:           data.GetName(),
+					Endpoints:      make(map[string]map[string]*store.Endpoints),
+					Services:       make(map[string]*store.Service),
+					Ingresses:      make(map[string]*store.Ingress),
+					Secret:         make(map[string]*store.Secret),
+					HAProxyRuntime: make(map[string]map[string]*store.RuntimeBackend),
 					CRs: &store.CustomResources{
 						Global:     make(map[string]*models.Global),
 						Defaults:   make(map[string]*models.Defaults),
@@ -144,11 +147,12 @@ func (k *K8s) EventsNamespaces(channel chan SyncDataEvent, stop chan struct{}, i
 				}
 				status := DELETED
 				item := &store.Namespace{
-					Name:      data.GetName(),
-					Endpoints: make(map[string]*store.Endpoints),
-					Services:  make(map[string]*store.Service),
-					Ingresses: make(map[string]*store.Ingress),
-					Secret:    make(map[string]*store.Secret),
+					Name:           data.GetName(),
+					Endpoints:      make(map[string]map[string]*store.Endpoints),
+					Services:       make(map[string]*store.Service),
+					Ingresses:      make(map[string]*store.Ingress),
+					Secret:         make(map[string]*store.Secret),
+					HAProxyRuntime: make(map[string]map[string]*store.RuntimeBackend),
 					CRs: &store.CustomResources{
 						Global:     make(map[string]*models.Global),
 						Defaults:   make(map[string]*models.Defaults),
@@ -191,7 +195,7 @@ func (k *K8s) EventsNamespaces(channel chan SyncDataEvent, stop chan struct{}, i
 	go informer.Run(stop)
 }
 
-func (k *K8s) EventsEndpoints(channel chan SyncDataEvent, stop chan struct{}, informer cache.SharedIndexInformer) {
+func (k *K8s) EventsEndpointSlices(channel chan SyncDataEvent, stop chan struct{}, informer cache.SharedIndexInformer) {
 	informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			item, err := k.convertToEndpoints(obj, ADDED)
@@ -227,44 +231,130 @@ func (k *K8s) EventsEndpoints(channel chan SyncDataEvent, stop chan struct{}, in
 }
 
 func (k *K8s) convertToEndpoints(obj interface{}, status store.Status) (*store.Endpoints, error) {
-	data, ok := obj.(*corev1.Endpoints)
-	if !ok {
-		k.Logger.Errorf("%s: Invalid data from k8s api, %s", ENDPOINTS, obj)
-		return nil, ErrIgnored
+	getServiceName := func(labels map[string]string) string {
+		return labels["kubernetes.io/service-name"]
 	}
-	if data.GetNamespace() == "kube-system" {
-		if data.ObjectMeta.Name == "kube-controller-manager" ||
-			data.ObjectMeta.Name == "kube-scheduler" ||
-			data.ObjectMeta.Name == "kubernetes-dashboard" ||
-			data.ObjectMeta.Name == "kube-dns" {
-			return nil, ErrIgnored
+
+	shouldIgnoreObject := func(namespace string, labels map[string]string) bool {
+		serviceName := getServiceName(labels)
+		if namespace == "kube-system" {
+			if serviceName == "kube-controller-manager" ||
+				serviceName == "kube-scheduler" ||
+				serviceName == "kubernetes-dashboard" ||
+				serviceName == "kube-dns" {
+				return true
+			}
 		}
+		return false
 	}
-	if data.ObjectMeta.GetDeletionTimestamp() != nil {
-		// detect endpoints that are in terminating state
-		status = DELETED
-	}
-	item := &store.Endpoints{
-		Namespace: data.GetNamespace(),
-		Service:   data.GetName(),
-		Ports:     make(map[string]*store.PortEndpoints),
-		Status:    status,
-	}
-	for _, subset := range data.Subsets {
-		for _, port := range subset.Ports {
-			addresses := make(map[string]struct{})
-			for _, address := range subset.Addresses {
-				addresses[address.IP] = struct{}{}
-			}
-			item.Ports[port.Name] = &store.PortEndpoints{
-				Port:        int64(port.Port),
-				AddrCount:   len(addresses),
-				AddrNew:     addresses,
-				HAProxySrvs: make([]*store.HAProxySrv, 0, len(addresses)),
+	switch data := obj.(type) {
+	case *discoveryv1beta1.EndpointSlice:
+		if shouldIgnoreObject(data.GetNamespace(), data.GetLabels()) {
+			return nil, ErrIgnored
+		}
+		item := &store.Endpoints{
+			SliceName: data.Name,
+			Namespace: data.GetNamespace(),
+			Service:   getServiceName(data.GetLabels()),
+			Ports:     make(map[string]*store.PortEndpoints),
+			Status:    status,
+		}
+		addresses := make(map[string]struct{})
+		for _, endpoints := range data.Endpoints {
+			for _, address := range endpoints.Addresses {
+				addresses[address] = struct{}{}
+			}
+		}
+		for _, port := range data.Ports {
+			item.Ports[*port.Name] = &store.PortEndpoints{
+				Port:      int64(*port.Port),
+				Addresses: addresses,
+			}
+		}
+		return item, nil
+	case *discoveryv1.EndpointSlice:
+		if shouldIgnoreObject(data.GetNamespace(), data.GetLabels()) {
+			return nil, ErrIgnored
+		}
+		item := &store.Endpoints{
+			SliceName: data.Name,
+			Namespace: data.GetNamespace(),
+			Service:   getServiceName(data.GetLabels()),
+			Ports:     make(map[string]*store.PortEndpoints),
+			Status:    status,
+		}
+		addresses := make(map[string]struct{})
+		for _, endpoints := range data.Endpoints {
+			for _, address := range endpoints.Addresses {
+				addresses[address] = struct{}{}
+			}
+		}
+		for _, port := range data.Ports {
+			item.Ports[*port.Name] = &store.PortEndpoints{
+				Port:      int64(*port.Port),
+				Addresses: addresses,
+			}
+		}
+		return item, nil
+	case *corev1.Endpoints:
+		item := &store.Endpoints{
+			Namespace: data.GetNamespace(),
+			Service:   data.GetName(),
+			Ports:     make(map[string]*store.PortEndpoints),
+			Status:    status,
+		}
+		for _, subset := range data.Subsets {
+			for _, port := range subset.Ports {
+				addresses := make(map[string]struct{})
+				for _, address := range subset.Addresses {
+					addresses[address.IP] = struct{}{}
+				}
+				item.Ports[port.Name] = &store.PortEndpoints{
+					Port:      int64(port.Port),
+					Addresses: addresses,
+				}
 			}
 		}
+		return item, nil
+	default:
+		k.Logger.Errorf("%s: Invalid data from k8s api, %s", ENDPOINTS, obj)
+		return nil, ErrIgnored
 	}
-	return item, nil
+}
+
+func (k *K8s) EventsEndpoints(channel chan SyncDataEvent, stop chan struct{}, informer cache.SharedIndexInformer) {
+	informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			item, err := k.convertToEndpoints(obj, ADDED)
+			if errors.Is(err, ErrIgnored) {
+				return
+			}
+			k.Logger.Tracef("%s %s: %s", ENDPOINTS, item.Status, item.Service)
+			channel <- SyncDataEvent{SyncType: ENDPOINTS, Namespace: item.Namespace, Data: item}
+		},
+		DeleteFunc: func(obj interface{}) {
+			item, err := k.convertToEndpoints(obj, DELETED)
+			if errors.Is(err, ErrIgnored) {
+				return
+			}
+			k.Logger.Tracef("%s %s: %s", ENDPOINTS, item.Status, item.Service)
+			channel <- SyncDataEvent{SyncType: ENDPOINTS, Namespace: item.Namespace, Data: item}
+		},
+		UpdateFunc: func(oldObj, newObj interface{}) {
+			item1, err := k.convertToEndpoints(oldObj, EMPTY)
+			if errors.Is(err, ErrIgnored) {
+				return
+			}
+			item2, _ := k.convertToEndpoints(newObj, MODIFIED)
+			if item2.Equal(item1) {
+				return
+			}
+			// fix modified state for ones that are deleted,new,same
+			k.Logger.Tracef("%s %s: %s", ENDPOINTS, item2.Status, item2.Service)
+			channel <- SyncDataEvent{SyncType: ENDPOINTS, Namespace: item2.Namespace, Data: item2}
+		},
+	})
+	go informer.Run(stop)
 }
 
 func (k *K8s) EventsIngressClass(channel chan SyncDataEvent, stop chan struct{}, informer cache.SharedIndexInformer) {

controller/monitor.go

@@ -37,8 +37,13 @@ func (c *HAProxyController) monitorChanges() {
 	for _, namespace := range c.getWhitelistedNamespaces() {
 		factory := informers.NewSharedInformerFactoryWithOptions(c.k8s.API, c.OSArgs.CacheResyncPeriod, informers.WithNamespace(namespace))
 
-		pi := factory.Core().V1().Endpoints().Informer()
-		c.k8s.EventsEndpoints(c.eventChan, stop, pi)
+		pi := c.getEndpointSlicesSharedInformer(factory)
+		if pi != nil {
+			c.k8s.EventsEndpointSlices(c.eventChan, stop, pi)
+		} else {
+			pi = factory.Core().V1().Endpoints().Informer()
+			c.k8s.EventsEndpoints(c.eventChan, stop, pi)
+		}
 
 		svci := factory.Core().V1().Services().Informer()
 		c.k8s.EventsServices(c.eventChan, c.statusChan, stop, svci, c.PublishService)
@@ -152,6 +157,30 @@ func (c *HAProxyController) getIngressSharedInformers(factory informers.SharedIn
 	return ii, ici
 }
 
+func (c *HAProxyController) getEndpointSlicesSharedInformer(factory informers.SharedInformerFactory) cache.SharedIndexInformer {
+	for i, apiGroup := range []string{"discovery.k8s.io/v1", "discovery.k8s.io/v1beta1"} {
+		resources, err := c.k8s.API.ServerResourcesForGroupVersion(apiGroup)
+		if err != nil {
+			continue
+		}
+
+		for _, rs := range resources.APIResources {
+			if rs.Name == "endpointslices" {
+				switch i {
+				case 0:
+					logger.Debug("Using discovery.k8s.io/v1 endpointslices")
+					return factory.Discovery().V1().EndpointSlices().Informer()
+
+				case 1:
+					logger.Debug("Using discovery.k8s.io/v1beta1 endpointslices")
+					return factory.Discovery().V1beta1().EndpointSlices().Informer()
+				}
+			}
+		}
+	}
+	return nil
+}
+
 func (c *HAProxyController) getWhitelistedNamespaces() []string {
 	if len(c.Store.NamespacesAccess.Whitelist) == 0 {
 		return []string{""}