Repository: submariner-io/lighthouse
Pull request #1144 (2023-03-15 20:21:39)

Implement ServiceImport aggregation
This PR contains two main commits: the [first](https://github.com/submariner-io/lighthouse/pull/1144/commits/288a00ab9fc1673d7dbd20f8c0e43191635aee45) to implement `ServiceImport` aggregation in the agent controller and the [second](https://github.com/submariner-io/lighthouse/pull/1144/commits/ec8c07d7ac0a0746e58641a7de15f3ee34d22bfb) to use the aggregated `ServiceImports` in the CoreDNS plugin. See commits for details.

There's still remaining work described as TODOs in the commits that will be completed in subsequent PRs.

Related to https://github.com/submariner-io/lighthouse/issues/214
Depends on https://github.com/submariner-io/admiral/pull/571


submariner-bot: 🤖 Created branch: z_pr1144/tpantelis/si_aggregate 
🚀 Full E2E won't run until the "ready-to-test" label is applied. I will add it automatically once the PR has 2 approvals, or you can add it manually.
github-actions[bot]: This PR/issue depends on:
* submariner-io/admiral#571
By **[Dependent Issues](https://github.com/z0al/dependent-issues)** (🤖). Happy coding!
<!-- By Dependent Issues (Action) - DO NOT REMOVE -->
tpantelis: > Doesn't this depend on https://github.com/submariner-io/admiral/pull/572 too?

It didn't really need to but now that's it's been merged I can remove pending from the test.


github-actions[bot]: This PR/issue depends on:
* ~~submariner-io/admiral#571~~
By **[Dependent Issues](https://github.com/z0al/dependent-issues)** (🤖). Happy coding!
<!-- By Dependent Issues (Action) - DO NOT REMOVE -->

Patch:
diff --git a/Makefile b/Makefile
index 4af1583..339e3ea 100644
--- a/Makefile
+++ b/Makefile
@@ -59,7 +59,7 @@ export-nginx: deploy-latest
 
 check-nginx:
 	KUBECONFIG=output/kubeconfigs/kind-config-cluster1 kubectl get serviceexports.multicluster.x-k8s.io -n default nginx-upgrade
-	KUBECONFIG=output/kubeconfigs/kind-config-cluster2 kubectl get serviceimports.multicluster.x-k8s.io -n submariner-operator nginx-upgrade-default-cluster1
+	KUBECONFIG=output/kubeconfigs/kind-config-cluster2 kubectl get serviceimports.multicluster.x-k8s.io -n default nginx-upgrade
 
 $(TARGETS):
 	./scripts/$@
diff --git a/coredns/constants/constants.go b/coredns/constants/constants.go
index 3272565..1f48db7 100644
--- a/coredns/constants/constants.go
+++ b/coredns/constants/constants.go
@@ -19,9 +19,9 @@ limitations under the License.
 package constants
 
 const (
-	LoadBalancerWeightAnnotationPrefix = "lighthouse-lb-weight.submariner.io"
-	KubernetesServiceName              = "kubernetes.io/service-name"
-	LabelValueManagedBy                = "lighthouse-agent.submariner.io"
-	LabelSourceNamespace               = "lighthouse.submariner.io/sourceNamespace"
-	MCSLabelSourceCluster              = "multicluster.kubernetes.io/source-cluster"
+	KubernetesServiceName = "kubernetes.io/service-name"
+	LabelValueManagedBy   = "lighthouse-agent.submariner.io"
+	LabelSourceNamespace  = "lighthouse.submariner.io/sourceNamespace"
+	MCSLabelSourceCluster = "multicluster.kubernetes.io/source-cluster"
+	LabelIsHeadless       = "lighthouse.submariner.io/is-headless"
 )
diff --git a/coredns/resolver/endpoint_slice.go b/coredns/resolver/endpoint_slice.go
index 663c28b..fc9d5a4 100644
--- a/coredns/resolver/endpoint_slice.go
+++ b/coredns/resolver/endpoint_slice.go
@@ -33,13 +33,17 @@ import (
 )
 
 func (i *Interface) PutEndpointSlice(endpointSlice *discovery.EndpointSlice) bool {
-	logger.Infof("Put EndpointSlice: %#v", endpointSlice.ObjectMeta)
-
 	key, clusterID, ok := getKeyInfoFrom(endpointSlice)
 	if !ok {
 		return false
 	}
 
+	if ignoreEndpointSlice(endpointSlice) {
+		return false
+	}
+
+	logger.Infof("Put EndpointSlice %q on cluster %q", key, clusterID)
+
 	var localEndpointSliceErr error
 	var localEndpointSlice *discovery.EndpointSlice
 
@@ -78,14 +82,42 @@ func (i *Interface) PutEndpointSlice(endpointSlice *discovery.EndpointSlice) boo
 }
 
 func (i *Interface) putClusterIPEndpointSlice(key, clusterID string, endpointSlice *discovery.EndpointSlice, serviceInfo *serviceInfo) bool {
-	clusterInfo, found := serviceInfo.clusters[clusterID]
+	_, found := endpointSlice.Labels[constants.LabelIsHeadless]
 	if !found {
-		logger.Infof("Cluster %q not found for EndpointSlice %q - requeuing", clusterID, key)
-		return true
+		// This is a legacy pre-0.15 EndpointSlice.
+		clusterInfo, found := serviceInfo.clusters[clusterID]
+		if !found {
+			logger.Infof("Cluster %q not found for EndpointSlice %q - requeuing", clusterID, key)
+			return true
+		}
+
+		// For a ClusterIPService we really only care if there are any backing endpoints.
+		clusterInfo.endpointsHealthy = len(endpointSlice.Endpoints) > 0
+
+		return false
+	}
+
+	if len(endpointSlice.Endpoints) == 0 {
+		// This shouldn't happen - we expect the service IP endpoint to always be present.
+		logger.Errorf(nil, "Missing service IP endpoint in EndpointSlice %q", key)
+
+		return false
 	}
 
-	// For a ClusterIPService we really only care if there are any backing endpoints.
-	clusterInfo.endpointsHealthy = len(endpointSlice.Endpoints) > 0
+	clusterInfo := serviceInfo.ensureClusterInfo(clusterID)
+	clusterInfo.endpointRecords = []DNSRecord{{
+		IP:          endpointSlice.Endpoints[0].Addresses[0],
+		Ports:       mcsServicePortsFrom(endpointSlice.Ports),
+		ClusterName: clusterID,
+	}}
+
+	clusterInfo.endpointsHealthy = endpointSlice.Endpoints[0].Conditions.Ready == nil || *endpointSlice.Endpoints[0].Conditions.Ready
+
+	serviceInfo.mergePorts()
+	serviceInfo.resetLoadBalancing()
+
+	logger.Infof("Added DNSRecord with service IP %q for EndpointSlice %q on cluster %q, endpointsHealthy: %v, ports: %#v",
+		clusterInfo.endpointRecords[0].IP, key, clusterID, clusterInfo.endpointsHealthy, clusterInfo.endpointRecords[0].Ports)
 
 	return false
 }
@@ -98,15 +130,7 @@ func (i *Interface) putHeadlessEndpointSlice(key, clusterID string, endpointSlic
 
 	serviceInfo.clusters[clusterID] = clusterInfo
 
-	mcsPorts := make([]mcsv1a1.ServicePort, len(endpointSlice.Ports))
-	for i, port := range endpointSlice.Ports {
-		mcsPorts[i] = mcsv1a1.ServicePort{
-			Name:        *port.Name,
-			Protocol:    *port.Protocol,
-			AppProtocol: port.AppProtocol,
-			Port:        *port.Port,
-		}
-	}
+	mcsPorts := mcsServicePortsFrom(endpointSlice.Ports)
 
 	for i := range endpointSlice.Endpoints {
 		endpoint := &endpointSlice.Endpoints[i]
@@ -184,13 +208,17 @@ func (i *Interface) getLocalEndpointSlice(from *discovery.EndpointSlice) (*disco
 }
 
 func (i *Interface) RemoveEndpointSlice(endpointSlice *discovery.EndpointSlice) {
-	logger.Infof("Remove EndpointSlice: %#v", endpointSlice.ObjectMeta)
-
 	key, clusterID, ok := getKeyInfoFrom(endpointSlice)
 	if !ok {
 		return
 	}
 
+	if ignoreEndpointSlice(endpointSlice) {
+		return
+	}
+
+	logger.Infof("Remove EndpointSlice %q on cluster %q", key, clusterID)
+
 	i.mutex.Lock()
 	defer i.mutex.Unlock()
 
@@ -199,17 +227,12 @@ func (i *Interface) RemoveEndpointSlice(endpointSlice *discovery.EndpointSlice)
 		return
 	}
 
-	clusterInfo, found := serviceInfo.clusters[clusterID]
-	if !found {
-		return
-	}
+	delete(serviceInfo.clusters, clusterID)
 
 	if !serviceInfo.isHeadless {
-		clusterInfo.endpointsHealthy = false
-		return
+		serviceInfo.mergePorts()
+		serviceInfo.resetLoadBalancing()
 	}
-
-	delete(serviceInfo.clusters, clusterID)
 }
 
 func getKeyInfoFrom(es *discovery.EndpointSlice) (string, string, bool) {
@@ -233,3 +256,22 @@ func getKeyInfoFrom(es *discovery.EndpointSlice) (string, string, bool) {
 
 	return keyFunc(namespace, name), clusterID, true
 }
+
+func mcsServicePortsFrom(ports []discovery.EndpointPort) []mcsv1a1.ServicePort {
+	mcsPorts := make([]mcsv1a1.ServicePort, len(ports))
+	for i, port := range ports {
+		mcsPorts[i] = mcsv1a1.ServicePort{
+			Name:        *port.Name,
+			Protocol:    *port.Protocol,
+			AppProtocol: port.AppProtocol,
+			Port:        *port.Port,
+		}
+	}
+
+	return mcsPorts
+}
+
+func ignoreEndpointSlice(eps *discovery.EndpointSlice) bool {
+	isOnBroker := eps.Namespace != eps.Labels[constants.LabelSourceNamespace]
+	return isOnBroker
+}
diff --git a/coredns/resolver/service_import.go b/coredns/resolver/service_import.go
index 583d05d..5a20a52 100644
--- a/coredns/resolver/service_import.go
+++ b/coredns/resolver/service_import.go
@@ -19,22 +19,18 @@ limitations under the License.
 package resolver
 
 import (
-	"strconv"
-
-	"github.com/submariner-io/lighthouse/coredns/constants"
 	"github.com/submariner-io/lighthouse/coredns/loadbalancer"
 	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
 func (i *Interface) PutServiceImport(serviceImport *mcsv1a1.ServiceImport) {
-	logger.Infof("Put %#v", serviceImport)
-
-	name, ok := getSourceName(serviceImport)
-	if !ok {
+	if ignoreServiceImport(serviceImport) {
 		return
 	}
 
-	key := keyFunc(getSourceNamespace(serviceImport), name)
+	key, isLegacy := getServiceImportKey(serviceImport)
+
+	logger.Infof("Put ServiceImport %q", key)
 
 	i.mutex.Lock()
 	defer i.mutex.Unlock()
@@ -51,96 +47,56 @@ func (i *Interface) PutServiceImport(serviceImport *mcsv1a1.ServiceImport) {
 		i.serviceMap[key] = svcInfo
 	}
 
-	if !svcInfo.isHeadless {
-		clusterName := getSourceCluster(serviceImport)
-
-		clusterInfo := svcInfo.ensureClusterInfo(clusterName)
-		clusterInfo.weight = getServiceWeightFrom(serviceImport, i.clusterStatus.GetLocalClusterID())
-		clusterInfo.endpointRecords = []DNSRecord{{
-			IP:          serviceImport.Spec.IPs[0],
-			Ports:       serviceImport.Spec.Ports,
-			ClusterName: clusterName,
-		}}
-
-		svcInfo.resetLoadBalancing()
-		svcInfo.mergePorts()
-
+	if svcInfo.isHeadless || !isLegacy {
 		return
 	}
-}
 
-func (i *Interface) RemoveServiceImport(serviceImport *mcsv1a1.ServiceImport) {
-	logger.Infof("Remove %#v", serviceImport)
+	// This is a legacy pre-0.15 remote cluster ServiceImport - initialize the cluster info to maintain backwards compatibility
+	// while roling upgrade is in progress.
 
-	name, found := getSourceName(serviceImport)
-	if !found {
-		return
-	}
+	clusterName := serviceImport.Labels["lighthouse.submariner.io/sourceCluster"]
 
-	key := keyFunc(getSourceNamespace(serviceImport), name)
+	clusterInfo := svcInfo.ensureClusterInfo(clusterName)
+	clusterInfo.endpointRecords = []DNSRecord{{
+		IP:          serviceImport.Spec.IPs[0],
+		Ports:       serviceImport.Spec.Ports,
+		ClusterName: clusterName,
+	}}
 
-	i.mutex.Lock()
-	defer i.mutex.Unlock()
+	svcInfo.mergePorts()
+	svcInfo.resetLoadBalancing()
+}
 
-	serviceInfo, found := i.serviceMap[key]
-	if !found {
+func (i *Interface) RemoveServiceImport(serviceImport *mcsv1a1.ServiceImport) {
+	if ignoreServiceImport(serviceImport) {
 		return
 	}
 
-	for _, info := range serviceImport.Status.Clusters {
-		delete(serviceInfo.clusters, info.Cluster)
-	}
-
-	if len(serviceInfo.clusters) == 0 {
-		delete(i.serviceMap, key)
+	key, isLegacy := getServiceImportKey(serviceImport)
+	if isLegacy {
 		return
 	}
 
-	if !serviceInfo.isHeadless {
-		serviceInfo.resetLoadBalancing()
-	}
-
-	serviceInfo.mergePorts()
-}
-
-func getSourceName(from *mcsv1a1.ServiceImport) (string, bool) {
-	name, ok := from.Labels[mcsv1a1.LabelServiceName]
-	if ok {
-		return name, true
-	}
-
-	name, ok = from.Annotations["origin-name"]
-	return name, ok
-}
+	logger.Infof("Remove ServiceImport %q", key)
 
-func getSourceNamespace(from *mcsv1a1.ServiceImport) string {
-	ns, ok := from.Labels[constants.LabelSourceNamespace]
-	if ok {
-		return ns
-	}
+	i.mutex.Lock()
+	defer i.mutex.Unlock()
 
-	return from.Annotations["origin-namespace"]
+	delete(i.serviceMap, key)
 }
 
-func getSourceCluster(from *mcsv1a1.ServiceImport) string {
-	c, ok := from.Labels[constants.MCSLabelSourceCluster]
+func getServiceImportKey(from *mcsv1a1.ServiceImport) (string, bool) {
+	name, ok := from.Annotations["origin-name"]
 	if ok {
-		return c
+		return keyFunc(from.Annotations["origin-namespace"], name), true
 	}
 
-	return from.Labels["lighthouse.submariner.io/sourceCluster"]
+	return keyFunc(from.Namespace, from.Name), false
 }
 
-func getServiceWeightFrom(si *mcsv1a1.ServiceImport, forClusterName string) int64 {
-	weightKey := constants.LoadBalancerWeightAnnotationPrefix + "/" + forClusterName
-	if val, ok := si.Annotations[weightKey]; ok {
-		f, err := strconv.ParseInt(val, 0, 64)
-		if err != nil {
-			return f
-		}
-
-		logger.Errorf(err, "Error parsing the %q annotation from ServiceImport %q", weightKey, si.Name)
-	}
+func ignoreServiceImport(serviceImport *mcsv1a1.ServiceImport) bool {
+	_, isLocal := serviceImport.Labels[mcsv1a1.LabelServiceName]
+	_, isOnBroker := serviceImport.Annotations[mcsv1a1.LabelServiceName]
 
-	return 1 // Zero will cause no selection
+	return isLocal || isOnBroker
 }
diff --git a/coredns/resolver/service_info.go b/coredns/resolver/service_info.go
index 42f472a..46511c6 100644
--- a/coredns/resolver/service_info.go
+++ b/coredns/resolver/service_info.go
@@ -56,6 +56,7 @@ func (si *serviceInfo) ensureClusterInfo(name string) *clusterInfo {
 	if !ok {
 		info = &clusterInfo{
 			endpointRecordsByHost: make(map[string][]DNSRecord),
+			weight:                1,
 		}
 
 		si.clusters[name] = info
diff --git a/go.mod b/go.mod
index 4640eae..34537e5 100644
--- a/go.mod
+++ b/go.mod
@@ -8,7 +8,7 @@ require (
 	github.com/onsi/gomega v1.27.4
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.14.0
-	github.com/submariner-io/admiral v0.15.0-m4
+	github.com/submariner-io/admiral v0.15.0-m4.0.20230323103716-69b73fbef249
 	github.com/submariner-io/shipyard v0.15.0-m4
 	github.com/uw-labs/lichen v0.1.7
 	k8s.io/api v0.26.3
diff --git a/go.sum b/go.sum
index 6a6c9a0..c20455b 100644
--- a/go.sum
+++ b/go.sum
@@ -494,8 +494,8 @@ github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81P
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
-github.com/submariner-io/admiral v0.15.0-m4 h1:4sPyxfDBi1bNV8S0vLPYKOVwdOlIBeXxvMjVAALoydY=
-github.com/submariner-io/admiral v0.15.0-m4/go.mod h1:RKemq7RSm8jS78zqcFZ96ibko59yNJ1mZcgbDRPDCzc=
+github.com/submariner-io/admiral v0.15.0-m4.0.20230323103716-69b73fbef249 h1:Ka6SroukPiFmuZ9P0bD/UIVcMzpF0jT2OKZcU0qXJOo=
+github.com/submariner-io/admiral v0.15.0-m4.0.20230323103716-69b73fbef249/go.mod h1:gg5ngIypA/bAzVEOiFynAIDdUAEbrjxdqecgDFgYvMo=
 github.com/submariner-io/shipyard v0.15.0-m4 h1:ji+WLE9yvOqUQJnKYhLm0Ajb5WTobkuxrDHSiIkGiEg=
 github.com/submariner-io/shipyard v0.15.0-m4/go.mod h1:txktXeFOmI8g5wOxmuvNgheyZLyRloXbdo4i5gQwGNY=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
diff --git a/pkg/agent/controller/agent.go b/pkg/agent/controller/agent.go
index 2c1107f..f85fc8c 100644
--- a/pkg/agent/controller/agent.go
+++ b/pkg/agent/controller/agent.go
@@ -20,19 +20,18 @@ package controller
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"reflect"
 
 	"github.com/pkg/errors"
-	"github.com/prometheus/client_golang/prometheus"
+	"github.com/submariner-io/admiral/pkg/federate"
 	"github.com/submariner-io/admiral/pkg/log"
-	"github.com/submariner-io/admiral/pkg/resource"
 	"github.com/submariner-io/admiral/pkg/syncer"
 	"github.com/submariner-io/admiral/pkg/syncer/broker"
 	"github.com/submariner-io/admiral/pkg/util"
 	"github.com/submariner-io/lighthouse/pkg/constants"
 	corev1 "k8s.io/api/core/v1"
-	discovery "k8s.io/api/discovery/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -75,66 +74,26 @@ func New(spec *AgentSpecification, syncerConf broker.SyncerConfig, syncerMetricN
 
 	agentController.serviceExportClient = syncerConf.LocalClient.Resource(*gvr)
 
-	syncerConf.LocalNamespace = spec.Namespace
-	syncerConf.LocalClusterID = spec.ClusterID
-
-	syncerConf.ResourceConfigs = []broker.ResourceConfig{
-		{
-			LocalSourceNamespace:  metav1.NamespaceAll,
-			LocalResourceType:     &mcsv1a1.ServiceImport{},
-			LocalTransform:        agentController.onLocalServiceImport,
-			LocalOnSuccessfulSync: agentController.onSuccessfulServiceImportSync,
-			BrokerResourceType:    &mcsv1a1.ServiceImport{},
-			SyncCounterOpts: &prometheus.GaugeOpts{
-				Name: syncerMetricNames.ServiceImportCounterName,
-				Help: "Count of imported services",
-			},
-		},
-	}
-
-	agentController.serviceImportSyncer, err = broker.NewSyncer(syncerConf)
+	agentController.endpointSliceController, err = newEndpointSliceController(spec, syncerConf, agentController.updateExportedServiceStatus)
 	if err != nil {
-		return nil, errors.Wrap(err, "error creating ServiceImport syncer")
-	}
-
-	syncerConf.LocalNamespace = metav1.NamespaceAll
-	syncerConf.ResourceConfigs = []broker.ResourceConfig{
-		{
-			LocalSourceNamespace: metav1.NamespaceAll,
-			LocalResourceType:    &discovery.EndpointSlice{},
-			LocalTransform:       agentController.filterLocalEndpointSlices,
-			LocalResourcesEquivalent: func(obj1, obj2 *unstructured.Unstructured) bool {
-				return false
-			},
-			BrokerResourceType: &discovery.EndpointSlice{},
-			BrokerResourcesEquivalent: func(obj1, obj2 *unstructured.Unstructured) bool {
-				return false
-			},
-			BrokerTransform: agentController.remoteEndpointSliceToLocal,
-		},
+		return nil, err
 	}
 
-	agentController.endpointSliceSyncer, err = broker.NewSyncer(syncerConf)
-	if err != nil {
-		return nil, errors.Wrap(err, "error creating EndpointSlice syncer")
-	}
+	agentController.localServiceImportFederator = federate.NewCreateOrUpdateFederator(syncerConf.LocalClient, syncerConf.RestMapper,
+		spec.Namespace, "")
 
 	agentController.serviceExportSyncer, err = syncer.NewResourceSyncer(&syncer.ResourceSyncerConfig{
 		Name:            "ServiceExport -> ServiceImport",
 		SourceClient:    syncerConf.LocalClient,
 		SourceNamespace: metav1.NamespaceAll,
 		RestMapper:      syncerConf.RestMapper,
-		Federator:       agentController.serviceImportSyncer.GetLocalFederator(),
+		Federator:       agentController.localServiceImportFederator,
 		ResourceType:    &mcsv1a1.ServiceExport{},
 		Transform:       agentController.serviceExportToServiceImport,
 		ResourcesEquivalent: func(oldObj, newObj *unstructured.Unstructured) bool {
 			return !agentController.shouldProcessServiceExportUpdate(oldObj, newObj)
 		},
 		Scheme: syncerConf.Scheme,
-		SyncCounterOpts: &prometheus.GaugeOpts{
-			Name: syncerMetricNames.ServiceExportCounterName,
-			Help: "Count of exported services",
-		},
 	})
 	if err != nil {
 		return nil, errors.Wrap(err, "error creating ServiceExport syncer")
@@ -145,7 +104,7 @@ func New(spec *AgentSpecification, syncerConf broker.SyncerConfig, syncerMetricN
 		SourceClient:    syncerConf.LocalClient,
 		SourceNamespace: metav1.NamespaceAll,
 		RestMapper:      syncerConf.RestMapper,
-		Federator:       agentController.serviceImportSyncer.GetLocalFederator(),
+		Federator:       agentController.localServiceImportFederator,
 		ResourceType:    &corev1.Service{},
 		Transform:       agentController.serviceToRemoteServiceImport,
 		Scheme:          syncerConf.Scheme,
@@ -154,8 +113,9 @@ func New(spec *AgentSpecification, syncerConf broker.SyncerConfig, syncerMetricN
 		return nil, errors.Wrap(err, "error creating Service syncer")
 	}
 
-	agentController.serviceImportController, err = newServiceImportController(spec, agentController.serviceSyncer,
-		syncerConf.RestMapper, syncerConf.LocalClient, syncerConf.Scheme)
+	agentController.serviceImportController, err = newServiceImportController(spec, syncerMetricNames, syncerConf,
+		agentController.endpointSliceController.syncer.GetBrokerClient(),
+		agentController.endpointSliceController.syncer.GetBrokerNamespace(), agentController.updateExportedServiceStatus)
 	if err != nil {
 		return nil, err
 	}
@@ -177,30 +137,16 @@ func (a *Controller) Start(stopCh <-chan struct{}) error {
 		return errors.Wrap(err, "error starting Service syncer")
 	}
 
-	if err := a.endpointSliceSyncer.Start(stopCh); err != nil {
+	if err := a.endpointSliceController.start(stopCh); err != nil {
 		return errors.Wrap(err, "error starting EndpointSlice syncer")
 	}
 
-	if err := a.serviceImportSyncer.Start(stopCh); err != nil {
-		return errors.Wrap(err, "error starting ServiceImport syncer")
-	}
-
 	if err := a.serviceImportController.start(stopCh); err != nil {
 		return errors.Wrap(err, "error starting ServiceImport controller")
 	}
 
-	// This function also checks the legacy source name label for migration - this can be removed after 0.15.
-	serviceImportSourceName := func(serviceImport *mcsv1a1.ServiceImport) string {
-		name, ok := serviceImport.Labels[mcsv1a1.LabelServiceName]
-		if ok {
-			return name
-		}
-
-		return serviceImport.GetLabels()["lighthouse.submariner.io/sourceName"]
-	}
-
 	a.serviceExportSyncer.Reconcile(func() []runtime.Object {
-		return a.serviceImportLister(func(si *mcsv1a1.ServiceImport) runtime.Object {
+		return a.serviceImportController.localServiceImportLister(func(si *mcsv1a1.ServiceImport) runtime.Object {
 			return &mcsv1a1.ServiceExport{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      serviceImportSourceName(si),
@@ -211,7 +157,7 @@ func (a *Controller) Start(stopCh <-chan struct{}) error {
 	})
 
 	a.serviceSyncer.Reconcile(func() []runtime.Object {
-		return a.serviceImportLister(func(si *mcsv1a1.ServiceImport) runtime.Object {
+		return a.serviceImportController.localServiceImportLister(func(si *mcsv1a1.ServiceImport) runtime.Object {
 			return &corev1.Service{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      serviceImportSourceName(si),
@@ -226,24 +172,6 @@ func (a *Controller) Start(stopCh <-chan struct{}) error {
 	return nil
 }
 
-func (a *Controller) serviceImportLister(transform func(si *mcsv1a1.ServiceImport) runtime.Object) []runtime.Object {
-	siList, err := a.serviceImportSyncer.ListLocalResources(&mcsv1a1.ServiceImport{})
-	if err != nil {
-		logger.Error(err, "Error listing serviceImports")
-		return nil
-	}
-
-	retList := make([]runtime.Object, 0, len(siList))
-
-	for _, obj := range siList {
-		si := obj.(*mcsv1a1.ServiceImport)
-
-		retList = append(retList, transform(si))
-	}
-
-	return retList
-}
-
 func (a *Controller) serviceExportToServiceImport(obj runtime.Object, numRequeues int, op syncer.Operation) (runtime.Object, bool) {
 	svcExport := obj.(*mcsv1a1.ServiceExport)
 
@@ -256,8 +184,8 @@ func (a *Controller) serviceExportToServiceImport(obj runtime.Object, numRequeue
 	obj, found, err := a.serviceSyncer.GetResource(svcExport.Name, svcExport.Namespace)
 	if err != nil {
 		// some other error. Log and requeue
-		a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid, corev1.ConditionUnknown,
-			"ServiceRetrievalFailed", fmt.Sprintf("Error retrieving the Service: %v", err))
+		a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, newServiceExportCondition(mcsv1a1.ServiceExportValid,
+			corev1.ConditionUnknown, "ServiceRetrievalFailed", fmt.Sprintf("Error retrieving the Service: %v", err)))
 		logger.Errorf(err, "Error retrieving Service %s/%s", svcExport.Namespace, svcExport.Name)
 
 		return nil, true
@@ -265,8 +193,8 @@ func (a *Controller) serviceExportToServiceImport(obj runtime.Object, numRequeue
 
 	if !found {
 		logger.V(log.DEBUG).Infof("Service to be exported (%s/%s) doesn't exist", svcExport.Namespace, svcExport.Name)
-		a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid, corev1.ConditionFalse,
-			serviceUnavailable, "Service to be exported doesn't exist")
+		a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, newServiceExportCondition(mcsv1a1.ServiceExportValid,
+			corev1.ConditionFalse, serviceUnavailable, "Service to be exported doesn't exist"))
 
 		return nil, false
 	}
@@ -276,11 +204,11 @@ func (a *Controller) serviceExportToServiceImport(obj runtime.Object, numRequeue
 	svcType, ok := getServiceImportType(svc)
 
 	if !ok {
-		a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid, corev1.ConditionFalse,
-			invalidServiceType, fmt.Sprintf("Service of type %v not supported", svc.Spec.Type))
+		a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, newServiceExportCondition(mcsv1a1.ServiceExportValid,
+			corev1.ConditionFalse, invalidServiceType, fmt.Sprintf("Service of type %v not supported", svc.Spec.Type)))
 		logger.Errorf(nil, "Service type %q not supported for Service (%s/%s)", svc.Spec.Type, svcExport.Namespace, svcExport.Name)
 
-		err = a.serviceImportSyncer.GetLocalFederator().Delete(a.newServiceImport(svcExport.Name, svcExport.Namespace))
+		err = a.localServiceImportFederator.Delete(a.newServiceImport(svcExport.Name, svcExport.Namespace))
 		if err == nil || apierrors.IsNotFound(err) {
 			return nil, false
 		}
@@ -313,8 +241,8 @@ func (a *Controller) serviceExportToServiceImport(obj runtime.Object, numRequeue
 				logger.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)
+				a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, newServiceExportCondition(mcsv1a1.ServiceExportValid,
+					corev1.ConditionFalse, reason, msg))
 
 				return nil, true
 			}
@@ -327,9 +255,10 @@ func (a *Controller) serviceExportToServiceImport(obj runtime.Object, numRequeue
 		serviceImport.Spec.Ports = a.getPortsForService(svc)
 	}
 
-	a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid, corev1.ConditionTrue, "", "")
+	a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, newServiceExportCondition(mcsv1a1.ServiceExportValid,
+		corev1.ConditionTrue, "", ""))
 
-	logger.V(log.DEBUG).Infof("Returning ServiceImport : %#v", serviceImport)
+	logger.V(log.DEBUG).Infof("Returning ServiceImport: %s", serviceImportStringer{serviceImport})
 
 	return serviceImport, false
 }
@@ -346,45 +275,6 @@ func getServiceImportType(service *corev1.Service) (mcsv1a1.ServiceImportType, b
 	return mcsv1a1.ClusterSetIP, true
 }
 
-func (a *Controller) onLocalServiceImport(obj runtime.Object, _ int, op syncer.Operation) (runtime.Object, bool) {
-	serviceImport := obj.(*mcsv1a1.ServiceImport)
-
-	if op == syncer.Delete {
-		a.updateExportedServiceStatus(serviceImport.Labels[mcsv1a1.LabelServiceName],
-			serviceImport.Labels[constants.LabelSourceNamespace], constants.ServiceExportSynced, corev1.ConditionFalse,
-			"NoServiceImport", "ServiceImport was deleted")
-
-		return obj, false
-	}
-
-	serviceName, ok := serviceImport.Labels[mcsv1a1.LabelServiceName]
-	if !ok {
-		// The label is missing - most likely b/c the ServiceImport hasn't yet been migrated from the legacy labels.
-		logger.Infof("Label %q missing from ServiceImport (%s/%s) - not syncing", mcsv1a1.LabelServiceName,
-			serviceImport.Namespace, serviceImport.Name)
-
-		return nil, false
-	}
-
-	a.updateExportedServiceStatus(serviceName,
-		serviceImport.Labels[constants.LabelSourceNamespace], constants.ServiceExportSynced, corev1.ConditionFalse,
-		"AwaitingSync", fmt.Sprintf("ServiceImport %sd - awaiting sync to the broker", op))
-
-	return obj, false
-}
-
-func (a *Controller) onSuccessfulServiceImportSync(synced runtime.Object, op syncer.Operation) {
-	if op == syncer.Delete {
-		return
-	}
-
-	serviceImport := synced.(*mcsv1a1.ServiceImport)
-
-	a.updateExportedServiceStatus(serviceImport.Labels[mcsv1a1.LabelServiceName],
-		serviceImport.Labels[constants.LabelSourceNamespace], constants.ServiceExportSynced, corev1.ConditionTrue, "",
-		"ServiceImport was successfully synced to the broker")
-}
-
 func (a *Controller) shouldProcessServiceExportUpdate(oldObj, newObj *unstructured.Unstructured) bool {
 	oldValidCond := FindServiceExportStatusCondition(a.toServiceExport(oldObj).Status.Conditions, mcsv1a1.ServiceExportValid)
 	newValidCond := FindServiceExportStatusCondition(a.toServiceExport(newObj).Status.Conditions, mcsv1a1.ServiceExportValid)
@@ -432,80 +322,57 @@ func (a *Controller) serviceToRemoteServiceImport(obj runtime.Object, numRequeue
 	serviceImport := a.newServiceImport(svcExport.Name, svcExport.Namespace)
 
 	// Update the status and requeue
-	a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid, corev1.ConditionFalse,
-		serviceUnavailable, "Service to be exported doesn't exist")
+	a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, newServiceExportCondition(mcsv1a1.ServiceExportValid,
+		corev1.ConditionFalse, serviceUnavailable, "Service to be exported doesn't exist"))
 
 	return serviceImport, false
 }
 
-func (a *Controller) updateExportedServiceStatus(name, namespace string, condType mcsv1a1.ServiceExportConditionType,
-	status corev1.ConditionStatus, reason, msg string,
-) {
-	logger.V(log.DEBUG).Infof("updateExportedServiceStatus for (%s/%s) - Type: %q, Status: %q, Reason: %q, Message: %q",
-		namespace, name, condType, status, reason, msg)
-
-	retryErr := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		toUpdate, err := a.getServiceExport(name, namespace)
+func (a *Controller) updateExportedServiceStatus(name, namespace string, conditions ...mcsv1a1.ServiceExportCondition) {
+	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		obj, err := a.serviceExportClient.Namespace(namespace).Get(context.TODO(), name, metav1.GetOptions{})
 		if apierrors.IsNotFound(err) {
 			logger.Infof("ServiceExport (%s/%s) not found - unable to update status", namespace, name)
 			return nil
 		} else if err != nil {
-			return err
+			return errors.Wrap(err, "error retrieving ServiceExport")
 		}
 
-		now := metav1.Now()
-		newCondition := mcsv1a1.ServiceExportCondition{
-			Type:               condType,
-			Status:             status,
-			LastTransitionTime: &now,
-			Reason:             &reason,
-			Message:            &msg,
-		}
+		toUpdate := a.toServiceExport(obj)
 
-		prevCond := FindServiceExportStatusCondition(toUpdate.Status.Conditions, condType)
-		if prevCond == nil {
-			toUpdate.Status.Conditions = append(toUpdate.Status.Conditions, newCondition)
-		} else if serviceExportConditionEqual(prevCond, &newCondition) {
-			logger.V(log.TRACE).Infof("Last ServiceExportCondition for (%s/%s) is equal - not updating status: %#v",
-				namespace, name, toUpdate.Status.Conditions[0])
-			return nil
-		} else {
-			*prevCond = newCondition
+		updated := false
+
+		for i := range conditions {
+			condition := &conditions[i]
+
+			logger.V(log.DEBUG).Infof("updateExportedServiceStatus for (%s/%s) - Type: %q, Status: %q, Reason: %q, Message: %q",
+				namespace, name, condition.Type, condition.Status, *condition.Reason, *condition.Message)
+
+			prevCond := FindServiceExportStatusCondition(toUpdate.Status.Conditions, condition.Type)
+			if prevCond == nil {
+				toUpdate.Status.Conditions = append(toUpdate.Status.Conditions, *condition)
+				updated = true
+			} else if serviceExportConditionEqual(prevCond, condition) {
+				logger.V(log.TRACE).Infof("Last ServiceExportCondition for (%s/%s) is equal - not updating status: %#v",
+					namespace, name, prevCond)
+			} else {
+				*prevCond = *condition
+				updated = true
+			}
 		}
 
-		raw, err := resource.ToUnstructured(toUpdate)
-		if err != nil {
-			return errors.Wrap(err, "error converting resource")
+		if !updated {
+			return nil
 		}
 
-		_, err = a.serviceExportClient.Namespace(toUpdate.Namespace).UpdateStatus(context.TODO(), raw, metav1.UpdateOptions{})
+		_, err = a.serviceExportClient.Namespace(toUpdate.Namespace).UpdateStatus(context.TODO(),
+			a.serviceImportController.converter.toUnstructured(toUpdate), metav1.UpdateOptions{})
 
 		return errors.Wrap(err, "error from UpdateStatus")
 	})
-
-	if retryErr != nil {
-		logger.Errorf(retryErr, "Error updating status for ServiceExport (%s/%s)", namespace, name)
-	}
-}
-
-func (a *Controller) getServiceExport(name, namespace string) (*mcsv1a1.ServiceExport, error) {
-	obj, err := a.serviceExportClient.Namespace(namespace).Get(context.TODO(), name, metav1.GetOptions{})
-	if err != nil {
-		return nil, errors.Wrap(err, "error retrieving ServiceExport")
-	}
-
-	return a.toServiceExport(obj), nil
-}
-
-func (a *Controller) toServiceExport(obj *unstructured.Unstructured) *mcsv1a1.ServiceExport {
-	se := &mcsv1a1.ServiceExport{}
-
-	err := a.serviceImportController.scheme.Convert(obj, se, nil)
 	if err != nil {
-		panic(errors.WithMessagef(err, "Error converting %#v to ServiceExport", obj))
+		logger.Errorf(err, "Error updating status for ServiceExport (%s/%s)", namespace, name)
 	}
-
-	return se
 }
 
 func serviceExportConditionEqual(c1, c2 *mcsv1a1.ServiceExportCondition) bool {
@@ -545,37 +412,6 @@ func (a *Controller) getObjectNameWithClusterID(name, namespace string) string {
 	return name + "-" + namespace + "-" + a.clusterID
 }
 
-func (a *Controller) remoteEndpointSliceToLocal(obj runtime.Object, numRequeues int, op syncer.Operation) (runtime.Object, bool) {
-	endpointSlice := obj.(*discovery.EndpointSlice)
-	endpointSlice.Namespace = endpointSlice.GetObjectMeta().GetLabels()[constants.LabelSourceNamespace]
-
-	return endpointSlice, false
-}
-
-func (a *Controller) filterLocalEndpointSlices(obj runtime.Object, numRequeues int, op syncer.Operation) (runtime.Object, bool) {
-	endpointSlice := obj.(*discovery.EndpointSlice)
-	labels := endpointSlice.GetObjectMeta().GetLabels()
-
-	if labels[discovery.LabelManagedBy] != constants.LabelValueManagedBy {
-		return nil, false
-	}
-
-	oldName := labels[mcsv1a1.LabelServiceName] + "-" + labels[constants.MCSLabelSourceCluster]
-	if op != syncer.Delete && endpointSlice.Name == oldName {
-		logger.Infof("EndpointSlice %s/%s has the old naming convention sans namespace - deleting it",
-			endpointSlice.Namespace, endpointSlice.Name)
-
-		err := a.endpointSliceSyncer.GetLocalFederator().Delete(endpointSlice)
-		if err != nil {
-			logger.Errorf(err, "Error deleting local EndpointSlice %s/%s", endpointSlice.Namespace, endpointSlice.Name)
-		}
-
-		return nil, false
-	}
-
-	return obj, false
-}
-
 func (a *Controller) getGlobalIP(service *corev1.Service) (ip, reason, msg string) {
 	if a.globalnetEnabled {
 		ingressIP, found := a.getIngressIP(service.Name, service.Namespace)
@@ -597,3 +433,61 @@ func (a *Controller) getIngressIP(name, namespace string) (*IngressIP, bool) {
 
 	return parseIngressIP(obj), true
 }
+
+func (a *Controller) toServiceExport(obj runtime.Object) *mcsv1a1.ServiceExport {
+	return a.serviceImportController.converter.toServiceExport(obj)
+}
+
+func newServiceExportCondition(condType mcsv1a1.ServiceExportConditionType, status corev1.ConditionStatus,
+	reason, msg string,
+) mcsv1a1.ServiceExportCondition {
+	now := metav1.Now()
+
+	return mcsv1a1.ServiceExportCondition{
+		Type:               condType,
+		Status:             status,
+		LastTransitionTime: &now,
+		Reason:             &reason,
+		Message:            &msg,
+	}
+}
+
+// This function also checks the legacy source name label for migration - this can be removed after 0.15.
+func serviceImportSourceName(serviceImport *mcsv1a1.ServiceImport) string {
+	name, ok := serviceImport.Labels[mcsv1a1.LabelServiceName]
+	if ok {
+		return name
+	}
+
+	return serviceImport.GetLabels()["lighthouse.submariner.io/sourceName"]
+}
+
+func (c converter) toServiceImport(obj runtime.Object) *mcsv1a1.ServiceImport {
+	to := &mcsv1a1.ServiceImport{}
+	utilruntime.Must(c.scheme.Convert(obj, to, nil))
+
+	return to
+}
+
+func (c converter) toUnstructured(obj runtime.Object) *unstructured.Unstructured {
+	to := &unstructured.Unstructured{}
+	utilruntime.Must(c.scheme.Convert(obj, to, nil))
+
+	return to
+}
+
+func (c converter) toServiceExport(obj runtime.Object) *mcsv1a1.ServiceExport {
+	to := &mcsv1a1.ServiceExport{}
+	utilruntime.Must(c.scheme.Convert(obj, to, nil))
+
+	return to
+}
+
+type serviceImportStringer struct {
+	*mcsv1a1.ServiceImport
+}
+
+func (s serviceImportStringer) String() string {
+	spec, _ := json.MarshalIndent(&s.Spec, "", "  ")
+	return "spec: " + string(spec)
+}
diff --git a/pkg/agent/controller/cleanup.go b/pkg/agent/controller/cleanup.go
index 24af273..6ece417 100644
--- a/pkg/agent/controller/cleanup.go
+++ b/pkg/agent/controller/cleanup.go
@@ -26,6 +26,7 @@ import (
 	discovery "k8s.io/api/discovery/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	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/schema"
@@ -50,28 +51,39 @@ var (
 func (a *Controller) Cleanup() error {
 	// Delete all ServiceImports from the local cluster skipping those in the broker namespace if the broker is on the
 	// local cluster.
-	err := deleteResources(a.serviceImportSyncer.GetLocalClient().Resource(serviceImportGVR), metav1.NamespaceAll,
+	siClient := a.serviceImportController.localClient.Resource(serviceImportGVR)
+
+	list, err := listResources(siClient, metav1.NamespaceAll,
 		&metav1.ListOptions{
-			FieldSelector: fields.OneTermNotEqualSelector("metadata.namespace", a.serviceImportSyncer.GetBrokerNamespace()).String(),
+			FieldSelector: fields.OneTermNotEqualSelector("metadata.namespace",
+				a.serviceImportController.serviceImportAggregator.brokerNamespace).String(),
 		})
 	if err != nil {
-		return errors.Wrap(err, "error deleting local ServiceImports")
+		return errors.Wrap(err, "error listing local ServiceImports")
 	}
 
-	// Delete all local ServiceImports from the broker.
-	err = deleteResources(a.serviceImportSyncer.GetBrokerClient().Resource(serviceImportGVR), a.serviceImportSyncer.GetBrokerNamespace(),
-		&metav1.ListOptions{
-			LabelSelector: labels.Set(map[string]string{constants.MCSLabelSourceCluster: a.clusterID}).String(),
-		})
-	if err != nil {
-		return errors.Wrap(err, "error deleting remote ServiceImports")
+	for i := range list {
+		_, ok := list[i].GetLabels()[mcsv1a1.LabelServiceName]
+		if ok {
+			err = a.serviceImportController.Delete(&list[i])
+			if err != nil && !apierrors.IsNotFound(err) {
+				return err
+			}
+		}
+
+		err = siClient.Namespace(list[i].GetNamespace()).Delete(context.TODO(), list[i].GetName(), metav1.DeleteOptions{})
+
+		if err != nil && !apierrors.IsNotFound(err) {
+			return err //nolint:wrapcheck // Let the caller wrap
+		}
 	}
 
 	// Delete all EndpointSlices from the local cluster skipping those in the broker namespace if the broker is on the
 	// local cluster.
-	err = deleteResources(a.endpointSliceSyncer.GetLocalClient().Resource(endpointSliceGVR), metav1.NamespaceAll,
+	err = deleteResources(a.endpointSliceController.syncer.GetLocalClient().Resource(endpointSliceGVR), metav1.NamespaceAll,
 		&metav1.ListOptions{
-			FieldSelector: fields.OneTermNotEqualSelector("metadata.namespace", a.serviceImportSyncer.GetBrokerNamespace()).String(),
+			FieldSelector: fields.OneTermNotEqualSelector("metadata.namespace",
+				a.serviceImportController.serviceImportAggregator.brokerNamespace).String(),
 			LabelSelector: labels.Set(map[string]string{discovery.LabelManagedBy: constants.LabelValueManagedBy}).String(),
 		})
 	if err != nil {
@@ -79,7 +91,8 @@ func (a *Controller) Cleanup() error {
 	}
 
 	// Delete all local EndpointSlices from the broker.
-	err = deleteResources(a.endpointSliceSyncer.GetBrokerClient().Resource(endpointSliceGVR), a.endpointSliceSyncer.GetBrokerNamespace(),
+	err = deleteResources(a.endpointSliceController.syncer.GetBrokerClient().Resource(endpointSliceGVR),
+		a.endpointSliceController.syncer.GetBrokerNamespace(),
 		&metav1.ListOptions{
 			LabelSelector: labels.Set(map[string]string{constants.MCSLabelSourceCluster: a.clusterID}).String(),
 		})
@@ -88,13 +101,13 @@ func (a *Controller) Cleanup() error {
 }
 
 func deleteResources(client dynamic.NamespaceableResourceInterface, ns string, options *metav1.ListOptions) error {
-	list, err := client.Namespace(ns).List(context.TODO(), *options)
-	if err != nil && !apierrors.IsNotFound(err) {
-		return err //nolint:wrapcheck // Let the caller wrap
+	list, err := listResources(client, ns, options)
+	if err != nil {
+		return err
 	}
 
-	for i := range list.Items {
-		err = client.Namespace(list.Items[i].GetNamespace()).Delete(context.TODO(), list.Items[i].GetName(), metav1.DeleteOptions{})
+	for i := range list {
+		err = client.Namespace(list[i].GetNamespace()).Delete(context.TODO(), list[i].GetName(), metav1.DeleteOptions{})
 		if err != nil && !apierrors.IsNotFound(err) {
 			return err //nolint:wrapcheck // Let the caller wrap
 		}
@@ -102,3 +115,14 @@ func deleteResources(client dynamic.NamespaceableResourceInterface, ns string, o
 
 	return nil
 }
+
+func listResources(client dynamic.NamespaceableResourceInterface, ns string,
+	options *metav1.ListOptions,
+) ([]unstructured.Unstructured, error) {
+	list, err := client.Namespace(ns).List(context.TODO(), *options)
+	if err != nil && !apierrors.IsNotFound(err) {
+		return nil, err //nolint:wrapcheck // Let the caller wrap
+	}
+
+	return list.Items, nil
+}
diff --git a/pkg/agent/controller/endpoint.go b/pkg/agent/controller/endpoint.go
index 6f5e755..2c1c6d3 100644
--- a/pkg/agent/controller/endpoint.go
+++ b/pkg/agent/controller/endpoint.go
@@ -20,11 +20,14 @@ package controller
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
+	"strconv"
 
 	"github.com/pkg/errors"
+	"github.com/submariner-io/admiral/pkg/federate"
 	"github.com/submariner-io/admiral/pkg/log"
 	"github.com/submariner-io/admiral/pkg/syncer"
-	"github.com/submariner-io/admiral/pkg/syncer/broker"
 	"github.com/submariner-io/lighthouse/pkg/constants"
 	corev1 "k8s.io/api/core/v1"
 	discovery "k8s.io/api/discovery/v1"
@@ -38,27 +41,29 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/dynamic"
 	utilnet "k8s.io/utils/net"
+	"k8s.io/utils/pointer"
 	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
 func startEndpointController(localClient dynamic.Interface, restMapper meta.RESTMapper, scheme *runtime.Scheme,
-	serviceImport *mcsv1a1.ServiceImport, serviceImportNameSpace, serviceName, clusterID string,
-	globalIngressIPCache *globalIngressIPCache,
+	serviceImport *mcsv1a1.ServiceImport, clusterID string, globalIngressIPCache *globalIngressIPCache,
 ) (*EndpointController, error) {
-	logger.V(log.DEBUG).Infof("Starting Endpoints controller for service %s/%s", serviceImportNameSpace, serviceName)
+	serviceNamespace := serviceImport.Labels[constants.LabelSourceNamespace]
+	serviceName := serviceImport.Labels[mcsv1a1.LabelServiceName]
+
+	logger.V(log.DEBUG).Infof("Starting Endpoints controller for service %s/%s", serviceNamespace, serviceName)
 
 	globalIngressIPGVR, _ := schema.ParseResourceArg("globalingressips.v1.submariner.io")
 
 	controller := &EndpointController{
-		clusterID:                    clusterID,
-		serviceImportName:            serviceImport.Name,
-		serviceImportSourceNameSpace: serviceImportNameSpace,
-		serviceName:                  serviceName,
-		stopCh:                       make(chan struct{}),
-		isHeadless:                   serviceImport.Spec.Type == mcsv1a1.Headless,
-		globalIngressIPCache:         globalIngressIPCache,
-		localClient:                  localClient,
-		ingressIPClient:              localClient.Resource(*globalIngressIPGVR),
+		clusterID:            clusterID,
+		serviceNamespace:     serviceNamespace,
+		serviceName:          serviceName,
+		serviceImportSpec:    &serviceImport.Spec,
+		stopCh:               make(chan struct{}),
+		globalIngressIPCache: globalIngressIPCache,
+		localClient:          localClient,
+		ingressIPClient:      localClient.Resource(*globalIngressIPGVR),
 	}
 
 	nameSelector := fields.OneTermEqualSelector("metadata.name", serviceName)
@@ -66,11 +71,10 @@ func startEndpointController(localClient dynamic.Interface, restMapper meta.REST
 	epsSyncer, err := syncer.NewResourceSyncer(&syncer.ResourceSyncerConfig{
 		Name:                "Endpoints -> EndpointSlice",
 		SourceClient:        localClient,
-		SourceNamespace:     serviceImportNameSpace,
+		SourceNamespace:     serviceNamespace,
 		SourceFieldSelector: nameSelector.String(),
-		Direction:           syncer.LocalToRemote,
 		RestMapper:          restMapper,
-		Federator:           broker.NewFederator(localClient, restMapper, serviceImportNameSpace, "", "ownerReferences"),
+		Federator:           federate.NewCreateOrUpdateFederator(localClient, restMapper, serviceNamespace, "", "ownerReferences"),
 		ResourceType:        &corev1.Endpoints{},
 		Transform:           controller.endpointsToEndpointSlice,
 		Scheme:              scheme,
@@ -89,29 +93,43 @@ func startEndpointController(localClient dynamic.Interface, restMapper meta.REST
 func (e *EndpointController) stop() {
 	e.stopOnce.Do(func() {
 		close(e.stopCh)
-		e.cleanup()
 	})
 }
 
-func (e *EndpointController) cleanup() {
+func (e *EndpointController) cleanup() (bool, error) {
 	resourceClient := e.localClient.Resource(schema.GroupVersionResource{
 		Group:    discovery.SchemeGroupVersion.Group,
 		Version:  discovery.SchemeGroupVersion.Version,
 		Resource: "endpointslices",
-	}).Namespace(e.serviceImportSourceNameSpace)
+	}).Namespace(e.serviceNamespace)
 
-	// MCS-compliant labels
-	err := resourceClient.DeleteCollection(context.TODO(), metav1.DeleteOptions{}, metav1.ListOptions{
+	listOptions := metav1.ListOptions{
 		LabelSelector: labels.SelectorFromSet(map[string]string{
-			constants.LabelSourceNamespace:  e.serviceImportSourceNameSpace,
+			discovery.LabelManagedBy:        constants.LabelValueManagedBy,
+			constants.LabelSourceNamespace:  e.serviceNamespace,
 			constants.MCSLabelSourceCluster: e.clusterID,
 			mcsv1a1.LabelServiceName:        e.serviceName,
 		}).String(),
-	})
+	}
+
+	list, err := resourceClient.List(context.Background(), listOptions)
+	if err != nil {
+		return false, errors.Wrapf(err, "error listing the EndpointSlices associated with service %s/%s",
+			e.serviceNamespace, e.serviceName)
+	}
+
+	if len(list.Items) == 0 {
+		return false, nil
+	}
+
+	err = resourceClient.DeleteCollection(context.Background(), metav1.DeleteOptions{}, listOptions)
 
 	if err != nil && !apierrors.IsNotFound(err) {
-		logger.Errorf(err, "Error deleting the EndpointSlices associated with ServiceImport %q", e.serviceImportName)
+		return false, errors.Wrapf(err, "error deleting the EndpointSlices associated with service %s/%s",
+			e.serviceNamespace, e.serviceName)
 	}
+
+	return true, nil
 }
 
 func (e *EndpointController) endpointsToEndpointSlice(obj runtime.Object, numRequeues int, op syncer.Operation) (runtime.Object, bool) {
@@ -140,26 +158,31 @@ func (e *EndpointController) endpointsToEndpointSlice(obj runtime.Object, numReq
 func (e *EndpointController) endpointSliceFromEndpoints(endpoints *corev1.Endpoints, op syncer.Operation) (
 	runtime.Object, bool,
 ) {
-	endpointSlice := &discovery.EndpointSlice{}
-
-	endpointSlice.Name = e.endpointSliceNameFrom(endpoints)
-	endpointSlice.Labels = map[string]string{
-		discovery.LabelManagedBy:        constants.LabelValueManagedBy,
-		constants.LabelSourceNamespace:  e.serviceImportSourceNameSpace,
-		constants.MCSLabelSourceCluster: e.clusterID,
-		mcsv1a1.LabelServiceName:        e.serviceName,
+	endpointSlice := &discovery.EndpointSlice{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: e.endpointSliceNameFrom(endpoints),
+			Labels: map[string]string{
+				discovery.LabelManagedBy:        constants.LabelValueManagedBy,
+				constants.LabelSourceNamespace:  e.serviceNamespace,
+				constants.MCSLabelSourceCluster: e.clusterID,
+				mcsv1a1.LabelServiceName:        e.serviceName,
+				constants.LabelIsHeadless:       strconv.FormatBool(e.isHeadless()),
+			},
+		},
+		AddressType: discovery.AddressTypeIPv4,
 	}
 
-	endpointSlice.AddressType = discovery.AddressTypeIPv4
-
 	if len(endpoints.Subsets) > 0 {
 		subset := endpoints.Subsets[0]
-		for i := range subset.Ports {
-			endpointSlice.Ports = append(endpointSlice.Ports, discovery.EndpointPort{
-				Port:     &subset.Ports[i].Port,
-				Name:     &subset.Ports[i].Name,
-				Protocol: &subset.Ports[i].Protocol,
-			})
+
+		if e.isHeadless() {
+			for i := range subset.Ports {
+				endpointSlice.Ports = append(endpointSlice.Ports, discovery.EndpointPort{
+					Port:     &subset.Ports[i].Port,
+					Name:     &subset.Ports[i].Name,
+					Protocol: &subset.Ports[i].Protocol,
+				})
+			}
 		}
 
 		if allAddressesIPv6(subset.Addresses) {
@@ -174,10 +197,27 @@ func (e *EndpointController) endpointSliceFromEndpoints(endpoints *corev1.Endpoi
 		endpointSlice.Endpoints = append(endpointSlice.Endpoints, newEndpoints...)
 	}
 
+	if !e.isHeadless() {
+		endpointSlice.Endpoints = []discovery.Endpoint{{
+			Addresses: []string{e.serviceImportSpec.IPs[0]},
+			Conditions: discovery.EndpointConditions{
+				Ready: pointer.Bool(len(endpointSlice.Endpoints) > 0),
+			},
+		}}
+
+		for i := range e.serviceImportSpec.Ports {
+			endpointSlice.Ports = append(endpointSlice.Ports, discovery.EndpointPort{
+				Port:     &e.serviceImportSpec.Ports[i].Port,
+				Name:     &e.serviceImportSpec.Ports[i].Name,
+				Protocol: &e.serviceImportSpec.Ports[i].Protocol,
+			})
+		}
+	}
+
 	if op == syncer.Create {
-		logger.V(log.DEBUG).Infof("Returning EndpointSlice: %#v", endpointSlice)
+		logger.V(log.DEBUG).Infof("Returning EndpointSlice: %s", endpointSliceStringer{endpointSlice})
 	} else {
-		logger.V(log.TRACE).Infof("Returning EndpointSlice: %#v", endpointSlice)
+		logger.V(log.TRACE).Infof("Returning EndpointSlice: %s", endpointSliceStringer{endpointSlice})
 	}
 
 	return endpointSlice, false
@@ -253,8 +293,8 @@ func allAddressesIPv6(addresses []corev1.EndpointAddress) bool {
 }
 
 func (e *EndpointController) getIP(address *corev1.EndpointAddress) string {
-	if e.isHeadless && e.globalIngressIPCache != nil {
-		obj, found := e.globalIngressIPCache.getForPod(e.serviceImportSourceNameSpace, address.TargetRef.Name)
+	if e.isHeadless() && e.globalIngressIPCache != nil {
+		obj, found := e.globalIngressIPCache.getForPod(e.serviceNamespace, address.TargetRef.Name)
 
 		var ip string
 		if found {
@@ -270,3 +310,18 @@ func (e *EndpointController) getIP(address *corev1.EndpointAddress) string {
 
 	return address.IP
 }
+
+func (e *EndpointController) isHeadless() bool {
+	return e.serviceImportSpec.Type == mcsv1a1.Headless
+}
+
+type endpointSliceStringer struct {
+	*discovery.EndpointSlice
+}
+
+func (s endpointSliceStringer) String() string {
+	ports, _ := json.MarshalIndent(&s.Ports, "", "  ")
+	endpoints, _ := json.MarshalIndent(&s.Endpoints, "", "  ")
+
+	return fmt.Sprintf("endpoints: %s\nports: %s", endpoints, ports)
+}
diff --git a/pkg/agent/controller/endpoint_slice.go b/pkg/agent/controller/endpoint_slice.go
new file mode 100644
index 0000000..1eccdd9
--- /dev/null
+++ b/pkg/agent/controller/endpoint_slice.go
@@ -0,0 +1,138 @@
+/*
+SPDX-License-Identifier: Apache-2.0
+
+Copyright Contributors to the Submariner project.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"fmt"
+
+	"github.com/pkg/errors"
+	"github.com/submariner-io/admiral/pkg/log"
+	"github.com/submariner-io/admiral/pkg/syncer"
+	"github.com/submariner-io/admiral/pkg/syncer/broker"
+	"github.com/submariner-io/lighthouse/pkg/constants"
+	corev1 "k8s.io/api/core/v1"
+	discovery "k8s.io/api/discovery/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8slabels "k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
+	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
+)
+
+//nolint:gocritic // (hugeParam) This function modifies syncerConf so we don't want to pass by pointer.
+func newEndpointSliceController(spec *AgentSpecification, syncerConfig broker.SyncerConfig,
+	updateExportedServiceStatus updateExportedServiceStatusFn,
+) (*EndpointSliceController, error) {
+	c := &EndpointSliceController{
+		clusterID:                   spec.ClusterID,
+		updateExportedServiceStatus: updateExportedServiceStatus,
+	}
+
+	syncerConfig.LocalNamespace = metav1.NamespaceAll
+	syncerConfig.LocalClusterID = spec.ClusterID
+	syncerConfig.ResourceConfigs = []broker.ResourceConfig{
+		{
+			LocalSourceNamespace: metav1.NamespaceAll,
+			LocalSourceLabelSelector: k8slabels.SelectorFromSet(map[string]string{
+				discovery.LabelManagedBy: constants.LabelValueManagedBy,
+			}).String(),
+			LocalResourceType:     &discovery.EndpointSlice{},
+			LocalTransform:        c.onLocalEndpointSlice,
+			LocalOnSuccessfulSync: c.onLocalEndpointSliceSynced,
+			BrokerResourceType:    &discovery.EndpointSlice{},
+			BrokerTransform:       c.onRemoteEndpointSlice,
+		},
+	}
+
+	var err error
+
+	c.syncer, err = broker.NewSyncer(syncerConfig)
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating EndpointSlice syncer")
+	}
+
+	c.serviceImportAggregator = newServiceImportAggregator(c.syncer.GetBrokerClient(), c.syncer.GetBrokerNamespace(),
+		spec.ClusterID, syncerConfig.Scheme)
+
+	return c, nil
+}
+
+func (c *EndpointSliceController) start(stopCh <-chan struct{}) error {
+	if err := c.syncer.Start(stopCh); err != nil {
+		return errors.Wrap(err, "error starting EndpointSlice syncer")
+	}
+
+	return nil
+}
+
+func (c *EndpointSliceController) onLocalEndpointSlice(obj runtime.Object, _ int, op syncer.Operation) (runtime.Object, bool) {
+	endpointSlice := obj.(*discovery.EndpointSlice)
+	labels := endpointSlice.GetObjectMeta().GetLabels()
+
+	oldName := labels[mcsv1a1.LabelServiceName] + "-" + labels[constants.MCSLabelSourceCluster]
+	if op != syncer.Delete && endpointSlice.Name == oldName {
+		logger.Infof("EndpointSlice %s/%s has the old naming convention sans namespace - deleting it",
+			endpointSlice.Namespace, endpointSlice.Name)
+
+		err := c.syncer.GetLocalFederator().Delete(endpointSlice)
+		if err != nil {
+			logger.Errorf(err, "Error deleting local EndpointSlice %s/%s", endpointSlice.Namespace, endpointSlice.Name)
+		}
+
+		return nil, false
+	}
+
+	return obj, false
+}
+
+func (c *EndpointSliceController) onRemoteEndpointSlice(obj runtime.Object, _ int, _ syncer.Operation) (runtime.Object, bool) {
+	endpointSlice := obj.(*discovery.EndpointSlice)
+	endpointSlice.Namespace = endpointSlice.GetObjectMeta().GetLabels()[constants.LabelSourceNamespace]
+
+	return endpointSlice, false
+}
+
+func (c *EndpointSliceController) onLocalEndpointSliceSynced(obj runtime.Object, op syncer.Operation) bool {
+	endpointSlice := obj.(*discovery.EndpointSlice)
+
+	serviceName := endpointSlice.Labels[mcsv1a1.LabelServiceName]
+	serviceNamespace := endpointSlice.Labels[constants.LabelSourceNamespace]
+
+	logger.V(log.DEBUG).Infof("Local EndpointSlice for service \"%s/%s\" %sd on broker", serviceNamespace, serviceName, op)
+
+	var err error
+
+	if op == syncer.Delete {
+		err = c.serviceImportAggregator.updateOnDelete(serviceName, serviceNamespace)
+	} else {
+		err = c.serviceImportAggregator.updateOnCreateOrUpdate(serviceName, serviceNamespace)
+		if err != nil {
+			c.updateExportedServiceStatus(serviceName, serviceNamespace, newServiceExportCondition(constants.ServiceExportSynced,
+				corev1.ConditionFalse, exportFailedReason, fmt.Sprintf("Unable to export: %v", err)))
+		} else {
+			c.updateExportedServiceStatus(serviceName, serviceNamespace, newServiceExportCondition(constants.ServiceExportSynced,
+				corev1.ConditionTrue, "", "Service was successfully exported to the broker"))
+		}
+	}
+
+	if err != nil {
+		logger.Errorf(err, "Error processing %sd EndpointSlice for service \"%s/%s\"", op, serviceNamespace, serviceName)
+	}
+
+	return err != nil
+}
diff --git a/pkg/agent/controller/service_import.go b/pkg/agent/controller/service_import.go
new file mode 100644
index 0000000..77e8d8c
--- /dev/null
+++ b/pkg/agent/controller/service_import.go
@@ -0,0 +1,410 @@
+/*
+SPDX-License-Identifier: Apache-2.0
+
+Copyright Contributors to the Submariner project.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/submariner-io/admiral/pkg/federate"
+	"github.com/submariner-io/admiral/pkg/log"
+	"github.com/submariner-io/admiral/pkg/resource"
+	"github.com/submariner-io/admiral/pkg/slices"
+	"github.com/submariner-io/admiral/pkg/syncer"
+	"github.com/submariner-io/admiral/pkg/syncer/broker"
+	"github.com/submariner-io/admiral/pkg/util"
+	"github.com/submariner-io/admiral/pkg/watcher"
+	"github.com/submariner-io/lighthouse/pkg/constants"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/tools/cache"
+	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
+)
+
+//nolint:gocritic // (hugeParam) This function modifies syncerConf so we don't want to pass by pointer.
+func newServiceImportController(spec *AgentSpecification, syncerMetricNames AgentConfig, syncerConfig broker.SyncerConfig,
+	brokerClient dynamic.Interface, brokerNamespace string, updateExportedServiceStatus updateExportedServiceStatusFn,
+) (*ServiceImportController, error) {
+	controller := &ServiceImportController{
+		localClient:                 syncerConfig.LocalClient,
+		restMapper:                  syncerConfig.RestMapper,
+		clusterID:                   spec.ClusterID,
+		localNamespace:              spec.Namespace,
+		converter:                   converter{scheme: syncerConfig.Scheme},
+		serviceImportAggregator:     newServiceImportAggregator(brokerClient, brokerNamespace, spec.ClusterID, syncerConfig.Scheme),
+		updateExportedServiceStatus: updateExportedServiceStatus,
+	}
+
+	syncCounter := prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: syncerMetricNames.ServiceImportCounterName,
+			Help: "Count of imported services",
+		},
+		[]string{
+			syncer.DirectionLabel,
+			syncer.OperationLabel,
+			syncer.SyncerNameLabel,
+		},
+	)
+	prometheus.MustRegister(syncCounter)
+
+	var err error
+
+	controller.localSyncer, err = syncer.NewResourceSyncer(&syncer.ResourceSyncerConfig{
+		Name:            "Local ServiceImport",
+		SourceClient:    syncerConfig.LocalClient,
+		SourceNamespace: controller.localNamespace,
+		Direction:       syncer.LocalToRemote,
+		RestMapper:      syncerConfig.RestMapper,
+		Federator:       controller,
+		ResourceType:    &mcsv1a1.ServiceImport{},
+		Transform:       controller.onLocalServiceImport,
+		Scheme:          syncerConfig.Scheme,
+		SyncCounter:     syncCounter,
+	})
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating local ServiceImport syncer")
+	}
+
+	controller.remoteSyncer, err = syncer.NewResourceSyncer(&syncer.ResourceSyncerConfig{
+		Name:            "Remote ServiceImport",
+		SourceClient:    brokerClient,
+		SourceNamespace: brokerNamespace,
+		RestMapper:      syncerConfig.RestMapper,
+		Federator:       federate.NewCreateOrUpdateFederator(syncerConfig.LocalClient, syncerConfig.RestMapper, corev1.NamespaceAll, ""),
+		ResourceType:    &mcsv1a1.ServiceImport{},
+		Transform:       controller.onRemoteServiceImport,
+		Scheme:          syncerConfig.Scheme,
+	})
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating ServiceImport watcher")
+	}
+
+	if spec.GlobalnetEnabled {
+		controller.globalIngressIPCache, err = newGlobalIngressIPCache(watcher.Config{
+			RestMapper: syncerConfig.RestMapper,
+			Client:     syncerConfig.LocalClient,
+			Scheme:     syncerConfig.Scheme,
+		})
+	}
+
+	return controller, err
+}
+
+func (c *ServiceImportController) start(stopCh <-chan struct{}) error {
+	if c.globalIngressIPCache != nil {
+		if err := c.globalIngressIPCache.start(stopCh); err != nil {
+			return err
+		}
+	}
+
+	go func() {
+		<-stopCh
+
+		c.endpointControllers.Range(func(key, value interface{}) bool {
+			value.(*EndpointController).stop()
+			return true
+		})
+
+		logger.Info("ServiceImport Controller stopped")
+	}()
+
+	if err := c.localSyncer.Start(stopCh); err != nil {
+		return errors.Wrap(err, "error starting local ServiceImport syncer")
+	}
+
+	if err := c.remoteSyncer.Start(stopCh); err != nil {
+		return errors.Wrap(err, "error starting remote ServiceImport syncer")
+	}
+
+	c.reconcileLocalAggregatedServiceImports()
+	c.reconcileRemoteAggregatedServiceImports()
+
+	return nil
+}
+
+func (c *ServiceImportController) reconcileRemoteAggregatedServiceImports() {
+	c.localSyncer.Reconcile(func() []runtime.Object {
+		siList, err := c.remoteSyncer.ListResources()
+		if err != nil {
+			logger.Error(err, "Error listing serviceImports")
+			return nil
+		}
+
+		retList := make([]runtime.Object, 0, len(siList))
+		for i := range siList {
+			si := c.converter.toServiceImport(siList[i])
+
+			serviceName, ok := si.Annotations[mcsv1a1.LabelServiceName]
+			if !ok {
+				// This is not an aggregated ServiceImport.
+				continue
+			}
+
+			if slices.IndexOf(si.Status.Clusters, c.clusterID, clusterStatusKey) < 0 {
+				continue
+			}
+
+			si.Name = serviceName + "-" + si.Annotations[constants.LabelSourceNamespace] + "-" + c.clusterID
+			si.Namespace = c.localNamespace
+			si.Labels = map[string]string{
+				mcsv1a1.LabelServiceName:       serviceName,
+				constants.LabelSourceNamespace: si.Annotations[constants.LabelSourceNamespace],
+			}
+
+			retList = append(retList, si)
+		}
+
+		return retList
+	})
+}
+
+func (c *ServiceImportController) reconcileLocalAggregatedServiceImports() {
+	c.remoteSyncer.Reconcile(func() []runtime.Object {
+		siList, err := c.localClient.Resource(serviceImportGVR).Namespace(corev1.NamespaceAll).List(context.TODO(), metav1.ListOptions{})
+		if err != nil {
+			logger.Error(err, "Error listing ServiceImports")
+			return nil
+		}
+
+		retList := make([]runtime.Object, 0, len(siList.Items))
+		for i := range siList.Items {
+			si := c.converter.toServiceImport(&siList.Items[i])
+
+			_, ok := si.Labels[mcsv1a1.LabelServiceName]
+			if ok {
+				// This is not an aggregated ServiceImport.
+				continue
+			}
+
+			si.Annotations = map[string]string{
+				mcsv1a1.LabelServiceName:       si.Name,
+				constants.LabelSourceNamespace: si.Namespace,
+			}
+
+			si.Name = fmt.Sprintf("%s-%s", si.Name, si.Namespace)
+			si.Namespace = c.serviceImportAggregator.brokerNamespace
+
+			retList = append(retList, si)
+		}
+
+		return retList
+	})
+}
+
+func (c *ServiceImportController) startEndpointsController(serviceImport *mcsv1a1.ServiceImport) error {
+	key, _ := cache.MetaNamespaceKeyFunc(serviceImport)
+
+	if obj, found := c.endpointControllers.LoadAndDelete(key); found {
+		logger.V(log.DEBUG).Infof("Stopping previous endpoints controller for %q", key)
+		obj.(*EndpointController).stop()
+	}
+
+	endpointController, err := startEndpointController(c.localClient, c.restMapper, c.converter.scheme,
+		serviceImport, c.clusterID, c.globalIngressIPCache)
+	if err != nil {
+		return errors.Wrapf(err, "failed to start endpoints controller for %q", key)
+	}
+
+	c.endpointControllers.Store(key, endpointController)
+
+	return nil
+}
+
+func (c *ServiceImportController) stopEndpointsController(key string) (bool, error) {
+	if obj, found := c.endpointControllers.Load(key); found {
+		endpointController := obj.(*EndpointController)
+		endpointController.stop()
+
+		found, err := endpointController.cleanup()
+		if err == nil {
+			c.endpointControllers.Delete(key)
+		}
+
+		return found, err
+	}
+
+	return false, nil
+}
+
+func (c *ServiceImportController) onLocalServiceImport(obj runtime.Object, _ int, op syncer.Operation) (runtime.Object, bool) {
+	serviceImport := obj.(*mcsv1a1.ServiceImport)
+	key, _ := cache.MetaNamespaceKeyFunc(serviceImport)
+
+	logger.V(log.DEBUG).Infof("Local ServiceImport %q %sd", key, op)
+
+	if op == syncer.Delete {
+		c.updateExportedServiceStatus(serviceImport.Labels[mcsv1a1.LabelServiceName],
+			serviceImport.Labels[constants.LabelSourceNamespace], newServiceExportCondition(constants.ServiceExportSynced,
+				corev1.ConditionFalse, "NoServiceImport", "ServiceImport was deleted"))
+
+		return obj, false
+	}
+
+	sourceCluster := serviceImport.Labels[constants.MCSLabelSourceCluster]
+	if sourceCluster == "" || sourceCluster != c.clusterID {
+		// TODO - handle migration of legacy per-cluster ServiceImports
+		return nil, false
+	}
+
+	serviceName, ok := serviceImport.Labels[mcsv1a1.LabelServiceName]
+	if !ok {
+		// The label is missing - most likely b/c the ServiceImport hasn't yet been migrated from the legacy labels.
+		logger.Infof("Label %q missing from ServiceImport (%s/%s) - not syncing", mcsv1a1.LabelServiceName,
+			serviceImport.Namespace, serviceImport.Name)
+
+		return nil, false
+	}
+
+	c.updateExportedServiceStatus(serviceName,
+		serviceImport.Labels[constants.LabelSourceNamespace], newServiceExportCondition(constants.ServiceExportSynced,
+			corev1.ConditionFalse, "AwaitingExport", fmt.Sprintf("ServiceImport %sd - awaiting aggregation on the broker", op)))
+
+	return obj, false
+}
+
+func (c *ServiceImportController) Distribute(obj runtime.Object) error {
+	localServiceImport := c.converter.toServiceImport(obj)
+	key, _ := cache.MetaNamespaceKeyFunc(localServiceImport)
+
+	logger.V(log.DEBUG).Infof("Distribute for local ServiceImport %q", key)
+
+	serviceName := serviceImportSourceName(localServiceImport)
+	aggregate := &mcsv1a1.ServiceImport{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: fmt.Sprintf("%s-%s", serviceName,
+				localServiceImport.Labels[constants.LabelSourceNamespace]),
+			Annotations: map[string]string{
+				mcsv1a1.LabelServiceName:       serviceName,
+				constants.LabelSourceNamespace: localServiceImport.Labels[constants.LabelSourceNamespace],
+			},
+		},
+		Spec: mcsv1a1.ServiceImportSpec{
+			Type:  localServiceImport.Spec.Type,
+			Ports: []mcsv1a1.ServicePort{},
+		},
+	}
+
+	// Here we just create the aggregated ServiceImport on the broker. We don't merge the local service info until we've
+	// successfully synced our local EndpointSlice to the broker. This is mainly done b/c the aggregated port information
+	// is determined from the constituent clusters' EndpointSlices, thus each cluster must have a consistent view of all
+	// the EndpointSlices in order for the aggregated port information to be eventually consistent.
+
+	result, err := util.CreateOrUpdate(context.Background(), resource.ForDynamic(c.serviceImportAggregator.brokerServiceImportClient()),
+		c.converter.toUnstructured(aggregate),
+		func(obj runtime.Object) (runtime.Object, error) {
+			// TODO - check for service type conflict.
+
+			return obj, nil
+		})
+	if err == nil {
+		err = c.startEndpointsController(localServiceImport)
+	}
+
+	if err != nil {
+		c.updateExportedServiceStatus(localServiceImport.Labels[mcsv1a1.LabelServiceName],
+			localServiceImport.Labels[constants.LabelSourceNamespace], newServiceExportCondition(constants.ServiceExportSynced,
+				corev1.ConditionFalse, exportFailedReason, fmt.Sprintf("Unable to export: %v", err)))
+	}
+
+	if result == util.OperationResultCreated {
+		logger.V(log.DEBUG).Infof("Created aggregated ServiceImport %q", aggregate.Name)
+	}
+
+	return err
+}
+
+func (c *ServiceImportController) Delete(obj runtime.Object) error {
+	localServiceImport := c.converter.toServiceImport(obj)
+	key, _ := cache.MetaNamespaceKeyFunc(localServiceImport)
+
+	logger.V(log.DEBUG).Infof("Delete for local ServiceImport %q", key)
+
+	// For consistency, we let the EndpointSlice controller handle removing the local service info from the aggregated
+	// ServiceImport on the broker after we delete the local EndpointSlice here. However, if the Endpoints controller
+	// was never started or if there are no local EndpointSlices, which can happen during reconciliation on startup or
+	// during clean up on uninstall, then we handle removal here.
+
+	found, err := c.stopEndpointsController(key)
+	if err != nil {
+		return err
+	}
+
+	if !found {
+		err = c.serviceImportAggregator.updateOnDelete(serviceImportSourceName(localServiceImport),
+			localServiceImport.Labels[constants.LabelSourceNamespace])
+	}
+
+	return err
+}
+
+func (c *ServiceImportController) onRemoteServiceImport(obj runtime.Object, _ int, _ syncer.Operation) (runtime.Object, bool) {
+	serviceImport := obj.(*mcsv1a1.ServiceImport)
+
+	serviceName, ok := serviceImport.Annotations[mcsv1a1.LabelServiceName]
+	if ok {
+		// This is an aggregated ServiceImport - sync it to the local service namespace.
+		serviceImport.Name = serviceName
+		serviceImport.Namespace = serviceImport.Annotations[constants.LabelSourceNamespace]
+
+		delete(serviceImport.Annotations, mcsv1a1.LabelServiceName)
+		delete(serviceImport.Annotations, constants.LabelSourceNamespace)
+
+		return serviceImport, false
+	}
+
+	return nil, false
+}
+
+func (c *ServiceImportController) localServiceImportLister(transform func(si *mcsv1a1.ServiceImport) runtime.Object) []runtime.Object {
+	siList, err := c.localSyncer.ListResources()
+	if err != nil {
+		logger.Error(err, "Error listing serviceImports")
+		return nil
+	}
+
+	// This function also checks the legacy source name label for migration - this can be removed after 0.15.
+	sourceClusterName := func(serviceImport *mcsv1a1.ServiceImport) string {
+		name, ok := serviceImport.Labels[constants.MCSLabelSourceCluster]
+		if ok {
+			return name
+		}
+
+		return serviceImport.Labels["lighthouse.submariner.io/sourceCluster"]
+	}
+
+	retList := make([]runtime.Object, 0, len(siList))
+
+	for _, obj := range siList {
+		si := obj.(*mcsv1a1.ServiceImport)
+
+		clusterID := sourceClusterName(si)
+		if clusterID != c.clusterID {
+			continue
+		}
+
+		retList = append(retList, transform(si))
+	}
+
+	return retList
+}
diff --git a/pkg/agent/controller/service_import_aggregator.go b/pkg/agent/controller/service_import_aggregator.go
new file mode 100644
index 0000000..6719fbf
--- /dev/null
+++ b/pkg/agent/controller/service_import_aggregator.go
@@ -0,0 +1,120 @@
+/*
+SPDX-License-Identifier: Apache-2.0
+
+Copyright Contributors to the Submariner project.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/pkg/errors"
+	"github.com/submariner-io/admiral/pkg/log"
+	"github.com/submariner-io/admiral/pkg/resource"
+	"github.com/submariner-io/admiral/pkg/slices"
+	"github.com/submariner-io/admiral/pkg/util"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/utils/pointer"
+	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
+)
+
+func newServiceImportAggregator(brokerClient dynamic.Interface, brokerNamespace, clusterID string, scheme *runtime.Scheme,
+) *ServiceImportAggregator {
+	return &ServiceImportAggregator{
+		clusterID:       clusterID,
+		converter:       converter{scheme: scheme},
+		brokerClient:    brokerClient,
+		brokerNamespace: brokerNamespace,
+	}
+}
+
+func (a *ServiceImportAggregator) updateOnCreateOrUpdate(name, namespace string) error {
+	return a.update(name, namespace, func(existing *mcsv1a1.ServiceImport) {
+		var added bool
+
+		existing.Status.Clusters, added = slices.AppendIfNotPresent(existing.Status.Clusters,
+			mcsv1a1.ClusterStatus{Cluster: a.clusterID}, clusterStatusKey)
+
+		if added {
+			logger.V(log.DEBUG).Infof("Added cluster name %q to aggregated ServiceImport %q. New status: %#v",
+				a.clusterID, existing.Name, existing.Status.Clusters)
+		}
+
+		// TODO - merge port info
+	})
+}
+
+func (a *ServiceImportAggregator) updateOnDelete(name, namespace string) error {
+	return a.update(name, namespace, func(existing *mcsv1a1.ServiceImport) {
+		var removed bool
+
+		existing.Status.Clusters, removed = slices.Remove(existing.Status.Clusters, mcsv1a1.ClusterStatus{Cluster: a.clusterID},
+			clusterStatusKey)
+		if !removed {
+			return
+		}
+
+		logger.V(log.DEBUG).Infof("Removed cluster name %q from aggregated ServiceImport %q. New status: %#v",
+			a.clusterID, existing.Name, existing.Status.Clusters)
+
+		// TODO - remove port info
+	})
+}
+
+func (a *ServiceImportAggregator) update(name, namespace string, mutate func(*mcsv1a1.ServiceImport)) error {
+	aggregate := &mcsv1a1.ServiceImport{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: fmt.Sprintf("%s-%s", name, namespace),
+		},
+	}
+
+	//nolint:wrapcheck // No need to wrap the return error.
+	return util.Update(context.Background(), resource.ForDynamic(a.brokerServiceImportClient()), a.converter.toUnstructured(aggregate),
+		func(obj runtime.Object) (runtime.Object, error) {
+			existing := a.converter.toServiceImport(obj)
+
+			mutate(existing)
+
+			if len(existing.Status.Clusters) == 0 {
+				logger.V(log.DEBUG).Infof("Deleting aggregated ServiceImport %q", existing.Name)
+
+				err := a.brokerServiceImportClient().Delete(context.Background(), existing.Name, metav1.DeleteOptions{
+					Preconditions: &metav1.Preconditions{
+						ResourceVersion: pointer.String(existing.ResourceVersion),
+					},
+				})
+				if apierrors.IsNotFound(err) {
+					err = nil
+				}
+
+				return obj, errors.Wrapf(err, "error deleting aggregated ServiceImport %q", existing.Name)
+			}
+
+			return a.converter.toUnstructured(existing), nil
+		})
+}
+
+func (a *ServiceImportAggregator) brokerServiceImportClient() dynamic.ResourceInterface {
+	return a.brokerClient.Resource(serviceImportGVR).Namespace(a.brokerNamespace)
+}
+
+func clusterStatusKey(c mcsv1a1.ClusterStatus) string {
+	return c.Cluster
+}
diff --git a/pkg/agent/controller/serviceimport.go b/pkg/agent/controller/serviceimport.go
deleted file mode 100644
index 73e9f90..0000000
--- a/pkg/agent/controller/serviceimport.go
+++ /dev/null
@@ -1,147 +0,0 @@
-/*
-SPDX-License-Identifier: Apache-2.0
-
-Copyright Contributors to the Submariner project.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package controller
-
-import (
-	"github.com/pkg/errors"
-	"github.com/submariner-io/admiral/pkg/federate"
-	"github.com/submariner-io/admiral/pkg/log"
-	"github.com/submariner-io/admiral/pkg/syncer"
-	"github.com/submariner-io/admiral/pkg/watcher"
-	"github.com/submariner-io/lighthouse/pkg/constants"
-	"k8s.io/apimachinery/pkg/api/meta"
-	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/client-go/dynamic"
-	"k8s.io/client-go/tools/cache"
-	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
-)
-
-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,
-	}
-
-	var err error
-
-	controller.serviceImportSyncer, err = syncer.NewResourceSyncer(&syncer.ResourceSyncerConfig{
-		Name:            "ServiceImport watcher",
-		SourceClient:    localClient,
-		SourceNamespace: spec.Namespace,
-		Direction:       syncer.LocalToRemote,
-		RestMapper:      restMapper,
-		Federator:       federate.NewNoopFederator(),
-		ResourceType:    &mcsv1a1.ServiceImport{},
-		Transform:       controller.serviceImportToEndpointController,
-		Scheme:          scheme,
-	})
-	if err != nil {
-		return nil, errors.Wrap(err, "error creating ServiceImport watcher")
-	}
-
-	if spec.GlobalnetEnabled {
-		controller.globalIngressIPCache, err = newGlobalIngressIPCache(watcher.Config{
-			RestMapper: restMapper,
-			Client:     localClient,
-			Scheme:     scheme,
-		})
-	}
-
-	return controller, err
-}
-
-func (c *ServiceImportController) start(stopCh <-chan struct{}) error {
-	if c.globalIngressIPCache != nil {
-		if err := c.globalIngressIPCache.start(stopCh); err != nil {
-			return err
-		}
-	}
-
-	go func() {
-		<-stopCh
-
-		c.endpointControllers.Range(func(key, value interface{}) bool {
-			value.(*EndpointController).stop()
-			return true
-		})
-
-		logger.Info("ServiceImport Controller stopped")
-	}()
-
-	if err := c.serviceImportSyncer.Start(stopCh); err != nil {
-		return errors.Wrap(err, "error starting ServiceImport watcher")
-	}
-
-	return nil
-}
-
-func (c *ServiceImportController) serviceImportCreatedOrUpdated(serviceImport *mcsv1a1.ServiceImport, key string) bool {
-	if _, found := c.endpointControllers.Load(key); found {
-		logger.V(log.DEBUG).Infof("The endpoint controller is already running for %q", key)
-		return false
-	}
-
-	sourceCluster := serviceImport.Labels[constants.MCSLabelSourceCluster]
-	if sourceCluster == "" || sourceCluster != c.clusterID {
-		return false
-	}
-
-	serviceNameSpace := serviceImport.Labels[constants.LabelSourceNamespace]
-	serviceName := serviceImport.Labels[mcsv1a1.LabelServiceName]
-
-	endpointController, err := startEndpointController(c.localClient, c.restMapper, c.scheme,
-		serviceImport, serviceNameSpace, serviceName, c.clusterID, c.globalIngressIPCache)
-	if err != nil {
-		logger.Error(err, "Failed to start endpoint controller")
-		return true
-	}
-
-	c.endpointControllers.Store(key, endpointController)
-
-	return false
-}
-
-func (c *ServiceImportController) serviceImportDeleted(key string) {
-	if obj, found := c.endpointControllers.LoadAndDelete(key); found {
-		endpointController := obj.(*EndpointController)
-		endpointController.stop()
-	}
-}
-
-func (c *ServiceImportController) serviceImportToEndpointController(obj runtime.Object, numRequeues int,
-	op syncer.Operation,
-) (runtime.Object, bool) {
-	serviceImport := obj.(*mcsv1a1.ServiceImport)
-	key, _ := cache.MetaNamespaceKeyFunc(serviceImport)
-
-	logger.V(log.DEBUG).Infof("ServiceImport %s %sd", key, op)
-
-	if op == syncer.Create || op == syncer.Update {
-		return nil, c.serviceImportCreatedOrUpdated(serviceImport, key)
-	}
-
-	c.serviceImportDeleted(key)
-
-	return nil, false
-}
diff --git a/pkg/agent/controller/types.go b/pkg/agent/controller/types.go
index a82966d..b0164cd 100644
--- a/pkg/agent/controller/types.go
+++ b/pkg/agent/controller/types.go
@@ -21,24 +21,34 @@ package controller
 import (
 	"sync"
 
+	"github.com/submariner-io/admiral/pkg/federate"
 	"github.com/submariner-io/admiral/pkg/syncer"
 	"github.com/submariner-io/admiral/pkg/syncer/broker"
 	"github.com/submariner-io/admiral/pkg/watcher"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/dynamic"
+	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
+const exportFailedReason = "ExportFailed"
+
+type updateExportedServiceStatusFn func(name, namespace string, condition ...mcsv1a1.ServiceExportCondition)
+
+type converter struct {
+	scheme *runtime.Scheme
+}
+
 type Controller struct {
-	clusterID               string
-	globalnetEnabled        bool
-	namespace               string
-	serviceExportClient     dynamic.NamespaceableResourceInterface
-	serviceExportSyncer     syncer.Interface
-	serviceImportSyncer     *broker.Syncer
-	endpointSliceSyncer     *broker.Syncer
-	serviceSyncer           syncer.Interface
-	serviceImportController *ServiceImportController
+	clusterID                   string
+	globalnetEnabled            bool
+	namespace                   string
+	serviceExportClient         dynamic.NamespaceableResourceInterface
+	serviceExportSyncer         syncer.Interface
+	endpointSliceController     *EndpointSliceController
+	serviceSyncer               syncer.Interface
+	serviceImportController     *ServiceImportController
+	localServiceImportFederator federate.Federator
 }
 
 type AgentSpecification struct {
@@ -48,34 +58,51 @@ type AgentSpecification struct {
 	Uninstall        bool
 }
 
-// The ServiceImportController listens for ServiceImport resources created in the target namespace
-// and creates an EndpointController in response. The EndpointController will use the app label as filter
-// to listen only for the endpoints event related to ServiceImport created.
+type ServiceImportAggregator struct {
+	clusterID       string
+	converter       converter
+	brokerClient    dynamic.Interface
+	brokerNamespace string
+}
+
+// The ServiceImportController encapsulates two resource syncers; one that watches for local cluster ServiceImports
+// from the submariner namespace and creates/updates the aggregated ServiceImport on the broker; the other that syncs
+// aggregated ServiceImports from the broker to the local service namespace. It also creates an EndpointController.
 type ServiceImportController struct {
-	serviceSyncer        syncer.Interface
-	localClient          dynamic.Interface
-	restMapper           meta.RESTMapper
-	serviceImportSyncer  syncer.Interface
-	endpointControllers  sync.Map
+	localClient                 dynamic.Interface
+	restMapper                  meta.RESTMapper
+	serviceImportAggregator     *ServiceImportAggregator
+	updateExportedServiceStatus updateExportedServiceStatusFn
+	localSyncer                 syncer.Interface
+	remoteSyncer                syncer.Interface
+	endpointControllers         sync.Map
+	clusterID                   string
+	localNamespace              string
+	converter                   converter
+	globalIngressIPCache        *globalIngressIPCache
+}
+
+// Each EndpointController watches for the Endpoints that backs a Service and have a ServiceImport by using a filter
+// that listen only for the Service's endpoints. It creates an EndpointSlice corresponding to an Endpoints object that is
+// distributed to other clusters.
+type EndpointController struct {
 	clusterID            string
-	scheme               *runtime.Scheme
+	serviceName          string
+	serviceNamespace     string
+	serviceImportSpec    *mcsv1a1.ServiceImportSpec
+	stopCh               chan struct{}
+	stopOnce             sync.Once
+	localClient          dynamic.Interface
+	ingressIPClient      dynamic.NamespaceableResourceInterface
 	globalIngressIPCache *globalIngressIPCache
 }
 
-// Each EndpointController listens for the endpoints that backs a service and have a ServiceImport
-// It will create an endpoint slice corresponding to an endpoint object and set the owner references
-// to ServiceImport. The app label from the endpoint will be added to endpoint slice as well.
-type EndpointController struct {
-	clusterID                    string
-	serviceImportName            string
-	serviceName                  string
-	serviceImportSourceNameSpace string
-	stopCh                       chan struct{}
-	stopOnce                     sync.Once
-	isHeadless                   bool
-	localClient                  dynamic.Interface
-	ingressIPClient              dynamic.NamespaceableResourceInterface
-	globalIngressIPCache         *globalIngressIPCache
+// EndpointSliceController encapsulates a syncer that syncs EndpointSlices to and from that broker.
+type EndpointSliceController struct {
+	clusterID                   string
+	syncer                      *broker.Syncer
+	serviceImportAggregator     *ServiceImportAggregator
+	updateExportedServiceStatus updateExportedServiceStatusFn
 }
 
 type globalIngressIPCache struct {
diff --git a/pkg/constants/constants.go b/pkg/constants/constants.go
index 962e613..24dc4f0 100644
--- a/pkg/constants/constants.go
+++ b/pkg/constants/constants.go
@@ -26,6 +26,7 @@ const (
 	LabelSourceNamespace  = "lighthouse.submariner.io/sourceNamespace"
 	LabelValueManagedBy   = "lighthouse-agent.submariner.io"
 	MCSLabelSourceCluster = "multicluster.kubernetes.io/source-cluster"
+	LabelIsHeadless       = "lighthouse.submariner.io/is-headless"
 )
 
 const ServiceExportSynced v1alpha1.ServiceExportConditionType = "Synced"


Test patch:
diff --git a/coredns/plugin/handler_test.go b/coredns/plugin/handler_test.go
index 2343217..6aa0576 100644
--- a/coredns/plugin/handler_test.go
+++ b/coredns/plugin/handler_test.go
@@ -43,18 +43,17 @@ import (
 )
 
 const (
-	service1       = "service1"
-	namespace1     = "namespace1"
-	namespace2     = "namespace2"
-	serviceIP      = "100.96.156.101"
-	serviceIP2     = "100.96.156.102"
-	localClusterID = "local"
-	clusterID      = "cluster1"
-	clusterID2     = "cluster2"
-	endpointIP     = "100.96.157.101"
-	endpointIP2    = "100.96.157.102"
-	hostName1      = "hostName1"
-	hostName2      = "hostName2"
+	service1    = "service1"
+	namespace1  = "namespace1"
+	namespace2  = "namespace2"
+	serviceIP   = "100.96.156.101"
+	serviceIP2  = "100.96.156.102"
+	clusterID   = "cluster1"
+	clusterID2  = "cluster2"
+	endpointIP  = "100.96.157.101"
+	endpointIP2 = "100.96.157.102"
+	hostName1   = "hostName1"
+	hostName2   = "hostName2"
 )
 
 var (
@@ -99,7 +98,7 @@ func testWithoutFallback() {
 		t = newHandlerTestDriver()
 		t.mockCs.ConnectClusterID(clusterID)
 
-		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, clusterID, serviceIP, mcsv1a1.ClusterSetIP, port1))
+		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, mcsv1a1.ClusterSetIP))
 
 		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1},
 			newEndpoint(serviceIP, "", true)))
@@ -163,7 +162,7 @@ func testWithoutFallback() {
 		qname := fmt.Sprintf("%s.%s.svc.clusterset.local.", service1, namespace2)
 
 		BeforeEach(func() {
-			t.lh.Resolver.PutServiceImport(newServiceImport(namespace2, service1, clusterID, serviceIP, mcsv1a1.ClusterSetIP, port1))
+			t.lh.Resolver.PutServiceImport(newServiceImport(namespace2, service1, mcsv1a1.ClusterSetIP))
 
 			t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace2, service1, clusterID, []mcsv1a1.ServicePort{port1},
 				newEndpoint(serviceIP, "", true)))
@@ -321,7 +320,7 @@ func testWithFallback() {
 			return dns.RcodeBadCookie, nil
 		})
 
-		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, clusterID, serviceIP, mcsv1a1.ClusterSetIP, port1))
+		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, mcsv1a1.ClusterSetIP))
 
 		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1},
 			newEndpoint(serviceIP, "", true)))
@@ -442,14 +441,12 @@ func testClusterStatus() {
 		t.mockCs.ConnectClusterID(clusterID)
 		t.mockCs.ConnectClusterID(clusterID2)
 
-		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, clusterID, serviceIP, mcsv1a1.ClusterSetIP, port1))
+		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, mcsv1a1.ClusterSetIP))
 
 		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1},
 			newEndpoint(serviceIP, "", true)))
 
-		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, clusterID2, serviceIP2, mcsv1a1.ClusterSetIP, port1, port2))
-
-		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID2, []mcsv1a1.ServicePort{port2},
+		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID2, []mcsv1a1.ServicePort{port1, port2},
 			newEndpoint(serviceIP2, "", true)))
 
 		rec = dnstest.NewRecorder(&test.ResponseWriter{})
@@ -482,7 +479,7 @@ func testClusterStatus() {
 		})
 	})
 
-	When("service is in two clusters and only one is connected", func() {
+	When("a service is in two clusters and only one is connected", func() {
 		JustBeforeEach(func() {
 			t.mockCs.DisconnectClusterID(clusterID)
 		})
@@ -512,7 +509,7 @@ func testClusterStatus() {
 		})
 	})
 
-	When("service is present in two clusters and both are disconnected", func() {
+	When("a service is present in two clusters and both are disconnected", func() {
 		JustBeforeEach(func() {
 			t.mockCs.DisconnectAll()
 		})
@@ -538,11 +535,11 @@ func testClusterStatus() {
 		})
 	})
 
-	When("service is present in one cluster and it is disconnected", func() {
+	When("a service is present in one cluster and it is disconnected", func() {
 		JustBeforeEach(func() {
 			t.mockCs.DisconnectClusterID(clusterID)
 
-			t.lh.Resolver.RemoveServiceImport(newServiceImport(namespace1, service1, clusterID2, serviceIP2, mcsv1a1.ClusterSetIP))
+			t.lh.Resolver.RemoveEndpointSlice(newEndpointSlice(namespace1, service1, clusterID2, []mcsv1a1.ServicePort{}))
 		})
 
 		qname := fmt.Sprintf("%s.%s.svc.clusterset.local.", service1, namespace1)
@@ -581,7 +578,7 @@ func testHeadlessService() {
 
 		t.mockCs.ConnectClusterID(clusterID)
 
-		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, clusterID, serviceIP, mcsv1a1.Headless, port1))
+		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, mcsv1a1.Headless))
 
 		rec = dnstest.NewRecorder(&test.ResponseWriter{})
 	})
@@ -721,7 +718,7 @@ func testHeadlessService() {
 
 	When("headless service is present in two clusters", func() {
 		BeforeEach(func() {
-			t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, clusterID2, serviceIP, mcsv1a1.Headless, port1))
+			t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, mcsv1a1.Headless))
 
 			t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID2, []mcsv1a1.ServicePort{port1},
 				newEndpoint(endpointIP2, hostName2, true)))
@@ -775,12 +772,12 @@ func testLocalService() {
 		t.mockCs.ConnectClusterID(clusterID)
 		t.mockCs.ConnectClusterID(clusterID2)
 
-		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, clusterID, serviceIP, mcsv1a1.ClusterSetIP, port1))
+		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, mcsv1a1.ClusterSetIP))
 
 		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1},
 			newEndpoint(serviceIP, "", true)))
 
-		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, clusterID2, serviceIP2, mcsv1a1.ClusterSetIP, port1, port2))
+		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, mcsv1a1.ClusterSetIP))
 
 		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID2, []mcsv1a1.ServicePort{port1, port2},
 			newEndpoint(serviceIP2, "", true)))
@@ -856,7 +853,8 @@ func testLocalService() {
 
 	When("service is in local and remote clusters and local has no active endpoints", func() {
 		BeforeEach(func() {
-			t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1}))
+			t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1},
+				newEndpoint(serviceIP, "", false)))
 		})
 
 		qname := fmt.Sprintf("%s.%s.svc.clusterset.local.", service1, namespace1)
@@ -884,7 +882,7 @@ func testSRVMultiplePorts() {
 		t = newHandlerTestDriver()
 		t.mockCs.ConnectClusterID(clusterID)
 
-		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, clusterID, serviceIP, mcsv1a1.ClusterSetIP, port1, port2))
+		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, mcsv1a1.ClusterSetIP))
 
 		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1, port2},
 			newEndpoint(endpointIP, "", true)))
@@ -994,29 +992,14 @@ func (t *handlerTestDriver) executeTestCase(rec *dnstest.Recorder, tc test.Case)
 }
 
 //nolint:unparam // `name` always receives `service1'.
-func newServiceImport(namespace, name, clusterID, serviceIP string, siType mcsv1a1.ServiceImportType,
-	ports ...mcsv1a1.ServicePort) *mcsv1a1.ServiceImport {
+func newServiceImport(namespace, name string, siType mcsv1a1.ServiceImportType) *mcsv1a1.ServiceImport {
 	return &mcsv1a1.ServiceImport{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      name,
 			Namespace: namespace,
-			Labels: map[string]string{
-				mcsv1a1.LabelServiceName:        name,
-				constants.LabelSourceNamespace:  namespace,
-				constants.MCSLabelSourceCluster: clusterID,
-			},
 		},
 		Spec: mcsv1a1.ServiceImportSpec{
-			Type:  siType,
-			IPs:   []string{serviceIP},
-			Ports: ports,
-		},
-		Status: mcsv1a1.ServiceImportStatus{
-			Clusters: []mcsv1a1.ClusterStatus{
-				{
-					Cluster: clusterID,
-				},
-			},
+			Type: siType,
 		},
 	}
 }
@@ -1043,6 +1026,7 @@ func newEndpointSlice(namespace, name, clusterID string, ports []mcsv1a1.Service
 				constants.LabelSourceNamespace:  namespace,
 				constants.MCSLabelSourceCluster: clusterID,
 				mcsv1a1.LabelServiceName:        name,
+				constants.LabelIsHeadless:       "false",
 			},
 		},
 		AddressType: discovery.AddressTypeIPv4,
diff --git a/coredns/resolver/clusterip_service_test.go b/coredns/resolver/clusterip_service_test.go
index 4259e44..1739889 100644
--- a/coredns/resolver/clusterip_service_test.go
+++ b/coredns/resolver/clusterip_service_test.go
@@ -21,7 +21,9 @@ package resolver_test
 import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"github.com/submariner-io/admiral/pkg/syncer/test"
 	"github.com/submariner-io/lighthouse/coredns/resolver"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
@@ -45,10 +47,10 @@ func testClusterIPServiceInOneCluster() {
 	}
 
 	BeforeEach(func() {
-		t.resolver.PutServiceImport(newClusterServiceImport(namespace1, service1, expDNSRecord.IP, expDNSRecord.ClusterName,
-			expDNSRecord.Ports...))
+		t.resolver.PutServiceImport(newAggregatedServiceImport(namespace1, service1))
 
-		t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, expDNSRecord.ClusterName, expDNSRecord.IP, true))
+		t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, expDNSRecord.ClusterName, expDNSRecord.IP,
+			true, expDNSRecord.Ports...))
 	})
 
 	Context("and no specific cluster is requested", func() {
@@ -85,7 +87,7 @@ func testClusterIPServiceInOneCluster() {
 
 	Context("and it becomes unhealthy", func() {
 		BeforeEach(func() {
-			t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, false))
+			t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, false, expDNSRecord.Ports...))
 		})
 
 		It("should return no DNS records", func() {
@@ -101,7 +103,7 @@ func testClusterIPServiceInOneCluster() {
 
 	Context("and the service information is updated", func() {
 		BeforeEach(func() {
-			t.resolver.PutServiceImport(newClusterServiceImport(namespace1, service1, serviceIP2, clusterID1, port2))
+			t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP2, true, port2))
 		})
 
 		It("should return the correct DNS record information", func() {
@@ -124,11 +126,10 @@ func testClusterIPServiceInTwoClusters() {
 	t := newTestDriver()
 
 	BeforeEach(func() {
-		t.resolver.PutServiceImport(newClusterServiceImport(namespace1, service1, serviceIP1, clusterID1, port1))
-		t.resolver.PutServiceImport(newClusterServiceImport(namespace1, service1, serviceIP2, clusterID2, port1))
+		t.resolver.PutServiceImport(newAggregatedServiceImport(namespace1, service1))
 
-		t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true))
-		t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID2, serviceIP2, true))
+		t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true, port1))
+		t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID2, serviceIP2, true, port1))
 	})
 
 	Context("and no specific cluster is requested", func() {
@@ -203,7 +204,7 @@ func testClusterIPServiceInTwoClusters() {
 		}
 
 		BeforeEach(func() {
-			t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID2, serviceIP2, false))
+			t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID2, serviceIP2, false, port1))
 		})
 
 		Context("and no specific cluster is requested", func() {
@@ -239,7 +240,7 @@ func testClusterIPServiceInTwoClusters() {
 		}
 
 		BeforeEach(func() {
-			t.resolver.RemoveServiceImport(newClusterServiceImport(namespace1, service1, serviceIP2, clusterID2))
+			t.resolver.RemoveEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID2, serviceIP2, true, port1))
 		})
 
 		It("should consistently return the DNS record of the remaining cluster", func() {
@@ -264,13 +265,11 @@ func testClusterIPServiceInThreeClusters() {
 	t := newTestDriver()
 
 	BeforeEach(func() {
-		t.resolver.PutServiceImport(newClusterServiceImport(namespace1, service1, serviceIP1, clusterID1, port1))
-		t.resolver.PutServiceImport(newClusterServiceImport(namespace1, service1, serviceIP2, clusterID2, port1, port2))
-		t.resolver.PutServiceImport(newClusterServiceImport(namespace1, service1, serviceIP3, clusterID3, port1))
+		t.resolver.PutServiceImport(newAggregatedServiceImport(namespace1, service1))
 
-		t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true))
-		t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID2, serviceIP2, true))
-		t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID3, serviceIP3, true))
+		t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true, port1))
+		t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID2, serviceIP2, true, port1, port2))
+		t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID3, serviceIP3, true, port1))
 	})
 
 	Context("and no specific cluster is requested", func() {
@@ -311,7 +310,7 @@ func testClusterIPServiceInThreeClusters() {
 
 	Context("and one becomes unhealthy", func() {
 		BeforeEach(func() {
-			t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID2, serviceIP2, false))
+			t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID2, serviceIP2, false, port1, port2))
 		})
 
 		It("should consistently return the healthy clusters' DNS records round-robin", func() {
@@ -324,7 +323,7 @@ func testClusterIPServiceInThreeClusters() {
 					Expect(t.getNonHeadlessDNSRecord(namespace1, service1, "").IP).To(Or(Equal(serviceIP1), Equal(serviceIP3)))
 				}
 
-				t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID2, serviceIP2, true))
+				t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID2, serviceIP2, true, port1, port2))
 
 				t.testRoundRobin(namespace1, service1, serviceIP1, serviceIP2, serviceIP3)
 			})
@@ -334,7 +333,7 @@ func testClusterIPServiceInThreeClusters() {
 	Context("and one becomes disconnected and one becomes unhealthy", func() {
 		BeforeEach(func() {
 			t.clusterStatus.DisconnectClusterID(clusterID2)
-			t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID3, serviceIP3, false))
+			t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID3, serviceIP3, false, port1))
 		})
 
 		It("should consistently return the remaining cluster's DNS record round-robin", func() {
@@ -348,56 +347,97 @@ func testClusterIPServiceMisc() {
 
 	When("a service exists in two namespaces", func() {
 		BeforeEach(func() {
-			t.resolver.PutServiceImport(newClusterServiceImport(namespace1, service1, serviceIP1, clusterID1))
-			t.resolver.PutServiceImport(newClusterServiceImport(namespace2, service1, serviceIP2, clusterID1))
+			t.resolver.PutServiceImport(newAggregatedServiceImport(namespace1, service1))
+			t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true))
+
+			t.resolver.PutServiceImport(newAggregatedServiceImport(namespace2, service1))
+			t.putEndpointSlice(newClusterIPEndpointSlice(namespace2, service1, clusterID1, serviceIP2, true))
 		})
 
 		It("should return the correct DNS record for each namespace", func() {
 			t.assertDNSRecordsFound(namespace1, service1, clusterID1, "", false, resolver.DNSRecord{
 				IP:          serviceIP1,
+				Ports:       []mcsv1a1.ServicePort{},
 				ClusterName: clusterID1,
 			})
 
 			t.assertDNSRecordsFound(namespace2, service1, clusterID1, "", false, resolver.DNSRecord{
 				IP:          serviceIP2,
+				Ports:       []mcsv1a1.ServicePort{},
 				ClusterName: clusterID1,
 			})
 		})
 	})
 
-	When("a per-cluster ServiceImport has the legacy annotations", func() {
-		BeforeEach(func() {
-			si := newClusterServiceImport(namespace1, service1, serviceIP1, clusterID1, port1)
-			si.Labels = map[string]string{"lighthouse.submariner.io/sourceCluster": clusterID1}
-			si.Annotations = map[string]string{"origin-name": service1, "origin-namespace": namespace1}
-			t.resolver.PutServiceImport(si)
-		})
-
-		It("should correctly process it and return its DNS record", func() {
-			t.assertDNSRecordsFound(namespace1, service1, clusterID1, "", false, resolver.DNSRecord{
-				IP:          serviceIP1,
-				Ports:       []mcsv1a1.ServicePort{port1},
-				ClusterName: clusterID1,
-			})
-		})
-	})
-
-	When("a cluster's EndpointSlice is created before its ServiceImport", func() {
-		It("should correctly process them and return its DNS record", func() {
-			t.resolver.PutServiceImport(newClusterServiceImport(namespace1, service1, serviceIP2, clusterID2))
-
+	When("a cluster's EndpointSlice is initially created before the ServiceImport", func() {
+		It("should eventually process them and return its DNS record", func() {
 			es := newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true)
 			Expect(t.resolver.PutEndpointSlice(es)).To(BeTrue())
 
 			t.awaitDNSRecords(namespace1, service1, clusterID1, "", false)
 
-			t.resolver.PutServiceImport(newClusterServiceImport(namespace1, service1, serviceIP1, clusterID1))
+			t.resolver.PutServiceImport(newAggregatedServiceImport(namespace1, service1))
 			t.putEndpointSlice(es)
 
 			t.assertDNSRecordsFound(namespace1, service1, "", "", false, resolver.DNSRecord{
 				IP:          serviceIP1,
+				Ports:       []mcsv1a1.ServicePort{},
 				ClusterName: clusterID1,
 			})
 		})
 	})
+
+	When("a local cluster ServiceImport is created", func() {
+		It("should ignore it", func() {
+			serviceImport := &mcsv1a1.ServiceImport{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      service1 + "-" + namespace1 + "-" + clusterID1,
+					Namespace: submarinerNamespace,
+					Labels: map[string]string{
+						mcsv1a1.LabelServiceName: service1,
+					},
+				},
+				Spec: mcsv1a1.ServiceImportSpec{
+					Type: mcsv1a1.ClusterSetIP,
+					IPs:  []string{serviceIP1},
+				},
+				Status: mcsv1a1.ServiceImportStatus{
+					Clusters: []mcsv1a1.ClusterStatus{
+						{
+							Cluster: clusterID1,
+						},
+					},
+				},
+			}
+
+			t.resolver.PutServiceImport(serviceImport)
+
+			Expect(t.resolver.PutEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true))).To(BeTrue())
+
+			t.resolver.RemoveServiceImport(serviceImport)
+		})
+	})
+
+	When("an aggregated ServiceImport on the broker is created", func() {
+		It("should ignore it", func() {
+			serviceImport := &mcsv1a1.ServiceImport{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      service1 + "-" + namespace1,
+					Namespace: test.RemoteNamespace,
+					Annotations: map[string]string{
+						mcsv1a1.LabelServiceName: service1,
+					},
+				},
+				Spec: mcsv1a1.ServiceImportSpec{
+					Type: mcsv1a1.ClusterSetIP,
+				},
+			}
+
+			t.resolver.PutServiceImport(serviceImport)
+
+			Expect(t.resolver.PutEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true))).To(BeTrue())
+
+			t.resolver.RemoveServiceImport(serviceImport)
+		})
+	})
 }
diff --git a/coredns/resolver/controller_test.go b/coredns/resolver/controller_test.go
index dd9cad3..e273b1b 100644
--- a/coredns/resolver/controller_test.go
+++ b/coredns/resolver/controller_test.go
@@ -23,9 +23,11 @@ import (
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"github.com/submariner-io/admiral/pkg/syncer/test"
 	"github.com/submariner-io/lighthouse/coredns/resolver"
 	discovery "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
 	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
@@ -38,46 +40,22 @@ var _ = Describe("Controller", func() {
 		ClusterName: clusterID1,
 	}
 
-	When("a ServiceImport is created", func() {
-		var serviceImport *mcsv1a1.ServiceImport
-
-		BeforeEach(func() {
-			serviceImport = newClusterServiceImport(namespace1, service1, serviceIP1, clusterID1, port1)
-			t.createServiceImport(serviceImport)
-		})
-
-		Specify("GetDNSRecords should return the cluster's DNS record when requested", func() {
-			t.awaitDNSRecordsFound(namespace1, service1, clusterID1, "", false, expDNSRecord)
-		})
-
-		Context("and is subsequently deleted", func() {
-			Specify("GetDNSRecords should eventually return no DNS record found", func() {
-				t.awaitDNSRecords(namespace1, service1, clusterID1, "", true)
-
-				err := t.serviceImports.Namespace(serviceImport.Namespace).Delete(context.TODO(), serviceImport.Name, metav1.DeleteOptions{})
-				Expect(err).To(Succeed())
-
-				t.awaitDNSRecords(namespace1, service1, clusterID1, "", false)
-			})
-		})
-	})
-
 	When("an EndpointSlice is created", func() {
 		var endpointSlice *discovery.EndpointSlice
 
 		JustBeforeEach(func() {
-			endpointSlice = newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true)
+			endpointSlice = newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true, port1)
 			t.createEndpointSlice(endpointSlice)
 		})
 
-		Context("before a ServiceImport", func() {
-			Specify("GetDNSRecords should eventually deem the cluster healthy and return its DNS record", func() {
+		Context("before the ServiceImport", func() {
+			Specify("GetDNSRecords should eventually return its DNS record", func() {
 				Consistently(func() bool {
 					_, _, found := t.resolver.GetDNSRecords(namespace1, service1, "", "")
 					return found
 				}).Should(BeFalse())
 
-				t.createServiceImport(newClusterServiceImport(namespace1, service1, serviceIP1, clusterID1, port1))
+				t.createServiceImport(newAggregatedServiceImport(namespace1, service1))
 
 				t.awaitDNSRecordsFound(namespace1, service1, clusterID1, "", false, expDNSRecord)
 			})
@@ -85,11 +63,11 @@ var _ = Describe("Controller", func() {
 
 		Context("after a ServiceImport", func() {
 			BeforeEach(func() {
-				t.createServiceImport(newClusterServiceImport(namespace1, service1, serviceIP1, clusterID1, port1))
+				t.createServiceImport(newAggregatedServiceImport(namespace1, service1))
 			})
 
-			Context("and then deleted", func() {
-				Specify("GetDNSRecords should eventually deem the cluster unhealthy and return no DNS record", func() {
+			Context("and then the EndpointSlice is deleted", func() {
+				Specify("GetDNSRecords should eventually return no DNS record", func() {
 					t.awaitDNSRecordsFound(namespace1, service1, clusterID1, "", false, expDNSRecord)
 
 					err := t.endpointSlices.Namespace(namespace1).Delete(context.TODO(), endpointSlice.Name, metav1.DeleteOptions{})
@@ -99,13 +77,24 @@ var _ = Describe("Controller", func() {
 				})
 			})
 
-			Context("and then updated to unhealthy", func() {
-				Specify("GetDNSRecords should eventually return no DNS record", func() {
+			Context("and then the ServiceImport is deleted", func() {
+				Specify("GetDNSRecords should eventually return not found", func() {
 					t.awaitDNSRecordsFound(namespace1, service1, clusterID1, "", false, expDNSRecord)
 
-					err := t.endpointSlices.Namespace(namespace1).Delete(context.TODO(), endpointSlice.Name, metav1.DeleteOptions{})
+					err := t.serviceImports.Namespace(namespace1).Delete(context.TODO(), service1, metav1.DeleteOptions{})
 					Expect(err).To(Succeed())
 
+					t.awaitDNSRecords(namespace1, service1, clusterID1, "", false)
+				})
+			})
+
+			Context("and then the EndpointSlice is updated to unhealthy", func() {
+				Specify("GetDNSRecords should eventually return no DNS record", func() {
+					t.awaitDNSRecordsFound(namespace1, service1, clusterID1, "", false, expDNSRecord)
+
+					endpointSlice.Endpoints[0].Conditions.Ready = pointer.Bool(false)
+					test.UpdateResource(t.endpointSlices.Namespace(namespace1), endpointSlice)
+
 					t.awaitDNSRecordsFound(namespace1, service1, "", "", false)
 				})
 			})
diff --git a/coredns/resolver/endpoint_slice_test.go b/coredns/resolver/endpoint_slice_test.go
index 0eb069a..e43f9ee 100644
--- a/coredns/resolver/endpoint_slice_test.go
+++ b/coredns/resolver/endpoint_slice_test.go
@@ -20,6 +20,7 @@ package resolver_test
 
 import (
 	. "github.com/onsi/ginkgo/v2"
+	"github.com/submariner-io/admiral/pkg/syncer/test"
 	"github.com/submariner-io/lighthouse/coredns/constants"
 	discovery "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -65,6 +66,22 @@ var _ = Describe("PutEndpointSlice", func() {
 			})
 		})
 	})
+
+	When("the EndpointSlice is on the broker", func() {
+		It("should not process it", func() {
+			t.putEndpointSlice(&discovery.EndpointSlice{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test",
+					Namespace: test.RemoteNamespace,
+					Labels: map[string]string{
+						constants.MCSLabelSourceCluster: "test",
+						mcsv1a1.LabelServiceName:        "test",
+						constants.LabelSourceNamespace:  namespace1,
+					},
+				},
+			})
+		})
+	})
 })
 
 var _ = Describe("RemoveEndpointSlice", func() {
@@ -92,9 +109,25 @@ var _ = Describe("RemoveEndpointSlice", func() {
 
 	When("the cluster information doesn't exist", func() {
 		It("should not process it", func() {
-			t.resolver.PutServiceImport(newClusterHeadlessServiceImport(namespace1, service1, clusterID1))
+			t.resolver.PutServiceImport(newAggregatedServiceImport(namespace1, service1))
 
 			t.resolver.RemoveEndpointSlice(newEndpointSlice(namespace1, service1, clusterID1, nil))
 		})
 	})
+
+	When("the EndpointSlice is on the broker", func() {
+		It("should not process it", func() {
+			t.resolver.RemoveEndpointSlice(&discovery.EndpointSlice{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test",
+					Namespace: test.RemoteNamespace,
+					Labels: map[string]string{
+						constants.MCSLabelSourceCluster: "test",
+						mcsv1a1.LabelServiceName:        "test",
+						constants.LabelSourceNamespace:  namespace1,
+					},
+				},
+			})
+		})
+	})
 })
diff --git a/coredns/resolver/headless_service_test.go b/coredns/resolver/headless_service_test.go
index a44dbc7..61eedc8 100644
--- a/coredns/resolver/headless_service_test.go
+++ b/coredns/resolver/headless_service_test.go
@@ -77,11 +77,11 @@ func testHeadlessServiceInMultipleClusters() {
 		HostName:    hostName2,
 	}
 
-	JustBeforeEach(func() {
-		t.resolver.PutServiceImport(newClusterHeadlessServiceImport(namespace1, service1, clusterID1))
-		t.resolver.PutServiceImport(newClusterHeadlessServiceImport(namespace1, service1, clusterID2))
-		t.resolver.PutServiceImport(newClusterHeadlessServiceImport(namespace1, service1, clusterID3))
+	BeforeEach(func() {
+		t.resolver.PutServiceImport(newHeadlessAggregatedServiceImport(namespace1, service1))
+	})
 
+	JustBeforeEach(func() {
 		t.putEndpointSlice(newEndpointSlice(namespace1, service1, clusterID1, []mcsv1a1.ServicePort{port1}, discovery.Endpoint{
 			Addresses: []string{endpointIP1},
 		}))
@@ -141,7 +141,6 @@ func testHeadlessServiceInMultipleClusters() {
 
 			// If the local cluster EndpointSlice is created before the local K8s EndpointSlice, PutEndpointSlice should
 			// return true to requeue.
-			t.resolver.PutServiceImport(newClusterHeadlessServiceImport(namespace1, service1, clusterID3))
 			Expect(t.resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID3, nil))).To(BeTrue())
 
 			t.createEndpointSlice(&discovery.EndpointSlice{
diff --git a/coredns/resolver/resolver_suite_test.go b/coredns/resolver/resolver_suite_test.go
index e39bb70..ef6b03f 100644
--- a/coredns/resolver/resolver_suite_test.go
+++ b/coredns/resolver/resolver_suite_test.go
@@ -41,6 +41,7 @@ import (
 	"k8s.io/client-go/dynamic"
 	fakeClient "k8s.io/client-go/dynamic/fake"
 	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/utils/pointer"
 	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
@@ -256,39 +257,20 @@ func (t *testDriver) putEndpointSlice(es *discovery.EndpointSlice) {
 	Expect(t.resolver.PutEndpointSlice(es)).To(BeFalse())
 }
 
-func newClusterServiceImport(namespace, name, serviceIP, clusterID string, ports ...mcsv1a1.ServicePort) *mcsv1a1.ServiceImport {
-	var ips []string
-	if serviceIP != "" {
-		ips = []string{serviceIP}
-	}
-
+func newAggregatedServiceImport(namespace, name string) *mcsv1a1.ServiceImport {
 	return &mcsv1a1.ServiceImport{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      name + "-" + namespace + "-" + clusterID,
-			Namespace: submarinerNamespace,
-			Labels: map[string]string{
-				mcsv1a1.LabelServiceName:        name,
-				constants.LabelSourceNamespace:  namespace,
-				constants.MCSLabelSourceCluster: clusterID,
-			},
+			Name:      name,
+			Namespace: namespace,
 		},
 		Spec: mcsv1a1.ServiceImportSpec{
-			Type:  mcsv1a1.ClusterSetIP,
-			IPs:   ips,
-			Ports: ports,
-		},
-		Status: mcsv1a1.ServiceImportStatus{
-			Clusters: []mcsv1a1.ClusterStatus{
-				{
-					Cluster: clusterID,
-				},
-			},
+			Type: mcsv1a1.ClusterSetIP,
 		},
 	}
 }
 
-func newClusterHeadlessServiceImport(namespace, name, clusterID string) *mcsv1a1.ServiceImport {
-	si := newClusterServiceImport(namespace, name, "", clusterID)
+func newHeadlessAggregatedServiceImport(namespace, name string) *mcsv1a1.ServiceImport {
+	si := newAggregatedServiceImport(namespace, name)
 	si.Spec.Type = mcsv1a1.Headless
 
 	return si
@@ -296,13 +278,14 @@ func newClusterHeadlessServiceImport(namespace, name, clusterID string) *mcsv1a1
 
 func newClusterIPEndpointSlice(namespace, name, clusterID, clusterIP string, isHealthy bool,
 	ports ...mcsv1a1.ServicePort) *discovery.EndpointSlice {
-	if isHealthy {
-		return newEndpointSlice(namespace, name, clusterID, ports, discovery.Endpoint{
-			Addresses: []string{clusterIP},
-		})
-	}
+	eps := newEndpointSlice(namespace, name, clusterID, ports, discovery.Endpoint{
+		Addresses:  []string{clusterIP},
+		Conditions: discovery.EndpointConditions{Ready: pointer.Bool(isHealthy)},
+	})
+
+	eps.Labels[constants.LabelIsHeadless] = "false"
 
-	return newEndpointSlice(namespace, name, clusterID, ports)
+	return eps
 }
 
 func newEndpointSlice(namespace, name, clusterID string, ports []mcsv1a1.ServicePort,
@@ -326,6 +309,7 @@ func newEndpointSlice(namespace, name, clusterID string, ports []mcsv1a1.Service
 				constants.LabelSourceNamespace:  namespace,
 				constants.MCSLabelSourceCluster: clusterID,
 				mcsv1a1.LabelServiceName:        name,
+				constants.LabelIsHeadless:       "true",
 			},
 		},
 		AddressType: discovery.AddressTypeIPv4,
diff --git a/coredns/resolver/service_import_migration_test.go b/coredns/resolver/service_import_migration_test.go
new file mode 100644
index 0000000..d3cef48
--- /dev/null
+++ b/coredns/resolver/service_import_migration_test.go
@@ -0,0 +1,138 @@
+/*
+SPDX-License-Identifier: Apache-2.0
+
+Copyright Contributors to the Submariner project.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package resolver_test
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	"github.com/submariner-io/lighthouse/coredns/constants"
+	"github.com/submariner-io/lighthouse/coredns/resolver"
+	discovery "k8s.io/api/discovery/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
+	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
+)
+
+var _ = Describe("ServiceImport migration", func() {
+	t := newTestDriver()
+
+	var (
+		legacyServiceImport *mcsv1a1.ServiceImport
+		legacyEndpointSlice *discovery.EndpointSlice
+	)
+
+	cluster1DNSRecord := resolver.DNSRecord{
+		IP:          serviceIP1,
+		Ports:       []mcsv1a1.ServicePort{port1},
+		ClusterName: clusterID1,
+	}
+
+	cluster2DNSRecord := resolver.DNSRecord{
+		IP:          serviceIP2,
+		Ports:       []mcsv1a1.ServicePort{port1},
+		ClusterName: clusterID2,
+	}
+
+	BeforeEach(func() {
+		legacyServiceImport = newLegacyServiceImport(namespace1, service1, serviceIP1, clusterID1, port1)
+		t.resolver.PutServiceImport(legacyServiceImport)
+
+		legacyEndpointSlice = newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true, port1)
+		legacyEndpointSlice.Endpoints = []discovery.Endpoint{{
+			Addresses:  []string{"1.2.3.4"},
+			Conditions: discovery.EndpointConditions{Ready: pointer.Bool(true)},
+		}}
+
+		delete(legacyEndpointSlice.Labels, constants.LabelIsHeadless)
+		t.resolver.PutEndpointSlice(legacyEndpointSlice)
+	})
+
+	When("a legacy per-cluster ServiceImport and EndpointSlice are created", func() {
+		Specify("GetDNSRecords should return the cluster's DNS record when requested", func() {
+			t.assertDNSRecordsFound(namespace1, service1, clusterID1, "", false, cluster1DNSRecord)
+		})
+
+		Context("and subsequently deleted", func() {
+			Specify("GetDNSRecords should return not found after the EndpointSlice is deleted", func() {
+				t.awaitDNSRecords(namespace1, service1, clusterID1, "", true)
+
+				t.resolver.RemoveServiceImport(legacyServiceImport)
+				t.awaitDNSRecords(namespace1, service1, clusterID1, "", true)
+
+				t.resolver.RemoveEndpointSlice(legacyEndpointSlice)
+				t.awaitDNSRecords(namespace1, service1, clusterID1, "", false)
+			})
+		})
+	})
+
+	Context("with a mix of upgraded and legacy cluster resources", func() {
+		BeforeEach(func() {
+			t.resolver.PutServiceImport(newAggregatedServiceImport(namespace1, service1))
+
+			t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID2, serviceIP2, true, port1))
+		})
+
+		Specify("GetDNSRecords should return the correct DNS records before and after the legacy cluster is upgraded", func() {
+			t.testRoundRobin(namespace1, service1, serviceIP1, serviceIP2)
+			t.assertDNSRecordsFound(namespace1, service1, clusterID1, "", false, cluster1DNSRecord)
+			t.assertDNSRecordsFound(namespace1, service1, clusterID2, "", false, cluster2DNSRecord)
+
+			t.resolver.RemoveServiceImport(legacyServiceImport)
+			t.putEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true, port1))
+
+			t.testRoundRobin(namespace1, service1, serviceIP1, serviceIP2)
+			t.assertDNSRecordsFound(namespace1, service1, clusterID1, "", false, cluster1DNSRecord)
+			t.assertDNSRecordsFound(namespace1, service1, clusterID2, "", false, cluster2DNSRecord)
+		})
+	})
+})
+
+func newLegacyServiceImport(namespace, name, serviceIP, clusterID string, ports ...mcsv1a1.ServicePort) *mcsv1a1.ServiceImport {
+	var ips []string
+	if serviceIP != "" {
+		ips = []string{serviceIP}
+	}
+
+	return &mcsv1a1.ServiceImport{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name + "-" + namespace + "-" + clusterID,
+			Namespace: submarinerNamespace,
+			Annotations: map[string]string{
+				"origin-name":      name,
+				"origin-namespace": namespace,
+			},
+			Labels: map[string]string{
+				"lighthouse.submariner.io/sourceName":    name,
+				constants.LabelSourceNamespace:           namespace,
+				"lighthouse.submariner.io/sourceCluster": clusterID,
+			},
+		},
+		Spec: mcsv1a1.ServiceImportSpec{
+			Type:  mcsv1a1.ClusterSetIP,
+			IPs:   ips,
+			Ports: ports,
+		},
+		Status: mcsv1a1.ServiceImportStatus{
+			Clusters: []mcsv1a1.ClusterStatus{
+				{
+					Cluster: clusterID,
+				},
+			},
+		},
+	}
+}
diff --git a/pkg/agent/controller/cleanup_test.go b/pkg/agent/controller/cleanup_test.go
index 8382b0f..bf993cd 100644
--- a/pkg/agent/controller/cleanup_test.go
+++ b/pkg/agent/controller/cleanup_test.go
@@ -19,30 +19,39 @@ limitations under the License.
 package controller_test
 
 import (
-	"time"
+	"context"
+	"fmt"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"github.com/submariner-io/admiral/pkg/syncer/test"
 	"github.com/submariner-io/lighthouse/pkg/constants"
 	discovery "k8s.io/api/discovery/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/kubernetes/scheme"
 	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
 var _ = Describe("Cleanup", func() {
 	var (
-		t                                    *testDriver
-		existingLocalServiceImport           *mcsv1a1.ServiceImport
-		existingRemoteServiceImport          *mcsv1a1.ServiceImport
-		existingLocalServiceImportInRemoteNS *mcsv1a1.ServiceImport
-		existingLocalEndpointSlice           *discovery.EndpointSlice
-		existingLocalEndpointSliceInRemoteNS *discovery.EndpointSlice
-		existingRemoteEndpointSlice          *discovery.EndpointSlice
-		nonLHEndpointSlice                   *discovery.EndpointSlice
-		remoteNSServiceImportClient          dynamic.ResourceInterface
-		remoteNSEndpointSliceClient          dynamic.ResourceInterface
+		t                                          *testDriver
+		localServiceImport1                        *mcsv1a1.ServiceImport
+		localAggregatedServiceImport1              *mcsv1a1.ServiceImport
+		aggregatedServiceImportOnRemoteBroker1     *mcsv1a1.ServiceImport
+		localServiceImport2                        *mcsv1a1.ServiceImport
+		localAggregatedServiceImport2              *mcsv1a1.ServiceImport
+		aggregatedServiceImportOnRemoteBroker2     *mcsv1a1.ServiceImport
+		remoteAggregatedServiceImportOnLocalBroker *mcsv1a1.ServiceImport
+		localEndpointSlice                         *discovery.EndpointSlice
+		remoteEndpointSliceOnLocalBroker           *discovery.EndpointSlice
+		remoteEndpointSliceInLocalNS               *discovery.EndpointSlice
+		nonLHEndpointSlice                         *discovery.EndpointSlice
+		localBrokerServiceImportClient             dynamic.ResourceInterface
+		localBrokerEndpointSliceClient             dynamic.ResourceInterface
 	)
 
 	BeforeEach(func() {
@@ -53,47 +62,117 @@ var _ = Describe("Cleanup", func() {
 	JustBeforeEach(func() {
 		t.justBeforeEach()
 
-		existingLocalServiceImport = &mcsv1a1.ServiceImport{
+		// Exported ServiceImport in one cluster
+
+		localServiceImport1 = &mcsv1a1.ServiceImport{
 			ObjectMeta: metav1.ObjectMeta{
-				Name: "nginx-" + serviceNamespace + "-" + clusterID1,
+				Name: fmt.Sprintf("%s-%s-%s", serviceName, serviceNamespace, clusterID1),
 				Labels: map[string]string{
+					mcsv1a1.LabelServiceName:        serviceName,
+					constants.LabelSourceNamespace:  serviceNamespace,
 					constants.MCSLabelSourceCluster: clusterID1,
 				},
 			},
 		}
 
-		test.CreateResource(t.cluster1.localServiceImportClient, existingLocalServiceImport)
-		test.CreateResource(t.brokerServiceImportClient, test.SetClusterIDLabel(existingLocalServiceImport, clusterID1))
+		test.CreateResource(t.cluster1.localServiceImportClient.Namespace(test.LocalNamespace), localServiceImport1)
 
-		remoteNSServiceImportClient = t.cluster1.localDynClient.Resource(*test.GetGroupVersionResourceFor(t.syncerConfig.RestMapper,
-			&mcsv1a1.ServiceImport{})).Namespace(test.RemoteNamespace)
+		localAggregatedServiceImport1 = &mcsv1a1.ServiceImport{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: serviceName,
+			},
+			Status: mcsv1a1.ServiceImportStatus{Clusters: []mcsv1a1.ClusterStatus{
+				{
+					Cluster: clusterID1,
+				},
+			}},
+		}
 
-		existingLocalServiceImportInRemoteNS = &mcsv1a1.ServiceImport{
+		test.CreateResource(t.cluster1.localServiceImportClient.Namespace(serviceNamespace), localAggregatedServiceImport1)
+
+		aggregatedServiceImportOnRemoteBroker1 = &mcsv1a1.ServiceImport{
 			ObjectMeta: metav1.ObjectMeta{
-				Name: "other-" + serviceNamespace + "-" + clusterID2,
-				Labels: map[string]string{
-					constants.MCSLabelSourceCluster: clusterID2,
+				Name: fmt.Sprintf("%s-%s", serviceName, serviceNamespace),
+				Annotations: map[string]string{
+					mcsv1a1.LabelServiceName:       serviceName,
+					constants.LabelSourceNamespace: serviceNamespace,
 				},
 			},
 		}
+		aggregatedServiceImportOnRemoteBroker1.Status = localAggregatedServiceImport1.Status
 
-		test.CreateResource(remoteNSServiceImportClient, test.SetClusterIDLabel(existingLocalServiceImportInRemoteNS, clusterID2))
+		test.CreateResource(t.brokerServiceImportClient.Namespace(test.RemoteNamespace), aggregatedServiceImportOnRemoteBroker1)
 
-		existingRemoteServiceImport = &mcsv1a1.ServiceImport{
+		// Exported ServiceImport in two clusters
+
+		serviceNamespace2 := "ns2"
+
+		localServiceImport2 = &mcsv1a1.ServiceImport{
 			ObjectMeta: metav1.ObjectMeta{
-				Name: "nginx2-" + serviceNamespace + "-" + clusterID2,
+				Name: fmt.Sprintf("%s-%s-%s", serviceName, serviceNamespace2, clusterID1),
 				Labels: map[string]string{
-					constants.MCSLabelSourceCluster: clusterID2,
+					mcsv1a1.LabelServiceName:        serviceName,
+					constants.LabelSourceNamespace:  serviceNamespace2,
+					constants.MCSLabelSourceCluster: clusterID1,
+				},
+			},
+		}
+
+		test.CreateResource(t.cluster1.localServiceImportClient.Namespace(test.LocalNamespace), localServiceImport2)
+
+		localAggregatedServiceImport2 = &mcsv1a1.ServiceImport{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: serviceName,
+			},
+			Status: mcsv1a1.ServiceImportStatus{Clusters: []mcsv1a1.ClusterStatus{
+				{
+					Cluster: clusterID1,
+				},
+				{
+					Cluster: clusterID2,
+				},
+			}},
+		}
+
+		test.CreateResource(t.cluster1.localServiceImportClient.Namespace(serviceNamespace2), localAggregatedServiceImport2)
+
+		aggregatedServiceImportOnRemoteBroker2 = &mcsv1a1.ServiceImport{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: fmt.Sprintf("%s-%s", serviceName, serviceNamespace2),
+				Annotations: map[string]string{
+					mcsv1a1.LabelServiceName:       serviceName,
+					constants.LabelSourceNamespace: serviceNamespace2,
 				},
 			},
 		}
+		aggregatedServiceImportOnRemoteBroker2.Status = localAggregatedServiceImport2.Status
+
+		test.CreateResource(t.brokerServiceImportClient.Namespace(test.RemoteNamespace), aggregatedServiceImportOnRemoteBroker2)
+
+		// Remote ServiceImport in local broker
+
+		serviceNamespace3 := "ns3"
 
-		test.CreateResource(t.cluster1.localServiceImportClient, test.SetClusterIDLabel(existingRemoteServiceImport, clusterID2))
-		test.CreateResource(t.brokerServiceImportClient, test.SetClusterIDLabel(existingRemoteServiceImport, clusterID2))
+		localBrokerServiceImportClient = t.cluster1.localDynClient.Resource(*test.GetGroupVersionResourceFor(t.syncerConfig.RestMapper,
+			&mcsv1a1.ServiceImport{})).Namespace(test.RemoteNamespace)
 
-		existingLocalEndpointSlice = &discovery.EndpointSlice{
+		remoteAggregatedServiceImportOnLocalBroker = &mcsv1a1.ServiceImport{
 			ObjectMeta: metav1.ObjectMeta{
-				Name: "nginx-" + clusterID1,
+				Name: fmt.Sprintf("%s-%s", serviceName, serviceNamespace3),
+				Annotations: map[string]string{
+					mcsv1a1.LabelServiceName:       serviceName,
+					constants.LabelSourceNamespace: serviceNamespace3,
+				},
+			},
+		}
+
+		test.CreateResource(localBrokerServiceImportClient, remoteAggregatedServiceImportOnLocalBroker)
+
+		// Local EndpointSlice in remote broker
+
+		localEndpointSlice = &discovery.EndpointSlice{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "nginx-ns1" + clusterID1,
 				Labels: map[string]string{
 					constants.MCSLabelSourceCluster: clusterID1,
 					discovery.LabelManagedBy:        constants.LabelValueManagedBy,
@@ -101,15 +180,17 @@ var _ = Describe("Cleanup", func() {
 			},
 		}
 
-		test.CreateResource(t.cluster1.localEndpointSliceClient, existingLocalEndpointSlice)
-		test.CreateResource(t.brokerEndpointSliceClient, test.SetClusterIDLabel(existingLocalEndpointSlice, clusterID1))
+		test.CreateResource(t.cluster1.localEndpointSliceClient, localEndpointSlice)
+		test.CreateResource(t.brokerEndpointSliceClient, test.SetClusterIDLabel(localEndpointSlice, clusterID1))
+
+		// Remote EndpointSlice in local broker
 
-		remoteNSEndpointSliceClient = t.cluster1.localDynClient.Resource(*test.GetGroupVersionResourceFor(t.syncerConfig.RestMapper,
+		localBrokerEndpointSliceClient = t.cluster1.localDynClient.Resource(*test.GetGroupVersionResourceFor(t.syncerConfig.RestMapper,
 			&discovery.EndpointSlice{})).Namespace(test.RemoteNamespace)
 
-		existingLocalEndpointSliceInRemoteNS = &discovery.EndpointSlice{
+		remoteEndpointSliceOnLocalBroker = &discovery.EndpointSlice{
 			ObjectMeta: metav1.ObjectMeta{
-				Name: "other-" + clusterID2,
+				Name: "nginx-ns3" + clusterID2,
 				Labels: map[string]string{
 					constants.MCSLabelSourceCluster: clusterID2,
 					discovery.LabelManagedBy:        constants.LabelValueManagedBy,
@@ -117,9 +198,11 @@ var _ = Describe("Cleanup", func() {
 			},
 		}
 
-		test.CreateResource(remoteNSEndpointSliceClient, test.SetClusterIDLabel(existingLocalEndpointSliceInRemoteNS, clusterID2))
+		test.CreateResource(localBrokerEndpointSliceClient, test.SetClusterIDLabel(remoteEndpointSliceOnLocalBroker, clusterID2))
 
-		existingRemoteEndpointSlice = &discovery.EndpointSlice{
+		// Remote EndpointSlice in local namespace
+
+		remoteEndpointSliceInLocalNS = &discovery.EndpointSlice{
 			ObjectMeta: metav1.ObjectMeta{
 				Name: "nginx2-" + clusterID2,
 				Labels: map[string]string{
@@ -129,8 +212,7 @@ var _ = Describe("Cleanup", func() {
 			},
 		}
 
-		test.CreateResource(t.cluster1.localEndpointSliceClient, test.SetClusterIDLabel(existingRemoteEndpointSlice, clusterID2))
-		test.CreateResource(t.brokerEndpointSliceClient, test.SetClusterIDLabel(existingRemoteEndpointSlice, clusterID2))
+		test.CreateResource(t.cluster1.localEndpointSliceClient, remoteEndpointSliceInLocalNS)
 
 		nonLHEndpointSlice = &discovery.EndpointSlice{
 			ObjectMeta: metav1.ObjectMeta{
@@ -145,28 +227,41 @@ var _ = Describe("Cleanup", func() {
 		t.afterEach()
 	})
 
-	It("should remove local LH ServiceImports and EndpointSlices from the remote datastore", func() {
+	It("should correctly remove local and remote ServiceImports and EndpointSlices", func() {
 		Expect(t.cluster1.agentController.Cleanup()).To(Succeed())
 
-		test.AwaitNoResource(t.brokerServiceImportClient, existingLocalServiceImport.GetName())
-		test.AwaitNoResource(t.brokerEndpointSliceClient, existingLocalEndpointSlice.GetName())
+		test.AwaitNoResource(t.cluster1.localServiceImportClient.Namespace(test.LocalNamespace), localServiceImport1.Name)
+		test.AwaitNoResource(t.cluster1.localServiceImportClient.Namespace(serviceNamespace), localAggregatedServiceImport1.Name)
+		test.AwaitNoResource(t.brokerServiceImportClient.Namespace(test.RemoteNamespace), aggregatedServiceImportOnRemoteBroker1.Name)
 
-		time.Sleep(300 * time.Millisecond)
-		test.AwaitResource(t.brokerServiceImportClient, existingRemoteServiceImport.GetName())
-		test.AwaitResource(t.brokerEndpointSliceClient, existingRemoteEndpointSlice.GetName())
-	})
+		test.AwaitNoResource(t.cluster1.localServiceImportClient.Namespace(test.LocalNamespace), localServiceImport2.Name)
+		test.AwaitNoResource(t.cluster1.localServiceImportClient.Namespace(serviceNamespace), localAggregatedServiceImport2.Name)
 
-	It("should remove all LH ServiceImports and EndpointSlices from the local datastore", func() {
-		Expect(t.cluster1.agentController.Cleanup()).To(Succeed())
+		si := ensureResource(t.brokerServiceImportClient.Namespace(test.RemoteNamespace), aggregatedServiceImportOnRemoteBroker2.Name,
+			&mcsv1a1.ServiceImport{}).(*mcsv1a1.ServiceImport)
+		Expect(si.Status.Clusters).To(HaveLen(1))
+		Expect(si.Status.Clusters).To(ContainElement(mcsv1a1.ClusterStatus{Cluster: clusterID2}))
 
-		test.AwaitNoResource(t.cluster1.localServiceImportClient, existingLocalServiceImport.GetName())
-		test.AwaitNoResource(t.cluster1.localServiceImportClient, existingRemoteServiceImport.GetName())
-		test.AwaitNoResource(t.cluster1.localEndpointSliceClient, existingLocalEndpointSlice.GetName())
-		test.AwaitNoResource(t.cluster1.localEndpointSliceClient, existingRemoteEndpointSlice.GetName())
+		ensureResource(localBrokerServiceImportClient, remoteAggregatedServiceImportOnLocalBroker.Name, &mcsv1a1.ServiceImport{})
 
-		time.Sleep(300 * time.Millisecond)
-		test.AwaitResource(t.cluster1.localEndpointSliceClient, nonLHEndpointSlice.GetName())
-		test.AwaitResource(remoteNSServiceImportClient, existingLocalServiceImportInRemoteNS.GetName())
-		test.AwaitResource(remoteNSEndpointSliceClient, existingLocalEndpointSliceInRemoteNS.GetName())
+		test.AwaitNoResource(t.brokerEndpointSliceClient, localEndpointSlice.GetName())
+		test.AwaitNoResource(t.cluster1.localEndpointSliceClient, remoteEndpointSliceInLocalNS.Name)
+
+		ensureResource(localBrokerEndpointSliceClient, remoteEndpointSliceOnLocalBroker.Name, &discovery.EndpointSlice{})
+		ensureResource(t.cluster1.localEndpointSliceClient, nonLHEndpointSlice.Name, &discovery.EndpointSlice{})
 	})
 })
+
+func ensureResource(client dynamic.ResourceInterface, name string, to runtime.Object) runtime.Object {
+	Consistently(func() bool {
+		_, err := client.Get(context.TODO(), name, metav1.GetOptions{})
+		return apierrors.IsNotFound(err)
+	}).Should(BeFalse(), "Expected resource %q not found", name)
+
+	obj, err := client.Get(context.TODO(), name, metav1.GetOptions{})
+	Expect(err).To(Succeed())
+
+	utilruntime.Must(scheme.Scheme.Convert(obj, to, nil))
+
+	return to
+}
diff --git a/pkg/agent/controller/clusterip_service_test.go b/pkg/agent/controller/clusterip_service_test.go
new file mode 100644
index 0000000..7277747
--- /dev/null
+++ b/pkg/agent/controller/clusterip_service_test.go
@@ -0,0 +1,249 @@
+/*
+SPDX-License-Identifier: Apache-2.0
+
+Copyright Contributors to the Submariner project.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller_test
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	"github.com/submariner-io/admiral/pkg/resource"
+	"github.com/submariner-io/admiral/pkg/syncer/test"
+	testutil "github.com/submariner-io/admiral/pkg/test"
+	"github.com/submariner-io/lighthouse/pkg/constants"
+	corev1 "k8s.io/api/core/v1"
+	discovery "k8s.io/api/discovery/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
+	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
+)
+
+var _ = Describe("ClusterIP Service export", func() {
+	Describe("Single cluster", testClusterIPServiceInOneCluster)
+})
+
+func testClusterIPServiceInOneCluster() {
+	var t *testDriver
+
+	BeforeEach(func() {
+		t = newTestDiver()
+
+		t.cluster1.createEndpoints()
+	})
+
+	JustBeforeEach(func() {
+		t.justBeforeEach()
+	})
+
+	AfterEach(func() {
+		t.afterEach()
+	})
+
+	When("a ServiceExport is created", func() {
+		Context("and the Service already exists", func() {
+			It("should export the service and update the ServiceExport status", func() {
+				t.cluster1.createService()
+				t.cluster1.createServiceExport()
+				t.awaitNonHeadlessServiceExported(&t.cluster1)
+			})
+		})
+
+		Context("and the Service doesn't initially exist", func() {
+			It("should eventually export the service", func() {
+				t.cluster1.createServiceExport()
+				t.cluster1.awaitServiceUnavailableStatus()
+
+				t.cluster1.createService()
+				t.awaitNonHeadlessServiceExported(&t.cluster1)
+			})
+		})
+	})
+
+	When("a ServiceExport is deleted after the service is exported", func() {
+		It("should unexport the service", func() {
+			t.cluster1.createService()
+			t.cluster1.createServiceExport()
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+
+			t.cluster1.deleteServiceExport()
+			t.awaitServiceUnexported(&t.cluster1)
+		})
+	})
+
+	When("an exported Service is deleted and recreated while the ServiceExport still exists", func() {
+		It("should unexport and re-export the service", func() {
+			t.cluster1.createService()
+			t.cluster1.createServiceExport()
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+			t.cluster1.localDynClient.Fake.ClearActions()
+
+			t.cluster1.deleteService()
+			t.cluster1.awaitServiceUnavailableStatus()
+			t.cluster1.awaitServiceExportCondition(newServiceExportSyncedCondition(corev1.ConditionFalse, "NoServiceImport"))
+			t.awaitServiceUnexported(&t.cluster1)
+
+			t.cluster1.createService()
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+		})
+	})
+
+	When("the type of an exported Service is updated to an unsupported type", func() {
+		It("should unexport the ServiceImport and update the ServiceExport status appropriately", func() {
+			t.cluster1.createService()
+			t.cluster1.createServiceExport()
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+
+			t.cluster1.service.Spec.Type = corev1.ServiceTypeNodePort
+			t.cluster1.updateService()
+
+			t.cluster1.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "UnsupportedServiceType"))
+			t.cluster1.awaitServiceExportCondition(newServiceExportSyncedCondition(corev1.ConditionFalse, "NoServiceImport"))
+			t.awaitServiceUnexported(&t.cluster1)
+		})
+	})
+
+	When("a ServiceExport is created for a Service whose type is unsupported", func() {
+		BeforeEach(func() {
+			t.cluster1.service.Spec.Type = corev1.ServiceTypeNodePort
+		})
+
+		JustBeforeEach(func() {
+			t.cluster1.createService()
+			t.cluster1.createServiceExport()
+		})
+
+		It("should update the ServiceExport status appropriately and not export the serviceImport", func() {
+			t.cluster1.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "UnsupportedServiceType"))
+			t.cluster1.ensureNoServiceExportCondition(constants.ServiceExportSynced)
+		})
+
+		Context("and is subsequently updated to a supported type", func() {
+			It("should eventually export the service and update the ServiceExport status appropriately", func() {
+				t.cluster1.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "UnsupportedServiceType"))
+
+				t.cluster1.service.Spec.Type = corev1.ServiceTypeClusterIP
+				t.cluster1.updateService()
+
+				t.awaitNonHeadlessServiceExported(&t.cluster1)
+			})
+		})
+	})
+
+	When("the backend Endpoints has no ready addresses", func() {
+		JustBeforeEach(func() {
+			t.cluster1.createService()
+			t.cluster1.createServiceExport()
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+		})
+
+		Specify("the EndpointSlice's service IP address should indicate not ready", func() {
+			t.cluster1.endpoints.Subsets[0].Addresses = nil
+
+			t.cluster1.updateEndpoints()
+			t.awaitEndpointSlice(&t.cluster1)
+		})
+	})
+
+	When("two Services with the same name in different namespaces are exported", func() {
+		It("should correctly export both services", func() {
+			t.cluster1.createService()
+			t.cluster1.createServiceExport()
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+
+			service := &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      t.cluster1.service.Name,
+					Namespace: "other-service-ns",
+				},
+				Spec: corev1.ServiceSpec{
+					ClusterIP: "10.253.9.2",
+				},
+			}
+
+			serviceExport := &mcsv1a1.ServiceExport{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      service.Name,
+					Namespace: service.Namespace,
+				},
+			}
+
+			endpoints := &corev1.Endpoints{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      service.Name,
+					Namespace: service.Namespace,
+				},
+			}
+
+			expServiceImport := &mcsv1a1.ServiceImport{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      service.Name,
+					Namespace: service.Namespace,
+				},
+				Spec: mcsv1a1.ServiceImportSpec{
+					Type:  mcsv1a1.ClusterSetIP,
+					Ports: []mcsv1a1.ServicePort{},
+				},
+				Status: mcsv1a1.ServiceImportStatus{
+					Clusters: []mcsv1a1.ClusterStatus{
+						{
+							Cluster: t.cluster1.clusterID,
+						},
+					},
+				},
+			}
+
+			expEndpointSlice := &discovery.EndpointSlice{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      service.Name,
+					Namespace: service.Namespace,
+					Labels: map[string]string{
+						discovery.LabelManagedBy:        constants.LabelValueManagedBy,
+						constants.MCSLabelSourceCluster: t.cluster1.clusterID,
+						mcsv1a1.LabelServiceName:        service.Name,
+						constants.LabelSourceNamespace:  service.Namespace,
+					},
+				},
+				AddressType: discovery.AddressTypeIPv4,
+				Endpoints: []discovery.Endpoint{
+					{
+						Addresses:  []string{service.Spec.ClusterIP},
+						Conditions: discovery.EndpointConditions{Ready: pointer.Bool(false)},
+					},
+				},
+			}
+
+			test.CreateResource(endpointsClientFor(t.cluster1.localDynClient, endpoints.Namespace), endpoints)
+			test.CreateResource(t.cluster1.dynamicServiceClientFor().Namespace(service.Namespace), service)
+			test.CreateResource(serviceExportClientFor(t.cluster1.localDynClient, service.Namespace), serviceExport)
+
+			awaitServiceImport(t.cluster2.localServiceImportClient, expServiceImport)
+			awaitEndpointSlice(endpointSliceClientFor(t.cluster2.localDynClient, endpoints.Namespace), expEndpointSlice)
+
+			// Ensure the resources for the first Service weren't overwritten
+			t.awaitAggregatedServiceImport(mcsv1a1.ClusterSetIP, t.cluster1.service.Name, t.cluster1.service.Namespace, &t.cluster1)
+		})
+	})
+
+	Specify("an EndpointSlice not managed by Lighthouse should not be synced to the broker", func() {
+		test.CreateResource(endpointSliceClientFor(t.cluster1.localDynClient, t.cluster1.service.Namespace),
+			&discovery.EndpointSlice{ObjectMeta: metav1.ObjectMeta{
+				Name:   "other-eps",
+				Labels: map[string]string{discovery.LabelManagedBy: "other"},
+			}})
+
+		testutil.EnsureNoResource(resource.ForDynamic(endpointSliceClientFor(t.syncerConfig.BrokerClient, test.RemoteNamespace)), "other-eps")
+	})
+}
diff --git a/pkg/agent/controller/controller_suite_test.go b/pkg/agent/controller/controller_suite_test.go
index ffee3c5..e4ae57e 100644
--- a/pkg/agent/controller/controller_suite_test.go
+++ b/pkg/agent/controller/controller_suite_test.go
@@ -27,7 +27,7 @@ import (
 	"fmt"
 	"math/big"
 	"reflect"
-	"sort"
+	"strconv"
 	"testing"
 	"time"
 
@@ -37,35 +37,55 @@ import (
 	"github.com/submariner-io/admiral/pkg/log/kzerolog"
 	"github.com/submariner-io/admiral/pkg/syncer/broker"
 	"github.com/submariner-io/admiral/pkg/syncer/test"
+	testutil "github.com/submariner-io/admiral/pkg/test"
 	"github.com/submariner-io/lighthouse/pkg/agent/controller"
 	"github.com/submariner-io/lighthouse/pkg/constants"
 	corev1 "k8s.io/api/core/v1"
 	discovery "k8s.io/api/discovery/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/dynamic"
+	dynamicfake "k8s.io/client-go/dynamic/fake"
 	"k8s.io/client-go/kubernetes/scheme"
-	"k8s.io/client-go/tools/cache"
+	k8stesting "k8s.io/client-go/testing"
+	"k8s.io/utils/pointer"
 	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
 const (
 	clusterID1       = "east"
 	clusterID2       = "west"
+	serviceName      = "nginx"
 	serviceNamespace = "service-ns"
 	globalIP1        = "242.254.1.1"
 	globalIP2        = "242.254.1.2"
-	globalIP3        = "242.254.1.3"
 )
 
 var (
 	nodeName = "my-node"
 	hostName = "my-host"
-	ready    = true
+
+	port1 = mcsv1a1.ServicePort{
+		Name:     "http",
+		Protocol: corev1.ProtocolTCP,
+		Port:     8080,
+	}
+
+	port2 = mcsv1a1.ServicePort{
+		Name:     "https",
+		Protocol: corev1.ProtocolTCP,
+		Port:     8443,
+	}
+
+	port3 = mcsv1a1.ServicePort{
+		Name:     "POP3",
+		Protocol: corev1.ProtocolUDP,
+		Port:     110,
+	}
 )
 
 func init() {
@@ -88,28 +108,32 @@ func TestController(t *testing.T) {
 }
 
 type cluster struct {
-	agentSpec                controller.AgentSpecification
-	localDynClient           dynamic.Interface
-	localServiceExportClient *fake.DynamicResourceClient
-	localServiceImportClient *fake.DynamicResourceClient
-	localIngressIPClient     *fake.DynamicResourceClient
-	localEndpointSliceClient dynamic.ResourceInterface
-	agentController          *controller.Controller
-	chanByCondType           map[mcsv1a1.ServiceExportConditionType]chan mcsv1a1.ServiceExportCondition
-}
-
-type testDriver struct {
-	cluster1                  cluster
-	cluster2                  cluster
-	brokerServiceImportClient *fake.DynamicResourceClient
-	brokerEndpointSliceClient *fake.DynamicResourceClient
+	agentSpec                 controller.AgentSpecification
+	localDynClient            *dynamicfake.FakeDynamicClient
+	localServiceExportClient  dynamic.ResourceInterface
+	localServiceImportClient  dynamic.NamespaceableResourceInterface
+	localIngressIPClient      dynamic.ResourceInterface
+	localEndpointSliceClient  dynamic.ResourceInterface
+	localServiceImportReactor *fake.FailingReactor
+	agentController           *controller.Controller
 	service                   *corev1.Service
+	serviceIP                 string
 	serviceExport             *mcsv1a1.ServiceExport
 	endpoints                 *corev1.Endpoints
-	stopCh                    chan struct{}
-	syncerConfig              *broker.SyncerConfig
-	endpointGlobalIPs         []string
-	doStart                   bool
+	clusterID                 string
+	endpointSliceAddresses    []discovery.Endpoint
+}
+
+type testDriver struct {
+	cluster1                   cluster
+	cluster2                   cluster
+	brokerServiceImportClient  dynamic.NamespaceableResourceInterface
+	brokerEndpointSliceClient  dynamic.ResourceInterface
+	stopCh                     chan struct{}
+	syncerConfig               *broker.SyncerConfig
+	doStart                    bool
+	brokerServiceImportReactor *fake.FailingReactor
+	aggregatedServicePorts     []mcsv1a1.ServicePort
 }
 
 func newTestDiver() *testDriver {
@@ -124,97 +148,158 @@ func newTestDiver() *testDriver {
 		Kind:    "GlobalIngressIPList",
 	}, &unstructured.UnstructuredList{})
 
+	brokerClient := dynamicfake.NewSimpleDynamicClient(syncerScheme)
+	fake.AddFilteringListReactor(&brokerClient.Fake)
+
 	t := &testDriver{
+		aggregatedServicePorts: []mcsv1a1.ServicePort{},
 		cluster1: cluster{
+			clusterID: clusterID1,
 			agentSpec: controller.AgentSpecification{
 				ClusterID:        clusterID1,
 				Namespace:        test.LocalNamespace,
 				GlobalnetEnabled: false,
 			},
+			service: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      serviceName,
+					Namespace: serviceNamespace,
+				},
+				Spec: corev1.ServiceSpec{
+					ClusterIP: "10.253.9.1",
+					Selector:  map[string]string{"app": "test"},
+					Ports:     []corev1.ServicePort{toServicePort(port1), toServicePort(port2)},
+				},
+			},
+			serviceExport: &mcsv1a1.ServiceExport{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      serviceName,
+					Namespace: serviceNamespace,
+				},
+			},
+			endpoints: &corev1.Endpoints{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      serviceName,
+					Namespace: serviceNamespace,
+					Labels:    map[string]string{"app": "test"},
+				},
+				Subsets: []corev1.EndpointSubset{
+					{
+						Ports: []corev1.EndpointPort{
+							{
+								Name:     port1.Name,
+								Protocol: port1.Protocol,
+								Port:     80,
+							},
+							{
+								Name:     port2.Name,
+								Protocol: port2.Protocol,
+								Port:     443,
+							},
+						},
+						Addresses: []corev1.EndpointAddress{
+							{
+								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",
+								},
+							},
+						},
+					},
+				},
+			},
 		},
 		cluster2: cluster{
+			clusterID: clusterID2,
 			agentSpec: controller.AgentSpecification{
 				ClusterID:        clusterID2,
 				Namespace:        test.LocalNamespace,
 				GlobalnetEnabled: false,
 			},
-		},
-		service: &corev1.Service{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:      "nginx",
-				Namespace: serviceNamespace,
+			service: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      serviceName,
+					Namespace: serviceNamespace,
+				},
+				Spec: corev1.ServiceSpec{
+					ClusterIP: "10.253.10.1",
+					Selector:  map[string]string{"app": "test"},
+				},
 			},
-			Spec: corev1.ServiceSpec{
-				ClusterIP: "10.253.9.1",
-				Selector:  map[string]string{"app": "test"},
+			endpoints: &corev1.Endpoints{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      serviceName,
+					Namespace: serviceNamespace,
+					Labels:    map[string]string{"app": "test"},
+				},
+				Subsets: []corev1.EndpointSubset{
+					{
+						Addresses: []corev1.EndpointAddress{
+							{
+								IP:       "192.168.5.3",
+								Hostname: hostName,
+							},
+						},
+					},
+				},
 			},
 		},
 		syncerConfig: &broker.SyncerConfig{
 			BrokerNamespace: test.RemoteNamespace,
 			RestMapper: test.GetRESTMapperFor(&mcsv1a1.ServiceExport{}, &mcsv1a1.ServiceImport{}, &corev1.Service{},
 				&corev1.Endpoints{}, &discovery.EndpointSlice{}, controller.GetGlobalIngressIPObj()),
-			BrokerClient: fake.NewDynamicClient(syncerScheme),
+			BrokerClient: brokerClient,
 			Scheme:       syncerScheme,
 		},
 		stopCh:  make(chan struct{}),
 		doStart: true,
 	}
 
-	t.serviceExport = &mcsv1a1.ServiceExport{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      t.service.Name,
-			Namespace: t.service.Namespace,
+	t.brokerServiceImportReactor = fake.NewFailingReactorForResource(&brokerClient.Fake, "serviceimports")
+
+	t.cluster1.endpointSliceAddresses = []discovery.Endpoint{
+		{
+			Addresses:  []string{t.cluster1.endpoints.Subsets[0].Addresses[0].IP},
+			Conditions: discovery.EndpointConditions{Ready: pointer.Bool(true)},
+			Hostname:   &t.cluster1.endpoints.Subsets[0].Addresses[0].Hostname,
+		},
+		{
+			Addresses:  []string{t.cluster1.endpoints.Subsets[0].Addresses[1].IP},
+			Conditions: discovery.EndpointConditions{Ready: pointer.Bool(true)},
+			Hostname:   &t.cluster1.endpoints.Subsets[0].Addresses[1].TargetRef.Name,
+			NodeName:   t.cluster1.endpoints.Subsets[0].Addresses[1].NodeName,
 		},
 	}
 
-	t.endpoints = &corev1.Endpoints{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      t.service.Name,
-			Namespace: t.service.Namespace,
-			Labels:    map[string]string{"app": "test"},
-		},
-		Subsets: []corev1.EndpointSubset{
-			{
-				Addresses: []corev1.EndpointAddress{
-					{
-						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{
-					{
-						Name:     "port-1",
-						Protocol: corev1.ProtocolTCP,
-						Port:     1234,
-					},
-				},
-			},
+	t.cluster2.endpointSliceAddresses = []discovery.Endpoint{
+		{
+			Addresses:  []string{t.cluster2.endpoints.Subsets[0].Addresses[0].IP},
+			Conditions: discovery.EndpointConditions{Ready: pointer.Bool(true)},
+			Hostname:   &t.cluster2.endpoints.Subsets[0].Addresses[0].Hostname,
 		},
 	}
 
 	t.brokerServiceImportClient = t.syncerConfig.BrokerClient.Resource(*test.GetGroupVersionResourceFor(t.syncerConfig.RestMapper,
-		&mcsv1a1.ServiceImport{})).Namespace(test.RemoteNamespace).(*fake.DynamicResourceClient)
+		&mcsv1a1.ServiceImport{}))
 
 	t.brokerEndpointSliceClient = t.syncerConfig.BrokerClient.Resource(*test.GetGroupVersionResourceFor(t.syncerConfig.RestMapper,
-		&discovery.EndpointSlice{})).Namespace(test.RemoteNamespace).(*fake.DynamicResourceClient)
+		&discovery.EndpointSlice{})).Namespace(test.RemoteNamespace)
 
 	t.cluster1.init(t.syncerConfig)
 	t.cluster2.init(t.syncerConfig)
@@ -222,32 +307,6 @@ func newTestDiver() *testDriver {
 	return t
 }
 
-func (t *testDriver) newGlobalIngressIP(name, ip string) *unstructured.Unstructured {
-	ingressIP := controller.GetGlobalIngressIPObj()
-	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, ip)
-	setIngressIPConditions(ingressIP, metav1.Condition{
-		Type:    "Allocated",
-		Status:  metav1.ConditionTrue,
-		Reason:  "Success",
-		Message: "Allocated global IP",
-	})
-
-	return ingressIP
-}
-
-func (t *testDriver) newHeadlessGlobalIngressIP(name, ip string) *unstructured.Unstructured {
-	ingressIP := t.newGlobalIngressIP("pod"+"-"+name, ip)
-	Expect(unstructured.SetNestedField(ingressIP.Object, controller.HeadlessServicePod, "spec", "target")).To(Succeed())
-	Expect(unstructured.SetNestedField(ingressIP.Object, name, "spec", "podRef", "name")).To(Succeed())
-
-	return ingressIP
-}
-
 func (t *testDriver) justBeforeEach() {
 	t.cluster1.start(t, *t.syncerConfig)
 	t.cluster2.start(t, *t.syncerConfig)
@@ -258,76 +317,32 @@ func (t *testDriver) afterEach() {
 }
 
 func (c *cluster) init(syncerConfig *broker.SyncerConfig) {
-	c.localDynClient = fake.NewDynamicClient(syncerConfig.Scheme)
+	c.serviceIP = c.service.Spec.ClusterIP
 
-	fake.AddDeleteCollectionReactor(&c.localDynClient.(*fake.DynamicClient).Fake,
+	c.localDynClient = dynamicfake.NewSimpleDynamicClient(syncerConfig.Scheme)
+	fake.AddFilteringListReactor(&c.localDynClient.Fake)
+	fake.NewWatchReactor(&c.localDynClient.Fake)
+
+	c.localServiceImportReactor = fake.NewFailingReactorForResource(&c.localDynClient.Fake, "serviceimports")
+
+	fake.AddDeleteCollectionReactor(&c.localDynClient.Fake,
 		discovery.SchemeGroupVersion.WithKind("EndpointSlice"))
 
 	c.localServiceExportClient = c.localDynClient.Resource(*test.GetGroupVersionResourceFor(syncerConfig.RestMapper,
-		&mcsv1a1.ServiceExport{})).Namespace(serviceNamespace).(*fake.DynamicResourceClient)
+		&mcsv1a1.ServiceExport{})).Namespace(serviceNamespace)
 
 	c.localServiceImportClient = c.localDynClient.Resource(*test.GetGroupVersionResourceFor(syncerConfig.RestMapper,
-		&mcsv1a1.ServiceImport{})).Namespace(test.LocalNamespace).(*fake.DynamicResourceClient)
+		&mcsv1a1.ServiceImport{}))
 
 	c.localEndpointSliceClient = c.localDynClient.Resource(*test.GetGroupVersionResourceFor(syncerConfig.RestMapper,
 		&discovery.EndpointSlice{})).Namespace(serviceNamespace)
 
 	c.localIngressIPClient = c.localDynClient.Resource(*test.GetGroupVersionResourceFor(syncerConfig.RestMapper,
-		controller.GetGlobalIngressIPObj())).Namespace(serviceNamespace).(*fake.DynamicResourceClient)
-
-	c.chanByCondType = map[mcsv1a1.ServiceExportConditionType]chan mcsv1a1.ServiceExportCondition{}
-	c.chanByCondType[mcsv1a1.ServiceExportValid] = make(chan mcsv1a1.ServiceExportCondition, 100)
-	c.chanByCondType[constants.ServiceExportSynced] = make(chan mcsv1a1.ServiceExportCondition, 100)
+		controller.GetGlobalIngressIPObj())).Namespace(serviceNamespace)
 }
 
 //nolint:gocritic // (hugeParam) This function modifies syncerConf so we don't want to pass by pointer.
 func (c *cluster) start(t *testDriver, syncerConfig broker.SyncerConfig) {
-	processServiceExportConditions := func(oldObj, newObj interface{}) {
-		defer GinkgoRecover()
-
-		var oldSE *mcsv1a1.ServiceExport
-		if oldObj != nil {
-			oldSE = toServiceExport(oldObj)
-		}
-
-		newSE := toServiceExport(newObj)
-
-		for i := range newSE.Status.Conditions {
-			newCond := &newSE.Status.Conditions[i]
-
-			condChan := c.chanByCondType[newCond.Type]
-			if condChan == nil {
-				continue
-			}
-
-			if oldSE != nil {
-				prevCond := controller.FindServiceExportStatusCondition(oldSE.Status.Conditions, newCond.Type)
-				if prevCond != nil && reflect.DeepEqual(prevCond, newCond) {
-					continue
-				}
-			}
-
-			condChan <- *newCond
-		}
-	}
-
-	_, informer := cache.NewInformer(&cache.ListWatch{
-		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
-			return t.cluster1.localServiceExportClient.List(context.TODO(), options)
-		},
-		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
-			return t.cluster1.localServiceExportClient.Watch(context.TODO(), options)
-		},
-	}, &unstructured.Unstructured{}, 0, cache.ResourceEventHandlerFuncs{
-		AddFunc: func(obj interface{}) {
-			processServiceExportConditions(nil, obj)
-		},
-		UpdateFunc: processServiceExportConditions,
-	})
-
-	go informer.Run(t.stopCh)
-	Expect(cache.WaitForCacheSync(t.stopCh, informer.HasSynced)).To(BeTrue())
-
 	syncerConfig.LocalClient = c.localDynClient
 	bigint, err := rand.Int(rand.Reader, big.NewInt(1000000))
 	Expect(err).To(Succeed())
@@ -352,258 +367,338 @@ func (c *cluster) start(t *testDriver, syncerConfig broker.SyncerConfig) {
 	}
 }
 
-func findServiceImport(client dynamic.ResourceInterface, namespace, name, sourceLabel string) *mcsv1a1.ServiceImport {
-	list, err := client.List(context.TODO(), metav1.ListOptions{})
-	Expect(err).To(Succeed())
+func (c *cluster) createService() {
+	test.CreateResource(c.dynamicServiceClientFor().Namespace(c.service.Namespace), c.service)
+}
 
-	for i := range list.Items {
-		if list.Items[i].GetLabels()[sourceLabel] == name &&
-			list.Items[i].GetLabels()[constants.LabelSourceNamespace] == namespace {
-			serviceImport := &mcsv1a1.ServiceImport{}
-			Expect(scheme.Scheme.Convert(&list.Items[i], serviceImport, nil)).To(Succeed())
+func (c *cluster) updateService() {
+	test.UpdateResource(c.dynamicServiceClientFor().Namespace(c.service.Namespace), c.service)
+}
 
-			return serviceImport
-		}
-	}
+func (c *cluster) deleteService() {
+	Expect(c.dynamicServiceClientFor().Namespace(c.service.Namespace).Delete(context.TODO(), c.service.Name,
+		metav1.DeleteOptions{})).To(Succeed())
+}
 
-	return nil
+func (c *cluster) createServiceExport() {
+	test.CreateResource(c.localServiceExportClient, c.serviceExport)
 }
 
-func awaitServiceImport(client dynamic.ResourceInterface, service *corev1.Service, sType mcsv1a1.ServiceImportType,
-	serviceIP string,
-) *mcsv1a1.ServiceImport {
-	var serviceImport *mcsv1a1.ServiceImport
+func (c *cluster) deleteServiceExport() {
+	Expect(c.localServiceExportClient.Delete(context.TODO(), c.serviceExport.GetName(), metav1.DeleteOptions{})).To(Succeed())
+}
 
-	Eventually(func() *mcsv1a1.ServiceImport {
-		serviceImport = findServiceImport(client, service.Namespace, service.Name, mcsv1a1.LabelServiceName)
-		return serviceImport
-	}, 5*time.Second).ShouldNot(BeNil(), "ServiceImport not found")
+func (c *cluster) createEndpoints() {
+	test.CreateResource(endpointsClientFor(c.localDynClient, c.endpoints.Namespace), c.endpoints)
+}
 
-	Expect(serviceImport.Spec.Type).To(Equal(sType))
+func (c *cluster) updateEndpoints() {
+	test.UpdateResource(endpointsClientFor(c.localDynClient, c.endpoints.Namespace), c.endpoints)
+}
 
-	Expect(serviceImport.Status.Clusters).To(HaveLen(1))
-	Expect(serviceImport.Status.Clusters[0].Cluster).To(Equal(clusterID1))
+func (c *cluster) createGlobalIngressIP(ingressIP *unstructured.Unstructured) {
+	test.CreateResource(c.localIngressIPClient, ingressIP)
+}
 
-	if serviceIP == "" {
-		Expect(len(serviceImport.Spec.IPs)).To(Equal(0))
-	} else {
-		Expect(serviceImport.Spec.IPs).To(Equal([]string{serviceIP}))
-	}
+func (c *cluster) newHeadlessGlobalIngressIP(target, ip string) *unstructured.Unstructured {
+	ingressIP := c.newGlobalIngressIP("pod"+"-"+target, ip)
+	Expect(unstructured.SetNestedField(ingressIP.Object, controller.HeadlessServicePod, "spec", "target")).To(Succeed())
+	Expect(unstructured.SetNestedField(ingressIP.Object, target, "spec", "podRef", "name")).To(Succeed())
 
-	Expect(serviceImport.Spec.Ports).To(HaveLen(len(service.Spec.Ports)))
+	return ingressIP
+}
 
-	for i := range service.Spec.Ports {
-		Expect(serviceImport.Spec.Ports[i].Name).To(Equal(service.Spec.Ports[i].Name))
-		Expect(serviceImport.Spec.Ports[i].Protocol).To(Equal(service.Spec.Ports[i].Protocol))
-		Expect(serviceImport.Spec.Ports[i].Port).To(Equal(service.Spec.Ports[i].Port))
-	}
+func (c *cluster) newGlobalIngressIP(name, ip string) *unstructured.Unstructured {
+	ingressIP := controller.GetGlobalIngressIPObj()
+	ingressIP.SetName(name)
+	ingressIP.SetNamespace(c.service.Namespace)
+	Expect(unstructured.SetNestedField(ingressIP.Object, controller.ClusterIPService, "spec", "target")).To(Succeed())
+	Expect(unstructured.SetNestedField(ingressIP.Object, c.service.Name, "spec", "serviceRef", "name")).To(Succeed())
 
-	labels := serviceImport.GetObjectMeta().GetLabels()
-	Expect(labels[constants.LabelSourceNamespace]).To(Equal(service.GetNamespace()))
-	Expect(labels[mcsv1a1.LabelServiceName]).To(Equal(service.GetName()))
-	Expect(labels[constants.MCSLabelSourceCluster]).To(Equal(clusterID1))
+	setIngressAllocatedIP(ingressIP, ip)
+	setIngressIPConditions(ingressIP, metav1.Condition{
+		Type:    "Allocated",
+		Status:  metav1.ConditionTrue,
+		Reason:  "Success",
+		Message: "Allocated global IP",
+	})
 
-	return serviceImport
+	return ingressIP
 }
 
-func (c *cluster) awaitServiceImport(service *corev1.Service, sType mcsv1a1.ServiceImportType, serviceIP string) *mcsv1a1.ServiceImport {
-	return awaitServiceImport(c.localServiceImportClient, service, sType, serviceIP)
+func (c *cluster) retrieveServiceExportCondition(condType mcsv1a1.ServiceExportConditionType) *mcsv1a1.ServiceExportCondition {
+	obj, err := c.localServiceExportClient.Get(context.TODO(), c.serviceExport.Name, metav1.GetOptions{})
+	Expect(err).To(Succeed())
+
+	return controller.FindServiceExportStatusCondition(toServiceExport(obj).Status.Conditions, condType)
 }
 
-func awaitUpdatedServiceImport(client dynamic.ResourceInterface, service *corev1.Service, serviceIP string) {
-	var serviceImport *mcsv1a1.ServiceImport
+func (c *cluster) awaitServiceExportCondition(expected ...*mcsv1a1.ServiceExportCondition) {
+	conditionsEqual := func(actual, expected *mcsv1a1.ServiceExportCondition) bool {
+		return actual != nil && actual.Type == expected.Type && actual.Status == expected.Status &&
+			reflect.DeepEqual(actual.Reason, expected.Reason)
+	}
 
-	err := wait.PollImmediate(50*time.Millisecond, 5*time.Second, func() (bool, error) {
-		serviceImport = findServiceImport(client, service.Namespace, service.Name, mcsv1a1.LabelServiceName)
-		Expect(serviceImport).ToNot(BeNil())
+	actual := make([]*mcsv1a1.ServiceExportCondition, len(expected))
+	lastIndex := -1
 
-		if serviceIP == "" {
-			return len(serviceImport.Spec.IPs) == 0, nil
-		}
+	for i := range expected {
+		expStr, _ := json.MarshalIndent(expected[i], "", "  ")
+		j := lastIndex + 1
 
-		return reflect.DeepEqual(serviceImport.Spec.IPs, []string{serviceIP}), nil
-	})
+		Eventually(func() interface{} {
+			actions := c.localDynClient.Fake.Actions()
+			for j < len(actions) {
+				a := actions[j]
+				j++
 
-	if errors.Is(err, wait.ErrWaitTimeout) {
-		if serviceIP == "" {
-			Expect(len(serviceImport.Spec.IPs)).To(Equal(0))
-		} else {
-			Expect(serviceImport.Spec.IPs).To(Equal([]string{serviceIP}))
-		}
+				if !a.Matches("update", "serviceexports") {
+					continue
+				}
+
+				actual[i] = controller.FindServiceExportStatusCondition(
+					toServiceExport(a.(k8stesting.UpdateActionImpl).Object).Status.Conditions, expected[i].Type)
+
+				if conditionsEqual(actual[i], expected[i]) {
+					lastIndex = j
+
+					return actual[i]
+				}
+			}
+
+			return nil
+		}).ShouldNot(BeNil(), fmt.Sprintf("ServiceExport condition not received. Expected: %s", expStr))
 	}
 
-	Expect(err).To(Succeed())
+	for i := range expected {
+		assertEquivalentConditions(actual[i], expected[i])
+	}
+
+	time.Sleep(time.Millisecond * 100)
+
+	lastCond := expected[len(expected)-1]
+	assertEquivalentConditions(c.retrieveServiceExportCondition(lastCond.Type), lastCond)
 }
 
-func (c *cluster) awaitUpdatedServiceImport(service *corev1.Service, serviceIP string) {
-	awaitUpdatedServiceImport(c.localServiceImportClient, service, serviceIP)
+func (c *cluster) ensureNoServiceExportCondition(condType mcsv1a1.ServiceExportConditionType) {
+	Consistently(func() interface{} {
+		return c.retrieveServiceExportCondition(condType)
+	}).Should(BeNil(), "Unexpected ServiceExport status condition")
 }
 
-func findEndpointSlice(client dynamic.ResourceInterface, namespace, name string) *discovery.EndpointSlice {
-	list, err := client.List(context.TODO(), metav1.ListOptions{})
+func (c *cluster) awaitServiceUnavailableStatus() {
+	c.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "ServiceUnavailable"))
+}
+
+func (c *cluster) findLocalServiceImport() *mcsv1a1.ServiceImport {
+	list, err := c.localServiceImportClient.Namespace(test.LocalNamespace).List(context.TODO(), metav1.ListOptions{})
 	Expect(err).To(Succeed())
 
 	for i := range list.Items {
-		if list.Items[i].GetLabels()[mcsv1a1.LabelServiceName] == name &&
-			list.Items[i].GetLabels()[constants.LabelSourceNamespace] == namespace {
-			endpointSlice := &discovery.EndpointSlice{}
-			Expect(scheme.Scheme.Convert(&list.Items[i], endpointSlice, nil)).To(Succeed())
+		if list.Items[i].GetLabels()[mcsv1a1.LabelServiceName] == c.service.Name &&
+			list.Items[i].GetLabels()[constants.LabelSourceNamespace] == c.service.Namespace {
+			serviceImport := &mcsv1a1.ServiceImport{}
+			Expect(scheme.Scheme.Convert(&list.Items[i], serviceImport, nil)).To(Succeed())
 
-			return endpointSlice
+			return serviceImport
 		}
 	}
 
 	return nil
 }
 
-func awaitEndpointSlice(client dynamic.ResourceInterface, namespace, name string) *discovery.EndpointSlice {
-	var endpointSlice *discovery.EndpointSlice
-
-	Eventually(func() *discovery.EndpointSlice {
-		endpointSlice = findEndpointSlice(client, namespace, name)
-		return endpointSlice
-	}, 5*time.Second).ShouldNot(BeNil(), "EndpointSlice not found")
-
-	return endpointSlice
+func (c *cluster) findLocalEndpointSlice() *discovery.EndpointSlice {
+	return findEndpointSlice(c.localEndpointSliceClient, c.endpoints.Namespace, c.endpoints.Name, c.clusterID)
 }
 
-func awaitAndVerifyEndpointSlice(endpointSliceClient dynamic.ResourceInterface, endpoints *corev1.Endpoints,
-	namespace string, globalIPs []string,
-) *discovery.EndpointSlice {
-	endpointSlice := awaitEndpointSlice(endpointSliceClient, endpoints.Namespace, endpoints.Name)
+func awaitServiceImport(client dynamic.NamespaceableResourceInterface, expected *mcsv1a1.ServiceImport) {
+	var serviceImport *mcsv1a1.ServiceImport
 
-	Expect(endpointSlice.Namespace).To(Equal(namespace))
+	err := wait.PollImmediate(50*time.Millisecond, 5*time.Second, func() (bool, error) {
+		obj, err := client.Namespace(expected.Namespace).Get(context.TODO(), expected.Name, metav1.GetOptions{})
+		if apierrors.IsNotFound(err) {
+			return false, nil
+		}
 
-	labels := endpointSlice.GetLabels()
-	Expect(labels).To(HaveKeyWithValue(discovery.LabelManagedBy, constants.LabelValueManagedBy))
-	Expect(labels).To(HaveKeyWithValue(constants.MCSLabelSourceCluster, clusterID1))
+		serviceImport = &mcsv1a1.ServiceImport{}
+		Expect(scheme.Scheme.Convert(obj, serviceImport, nil)).To(Succeed())
 
-	Expect(endpointSlice.AddressType).To(Equal(discovery.AddressTypeIPv4))
+		return reflect.DeepEqual(&expected.Spec, &serviceImport.Spec) && reflect.DeepEqual(&expected.Status, &serviceImport.Status), nil
+	})
 
-	addresses := globalIPs
-	if addresses == nil {
-		addresses = []string{
-			endpoints.Subsets[0].Addresses[0].IP, endpoints.Subsets[0].Addresses[1].IP,
-		}
+	if !errors.Is(err, wait.ErrWaitTimeout) {
+		Expect(err).To(Succeed())
+	}
+
+	if serviceImport == nil {
+		Fail(fmt.Sprintf("ServiceImport %s/%s not found", expected.Namespace, expected.Name))
 	}
 
-	Expect(endpointSlice.Endpoints).To(HaveLen(2))
-	Expect(endpointSlice.Endpoints[0]).To(Equal(discovery.Endpoint{
-		Addresses:  []string{addresses[0]},
-		Conditions: discovery.EndpointConditions{Ready: &ready},
-		Hostname:   &hostName,
-	}))
-	Expect(endpointSlice.Endpoints[1]).To(Equal(discovery.Endpoint{
-		Addresses:  []string{addresses[1]},
-		Hostname:   &endpoints.Subsets[0].Addresses[1].TargetRef.Name,
-		Conditions: discovery.EndpointConditions{Ready: &ready},
-		NodeName:   &nodeName,
-	}))
+	Expect(serviceImport.Spec).To(Equal(expected.Spec))
+	Expect(serviceImport.Status).To(Equal(expected.Status))
 
-	Expect(endpointSlice.Ports).To(HaveLen(1))
+	Expect(serviceImport.Labels).To(BeEmpty())
+}
 
-	name := "port-1"
-	protocol := corev1.ProtocolTCP
-	var port int32 = 1234
+func findEndpointSlice(client dynamic.ResourceInterface, namespace, name, clusterID string) *discovery.EndpointSlice {
+	list, err := client.List(context.TODO(), metav1.ListOptions{})
+	Expect(err).To(Succeed())
 
-	Expect(endpointSlice.Ports[0]).To(Equal(discovery.EndpointPort{
-		Name:     &name,
-		Protocol: &protocol,
-		Port:     &port,
-	}))
+	for i := range list.Items {
+		if list.Items[i].GetLabels()[mcsv1a1.LabelServiceName] == name &&
+			list.Items[i].GetLabels()[constants.LabelSourceNamespace] == namespace &&
+			list.Items[i].GetLabels()[constants.MCSLabelSourceCluster] == clusterID {
+			endpointSlice := &discovery.EndpointSlice{}
+			Expect(scheme.Scheme.Convert(&list.Items[i], endpointSlice, nil)).To(Succeed())
 
-	return endpointSlice
-}
+			return endpointSlice
+		}
+	}
 
-func (c *cluster) awaitEndpointSlice(t *testDriver) *discovery.EndpointSlice {
-	return awaitAndVerifyEndpointSlice(c.localEndpointSliceClient, t.endpoints, t.service.Namespace, t.endpointGlobalIPs)
+	return nil
 }
 
-func awaitUpdatedEndpointSlice(endpointSliceClient dynamic.ResourceInterface, endpoints *corev1.Endpoints, expectedIPs []string) {
-	sort.Strings(expectedIPs)
-
-	var actualIPs []string
+func awaitEndpointSlice(client dynamic.ResourceInterface, expected *discovery.EndpointSlice) {
+	var endpointSlice *discovery.EndpointSlice
 
 	err := wait.PollImmediate(50*time.Millisecond, 5*time.Second, func() (bool, error) {
-		endpointSlice := findEndpointSlice(endpointSliceClient, endpoints.Namespace, endpoints.Name)
-		Expect(endpointSlice).ToNot(BeNil())
-
-		actualIPs = nil
-		for _, ep := range endpointSlice.Endpoints {
-			actualIPs = append(actualIPs, ep.Addresses...)
+		endpointSlice = findEndpointSlice(client, expected.Namespace, expected.Name, expected.Labels[constants.MCSLabelSourceCluster])
+		if endpointSlice == nil {
+			return false, nil
 		}
 
-		sort.Strings(actualIPs)
-
-		return reflect.DeepEqual(actualIPs, expectedIPs), nil
+		return reflect.DeepEqual(expected.Endpoints, endpointSlice.Endpoints) &&
+			reflect.DeepEqual(expected.Ports, endpointSlice.Ports), nil
 	})
 
-	if errors.Is(err, wait.ErrWaitTimeout) {
-		Expect(actualIPs).To(Equal(expectedIPs))
+	if !errors.Is(err, wait.ErrWaitTimeout) {
+		Expect(err).To(Succeed())
 	}
 
-	Expect(err).To(Succeed())
+	if endpointSlice == nil {
+		Fail(fmt.Sprintf("EndpointSlice for %s/%s not found", expected.Namespace, expected.Name))
+	}
+
+	for k, v := range expected.Labels {
+		Expect(endpointSlice.Labels).To(HaveKeyWithValue(k, v))
+	}
+
+	Expect(endpointSlice.AddressType).To(Equal(expected.AddressType))
+	Expect(endpointSlice.Endpoints).To(Equal(expected.Endpoints))
+	Expect(endpointSlice.Ports).To(Equal(expected.Ports))
 }
 
-func (c *cluster) awaitUpdatedEndpointSlice(endpoints *corev1.Endpoints, expectedIPs []string) {
-	awaitUpdatedEndpointSlice(c.localEndpointSliceClient, endpoints, expectedIPs)
+func awaitNoEndpointSlice(client dynamic.ResourceInterface, ns, name, clusterID string) {
+	Eventually(func() interface{} {
+		return findEndpointSlice(client, ns, name, clusterID)
+	}).Should(BeNil(), "Unexpected EndpointSlice found for %s/%s", ns, name)
 }
 
-func (c *cluster) dynamicServiceClient() dynamic.NamespaceableResourceInterface {
+func (c *cluster) dynamicServiceClientFor() dynamic.NamespaceableResourceInterface {
 	return c.localDynClient.Resource(schema.GroupVersionResource{Version: "v1", Resource: "services"})
 }
 
-func (t *testDriver) awaitBrokerServiceImport(sType mcsv1a1.ServiceImportType, serviceIP string) *mcsv1a1.ServiceImport {
-	return awaitServiceImport(t.brokerServiceImportClient, t.service, sType, serviceIP)
-}
+func (t *testDriver) awaitAggregatedServiceImport(sType mcsv1a1.ServiceImportType, name, ns string, clusters ...*cluster) {
+	expServiceImport := &mcsv1a1.ServiceImport{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      fmt.Sprintf("%s-%s", name, ns),
+			Namespace: test.RemoteNamespace,
+		},
+		Spec: mcsv1a1.ServiceImportSpec{
+			Type:  sType,
+			Ports: []mcsv1a1.ServicePort{},
+		},
+	}
 
-func (t *testDriver) awaitBrokerEndpointSlice() *discovery.EndpointSlice {
-	return awaitAndVerifyEndpointSlice(t.brokerEndpointSliceClient, t.endpoints, test.RemoteNamespace, t.endpointGlobalIPs)
-}
+	if len(clusters) > 0 {
+		if sType == mcsv1a1.ClusterSetIP {
+			expServiceImport.Spec.Ports = t.aggregatedServicePorts
+		}
 
-func (t *testDriver) awaitUpdatedServiceImport(serviceIP string) {
-	awaitUpdatedServiceImport(t.brokerServiceImportClient, t.service, serviceIP)
-	t.cluster1.awaitUpdatedServiceImport(t.service, serviceIP)
-	t.cluster2.awaitUpdatedServiceImport(t.service, serviceIP)
-}
+		for _, c := range clusters {
+			expServiceImport.Status.Clusters = append(expServiceImport.Status.Clusters,
+				mcsv1a1.ClusterStatus{Cluster: c.clusterID})
+		}
+	}
 
-func (t *testDriver) awaitEndpointSlice() {
-	t.awaitBrokerEndpointSlice()
-	t.cluster1.awaitEndpointSlice(t)
-	t.cluster2.awaitEndpointSlice(t)
-}
+	awaitServiceImport(t.brokerServiceImportClient, expServiceImport)
 
-func (t *testDriver) awaitUpdatedEndpointSlice(expectedIPs []string) {
-	awaitUpdatedEndpointSlice(t.brokerEndpointSliceClient, t.endpoints, expectedIPs)
-	t.cluster1.awaitUpdatedEndpointSlice(t.endpoints, expectedIPs)
-	t.cluster2.awaitUpdatedEndpointSlice(t.endpoints, expectedIPs)
-}
+	expServiceImport.Name = name
+	expServiceImport.Namespace = ns
 
-func (t *testDriver) createService() {
-	test.CreateResource(t.cluster1.dynamicServiceClient().Namespace(t.service.Namespace), t.service)
+	awaitServiceImport(t.cluster1.localServiceImportClient, expServiceImport)
+	awaitServiceImport(t.cluster2.localServiceImportClient, expServiceImport)
 }
 
-func (t *testDriver) updateService() {
-	test.UpdateResource(t.cluster1.dynamicServiceClient().Namespace(t.service.Namespace), t.service)
+func (t *testDriver) awaitNoAggregatedServiceImport(c *cluster) {
+	test.AwaitNoResource(t.brokerServiceImportClient.Namespace(test.RemoteNamespace),
+		fmt.Sprintf("%s-%s", c.service.Name, c.service.Namespace))
+	test.AwaitNoResource(t.cluster1.localServiceImportClient.Namespace(c.service.Namespace), c.service.Name)
+	test.AwaitNoResource(t.cluster2.localServiceImportClient.Namespace(c.service.Namespace), c.service.Name)
 }
 
-func (t *testDriver) createEndpoints() {
-	test.CreateResource(t.dynamicEndpointsClient(), t.endpoints)
-}
+func (t *testDriver) awaitEndpointSlice(c *cluster) {
+	hasEndpoints := len(c.endpoints.Subsets[0].Addresses) > 0
+	isHeadless := c.service.Spec.ClusterIP == corev1.ClusterIPNone
+
+	expected := &discovery.EndpointSlice{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      c.service.Name,
+			Namespace: c.service.Namespace,
+			Labels: map[string]string{
+				discovery.LabelManagedBy:        constants.LabelValueManagedBy,
+				constants.MCSLabelSourceCluster: c.clusterID,
+				mcsv1a1.LabelServiceName:        c.service.Name,
+				constants.LabelSourceNamespace:  c.service.Namespace,
+				constants.LabelIsHeadless:       strconv.FormatBool(isHeadless),
+			},
+		},
+		AddressType: discovery.AddressTypeIPv4,
+	}
+
+	if isHeadless {
+		expected.Endpoints = c.endpointSliceAddresses
+
+		for i := range c.endpoints.Subsets[0].Ports {
+			expected.Ports = append(expected.Ports, discovery.EndpointPort{
+				Name:     &c.endpoints.Subsets[0].Ports[i].Name,
+				Protocol: &c.endpoints.Subsets[0].Ports[i].Protocol,
+				Port:     &c.endpoints.Subsets[0].Ports[i].Port,
+			})
+		}
+	} else {
+		expected.Endpoints = []discovery.Endpoint{
+			{
+				Addresses:  []string{c.serviceIP},
+				Conditions: discovery.EndpointConditions{Ready: pointer.Bool(hasEndpoints)},
+			},
+		}
 
-func (t *testDriver) updateEndpoints() {
-	test.UpdateResource(t.dynamicEndpointsClient(), t.endpoints)
+		for i := range c.service.Spec.Ports {
+			expected.Ports = append(expected.Ports, discovery.EndpointPort{
+				Name:     &c.service.Spec.Ports[i].Name,
+				Protocol: &c.service.Spec.Ports[i].Protocol,
+				Port:     &c.service.Spec.Ports[i].Port,
+			})
+		}
+	}
+
+	awaitEndpointSlice(t.brokerEndpointSliceClient, expected)
+	awaitEndpointSlice(t.cluster1.localEndpointSliceClient, expected)
+	awaitEndpointSlice(t.cluster2.localEndpointSliceClient, expected)
 }
 
-func (t *testDriver) dynamicEndpointsClient() dynamic.ResourceInterface {
-	return dynamicEndpointsClient(t.cluster1.localDynClient, t.service.Namespace)
+func (t *testDriver) awaitNoEndpointSlice(c *cluster) {
+	awaitNoEndpointSlice(t.cluster1.localEndpointSliceClient, c.service.Namespace, c.service.Name, c.clusterID)
+	awaitNoEndpointSlice(t.brokerEndpointSliceClient, c.service.Namespace, c.service.Name, c.clusterID)
+	awaitNoEndpointSlice(t.cluster2.localEndpointSliceClient, c.service.Namespace, c.service.Name, c.clusterID)
 }
 
-func dynamicEndpointsClient(client dynamic.Interface, namespace string) dynamic.ResourceInterface {
+func endpointsClientFor(client dynamic.Interface, namespace string) dynamic.ResourceInterface {
 	return client.Resource(corev1.SchemeGroupVersion.WithResource("endpoints")).Namespace(namespace)
 }
 
-func serviceExportClient(client dynamic.Interface, namespace string) dynamic.ResourceInterface {
+func serviceExportClientFor(client dynamic.Interface, namespace string) dynamic.ResourceInterface {
 	return client.Resource(schema.GroupVersionResource{
 		Group:    mcsv1a1.GroupVersion.Group,
 		Version:  mcsv1a1.GroupVersion.Version,
@@ -611,42 +706,10 @@ func serviceExportClient(client dynamic.Interface, namespace string) dynamic.Res
 	}).Namespace(namespace)
 }
 
-func endpointSliceClient(client dynamic.Interface, namespace string) dynamic.ResourceInterface {
+func endpointSliceClientFor(client dynamic.Interface, namespace string) dynamic.ResourceInterface {
 	return client.Resource(discovery.SchemeGroupVersion.WithResource("endpointslices")).Namespace(namespace)
 }
 
-func (t *testDriver) createServiceExport() {
-	test.CreateResource(t.cluster1.localServiceExportClient, t.serviceExport)
-}
-
-func (t *testDriver) deleteServiceExport() {
-	Expect(t.cluster1.localServiceExportClient.Delete(context.TODO(), t.service.GetName(), metav1.DeleteOptions{})).To(Succeed())
-}
-
-func (t *testDriver) deleteService() {
-	Expect(t.cluster1.dynamicServiceClient().Namespace(t.service.Namespace).Delete(context.TODO(), t.service.Name,
-		metav1.DeleteOptions{})).To(Succeed())
-}
-
-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.newHeadlessGlobalIngressIP("one", globalIP1))
-	t.createGlobalIngressIP(t.newHeadlessGlobalIngressIP("two", globalIP2))
-	t.createGlobalIngressIP(t.newHeadlessGlobalIngressIP("not-ready", globalIP3))
-}
-
-func (t *testDriver) awaitNoServiceImport(client dynamic.ResourceInterface) {
-	test.AwaitNoResource(client, t.service.Name+"-"+t.service.Namespace+"-"+clusterID1)
-}
-
-func (t *testDriver) awaitNoEndpointSlice(client dynamic.ResourceInterface) {
-	test.AwaitNoResource(client, t.endpoints.Name+"-"+clusterID1)
-}
-
 func assertEquivalentConditions(actual, expected *mcsv1a1.ServiceExportCondition) {
 	out, _ := json.MarshalIndent(actual, "", "  ")
 
@@ -668,119 +731,44 @@ func toServiceExport(obj interface{}) *mcsv1a1.ServiceExport {
 	return se
 }
 
-func (t *testDriver) awaitServiceExportCondition(expected ...*mcsv1a1.ServiceExportCondition) {
-	conditionsEqual := func(actual, expected *mcsv1a1.ServiceExportCondition) bool {
-		return actual != nil && actual.Type == expected.Type && actual.Status == expected.Status &&
-			reflect.DeepEqual(actual.Reason, expected.Reason)
-	}
-
-	actual := make([]*mcsv1a1.ServiceExportCondition, len(expected))
-
-	for i := range expected {
-		actual[i] = t.retrieveServiceExportCondition(expected[i].Type)
-		if conditionsEqual(actual[i], expected[i]) {
-			continue
-		}
-
-		condChan := t.cluster1.chanByCondType[expected[i].Type]
-
-		var received *mcsv1a1.ServiceExportCondition
-		nIter := 0
-
-		for received == nil && nIter <= 50 {
-			select {
-			case c := <-condChan:
-				if conditionsEqual(&c, expected[i]) {
-					received = &c
-				}
-			case <-time.After(100 * time.Millisecond):
-			}
-
-			nIter++
-		}
-
-		if received == nil {
-			out, _ := json.MarshalIndent(expected[i], "", "  ")
-			Fail(fmt.Sprintf("ServiceExport condition not received. Expected: %s", out))
-		}
-
-		actual[i] = received
-	}
-
-	for i := range expected {
-		assertEquivalentConditions(actual[i], expected[i])
-	}
-
-	time.Sleep(time.Millisecond * 100)
-
-	lastCond := expected[len(expected)-1]
-	assertEquivalentConditions(t.retrieveServiceExportCondition(lastCond.Type), lastCond)
-}
-
-func (t *testDriver) ensureNoServiceExportCondition(condType mcsv1a1.ServiceExportConditionType) {
-	Consistently(func() interface{} {
-		return t.retrieveServiceExportCondition(condType)
-	}).Should(BeNil(), "Unexpected ServiceExport status condition")
+func (t *testDriver) awaitNonHeadlessServiceExported(c *cluster) {
+	t.awaitServiceExported(mcsv1a1.ClusterSetIP, c)
 }
 
-func (t *testDriver) retrieveServiceExportCondition(condType mcsv1a1.ServiceExportConditionType) *mcsv1a1.ServiceExportCondition {
-	obj, err := t.cluster1.localServiceExportClient.Get(context.TODO(), t.service.Name, metav1.GetOptions{})
-	Expect(err).To(Succeed())
-
-	return controller.FindServiceExportStatusCondition(toServiceExport(obj).Status.Conditions, condType)
+func (t *testDriver) awaitHeadlessServiceExported(c *cluster) {
+	t.awaitServiceExported(mcsv1a1.Headless, c)
 }
 
-func (t *testDriver) awaitServiceExported(serviceIP string) {
-	t.cluster1.awaitServiceImport(t.service, mcsv1a1.ClusterSetIP, serviceIP)
+func (t *testDriver) awaitServiceExported(sType mcsv1a1.ServiceImportType, c *cluster) {
+	t.awaitAggregatedServiceImport(sType, c.service.Name, c.service.Namespace, c)
 
-	t.awaitBrokerServiceImport(mcsv1a1.ClusterSetIP, serviceIP)
-	t.cluster2.awaitServiceImport(t.service, mcsv1a1.ClusterSetIP, serviceIP)
+	t.awaitEndpointSlice(c)
 
-	t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionTrue, ""))
-	t.awaitServiceExportCondition(newServiceExportSyncedCondition(corev1.ConditionFalse, "AwaitingSync"),
+	c.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionTrue, ""))
+	c.awaitServiceExportCondition(newServiceExportSyncedCondition(corev1.ConditionFalse, "AwaitingExport"),
 		newServiceExportSyncedCondition(corev1.ConditionTrue, ""))
 }
 
-func (t *testDriver) awaitHeadlessServiceImport() {
-	serviceIP := ""
-	t.awaitBrokerServiceImport(mcsv1a1.Headless, serviceIP)
-	t.cluster1.awaitServiceImport(t.service, mcsv1a1.Headless, serviceIP)
-	t.cluster2.awaitServiceImport(t.service, mcsv1a1.Headless, serviceIP)
-}
-
-func (t *testDriver) awaitServiceUnexported() {
-	t.awaitNoServiceImport(t.brokerServiceImportClient)
-	t.awaitNoServiceImport(t.cluster1.localServiceImportClient)
-	t.awaitNoServiceImport(t.cluster2.localServiceImportClient)
-}
+func (t *testDriver) awaitServiceUnexported(c *cluster) {
+	t.awaitNoEndpointSlice(c)
 
-func (t *testDriver) awaitHeadlessServiceUnexported() {
-	t.awaitServiceUnexported()
+	t.awaitNoAggregatedServiceImport(c)
 
-	t.awaitNoEndpointSlice(t.cluster1.localEndpointSliceClient)
-	t.awaitNoEndpointSlice(t.brokerEndpointSliceClient)
-	t.awaitNoEndpointSlice(t.cluster2.localEndpointSliceClient)
+	t.cluster1.localDynClient.Fake.ClearActions()
 
-	// Ensure the service's Endpoints are no longer being watched by updating the Endpoints and verifying the
-	// EndpointSlice isn't recreated.
-	t.endpoints.Subsets[0].Addresses = append(t.endpoints.Subsets[0].Addresses, corev1.EndpointAddress{IP: "192.168.5.10"})
-	t.updateEndpoints()
+	_, err := endpointsClientFor(c.localDynClient, c.endpoints.Namespace).Get(context.TODO(), c.endpoints.Name, metav1.GetOptions{})
+	if err == nil {
+		// Ensure the service's Endpoints are no longer being watched by updating the Endpoints and verifying the
+		// EndpointSlice isn't recreated.
+		t.cluster1.localDynClient.Fake.ClearActions()
 
-	time.Sleep(200 * time.Millisecond)
-	t.awaitNoEndpointSlice(t.cluster1.localEndpointSliceClient)
-}
+		c.endpoints.Subsets[0].Addresses = append(c.endpoints.Subsets[0].Addresses, corev1.EndpointAddress{IP: "192.168.5.10"})
+		c.updateEndpoints()
 
-func (t *testDriver) awaitServiceUnavailableStatus() {
-	t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "ServiceUnavailable"))
-}
-
-func (t *testDriver) endpointIPs() []string {
-	ips := []string{}
-	for _, a := range t.endpoints.Subsets[0].Addresses {
-		ips = append(ips, a.IP)
+		testutil.EnsureNoActionsForResource(&t.cluster1.localDynClient.Fake, "endpointslices", "create")
+	} else if !apierrors.IsNotFound(err) {
+		Expect(err).To(Succeed())
 	}
-
-	return ips
 }
 
 func newServiceExportValidCondition(status corev1.ConditionStatus, reason string) *mcsv1a1.ServiceExportCondition {
@@ -814,3 +802,11 @@ func setIngressIPConditions(ingressIP *unstructured.Unstructured, conditions ...
 func setIngressAllocatedIP(ingressIP *unstructured.Unstructured, ip string) {
 	Expect(unstructured.SetNestedField(ingressIP.Object, ip, "status", "allocatedIP")).To(Succeed())
 }
+
+func toServicePort(port mcsv1a1.ServicePort) corev1.ServicePort {
+	return corev1.ServicePort{
+		Name:     port.Name,
+		Protocol: port.Protocol,
+		Port:     port.Port,
+	}
+}
diff --git a/pkg/agent/controller/globalnet_test.go b/pkg/agent/controller/globalnet_test.go
index 1046ad9..04d9e52 100644
--- a/pkg/agent/controller/globalnet_test.go
+++ b/pkg/agent/controller/globalnet_test.go
@@ -19,9 +19,8 @@ limitations under the License.
 package controller_test
 
 import (
-	"time"
-
 	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
 	"github.com/submariner-io/admiral/pkg/syncer/test"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -39,8 +38,9 @@ var _ = Describe("Globalnet enabled", func() {
 
 	JustBeforeEach(func() {
 		t.justBeforeEach()
-		t.createService()
-		t.createServiceExport()
+		t.cluster1.createEndpoints()
+		t.cluster1.createService()
+		t.cluster1.createServiceExport()
 	})
 
 	AfterEach(func() {
@@ -51,28 +51,29 @@ var _ = Describe("Globalnet enabled", func() {
 		var ingressIP *unstructured.Unstructured
 
 		BeforeEach(func() {
-			ingressIP = t.newGlobalIngressIP(t.service.Name, globalIP1)
+			ingressIP = t.cluster1.newGlobalIngressIP(t.cluster1.service.Name, globalIP1)
+			t.cluster1.serviceIP = globalIP1
 		})
 
 		Context("and it has a global IP", func() {
 			Context("via a GlobalIngressIP", func() {
 				BeforeEach(func() {
-					t.createGlobalIngressIP(ingressIP)
+					t.cluster1.createGlobalIngressIP(ingressIP)
 				})
 
-				It("should sync a ServiceImport with the global IP", func() {
-					t.awaitServiceExported(globalIP1)
+				It("should export the service with the global IP", func() {
+					t.awaitNonHeadlessServiceExported(&t.cluster1)
 				})
 			})
 		})
 
 		Context("and it does not initially have a global IP", func() {
 			Context("due to missing GlobalIngressIP", func() {
-				It("should update the ServiceExport status appropriately and eventually sync a ServiceImport", func() {
-					t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "ServiceGlobalIPUnavailable"))
+				It("should update the ServiceExport status appropriately and eventually export the service", func() {
+					t.cluster1.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "ServiceGlobalIPUnavailable"))
 
-					t.createGlobalIngressIP(ingressIP)
-					t.awaitServiceExported(globalIP1)
+					t.cluster1.createGlobalIngressIP(ingressIP)
+					t.awaitNonHeadlessServiceExported(&t.cluster1)
 				})
 			})
 
@@ -80,15 +81,15 @@ var _ = Describe("Globalnet enabled", func() {
 				BeforeEach(func() {
 					setIngressAllocatedIP(ingressIP, "")
 					setIngressIPConditions(ingressIP)
-					t.createGlobalIngressIP(ingressIP)
+					t.cluster1.createGlobalIngressIP(ingressIP)
 				})
 
-				It("should update the ServiceExport status appropriately and eventually sync a ServiceImport", func() {
-					t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "ServiceGlobalIPUnavailable"))
+				It("should update the ServiceExport status appropriately and eventually export the service", func() {
+					t.cluster1.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "ServiceGlobalIPUnavailable"))
 
 					setIngressAllocatedIP(ingressIP, globalIP1)
 					test.UpdateResource(t.cluster1.localIngressIPClient, ingressIP)
-					t.awaitServiceExported(globalIP1)
+					t.awaitNonHeadlessServiceExported(&t.cluster1)
 				})
 			})
 		})
@@ -104,46 +105,53 @@ var _ = Describe("Globalnet enabled", func() {
 			BeforeEach(func() {
 				setIngressAllocatedIP(ingressIP, "")
 				setIngressIPConditions(ingressIP, condition)
-				t.createGlobalIngressIP(ingressIP)
+				t.cluster1.createGlobalIngressIP(ingressIP)
 			})
 
 			It("should update the ServiceExport status with the condition details", func() {
 				c := newServiceExportValidCondition(corev1.ConditionFalse, condition.Reason)
 				c.Message = &condition.Message
-				t.awaitServiceExportCondition(c)
+				t.cluster1.awaitServiceExportCondition(c)
 			})
 		})
 	})
 
-	When("a headless Service is exported", func() {
+	When("a Headless Service is exported", func() {
 		BeforeEach(func() {
-			t.service.Spec.ClusterIP = corev1.ClusterIPNone
-		})
+			t.cluster1.service.Spec.ClusterIP = corev1.ClusterIPNone
 
-		JustBeforeEach(func() {
-			t.createEndpoints()
+			t.cluster1.endpointSliceAddresses[0].Addresses = []string{globalIP1}
+			t.cluster1.endpointSliceAddresses[1].Addresses = []string{globalIP2}
 		})
 
-		Context("and it has a global IP for all endpoint addresses", func() {
+		Context("and it has global IPs for all endpoint addresses", func() {
 			BeforeEach(func() {
-				t.createEndpointIngressIPs()
+				t.cluster1.createGlobalIngressIP(t.cluster1.newHeadlessGlobalIngressIP("one", globalIP1))
+				t.cluster1.createGlobalIngressIP(t.cluster1.newHeadlessGlobalIngressIP("two", globalIP2))
 			})
 
-			It("should sync a ServiceImport and EndpointSlice with the global IPs", func() {
-				t.awaitHeadlessServiceImport()
-				t.awaitEndpointSlice()
+			It("should export the service with the global IPs", func() {
+				t.awaitHeadlessServiceExported(&t.cluster1)
 			})
 		})
 
 		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)
+			It("should eventually export the service with the global IPs", func() {
+				Consistently(func() interface{} {
+					return findEndpointSlice(t.cluster1.localEndpointSliceClient, t.cluster1.endpoints.Namespace,
+						t.cluster1.endpoints.Name, t.cluster1.clusterID)
+				}).Should(BeNil(), "Unexpected EndpointSlice")
+
+				t.cluster1.createGlobalIngressIP(t.cluster1.newHeadlessGlobalIngressIP("one", globalIP1))
+
+				Consistently(func() interface{} {
+					return findEndpointSlice(t.cluster1.localEndpointSliceClient, t.cluster1.endpoints.Namespace,
+						t.cluster1.endpoints.Name, t.cluster1.clusterID)
+				}).Should(BeNil(), "Unexpected EndpointSlice")
 
-				t.createEndpointIngressIPs()
+				t.cluster1.createGlobalIngressIP(t.cluster1.newHeadlessGlobalIngressIP("two", globalIP2))
 
-				t.awaitHeadlessServiceImport()
-				t.awaitEndpointSlice()
+				t.awaitEndpointSlice(&t.cluster1)
 			})
 		})
 	})
diff --git a/pkg/agent/controller/headless_service_test.go b/pkg/agent/controller/headless_service_test.go
index 245d74a..bfea061 100644
--- a/pkg/agent/controller/headless_service_test.go
+++ b/pkg/agent/controller/headless_service_test.go
@@ -21,19 +21,22 @@ package controller_test
 import (
 	. "github.com/onsi/ginkgo/v2"
 	corev1 "k8s.io/api/core/v1"
+	discovery "k8s.io/api/discovery/v1"
+	"k8s.io/utils/pointer"
+	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
-var _ = Describe("Headless service syncing", func() {
+var _ = Describe("Headless Service export", func() {
 	var t *testDriver
 
 	BeforeEach(func() {
 		t = newTestDiver()
-		t.service.Spec.ClusterIP = corev1.ClusterIPNone
+		t.cluster1.service.Spec.ClusterIP = corev1.ClusterIPNone
 	})
 
 	JustBeforeEach(func() {
 		t.justBeforeEach()
-		t.createService()
+		t.cluster1.createService()
 	})
 
 	AfterEach(func() {
@@ -41,52 +44,51 @@ var _ = Describe("Headless service syncing", func() {
 	})
 
 	When("a ServiceExport is created", func() {
-		When("the Endpoints already exists", func() {
-			It("should correctly sync a ServiceImport and EndpointSlice", func() {
-				t.createEndpoints()
-				t.createServiceExport()
+		Context("and the Service already exists", func() {
+			It("should export the service", func() {
+				t.cluster1.createEndpoints()
+				t.cluster1.createServiceExport()
 
-				t.awaitHeadlessServiceImport()
-				t.awaitEndpointSlice()
+				t.awaitHeadlessServiceExported(&t.cluster1)
 			})
 		})
 
-		When("the Endpoints doesn't initially exist", func() {
-			It("should eventually sync a correct ServiceImport and EndpointSlice", func() {
-				t.createServiceExport()
-				t.awaitHeadlessServiceImport()
+		Context("and the Endpoints doesn't initially exist", func() {
+			It("should eventually export the EndpointSlice", func() {
+				t.cluster1.createServiceExport()
+				t.awaitAggregatedServiceImport(mcsv1a1.Headless, t.cluster1.service.Name, t.cluster1.service.Namespace)
 
-				t.createEndpoints()
-				t.awaitUpdatedServiceImport("")
-				t.awaitEndpointSlice()
+				t.cluster1.createEndpoints()
+				t.awaitEndpointSlice(&t.cluster1)
 			})
 		})
 	})
 
 	When("the Endpoints for a service are updated", func() {
-		It("should update the ServiceImport and EndpointSlice", func() {
-			t.createEndpoints()
-			t.createServiceExport()
-
-			t.awaitHeadlessServiceImport()
-			t.awaitEndpointSlice()
+		It("should update the exported EndpointSlice", func() {
+			t.cluster1.createEndpoints()
+			t.cluster1.createServiceExport()
+			t.awaitHeadlessServiceExported(&t.cluster1)
+
+			t.cluster1.endpoints.Subsets[0].Addresses = append(t.cluster1.endpoints.Subsets[0].Addresses, corev1.EndpointAddress{IP: "192.168.5.3"})
+			t.cluster1.endpointSliceAddresses = append(t.cluster1.endpointSliceAddresses, discovery.Endpoint{
+				Addresses:  []string{"192.168.5.3"},
+				Conditions: discovery.EndpointConditions{Ready: pointer.Bool(true)},
+			})
 
-			t.endpoints.Subsets[0].Addresses = append(t.endpoints.Subsets[0].Addresses, corev1.EndpointAddress{IP: "192.168.5.3"})
-			t.updateEndpoints()
-			t.awaitUpdatedServiceImport("")
-			t.awaitUpdatedEndpointSlice(t.endpointIPs())
+			t.cluster1.updateEndpoints()
+			t.awaitEndpointSlice(&t.cluster1)
 		})
 	})
 
 	When("a ServiceExport is deleted", func() {
-		It("should delete the ServiceImport and EndpointSlice", func() {
-			t.createEndpoints()
-			t.createServiceExport()
-			t.awaitHeadlessServiceImport()
-			t.awaitEndpointSlice()
-
-			t.deleteServiceExport()
-			t.awaitHeadlessServiceUnexported()
+		It("should unexport the service", func() {
+			t.cluster1.createEndpoints()
+			t.cluster1.createServiceExport()
+			t.awaitHeadlessServiceExported(&t.cluster1)
+
+			t.cluster1.deleteServiceExport()
+			t.awaitServiceUnexported(&t.cluster1)
 		})
 	})
 })
diff --git a/pkg/agent/controller/reconciliation_test.go b/pkg/agent/controller/reconciliation_test.go
index 008aa8a..81eddbe 100644
--- a/pkg/agent/controller/reconciliation_test.go
+++ b/pkg/agent/controller/reconciliation_test.go
@@ -19,15 +19,20 @@ limitations under the License.
 package controller_test
 
 import (
+	"context"
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"github.com/submariner-io/admiral/pkg/syncer/test"
+	testutil "github.com/submariner-io/admiral/pkg/test"
 	"github.com/submariner-io/lighthouse/pkg/constants"
 	corev1 "k8s.io/api/core/v1"
 	discovery "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/client-go/dynamic/fake"
+	"k8s.io/client-go/kubernetes/scheme"
 	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
@@ -45,95 +50,150 @@ var _ = Describe("Reconciliation", func() {
 
 	JustBeforeEach(func() {
 		t.justBeforeEach()
-		t.createService()
-		t.createServiceExport()
+		t.cluster1.createEndpoints()
+		t.cluster1.createService()
+		t.cluster1.createServiceExport()
 	})
 
 	AfterEach(func() {
 		t.afterEach()
 	})
 
-	When("a local headless ServiceImport is stale on startup due to a missed ServiceExport delete event", func() {
-		BeforeEach(func() {
-			t.service.Spec.ClusterIP = corev1.ClusterIPNone
-		})
+	Context("on restart after a service was exported", func() {
+		It("should retain the exported resources on reconciliation", func() {
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
 
-		JustBeforeEach(func() {
-			t.createEndpoints()
-		})
+			localServiceImport := t.cluster1.findLocalServiceImport()
+			Expect(localServiceImport).ToNot(BeNil())
 
-		It("should delete the ServiceImport and EndpointSlice on reconciliation", func() {
-			serviceImport := t.cluster1.awaitServiceImport(t.service, mcsv1a1.Headless, "")
-			endpointSlice := t.cluster1.awaitEndpointSlice(t)
+			localEndpointSlice := t.cluster1.findLocalEndpointSlice()
+			Expect(localEndpointSlice).ToNot(BeNil())
+
+			brokerServiceImports, err := t.brokerServiceImportClient.Namespace(test.RemoteNamespace).List(context.TODO(), metav1.ListOptions{})
+			Expect(err).To(Succeed())
+
+			brokerEndpointSlices, err := t.brokerEndpointSliceClient.List(context.TODO(), metav1.ListOptions{})
+			Expect(err).To(Succeed())
 
 			t.afterEach()
 			t = newTestDiver()
 
-			test.CreateResource(t.cluster1.localServiceImportClient, serviceImport)
-			test.CreateResource(t.cluster1.localEndpointSliceClient, endpointSlice)
-			t.createService()
+			test.CreateResource(t.cluster1.localServiceImportClient.Namespace(test.LocalNamespace), localServiceImport)
+			test.CreateResource(t.cluster1.localEndpointSliceClient, localEndpointSlice)
+
+			for i := range brokerServiceImports.Items {
+				test.CreateResource(t.brokerServiceImportClient.Namespace(test.RemoteNamespace), &brokerServiceImports.Items[i])
+			}
+
+			for i := range brokerEndpointSlices.Items {
+				test.CreateResource(t.brokerEndpointSliceClient, &brokerEndpointSlices.Items[i])
+			}
+
+			t.cluster1.createEndpoints()
+			t.cluster1.createService()
+			t.cluster1.createServiceExport()
+
 			t.cluster1.start(t, *t.syncerConfig)
+			t.cluster2.start(t, *t.syncerConfig)
+
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+
+			testutil.EnsureNoActionsForResource(&t.cluster1.localDynClient.Fake, "serviceimports", "delete")
+			testutil.EnsureNoActionsForResource(&t.cluster1.localDynClient.Fake, "endpointslices", "delete")
 
-			t.awaitServiceUnexported()
-			t.awaitNoEndpointSlice(t.cluster1.localEndpointSliceClient)
+			brokerDynClient := t.syncerConfig.BrokerClient.(*fake.FakeDynamicClient)
+			testutil.EnsureNoActionsForResource(&brokerDynClient.Fake, "serviceimports", "delete")
+			testutil.EnsureNoActionsForResource(&brokerDynClient.Fake, "endpointslices", "delete")
 		})
 	})
 
-	When("a local ServiceImport is stale on startup due to a missed Service delete event", func() {
-		It("should delete the ServiceImport on reconciliation", func() {
-			serviceImport := t.cluster1.awaitServiceImport(t.service, mcsv1a1.ClusterSetIP, t.service.Spec.ClusterIP)
+	When("a local ServiceImport is stale on startup due to a missed ServiceExport delete event", func() {
+		It("should unexport the service on reconciliation", func() {
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+
+			serviceImport := t.cluster1.findLocalServiceImport()
+			Expect(serviceImport).ToNot(BeNil())
+
+			endpointSlice := t.cluster1.findLocalEndpointSlice()
+			Expect(endpointSlice).ToNot(BeNil())
 
 			t.afterEach()
 			t = newTestDiver()
 
-			test.CreateResource(t.cluster1.localServiceImportClient, serviceImport)
-			t.createServiceExport()
+			test.CreateResource(t.cluster1.localServiceImportClient.Namespace(test.LocalNamespace), serviceImport)
+			test.CreateResource(t.cluster1.localEndpointSliceClient, endpointSlice)
+			t.cluster1.createService()
 			t.cluster1.start(t, *t.syncerConfig)
 
-			t.awaitServiceUnexported()
-			t.awaitServiceExportCondition(newServiceExportSyncedCondition(corev1.ConditionFalse, "NoServiceImport"))
+			t.awaitServiceUnexported(&t.cluster1)
 		})
 	})
 
-	When("a synced local ServiceImport is stale in the broker datastore on startup", func() {
-		It("should delete it from the broker datastore on reconciliation", func() {
-			serviceImport := t.awaitBrokerServiceImport(mcsv1a1.ClusterSetIP, t.service.Spec.ClusterIP)
+	When("a local ServiceImport is stale on startup due to a missed Service delete event", func() {
+		It("should unexport the service on reconciliation", func() {
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+			serviceImport := t.cluster1.findLocalServiceImport()
+			Expect(serviceImport).ToNot(BeNil())
 
 			t.afterEach()
 			t = newTestDiver()
 
-			test.CreateResource(t.brokerServiceImportClient, serviceImport)
-			t.justBeforeEach()
+			test.CreateResource(t.cluster1.localServiceImportClient.Namespace(test.LocalNamespace), serviceImport)
+			t.cluster1.createServiceExport()
+			t.cluster1.start(t, *t.syncerConfig)
 
-			t.awaitServiceUnexported()
+			t.cluster1.awaitServiceExportCondition(newServiceExportSyncedCondition(corev1.ConditionFalse, "NoServiceImport"))
+			t.awaitServiceUnexported(&t.cluster1)
 		})
 	})
 
-	When("a synced remote ServiceImport is stale in the local datastore on startup", func() {
+	When("a remote aggregated ServiceImport is stale in the local datastore on startup", func() {
 		It("should delete it from the local datastore on reconciliation", func() {
-			serviceImport := t.cluster2.awaitServiceImport(t.service, mcsv1a1.ClusterSetIP, t.service.Spec.ClusterIP)
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+
+			obj, err := t.cluster2.localServiceImportClient.Namespace(t.cluster1.service.Namespace).Get(context.TODO(),
+				t.cluster1.service.Name, metav1.GetOptions{})
+			Expect(err).To(Succeed())
+
+			serviceImport := &mcsv1a1.ServiceImport{}
+			Expect(scheme.Scheme.Convert(obj, serviceImport, nil)).To(Succeed())
 
 			t.afterEach()
 			t = newTestDiver()
 
-			test.CreateResource(t.cluster2.localServiceImportClient, serviceImport)
+			test.CreateResource(t.cluster2.localServiceImportClient.Namespace(t.cluster1.service.Namespace), serviceImport)
 			t.cluster2.start(t, *t.syncerConfig)
 
-			t.awaitServiceUnexported()
+			t.awaitNoAggregatedServiceImport(&t.cluster1)
 		})
 	})
 
-	When("a synced local EndpointSlice is stale in the broker datastore on startup", func() {
-		BeforeEach(func() {
-			t.service.Spec.ClusterIP = corev1.ClusterIPNone
-		})
+	When("a remote aggregated ServiceImport in the broker datastore contains a stale cluster name on startup", func() {
+		It("should delete it on reconciliation", func() {
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
 
-		JustBeforeEach(func() {
-			t.createEndpoints()
+			brokerServiceImports, err := t.brokerServiceImportClient.Namespace(test.RemoteNamespace).List(context.TODO(), metav1.ListOptions{})
+			Expect(err).To(Succeed())
+
+			t.afterEach()
+			t = newTestDiver()
+
+			for i := range brokerServiceImports.Items {
+				test.CreateResource(t.brokerServiceImportClient.Namespace(test.RemoteNamespace), &brokerServiceImports.Items[i])
+			}
+
+			t.justBeforeEach()
+
+			t.awaitNoAggregatedServiceImport(&t.cluster1)
 		})
+	})
 
+	When("a local EndpointSlice is stale in the broker datastore on startup", func() {
 		It("should delete it from the broker datastore on reconciliation", func() {
-			endpointSlice := t.awaitBrokerEndpointSlice()
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+			endpointSlice := findEndpointSlice(t.brokerEndpointSliceClient, t.cluster1.endpoints.Namespace,
+				t.cluster1.endpoints.Name, t.cluster1.clusterID)
+			Expect(endpointSlice).ToNot(BeNil())
 
 			t.afterEach()
 			t = newTestDiver()
@@ -141,22 +201,16 @@ var _ = Describe("Reconciliation", func() {
 			test.CreateResource(t.brokerEndpointSliceClient, endpointSlice)
 			t.justBeforeEach()
 
-			t.awaitNoEndpointSlice(t.brokerEndpointSliceClient)
-			t.awaitNoEndpointSlice(t.cluster2.localEndpointSliceClient)
+			t.awaitNoEndpointSlice(&t.cluster1)
 		})
 	})
 
-	When("a synced remote EndpointSlice is stale in the local datastore on startup", func() {
-		BeforeEach(func() {
-			t.service.Spec.ClusterIP = corev1.ClusterIPNone
-		})
-
-		JustBeforeEach(func() {
-			t.createEndpoints()
-		})
-
+	When("a remote EndpointSlice is stale in the local datastore on startup", func() {
 		It("should delete it from the local datastore on reconciliation", func() {
-			endpointSlice := t.cluster2.awaitEndpointSlice(t)
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+			endpointSlice := findEndpointSlice(t.cluster2.localEndpointSliceClient, t.cluster1.endpoints.Namespace,
+				t.cluster1.endpoints.Name, t.cluster1.clusterID)
+			Expect(endpointSlice).ToNot(BeNil())
 
 			t.afterEach()
 			t = newTestDiver()
@@ -164,7 +218,8 @@ var _ = Describe("Reconciliation", func() {
 			test.CreateResource(t.cluster2.localEndpointSliceClient, endpointSlice)
 			t.cluster2.start(t, *t.syncerConfig)
 
-			t.awaitNoEndpointSlice(t.cluster2.localEndpointSliceClient)
+			awaitNoEndpointSlice(t.cluster2.localEndpointSliceClient, t.cluster1.service.Namespace,
+				t.cluster1.service.Name, t.cluster1.clusterID)
 		})
 	})
 
@@ -197,10 +252,20 @@ var _ = Describe("Reconciliation", func() {
 	})
 
 	When("a local ServiceImport with the legacy source labels exists after restart", func() {
-		var serviceImport *mcsv1a1.ServiceImport
+		var (
+			serviceImport        *mcsv1a1.ServiceImport
+			brokerServiceImports *unstructured.UnstructuredList
+		)
 
 		JustBeforeEach(func() {
-			serviceImport = t.cluster1.awaitServiceImport(t.service, mcsv1a1.ClusterSetIP, t.service.Spec.ClusterIP)
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+			serviceImport = t.cluster1.findLocalServiceImport()
+			Expect(serviceImport).ToNot(BeNil())
+
+			var err error
+			brokerServiceImports, err = t.brokerServiceImportClient.Namespace(test.RemoteNamespace).List(context.TODO(), metav1.ListOptions{})
+
+			Expect(err).To(Succeed())
 
 			t.afterEach()
 			t = newTestDiver()
@@ -211,8 +276,8 @@ var _ = Describe("Reconciliation", func() {
 			serviceImport.Labels[legacySourceClusterLabel] = serviceImport.Labels[constants.MCSLabelSourceCluster]
 			delete(serviceImport.Labels, constants.MCSLabelSourceCluster)
 
-			t.createService()
-			t.createEndpoints()
+			t.cluster1.createService()
+			t.cluster1.createEndpoints()
 
 			By("Restarting controller")
 		})
@@ -227,48 +292,38 @@ var _ = Describe("Reconciliation", func() {
 			// haven't been migrated yet. This occurs when the ServiceExport is processed so to force the sequencing
 			// create the ServiceImport with the legacy labels first then create the ServiceExport.
 
-			test.CreateResource(t.cluster1.localServiceImportClient, serviceImport)
+			test.CreateResource(t.cluster1.localServiceImportClient.Namespace(test.LocalNamespace), serviceImport)
 
 			// The ServiceImport still has the legacy labels so shouldn't be synced to the broker yet.
-			Eventually(func() *mcsv1a1.ServiceImport {
-				return findServiceImport(t.brokerServiceImportClient, t.service.Namespace, t.service.Name, legacySourceNameLabel)
-			}, 5*time.Second).Should(BeNil(), "Unexpected ServiceImport found")
+			Eventually(func() int {
+				list, _ := t.brokerServiceImportClient.Namespace(test.RemoteNamespace).List(context.TODO(), metav1.ListOptions{})
+				return len(list.Items)
+			}, 5*time.Second).Should(BeZero(), "Unexpected ServiceImport found")
 
 			// The EndpointSlice shouldn't be created yet since the ServiceImport still has the legacy source cluster label.
 			Consistently(func() *discovery.EndpointSlice {
-				return findEndpointSlice(t.cluster1.localEndpointSliceClient, t.endpoints.Namespace, t.endpoints.Name)
+				return t.cluster1.findLocalEndpointSlice()
 			}).Should(BeNil(), "Unexpected EndpointSlice found")
 
 			By("Create the ServiceExport")
 
-			t.createServiceExport()
+			t.cluster1.createServiceExport()
 
-			serviceImport = t.cluster1.awaitServiceImport(t.service, mcsv1a1.ClusterSetIP, t.service.Spec.ClusterIP)
-			Expect(serviceImport.Labels).ToNot(HaveKey(legacySourceNameLabel))
-			Expect(serviceImport.Labels).ToNot(HaveKey(legacySourceClusterLabel))
-
-			t.awaitServiceExported(t.service.Spec.ClusterIP)
-			t.awaitEndpointSlice()
+			t.awaitNoEndpointSlice(&t.cluster1)
 		})
 
 		Context("and the ServiceExport no longer exists", func() {
 			It("should delete the ServiceImport on reconciliation", func() {
-				test.CreateResource(t.cluster1.localServiceImportClient, serviceImport)
+				test.CreateResource(t.cluster1.localServiceImportClient.Namespace(test.LocalNamespace), serviceImport)
 
-				brokerSI := serviceImport.DeepCopy()
-				brokerSI.Namespace = test.RemoteNamespace
-				test.CreateResource(t.brokerServiceImportClient, brokerSI)
+				for i := range brokerServiceImports.Items {
+					test.CreateResource(t.brokerServiceImportClient.Namespace(test.RemoteNamespace), &brokerServiceImports.Items[i])
+				}
 
 				t.cluster1.start(t, *t.syncerConfig)
+				t.cluster2.start(t, *t.syncerConfig)
 
-				Eventually(func() *mcsv1a1.ServiceImport {
-					return findServiceImport(t.cluster1.localServiceImportClient, t.service.Namespace, t.service.Name,
-						legacySourceNameLabel)
-				}, 5*time.Second).Should(BeNil(), "Unexpected ServiceImport found")
-
-				Eventually(func() *mcsv1a1.ServiceImport {
-					return findServiceImport(t.brokerServiceImportClient, t.service.Namespace, t.service.Name, legacySourceNameLabel)
-				}, 5*time.Second).Should(BeNil(), "Unexpected ServiceImport found")
+				t.awaitNoAggregatedServiceImport(&t.cluster1)
 			})
 		})
 	})
diff --git a/pkg/agent/controller/service_export_failures_test.go b/pkg/agent/controller/service_export_failures_test.go
index eb7bb3c..c9aa492 100644
--- a/pkg/agent/controller/service_export_failures_test.go
+++ b/pkg/agent/controller/service_export_failures_test.go
@@ -22,6 +22,9 @@ import (
 	"errors"
 
 	. "github.com/onsi/ginkgo/v2"
+	"github.com/submariner-io/admiral/pkg/fake"
+	"github.com/submariner-io/lighthouse/pkg/constants"
+	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 )
@@ -35,23 +38,90 @@ var _ = Describe("Service export failures", func() {
 
 	JustBeforeEach(func() {
 		t.justBeforeEach()
-		t.createService()
-		t.createEndpoints()
-		t.createServiceExport()
+		t.cluster1.createService()
+		t.cluster1.createEndpoints()
+		t.cluster1.createServiceExport()
 	})
 
 	AfterEach(func() {
 		t.afterEach()
 	})
 
+	When("the local ServiceImport creation initially fails", func() {
+		BeforeEach(func() {
+			t.cluster1.localServiceImportReactor.SetFailOnCreate(errors.New("mock create error"))
+		})
+
+		It("should eventually export the service", func() {
+			t.cluster1.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionTrue, ""))
+			t.cluster1.ensureNoServiceExportCondition(constants.ServiceExportSynced)
+
+			t.cluster1.localServiceImportReactor.SetFailOnCreate(nil)
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+		})
+	})
+
+	When("the aggregated ServiceImport creation initially fails", func() {
+		BeforeEach(func() {
+			t.brokerServiceImportReactor.SetFailOnCreate(errors.New("mock create error"))
+		})
+
+		It("should eventually export the service", func() {
+			t.cluster1.awaitServiceExportCondition(newServiceExportSyncedCondition(corev1.ConditionFalse, "ExportFailed"))
+
+			t.brokerServiceImportReactor.SetFailOnCreate(nil)
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+		})
+	})
+
+	When("the aggregated ServiceImport update initially fails", func() {
+		BeforeEach(func() {
+			t.brokerServiceImportReactor.SetFailOnUpdate(errors.New("mock update error"))
+		})
+
+		It("should eventually export the service", func() {
+			t.cluster1.awaitServiceExportCondition(newServiceExportSyncedCondition(corev1.ConditionFalse, "ExportFailed"))
+
+			t.brokerServiceImportReactor.SetFailOnUpdate(nil)
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+		})
+	})
+
+	When("the aggregated ServiceImport delete initially fails", func() {
+		BeforeEach(func() {
+			t.brokerServiceImportReactor.SetFailOnDelete(errors.New("mock delete error"))
+			t.brokerServiceImportReactor.SetResetOnFailure(true)
+		})
+
+		It("should eventually unexport the service", func() {
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+			t.cluster1.deleteServiceExport()
+			t.awaitServiceUnexported(&t.cluster1)
+		})
+	})
+
 	When("a conflict initially occurs when updating the ServiceExport status", func() {
 		BeforeEach(func() {
-			t.cluster1.localServiceExportClient.FailOnUpdate = apierrors.NewConflict(schema.GroupResource{}, t.serviceExport.Name,
-				errors.New("fake conflict"))
+			t.cluster1.localServiceImportReactor.SetFailOnUpdate(apierrors.NewConflict(schema.GroupResource{}, t.cluster1.serviceExport.Name,
+				errors.New("fake conflict")))
+			t.cluster1.localServiceImportReactor.SetResetOnFailure(true)
 		})
 
 		It("should eventually update the ServiceExport status", func() {
-			t.awaitServiceExported(t.service.Spec.ClusterIP)
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+		})
+	})
+
+	When("deleting the local EndpointSlice on unexport initially fails", func() {
+		JustBeforeEach(func() {
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+
+			fake.FailOnAction(&t.cluster1.localDynClient.Fake, "endpointslices", "delete-collection", nil, true)
+		})
+
+		It("should eventually unexport the service", func() {
+			t.cluster1.deleteServiceExport()
+			t.awaitServiceUnexported(&t.cluster1)
 		})
 	})
 })
diff --git a/pkg/agent/controller/service_import_sync_test.go b/pkg/agent/controller/service_import_sync_test.go
deleted file mode 100644
index 50e21f5..0000000
--- a/pkg/agent/controller/service_import_sync_test.go
+++ /dev/null
@@ -1,225 +0,0 @@
-/*
-SPDX-License-Identifier: Apache-2.0
-
-Copyright Contributors to the Submariner project.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package controller_test
-
-import (
-	. "github.com/onsi/ginkgo/v2"
-	"github.com/submariner-io/admiral/pkg/syncer/test"
-	"github.com/submariner-io/lighthouse/pkg/constants"
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
-)
-
-var _ = Describe("ServiceImport syncing", func() {
-	var t *testDriver
-
-	BeforeEach(func() {
-		t = newTestDiver()
-	})
-
-	JustBeforeEach(func() {
-		t.justBeforeEach()
-	})
-
-	AfterEach(func() {
-		t.afterEach()
-	})
-
-	When("a ServiceExport is created", func() {
-		Context("and the Service already exists", func() {
-			It("should correctly sync a ServiceImport and update the ServiceExport status", func() {
-				t.createService()
-				t.createServiceExport()
-				t.awaitServiceExported(t.service.Spec.ClusterIP)
-			})
-		})
-
-		Context("and the Service doesn't initially exist", func() {
-			It("should initially update the ServiceExport status to Initialized and eventually sync a ServiceImport", func() {
-				t.createServiceExport()
-				t.awaitServiceUnavailableStatus()
-
-				t.createService()
-				t.awaitServiceExported(t.service.Spec.ClusterIP)
-			})
-		})
-	})
-
-	When("a ServiceExport is deleted after a ServiceImport is synced", func() {
-		It("should delete the ServiceImport", func() {
-			t.createService()
-			t.createServiceExport()
-			t.awaitServiceExported(t.service.Spec.ClusterIP)
-
-			t.deleteServiceExport()
-			t.awaitServiceUnexported()
-		})
-	})
-
-	When("an exported Service is deleted and recreated while the ServiceExport still exists", func() {
-		It("should delete and recreate the ServiceImport", func() {
-			t.createService()
-			t.createServiceExport()
-			t.awaitServiceExported(t.service.Spec.ClusterIP)
-
-			t.deleteService()
-			t.awaitServiceUnexported()
-			t.awaitServiceUnavailableStatus()
-			t.awaitServiceExportCondition(newServiceExportSyncedCondition(corev1.ConditionFalse, "NoServiceImport"))
-
-			t.createService()
-			t.awaitServiceExported(t.service.Spec.ClusterIP)
-		})
-	})
-
-	When("the type of an exported Service is updated to an unsupported type", func() {
-		It("should delete the ServiceImport and update the ServiceExport status", func() {
-			t.createService()
-			t.createServiceExport()
-			t.awaitServiceExported(t.service.Spec.ClusterIP)
-
-			t.service.Spec.Type = corev1.ServiceTypeNodePort
-			t.updateService()
-
-			t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "UnsupportedServiceType"))
-			t.awaitServiceUnexported()
-			t.awaitServiceExportCondition(newServiceExportSyncedCondition(corev1.ConditionFalse, "NoServiceImport"))
-		})
-	})
-
-	When("the ServiceImport sync initially fails", func() {
-		BeforeEach(func() {
-			t.cluster1.localServiceImportClient.PersistentFailOnCreate.Store("mock create error")
-		})
-
-		It("should eventually update the ServiceExport status to successfully synced", func() {
-			t.createService()
-			t.createServiceExport()
-
-			t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionTrue, ""))
-			t.ensureNoServiceExportCondition(constants.ServiceExportSynced)
-
-			t.cluster1.localServiceImportClient.PersistentFailOnCreate.Store("")
-			t.awaitServiceExported(t.service.Spec.ClusterIP)
-		})
-	})
-
-	When("a ServiceExport is created for a Service whose type is unsupported", func() {
-		BeforeEach(func() {
-			t.service.Spec.Type = corev1.ServiceTypeNodePort
-		})
-
-		JustBeforeEach(func() {
-			t.createService()
-			t.createServiceExport()
-		})
-
-		It("should update the ServiceExport status appropriately and not sync a ServiceImport", func() {
-			t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "UnsupportedServiceType"))
-			t.awaitNoServiceImport(t.brokerServiceImportClient)
-			t.ensureNoServiceExportCondition(constants.ServiceExportSynced)
-		})
-
-		Context("and is subsequently updated to a supported type", func() {
-			It("should eventually sync a ServiceImport and update the ServiceExport status appropriately", func() {
-				t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "UnsupportedServiceType"))
-
-				t.service.Spec.Type = corev1.ServiceTypeClusterIP
-				t.updateService()
-
-				t.awaitServiceExported(t.service.Spec.ClusterIP)
-			})
-		})
-	})
-
-	When("a Service has port information", func() {
-		BeforeEach(func() {
-			t.service.Spec.Ports = []corev1.ServicePort{
-				{
-					Name:     "eth0",
-					Protocol: corev1.ProtocolTCP,
-					Port:     123,
-				},
-				{
-					Name:     "eth1",
-					Protocol: corev1.ProtocolTCP,
-					Port:     1234,
-				},
-			}
-		})
-
-		It("should set the appropriate port information in the ServiceImport", func() {
-			t.createService()
-			t.createServiceExport()
-			t.awaitServiceExported(t.service.Spec.ClusterIP)
-		})
-	})
-
-	When("two Services with the same name in different namespaces are exported", func() {
-		It("should sync ServiceImport and EndpointSlice resources for each", func() {
-			t.createEndpoints()
-			t.createService()
-			t.createServiceExport()
-			t.awaitServiceExported(t.service.Spec.ClusterIP)
-			t.awaitEndpointSlice()
-
-			service := &corev1.Service{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      t.service.Name,
-					Namespace: "other-service-ns",
-				},
-				Spec: corev1.ServiceSpec{
-					ClusterIP: "10.253.9.2",
-				},
-			}
-
-			serviceExport := &mcsv1a1.ServiceExport{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      service.Name,
-					Namespace: service.Namespace,
-				},
-			}
-
-			endpoints := &corev1.Endpoints{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      service.Name,
-					Namespace: service.Namespace,
-				},
-			}
-
-			test.CreateResource(dynamicEndpointsClient(t.cluster1.localDynClient, endpoints.Namespace), endpoints)
-			test.CreateResource(t.cluster1.dynamicServiceClient().Namespace(service.Namespace), service)
-			test.CreateResource(serviceExportClient(t.cluster1.localDynClient, service.Namespace), serviceExport)
-
-			t.cluster1.awaitServiceImport(service, mcsv1a1.ClusterSetIP, service.Spec.ClusterIP)
-			awaitServiceImport(t.brokerServiceImportClient, service, mcsv1a1.ClusterSetIP, service.Spec.ClusterIP)
-			t.cluster2.awaitServiceImport(service, mcsv1a1.ClusterSetIP, service.Spec.ClusterIP)
-
-			awaitEndpointSlice(endpointSliceClient(t.cluster1.localDynClient, service.Namespace), endpoints.Namespace, endpoints.Name)
-			awaitEndpointSlice(t.brokerEndpointSliceClient, endpoints.Namespace, endpoints.Name)
-			awaitEndpointSlice(endpointSliceClient(t.cluster2.localDynClient, service.Namespace), endpoints.Namespace, endpoints.Name)
-
-			// Ensure the resources for the first Service weren't overwritten
-			t.cluster1.awaitServiceImport(t.service, mcsv1a1.ClusterSetIP, t.service.Spec.ClusterIP)
-			t.awaitBrokerServiceImport(mcsv1a1.ClusterSetIP, t.service.Spec.ClusterIP)
-			t.awaitEndpointSlice()
-		})
-	})
-})
diff --git a/test/e2e/discovery/headless_services.go b/test/e2e/discovery/headless_services.go
index 9c02d97..60959ff 100644
--- a/test/e2e/discovery/headless_services.go
+++ b/test/e2e/discovery/headless_services.go
@@ -97,7 +97,7 @@ func RunHeadlessDiscoveryTest(f *lhframework.Framework) {
 		clusterBName, false, false, true)
 
 	f.DeleteServiceExport(framework.ClusterB, nginxHeadlessClusterB.Name, nginxHeadlessClusterB.Namespace)
-	f.AwaitServiceImportCount(framework.ClusterA, nginxHeadlessClusterB.Name, nginxHeadlessClusterB.Namespace, 0)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxHeadlessClusterB, 0)
 
 	f.VerifyIPsWithDig(framework.ClusterA, nginxHeadlessClusterB, netshootPodList, ipList, checkedDomains,
 		"", false)
@@ -117,6 +117,7 @@ func RunHeadlessDiscoveryLocalAndRemoteTest(f *lhframework.Framework) {
 
 	f.NewServiceExport(framework.ClusterB, nginxHeadlessClusterB.Name, nginxHeadlessClusterB.Namespace)
 	f.AwaitServiceExportedStatusCondition(framework.ClusterB, nginxHeadlessClusterB.Name, nginxHeadlessClusterB.Namespace)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxHeadlessClusterB, 1)
 
 	By(fmt.Sprintf("Creating an Nginx Deployment on %q", clusterAName))
 	f.NewNginxDeployment(framework.ClusterA)
@@ -126,6 +127,7 @@ func RunHeadlessDiscoveryLocalAndRemoteTest(f *lhframework.Framework) {
 
 	f.NewServiceExport(framework.ClusterA, nginxHeadlessClusterA.Name, nginxHeadlessClusterA.Namespace)
 	f.AwaitServiceExportedStatusCondition(framework.ClusterA, nginxHeadlessClusterA.Name, nginxHeadlessClusterA.Namespace)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxHeadlessClusterA, 2)
 
 	By(fmt.Sprintf("Creating a Netshoot Deployment on %q", clusterAName))
 
@@ -146,7 +148,7 @@ func RunHeadlessDiscoveryLocalAndRemoteTest(f *lhframework.Framework) {
 	verifyHeadlessSRVRecordsWithDig(f.Framework, framework.ClusterA, nginxHeadlessClusterB, netshootPodList, hostNameListA, checkedDomains,
 		clusterAName, true, false, true)
 	f.DeleteServiceExport(framework.ClusterB, nginxHeadlessClusterB.Name, nginxHeadlessClusterB.Namespace)
-	f.AwaitServiceImportCount(framework.ClusterA, nginxHeadlessClusterB.Name, nginxHeadlessClusterB.Namespace, 1)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxHeadlessClusterB, 1)
 
 	f.VerifyIPsWithDig(framework.ClusterA, nginxHeadlessClusterB, netshootPodList, ipListB, checkedDomains,
 		"", false)
diff --git a/test/e2e/discovery/service_discovery.go b/test/e2e/discovery/service_discovery.go
index 4483910..510270e 100644
--- a/test/e2e/discovery/service_discovery.go
+++ b/test/e2e/discovery/service_discovery.go
@@ -137,7 +137,7 @@ func RunServiceDiscoveryTest(f *lhframework.Framework) {
 	Expect(err).NotTo(HaveOccurred())
 
 	nginxServiceClusterB = svc
-	f.AwaitServiceImportIP(framework.ClusterB, framework.ClusterA, nginxServiceClusterB)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxServiceClusterB, 1)
 	f.AwaitEndpointSlices(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 1, 1)
 	f.AwaitEndpointSlices(framework.ClusterA, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 1, 1)
 
@@ -150,7 +150,7 @@ func RunServiceDiscoveryTest(f *lhframework.Framework) {
 		false, true)
 
 	f.DeleteServiceExport(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace)
-	f.AwaitServiceImportDelete(framework.ClusterA, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxServiceClusterB, 0)
 
 	f.DeleteService(framework.ClusterB, nginxServiceClusterB.Name)
 
@@ -200,7 +200,7 @@ func RunServiceDiscoveryLocalTest(f *lhframework.Framework) {
 	Expect(err).NotTo(HaveOccurred())
 
 	nginxServiceClusterB = svc
-	f.AwaitServiceImportIP(framework.ClusterB, framework.ClusterA, nginxServiceClusterB)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxServiceClusterB, 1)
 	f.AwaitEndpointSlices(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 1, 1)
 	f.AwaitEndpointSlices(framework.ClusterA, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 1, 1)
 
@@ -208,7 +208,7 @@ func RunServiceDiscoveryLocalTest(f *lhframework.Framework) {
 		"", true)
 
 	f.DeleteServiceExport(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace)
-	f.AwaitServiceImportDelete(framework.ClusterA, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxServiceClusterB, 0)
 
 	f.DeleteService(framework.ClusterB, nginxServiceClusterB.Name)
 
@@ -236,7 +236,7 @@ func RunServiceExportTest(f *lhframework.Framework) {
 	Expect(err).NotTo(HaveOccurred())
 
 	nginxServiceClusterB = svc
-	f.AwaitServiceImportIP(framework.ClusterB, framework.ClusterA, nginxServiceClusterB)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxServiceClusterB, 1)
 	f.AwaitEndpointSlices(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 1, 1)
 	f.AwaitEndpointSlices(framework.ClusterA, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 1, 1)
 
@@ -244,7 +244,7 @@ func RunServiceExportTest(f *lhframework.Framework) {
 		"", true)
 
 	f.DeleteServiceExport(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace)
-	f.AwaitServiceImportDelete(framework.ClusterA, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxServiceClusterB, 0)
 
 	f.VerifyIPWithDig(framework.ClusterA, nginxServiceClusterB, netshootPodList, checkedDomains, "", "", true)
 }
@@ -271,7 +271,7 @@ func RunServicesPodAvailabilityTest(f *lhframework.Framework) {
 	Expect(err).NotTo(HaveOccurred())
 
 	nginxServiceClusterB = svc
-	f.AwaitServiceImportIP(framework.ClusterB, framework.ClusterA, nginxServiceClusterB)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxServiceClusterB, 1)
 	f.AwaitEndpointSlices(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 1, 1)
 	f.AwaitEndpointSlices(framework.ClusterA, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 1, 1)
 
@@ -329,14 +329,14 @@ func RunServicesPodAvailabilityMultiClusterTest(f *lhframework.Framework) {
 	Expect(err).NotTo(HaveOccurred())
 
 	nginxServiceClusterC = svc
-	f.AwaitServiceImportIP(framework.ClusterC, framework.ClusterC, nginxServiceClusterC)
+	f.AwaitAggregatedServiceImport(framework.ClusterC, nginxServiceClusterC, 2)
 	f.AwaitEndpointSlices(framework.ClusterC, nginxServiceClusterC.Name, nginxServiceClusterC.Namespace, 2, 2)
 
 	svc, err = f.GetService(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace)
 	Expect(err).NotTo(HaveOccurred())
 
 	nginxServiceClusterB = svc
-	f.AwaitServiceImportIP(framework.ClusterB, framework.ClusterB, nginxServiceClusterB)
+	f.AwaitAggregatedServiceImport(framework.ClusterB, nginxServiceClusterB, 2)
 	f.AwaitEndpointSlices(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 2, 2)
 
 	f.VerifyServiceIPWithDig(framework.ClusterA, framework.ClusterB, nginxServiceClusterB, netshootPodList, checkedDomains,
@@ -403,7 +403,7 @@ func RunServiceDiscoveryClusterNameTest(f *lhframework.Framework) {
 	Expect(err).NotTo(HaveOccurred())
 
 	nginxServiceClusterA = svc
-	f.AwaitServiceImportIP(framework.ClusterA, framework.ClusterA, nginxServiceClusterA)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxServiceClusterA, 2)
 	f.AwaitEndpointSlices(framework.ClusterA, nginxServiceClusterA.Name, nginxServiceClusterA.Namespace, 2, 2)
 
 	f.VerifyServiceIPWithDig(framework.ClusterA, framework.ClusterA, nginxServiceClusterA, netshootPodList, checkedDomains,
@@ -415,7 +415,7 @@ func RunServiceDiscoveryClusterNameTest(f *lhframework.Framework) {
 	Expect(err).NotTo(HaveOccurred())
 
 	nginxServiceClusterB = svc
-	f.AwaitServiceImportIP(framework.ClusterB, framework.ClusterA, nginxServiceClusterB)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxServiceClusterB, 2)
 	f.AwaitEndpointSlices(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 2, 2)
 
 	f.VerifyServiceIPWithDig(framework.ClusterA, framework.ClusterB, nginxServiceClusterB, netshootPodList, checkedDomains,
@@ -462,14 +462,14 @@ func RunServiceDiscoveryRoundRobinTest(f *lhframework.Framework) {
 	Expect(err).NotTo(HaveOccurred())
 
 	nginxServiceClusterC = svc
-	f.AwaitServiceImportIP(framework.ClusterC, framework.ClusterC, nginxServiceClusterC)
+	f.AwaitAggregatedServiceImport(framework.ClusterC, nginxServiceClusterC, 2)
 	f.AwaitEndpointSlices(framework.ClusterC, nginxServiceClusterC.Name, nginxServiceClusterC.Namespace, 2, 2)
 
 	svc, err = f.GetService(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace)
 	Expect(err).NotTo(HaveOccurred())
 
 	nginxServiceClusterB = svc
-	f.AwaitServiceImportIP(framework.ClusterB, framework.ClusterB, nginxServiceClusterB)
+	f.AwaitAggregatedServiceImport(framework.ClusterB, nginxServiceClusterB, 2)
 	f.AwaitEndpointSlices(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 2, 2)
 
 	var serviceIPList []string
@@ -507,7 +507,7 @@ func RunServicesClusterAvailabilityMultiClusterTest(f *lhframework.Framework) {
 	Expect(err).NotTo(HaveOccurred())
 
 	nginxServiceClusterB = svc
-	f.AwaitServiceImportIP(framework.ClusterB, framework.ClusterA, nginxServiceClusterB)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxServiceClusterB, 1)
 	f.AwaitEndpointSlices(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 1, 1)
 	f.AwaitEndpointSlices(framework.ClusterA, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 1, 1)
 
@@ -525,7 +525,7 @@ func RunServicesClusterAvailabilityMultiClusterTest(f *lhframework.Framework) {
 	Expect(err).NotTo(HaveOccurred())
 
 	nginxServiceClusterC = svc
-	f.AwaitServiceImportIP(framework.ClusterC, framework.ClusterA, nginxServiceClusterC)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxServiceClusterC, 2)
 	f.AwaitEndpointSlices(framework.ClusterA, nginxServiceClusterC.Name, nginxServiceClusterC.Namespace, 2, 2)
 
 	f.VerifyServiceIPWithDig(framework.ClusterA, framework.ClusterB, nginxServiceClusterB, netshootPodList,
diff --git a/test/e2e/discovery/statefulsets.go b/test/e2e/discovery/statefulsets.go
index 4af90f1..194635a 100644
--- a/test/e2e/discovery/statefulsets.go
+++ b/test/e2e/discovery/statefulsets.go
@@ -78,7 +78,7 @@ func RunSSDiscoveryTest(f *lhframework.Framework) {
 	verifyEndpointSlices(f, framework.ClusterA, netshootPodList, endpointSlices, nginxServiceClusterB, 1, true, clusterAName)
 
 	f.DeleteServiceExport(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace)
-	f.AwaitServiceImportCount(framework.ClusterA, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 0)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxServiceClusterB, 0)
 	f.AwaitEndpointSlices(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 0, 0)
 
 	verifyEndpointSlices(f, framework.ClusterA, netshootPodList, endpointSlices, nginxServiceClusterB, 1, false, clusterAName)
@@ -124,7 +124,7 @@ func RunSSDiscoveryLocalTest(f *lhframework.Framework) {
 	verifyEndpointSlices(f, framework.ClusterA, netshootPodList, endpointSlices, nginxServiceClusterB, 2, true, clusterAName)
 
 	f.DeleteServiceExport(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace)
-	f.AwaitServiceImportCount(framework.ClusterA, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 1)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxServiceClusterB, 1)
 
 	verifyCount := 0
 
@@ -142,7 +142,7 @@ func RunSSDiscoveryLocalTest(f *lhframework.Framework) {
 	Expect(verifyCount).To(Equal(2), "Mismatch in count of IPs to be validated with dig")
 
 	f.DeleteServiceExport(framework.ClusterA, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace)
-	f.AwaitServiceImportCount(framework.ClusterA, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 0)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxServiceClusterB, 0)
 	f.AwaitEndpointSlices(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 0, 0)
 }
 
@@ -186,7 +186,7 @@ func RunSSPodsAvailabilityTest(f *lhframework.Framework) {
 	}
 
 	f.DeleteServiceExport(framework.ClusterB, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace)
-	f.AwaitServiceImportCount(framework.ClusterA, nginxServiceClusterB.Name, nginxServiceClusterB.Namespace, 0)
+	f.AwaitAggregatedServiceImport(framework.ClusterA, nginxServiceClusterB, 0)
 }
 
 //nolint:unparam //  `targetCluster` always receives `framework.ClusterA`.
diff --git a/test/e2e/framework/framework.go b/test/e2e/framework/framework.go
index 9b2f12f..0360cc7 100644
--- a/test/e2e/framework/framework.go
+++ b/test/e2e/framework/framework.go
@@ -31,6 +31,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	discovery "k8s.io/api/discovery/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
@@ -160,7 +161,10 @@ func (f *Framework) AwaitServiceExportedStatusCondition(cluster framework.Cluste
 			}
 		}
 
-		return false, fmt.Sprintf("ServiceExport %s condition status not found", constants.ServiceExportSynced), nil
+		out, _ := json.MarshalIndent(se.Status.Conditions, "", " ")
+
+		return false, fmt.Sprintf("ServiceExport %s condition status not found. Actual: %s",
+			constants.ServiceExportSynced, out), nil
 	})
 }
 
@@ -177,76 +181,37 @@ func (f *Framework) GetService(cluster framework.ClusterIndex, name, namespace s
 	return framework.KubeClients[cluster].CoreV1().Services(namespace).Get(context.TODO(), name, metav1.GetOptions{})
 }
 
-func (f *Framework) AwaitServiceImportIP(srcCluster, targetCluster framework.ClusterIndex, svc *v1.Service) *mcsv1a1.ServiceImport {
-	serviceIP := f.GetServiceIP(srcCluster, svc, false)
-
-	return f.AwaitServiceImportWithIP(targetCluster, svc, serviceIP)
-}
+func (f *Framework) AwaitAggregatedServiceImport(targetCluster framework.ClusterIndex, svc *v1.Service, clusterCount int) {
+	By(fmt.Sprintf("Retrieving ServiceImport for %q in ns %q on %q", svc.Name, svc.Namespace,
+		framework.TestContext.ClusterIDs[targetCluster]))
 
-func (f *Framework) AwaitServiceImportWithIP(targetCluster framework.ClusterIndex, svc *v1.Service,
-	serviceIP string,
-) *mcsv1a1.ServiceImport {
-	var retServiceImport *mcsv1a1.ServiceImport
+	si := MCSClients[targetCluster].MulticlusterV1alpha1().ServiceImports(svc.Namespace)
 
-	siNamePrefix := svc.Name + "-" + svc.Namespace + "-"
-	si := MCSClients[targetCluster].MulticlusterV1alpha1().ServiceImports(framework.TestContext.SubmarinerNamespace)
-	By(fmt.Sprintf("Retrieving ServiceImport for %s on %q", siNamePrefix, framework.TestContext.ClusterIDs[targetCluster]))
 	framework.AwaitUntil("retrieve ServiceImport", func() (interface{}, error) {
-		return si.List(context.TODO(), metav1.ListOptions{})
-	}, func(result interface{}) (bool, string, error) {
-		siList := result.(*mcsv1a1.ServiceImportList)
-		if len(siList.Items) < 1 {
-			return false, fmt.Sprintf("ServiceImport with name prefix %s not found", siNamePrefix), nil
-		}
-		for i := range siList.Items {
-			si := &siList.Items[i]
-			if strings.HasPrefix(si.Name, siNamePrefix) {
-				if si.Spec.IPs[0] == serviceIP {
-					retServiceImport = &siList.Items[i]
-					return true, "", nil
-				}
-			}
+		obj, err := si.Get(context.TODO(), svc.Name, metav1.GetOptions{})
+		if apierrors.IsNotFound(err) {
+			return nil, nil //nolint:nilnil // Intentional
 		}
 
-		return false, fmt.Sprintf("Failed to find ServiceImport with IP %s", serviceIP), nil
-	})
-
-	return retServiceImport
-}
-
-func (f *Framework) AwaitServiceImportDelete(targetCluster framework.ClusterIndex, name, namespace string) {
-	siNamePrefix := name + "-" + namespace
-	si := MCSClients[targetCluster].MulticlusterV1alpha1().ServiceImports(framework.TestContext.SubmarinerNamespace)
-	framework.AwaitUntil("retrieve ServiceImport", func() (interface{}, error) {
-		return si.List(context.TODO(), metav1.ListOptions{})
+		return obj, err
 	}, func(result interface{}) (bool, string, error) {
-		siList := result.(*mcsv1a1.ServiceImportList)
-		for i := range siList.Items {
-			si := &siList.Items[i]
-			if strings.HasPrefix(si.Name, siNamePrefix) {
-				return false, fmt.Sprintf("ServiceImport with name prefix %s still exists", siNamePrefix), nil
+		if clusterCount == 0 {
+			if result != nil {
+				return false, "ServiceImport still exists", nil
 			}
+
+			return true, "", nil
 		}
 
-		return true, "", nil
-	})
-}
+		if result == nil {
+			return false, "ServiceImport not found", nil
+		}
 
-func (f *Framework) AwaitServiceImportCount(targetCluster framework.ClusterIndex, name, namespace string, count int) {
-	labelMap := map[string]string{
-		mcsv1a1.LabelServiceName:       name,
-		constants.LabelSourceNamespace: namespace,
-	}
-	siListOptions := metav1.ListOptions{
-		LabelSelector: labels.Set(labelMap).String(),
-	}
-	si := MCSClients[targetCluster].MulticlusterV1alpha1().ServiceImports(framework.TestContext.SubmarinerNamespace)
-	framework.AwaitUntil("retrieve ServiceImport", func() (interface{}, error) {
-		return si.List(context.TODO(), siListOptions)
-	}, func(result interface{}) (bool, string, error) {
-		siList := result.(*mcsv1a1.ServiceImportList)
-		if len(siList.Items) != count {
-			return false, fmt.Sprintf("ServiceImport count was %v instead of %v", len(siList.Items), count), nil
+		si := result.(*mcsv1a1.ServiceImport)
+
+		if len(si.Status.Clusters) != clusterCount {
+			return false, fmt.Sprintf("Actual cluster count %d does not match expected %d",
+				len(si.Status.Clusters), clusterCount), nil
 		}
 
 		return true, "", nil
@@ -438,7 +403,7 @@ func create(f *Framework, cluster framework.ClusterIndex, statefulSet *appsv1.St
 }
 
 func (f *Framework) AwaitEndpointSlices(targetCluster framework.ClusterIndex, name, namespace string,
-	expSliceCount, expEpCount int,
+	expSliceCount, expReadyCount int,
 ) (endpointSliceList *discovery.EndpointSliceList) {
 	ep := framework.KubeClients[targetCluster].DiscoveryV1().EndpointSlices(namespace)
 	labelMap := map[string]string{
@@ -455,22 +420,27 @@ func (f *Framework) AwaitEndpointSlices(targetCluster framework.ClusterIndex, na
 	}, func(result interface{}) (bool, string, error) {
 		endpointSliceList = result.(*discovery.EndpointSliceList)
 		sliceCount := 0
-		epCount := 0
+		readyCount := 0
 
 		for i := range endpointSliceList.Items {
 			es := &endpointSliceList.Items[i]
-			if name == "" || strings.HasPrefix(es.Name, name) {
+			if name == "" || es.Labels[mcsv1a1.LabelServiceName] == name {
 				sliceCount++
-				epCount += len(es.Endpoints)
+
+				for j := range es.Endpoints {
+					if es.Endpoints[j].Conditions.Ready == nil || *es.Endpoints[j].Conditions.Ready {
+						readyCount++
+					}
+				}
 			}
 		}
 
 		if expSliceCount != anyCount && sliceCount != expSliceCount {
-			return false, fmt.Sprintf("%d EndpointSlices found when expected %d", len(endpointSliceList.Items), expSliceCount), nil
+			return false, fmt.Sprintf("%d EndpointSlices found when expected %d", sliceCount, expSliceCount), nil
 		}
 
-		if expEpCount != anyCount && epCount != expEpCount {
-			return false, fmt.Sprintf("%d total Endpoints found when expected %d", epCount, expEpCount), nil
+		if expReadyCount != anyCount && readyCount != expReadyCount {
+			return false, fmt.Sprintf("%d ready Endpoints found when expected %d", readyCount, expReadyCount), nil
 		}
 
 		return true, "", nil