Repository: submariner-io/lighthouse
Pull request #508 (2021-04-16 21:59:54)

Use admiral syncer reconciliation
The stand-alone local resource syncers (eg `ServiceExport` -> `ServiceImport`) need to manually invoke reconciliation. The broker syncers work as is. Also added a unit tests to cover the various reconciliation scenarios.

Depends on https://github.com/submariner-io/admiral/pull/209/

Related to https://github.com/submariner-io/submariner/issues/1244

submariner-bot: 🤖 Created branch: z_pr508/tpantelis/use_admiral_reconciliation
github-actions[bot]: :hourglass_flowing_sand: Alright! Looks like we need to wait for some *dependencies*:

* submariner-io/admiral#209

:bulb: Don't worry, I will continue watching the list above and keep this comment updated. To add or remove a dependency please update this issue/PR description.

Brought to you by **[Dependent Issues](https://github.com/z0al/dependent-issues)** (:robot: ). Happy coding!
<!-- By Dependent Issues (Action) - DO NOT REMOVE -->
github-actions[bot]: :tada: Great news! Looks like all the *dependencies* have been resolved:

* ~~submariner-io/admiral#209~~

:bulb: To add or remove a dependency please update this issue/PR description.

Brought to you by **[Dependent Issues](https://github.com/z0al/dependent-issues)** (:robot: ). Happy coding!
<!-- By Dependent Issues (Action) - DO NOT REMOVE -->

Patch:
diff --git a/go.mod b/go.mod
index 07b0a68..0f880a3 100644
--- a/go.mod
+++ b/go.mod
@@ -16,7 +16,7 @@ require (
 	github.com/onsi/gomega v1.11.0
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.10.0
-	github.com/submariner-io/admiral v0.9.0-m2
+	github.com/submariner-io/admiral v0.9.0-m2.0.20210419083944-b1dcdc167179
 	github.com/submariner-io/shipyard v0.9.0-m2
 	go.uber.org/zap v1.15.0 // indirect
 	golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e // indirect
diff --git a/go.sum b/go.sum
index d7f5122..f5091c5 100644
--- a/go.sum
+++ b/go.sum
@@ -684,6 +684,14 @@ github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/submariner-io/admiral v0.9.0-m2 h1:Gth1WC3rvsU3iSqWjEqOh+HyUYurGzResrdy95Aabu4=
 github.com/submariner-io/admiral v0.9.0-m2/go.mod h1:NeABccXoo4w9YlguXd9u1Ztw1hkZqsL5EZymW4ddbbs=
+github.com/submariner-io/admiral v0.9.0-m2.0.20210415113737-8be404dbde8a h1:IDQb/Wi9/BYbiX/eSHd/yWnSbXzj4LXB11cIl3UYNF8=
+github.com/submariner-io/admiral v0.9.0-m2.0.20210415113737-8be404dbde8a/go.mod h1:NccJ4ZLR/5R4FDfqX17RO1u9RDQ6dF7SKmCOukivQ1U=
+github.com/submariner-io/admiral v0.9.0-m2.0.20210416033405-d74ba7a14d7a h1:35kpBkr1VhAO6njRBTT0DK83cYbuWUtBKjWGJatR7Kw=
+github.com/submariner-io/admiral v0.9.0-m2.0.20210416033405-d74ba7a14d7a/go.mod h1:NccJ4ZLR/5R4FDfqX17RO1u9RDQ6dF7SKmCOukivQ1U=
+github.com/submariner-io/admiral v0.9.0-m2.0.20210416224604-623e85271ca8 h1:ll0clC5gr6lSPJ0u7ZuSweTLxBuek7tELahkuZpXZzk=
+github.com/submariner-io/admiral v0.9.0-m2.0.20210416224604-623e85271ca8/go.mod h1:NccJ4ZLR/5R4FDfqX17RO1u9RDQ6dF7SKmCOukivQ1U=
+github.com/submariner-io/admiral v0.9.0-m2.0.20210419083944-b1dcdc167179 h1:GD/y4CyK/psteZtQDnl8/t+UbNHAdo6uO8/wRWFjha8=
+github.com/submariner-io/admiral v0.9.0-m2.0.20210419083944-b1dcdc167179/go.mod h1:NccJ4ZLR/5R4FDfqX17RO1u9RDQ6dF7SKmCOukivQ1U=
 github.com/submariner-io/shipyard v0.9.0-m2 h1:Kk8Qc37SSbfQmJ++gfukxvXB+D49oF27g9l5FcKnJ28=
 github.com/submariner-io/shipyard v0.9.0-m2/go.mod h1:y5q1jef9Hph3VGqD7d9W3fXtdLNZ4GR7b7Rf/lrA/go=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
diff --git a/pkg/agent/controller/agent.go b/pkg/agent/controller/agent.go
index f59f6de..29f48f2 100644
--- a/pkg/agent/controller/agent.go
+++ b/pkg/agent/controller/agent.go
@@ -116,7 +116,6 @@ func New(spec *AgentSpecification, syncerConf broker.SyncerConfig, kubeClientSet
 		Name:             "ServiceExport -> ServiceImport",
 		SourceClient:     syncerConf.LocalClient,
 		SourceNamespace:  metav1.NamespaceAll,
-		Direction:        syncer.RemoteToLocal,
 		RestMapper:       syncerConf.RestMapper,
 		Federator:        agentController.serviceImportSyncer.GetLocalFederator(),
 		ResourceType:     &mcsv1a1.ServiceExport{},
@@ -136,7 +135,6 @@ func New(spec *AgentSpecification, syncerConf broker.SyncerConfig, kubeClientSet
 		Name:            "Service deletion",
 		SourceClient:    syncerConf.LocalClient,
 		SourceNamespace: metav1.NamespaceAll,
-		Direction:       syncer.RemoteToLocal,
 		RestMapper:      syncerConf.RestMapper,
 		Federator:       agentController.serviceImportSyncer.GetLocalFederator(),
 		ResourceType:    &corev1.Service{},
@@ -182,11 +180,51 @@ func (a *Controller) Start(stopCh <-chan struct{}) error {
 		return err
 	}
 
+	a.serviceExportSyncer.Reconcile(func() []runtime.Object {
+		return a.serviceImportLister(func(si *mcsv1a1.ServiceImport) runtime.Object {
+			return &mcsv1a1.ServiceExport{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      si.GetAnnotations()[lhconstants.OriginName],
+					Namespace: si.GetAnnotations()[lhconstants.OriginNamespace],
+				},
+			}
+		})
+	})
+
+	a.serviceSyncer.Reconcile(func() []runtime.Object {
+		return a.serviceImportLister(func(si *mcsv1a1.ServiceImport) runtime.Object {
+			return &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      si.GetAnnotations()[lhconstants.OriginName],
+					Namespace: si.GetAnnotations()[lhconstants.OriginNamespace],
+				},
+			}
+		})
+	})
+
 	klog.Info("Agent controller started")
 
 	return nil
 }
 
+func (a *Controller) serviceImportLister(transform func(si *mcsv1a1.ServiceImport) runtime.Object) []runtime.Object {
+	siList, err := a.serviceImportSyncer.ListLocalResources(&mcsv1a1.ServiceImport{})
+	if err != nil {
+		klog.Errorf("Error listing serviceImports: %v", err)
+		return nil
+	}
+
+	retList := make([]runtime.Object, 0, len(siList))
+
+	for _, obj := range siList {
+		si := obj.(*mcsv1a1.ServiceImport)
+
+		retList = append(retList, transform(si))
+	}
+
+	return retList
+}
+
 func (a *Controller) serviceExportToServiceImport(obj runtime.Object, numRequeues int, op syncer.Operation) (runtime.Object, bool) {
 	svcExport := obj.(*mcsv1a1.ServiceExport)
 


Test patch:
diff --git a/pkg/agent/controller/agent_test.go b/pkg/agent/controller/agent_test.go
index a5301ba..696ea0e 100644
--- a/pkg/agent/controller/agent_test.go
+++ b/pkg/agent/controller/agent_test.go
@@ -200,6 +200,138 @@ var _ = Describe("ServiceImport syncing", func() {
 	})
 })
 
+var _ = Describe("Reconciliation", func() {
+	var t *testDriver
+
+	BeforeEach(func() {
+		t = newTestDiver()
+	})
+
+	JustBeforeEach(func() {
+		t.justBeforeEach()
+		t.createService()
+		t.createServiceExport()
+	})
+
+	AfterEach(func() {
+		t.afterEach()
+	})
+
+	When("a local headless ServiceImport is stale on startup due to a missed ServiceExport delete event", func() {
+		BeforeEach(func() {
+			t.service.Spec.ClusterIP = corev1.ClusterIPNone
+		})
+
+		JustBeforeEach(func() {
+			t.createEndpoints()
+		})
+
+		It("should delete the ServiceImport and EndpointSlice on reconciliation", func() {
+			serviceImport := t.cluster1.awaitServiceImport(t.service, mcsv1a1.Headless, "")
+			endpointSlice := t.cluster1.awaitEndpointSlice(t.endpoints, t.service)
+
+			t.afterEach()
+			t = newTestDiver()
+
+			test.CreateResource(t.cluster1.localServiceImportClient, serviceImport)
+			test.CreateResource(t.cluster1.localEndpointSliceClient, endpointSlice)
+			t.createService()
+			t.cluster1.start(t, *t.syncerConfig)
+
+			t.awaitServiceUnexported()
+			t.awaitNoEndpointSlice(t.cluster1.localEndpointSliceClient)
+		})
+	})
+
+	When("a local ServiceImport is stale on startup due to a missed Service delete event", func() {
+		It("should delete the ServiceImport on reconciliation", func() {
+			serviceImport := t.cluster1.awaitServiceImport(t.service, mcsv1a1.ClusterSetIP, t.service.Spec.ClusterIP)
+
+			t.afterEach()
+			t = newTestDiver()
+
+			test.CreateResource(t.cluster1.localServiceImportClient, serviceImport)
+			t.createServiceExport()
+			t.cluster1.start(t, *t.syncerConfig)
+
+			t.awaitServiceUnexported()
+		})
+	})
+
+	When("a synced local ServiceImport is stale in the broker datastore on startup", func() {
+		It("should delete it from the broker datastore on reconciliation", func() {
+			serviceImport := t.awaitBrokerServiceImport(mcsv1a1.ClusterSetIP, t.service.Spec.ClusterIP)
+
+			t.afterEach()
+			t = newTestDiver()
+
+			test.CreateResource(t.brokerServiceImportClient, serviceImport)
+			t.justBeforeEach()
+
+			t.awaitServiceUnexported()
+		})
+	})
+
+	When("a synced remote ServiceImport is stale in the local datastore on startup", func() {
+		It("should delete it from the local datastore on reconciliation", func() {
+			serviceImport := t.cluster2.awaitServiceImport(t.service, mcsv1a1.ClusterSetIP, t.service.Spec.ClusterIP)
+
+			t.afterEach()
+			t = newTestDiver()
+
+			test.CreateResource(t.cluster2.localServiceImportClient, serviceImport)
+			t.cluster2.start(t, *t.syncerConfig)
+
+			t.awaitServiceUnexported()
+		})
+	})
+
+	When("a synced local EndpointSlice is stale in the broker datastore on startup", func() {
+		BeforeEach(func() {
+			t.service.Spec.ClusterIP = corev1.ClusterIPNone
+		})
+
+		JustBeforeEach(func() {
+			t.createEndpoints()
+		})
+
+		It("should delete it from the broker datastore on reconciliation", func() {
+			endpointSlice := t.awaitBrokerEndpointSlice()
+
+			t.afterEach()
+			t = newTestDiver()
+
+			test.CreateResource(t.brokerEndpointSliceClient, endpointSlice)
+			t.justBeforeEach()
+
+			t.awaitNoEndpointSlice(t.brokerEndpointSliceClient)
+			t.awaitNoEndpointSlice(t.cluster2.localEndpointSliceClient)
+		})
+	})
+
+	When("a synced remote EndpointSlice is stale in the local datastore on startup", func() {
+		BeforeEach(func() {
+			t.service.Spec.ClusterIP = corev1.ClusterIPNone
+		})
+
+		JustBeforeEach(func() {
+			t.createEndpoints()
+		})
+
+		It("should delete it from the local datastore on reconciliation", func() {
+			endpointSlice := t.cluster2.awaitEndpointSlice(t.endpoints, t.service)
+
+			t.afterEach()
+			t = newTestDiver()
+
+			test.CreateResource(t.cluster2.localEndpointSliceClient, endpointSlice)
+			t.cluster2.start(t, *t.syncerConfig)
+
+			t.awaitNoEndpointSlice(t.cluster2.localEndpointSliceClient)
+		})
+	})
+})
+
 var _ = Describe("Globalnet enabled", func() {
 	globalIP := "192.168.10.34"
 	var t *testDriver
@@ -567,8 +699,8 @@ func awaitServiceImport(client dynamic.ResourceInterface, service *corev1.Servic
 	return serviceImport
 }
 
-func (c *cluster) awaitServiceImport(service *corev1.Service, sType mcsv1a1.ServiceImportType, serviceIP string) {
-	awaitServiceImport(c.localServiceImportClient, service, sType, serviceIP)
+func (c *cluster) awaitServiceImport(service *corev1.Service, sType mcsv1a1.ServiceImportType, serviceIP string) *mcsv1a1.ServiceImport {
+	return awaitServiceImport(c.localServiceImportClient, service, sType, serviceIP)
 }
 
 func awaitUpdatedServiceImport(client dynamic.ResourceInterface, service *corev1.Service,
@@ -609,7 +741,7 @@ func (c *cluster) awaitUpdatedServiceImport(service *corev1.Service, serviceIP s
 }
 
 func awaitEndpointSlice(endpointSliceClient, serviceImportClient dynamic.ResourceInterface,
-	endpoints *corev1.Endpoints, service *corev1.Service, namespace string, expectOwnerRef bool) {
+	endpoints *corev1.Endpoints, service *corev1.Service, namespace string, expectOwnerRef bool) *discovery.EndpointSlice {
 	obj := test.AwaitResource(endpointSliceClient, endpoints.Name+"-"+clusterID1)
 
 	endpointSlice := &discovery.EndpointSlice{}
@@ -673,10 +805,12 @@ func awaitEndpointSlice(endpointSliceClient, serviceImportClient dynamic.Resourc
 		Protocol: &protocol,
 		Port:     &port,
 	}))
+
+	return endpointSlice
 }
 
-func (c *cluster) awaitEndpointSlice(endpoints *corev1.Endpoints, service *corev1.Service) {
-	awaitEndpointSlice(c.localEndpointSliceClient, c.localServiceImportClient, endpoints, service,
+func (c *cluster) awaitEndpointSlice(endpoints *corev1.Endpoints, service *corev1.Service) *discovery.EndpointSlice {
+	return awaitEndpointSlice(c.localEndpointSliceClient, c.localServiceImportClient, endpoints, service,
 		service.Namespace, c.agentSpec.ClusterID == clusterID1)
 }
 
@@ -715,8 +849,12 @@ func (c *cluster) awaitUpdatedEndpointSlice(endpoints *corev1.Endpoints, expecte
 	awaitUpdatedEndpointSlice(c.localEndpointSliceClient, endpoints, expectedIPs)
 }
 
-func (t *testDriver) awaitBrokerServiceImport(sType mcsv1a1.ServiceImportType, serviceIP string) {
-	awaitServiceImport(t.brokerServiceImportClient, t.service, sType, serviceIP)
+func (t *testDriver) awaitBrokerServiceImport(sType mcsv1a1.ServiceImportType, serviceIP string) *mcsv1a1.ServiceImport {
+	return awaitServiceImport(t.brokerServiceImportClient, t.service, sType, serviceIP)
+}
+
+func (t *testDriver) awaitBrokerEndpointSlice() *discovery.EndpointSlice {
+	return awaitEndpointSlice(t.brokerEndpointSliceClient, t.brokerServiceImportClient, t.endpoints, t.service, test.RemoteNamespace, false)
 }
 
 func (t *testDriver) awaitUpdatedServiceImport(serviceIP string) {
@@ -725,8 +863,8 @@ func (t *testDriver) awaitUpdatedServiceImport(serviceIP string) {
 	t.cluster2.awaitUpdatedServiceImport(t.service, serviceIP)
 }
 
-func (t *testDriver) awaitEndpointSlice(serviceIPs ...string) {
-	awaitEndpointSlice(t.brokerEndpointSliceClient, t.brokerServiceImportClient, t.endpoints, t.service, test.RemoteNamespace, false)
+func (t *testDriver) awaitEndpointSlice() {
+	t.awaitBrokerEndpointSlice()
 	t.cluster1.awaitEndpointSlice(t.endpoints, t.service)
 	t.cluster2.awaitEndpointSlice(t.endpoints, t.service)
 }