Repository: submariner-io/lighthouse (go)

Issue:
Support Globalnet with Headless services
**What would you like to be added**:
Add support for Globalnet + Headless services

**Why is this needed**:
Current Globalnet deployments don't support headless services due to technical limitations. But [Globalnet v2](https://github.com/submariner-io/submariner/issues/652) supports headless services. This require changes in Lighthouse too.

Pull request:
Globalnet v2 support for headless services
* Add globalnet support for headless services
* Remove unused old code

Fixes #558 

Signed-off-by: Vishal Thapar <5137689+vthapar@users.noreply.github.com>

<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our developer guide: https://submariner.io/development/
2. Ensure you have added the appropriate tests for your PR: https://submariner.io/development/code-review/#test-new-functionality
3. Read the code review guide to ease the review process: https://submariner.io/development/code-review/
4. If the PR is unfinished, mark it as a draft: https://submariner.io/development/code-review/#mark-work-in-progress-prs-as-drafts
5. If you are using CI to debug, use your private fork: https://submariner.io/development/code-review/#use-private-forks-for-debugging-prs-by-running-ci
6. Add labels to the PR as appropriate.

This template is based on the K8s/K8s template:

https://github.com/kubernetes/kubernetes/blob/master/.github/PULL_REQUEST_TEMPLATE.md
-->


Patch:
diff --git a/pkg/agent/controller/agent.go b/pkg/agent/controller/agent.go
index c05042c..f51d32a 100644
--- a/pkg/agent/controller/agent.go
+++ b/pkg/agent/controller/agent.go
@@ -276,40 +276,6 @@ func (a *Controller) serviceExportToServiceImport(obj runtime.Object, numRequeue
 		return nil, false
 	}
 
-	if a.globalnetEnabled && svcType == mcsv1a1.Headless {
-		klog.Infof("Headless Services not supported with globalnet yet")
-		a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid,
-			corev1.ConditionFalse, invalidServiceType, "Headless Services not supported with Globalnet IP")
-
-		return nil, false
-	}
-
-	var ips []string
-	var ports []mcsv1a1.ServicePort
-	if a.globalnetEnabled {
-		ip, reason, msg := a.getGlobalIP(svc)
-		if ip == "" {
-			klog.V(log.DEBUG).Infof("Service to be exported (%s/%s) doesn't have a global IP yet", svcExport.Namespace, svcExport.Name)
-			// Globalnet enabled but service doesn't have globalIp yet, Update the status and requeue
-			a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid,
-				corev1.ConditionFalse, reason, msg)
-
-			return nil, true
-		}
-
-		ips = []string{ip}
-		ports = a.getPortsForService(svc)
-	} else {
-		ips, ports, err = a.getIPsAndPortsForService(svc, svcType)
-		if err != nil {
-			// Failed to get ips for some reason, requeue
-			a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid,
-				corev1.ConditionUnknown, "ServiceRetrievalFailed", err.Error())
-
-			return nil, true
-		}
-	}
-
 	serviceImport := a.newServiceImport(svcExport.Name, svcExport.Namespace)
 
 	serviceImport.Spec = mcsv1a1.ServiceImportSpec{
@@ -327,12 +293,27 @@ func (a *Controller) serviceExportToServiceImport(obj runtime.Object, numRequeue
 	}
 
 	if svcType == mcsv1a1.ClusterSetIP {
-		serviceImport.Spec.IPs = ips
-		serviceImport.Spec.Ports = ports
+		if a.globalnetEnabled {
+			ip, reason, msg := a.getGlobalIP(svc)
+			if ip == "" {
+				klog.V(log.DEBUG).Infof("Service to be exported (%s/%s) doesn't have a global IP yet", svcExport.Namespace, svcExport.Name)
+				// Globalnet enabled but service doesn't have globalIp yet, Update the status and requeue
+				a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid,
+					corev1.ConditionFalse, reason, msg)
+
+				return nil, true
+			}
+
+			serviceImport.Spec.IPs = []string{ip}
+		} else {
+			serviceImport.Spec.IPs = []string{svc.Spec.ClusterIP}
+		}
+
+		serviceImport.Spec.Ports = a.getPortsForService(svc)
 		/* We also store the clusterIP in an annotation as an optimization to recover it in case the IPs are
 		cleared out when here's no backing Endpoint pods.
 		*/
-		serviceImport.Annotations[clusterIP] = ips[0]
+		serviceImport.Annotations[clusterIP] = serviceImport.Spec.IPs[0]
 	}
 
 	a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid,
@@ -496,27 +477,6 @@ func (a *Controller) newServiceImport(name, namespace string) *mcsv1a1.ServiceIm
 	}
 }
 
-func (a *Controller) getIPsAndPortsForService(service *corev1.Service, siType mcsv1a1.ServiceImportType) (
-	[]string, []mcsv1a1.ServicePort, error) {
-	mcsPorts := a.getPortsForService(service)
-
-	if siType == mcsv1a1.ClusterSetIP {
-		return []string{service.Spec.ClusterIP}, mcsPorts, nil
-	}
-
-	endpoint, err := a.kubeClientSet.CoreV1().Endpoints(service.Namespace).Get(context.TODO(), service.Name, metav1.GetOptions{})
-	if err != nil {
-		if !apierrors.IsNotFound(err) {
-			klog.Errorf("Error retrieving Endpoints for Service (%s/%s): %v", service.Namespace, service.Name, err)
-			return nil, nil, errors.WithMessage(err, "Error retrieving the Endpoints for the Service")
-		}
-
-		return make([]string, 0), nil, nil
-	}
-
-	return getIPsFromEndpoint(endpoint), mcsPorts, nil
-}
-
 func (a *Controller) getPortsForService(service *corev1.Service) []mcsv1a1.ServicePort {
 	var mcsPorts = make([]mcsv1a1.ServicePort, 0, len(service.Spec.Ports))
 
@@ -535,20 +495,6 @@ func (a *Controller) getObjectNameWithClusterID(name, namespace string) string {
 	return name + "-" + namespace + "-" + a.clusterID
 }
 
-func getIPsFromEndpoint(endpoint *corev1.Endpoints) []string {
-	ipList := make([]string, 0)
-
-	for _, eps := range endpoint.Subsets {
-		for _, addr := range eps.Addresses {
-			ipList = append(ipList, addr.IP)
-		}
-	}
-
-	klog.V(log.DEBUG).Infof("IPList %v for endpoint (%s/%s)", ipList, endpoint.Namespace, endpoint.Name)
-
-	return ipList
-}
-
 func (a *Controller) remoteEndpointSliceToLocal(obj runtime.Object, numRequeues int, op syncer.Operation) (runtime.Object, bool) {
 	endpointSlice := obj.(*discovery.EndpointSlice)
 	endpointSlice.Namespace = endpointSlice.GetObjectMeta().GetLabels()[lhconstants.LabelSourceNamespace]
diff --git a/pkg/agent/controller/endpoint.go b/pkg/agent/controller/endpoint.go
index 32b855d..0cbb03b 100644
--- a/pkg/agent/controller/endpoint.go
+++ b/pkg/agent/controller/endpoint.go
@@ -29,27 +29,35 @@ import (
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/klog"
 	utilnet "k8s.io/utils/net"
+	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
 func startEndpointController(localClient dynamic.Interface, restMapper meta.RESTMapper, scheme *runtime.Scheme,
-	serviceImportUID types.UID, serviceImportName, serviceImportNameSpace, serviceName, clusterID string) (*EndpointController, error) {
+	serviceImport *mcsv1a1.ServiceImport, serviceImportNameSpace, serviceName, clusterID string,
+	globalnetEnabled bool) (*EndpointController, error) {
 	klog.V(log.DEBUG).Infof("Starting Endpoints controller for service %s/%s", serviceImportNameSpace, serviceName)
 
+	globalIngressIPGVR, _ := schema.ParseResourceArg("globalingressips.v1.submariner.io")
+
 	controller := &EndpointController{
 		clusterID:                    clusterID,
-		serviceImportUID:             serviceImportUID,
-		serviceImportName:            serviceImportName,
+		serviceImportUID:             serviceImport.UID,
+		serviceImportName:            serviceImport.Name,
 		serviceImportSourceNameSpace: serviceImportNameSpace,
 		serviceName:                  serviceName,
 		stopCh:                       make(chan struct{}),
+		isHeadless:                   serviceImport.Spec.Type == mcsv1a1.Headless,
+		globalnetEnabled:             globalnetEnabled,
+		localClient:                  localClient,
+		ingressIPClient:              localClient.Resource(*globalIngressIPGVR),
 	}
 
 	nameSelector := fields.OneTermEqualSelector("metadata.name", serviceName)
@@ -74,8 +82,6 @@ func startEndpointController(localClient dynamic.Interface, restMapper meta.REST
 		return nil, err
 	}
 
-	controller.localClient = localClient
-
 	return controller, nil
 }
 
@@ -122,10 +128,11 @@ func (e *EndpointController) endpointsToEndpointSlice(obj runtime.Object, numReq
 		klog.V(log.TRACE).Infof("Endpoints %s/%s updated", endPoints.Namespace, endPoints.Name)
 	}
 
-	return e.endpointSliceFromEndpoints(endPoints, op), false
+	return e.endpointSliceFromEndpoints(endPoints, op)
 }
 
-func (e *EndpointController) endpointSliceFromEndpoints(endpoints *corev1.Endpoints, op syncer.Operation) *discovery.EndpointSlice {
+func (e *EndpointController) endpointSliceFromEndpoints(endpoints *corev1.Endpoints, op syncer.Operation) (
+	runtime.Object, bool) {
 	endpointSlice := &discovery.EndpointSlice{}
 
 	endpointSlice.Name = endpoints.Name + "-" + e.clusterID
@@ -153,9 +160,19 @@ func (e *EndpointController) endpointSliceFromEndpoints(endpoints *corev1.Endpoi
 			endpointSlice.AddressType = discovery.AddressTypeIPv6
 		}
 
-		endpointSlice.Endpoints = append(endpointSlice.Endpoints, getEndpointsFromAddresses(subset.Addresses, endpointSlice.AddressType, true)...)
-		endpointSlice.Endpoints = append(endpointSlice.Endpoints, getEndpointsFromAddresses(subset.NotReadyAddresses,
-			endpointSlice.AddressType, false)...)
+		newEndpoints, retry := e.getEndpointsFromAddresses(subset.Addresses, endpointSlice.AddressType, true)
+		if retry {
+			return nil, true
+		}
+
+		endpointSlice.Endpoints = append(endpointSlice.Endpoints, newEndpoints...)
+		newEndpoints, retry = e.getEndpointsFromAddresses(subset.NotReadyAddresses, endpointSlice.AddressType, false)
+		if retry {
+			// TODO: We may not want unready endpoints at all
+			return nil, true
+		}
+
+		endpointSlice.Endpoints = append(endpointSlice.Endpoints, newEndpoints...)
 	}
 
 	if op == syncer.Create {
@@ -164,30 +181,42 @@ func (e *EndpointController) endpointSliceFromEndpoints(endpoints *corev1.Endpoi
 		klog.V(log.TRACE).Infof("Returning EndpointSlice: %#v", endpointSlice)
 	}
 
-	return endpointSlice
+	return endpointSlice, false
 }
 
-func getEndpointsFromAddresses(addresses []corev1.EndpointAddress, addressType discovery.AddressType, ready bool) []discovery.Endpoint {
+func (e *EndpointController) getEndpointsFromAddresses(addresses []corev1.EndpointAddress, addressType discovery.AddressType,
+	ready bool) ([]discovery.Endpoint, bool) {
 	endpoints := []discovery.Endpoint{}
 	isIPv6AddressType := addressType == discovery.AddressTypeIPv6
 
 	for _, address := range addresses {
 		if utilnet.IsIPv6String(address.IP) == isIPv6AddressType {
-			endpoints = append(endpoints, endpointFromAddress(address, ready))
+			endpoint, retry := e.endpointFromAddress(address, ready)
+			if retry {
+				return nil, true
+			}
+
+			endpoints = append(endpoints, *endpoint)
 		}
 	}
 
-	return endpoints
+	return endpoints, false
 }
 
-func endpointFromAddress(address corev1.EndpointAddress, ready bool) discovery.Endpoint {
+func (e *EndpointController) endpointFromAddress(address corev1.EndpointAddress, ready bool) (*discovery.Endpoint, bool) {
 	topology := map[string]string{}
 	if address.NodeName != nil {
 		topology["kubernetes.io/hostname"] = *address.NodeName
 	}
 
-	endpoint := discovery.Endpoint{
-		Addresses:  []string{address.IP},
+	ip := e.getIP(address)
+
+	if ip == "" {
+		return nil, true
+	}
+
+	endpoint := &discovery.Endpoint{
+		Addresses:  []string{ip},
 		Conditions: discovery.EndpointConditions{Ready: &ready},
 		Topology:   topology,
 	}
@@ -206,7 +235,7 @@ func endpointFromAddress(address corev1.EndpointAddress, ready bool) discovery.E
 		endpoint.Hostname = &address.TargetRef.Name
 	}
 
-	return endpoint
+	return endpoint, false
 }
 
 func allAddressesIPv6(addresses []corev1.EndpointAddress) bool {
@@ -222,3 +251,25 @@ func allAddressesIPv6(addresses []corev1.EndpointAddress) bool {
 
 	return true
 }
+
+func (e *EndpointController) getIP(address corev1.EndpointAddress) string {
+	if e.isHeadless && e.globalnetEnabled {
+		var ip string
+
+		name := "pod-" + address.TargetRef.Name
+
+		// TODO: Replace with a syncer.GetResource() to reduce API calls
+		obj, err := e.ingressIPClient.Namespace(e.serviceImportSourceNameSpace).Get(context.TODO(), name, metav1.GetOptions{})
+		if err == nil {
+			ip, _, _ = unstructured.NestedString(obj.Object, "status", "allocatedIP")
+		}
+
+		if ip == "" {
+			klog.Infof("GlobalIP for %s not allocated yet", name)
+		}
+
+		return ip
+	}
+
+	return address.IP
+}
diff --git a/pkg/agent/controller/serviceimport.go b/pkg/agent/controller/serviceimport.go
index d0e62e1..ba3ad28 100644
--- a/pkg/agent/controller/serviceimport.go
+++ b/pkg/agent/controller/serviceimport.go
@@ -34,11 +34,12 @@ import (
 func newServiceImportController(spec *AgentSpecification, serviceSyncer syncer.Interface, restMapper meta.RESTMapper,
 	localClient dynamic.Interface, scheme *runtime.Scheme) (*ServiceImportController, error) {
 	controller := &ServiceImportController{
-		serviceSyncer: serviceSyncer,
-		localClient:   localClient,
-		restMapper:    restMapper,
-		clusterID:     spec.ClusterID,
-		scheme:        scheme,
+		serviceSyncer:    serviceSyncer,
+		localClient:      localClient,
+		restMapper:       restMapper,
+		clusterID:        spec.ClusterID,
+		scheme:           scheme,
+		globalnetEnabled: spec.GlobalnetEnabled,
 	}
 
 	var err error
@@ -112,7 +113,7 @@ func (c *ServiceImportController) serviceImportCreatedOrUpdated(serviceImport *m
 	}
 
 	endpointController, err := startEndpointController(c.localClient, c.restMapper, c.scheme,
-		serviceImport.ObjectMeta.UID, serviceImport.ObjectMeta.Name, serviceNameSpace, serviceName, c.clusterID)
+		serviceImport, serviceNameSpace, serviceName, c.clusterID, c.globalnetEnabled)
 	if err != nil {
 		klog.Errorf(err.Error())
 		return true
diff --git a/pkg/agent/controller/types.go b/pkg/agent/controller/types.go
index 9f4c435..7030600 100644
--- a/pkg/agent/controller/types.go
+++ b/pkg/agent/controller/types.go
@@ -61,6 +61,7 @@ type ServiceImportController struct {
 	endpointControllers sync.Map
 	clusterID           string
 	scheme              *runtime.Scheme
+	globalnetEnabled    bool
 }
 
 // Each EndpointController listens for the endpoints that backs a service and have a ServiceImport
@@ -74,4 +75,7 @@ type EndpointController struct {
 	serviceImportSourceNameSpace string
 	stopCh                       chan struct{}
 	localClient                  dynamic.Interface
+	ingressIPClient              dynamic.NamespaceableResourceInterface
+	isHeadless                   bool
+	globalnetEnabled             bool
 }


Test patch:
diff --git a/pkg/agent/controller/controller_suite_test.go b/pkg/agent/controller/controller_suite_test.go
index 9e26aa2..14b1623 100644
--- a/pkg/agent/controller/controller_suite_test.go
+++ b/pkg/agent/controller/controller_suite_test.go
@@ -54,7 +54,9 @@ import (
 const clusterID1 = "east"
 const clusterID2 = "west"
 const serviceNamespace = "service-ns"
-const globalIP = "192.168.10.34"
+const globalIP1 = "242.254.1.1"
+const globalIP2 = "242.254.1.2"
+const globalIP3 = "242.254.1.3"
 
 var nodeName = "my-node"
 var hostName = "my-host"
@@ -94,9 +96,9 @@ type testDriver struct {
 	service                   *corev1.Service
 	serviceExport             *mcsv1a1.ServiceExport
 	endpoints                 *corev1.Endpoints
-	ingressIP                 *unstructured.Unstructured
 	stopCh                    chan struct{}
 	syncerConfig              *broker.SyncerConfig
+	endpointGlobalIPs         []string
 }
 
 func newTestDiver() *testDriver {
@@ -159,15 +161,24 @@ func newTestDiver() *testDriver {
 					{
 						IP:       "192.168.5.1",
 						Hostname: hostName,
+						TargetRef: &corev1.ObjectReference{
+							Name: "one",
+						},
 					},
 					{
 						IP:       "192.168.5.2",
 						NodeName: &nodeName,
+						TargetRef: &corev1.ObjectReference{
+							Name: "two",
+						},
 					},
 				},
 				NotReadyAddresses: []corev1.EndpointAddress{
 					{
 						IP: "10.253.6.1",
+						TargetRef: &corev1.ObjectReference{
+							Name: "not-ready",
+						},
 					},
 				},
 				Ports: []corev1.EndpointPort{
@@ -181,7 +192,6 @@ func newTestDiver() *testDriver {
 		},
 	}
 
-	t.ingressIP = t.newGlobalIngressIP()
 	t.brokerServiceImportClient = t.syncerConfig.BrokerClient.Resource(*test.GetGroupVersionResourceFor(t.syncerConfig.RestMapper,
 		&mcsv1a1.ServiceImport{})).Namespace(test.RemoteNamespace).(*fake.DynamicResourceClient)
 
@@ -194,14 +204,14 @@ func newTestDiver() *testDriver {
 	return t
 }
 
-func (t *testDriver) newGlobalIngressIP() *unstructured.Unstructured {
+func (t *testDriver) newGlobalIngressIP(name, ip string) *unstructured.Unstructured {
 	ingressIP := controller.GetGlobalIngressIPObj()
-	ingressIP.SetName(t.service.Name)
+	ingressIP.SetName(name)
 	ingressIP.SetNamespace(t.service.Namespace)
 	Expect(unstructured.SetNestedField(ingressIP.Object, controller.ClusterIPService, "spec", "target")).To(Succeed())
 	Expect(unstructured.SetNestedField(ingressIP.Object, t.service.Name, "spec", "serviceRef", "name")).To(Succeed())
 
-	setIngressAllocatedIP(ingressIP, globalIP)
+	setIngressAllocatedIP(ingressIP, ip)
 	setIngressIPConditions(ingressIP, metav1.Condition{
 		Type:    "Allocated",
 		Status:  metav1.ConditionTrue,
@@ -339,8 +349,8 @@ func (c *cluster) awaitUpdatedServiceImport(service *corev1.Service, serviceIP s
 	awaitUpdatedServiceImport(c.localServiceImportClient, service, serviceIP)
 }
 
-func awaitEndpointSlice(endpointSliceClient, serviceImportClient dynamic.ResourceInterface, endpoints *corev1.Endpoints,
-	service *corev1.Service, namespace string) *discovery.EndpointSlice {
+func awaitEndpointSlice(endpointSliceClient dynamic.ResourceInterface, endpoints *corev1.Endpoints,
+	service *corev1.Service, namespace string, globalIPs []string) *discovery.EndpointSlice {
 	obj := test.AwaitResource(endpointSliceClient, endpoints.Name+"-"+clusterID1)
 
 	endpointSlice := &discovery.EndpointSlice{}
@@ -358,19 +368,27 @@ func awaitEndpointSlice(endpointSliceClient, serviceImportClient dynamic.Resourc
 
 	Expect(endpointSlice.AddressType).To(Equal(discovery.AddressTypeIPv4))
 
+	addresses := globalIPs
+	if addresses == nil {
+		addresses = []string{endpoints.Subsets[0].Addresses[0].IP, endpoints.Subsets[0].Addresses[1].IP,
+			endpoints.Subsets[0].NotReadyAddresses[0].IP}
+	}
+
 	Expect(endpointSlice.Endpoints).To(HaveLen(3))
 	Expect(endpointSlice.Endpoints[0]).To(Equal(discovery.Endpoint{
-		Addresses:  []string{"192.168.5.1"},
+		Addresses:  []string{addresses[0]},
 		Conditions: discovery.EndpointConditions{Ready: &ready},
 		Hostname:   &hostName,
 	}))
 	Expect(endpointSlice.Endpoints[1]).To(Equal(discovery.Endpoint{
-		Addresses:  []string{"192.168.5.2"},
+		Addresses:  []string{addresses[1]},
+		Hostname:   &endpoints.Subsets[0].Addresses[1].TargetRef.Name,
 		Conditions: discovery.EndpointConditions{Ready: &ready},
 		Topology:   map[string]string{"kubernetes.io/hostname": nodeName},
 	}))
 	Expect(endpointSlice.Endpoints[2]).To(Equal(discovery.Endpoint{
-		Addresses:  []string{"10.253.6.1"},
+		Addresses:  []string{addresses[2]},
+		Hostname:   &endpoints.Subsets[0].NotReadyAddresses[0].TargetRef.Name,
 		Conditions: discovery.EndpointConditions{Ready: &notReady},
 	}))
 
@@ -389,8 +407,8 @@ func awaitEndpointSlice(endpointSliceClient, serviceImportClient dynamic.Resourc
 	return endpointSlice
 }
 
-func (c *cluster) awaitEndpointSlice(endpoints *corev1.Endpoints, service *corev1.Service) *discovery.EndpointSlice {
-	return awaitEndpointSlice(c.localEndpointSliceClient, c.localServiceImportClient, endpoints, service, service.Namespace)
+func (c *cluster) awaitEndpointSlice(t *testDriver) *discovery.EndpointSlice {
+	return awaitEndpointSlice(c.localEndpointSliceClient, t.endpoints, t.service, t.service.Namespace, t.endpointGlobalIPs)
 }
 
 func awaitUpdatedEndpointSlice(endpointSliceClient dynamic.ResourceInterface, endpoints *corev1.Endpoints, expectedIPs []string) {
@@ -433,7 +451,7 @@ func (t *testDriver) awaitBrokerServiceImport(sType mcsv1a1.ServiceImportType, s
 }
 
 func (t *testDriver) awaitBrokerEndpointSlice() *discovery.EndpointSlice {
-	return awaitEndpointSlice(t.brokerEndpointSliceClient, t.brokerServiceImportClient, t.endpoints, t.service, test.RemoteNamespace)
+	return awaitEndpointSlice(t.brokerEndpointSliceClient, t.endpoints, t.service, test.RemoteNamespace, t.endpointGlobalIPs)
 }
 
 func (t *testDriver) awaitUpdatedServiceImport(serviceIP string) {
@@ -444,8 +462,8 @@ func (t *testDriver) awaitUpdatedServiceImport(serviceIP string) {
 
 func (t *testDriver) awaitEndpointSlice() {
 	t.awaitBrokerEndpointSlice()
-	t.cluster1.awaitEndpointSlice(t.endpoints, t.service)
-	t.cluster2.awaitEndpointSlice(t.endpoints, t.service)
+	t.cluster1.awaitEndpointSlice(t)
+	t.cluster2.awaitEndpointSlice(t)
 }
 
 func (t *testDriver) awaitUpdatedEndpointSlice(expectedIPs []string) {
@@ -494,8 +512,15 @@ func (t *testDriver) deleteService() {
 		metav1.DeleteOptions{})).To(Succeed())
 }
 
-func (t *testDriver) createGlobalIngressIP() {
-	test.CreateResource(t.cluster1.localIngressIPClient, t.ingressIP)
+func (t *testDriver) createGlobalIngressIP(ingressIP *unstructured.Unstructured) {
+	test.CreateResource(t.cluster1.localIngressIPClient, ingressIP)
+}
+
+func (t *testDriver) createEndpointIngressIPs() {
+	t.endpointGlobalIPs = []string{globalIP1, globalIP2, globalIP3}
+	t.createGlobalIngressIP(t.newGlobalIngressIP("pod-one", globalIP1))
+	t.createGlobalIngressIP(t.newGlobalIngressIP("pod-two", globalIP2))
+	t.createGlobalIngressIP(t.newGlobalIngressIP("pod-not-ready", globalIP3))
 }
 
 func (t *testDriver) dynamicServiceClient() dynamic.ResourceInterface {
diff --git a/pkg/agent/controller/globalnet_test.go b/pkg/agent/controller/globalnet_test.go
index 29e6bc9..43b150f 100644
--- a/pkg/agent/controller/globalnet_test.go
+++ b/pkg/agent/controller/globalnet_test.go
@@ -18,10 +18,13 @@ limitations under the License.
 package controller_test
 
 import (
+	"time"
+
 	. "github.com/onsi/ginkgo"
 	"github.com/submariner-io/admiral/pkg/syncer/test"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
@@ -45,14 +48,20 @@ var _ = Describe("Globalnet enabled", func() {
 	})
 
 	When("a ClusterIP Service is exported", func() {
+		var ingressIP *unstructured.Unstructured
+
+		BeforeEach(func() {
+			ingressIP = t.newGlobalIngressIP(t.service.Name, globalIP1)
+		})
+
 		Context("and it has a global IP", func() {
 			Context("via a GlobalIngressIP", func() {
 				BeforeEach(func() {
-					t.createGlobalIngressIP()
+					t.createGlobalIngressIP(ingressIP)
 				})
 
 				It("should sync a ServiceImport with the global IP", func() {
-					t.awaitServiceExported(globalIP, 0)
+					t.awaitServiceExported(globalIP1, 0)
 				})
 			})
 		})
@@ -63,25 +72,25 @@ var _ = Describe("Globalnet enabled", func() {
 					t.awaitServiceExportStatus(0, newServiceExportCondition(mcsv1a1.ServiceExportValid,
 						corev1.ConditionFalse, "ServiceGlobalIPUnavailable"))
 
-					t.createGlobalIngressIP()
-					t.awaitServiceExported(globalIP, 1)
+					t.createGlobalIngressIP(ingressIP)
+					t.awaitServiceExported(globalIP1, 1)
 				})
 			})
 
 			Context("due to no AllocatedIP in the GlobalIngressIP", func() {
 				BeforeEach(func() {
-					setIngressAllocatedIP(t.ingressIP, "")
-					setIngressIPConditions(t.ingressIP)
-					t.createGlobalIngressIP()
+					setIngressAllocatedIP(ingressIP, "")
+					setIngressIPConditions(ingressIP)
+					t.createGlobalIngressIP(ingressIP)
 				})
 
 				It("should update the ServiceExport status appropriately and eventually sync a ServiceImport", func() {
 					t.awaitServiceExportStatus(0, newServiceExportCondition(mcsv1a1.ServiceExportValid,
 						corev1.ConditionFalse, "ServiceGlobalIPUnavailable"))
 
-					setIngressAllocatedIP(t.ingressIP, globalIP)
-					test.UpdateResource(t.cluster1.localIngressIPClient, t.ingressIP)
-					t.awaitServiceExported(globalIP, 1)
+					setIngressAllocatedIP(ingressIP, globalIP1)
+					test.UpdateResource(t.cluster1.localIngressIPClient, ingressIP)
+					t.awaitServiceExported(globalIP1, 1)
 				})
 			})
 		})
@@ -95,9 +104,9 @@ var _ = Describe("Globalnet enabled", func() {
 			}
 
 			BeforeEach(func() {
-				setIngressAllocatedIP(t.ingressIP, "")
-				setIngressIPConditions(t.ingressIP, condition)
-				t.createGlobalIngressIP()
+				setIngressAllocatedIP(ingressIP, "")
+				setIngressIPConditions(ingressIP, condition)
+				t.createGlobalIngressIP(ingressIP)
 			})
 
 			It("should update the ServiceExport status with the condition details", func() {
@@ -113,11 +122,31 @@ var _ = Describe("Globalnet enabled", func() {
 			t.service.Spec.ClusterIP = corev1.ClusterIPNone
 		})
 
-		It("should update the ServiceExport status and not sync a ServiceImport", func() {
-			t.awaitServiceExportStatus(0, newServiceExportCondition(mcsv1a1.ServiceExportValid,
-				corev1.ConditionFalse, "UnsupportedServiceType"))
+		JustBeforeEach(func() {
+			t.createEndpoints()
+		})
 
-			t.awaitNoServiceImport(t.brokerServiceImportClient)
+		Context("and it has a global IP for all endpoint addresses", func() {
+			BeforeEach(func() {
+				t.createEndpointIngressIPs()
+			})
+
+			It("should sync a ServiceImport and EndpointSlice with the global IPs", func() {
+				t.awaitHeadlessServiceImport("")
+				t.awaitEndpointSlice()
+			})
+		})
+
+		Context("and it initially does not have a global IP for all endpoint addresses", func() {
+			It("should eventually sync a ServiceImport and EndpointSlice with the global IPs", func() {
+				time.Sleep(time.Millisecond * 300)
+				t.awaitNoEndpointSlice(t.cluster1.localEndpointSliceClient)
+
+				t.createEndpointIngressIPs()
+
+				t.awaitHeadlessServiceImport("")
+				t.awaitEndpointSlice()
+			})
 		})
 	})
 })
diff --git a/pkg/agent/controller/reconciliation_test.go b/pkg/agent/controller/reconciliation_test.go
index 65384da..5ad5b39 100644
--- a/pkg/agent/controller/reconciliation_test.go
+++ b/pkg/agent/controller/reconciliation_test.go
@@ -52,7 +52,7 @@ var _ = Describe("Reconciliation", func() {
 
 		It("should delete the ServiceImport and EndpointSlice on reconciliation", func() {
 			serviceImport := t.cluster1.awaitServiceImport(t.service, mcsv1a1.Headless, "")
-			endpointSlice := t.cluster1.awaitEndpointSlice(t.endpoints, t.service)
+			endpointSlice := t.cluster1.awaitEndpointSlice(t)
 
 			t.afterEach()
 			t = newTestDiver()
@@ -143,7 +143,7 @@ var _ = Describe("Reconciliation", func() {
 		})
 
 		It("should delete it from the local datastore on reconciliation", func() {
-			endpointSlice := t.cluster2.awaitEndpointSlice(t.endpoints, t.service)
+			endpointSlice := t.cluster2.awaitEndpointSlice(t)
 
 			t.afterEach()
 			t = newTestDiver()
diff --git a/pkg/agent/controller/service_export_failures_test.go b/pkg/agent/controller/service_export_failures_test.go
index 6168689..1108765 100644
--- a/pkg/agent/controller/service_export_failures_test.go
+++ b/pkg/agent/controller/service_export_failures_test.go
@@ -21,10 +21,8 @@ import (
 	"errors"
 
 	. "github.com/onsi/ginkgo"
-	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
 var _ = Describe("Service export failures", func() {
@@ -45,20 +43,6 @@ var _ = Describe("Service export failures", func() {
 		t.afterEach()
 	})
 
-	When("Endpoints retrieval initially fails", func() {
-		BeforeEach(func() {
-			t.service.Spec.ClusterIP = corev1.ClusterIPNone
-			t.cluster1.endpointsReactor.SetFailOnGet(errors.New("fake Get error"))
-		})
-
-		It("should update the ServiceExport status and eventually sync a ServiceImport", func() {
-			t.awaitServiceExportStatus(0, newServiceExportCondition(mcsv1a1.ServiceExportValid,
-				corev1.ConditionUnknown, "ServiceRetrievalFailed"))
-			t.cluster1.endpointsReactor.SetResetOnFailure(true)
-			t.awaitHeadlessServiceImport("")
-		})
-	})
-
 	When("a conflict initially occurs when updating the ServiceExport status", func() {
 		BeforeEach(func() {
 			t.cluster1.localServiceExportClient.FailOnUpdate = apierrors.NewConflict(schema.GroupResource{}, t.serviceExport.Name,