Repository: submariner-io/lighthouse
Pull request #315 (2020-09-30 21:38:15)

Convert EndpointController to use ResourceSyncer


Patch:
diff --git a/go.mod b/go.mod
index c4b4abe..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.3
+	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 3c3575a..a6b6446 100644
--- a/go.sum
+++ b/go.sum
@@ -538,8 +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.3 h1:96wOqGEDKv0Oq6QXozE6C7/HJ87kEIdHZp4AutykC5s=
-github.com/submariner-io/admiral v0.6.3/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/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 1d581e3..7ca02d4 100644
--- a/pkg/agent/controller/serviceimport.go
+++ b/pkg/agent/controller/serviceimport.go
@@ -21,7 +21,10 @@ func newServiceImportController(spec *AgentSpecification, kubeClientSet kubernet
 	localClient dynamic.Interface, scheme *runtime.Scheme) (*ServiceImportController, error) {
 	controller := &ServiceImportController{
 		kubeClientSet: kubeClientSet,
+		localClient:   localClient,
+		restMapper:    restMapper,
 		clusterID:     spec.ClusterID,
+		scheme:        scheme,
 	}
 
 	var err error
@@ -49,7 +52,7 @@ func (c *ServiceImportController) start(stopCh <-chan struct{}) error {
 		<-stopCh
 
 		c.endpointControllers.Range(func(key, value interface{}) bool {
-			value.(*EndpointController).Stop()
+			value.(*EndpointController).stop()
 			return true
 		})
 
@@ -95,10 +98,13 @@ func (c *ServiceImportController) serviceImportCreatedOrUpdated(serviceImport *l
 		return false
 	}
 
-	endpointController := newEndpointController(c.kubeClientSet, serviceImport.ObjectMeta.UID,
-		serviceImport.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 false
@@ -111,7 +117,7 @@ func (c *ServiceImportController) serviceImportDeleted(serviceImport *lighthouse
 
 	if obj, found := c.endpointControllers.Load(key); found {
 		endpointController := obj.(*EndpointController)
-		endpointController.Stop()
+		endpointController.stop()
 		c.endpointControllers.Delete(key)
 	}
 
diff --git a/pkg/agent/controller/types.go b/pkg/agent/controller/types.go
index 7a5c7ff..eecb4c6 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 {
@@ -36,24 +37,22 @@ type AgentSpecification struct {
 // to listen only for the endpoints event related to ServiceImport created
 type ServiceImportController struct {
 	kubeClientSet       kubernetes.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..b9f2e89 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
 }
 
@@ -914,8 +880,7 @@ func (t *testDriver) awaitHeadlessServiceUnexported() {
 		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")