Repository: submariner-io/lighthouse
Pull request #303 (2020-09-21 19:05:35)

Use Syncer cache
Get `ServiceExport` and `Service` resources from the Syncer informer caches to avoid accessing from the API server.


Patch:
diff --git a/pkg/agent/controller/agent.go b/pkg/agent/controller/agent.go
index 4086671..d14b623 100644
--- a/pkg/agent/controller/agent.go
+++ b/pkg/agent/controller/agent.go
@@ -209,21 +209,26 @@ func (a *Controller) serviceExportToRemoteServiceImport(obj runtime.Object, op s
 		return serviceImport, false
 	}
 
-	svc, err := a.kubeClientSet.CoreV1().Services(svcExport.Namespace).Get(svcExport.Name, metav1.GetOptions{})
-	if apierrors.IsNotFound(err) {
-		klog.V(log.DEBUG).Infof("Service to be exported (%s/%s) doesn't exist", svcExport.Namespace, svcExport.Name)
-		a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, lighthousev2a1.ServiceExportInitialized,
-			corev1.ConditionFalse, serviceUnavailable, "Service to be exported doesn't exist")
-
-		return nil, true
-	} else if err != nil {
+	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, lighthousev2a1.ServiceExportInitialized,
 			corev1.ConditionUnknown, "ServiceRetrievalFailed", fmt.Sprintf("Error retrieving the Service: %v", err))
 		klog.Errorf("Error retrieving Service (%s/%s): %v", svcExport.Namespace, svcExport.Name, err)
+
+		return nil, true
+	}
+
+	if !found {
+		klog.V(log.DEBUG).Infof("Service to be exported (%s/%s) doesn't exist", svcExport.Namespace, svcExport.Name)
+		a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, lighthousev2a1.ServiceExportInitialized,
+			corev1.ConditionFalse, serviceUnavailable, "Service to be exported doesn't exist")
+
 		return nil, true
 	}
 
+	svc := obj.(*corev1.Service)
+
 	svcType, ok := getServiceImportType(svc)
 
 	if !ok {
@@ -319,16 +324,20 @@ func (a *Controller) serviceToRemoteServiceImport(obj runtime.Object, op syncer.
 	}
 
 	svc := obj.(*corev1.Service)
-	svcExport, err := a.lighthouseClient.LighthouseV2alpha1().ServiceExports(svc.Namespace).Get(svc.Name, metav1.GetOptions{})
-	if apierrors.IsNotFound(err) {
-		// Service Export not created yet
-		return nil, false
-	} else if err != nil {
+	obj, found, err := a.serviceExportSyncer.GetLocalResource(svc.Name, svc.Namespace, &lighthousev2a1.ServiceExport{})
+	if err != nil {
 		// some other error. Log and requeue
 		klog.Errorf("Error retrieving ServiceExport for Service (%s/%s): %v", svc.Namespace, svc.Name, err)
 		return nil, true
 	}
 
+	if !found {
+		// Service Export not created yet
+		return nil, false
+	}
+
+	svcExport := obj.(*lighthousev2a1.ServiceExport)
+
 	serviceImport := a.newServiceImport(svcExport)
 
 	// Update the status and requeue
@@ -340,16 +349,19 @@ func (a *Controller) serviceToRemoteServiceImport(obj runtime.Object, op syncer.
 
 func (a *Controller) endpointToRemoteServiceImport(obj runtime.Object, op syncer.Operation) (runtime.Object, bool) {
 	ep := obj.(*corev1.Endpoints)
-	svcExport, err := a.lighthouseClient.LighthouseV2alpha1().ServiceExports(ep.Namespace).Get(ep.Name, metav1.GetOptions{})
-	if apierrors.IsNotFound(err) {
-		// Service Export not created yet
-		return nil, false
-	} else if err != nil {
-		// some other error. Log and requeue
+	obj, found, err := a.serviceExportSyncer.GetLocalResource(ep.Name, ep.Namespace, &lighthousev2a1.ServiceExport{})
+	if err != nil {
 		klog.Errorf("Error retrieving ServiceExport for Endpoints (%s/%s): %v", ep.Namespace, ep.Name, err)
 		return nil, true
 	}
 
+	if !found {
+		// Service Export not created yet
+		return nil, false
+	}
+
+	svcExport := obj.(*lighthousev2a1.ServiceExport)
+
 	serviceImport, err := a.getServiceImport(svcExport)
 
 	if err != nil {


Test patch:
diff --git a/pkg/agent/controller/agent_test.go b/pkg/agent/controller/agent_test.go
index 45508c4..66802f0 100644
--- a/pkg/agent/controller/agent_test.go
+++ b/pkg/agent/controller/agent_test.go
@@ -192,8 +192,7 @@ var _ = Describe("Globalnet enabled", func() {
 				corev1.ConditionFalse, "ServiceGlobalIPUnavailable"))
 
 			t.service.SetAnnotations(map[string]string{"submariner.io/globalIp": globalIP})
-			_, err := t.cluster1.localKubeClient.CoreV1().Services(t.service.Namespace).Update(t.service)
-			Expect(err).To(Succeed())
+			test.UpdateResource(t.dynamicServiceClient(), t.service)
 
 			t.awaitServiceExported(globalIP, 1)
 		})
@@ -284,19 +283,6 @@ var _ = Describe("Service export failures", func() {
 		t.createServiceExport()
 	})
 
-	When("Service retrieval initially fails", func() {
-		BeforeEach(func() {
-			t.cluster1.servicesReactor.SetFailOnGet(errors.New("fake Get error"))
-		})
-
-		It("should update the ServiceExport status and eventually sync a ServiceImport", func() {
-			t.awaitServiceExportStatus(0, newServiceExportCondition(lighthousev2a1.ServiceExportInitialized,
-				corev1.ConditionUnknown, "ServiceRetrievalFailed"))
-			t.cluster1.servicesReactor.SetResetOnFailure(true)
-			t.awaitServiceExported(t.service.Spec.ClusterIP, 1)
-		})
-	})
-
 	When("Endpoints retrieval initially fails", func() {
 		BeforeEach(func() {
 			t.service.Spec.ClusterIP = corev1.ClusterIPNone
@@ -311,34 +297,6 @@ var _ = Describe("Service export failures", func() {
 		})
 	})
 
-	When("an exported Service is deleted and ServiceExport retrieval initially fails", func() {
-		It("should eventually delete the ServiceImport", func() {
-			t.awaitServiceExported(t.service.Spec.ClusterIP, 0)
-
-			t.cluster1.serviceExportReactor.SetFailOnGet(errors.New("fake Get error"))
-			t.cluster1.serviceExportReactor.SetResetOnFailure(true)
-			t.deleteService()
-			t.awaitNoServiceImport(t.cluster1.localServiceImportClient)
-		})
-	})
-
-	When("Endpoints is updated and ServiceExport retrieval initially fails", func() {
-		BeforeEach(func() {
-			t.service.Spec.ClusterIP = corev1.ClusterIPNone
-		})
-
-		It("should eventually update the ServiceImport", func() {
-			t.awaitHeadlessServiceImport(t.endpointIPs()...)
-
-			t.cluster1.serviceExportReactor.SetFailOnGet(errors.New("fake Get error"))
-			t.cluster1.serviceExportReactor.SetResetOnFailure(true)
-			t.endpoints.Subsets[0].Addresses = append(t.endpoints.Subsets[0].Addresses, corev1.EndpointAddress{IP: "192.168.5.3"})
-			t.updateEndpoints()
-
-			t.awaitUpdatedServiceImport(t.endpointIPs()...)
-		})
-	})
-
 	When("Endpoints is updated and ServiceImport retrieval initially fails", func() {
 		BeforeEach(func() {
 			t.service.Spec.ClusterIP = corev1.ClusterIPNone