Repository: submariner-io/lighthouse
Pull request #310 (2020-09-28 12:53:30)

Add build Make target for building both binaries
Add a "build" Make target to group building all binaries of this repo
under a single well-known command.

Signed-off-by: Daniel Farrell <dfarrell@redhat.com>

dfarrell07: Trying to make all repos conform to the building/testing pattern described in https://github.com/submariner-io/submariner-website/pull/285.

Patch:
diff --git a/.github/stale.yml b/.github/stale.yml
new file mode 100644
index 0000000..41321d4
--- /dev/null
+++ b/.github/stale.yml
@@ -0,0 +1,57 @@
+---
+# Configuration for probot-stale - https://github.com/probot/stale
+
+# Number of days of inactivity before an Issue becomes stale (pull request is overridden later)
+daysUntilStale: 60
+
+# Number of days of inactivity before an Issue or Pull Request with the stale label is closed.
+# Set to false to disable. If disabled, issues still need to be closed manually, but will remain marked as stale.
+daysUntilClose: 7
+
+# Only issues or pull requests with all of these labels are check if stale. Defaults to `[]` (disabled)
+onlyLabels: []
+
+# Issues or Pull Requests with these labels will never be considered stale. Set to `[]` to disable
+exemptLabels:
+  - security
+  - confirmed
+
+# Set to true to ignore issues in a project (defaults to false)
+exemptProjects: false
+
+# Set to true to ignore issues in a milestone (defaults to false)
+exemptMilestones: false
+
+# Set to true to ignore issues with an assignee (defaults to false)
+exemptAssignees: false
+
+# Label to use when marking as stale
+staleLabel: wontfix
+
+# Comment to post when marking as stale. Set to `false` to disable
+markComment: >
+  This issue has been automatically marked as stale because it has not had
+  activity for 60 days. It will be closed if no further activity occurs.
+  Please make a comment if this issue/pr is still valid. Thank you
+  for your contributions.
+
+# Comment to post when removing the stale label.
+# unmarkComment: >
+#   Your comment here.
+
+# Comment to post when closing a stale Issue or Pull Request.
+# closeComment: >
+#   Your comment here.
+
+# Limit the number of actions per hour, from 1-30. Default is 30
+limitPerRun: 30
+
+# Limit to only `issues` or `pulls`
+# only: issues
+
+pulls:
+  daysUntilStale: 30
+  markComment: >
+    This pull request has been automatically marked as stale because it has not had
+    recent activity. It will be closed if no further activity occurs. Thank you
+    for your contributions.
diff --git a/Makefile b/Makefile
index 69d79c7..b2d6e0a 100644
--- a/Makefile
+++ b/Makefile
@@ -29,6 +29,8 @@ package/.image.lighthouse-agent: bin/lighthouse-agent
 
 package/.image.lighthouse-coredns: bin/lighthouse-coredns
 
+build: bin/lighthouse-agent bin/lighthouse-coredns
+
 bin/lighthouse-agent: vendor/modules.txt $(shell find pkg/agent)
 	${SCRIPTS_DIR}/compile.sh $@ pkg/agent/main.go
 
diff --git a/README.md b/README.md
index 31dda1f..4ef39aa 100644
--- a/README.md
+++ b/README.md
@@ -9,7 +9,7 @@ The lighthouse architecture is explained in detail at [Service Discovery](https:
 
 ## Contribute
 
-We welcome any contributions and please refer [Contribution Guide](https://submariner-io.github.io/contributing/lighthouse/) for more details.
+We welcome any contributions. Please refer to the [Contribution Guide](https://submariner.io/contributing/) for more details.
 
 ## Build and Test
 
diff --git a/go.mod b/go.mod
index 4b2a40f..6d7dd5c 100644
--- a/go.mod
+++ b/go.mod
@@ -10,7 +10,7 @@ require (
 	github.com/onsi/ginkgo v1.14.1
 	github.com/onsi/gomega v1.10.2
 	github.com/pkg/errors v0.9.1
-	github.com/submariner-io/admiral v0.6.2
+	github.com/submariner-io/admiral v0.6.4
 	github.com/submariner-io/shipyard v0.6.1
 	k8s.io/api v0.17.2
 	k8s.io/apimachinery v0.17.2
diff --git a/go.sum b/go.sum
index e8b99b5..a6b6446 100644
--- a/go.sum
+++ b/go.sum
@@ -538,10 +538,8 @@ github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJy
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/submariner-io/admiral v0.6.1 h1:EWGnf7Yem6dEvbvF/MaQYcYhZ8rTflPgQukl2IojMhI=
-github.com/submariner-io/admiral v0.6.1/go.mod h1:bPMiLdXmGVj3EHnrPoVzn6jTukKeaH2IoWoZdFg2blM=
-github.com/submariner-io/admiral v0.6.2 h1:6iPGX8gSFQq02nNwkpj/zrqA1CbRkc+Too3TziI2HzA=
-github.com/submariner-io/admiral v0.6.2/go.mod h1:/lm9D8AUs/cZE0lhxBfKDwkaU1XhmLkRnRwoD6BNsfg=
+github.com/submariner-io/admiral v0.6.4 h1:Z+90xbsVqXs6qe3o6L2hscVo3FDz+Twjw+v26fVrQCI=
+github.com/submariner-io/admiral v0.6.4/go.mod h1:/lm9D8AUs/cZE0lhxBfKDwkaU1XhmLkRnRwoD6BNsfg=
 github.com/submariner-io/shipyard v0.6.1 h1:jkZc9WJpK+ZKjHmuDOM7BT0rkp41pOAsEGJtPAqFs14=
 github.com/submariner-io/shipyard v0.6.1/go.mod h1:9H9ctEv+K61bMy5teo/fG2azZswmXsPVfFTVdk8ZGk0=
 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 d14b623..e313046 100644
--- a/pkg/agent/controller/agent.go
+++ b/pkg/agent/controller/agent.go
@@ -158,7 +158,11 @@ func NewWithDetail(spec *AgentSpecification, syncerConf *broker.SyncerConfig, re
 		return nil, err
 	}
 
-	agentController.serviceImportController = newServiceImportController(spec, kubeClientSet, lighthouseClient)
+	agentController.serviceImportController, err = newServiceImportController(spec, agentController.serviceSyncer, restMapper,
+		localClient, scheme)
+	if err != nil {
+		return nil, err
+	}
 
 	return agentController, nil
 }
@@ -362,17 +366,19 @@ func (a *Controller) endpointToRemoteServiceImport(obj runtime.Object, op syncer
 
 	svcExport := obj.(*lighthousev2a1.ServiceExport)
 
-	serviceImport, err := a.getServiceImport(svcExport)
-
+	serviceImport, found, err := a.serviceImportController.getServiceImport(
+		a.getObjectNameWithClusterId(svcExport.Name, svcExport.Namespace), a.namespace)
 	if err != nil {
-		if !apierrors.IsNotFound(err) {
-			klog.Errorf("Error retrieving ServiceImport for Endpoints (%s/%s): %v", ep.Namespace, ep.Name, err)
-		}
+		klog.Errorf("Error retrieving ServiceImport for Endpoints (%s/%s): %v", ep.Namespace, ep.Name, err)
 
 		// Requeue
 		return nil, true
 	}
 
+	if !found {
+		return nil, true
+	}
+
 	if serviceImport.Spec.Type == lighthousev2a1.Headless && a.globalnetEnabled {
 		return nil, false
 	}
@@ -486,17 +492,6 @@ func (a *Controller) newServiceImport(svcExport *lighthousev2a1.ServiceExport) *
 	}
 }
 
-func (a *Controller) getServiceImport(svcExport *lighthousev2a1.ServiceExport) (*lighthousev2a1.ServiceImport, error) {
-	siName := a.getObjectNameWithClusterId(svcExport.Name, svcExport.Namespace)
-	serviceImport, err := a.lighthouseClient.LighthouseV2alpha1().ServiceImports(a.namespace).Get(siName, metav1.GetOptions{})
-
-	if err != nil {
-		return nil, err
-	}
-
-	return serviceImport, nil
-}
-
 func (a *Controller) getIPsForService(service *corev1.Service, siType lighthousev2a1.ServiceImportType) ([]string, error) {
 	if siType == lighthousev2a1.ClusterSetIP {
 		mcsIp := getGlobalIpFromService(service)
diff --git a/pkg/agent/controller/endpoint.go b/pkg/agent/controller/endpoint.go
index d1dc6ab..9117b0a 100644
--- a/pkg/agent/controller/endpoint.go
+++ b/pkg/agent/controller/endpoint.go
@@ -2,197 +2,77 @@ package controller
 
 import (
 	"github.com/submariner-io/admiral/pkg/log"
+	"github.com/submariner-io/admiral/pkg/syncer"
+	"github.com/submariner-io/admiral/pkg/syncer/broker"
 	lhconstants "github.com/submariner-io/lighthouse/pkg/constants"
 	corev1 "k8s.io/api/core/v1"
 	discovery "k8s.io/api/discovery/v1beta1"
-	"k8s.io/apimachinery/pkg/api/equality"
-	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/watch"
-	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/tools/cache"
-	"k8s.io/client-go/util/retry"
-	"k8s.io/client-go/util/workqueue"
+	"k8s.io/client-go/dynamic"
 	"k8s.io/klog"
 	utilnet "k8s.io/utils/net"
 )
 
-func newEndpointController(kubeClientSet kubernetes.Interface, serviceImportuid types.UID, serviceImportName,
-	serviceName, serviceImportNameSpace, clusterId string) *EndpointController {
-	queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
-	endpointController := &EndpointController{
-		endPointqueue:                queue,
-		serviceImportUID:             serviceImportuid,
+func startEndpointController(localClient dynamic.Interface, restMapper meta.RESTMapper, scheme *runtime.Scheme,
+	serviceImportUID types.UID, serviceImportName, serviceImportNameSpace, serviceName, clusterID string) (*EndpointController, error) {
+	klog.V(log.DEBUG).Infof("Starting Endpoints controller for service %q", serviceName)
+
+	controller := &EndpointController{
+		clusterID:                    clusterID,
+		serviceImportUID:             serviceImportUID,
 		serviceImportName:            serviceImportName,
-		serviceName:                  serviceName,
 		serviceImportSourceNameSpace: serviceImportNameSpace,
-		kubeClientSet:                kubeClientSet,
-		clusterID:                    clusterId,
+		serviceName:                  serviceName,
 		stopCh:                       make(chan struct{}),
 	}
 
-	return endpointController
-}
-
-func (e *EndpointController) start(stopCh <-chan struct{}) {
-	nameSelector := fields.OneTermEqualSelector("metadata.name", e.serviceName)
-	e.store, e.endpointInformer = cache.NewInformer(
-		&cache.ListWatch{
-			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
-				options.FieldSelector = nameSelector.String()
-				return e.kubeClientSet.CoreV1().Endpoints(e.serviceImportSourceNameSpace).List(options)
-			},
-			WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
-				options.FieldSelector = nameSelector.String()
-				return e.kubeClientSet.CoreV1().Endpoints(e.serviceImportSourceNameSpace).Watch(options)
-			},
-		},
-		&corev1.Endpoints{},
-		0,
-		cache.ResourceEventHandlerFuncs{
-			AddFunc: func(obj interface{}) {
-				key, err := cache.MetaNamespaceKeyFunc(obj)
-				klog.V(log.DEBUG).Infof("Endpoint %q added", key)
-				if err == nil {
-					e.endPointqueue.Add(key)
-				}
-			},
-			UpdateFunc: func(obj interface{}, new interface{}) {
-				key, err := cache.MetaNamespaceKeyFunc(new)
-				klog.V(log.DEBUG).Infof("Endpoint %q updated", key)
-				if err == nil {
-					e.endPointqueue.Add(key)
-				}
-			},
-			DeleteFunc: func(obj interface{}) {
-				key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
-				klog.V(log.DEBUG).Infof("Endpoint %q deleted", key)
-				if err == nil {
-					var endPoints *corev1.Endpoints
-					var ok bool
-					if endPoints, ok = obj.(*corev1.Endpoints); !ok {
-						tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
-						if !ok {
-							klog.Errorf("Failed to get deleted endPoints object for key %s, serviceImport %v", key, endPoints)
-							return
-						}
-
-						endPoints, ok = tombstone.Obj.(*corev1.Endpoints)
-
-						if !ok {
-							klog.Errorf("Failed to convert deleted tombstone object %v  to endPoints", tombstone.Obj)
-							return
-						}
-					}
-					e.endpointDeletedMap.Store(endPoints, key)
-					e.endPointqueue.Add(key)
-				}
-			},
-		},
-	)
+	nameSelector := fields.OneTermEqualSelector("metadata.name", serviceName)
+
+	epsSyncer, err := syncer.NewResourceSyncer(&syncer.ResourceSyncerConfig{
+		Name:                "Endpoints -> EndpointSlice",
+		SourceClient:        localClient,
+		SourceNamespace:     serviceImportNameSpace,
+		SourceFieldSelector: nameSelector.String(),
+		Direction:           syncer.LocalToRemote,
+		RestMapper:          restMapper,
+		Federator:           broker.NewFederator(localClient, restMapper, serviceImportNameSpace, "", "ownerReferences"),
+		ResourceType:        &corev1.Endpoints{},
+		Transform:           controller.endpointsToEndpointSlice,
+		Scheme:              scheme,
+	})
+	if err != nil {
+		return nil, err
+	}
 
-	klog.V(log.DEBUG).Infof("Starting Endpoint watcher for %q", e.serviceImportName)
+	if err := epsSyncer.Start(controller.stopCh); err != nil {
+		return nil, err
+	}
 
-	go e.endpointInformer.Run(e.stopCh)
-	go e.runEndpointWorker(e.endpointInformer, e.endPointqueue)
+	return controller, nil
 }
 
-func (e *EndpointController) Stop() {
-	e.endPointqueue.ShutDown()
+func (e *EndpointController) stop() {
 	close(e.stopCh)
 }
 
-func (e *EndpointController) runEndpointWorker(informer cache.Controller, queue workqueue.RateLimitingInterface) {
-	for {
-		keyObj, shutdown := queue.Get()
-		if shutdown {
-			klog.V(log.DEBUG).Infof("Endpoints watcher for %q stopped", e.serviceImportName)
-			return
-		}
-
-		key := keyObj.(string)
-
-		func() {
-			defer queue.Done(key)
-			obj, exists, err := e.store.GetByKey(key)
-
-			if err != nil {
-				klog.Errorf("Error retrieving the object with key  %s from the cache: %v", key, err)
-				// requeue the item to work on later
-				queue.AddRateLimited(key)
-
-				return
-			}
-
-			if exists {
-				err = e.endPointCreatedOrUpdated(obj, key)
-			} else {
-				err = e.endPointDeleted(key)
-			}
-
-			if err != nil {
-				queue.AddRateLimited(key)
-			} else {
-				queue.Forget(key)
-			}
-		}()
-	}
-}
-
-func (e *EndpointController) endPointCreatedOrUpdated(obj interface{}, key string) error {
+func (e *EndpointController) endpointsToEndpointSlice(obj runtime.Object, op syncer.Operation) (runtime.Object, bool) {
 	endPoints := obj.(*corev1.Endpoints)
-
-	newEndPointSlice := e.endpointSliceFromEndpoints(endPoints)
-	retryErr := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		currentEndpointSice, err := e.kubeClientSet.DiscoveryV1beta1().EndpointSlices(endPoints.Namespace).
-			Get(newEndPointSlice.Name, metav1.GetOptions{})
-
-		if err != nil && !errors.IsNotFound(err) {
-			return err
-		}
-		if errors.IsNotFound(err) {
-			_, err = e.kubeClientSet.DiscoveryV1beta1().EndpointSlices(endPoints.Namespace).Create(newEndPointSlice)
-		} else {
-			if endpointSliceEquivalent(currentEndpointSice, newEndPointSlice) {
-				return nil
-			}
-			_, err = e.kubeClientSet.DiscoveryV1beta1().EndpointSlices(endPoints.Namespace).Update(newEndPointSlice)
-		}
-		if err != nil {
-			return err
-		}
-		return nil
-	})
-	if retryErr != nil {
-		klog.Errorf("Creating EndpointSlice %s from NameSpace %s failed after retry %v", newEndPointSlice.Name, endPoints.Namespace,
-			retryErr)
-	}
-
-	return retryErr
-}
-
-func (e *EndpointController) endPointDeleted(key string) error {
-	obj, found := e.endpointDeletedMap.Load(key)
-	if !found {
-		klog.Errorf("deleting endpointSlice for key %s failed since object not found", key)
-		return nil
-	}
-
-	endPoints := obj.(corev1.Endpoints)
 	endpointSliceName := endPoints.Name + "-" + e.clusterID
-	err := e.kubeClientSet.DiscoveryV1beta1().EndpointSlices(endPoints.Namespace).
-		Delete(endpointSliceName, &metav1.DeleteOptions{})
-	if err != nil && !errors.IsNotFound(err) {
-		klog.Errorf("Deleting EndpointSlice for endPoints %v from NameSpace %s failed due to %v", endPoints, endPoints.Namespace, err)
-		return err
-	}
 
-	e.endpointDeletedMap.Delete(key)
+	if op == syncer.Delete {
+		return &discovery.EndpointSlice{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      endpointSliceName,
+				Namespace: endPoints.Namespace,
+			},
+		}, false
+	}
 
-	return nil
+	return e.endpointSliceFromEndpoints(endPoints), false
 }
 
 func (e *EndpointController) endpointSliceFromEndpoints(endpoints *corev1.Endpoints) *discovery.EndpointSlice {
@@ -284,9 +164,3 @@ func allAddressesIPv6(addresses []corev1.EndpointAddress) bool {
 
 	return true
 }
-
-func endpointSliceEquivalent(obj1, obj2 *discovery.EndpointSlice) bool {
-	return equality.Semantic.DeepEqual(obj1.Endpoints, obj2.Endpoints) &&
-		equality.Semantic.DeepEqual(obj1.Ports, obj2.Ports) &&
-		equality.Semantic.DeepEqual(obj1.AddressType, obj2.AddressType)
-}
diff --git a/pkg/agent/controller/serviceimport.go b/pkg/agent/controller/serviceimport.go
index 0519fae..de77970 100644
--- a/pkg/agent/controller/serviceimport.go
+++ b/pkg/agent/controller/serviceimport.go
@@ -1,200 +1,141 @@
 package controller
 
 import (
+	"github.com/submariner-io/admiral/pkg/federate"
 	"github.com/submariner-io/admiral/pkg/log"
+	"github.com/submariner-io/admiral/pkg/syncer"
 	lighthousev2a1 "github.com/submariner-io/lighthouse/pkg/apis/lighthouse.submariner.io/v2alpha1"
-	lighthouseClientset "github.com/submariner-io/lighthouse/pkg/client/clientset/versioned"
-	"github.com/submariner-io/lighthouse/pkg/client/informers/externalversions"
 	lhconstants "github.com/submariner-io/lighthouse/pkg/constants"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/tools/cache"
-	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog"
 )
 
-func newServiceImportController(spec *AgentSpecification, kubeClientSet kubernetes.Interface,
-	lighthouseClient lighthouseClientset.Interface) *ServiceImportController {
-	return &ServiceImportController{
-		queue:            workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
-		kubeClientSet:    kubeClientSet,
-		lighthouseClient: lighthouseClient,
-		clusterID:        spec.ClusterID,
-		namespace:        spec.Namespace,
+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,
 	}
-}
 
-func (c *ServiceImportController) start(stopCh <-chan struct{}) error {
-	informerFactory := externalversions.NewSharedInformerFactoryWithOptions(c.lighthouseClient, 0,
-		externalversions.WithNamespace(c.namespace))
-	c.serviceInformer = informerFactory.Lighthouse().V2alpha1().ServiceImports().Informer()
-
-	c.serviceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc: func(obj interface{}) {
-			key, err := cache.MetaNamespaceKeyFunc(obj)
-			klog.V(log.DEBUG).Infof("ServiceImport %q added", key)
-			if err == nil {
-				c.queue.Add(key)
-			}
-		},
-		UpdateFunc: func(obj interface{}, new interface{}) {
-			key, err := cache.MetaNamespaceKeyFunc(new)
-			klog.V(log.DEBUG).Infof("ServiceImport %q updated", key)
-			if err == nil {
-				c.queue.Add(key)
-			}
-		},
-		DeleteFunc: func(obj interface{}) {
-			key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
-			klog.V(log.DEBUG).Infof("ServiceImport %q deleted", key)
-			if err == nil {
-				var si *lighthousev2a1.ServiceImport
-				var ok bool
-				if si, ok = obj.(*lighthousev2a1.ServiceImport); !ok {
-					tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
-					if !ok {
-						klog.Errorf("Failed to get deleted serviceimport object for key %s, serviceImport %v", key, si)
-						return
-					}
-
-					si, ok = tombstone.Obj.(*lighthousev2a1.ServiceImport)
-
-					if !ok {
-						klog.Errorf("Failed to convert deleted tombstone object %v  to serviceimport", tombstone.Obj)
-						return
-					}
-				}
-				if si.Spec.Type != lighthousev2a1.Headless {
-					return
-				}
-				c.serviceImportDeletedMap.Store(key, si)
-				c.queue.AddRateLimited(key)
-			}
-		},
+	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:    &lighthousev2a1.ServiceImport{},
+		Transform:       controller.serviceImportToEndpointController,
+		Scheme:          scheme,
 	})
+	if err != nil {
+		return nil, err
+	}
 
-	go c.serviceInformer.Run(stopCh)
-	go c.runServiceImportWorker()
+	return controller, nil
+}
 
-	go func(stopCh <-chan struct{}) {
+func (c *ServiceImportController) start(stopCh <-chan struct{}) error {
+	go func() {
 		<-stopCh
-		c.queue.ShutDown()
 
 		c.endpointControllers.Range(func(key, value interface{}) bool {
-			value.(*EndpointController).Stop()
+			value.(*EndpointController).stop()
 			return true
 		})
 
 		klog.Infof("ServiceImport Controller stopped")
-	}(stopCh)
-
-	return nil
-}
+	}()
 
-func (c *ServiceImportController) runServiceImportWorker() {
-	for {
-		keyObj, shutdown := c.queue.Get()
-		if shutdown {
-			klog.Infof("Lighthouse watcher for ServiceImports stopped")
-			return
-		}
-
-		key := keyObj.(string)
-
-		func() {
-			defer c.queue.Done(key)
-			obj, exists, err := c.serviceInformer.GetIndexer().GetByKey(key)
-
-			if err != nil {
-				klog.Errorf("Error retrieving the object with store is  %v from the cache: %v", c.serviceInformer.GetIndexer().ListKeys(), err)
-				// requeue the item to work on later
-				c.queue.AddRateLimited(key)
-
-				return
-			}
-
-			if exists {
-				err = c.serviceImportCreatedOrUpdated(obj, key)
-			} else {
-				err = c.serviceImportDeleted(key)
-			}
-
-			if err != nil {
-				c.queue.AddRateLimited(key)
-			} else {
-				c.queue.Forget(key)
-			}
-		}()
+	if err := c.serviceImportSyncer.Start(stopCh); err != nil {
+		return err
 	}
+
+	return nil
 }
 
-func (c *ServiceImportController) serviceImportCreatedOrUpdated(obj interface{}, key string) error {
+func (c *ServiceImportController) serviceImportCreatedOrUpdated(serviceImport *lighthousev2a1.ServiceImport, key string) bool {
 	if _, found := c.endpointControllers.Load(key); found {
-		klog.V(log.DEBUG).Infof("The endpoint controller is already running fof %q", key)
-		return nil
+		klog.V(log.DEBUG).Infof("The endpoint controller is already running for %q", key)
+		return false
 	}
 
-	serviceImportCreated := obj.(*lighthousev2a1.ServiceImport)
-	if serviceImportCreated.Spec.Type != lighthousev2a1.Headless ||
-		serviceImportCreated.GetLabels()[lhconstants.LabelSourceCluster] != c.clusterID {
-		return nil
+	if serviceImport.Spec.Type != lighthousev2a1.Headless ||
+		serviceImport.GetLabels()[lhconstants.LabelSourceCluster] != c.clusterID {
+		return false
 	}
 
-	annotations := serviceImportCreated.ObjectMeta.Annotations
+	annotations := serviceImport.ObjectMeta.Annotations
 	serviceNameSpace := annotations[lhconstants.OriginNamespace]
 	serviceName := annotations[lhconstants.OriginName]
-	var service *corev1.Service
 
-	service, err := c.kubeClientSet.CoreV1().Services(serviceNameSpace).Get(serviceName, metav1.GetOptions{})
+	obj, found, err := c.serviceSyncer.GetResource(serviceName, serviceNameSpace)
 	if err != nil {
-		if errors.IsNotFound(err) {
-			return nil
-		}
-
 		klog.Errorf("Error retrieving the service  %q from the namespace %q : %v", serviceName, serviceNameSpace, err)
 
-		return err
+		return true
+	}
+
+	if !found {
+		return false
 	}
 
+	service := obj.(*corev1.Service)
 	if service.Spec.Selector == nil {
 		klog.Errorf("The service %s/%s without a Selector is not supported", serviceNameSpace, serviceName)
-		return nil
+		return false
 	}
 
-	endpointController := newEndpointController(c.kubeClientSet, serviceImportCreated.ObjectMeta.UID,
-		serviceImportCreated.ObjectMeta.Name, serviceName, serviceNameSpace, c.clusterID)
+	endpointController, err := startEndpointController(c.localClient, c.restMapper, c.scheme,
+		serviceImport.ObjectMeta.UID, serviceImport.ObjectMeta.Name, serviceNameSpace, serviceName, c.clusterID)
+	if err != nil {
+		klog.Errorf(err.Error())
+		return true
+	}
 
-	endpointController.start(endpointController.stopCh)
 	c.endpointControllers.Store(key, endpointController)
 
-	return nil
+	return false
 }
 
-func (c *ServiceImportController) serviceImportDeleted(key string) error {
-	obj, found := c.serviceImportDeletedMap.Load(key)
-	if !found {
-		klog.Warningf("No endpoint controller found  for %q", key)
-		c.serviceImportDeletedMap.Delete(key)
-
-		return nil
-	}
-
-	si := obj.(*lighthousev2a1.ServiceImport)
-
-	if si.GetLabels()[lhconstants.LabelSourceCluster] != c.clusterID {
-		c.serviceImportDeletedMap.Delete(key)
-		return nil
+func (c *ServiceImportController) serviceImportDeleted(serviceImport *lighthousev2a1.ServiceImport, key string) bool {
+	if serviceImport.GetLabels()[lhconstants.LabelSourceCluster] != c.clusterID {
+		return false
 	}
 
 	if obj, found := c.endpointControllers.Load(key); found {
 		endpointController := obj.(*EndpointController)
-		endpointController.Stop()
+		endpointController.stop()
 		c.endpointControllers.Delete(key)
 	}
 
-	c.serviceImportDeletedMap.Delete(key)
+	return false
+}
 
-	return nil
+func (c *ServiceImportController) serviceImportToEndpointController(obj runtime.Object, op syncer.Operation) (runtime.Object, bool) {
+	serviceImport := obj.(*lighthousev2a1.ServiceImport)
+	key, _ := cache.MetaNamespaceKeyFunc(serviceImport)
+	if op == syncer.Create || op == syncer.Update {
+		return nil, c.serviceImportCreatedOrUpdated(serviceImport, key)
+	}
+
+	return nil, c.serviceImportDeleted(serviceImport, key)
+}
+
+func (c *ServiceImportController) getServiceImport(name, namespace string) (*lighthousev2a1.ServiceImport, bool, error) {
+	obj, found, err := c.serviceImportSyncer.GetResource(name, namespace)
+	if obj != nil {
+		return obj.(*lighthousev2a1.ServiceImport), found, err
+	}
+
+	return nil, found, err
 }
diff --git a/pkg/agent/controller/types.go b/pkg/agent/controller/types.go
index 8a2f04a..b5fdbf5 100644
--- a/pkg/agent/controller/types.go
+++ b/pkg/agent/controller/types.go
@@ -6,10 +6,11 @@ import (
 	"github.com/submariner-io/admiral/pkg/syncer"
 	"github.com/submariner-io/admiral/pkg/syncer/broker"
 	lighthouseClientset "github.com/submariner-io/lighthouse/pkg/client/clientset/versioned"
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/tools/cache"
-	"k8s.io/client-go/util/workqueue"
 )
 
 type Controller struct {
@@ -35,29 +36,23 @@ type AgentSpecification struct {
 // 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 ServiceImportController struct {
-	kubeClientSet           kubernetes.Interface
-	lighthouseClient        lighthouseClientset.Interface
-	serviceInformer         cache.SharedIndexInformer
-	queue                   workqueue.RateLimitingInterface
-	endpointControllers     sync.Map
-	serviceImportDeletedMap sync.Map
-	clusterID               string
-	namespace               string
+	serviceSyncer       syncer.Interface
+	localClient         dynamic.Interface
+	restMapper          meta.RESTMapper
+	serviceImportSyncer syncer.Interface
+	endpointControllers sync.Map
+	clusterID           string
+	scheme              *runtime.Scheme
 }
 
 // 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 {
-	kubeClientSet                kubernetes.Interface
-	endpointInformer             cache.Controller
-	store                        cache.Store
-	endPointqueue                workqueue.RateLimitingInterface
 	serviceImportUID             types.UID
 	clusterID                    string
 	serviceImportName            string
 	serviceName                  string
 	serviceImportSourceNameSpace string
-	endpointDeletedMap           sync.Map
 	stopCh                       chan struct{}
 }
diff --git a/scripts/resources/coredns-cm.yaml b/scripts/resources/coredns-cm.yaml
index 4927260..291d53e 100644
--- a/scripts/resources/coredns-cm.yaml
+++ b/scripts/resources/coredns-cm.yaml
@@ -1,6 +1,3 @@
     clusterset.local:53 {
         forward . dnsip
     }
-    supercluster.local:53 {
-        forward . dnsip
-    }


Test patch:
diff --git a/pkg/agent/controller/agent_test.go b/pkg/agent/controller/agent_test.go
index 1a373c2..79d916b 100644
--- a/pkg/agent/controller/agent_test.go
+++ b/pkg/agent/controller/agent_test.go
@@ -26,12 +26,10 @@ import (
 	"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"
 	"k8s.io/client-go/kubernetes"
 	fakeKubeClient "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/kubernetes/scheme"
-	"k8s.io/client-go/tools/cache"
 )
 
 const clusterID1 = "east"
@@ -473,38 +471,6 @@ func newTestDiver() *testDriver {
 	t.cluster1.init(t.restMapper)
 	t.cluster2.init(t.restMapper)
 
-	_, endpointSliceInformer := cache.NewInformer(
-		&cache.ListWatch{
-			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
-				return t.cluster1.localKubeClient.DiscoveryV1beta1().EndpointSlices(metav1.NamespaceAll).List(options)
-			},
-			WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
-				return t.cluster1.localKubeClient.DiscoveryV1beta1().EndpointSlices(metav1.NamespaceAll).Watch(options)
-			},
-		},
-		&discovery.EndpointSlice{}, 0,
-		cache.ResourceEventHandlerFuncs{
-			AddFunc: func(obj interface{}) {
-				defer GinkgoRecover()
-				_, err := t.cluster1.localEndpointSliceClient.Create(test.ToUnstructured(obj.(runtime.Object)),
-					metav1.CreateOptions{})
-				Expect(err).To(Succeed())
-			},
-			UpdateFunc: func(obj interface{}, new interface{}) {
-				defer GinkgoRecover()
-				_, err := t.cluster1.localEndpointSliceClient.Update(test.ToUnstructured(new.(runtime.Object)),
-					metav1.UpdateOptions{})
-				Expect(err).To(Succeed())
-			},
-			DeleteFunc: func(obj interface{}) {
-				defer GinkgoRecover()
-				Expect(t.cluster1.localEndpointSliceClient.Delete(obj.(*discovery.EndpointSlice).Name, nil)).To(Succeed())
-			},
-		},
-	)
-
-	go endpointSliceInformer.Run(t.stopCh)
-
 	return t
 }
 
@@ -583,10 +549,7 @@ func awaitServiceImport(client dynamic.ResourceInterface, service *corev1.Servic
 }
 
 func (c *cluster) awaitServiceImport(service *corev1.Service, sType lighthousev2a1.ServiceImportType, serviceIPs ...string) {
-	serviceImport := awaitServiceImport(c.localServiceImportClient, service, sType, serviceIPs...)
-
-	_, err := c.lighthouseClient.LighthouseV2alpha1().ServiceImports(serviceImport.Namespace).Create(serviceImport)
-	Expect(err).To(Succeed())
+	awaitServiceImport(c.localServiceImportClient, service, sType, serviceIPs...)
 }
 
 func awaitUpdatedServiceImport(client dynamic.ResourceInterface, service *corev1.Service,
@@ -619,10 +582,7 @@ func awaitUpdatedServiceImport(client dynamic.ResourceInterface, service *corev1
 }
 
 func (c *cluster) awaitUpdatedServiceImport(service *corev1.Service, serviceIPs ...string) {
-	serviceImport := awaitUpdatedServiceImport(c.localServiceImportClient, service, serviceIPs...)
-
-	_, err := c.lighthouseClient.LighthouseV2alpha1().ServiceImports(serviceImport.Namespace).Update(serviceImport)
-	Expect(err).To(Succeed())
+	awaitUpdatedServiceImport(c.localServiceImportClient, service, serviceIPs...)
 }
 
 func awaitEndpointSlice(endpointSliceClient, serviceImportClient dynamic.ResourceInterface,
@@ -910,12 +870,8 @@ func (t *testDriver) awaitServiceUnexported() {
 func (t *testDriver) awaitHeadlessServiceUnexported() {
 	t.awaitServiceUnexported()
 
-	Expect(t.cluster1.lighthouseClient.LighthouseV2alpha1().ServiceImports(t.service.Namespace).Delete(
-		t.service.Name+"-"+t.service.Namespace+"-"+clusterID1, nil)).To(Succeed())
-
 	// In a real k8s env, the EndpointSlice would be deleted via k8s GC as it is owned by the ServiceImport.
-	Expect(t.cluster1.localKubeClient.DiscoveryV1beta1().EndpointSlices(t.endpoints.Namespace).Delete(
-		t.endpoints.Name+"-"+clusterID1, nil)).To(Succeed())
+	Expect(t.cluster1.localEndpointSliceClient.Delete(t.endpoints.Name+"-"+clusterID1, nil)).To(Succeed())
 
 	t.awaitNoEndpointSlice(t.cluster1.localEndpointSliceClient)
 	t.awaitNoEndpointSlice(t.brokerEndpointSliceClient)
diff --git a/test/e2e/discovery/service_discovery.go b/test/e2e/discovery/service_discovery.go
index e2c0727..be6bb9b 100644
--- a/test/e2e/discovery/service_discovery.go
+++ b/test/e2e/discovery/service_discovery.go
@@ -14,11 +14,10 @@ import (
 const (
 	submarinerIpamGlobalIp = "submariner.io/globalIp"
 	clustersetDomain       = "clusterset.local"
-	superclusterDomain     = "supercluster.local"
 )
 
 // Both domains need to be checked, until the operator is updated to use clusterset
-var checkedDomains = []string{clustersetDomain, superclusterDomain}
+var checkedDomains = []string{clustersetDomain}
 
 var _ = Describe("[discovery] Test Service Discovery Across Clusters", func() {
 	f := lhframework.NewFramework("discovery")