Fix inadvertent deletion of aggregated ServiceImports on agent restart

If the broker co-exists on a managed cluster, on LH agent restart,
the aggregated ServiceImports on the broker are inadvertently deleted
during reconciliation. Reconciliation should only process local
aggregated ServiceImports and should ignore ServiceImports in the
broker namespace. The latter are distinguished by the presence of the
 "multicluster.kubernetes.io/service-name" annotation.

The reconciliation unit test was adjusted to cover this case.

Fixes submariner-io/submariner#3188

Signed-off-by: Tom Pantelis <[email protected]>

pkg/agent/controller/controller_suite_test.go

@@ -325,8 +325,8 @@ func newTestDiver() *testDriver {
 	t.brokerEndpointSliceClient = t.syncerConfig.BrokerClient.Resource(*test.GetGroupVersionResourceFor(t.syncerConfig.RestMapper,
 		&discovery.EndpointSlice{})).Namespace(test.RemoteNamespace)
 
-	t.cluster1.init(t.syncerConfig)
-	t.cluster2.init(t.syncerConfig)
+	t.cluster1.init(t.syncerConfig, nil)
+	t.cluster2.init(t.syncerConfig, nil)
 
 	return t
 }
@@ -340,15 +340,18 @@ func (t *testDriver) afterEach() {
 	close(t.stopCh)
 }
 
-func (c *cluster) init(syncerConfig *broker.SyncerConfig) {
+func (c *cluster) init(syncerConfig *broker.SyncerConfig, dynClient *dynamicfake.FakeDynamicClient) {
 	for k, v := range c.service.Labels {
 		c.serviceEndpointSlices[0].Labels[k] = v
 	}
 
 	c.serviceIP = c.service.Spec.ClusterIP
 
-	c.localDynClient = dynamicfake.NewSimpleDynamicClient(syncerConfig.Scheme)
-	fake.AddBasicReactors(&c.localDynClient.Fake)
+	c.localDynClient = dynClient
+	if c.localDynClient == nil {
+		c.localDynClient = dynamicfake.NewSimpleDynamicClient(syncerConfig.Scheme)
+		fake.AddBasicReactors(&c.localDynClient.Fake)
+	}
 
 	c.localServiceImportReactor = fake.NewFailingReactorForResource(&c.localDynClient.Fake, "serviceimports")
 

pkg/agent/controller/reconciliation_test.go

@@ -117,6 +117,11 @@ var _ = Describe("Reconciliation", func() {
 			t = newTestDiver()
 			t.useClusterSetIP = true
 
+			brokerDynClient := t.syncerConfig.BrokerClient.(*fake.FakeDynamicClient)
+
+			// Use the broker client for cluster1 to simulate the broker being on the same cluster.
+			t.cluster1.init(t.syncerConfig, brokerDynClient)
+
 			test.CreateResource(t.cluster1.localServiceImportClient.Namespace(test.LocalNamespace), localServiceImport)
 			test.CreateResource(t.cluster1.localEndpointSliceClient, localEndpointSlice)
 			test.CreateResource(t.cluster1.localServiceExportClient, serviceExport)
@@ -133,12 +138,6 @@ var _ = Describe("Reconciliation", func() {
 			t.cluster1.start(t, *t.syncerConfig)
 			t.cluster2.start(t, *t.syncerConfig)
 
-			t.awaitNonHeadlessServiceExported(&t.cluster1)
-
-			testutil.EnsureNoActionsForResource(&t.cluster1.localDynClient.Fake, "serviceimports", "delete")
-			testutil.EnsureNoActionsForResource(&t.cluster1.localDynClient.Fake, "endpointslices", "delete")
-
-			brokerDynClient := t.syncerConfig.BrokerClient.(*fake.FakeDynamicClient)
 			testutil.EnsureNoActionsForResource(&brokerDynClient.Fake, "endpointslices", "delete")
 
 			// For migration cleanup, it may attempt to delete a local legacy ServiceImport from the broker so ignore it.
@@ -153,6 +152,8 @@ var _ = Describe("Reconciliation", func() {
 
 				return false
 			}).Should(BeFalse())
+
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
 		})
 	})
 

pkg/agent/controller/service_import.go

@@ -247,8 +247,8 @@ func (c *ServiceImportController) reconcileLocalAggregatedServiceImports() {
 		for i := range siList.Items {
 			si := c.converter.toServiceImport(&siList.Items[i])
 
-			if serviceImportSourceName(si) != "" {
-				// This is not an aggregated ServiceImport.
+			if serviceImportSourceName(si) != "" || si.Annotations[mcsv1a1.LabelServiceName] != "" {
+				// This is not a local aggregated ServiceImport.
 				continue
 			}