Repository: submariner-io/lighthouse
Pull request #220 (2020-07-23 22:02:54)

Update status to Exported when successfully synced
Utilize the `OnSuccessfulSync` callback to properly update the exported status. Also prior to sync update the status to indicate it's awaiting sync.


submariner-bot: 🤖 Created branch: z_pr220/tpantelis/successful_serviceexport_status

Patch:
diff --git a/go.mod b/go.mod
index d15c4b5..ae56312 100644
--- a/go.mod
+++ b/go.mod
@@ -10,7 +10,7 @@ require (
 	github.com/onsi/ginkgo v1.14.0
 	github.com/onsi/gomega v1.10.1
 	github.com/pkg/errors v0.9.1
-	github.com/submariner-io/admiral v0.6.0-pre0
+	github.com/submariner-io/admiral v0.6.0-pre1
 	github.com/submariner-io/shipyard v0.5.0
 	k8s.io/api v0.0.0-20190918195907-bd6ac527cfd2
 	k8s.io/apimachinery v0.0.0-20190817020851-f2f3a405f61d
diff --git a/go.sum b/go.sum
index 009c28f..8c5d96d 100644
--- a/go.sum
+++ b/go.sum
@@ -345,6 +345,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/submariner-io/admiral v0.6.0-pre0 h1:/f/cs1qZPNHRQMl4yQxYMEZzOlKeZguGM9HSP84Rt/w=
 github.com/submariner-io/admiral v0.6.0-pre0/go.mod h1:cBv1Zvfai117idhYv5WPxTZEsgWQfeSqwPpxo0EMZXg=
+github.com/submariner-io/admiral v0.6.0-pre1 h1:v3KBPSQlfFovfkpyXnrwmWeAcSdbJ5zGBgslsLNSAMY=
+github.com/submariner-io/admiral v0.6.0-pre1/go.mod h1:cBv1Zvfai117idhYv5WPxTZEsgWQfeSqwPpxo0EMZXg=
 github.com/submariner-io/shipyard v0.5.0 h1:o4kTCubLFLeh+PaBDac1J7zNY3JKWwj7QrMLNb4g/p8=
 github.com/submariner-io/shipyard v0.5.0/go.mod h1:2OrKoDmJSo1jVurxiVH9FqGFmDwshIU7Vgu9pFLBsYo=
 github.com/timewasted/linode v0.0.0-20160829202747-37e84520dcf7/go.mod h1:imsgLplxEC/etjIhdr3dNzV3JeT27LbVu5pYWm0JCBY=
diff --git a/pkg/agent/controller/agent.go b/pkg/agent/controller/agent.go
index ac40132..3712c9f 100644
--- a/pkg/agent/controller/agent.go
+++ b/pkg/agent/controller/agent.go
@@ -25,6 +25,8 @@ import (
 const (
 	submarinerIpamGlobalIp = "submariner.io/globalIp"
 	serviceUnavailable     = "ServiceUnavailable"
+	originName             = "origin-name"
+	originNamespace        = "origin-namespace"
 )
 
 func New(spec *AgentSpecification, cfg *rest.Config) (*Controller, error) {
@@ -68,10 +70,11 @@ func NewWithDetail(spec *AgentSpecification, syncerConf *broker.SyncerConfig, re
 	}
 
 	svcExportResourceConfig := broker.ResourceConfig{
-		LocalSourceNamespace: metav1.NamespaceAll,
-		LocalResourceType:    &lighthousev2a1.ServiceExport{},
-		LocalTransform:       agentController.serviceExportToRemoteServiceImport,
-		BrokerResourceType:   &lighthousev2a1.ServiceImport{},
+		LocalSourceNamespace:  metav1.NamespaceAll,
+		LocalResourceType:     &lighthousev2a1.ServiceExport{},
+		LocalTransform:        agentController.serviceExportToRemoteServiceImport,
+		LocalOnSuccessfulSync: agentController.onSuccessfulServiceImportSync,
+		BrokerResourceType:    &lighthousev2a1.ServiceImport{},
 	}
 
 	syncerConf.Scheme = scheme
@@ -137,13 +140,13 @@ func (a *Controller) serviceExportToRemoteServiceImport(obj runtime.Object, op s
 	svc, err := a.kubeClientSet.CoreV1().Services(svcExport.Namespace).Get(svcExport.Name, metav1.GetOptions{})
 	if errors.IsNotFound(err) {
 		klog.V(log.DEBUG).Infof("Service to be exported (%s/%s) doesn't exist", svcExport.Namespace, svcExport.Name)
-		a.updateExportedServiceStatus(svcExport, lighthousev2a1.ServiceExportInitialized, corev1.ConditionFalse,
-			serviceUnavailable, "Service to be exported doesn't exist")
+		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 {
 		// some other error. Log and requeue
-		a.updateExportedServiceStatus(svcExport, lighthousev2a1.ServiceExportInitialized, corev1.ConditionUnknown,
-			"ServiceRetrievalFailed", fmt.Sprintf("Error retrieving the Service: %v", err))
+		a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, lighthousev2a1.ServiceExportInitialized,
+			corev1.ConditionUnknown, "ServiceRetrievalFailed", fmt.Sprintf("Error retrieving the Service: %v", err))
 		klog.Errorf("Unable to get service for %#v: %v", svc, err)
 		return nil, true
 	}
@@ -151,8 +154,8 @@ func (a *Controller) serviceExportToRemoteServiceImport(obj runtime.Object, op s
 		klog.V(log.DEBUG).Infof("Service to be exported (%s/%s) doesn't have a global IP yet", svcExport.Namespace, svcExport.Name)
 
 		// Globalnet enabled but service doesn't have globalIp yet, Update the status and requeue
-		a.updateExportedServiceStatus(svcExport, lighthousev2a1.ServiceExportInitialized, corev1.ConditionFalse,
-			"ServiceGlobalIPUnavailable", "Service doesn't have a global IP yet")
+		a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, lighthousev2a1.ServiceExportInitialized,
+			corev1.ConditionFalse, "ServiceGlobalIPUnavailable", "Service doesn't have a global IP yet")
 		return nil, true
 	}
 	serviceImport.Spec = lighthousev2a1.ServiceImportSpec{
@@ -163,12 +166,25 @@ func (a *Controller) serviceExportToRemoteServiceImport(obj runtime.Object, op s
 			a.newClusterStatus(svc, a.clusterID),
 		},
 	}
-	a.updateExportedServiceStatus(svcExport, lighthousev2a1.ServiceExportExported, corev1.ConditionTrue,
-		"", "Service was successfully synced to the broker")
+
+	a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, lighthousev2a1.ServiceExportInitialized,
+		corev1.ConditionTrue, "AwaitingSync", "Awaiting sync of the ServiceImport to the broker")
 
 	return serviceImport, false
 }
 
+func (a *Controller) onSuccessfulServiceImportSync(synced runtime.Object, op syncer.Operation) {
+	if op != syncer.Create {
+		return
+	}
+
+	serviceImport := synced.(*lighthousev2a1.ServiceImport)
+
+	a.updateExportedServiceStatus(serviceImport.GetAnnotations()[originName], serviceImport.GetAnnotations()[originNamespace],
+		lighthousev2a1.ServiceExportExported, corev1.ConditionTrue,
+		"", "Service was successfully synced to the broker")
+}
+
 func (a *Controller) serviceToRemoteServiceImport(obj runtime.Object, op syncer.Operation) (runtime.Object, bool) {
 	if op != syncer.Delete {
 		// Ignore create/update
@@ -189,18 +205,18 @@ func (a *Controller) serviceToRemoteServiceImport(obj runtime.Object, op syncer.
 	serviceImport := a.newServiceImport(svcExport)
 
 	// Update the status and requeue
-	a.updateExportedServiceStatus(svcExport, lighthousev2a1.ServiceExportInitialized, corev1.ConditionFalse,
-		serviceUnavailable, "Service to be exported doesn't exist")
+	a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, lighthousev2a1.ServiceExportInitialized,
+		corev1.ConditionFalse, serviceUnavailable, "Service to be exported doesn't exist")
 
 	return serviceImport, false
 }
 
-func (a *Controller) updateExportedServiceStatus(export *lighthousev2a1.ServiceExport, condType lighthousev2a1.ServiceExportConditionType,
+func (a *Controller) updateExportedServiceStatus(name, namespace string, condType lighthousev2a1.ServiceExportConditionType,
 	status corev1.ConditionStatus, reason, msg string) {
 	retryErr := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		toUpdate, err := a.lighthouseClient.LighthouseV2alpha1().ServiceExports(export.Namespace).Get(export.Name, metav1.GetOptions{})
+		toUpdate, err := a.lighthouseClient.LighthouseV2alpha1().ServiceExports(namespace).Get(name, metav1.GetOptions{})
 		if errors.IsNotFound(err) {
-			klog.Infof("ServiceExport (%s/%s) not found - unable to update status", export.Namespace, export.Name)
+			klog.Infof("ServiceExport (%s/%s) not found - unable to update status", namespace, name)
 			return nil
 		} else if err != nil {
 			return err
@@ -226,7 +242,7 @@ func (a *Controller) updateExportedServiceStatus(export *lighthousev2a1.ServiceE
 		return err
 	})
 	if retryErr != nil {
-		klog.Errorf("Error updating status for %#v: %v", export, retryErr)
+		klog.Errorf("Error updating status for ServiceExport (%s/%s): %v", namespace, name, retryErr)
 	}
 }
 
@@ -246,8 +262,8 @@ func (a *Controller) newServiceImport(svcExport *lighthousev2a1.ServiceExport) *
 		ObjectMeta: metav1.ObjectMeta{
 			Name: svcExport.Name + "-" + svcExport.Namespace + "-" + a.clusterID,
 			Annotations: map[string]string{
-				"origin-name":      svcExport.Name,
-				"origin-namespace": svcExport.Namespace,
+				originName:      svcExport.Name,
+				originNamespace: svcExport.Namespace,
 			},
 		},
 	}


Test patch:
diff --git a/pkg/agent/controller/agent_test.go b/pkg/agent/controller/agent_test.go
index 3d6fefe..6e2a5f2 100644
--- a/pkg/agent/controller/agent_test.go
+++ b/pkg/agent/controller/agent_test.go
@@ -1,6 +1,7 @@
 package controller_test
 
 import (
+	"fmt"
 	"time"
 
 	. "github.com/onsi/ginkgo"
@@ -14,6 +15,7 @@ import (
 	fakeLighthouseClientset "github.com/submariner-io/lighthouse/pkg/client/clientset/versioned/fake"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -74,6 +76,27 @@ var _ = Describe("ServiceImport syncing", func() {
 			t.awaitServiceUnavailableStatus()
 		})
 	})
+
+	When("the ServiceImport sync initially fails", func() {
+		BeforeEach(func() {
+			t.brokerServiceImportClient.PersistentFailOnCreate.Store("mock create error")
+		})
+
+		It("should not update the ServiceExport status to Exported until the sync is successful", func() {
+			t.createService()
+			t.createServiceExport()
+
+			t.awaitServiceExportStatus(lighthousev2a1.ServiceExportInitialized, corev1.ConditionTrue, "AwaitingSync")
+
+			t.awaitNotServiceExportStatus(&lighthousev2a1.ServiceExportCondition{
+				Type:   lighthousev2a1.ServiceExportExported,
+				Status: corev1.ConditionTrue,
+			})
+
+			t.brokerServiceImportClient.PersistentFailOnCreate.Store("")
+			t.awaitServiceExported(t.service.Spec.ClusterIP)
+		})
+	})
 })
 
 var _ = Describe("Globalnet enabled", func() {
@@ -99,12 +122,7 @@ var _ = Describe("Globalnet enabled", func() {
 
 	When("a local ServiceExport is created and the Service does not initially have a global IP", func() {
 		It("should eventually sync a ServiceImport with the global IP of the Service", func() {
-			reason := "ServiceGlobalIPUnavailable"
-			t.awaitServiceExportStatus(&lighthousev2a1.ServiceExportCondition{
-				Type:   lighthousev2a1.ServiceExportInitialized,
-				Status: corev1.ConditionFalse,
-				Reason: &reason,
-			})
+			t.awaitServiceExportStatus(lighthousev2a1.ServiceExportInitialized, corev1.ConditionFalse, "ServiceGlobalIPUnavailable")
 
 			t.service.SetAnnotations(map[string]string{"submariner.io/globalIp": globalIP})
 			_, err := t.localServiceClient.CoreV1().Services(t.service.Namespace).Update(t.service)
@@ -119,16 +137,17 @@ type testDriver struct {
 	agentController           *controller.Controller
 	agentSpec                 *controller.AgentSpecification
 	localDynClient            dynamic.Interface
+	brokerDynClient           dynamic.Interface
 	localServiceExportClient  dynamic.ResourceInterface
 	localServiceImportClient  dynamic.ResourceInterface
-	brokerServiceImportClient dynamic.ResourceInterface
+	brokerServiceImportClient *fake.DynamicResourceClient
 	localServiceClient        kubernetes.Interface
 	lighthouseClient          lighthouseClientset.Interface
-	initialResources          []runtime.Object
 	service                   *corev1.Service
 	serviceExport             *lighthousev2a1.ServiceExport
 	stopCh                    chan struct{}
 	now                       time.Time
+	restMapper                meta.RESTMapper
 }
 
 func newTestDiver() *testDriver {
@@ -140,7 +159,6 @@ func newTestDiver() *testDriver {
 
 	BeforeEach(func() {
 		t.now = time.Now()
-		t.initialResources = nil
 		t.stopCh = make(chan struct{})
 
 		t.service = &corev1.Service{
@@ -159,23 +177,23 @@ func newTestDiver() *testDriver {
 				Namespace: t.service.Namespace,
 			},
 		}
-	})
 
-	JustBeforeEach(func() {
-		restMapper := test.GetRESTMapperFor(&lighthousev2a1.ServiceExport{}, &lighthousev2a1.ServiceImport{}, &corev1.Service{})
+		t.restMapper = test.GetRESTMapperFor(&lighthousev2a1.ServiceExport{}, &lighthousev2a1.ServiceImport{}, &corev1.Service{})
 
-		t.localDynClient = fake.NewDynamicClient(test.PrepInitialClientObjs("", "", t.initialResources...)...)
-		brokerDynClient := fake.NewDynamicClient()
+		t.localDynClient = fake.NewDynamicClient()
+		t.brokerDynClient = fake.NewDynamicClient()
 
-		t.localServiceExportClient = t.localDynClient.Resource(*test.GetGroupVersionResourceFor(restMapper,
+		t.localServiceExportClient = t.localDynClient.Resource(*test.GetGroupVersionResourceFor(t.restMapper,
 			&lighthousev2a1.ServiceExport{})).Namespace(test.LocalNamespace)
 
-		t.localServiceImportClient = t.localDynClient.Resource(*test.GetGroupVersionResourceFor(restMapper,
+		t.localServiceImportClient = t.localDynClient.Resource(*test.GetGroupVersionResourceFor(t.restMapper,
 			&lighthousev2a1.ServiceImport{})).Namespace(test.LocalNamespace)
 
-		t.brokerServiceImportClient = brokerDynClient.Resource(*test.GetGroupVersionResourceFor(restMapper,
-			&lighthousev2a1.ServiceImport{})).Namespace(test.RemoteNamespace)
+		t.brokerServiceImportClient = t.brokerDynClient.Resource(*test.GetGroupVersionResourceFor(t.restMapper,
+			&lighthousev2a1.ServiceImport{})).Namespace(test.RemoteNamespace).(*fake.DynamicResourceClient)
+	})
 
+	JustBeforeEach(func() {
 		t.localServiceClient = fakeKubeCLient.NewSimpleClientset()
 
 		t.lighthouseClient = fakeLighthouseClientset.NewSimpleClientset(t.serviceExport)
@@ -189,9 +207,9 @@ func newTestDiver() *testDriver {
 		Expect(lighthousev2a1.AddToScheme(syncerScheme)).To(Succeed())
 
 		var err error
-		t.agentController, err = controller.NewWithDetail(t.agentSpec, syncerConfig, restMapper, t.localDynClient,
+		t.agentController, err = controller.NewWithDetail(t.agentSpec, syncerConfig, t.restMapper, t.localDynClient,
 			t.localServiceClient, t.lighthouseClient, syncerScheme, func(config *broker.SyncerConfig) (*broker.Syncer, error) {
-				return broker.NewSyncerWithDetail(config, t.localDynClient, brokerDynClient, restMapper)
+				return broker.NewSyncerWithDetail(config, t.localDynClient, t.brokerDynClient, t.restMapper)
 			})
 
 		Expect(err).To(Succeed())
@@ -255,7 +273,8 @@ func (t *testDriver) awaitNoServiceImport(client dynamic.ResourceInterface) {
 	test.WaitForNoResource(client, t.service.Name+"-"+t.service.Namespace+"-"+clusterID)
 }
 
-func (t *testDriver) awaitServiceExportStatus(expCond *lighthousev2a1.ServiceExportCondition) {
+func (t *testDriver) awaitServiceExportStatus(expType lighthousev2a1.ServiceExportConditionType,
+	expStatus corev1.ConditionStatus, expReason string) {
 	var found *lighthousev2a1.ServiceExport
 
 	err := wait.PollImmediate(50*time.Millisecond, 5*time.Second, func() (bool, error) {
@@ -270,8 +289,8 @@ func (t *testDriver) awaitServiceExportStatus(expCond *lighthousev2a1.ServiceExp
 
 		found = se
 
-		return len(se.Status.Conditions) > 0 && se.Status.Conditions[0].Type == expCond.Type &&
-			se.Status.Conditions[0].Status == expCond.Status, nil
+		return len(se.Status.Conditions) > 0 && se.Status.Conditions[0].Type == expType &&
+			se.Status.Conditions[0].Status == expStatus, nil
 	})
 
 	if err == wait.ErrWaitTimeout {
@@ -283,31 +302,50 @@ func (t *testDriver) awaitServiceExportStatus(expCond *lighthousev2a1.ServiceExp
 	}
 
 	Expect(found.Status.Conditions).To(HaveLen(1))
-	Expect(found.Status.Conditions[0].Type).To(Equal(expCond.Type))
-	Expect(found.Status.Conditions[0].Status).To(Equal(expCond.Status))
+	Expect(found.Status.Conditions[0].Type).To(Equal(expType))
+	Expect(found.Status.Conditions[0].Status).To(Equal(expStatus))
 	Expect(found.Status.Conditions[0].LastTransitionTime).To(Not(BeNil()))
 	Expect(found.Status.Conditions[0].LastTransitionTime.After(t.now)).To(BeTrue())
-	Expect(found.Status.Conditions[0].Reason).To(Equal(expCond.Reason))
+	Expect(found.Status.Conditions[0].Reason).To(Not(BeNil()))
+	Expect(*found.Status.Conditions[0].Reason).To(Equal(expReason))
 	Expect(found.Status.Conditions[0].Message).To(Not(BeNil()))
 }
 
+func (t *testDriver) awaitNotServiceExportStatus(notCond *lighthousev2a1.ServiceExportCondition) {
+	err := wait.PollImmediate(50*time.Millisecond, 300*time.Millisecond, func() (bool, error) {
+		se, err := t.lighthouseClient.LighthouseV2alpha1().ServiceExports(t.service.Namespace).Get(t.service.Name, metav1.GetOptions{})
+		if apierrors.IsNotFound(err) {
+			return false, nil
+		}
+
+		if err != nil {
+			return false, err
+		}
+
+		if len(se.Status.Conditions) == 0 {
+			return false, nil
+		}
+
+		last := &se.Status.Conditions[len(se.Status.Conditions)-1]
+		if last.Type == notCond.Type && last.Status == notCond.Status {
+			return false, fmt.Errorf("Received unexpected %#v", last)
+		}
+
+		return false, nil
+	})
+
+	if err != wait.ErrWaitTimeout {
+		Fail(err.Error())
+	}
+}
+
 func (t *testDriver) awaitServiceExported(serviceIP string) {
 	t.awaitServiceImport(t.brokerServiceImportClient, serviceIP)
 	t.awaitServiceImport(t.localServiceImportClient, serviceIP)
 
-	empty := ""
-	t.awaitServiceExportStatus(&lighthousev2a1.ServiceExportCondition{
-		Type:   lighthousev2a1.ServiceExportExported,
-		Status: corev1.ConditionTrue,
-		Reason: &empty,
-	})
+	t.awaitServiceExportStatus(lighthousev2a1.ServiceExportExported, corev1.ConditionTrue, "")
 }
 
 func (t *testDriver) awaitServiceUnavailableStatus() {
-	reason := "ServiceUnavailable"
-	t.awaitServiceExportStatus(&lighthousev2a1.ServiceExportCondition{
-		Type:   lighthousev2a1.ServiceExportInitialized,
-		Status: corev1.ConditionFalse,
-		Reason: &reason,
-	})
+	t.awaitServiceExportStatus(lighthousev2a1.ServiceExportInitialized, corev1.ConditionFalse, "ServiceUnavailable")
 }