Repository: submariner-io/lighthouse
Pull request #899 (2022-10-11 20:32:23)

Add Synced status condition type to ServiceExport
 After export validation, the `Valid` status condition type is also used to reflect the broker sync conditions, ie awaiting sync and successful sync. However these condition statuses aren't really related to validating export criteria. Thus a new condition type, `Synced`, is introduced and the logic to set the statuses are part of the broker syncer. The awaiting sync status is set in the `LocalTransform` function of the ServiceImport broker syncer and  the successful status is set in the `LocalOnSuccessfulSync` function.
    
Separating the condition statuses will also make it easier to handle exported Service updates (see https://github.com/submariner-io/lighthouse/pull/879).


submariner-bot: 🤖 Created branch: z_pr899/tpantelis/synced_condition_type 
🚀 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.

Patch:
diff --git a/pkg/agent/controller/agent.go b/pkg/agent/controller/agent.go
index fac4a60..3cc5c19 100644
--- a/pkg/agent/controller/agent.go
+++ b/pkg/agent/controller/agent.go
@@ -83,9 +83,11 @@ func New(spec *AgentSpecification, syncerConf broker.SyncerConfig, kubeClientSet
 
 	syncerConf.ResourceConfigs = []broker.ResourceConfig{
 		{
-			LocalSourceNamespace: metav1.NamespaceAll,
-			LocalResourceType:    &mcsv1a1.ServiceImport{},
-			BrokerResourceType:   &mcsv1a1.ServiceImport{},
+			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",
@@ -121,15 +123,17 @@ func New(spec *AgentSpecification, syncerConf broker.SyncerConfig, kubeClientSet
 	}
 
 	agentController.serviceExportSyncer, err = syncer.NewResourceSyncer(&syncer.ResourceSyncerConfig{
-		Name:             "ServiceExport -> ServiceImport",
-		SourceClient:     syncerConf.LocalClient,
-		SourceNamespace:  metav1.NamespaceAll,
-		RestMapper:       syncerConf.RestMapper,
-		Federator:        agentController.serviceImportSyncer.GetLocalFederator(),
-		ResourceType:     &mcsv1a1.ServiceExport{},
-		Transform:        agentController.serviceExportToServiceImport,
-		OnSuccessfulSync: agentController.onSuccessfulServiceImportSync,
-		Scheme:           syncerConf.Scheme,
+		Name:            "ServiceExport -> ServiceImport",
+		SourceClient:    syncerConf.LocalClient,
+		SourceNamespace: metav1.NamespaceAll,
+		RestMapper:      syncerConf.RestMapper,
+		Federator:       agentController.serviceImportSyncer.GetLocalFederator(),
+		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",
@@ -245,8 +249,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, corev1.ConditionUnknown, "ServiceRetrievalFailed",
-			fmt.Sprintf("Error retrieving the Service: %v", err))
+		a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid, 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
@@ -254,23 +258,19 @@ func (a *Controller) serviceExportToServiceImport(obj runtime.Object, numRequeue
 
 	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, corev1.ConditionFalse, serviceUnavailable,
-			"Service to be exported doesn't exist")
+		a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid, corev1.ConditionFalse,
+			serviceUnavailable, "Service to be exported doesn't exist")
 
 		return nil, true
 	}
 
-	if op == syncer.Update && getLastExportConditionReason(svcExport) != serviceUnavailable {
-		return nil, false
-	}
-
 	svc := obj.(*corev1.Service)
 
 	svcType, ok := getServiceImportType(svc)
 
 	if !ok {
-		a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, corev1.ConditionFalse, invalidServiceType,
-			fmt.Sprintf("Service of type %v not supported", svc.Spec.Type))
+		a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid, corev1.ConditionFalse,
+			invalidServiceType, fmt.Sprintf("Service of type %v not supported", svc.Spec.Type))
 		klog.Errorf("Service type %q not supported", svc.Spec.Type)
 
 		return nil, true
@@ -298,7 +298,8 @@ func (a *Controller) serviceExportToServiceImport(obj runtime.Object, numRequeue
 			if ip == "" {
 				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.Name, svcExport.Namespace, corev1.ConditionFalse, reason, msg)
+				a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid,
+					corev1.ConditionFalse, reason, msg)
 
 				return nil, true
 			}
@@ -315,23 +316,13 @@ func (a *Controller) serviceExportToServiceImport(obj runtime.Object, numRequeue
 		serviceImport.Annotations[clusterIP] = serviceImport.Spec.IPs[0]
 	}
 
-	a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, corev1.ConditionFalse, "AwaitingSync",
-		"Awaiting sync of the ServiceImport to the broker")
+	a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid, corev1.ConditionTrue, "", "")
 
 	klog.V(log.DEBUG).Infof("Returning ServiceImport: %#v", serviceImport)
 
 	return serviceImport, false
 }
 
-func getLastExportConditionReason(svcExport *mcsv1a1.ServiceExport) string {
-	numCond := len(svcExport.Status.Conditions)
-	if numCond > 0 && svcExport.Status.Conditions[numCond-1].Reason != nil {
-		return *svcExport.Status.Conditions[numCond-1].Reason
-	}
-
-	return ""
-}
-
 func getServiceImportType(service *corev1.Service) (mcsv1a1.ServiceImportType, bool) {
 	if service.Spec.Type != "" && service.Spec.Type != corev1.ServiceTypeClusterIP {
 		return "", false
@@ -344,6 +335,24 @@ 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.GetAnnotations()[lhconstants.OriginName],
+			serviceImport.GetAnnotations()[lhconstants.OriginNamespace], lhconstants.ServiceExportSynced, corev1.ConditionFalse,
+			"NoServiceImport", "ServiceImport was deleted")
+
+		return obj, false
+	}
+
+	a.updateExportedServiceStatus(serviceImport.GetAnnotations()[lhconstants.OriginName],
+		serviceImport.GetAnnotations()[lhconstants.OriginNamespace], lhconstants.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
@@ -352,8 +361,31 @@ func (a *Controller) onSuccessfulServiceImportSync(synced runtime.Object, op syn
 	serviceImport := synced.(*mcsv1a1.ServiceImport)
 
 	a.updateExportedServiceStatus(serviceImport.GetAnnotations()[lhconstants.OriginName],
-		serviceImport.GetAnnotations()[lhconstants.OriginNamespace], corev1.ConditionTrue, "",
-		"Service was successfully synced to the broker")
+		serviceImport.GetAnnotations()[lhconstants.OriginNamespace], lhconstants.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)
+
+	if newValidCond != nil && !reflect.DeepEqual(oldValidCond, newValidCond) && newValidCond.Status == corev1.ConditionFalse {
+		return true
+	}
+
+	return false
+}
+
+func findServiceExportStatusCondition(conditions []mcsv1a1.ServiceExportCondition,
+	condType mcsv1a1.ServiceExportConditionType,
+) *mcsv1a1.ServiceExportCondition {
+	for i := range conditions {
+		if conditions[i].Type == condType {
+			return &conditions[i]
+		}
+	}
+
+	return nil
 }
 
 func (a *Controller) serviceToRemoteServiceImport(obj runtime.Object, numRequeues int, op syncer.Operation) (runtime.Object, bool) {
@@ -381,15 +413,17 @@ 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, corev1.ConditionFalse, serviceUnavailable,
-		"Service to be exported doesn't exist")
+	a.updateExportedServiceStatus(svcExport.Name, svcExport.Namespace, mcsv1a1.ServiceExportValid, corev1.ConditionFalse,
+		serviceUnavailable, "Service to be exported doesn't exist")
 
 	return serviceImport, false
 }
 
-func (a *Controller) updateExportedServiceStatus(name, namespace string, status corev1.ConditionStatus, reason, msg string) {
+func (a *Controller) updateExportedServiceStatus(name, namespace string, condType mcsv1a1.ServiceExportConditionType,
+	status corev1.ConditionStatus, reason, msg string,
+) {
 	klog.V(log.DEBUG).Infof("updateExportedServiceStatus for (%s/%s) - Type: %q, Status: %q, Reason: %q, Message: %q",
-		namespace, name, mcsv1a1.ServiceExportValid, status, reason, msg)
+		namespace, name, condType, status, reason, msg)
 
 	retryErr := retry.RetryOnConflict(retry.DefaultRetry, func() error {
 		toUpdate, err := a.getServiceExport(name, namespace)
@@ -401,23 +435,25 @@ func (a *Controller) updateExportedServiceStatus(name, namespace string, status
 		}
 
 		now := metav1.Now()
-		exportCondition := mcsv1a1.ServiceExportCondition{
-			Type:               mcsv1a1.ServiceExportValid,
+		newCondition := mcsv1a1.ServiceExportCondition{
+			Type:               condType,
 			Status:             status,
 			LastTransitionTime: &now,
 			Reason:             &reason,
 			Message:            &msg,
 		}
 
-		// TODO: Currently we only check for conditionType Valid. Revisit this when Conflict is supported.
-		if len(toUpdate.Status.Conditions) > 0 && serviceExportConditionEqual(&toUpdate.Status.Conditions[0], &exportCondition) {
+		prevCond := findServiceExportStatusCondition(toUpdate.Status.Conditions, condType)
+		if prevCond == nil {
+			toUpdate.Status.Conditions = append(toUpdate.Status.Conditions, newCondition)
+		} else if serviceExportConditionEqual(prevCond, &newCondition) {
 			klog.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
 		}
 
-		toUpdate.Status.Conditions = []mcsv1a1.ServiceExportCondition{exportCondition}
-
 		raw, err := resource.ToUnstructured(toUpdate)
 		if err != nil {
 			return errors.Wrap(err, "error converting resource")
@@ -439,14 +475,18 @@ func (a *Controller) getServiceExport(name, namespace string) (*mcsv1a1.ServiceE
 		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)
+	err := a.serviceImportController.scheme.Convert(obj, se, nil)
 	if err != nil {
-		return nil, errors.WithMessagef(err, "Error converting %#v to ServiceExport", obj)
+		panic(errors.WithMessagef(err, "Error converting %#v to ServiceExport", obj))
 	}
 
-	return se, nil
+	return se
 }
 
 func serviceExportConditionEqual(c1, c2 *mcsv1a1.ServiceExportCondition) bool {
diff --git a/pkg/constants/constants.go b/pkg/constants/constants.go
index c09c3dc..b728a17 100644
--- a/pkg/constants/constants.go
+++ b/pkg/constants/constants.go
@@ -18,6 +18,8 @@ limitations under the License.
 
 package constants
 
+import "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
+
 const (
 	OriginName                         = "origin-name"
 	OriginNamespace                    = "origin-namespace"
@@ -30,3 +32,5 @@ const (
 	MCSLabelSourceCluster              = "multicluster.kubernetes.io/source-cluster"
 	KubernetesServiceName              = "kubernetes.io/service-name"
 )
+
+const ServiceExportSynced v1alpha1.ServiceExportConditionType = "Synced"
diff --git a/release-notes/20221011-new-synced-condition.md b/release-notes/20221011-new-synced-condition.md
new file mode 100644
index 0000000..09964c2
--- /dev/null
+++ b/release-notes/20221011-new-synced-condition.md
@@ -0,0 +1,3 @@
+<!-- markdownlint-disable MD041 -->
+A new ServiceExport status condition type named Synced was added that indicates whether or not the ServiceImport
+was successfully synced to the broker.


Test patch:
diff --git a/pkg/agent/controller/controller_suite_test.go b/pkg/agent/controller/controller_suite_test.go
index 52a8150..b594408 100644
--- a/pkg/agent/controller/controller_suite_test.go
+++ b/pkg/agent/controller/controller_suite_test.go
@@ -21,8 +21,8 @@ package controller_test
 import (
 	"context"
 	"crypto/rand"
+	"encoding/json"
 	"errors"
-	"fmt"
 	"math/big"
 	"reflect"
 	"sort"
@@ -45,10 +45,12 @@ 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"
 	"k8s.io/klog/v2"
 	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
@@ -565,10 +567,42 @@ func (t *testDriver) awaitNoEndpointSlice(client dynamic.ResourceInterface) {
 	test.AwaitNoResource(client, t.endpoints.Name+"-"+clusterID1)
 }
 
-func (t *testDriver) awaitServiceExportStatus(expCond ...*mcsv1a1.ServiceExportCondition) {
+func (t *testDriver) assertServiceExportCondition(se *mcsv1a1.ServiceExport, expCond *mcsv1a1.ServiceExportCondition) bool {
+	for i := range se.Status.Conditions {
+		actual := &se.Status.Conditions[i]
+		if actual.Type == expCond.Type {
+			out, _ := json.MarshalIndent(actual, "", "  ")
+
+			Expect(actual.Status).To(Equal(expCond.Status), "Actual: %s", out)
+			Expect(actual.LastTransitionTime).To(Not(BeNil()), "Actual: %s", out)
+			Expect(actual.Reason).To(Not(BeNil()), "Actual: %s", out)
+			Expect(*actual.Reason).To(Equal(*expCond.Reason), "Actual: %s", out)
+			Expect(actual.Message).To(Not(BeNil()), "Actual: %s", out)
+
+			if expCond.Message != nil {
+				Expect(*actual.Message).To(Equal(*expCond.Message), "Actual: %s", out)
+			}
+
+			return true
+		}
+	}
+
+	return false
+}
+
+func (t *testDriver) toServiceExport(obj interface{}) *mcsv1a1.ServiceExport {
+	se := &mcsv1a1.ServiceExport{}
+	Expect(scheme.Scheme.Convert(obj, se, nil)).To(Succeed())
+
+	return se
+}
+
+func (t *testDriver) awaitServiceExportCondition(expCond *mcsv1a1.ServiceExportCondition) {
 	var found *mcsv1a1.ServiceExport
 
 	err := wait.PollImmediate(50*time.Millisecond, 5*time.Second, func() (bool, error) {
+		defer GinkgoRecover()
+
 		obj, err := t.cluster1.localServiceExportClient.Get(context.TODO(), t.service.Name, metav1.GetOptions{})
 		if err != nil {
 			if apierrors.IsNotFound(err) {
@@ -578,29 +612,8 @@ func (t *testDriver) awaitServiceExportStatus(expCond ...*mcsv1a1.ServiceExportC
 			return false, err
 		}
 
-		se := &mcsv1a1.ServiceExport{}
-		Expect(scheme.Scheme.Convert(obj, se, nil)).To(Succeed())
-
-		found = se
-
-		if len(expCond) != len(se.Status.Conditions) {
-			return false, nil
-		}
-
-		exp := expCond[0]
-		actual := se.Status.Conditions[0]
-		Expect(actual.Type).To(Equal(exp.Type))
-		Expect(actual.Status).To(Equal(exp.Status))
-		Expect(actual.LastTransitionTime).To(Not(BeNil()))
-		Expect(actual.Reason).To(Not(BeNil()))
-		Expect(*actual.Reason).To(Equal(*exp.Reason))
-		Expect(actual.Message).To(Not(BeNil()))
-
-		if exp.Message != nil {
-			Expect(*actual.Message).To(Equal(*exp.Message))
-		}
-
-		return true, nil
+		found = t.toServiceExport(obj)
+		return t.assertServiceExportCondition(found, expCond), nil
 	})
 
 	if errors.Is(err, wait.ErrWaitTimeout) {
@@ -608,41 +621,55 @@ func (t *testDriver) awaitServiceExportStatus(expCond ...*mcsv1a1.ServiceExportC
 			Fail("ServiceExport not found")
 		}
 
-		Fail(format.Message(found.Status.Conditions, fmt.Sprintf("to contain at index %d", 0), expCond))
+		Fail(format.Message(found.Status.Conditions, "to contain", expCond))
 	} else {
 		Expect(err).To(Succeed())
 	}
 }
 
-func (t *testDriver) awaitNotServiceExportStatus(notCond *mcsv1a1.ServiceExportCondition) {
-	err := wait.PollImmediate(50*time.Millisecond, 300*time.Millisecond, func() (bool, error) {
+func (t *testDriver) ensureNoServiceExportCondition(condType mcsv1a1.ServiceExportConditionType) {
+	Consistently(func() interface{} {
 		obj, err := t.cluster1.localServiceExportClient.Get(context.TODO(), t.service.Name, metav1.GetOptions{})
-		if apierrors.IsNotFound(err) {
-			return false, nil
-		}
-
-		if err != nil {
-			return false, err
-		}
+		Expect(err).To(Succeed())
 
 		se := &mcsv1a1.ServiceExport{}
 		Expect(scheme.Scheme.Convert(obj, se, nil)).To(Succeed())
 
-		if len(se.Status.Conditions) == 0 {
-			return false, nil
+		for i := range se.Status.Conditions {
+			if se.Status.Conditions[i].Type == condType {
+				return &se.Status.Conditions[i]
+			}
 		}
 
-		last := &se.Status.Conditions[len(se.Status.Conditions)-1]
-		if last.Message == notCond.Message && last.Status == notCond.Status {
-			return false, fmt.Errorf("Received unexpected %#v", last)
-		}
+		return nil
+	}).Should(BeNil(), "Unexpected ServiceExport status condition")
+}
 
-		return false, nil
+func (t *testDriver) ensureServiceExportCondition(expCond *mcsv1a1.ServiceExportCondition) {
+	_, 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{}) {
+			defer GinkgoRecover()
+			Expect(t.assertServiceExportCondition(t.toServiceExport(obj), expCond)).To(BeTrue())
+		},
+		UpdateFunc: func(oldObj, newObj interface{}) {
+			defer GinkgoRecover()
+			Expect(t.assertServiceExportCondition(t.toServiceExport(newObj), expCond)).To(BeTrue())
+		},
 	})
 
-	if !errors.Is(err, wait.ErrWaitTimeout) {
-		Fail(err.Error())
-	}
+	stopCh := make(chan struct{})
+	go informer.Run(stopCh)
+	Expect(cache.WaitForCacheSync(stopCh, informer.HasSynced)).To(BeTrue())
+
+	time.Sleep(time.Millisecond * 100)
+	close(stopCh)
 }
 
 func (t *testDriver) awaitServiceExported(serviceIP string) {
@@ -651,7 +678,11 @@ func (t *testDriver) awaitServiceExported(serviceIP string) {
 	t.awaitBrokerServiceImport(mcsv1a1.ClusterSetIP, serviceIP)
 	t.cluster2.awaitServiceImport(t.service, mcsv1a1.ClusterSetIP, serviceIP)
 
-	t.awaitServiceExportStatus(newServiceExportCondition(corev1.ConditionTrue, ""))
+	t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionTrue, ""))
+
+	expCond := newServiceExportSyncedCondition(corev1.ConditionTrue, "")
+	t.awaitServiceExportCondition(expCond)
+	t.ensureServiceExportCondition(expCond)
 }
 
 func (t *testDriver) awaitHeadlessServiceImport() {
@@ -685,7 +716,7 @@ func (t *testDriver) awaitHeadlessServiceUnexported() {
 }
 
 func (t *testDriver) awaitServiceUnavailableStatus() {
-	t.awaitServiceExportStatus(newServiceExportCondition(corev1.ConditionFalse, "ServiceUnavailable"))
+	t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "ServiceUnavailable"))
 }
 
 func (t *testDriver) endpointIPs() []string {
@@ -697,7 +728,7 @@ func (t *testDriver) endpointIPs() []string {
 	return ips
 }
 
-func newServiceExportCondition(status corev1.ConditionStatus, reason string) *mcsv1a1.ServiceExportCondition {
+func newServiceExportValidCondition(status corev1.ConditionStatus, reason string) *mcsv1a1.ServiceExportCondition {
 	return &mcsv1a1.ServiceExportCondition{
 		Type:   mcsv1a1.ServiceExportValid,
 		Status: status,
@@ -705,6 +736,14 @@ func newServiceExportCondition(status corev1.ConditionStatus, reason string) *mc
 	}
 }
 
+func newServiceExportSyncedCondition(status corev1.ConditionStatus, reason string) *mcsv1a1.ServiceExportCondition {
+	return &mcsv1a1.ServiceExportCondition{
+		Type:   lhconstants.ServiceExportSynced,
+		Status: status,
+		Reason: &reason,
+	}
+}
+
 func setIngressIPConditions(ingressIP *unstructured.Unstructured, conditions ...metav1.Condition) {
 	var err error
 
diff --git a/pkg/agent/controller/globalnet_test.go b/pkg/agent/controller/globalnet_test.go
index cddcba6..af7db9d 100644
--- a/pkg/agent/controller/globalnet_test.go
+++ b/pkg/agent/controller/globalnet_test.go
@@ -69,7 +69,7 @@ var _ = Describe("Globalnet enabled", func() {
 		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.awaitServiceExportStatus(newServiceExportCondition(corev1.ConditionFalse, "ServiceGlobalIPUnavailable"))
+					t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "ServiceGlobalIPUnavailable"))
 
 					t.createGlobalIngressIP(ingressIP)
 					t.awaitServiceExported(globalIP1)
@@ -84,7 +84,7 @@ var _ = Describe("Globalnet enabled", func() {
 				})
 
 				It("should update the ServiceExport status appropriately and eventually sync a ServiceImport", func() {
-					t.awaitServiceExportStatus(newServiceExportCondition(corev1.ConditionFalse, "ServiceGlobalIPUnavailable"))
+					t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "ServiceGlobalIPUnavailable"))
 
 					setIngressAllocatedIP(ingressIP, globalIP1)
 					test.UpdateResource(t.cluster1.localIngressIPClient, ingressIP)
@@ -108,9 +108,9 @@ var _ = Describe("Globalnet enabled", func() {
 			})
 
 			It("should update the ServiceExport status with the condition details", func() {
-				c := newServiceExportCondition(corev1.ConditionFalse, condition.Reason)
+				c := newServiceExportValidCondition(corev1.ConditionFalse, condition.Reason)
 				c.Message = &condition.Message
-				t.awaitServiceExportStatus(c)
+				t.awaitServiceExportCondition(c)
 			})
 		})
 	})
diff --git a/pkg/agent/controller/reconciliation_test.go b/pkg/agent/controller/reconciliation_test.go
index 66c61b6..b9d1ade 100644
--- a/pkg/agent/controller/reconciliation_test.go
+++ b/pkg/agent/controller/reconciliation_test.go
@@ -80,6 +80,7 @@ var _ = Describe("Reconciliation", func() {
 			t.cluster1.start(t, *t.syncerConfig)
 
 			t.awaitServiceUnexported()
+			t.awaitServiceExportCondition(newServiceExportSyncedCondition(corev1.ConditionFalse, "NoServiceImport"))
 		})
 	})
 
diff --git a/pkg/agent/controller/service_import_sync_test.go b/pkg/agent/controller/service_import_sync_test.go
index 7ef52e8..1e16eb1 100644
--- a/pkg/agent/controller/service_import_sync_test.go
+++ b/pkg/agent/controller/service_import_sync_test.go
@@ -20,8 +20,8 @@ package controller_test
 
 import (
 	. "github.com/onsi/ginkgo"
+	"github.com/submariner-io/lighthouse/pkg/constants"
 	corev1 "k8s.io/api/core/v1"
-	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
 var _ = Describe("ServiceImport syncing", func() {
@@ -79,6 +79,7 @@ var _ = Describe("ServiceImport syncing", func() {
 			t.deleteService()
 			t.awaitServiceUnexported()
 			t.awaitServiceUnavailableStatus()
+			t.awaitServiceExportCondition(newServiceExportSyncedCondition(corev1.ConditionFalse, "NoServiceImport"))
 
 			t.createService()
 			t.awaitServiceExported(t.service.Spec.ClusterIP)
@@ -90,18 +91,12 @@ var _ = Describe("ServiceImport syncing", func() {
 			t.cluster1.localServiceImportClient.PersistentFailOnCreate.Store("mock create error")
 		})
 
-		It("should not update the ServiceExport status to Exported until the sync is successful", func() {
+		It("should eventually update the ServiceExport status to successfully synced", func() {
 			t.createService()
 			t.createServiceExport()
 
-			message := "AwaitingSync"
-			t.awaitServiceExportStatus(newServiceExportCondition(corev1.ConditionFalse, message))
-
-			t.awaitNotServiceExportStatus(&mcsv1a1.ServiceExportCondition{
-				Type:    mcsv1a1.ServiceExportValid,
-				Status:  corev1.ConditionTrue,
-				Message: &message,
-			})
+			t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionTrue, ""))
+			t.ensureNoServiceExportCondition(constants.ServiceExportSynced)
 
 			t.cluster1.localServiceImportClient.PersistentFailOnCreate.Store("")
 			t.awaitServiceExported(t.service.Spec.ClusterIP)
@@ -117,7 +112,7 @@ var _ = Describe("ServiceImport syncing", func() {
 			t.createService()
 			t.createServiceExport()
 
-			t.awaitServiceExportStatus(newServiceExportCondition(corev1.ConditionFalse, "UnsupportedServiceType"))
+			t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "UnsupportedServiceType"))
 			t.awaitNoServiceImport(t.brokerServiceImportClient)
 		})
 	})
diff --git a/test/e2e/framework/framework.go b/test/e2e/framework/framework.go
index 0068d1b..5874992 100644
--- a/test/e2e/framework/framework.go
+++ b/test/e2e/framework/framework.go
@@ -20,6 +20,7 @@ package framework
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"strings"
 
@@ -141,26 +142,27 @@ func (f *Framework) NewServiceExport(cluster framework.ClusterIndex, name, names
 }
 
 func (f *Framework) AwaitServiceExportedStatusCondition(cluster framework.ClusterIndex, name, namespace string) {
-	se := MCSClients[cluster].MulticlusterV1alpha1().ServiceExports(namespace)
 	By(fmt.Sprintf("Retrieving ServiceExport %s.%s on %q", name, namespace, framework.TestContext.ClusterIDs[cluster]))
+
+	se := MCSClients[cluster].MulticlusterV1alpha1().ServiceExports(namespace)
+
 	framework.AwaitUntil("retrieve ServiceExport", func() (interface{}, error) {
 		return se.Get(context.TODO(), name, metav1.GetOptions{})
 	}, func(result interface{}) (bool, string, error) {
 		se := result.(*mcsv1a1.ServiceExport)
-		if len(se.Status.Conditions) == 0 {
-			return false, "No ServiceExportConditions", nil
-		}
 
-		last := se.Status.Conditions[len(se.Status.Conditions)-1]
-		if last.Type != mcsv1a1.ServiceExportValid {
-			return false, fmt.Sprintf("ServiceExportCondition Type is %v", last.Type), nil
-		}
+		for i := range se.Status.Conditions {
+			if se.Status.Conditions[i].Type == lhconstants.ServiceExportSynced {
+				if se.Status.Conditions[i].Status != v1.ConditionTrue {
+					out, _ := json.MarshalIndent(se.Status.Conditions[i], "", "  ")
+					return false, fmt.Sprintf("ServiceExport %s condition status is %s", lhconstants.ServiceExportSynced, out), nil
+				}
 
-		if last.Status != v1.ConditionTrue {
-			return false, fmt.Sprintf("ServiceExportCondition Status is %v", last.Status), nil
+				return true, "", nil
+			}
 		}
 
-		return true, "", nil
+		return false, fmt.Sprintf("ServiceExport %s condition status not found", lhconstants.ServiceExportSynced), nil
 	})
 }