Update endpoint slices during service import reconciliation (#91)

astaticvoid · web-flow · commit 3c71e872172c · 2021-11-18T20:59:30.000-08:00
diff --git a/integration/scripts/run-tests.sh b/integration/scripts/run-tests.sh
@@ -42,12 +42,10 @@ if [ "$exit_code" -eq 0 ] ; then
   go run $SCENARIOS/runner/main.go $NAMESPACE $SERVICE $ENDPT_PORT $SERVICE_PORT "$updated_endpoints"
   exit_code=$?
 
-  # TODO: The reconciliation during the import is not implemented yet, uncomment below once done
-  #
-  #if [ "$exit_code" -eq 0 ] ; then
-  #  ./integration/scripts/test-import.sh "$UPDATED_ENDPOINT_COUNT" "$updated_endpoints"
-  #  exit_code=$?
-  #fi
+  if [ "$exit_code" -eq 0 ] ; then
+    ./integration/scripts/test-import.sh "$UPDATED_ENDPOINT_COUNT" "$updated_endpoints"
+    exit_code=$?
+  fi
 fi
 
 echo "killing controller PID:$CTL_PID"
diff --git a/pkg/controllers/cloudmap_controller.go b/pkg/controllers/cloudmap_controller.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/base32"
+	"fmt"
 	"github.com/aws/aws-cloud-map-mcs-controller-for-k8s/pkg/api/v1alpha1"
 	"github.com/aws/aws-cloud-map-mcs-controller-for-k8s/pkg/cloudmap"
 	"github.com/aws/aws-cloud-map-mcs-controller-for-k8s/pkg/model"
@@ -24,6 +25,8 @@ const (
 	// TODO move to configuration
 	syncPeriod = 2 * time.Second
 
+	maxEndpointsPerSlice = 100
+
 	// DerivedServiceAnnotation annotates a ServiceImport with derived Service name
 	DerivedServiceAnnotation = "multicluster.k8s.aws/derived-service"
 
@@ -156,7 +159,7 @@ func (r *CloudMapReconciler) reconcileService(ctx context.Context, svc *model.Se
 		return err
 	}
 
-	err = r.updateEndpointSlices(ctx, svcImport, svc, derivedService)
+	err = r.updateEndpointSlices(ctx, svcImport, svc.Endpoints, derivedService)
 	if err != nil {
 		return err
 	}
@@ -199,7 +202,7 @@ func (r *CloudMapReconciler) getDerivedService(ctx context.Context, namespace st
 }
 
 func (r *CloudMapReconciler) createAndGetDerivedService(ctx context.Context, svc *model.Service, svcImport *v1alpha1.ServiceImport) (*v1.Service, error) {
-	toCreate := createDerivedServiceStruct(svc, svcImport)
+	toCreate := createDerivedServiceStruct(svc.Endpoints, svcImport)
 	if err := r.Client.Create(ctx, toCreate); err != nil {
 		return nil, err
 	}
@@ -208,25 +211,100 @@ func (r *CloudMapReconciler) createAndGetDerivedService(ctx context.Context, svc
 	return r.getDerivedService(ctx, svc.Namespace, svcImport.Annotations[DerivedServiceAnnotation])
 }
 
-func (r *CloudMapReconciler) updateEndpointSlices(ctx context.Context, svcImport *v1alpha1.ServiceImport, cloudMapService *model.Service, svc *v1.Service) error {
+func (r *CloudMapReconciler) updateEndpointSlices(ctx context.Context, svcImport *v1alpha1.ServiceImport, desiredEndpoints []*model.Endpoint, svc *v1.Service) error {
 	existingSlicesList := discovery.EndpointSliceList{}
-	var existingSlices []*discovery.EndpointSlice
 	if err := r.Client.List(ctx, &existingSlicesList,
 		client.InNamespace(svc.Namespace), client.MatchingLabels{discovery.LabelServiceName: svc.Name}); err != nil {
 		return err
 	}
-	if len(existingSlicesList.Items) == 0 {
-		// create new endpoint slice
-		existingSlices = createEndpointSlicesStruct(svcImport, cloudMapService, svc)
-		for _, slice := range existingSlices {
-			if err := r.Client.Create(ctx, slice); err != nil {
-				return err
+
+	desiredPorts := extractEndpointPorts(desiredEndpoints)
+	matchedEndpoints := make(map[string]*discovery.Endpoint)
+	endpointsToCreate := make([]discovery.Endpoint, 0)
+
+	// populate map of existing endpoints in slices for lookup efficiency
+	existingEndpointMap := make(map[string]*discovery.Endpoint)
+	for _, existingSlice := range existingSlicesList.Items {
+		for _, existingEndpoint := range existingSlice.Endpoints {
+			ref := existingEndpoint
+			existingEndpointMap[ref.Addresses[0]] = &ref
+		}
+	}
+
+	// check if all desired endpoints are in an endpoint slice already
+	for _, desiredEndpoint := range desiredEndpoints {
+		match, exists := existingEndpointMap[desiredEndpoint.IP]
+		if exists {
+			matchedEndpoints[desiredEndpoint.IP] = match
+		} else {
+			endpointsToCreate = append(endpointsToCreate, createEndpointForSlice(svc, desiredEndpoint.IP))
+		}
+	}
+
+	// check if all endpoints in slices match a desired endpoint,
+	for _, existingSlice := range existingSlicesList.Items {
+		updatedEndpointList := make([]discovery.Endpoint, 0)
+		for _, existingEndpoint := range existingSlice.Endpoints {
+			keep, found := matchedEndpoints[existingEndpoint.Addresses[0]]
+			if found {
+				updatedEndpointList = append(updatedEndpointList, *keep)
+			}
+		}
+
+		endpointSliceNeedsUpdate := len(existingSlice.Endpoints) != len(updatedEndpointList)
+
+		// fill endpoint slice with endpoints to create if necessary and there is sufficient room
+		for _, endpointToCreate := range endpointsToCreate {
+			if len(updatedEndpointList) >= maxEndpointsPerSlice {
+				break
+			}
+			endpointSliceNeedsUpdate = true
+			updatedEndpointList = append(updatedEndpointList, endpointToCreate)
+			endpointsToCreate = endpointsToCreate[1:]
+		}
+
+		sliceToUpdate := existingSlice
+		sliceToUpdate.Endpoints = updatedEndpointList
+
+		// delete empty endpoint slice
+		if len(updatedEndpointList) == 0 {
+			r.Logger.Info("deleting EndpointSlice", "namespace", sliceToUpdate.Namespace, "name", sliceToUpdate.Name)
+			if err := r.Client.Delete(ctx, &sliceToUpdate); err != nil {
+				return fmt.Errorf("failed to delete EndpointSlice: %w", err)
+			}
+			continue
+		}
+
+		// needsUpdate = true if ports don't match
+		if !EndpointPortsAreEqualIgnoreOrder(desiredPorts, sliceToUpdate.Ports) {
+			sliceToUpdate.Ports = desiredPorts
+			endpointSliceNeedsUpdate = true
+		}
+
+		if endpointSliceNeedsUpdate {
+			r.Logger.Info("updating EndpointSlice", "namespace", sliceToUpdate.Namespace, "name", sliceToUpdate.Name)
+			if err := r.Client.Update(ctx, &sliceToUpdate); err != nil {
+				return fmt.Errorf("failed to update EndpointSlice: %w", err)
 			}
-			r.Logger.Info("created EndpointSlice", "namespace", slice.Namespace, "name", slice.Name)
 		}
 	}
 
-	// TODO check existing slices match Cloud Map endpoints
+	slicesToCreate := make([]*discovery.EndpointSlice, 0)
+	for len(endpointsToCreate) > maxEndpointsPerSlice {
+		slicesToCreate = append(slicesToCreate, createEndpointSliceStruct(svcImport, svc, endpointsToCreate[0:maxEndpointsPerSlice], desiredPorts))
+		endpointsToCreate = endpointsToCreate[maxEndpointsPerSlice:]
+	}
+
+	if len(endpointsToCreate) != 0 {
+		slicesToCreate = append(slicesToCreate, createEndpointSliceStruct(svcImport, svc, endpointsToCreate, desiredPorts))
+	}
+
+	for _, newSlice := range slicesToCreate {
+		r.Logger.Info("creating EndpointSlice", "namespace", newSlice.Namespace)
+		if err := r.Client.Create(ctx, newSlice); err != nil {
+			return fmt.Errorf("failed to create EndpointSlice: %w", err)
+		}
+	}
 
 	return nil
 }
@@ -238,7 +316,7 @@ func DerivedName(namespace string, name string) string {
 	return "imported-" + strings.ToLower(base32.HexEncoding.WithPadding(base32.NoPadding).EncodeToString(hash.Sum(nil)))[:10]
 }
 
-func createDerivedServiceStruct(svc *model.Service, svcImport *v1alpha1.ServiceImport) *v1.Service {
+func createDerivedServiceStruct(endpoints []*model.Endpoint, svcImport *v1alpha1.ServiceImport) *v1.Service {
 	ownerRef := metav1.NewControllerRef(svcImport, schema.GroupVersionKind{
 		Version: svcImport.TypeMeta.APIVersion,
 		Kind:    svcImport.TypeMeta.Kind,
@@ -252,37 +330,31 @@ func createDerivedServiceStruct(svc *model.Service, svcImport *v1alpha1.ServiceI
 		},
 		Spec: v1.ServiceSpec{
 			Type:  v1.ServiceTypeClusterIP,
-			Ports: extractServicePorts(svc),
+			Ports: extractServicePorts(endpoints),
 		},
 	}
 }
 
-func createEndpointSlicesStruct(svcImport *v1alpha1.ServiceImport, cloudMapSvc *model.Service, svc *v1.Service) []*discovery.EndpointSlice {
-	slices := make([]*discovery.EndpointSlice, 0)
-
+func createEndpointForSlice(svc *v1.Service, ip string) discovery.Endpoint {
 	t := true
 
-	endpoints := make([]discovery.Endpoint, 0)
-	for _, ep := range cloudMapSvc.Endpoints {
-		endpoints = append(endpoints, discovery.Endpoint{
-			Addresses: []string{ep.IP},
-			Conditions: discovery.EndpointConditions{
-				Ready: &t,
-			},
-			TargetRef: &v1.ObjectReference{
-				Kind:            "Service",
-				Namespace:       svc.Namespace,
-				Name:            svc.Name,
-				UID:             svc.ObjectMeta.UID,
-				ResourceVersion: svc.ObjectMeta.ResourceVersion,
-			},
-		})
+	return discovery.Endpoint{
+		Addresses: []string{ip},
+		Conditions: discovery.EndpointConditions{
+			Ready: &t,
+		},
+		TargetRef: &v1.ObjectReference{
+			Kind:            "Service",
+			Namespace:       svc.Namespace,
+			Name:            svc.Name,
+			UID:             svc.ObjectMeta.UID,
+			ResourceVersion: svc.ObjectMeta.ResourceVersion,
+		},
 	}
+}
 
-	// TODO split slices in case there are more than 1000 endpoints
-	// see https://github.com/kubernetes/enhancements/blob/master/keps/sig-network/0752-endpointslices/README.md
-
-	slices = append(slices, &discovery.EndpointSlice{
+func createEndpointSliceStruct(svcImport *v1alpha1.ServiceImport, svc *v1.Service, endpoints []discovery.Endpoint, ports []discovery.EndpointPort) *discovery.EndpointSlice {
+	return &discovery.EndpointSlice{
 		ObjectMeta: metav1.ObjectMeta{
 			Labels: map[string]string{
 				discovery.LabelServiceName: svc.Name,       // derived Service name
@@ -297,15 +369,13 @@ func createEndpointSlicesStruct(svcImport *v1alpha1.ServiceImport, cloudMapSvc *
 		},
 		AddressType: discovery.AddressTypeIPv4,
 		Endpoints:   endpoints,
-		Ports:       extractEndpointPorts(cloudMapSvc),
-	})
-
-	return slices
+		Ports:       ports,
+	}
 }
 
-func extractServicePorts(svc *model.Service) []v1.ServicePort {
+func extractServicePorts(endpoints []*model.Endpoint) []v1.ServicePort {
 	uniquePorts := make(map[string]model.Port)
-	for _, ep := range svc.Endpoints {
+	for _, ep := range endpoints {
 		uniquePorts[ep.ServicePort.GetID()] = ep.ServicePort
 	}
 
@@ -317,16 +387,17 @@ func extractServicePorts(svc *model.Service) []v1.ServicePort {
 	return servicePorts
 }
 
-func extractEndpointPorts(svc *model.Service) []discovery.EndpointPort {
+func extractEndpointPorts(endpoints []*model.Endpoint) []discovery.EndpointPort {
 	uniquePorts := make(map[string]model.Port)
-	for _, ep := range svc.Endpoints {
+	for _, ep := range endpoints {
 		uniquePorts[ep.EndpointPort.GetID()] = ep.EndpointPort
 	}
 
 	endpointPorts := make([]discovery.EndpointPort, 0, len(uniquePorts))
 	for _, endpointPort := range uniquePorts {
 		endpointPorts = append(endpointPorts, PortToEndpointPort(endpointPort))
 	}
+
 	return endpointPorts
 }
 
@@ -356,6 +427,7 @@ func portsEqual(svcImport *v1alpha1.ServiceImport, svc *v1.Service) bool {
 		svcPorts = append(svcPorts, servicePortToServiceImport(p))
 	}
 
+	// TODO: consider order
 	return reflect.DeepEqual(impPorts, svcPorts)
 }
 
diff --git a/pkg/controllers/utils.go b/pkg/controllers/utils.go
@@ -5,6 +5,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	discovery "k8s.io/api/discovery/v1beta1"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"reflect"
 )
 
 func ServicePortToPort(svcPort v1.ServicePort) model.Port {
@@ -67,3 +68,23 @@ func stringToProtocol(protocol string) v1.Protocol {
 		return ""
 	}
 }
+
+func EndpointPortsAreEqualIgnoreOrder(a, b []discovery.EndpointPort) (equal bool) {
+	if len(a) != len(b) {
+		return false
+	}
+
+	for _, aPort := range a {
+		match := false
+		for _, bPort := range b {
+			if reflect.DeepEqual(aPort, bPort) {
+				match = true
+				break
+			}
+		}
+		if !match {
+			return false
+		}
+	}
+	return true
+}
