Repository: submariner-io/lighthouse (go)

Issue:
Updating an exported service is not handled
**What happened**:
In a working setup, when a service is exported, we can successfully access it from remote clusters.
If we update the service (like the port numbers) in the origin cluster, the corresponding serviceImport is not updated.



Pull request:
Handle update of an exported service
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our developer guide: https://submariner.io/development/
2. Ensure you have added the appropriate tests for your PR: https://submariner.io/development/code-review/#test-new-functionality
3. Read the code review guide to ease the review process: https://submariner.io/development/code-review/
4. If the PR is unfinished, mark it as a draft: https://submariner.io/development/code-review/#mark-work-in-progress-prs-as-drafts
5. If you are using CI to debug, use your private fork: https://submariner.io/development/code-review/#use-private-forks-for-debugging-prs-by-running-ci
6. Add labels to the PR as appropriate.

This template is based on the K8s/K8s template:

https://github.com/kubernetes/kubernetes/blob/master/.github/PULL_REQUEST_TEMPLATE.md
-->

When an exported service update, Service deletion
will trigger ServiceExport update.

Fixes #645
Signed-off-by: blue-troy <12729455+blue-troy@users.noreply.github.com>


Patch:
diff --git a/pkg/agent/controller/agent.go b/pkg/agent/controller/agent.go
index 0ceb7b9..1ba6cc6 100644
--- a/pkg/agent/controller/agent.go
+++ b/pkg/agent/controller/agent.go
@@ -272,7 +272,14 @@ func (a *Controller) serviceExportToServiceImport(obj runtime.Object, numRequeue
 	if !ok {
 		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)
+		klog.Errorf("Service type %q not supported for Service (%s/%s)", svc.Spec.Type, svcExport.Namespace, svcExport.Name)
+
+		err = a.serviceImportSyncer.GetLocalFederator().Delete(a.newServiceImport(svcExport.Name, svcExport.Namespace))
+		if err == nil || apierrors.IsNotFound(err) {
+			return nil, false
+		}
+
+		klog.Errorf("Error deleting ServiceImport for Service (%s/%s)", svcExport.Namespace, svcExport.Name)
 
 		return nil, true
 	}
@@ -367,8 +374,8 @@ func (a *Controller) onSuccessfulServiceImportSync(synced runtime.Object, op syn
 }
 
 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)
+	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
@@ -377,7 +384,7 @@ func (a *Controller) shouldProcessServiceExportUpdate(oldObj, newObj *unstructur
 	return false
 }
 
-func findServiceExportStatusCondition(conditions []mcsv1a1.ServiceExportCondition,
+func FindServiceExportStatusCondition(conditions []mcsv1a1.ServiceExportCondition,
 	condType mcsv1a1.ServiceExportConditionType,
 ) *mcsv1a1.ServiceExportCondition {
 	for i := range conditions {
@@ -390,8 +397,8 @@ func findServiceExportStatusCondition(conditions []mcsv1a1.ServiceExportConditio
 }
 
 func (a *Controller) serviceToRemoteServiceImport(obj runtime.Object, numRequeues int, op syncer.Operation) (runtime.Object, bool) {
-	if op != syncer.Delete {
-		// Ignore create/update
+	if op == syncer.Create {
+		// Ignore create
 		return nil, false
 	}
 
@@ -409,6 +416,10 @@ func (a *Controller) serviceToRemoteServiceImport(obj runtime.Object, numRequeue
 		return nil, false
 	}
 
+	if op == syncer.Update {
+		return a.serviceExportToServiceImport(obj, numRequeues, op)
+	}
+
 	svcExport := obj.(*mcsv1a1.ServiceExport)
 
 	serviceImport := a.newServiceImport(svcExport.Name, svcExport.Namespace)
@@ -444,7 +455,7 @@ func (a *Controller) updateExportedServiceStatus(name, namespace string, condTyp
 			Message:            &msg,
 		}
 
-		prevCond := findServiceExportStatusCondition(toUpdate.Status.Conditions, condType)
+		prevCond := FindServiceExportStatusCondition(toUpdate.Status.Conditions, condType)
 		if prevCond == nil {
 			toUpdate.Status.Conditions = append(toUpdate.Status.Conditions, newCondition)
 		} else if serviceExportConditionEqual(prevCond, &newCondition) {
diff --git a/release-notes/20221018-service-updates.md b/release-notes/20221018-service-updates.md
new file mode 100644
index 0000000..7e1c6c4
--- /dev/null
+++ b/release-notes/20221018-service-updates.md
@@ -0,0 +1,2 @@
+<!-- markdownlint-disable MD041 -->
+Service Discovery now handles updates to an exported service and updates/deletes the corresponding ServiceImport accordingly.


Test patch:
diff --git a/pkg/agent/controller/controller_suite_test.go b/pkg/agent/controller/controller_suite_test.go
index e439e59..05dbdfe 100644
--- a/pkg/agent/controller/controller_suite_test.go
+++ b/pkg/agent/controller/controller_suite_test.go
@@ -23,6 +23,7 @@ import (
 	"crypto/rand"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"math/big"
 	"reflect"
 	"sort"
@@ -31,7 +32,6 @@ import (
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
-	"github.com/onsi/gomega/format"
 	"github.com/submariner-io/admiral/pkg/fake"
 	"github.com/submariner-io/admiral/pkg/syncer/broker"
 	"github.com/submariner-io/admiral/pkg/syncer/test"
@@ -40,7 +40,6 @@ import (
 	lhconstants "github.com/submariner-io/lighthouse/pkg/constants"
 	corev1 "k8s.io/api/core/v1"
 	discovery "k8s.io/api/discovery/v1"
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -96,6 +95,7 @@ type cluster struct {
 	localKubeClient          kubernetes.Interface
 	endpointsReactor         *fake.FailingReactor
 	agentController          *controller.Controller
+	chanByCondType           map[mcsv1a1.ServiceExportConditionType]chan mcsv1a1.ServiceExportCondition
 }
 
 type testDriver struct {
@@ -278,10 +278,60 @@ func (c *cluster) init(syncerConfig *broker.SyncerConfig) {
 	fakeCS := fakeKubeClient.NewSimpleClientset()
 	c.endpointsReactor = fake.NewFailingReactorForResource(&fakeCS.Fake, "endpoints")
 	c.localKubeClient = fakeCS
+
+	c.chanByCondType = map[mcsv1a1.ServiceExportConditionType]chan mcsv1a1.ServiceExportCondition{}
+	c.chanByCondType[mcsv1a1.ServiceExportValid] = make(chan mcsv1a1.ServiceExportCondition, 100)
+	c.chanByCondType[lhconstants.ServiceExportSynced] = make(chan mcsv1a1.ServiceExportCondition, 100)
 }
 
 // nolint:gocritic // (hugeParam) This function modifies syncerConf so we don't want to pass by pointer.
 func (c *cluster) start(t *testDriver, syncerConfig broker.SyncerConfig) {
+	processServiceExportConditions := func(oldObj, newObj interface{}) {
+		defer GinkgoRecover()
+
+		var oldSE *mcsv1a1.ServiceExport
+		if oldObj != nil {
+			oldSE = toServiceExport(oldObj)
+		}
+
+		newSE := toServiceExport(newObj)
+
+		for i := range newSE.Status.Conditions {
+			newCond := &newSE.Status.Conditions[i]
+
+			condChan := c.chanByCondType[newCond.Type]
+			if condChan == nil {
+				continue
+			}
+
+			if oldSE != nil {
+				prevCond := controller.FindServiceExportStatusCondition(oldSE.Status.Conditions, newCond.Type)
+				if prevCond != nil && reflect.DeepEqual(prevCond, newCond) {
+					continue
+				}
+			}
+
+			condChan <- *newCond
+		}
+	}
+
+	_, 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{}) {
+			processServiceExportConditions(nil, obj)
+		},
+		UpdateFunc: processServiceExportConditions,
+	})
+
+	go informer.Run(t.stopCh)
+	Expect(cache.WaitForCacheSync(t.stopCh, informer.HasSynced)).To(BeTrue())
+
 	syncerConfig.LocalClient = c.localDynClient
 	bigint, err := rand.Int(rand.Reader, big.NewInt(1000000))
 	Expect(err).To(Succeed())
@@ -515,6 +565,13 @@ func (t *testDriver) createService() {
 	test.CreateResource(t.cluster1.dynamicServiceClient().Namespace(t.service.Namespace), t.service)
 }
 
+func (t *testDriver) updateService() {
+	_, err := t.cluster1.localKubeClient.CoreV1().Services(t.service.Namespace).Update(context.TODO(), t.service, metav1.UpdateOptions{})
+	Expect(err).To(Succeed())
+
+	test.UpdateResource(t.cluster1.dynamicServiceClient().Namespace(t.service.Namespace), t.service)
+}
+
 func (t *testDriver) createEndpoints() {
 	_, err := t.cluster1.localKubeClient.CoreV1().Endpoints(t.endpoints.Namespace).Create(context.TODO(), t.endpoints, metav1.CreateOptions{})
 	Expect(err).To(Succeed())
@@ -568,109 +625,87 @@ func (t *testDriver) awaitNoEndpointSlice(client dynamic.ResourceInterface) {
 	test.AwaitNoResource(client, t.endpoints.Name+"-"+clusterID1)
 }
 
-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, "", "  ")
+func assertEquivalentConditions(actual, expected *mcsv1a1.ServiceExportCondition) {
+	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)
-			}
+	Expect(actual.Status).To(Equal(expected.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(*expected.Reason), "Actual: %s", out)
+	Expect(actual.Message).To(Not(BeNil()), "Actual: %s", out)
 
-			return true
-		}
+	if expected.Message != nil {
+		Expect(*actual.Message).To(Equal(*expected.Message), "Actual: %s", out)
 	}
-
-	return false
 }
 
-func (t *testDriver) toServiceExport(obj interface{}) *mcsv1a1.ServiceExport {
+func 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
+func (t *testDriver) awaitServiceExportCondition(expected ...*mcsv1a1.ServiceExportCondition) {
+	conditionsEqual := func(actual, expected *mcsv1a1.ServiceExportCondition) bool {
+		return actual != nil && actual.Type == expected.Type && actual.Status == expected.Status &&
+			reflect.DeepEqual(actual.Reason, expected.Reason)
+	}
 
-	err := wait.PollImmediate(50*time.Millisecond, 5*time.Second, func() (bool, error) {
-		defer GinkgoRecover()
+	actual := make([]*mcsv1a1.ServiceExportCondition, len(expected))
+
+	for i := range expected {
+		actual[i] = t.retrieveServiceExportCondition(expected[i].Type)
+		if conditionsEqual(actual[i], expected[i]) {
+			continue
+		}
+
+		condChan := t.cluster1.chanByCondType[expected[i].Type]
+
+		var received *mcsv1a1.ServiceExportCondition
+		nIter := 0
 
-		obj, err := t.cluster1.localServiceExportClient.Get(context.TODO(), t.service.Name, metav1.GetOptions{})
-		if err != nil {
-			if apierrors.IsNotFound(err) {
-				return false, nil
+		for received == nil && nIter <= 50 {
+			select {
+			case c := <-condChan:
+				if conditionsEqual(&c, expected[i]) {
+					received = &c
+				}
+			case <-time.After(100 * time.Millisecond):
 			}
 
-			return false, err
+			nIter++
 		}
 
-		found = t.toServiceExport(obj)
-		return t.assertServiceExportCondition(found, expCond), nil
-	})
-
-	if errors.Is(err, wait.ErrWaitTimeout) {
-		if found == nil {
-			Fail("ServiceExport not found")
+		if received == nil {
+			out, _ := json.MarshalIndent(expected[i], "", "  ")
+			Fail(fmt.Sprintf("ServiceExport condition not received. Expected: %s", out))
 		}
 
-		Fail(format.Message(found.Status.Conditions, "to contain", expCond))
-	} else {
-		Expect(err).To(Succeed())
+		actual[i] = received
 	}
-}
 
-func (t *testDriver) ensureNoServiceExportCondition(condType mcsv1a1.ServiceExportConditionType) {
-	Consistently(func() interface{} {
-		obj, err := t.cluster1.localServiceExportClient.Get(context.TODO(), t.service.Name, metav1.GetOptions{})
-		Expect(err).To(Succeed())
+	for i := range expected {
+		assertEquivalentConditions(actual[i], expected[i])
+	}
 
-		se := &mcsv1a1.ServiceExport{}
-		Expect(scheme.Scheme.Convert(obj, se, nil)).To(Succeed())
+	time.Sleep(time.Millisecond * 100)
 
-		for i := range se.Status.Conditions {
-			if se.Status.Conditions[i].Type == condType {
-				return &se.Status.Conditions[i]
-			}
-		}
+	lastCond := expected[len(expected)-1]
+	assertEquivalentConditions(t.retrieveServiceExportCondition(lastCond.Type), lastCond)
+}
 
-		return nil
+func (t *testDriver) ensureNoServiceExportCondition(condType mcsv1a1.ServiceExportConditionType) {
+	Consistently(func() interface{} {
+		return t.retrieveServiceExportCondition(condType)
 	}).Should(BeNil(), "Unexpected ServiceExport status condition")
 }
 
-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())
-		},
-	})
-
-	stopCh := make(chan struct{})
-	go informer.Run(stopCh)
-	Expect(cache.WaitForCacheSync(stopCh, informer.HasSynced)).To(BeTrue())
+func (t *testDriver) retrieveServiceExportCondition(condType mcsv1a1.ServiceExportConditionType) *mcsv1a1.ServiceExportCondition {
+	obj, err := t.cluster1.localServiceExportClient.Get(context.TODO(), t.service.Name, metav1.GetOptions{})
+	Expect(err).To(Succeed())
 
-	time.Sleep(time.Millisecond * 100)
-	close(stopCh)
+	return controller.FindServiceExportStatusCondition(toServiceExport(obj).Status.Conditions, condType)
 }
 
 func (t *testDriver) awaitServiceExported(serviceIP string) {
@@ -680,10 +715,8 @@ func (t *testDriver) awaitServiceExported(serviceIP string) {
 	t.cluster2.awaitServiceImport(t.service, mcsv1a1.ClusterSetIP, serviceIP)
 
 	t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionTrue, ""))
-
-	expCond := newServiceExportSyncedCondition(corev1.ConditionTrue, "")
-	t.awaitServiceExportCondition(expCond)
-	t.ensureServiceExportCondition(expCond)
+	t.awaitServiceExportCondition(newServiceExportSyncedCondition(corev1.ConditionFalse, "AwaitingSync"),
+		newServiceExportSyncedCondition(corev1.ConditionTrue, ""))
 }
 
 func (t *testDriver) awaitHeadlessServiceImport() {
diff --git a/pkg/agent/controller/service_import_sync_test.go b/pkg/agent/controller/service_import_sync_test.go
index 1e16eb1..36e3cac 100644
--- a/pkg/agent/controller/service_import_sync_test.go
+++ b/pkg/agent/controller/service_import_sync_test.go
@@ -40,7 +40,7 @@ var _ = Describe("ServiceImport syncing", func() {
 	})
 
 	When("a ServiceExport is created", func() {
-		When("the Service already exists", func() {
+		Context("and the Service already exists", func() {
 			It("should correctly sync a ServiceImport and update the ServiceExport status", func() {
 				t.createService()
 				t.createServiceExport()
@@ -48,7 +48,7 @@ var _ = Describe("ServiceImport syncing", func() {
 			})
 		})
 
-		When("the Service doesn't initially exist", func() {
+		Context("and the Service doesn't initially exist", func() {
 			It("should initially update the ServiceExport status to Initialized and eventually sync a ServiceImport", func() {
 				t.createServiceExport()
 				t.awaitServiceUnavailableStatus()
@@ -86,6 +86,21 @@ var _ = Describe("ServiceImport syncing", func() {
 		})
 	})
 
+	When("the type of an exported Service is updated to an unsupported type", func() {
+		It("should delete the ServiceImport and update the ServiceExport status", func() {
+			t.createService()
+			t.createServiceExport()
+			t.awaitServiceExported(t.service.Spec.ClusterIP)
+
+			t.service.Spec.Type = corev1.ServiceTypeNodePort
+			t.updateService()
+
+			t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "UnsupportedServiceType"))
+			t.awaitServiceUnexported()
+			t.awaitServiceExportCondition(newServiceExportSyncedCondition(corev1.ConditionFalse, "NoServiceImport"))
+		})
+	})
+
 	When("the ServiceImport sync initially fails", func() {
 		BeforeEach(func() {
 			t.cluster1.localServiceImportClient.PersistentFailOnCreate.Store("mock create error")
@@ -103,17 +118,31 @@ var _ = Describe("ServiceImport syncing", func() {
 		})
 	})
 
-	When("a ServiceExport is created for a Service whose type is other than ServiceTypeClusterIP", func() {
+	When("a ServiceExport is created for a Service whose type is unsupported", func() {
 		BeforeEach(func() {
 			t.service.Spec.Type = corev1.ServiceTypeNodePort
 		})
 
-		It("should update the ServiceExport status and not sync a ServiceImport", func() {
+		JustBeforeEach(func() {
 			t.createService()
 			t.createServiceExport()
+		})
 
+		It("should update the ServiceExport status appropriately and not sync a ServiceImport", func() {
 			t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "UnsupportedServiceType"))
 			t.awaitNoServiceImport(t.brokerServiceImportClient)
+			t.ensureNoServiceExportCondition(constants.ServiceExportSynced)
+		})
+
+		Context("and is subsequently updated to a supported type", func() {
+			It("should eventually sync a ServiceImport and update the ServiceExport status appropriately", func() {
+				t.awaitServiceExportCondition(newServiceExportValidCondition(corev1.ConditionFalse, "UnsupportedServiceType"))
+
+				t.service.Spec.Type = corev1.ServiceTypeClusterIP
+				t.updateService()
+
+				t.awaitServiceExported(t.service.Spec.ClusterIP)
+			})
 		})
 	})