Repository: submariner-io/lighthouse
Pull request #964 (2022-11-17 16:48:51)

Include namespace in EndpointSlice name
Fixes https://github.com/submariner-io/lighthouse/issues/957

Also remove prior `EndpointSlices` on migration.

Depends on https://github.com/submariner-io/subctl/pull/394


submariner-bot: 🤖 Created branch: z_pr964/tpantelis/fix_epslice_name 
🚀 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.
github-actions[bot]: This PR/issue depends on:
* submariner-io/subctl#394
By **[Dependent Issues](https://github.com/z0al/dependent-issues)** (🤖). Happy coding!
<!-- By Dependent Issues (Action) - DO NOT REMOVE -->
github-actions[bot]: This PR/issue depends on:
* ~~submariner-io/subctl#394~~
By **[Dependent Issues](https://github.com/z0al/dependent-issues)** (🤖). Happy coding!
<!-- By Dependent Issues (Action) - DO NOT REMOVE -->

Patch:
diff --git a/pkg/agent/controller/agent.go b/pkg/agent/controller/agent.go
index eeb5b58..66ac82a 100644
--- a/pkg/agent/controller/agent.go
+++ b/pkg/agent/controller/agent.go
@@ -558,6 +558,19 @@ func (a *Controller) filterLocalEndpointSlices(obj runtime.Object, numRequeues i
 		return nil, false
 	}
 
+	oldName := labels[constants.MCSLabelServiceName] + "-" + labels[constants.MCSLabelSourceCluster]
+	if op != syncer.Delete && endpointSlice.Name == oldName {
+		logger.Infof("EndpointSlice %s/%s has the old naming convention sans namespace - deleting it",
+			endpointSlice.Namespace, endpointSlice.Name)
+
+		err := a.endpointSliceSyncer.GetLocalFederator().Delete(endpointSlice)
+		if err != nil {
+			logger.Errorf(err, "Error deleting local EndpointSlice %s/%s", endpointSlice.Namespace, endpointSlice.Name)
+		}
+
+		return nil, false
+	}
+
 	return obj, false
 }
 
diff --git a/pkg/agent/controller/endpoint.go b/pkg/agent/controller/endpoint.go
index ebf8296..1b01149 100644
--- a/pkg/agent/controller/endpoint.go
+++ b/pkg/agent/controller/endpoint.go
@@ -131,14 +131,12 @@ func (e *EndpointController) cleanup() {
 func (e *EndpointController) endpointsToEndpointSlice(obj runtime.Object, numRequeues int, op syncer.Operation) (runtime.Object, bool) {
 	endPoints := obj.(*corev1.Endpoints)
 
-	endpointSliceName := endPoints.Name + "-" + e.clusterID
-
 	if op == syncer.Delete {
 		logger.V(log.DEBUG).Infof("Endpoints %s/%s deleted", endPoints.Namespace, endPoints.Name)
 
 		return &discovery.EndpointSlice{
 			ObjectMeta: metav1.ObjectMeta{
-				Name:      endpointSliceName,
+				Name:      e.endpointSliceNameFrom(endPoints),
 				Namespace: endPoints.Namespace,
 			},
 		}, false
@@ -158,7 +156,7 @@ func (e *EndpointController) endpointSliceFromEndpoints(endpoints *corev1.Endpoi
 ) {
 	endpointSlice := &discovery.EndpointSlice{}
 
-	endpointSlice.Name = endpoints.Name + "-" + e.clusterID
+	endpointSlice.Name = e.endpointSliceNameFrom(endpoints)
 	endpointSlice.Labels = map[string]string{
 		discovery.LabelManagedBy:        constants.LabelValueManagedBy,
 		constants.LabelSourceNamespace:  e.serviceImportSourceNameSpace,
@@ -207,6 +205,10 @@ func (e *EndpointController) endpointSliceFromEndpoints(endpoints *corev1.Endpoi
 	return endpointSlice, false
 }
 
+func (e *EndpointController) endpointSliceNameFrom(endpoints *corev1.Endpoints) string {
+	return endpoints.Name + "-" + endpoints.Namespace + "-" + e.clusterID
+}
+
 func (e *EndpointController) getEndpointsFromAddresses(addresses []corev1.EndpointAddress, addressType discovery.AddressType,
 	ready bool,
 ) ([]discovery.Endpoint, bool) {


Test patch:
diff --git a/pkg/agent/controller/controller_suite_test.go b/pkg/agent/controller/controller_suite_test.go
index 63520d8..c6be734 100644
--- a/pkg/agent/controller/controller_suite_test.go
+++ b/pkg/agent/controller/controller_suite_test.go
@@ -361,16 +361,33 @@ func (c *cluster) start(t *testDriver, syncerConfig broker.SyncerConfig) {
 	}
 }
 
+func findServiceImport(client dynamic.ResourceInterface, namespace, name string) *mcsv1a1.ServiceImport {
+	list, err := client.List(context.TODO(), metav1.ListOptions{})
+	Expect(err).To(Succeed())
+
+	for i := range list.Items {
+		if list.Items[i].GetAnnotations()[constants.OriginName] == name &&
+			list.Items[i].GetAnnotations()[constants.OriginNamespace] == namespace {
+			serviceImport := &mcsv1a1.ServiceImport{}
+			Expect(scheme.Scheme.Convert(&list.Items[i], serviceImport, nil)).To(Succeed())
+
+			return serviceImport
+		}
+	}
+
+	return nil
+}
+
 func awaitServiceImport(client dynamic.ResourceInterface, service *corev1.Service, sType mcsv1a1.ServiceImportType,
 	serviceIP string,
 ) *mcsv1a1.ServiceImport {
-	obj := test.AwaitResource(client, service.Name+"-"+service.Namespace+"-"+clusterID1)
+	var serviceImport *mcsv1a1.ServiceImport
 
-	serviceImport := &mcsv1a1.ServiceImport{}
-	Expect(scheme.Scheme.Convert(obj, serviceImport, nil)).To(Succeed())
+	Eventually(func() *mcsv1a1.ServiceImport {
+		serviceImport = findServiceImport(client, service.Namespace, service.Name)
+		return serviceImport
+	}, 5*time.Second).ShouldNot(BeNil(), "ServiceImport not found")
 
-	Expect(serviceImport.GetAnnotations()["origin-name"]).To(Equal(service.Name))
-	Expect(serviceImport.GetAnnotations()["origin-namespace"]).To(Equal(service.Namespace))
 	Expect(serviceImport.Spec.Type).To(Equal(sType))
 
 	Expect(serviceImport.Status.Clusters).To(HaveLen(1))
@@ -403,16 +420,11 @@ func (c *cluster) awaitServiceImport(service *corev1.Service, sType mcsv1a1.Serv
 }
 
 func awaitUpdatedServiceImport(client dynamic.ResourceInterface, service *corev1.Service, serviceIP string) {
-	name := service.Name + "-" + service.Namespace + "-" + clusterID1
-
 	var serviceImport *mcsv1a1.ServiceImport
 
 	err := wait.PollImmediate(50*time.Millisecond, 5*time.Second, func() (bool, error) {
-		obj, err := client.Get(context.TODO(), name, metav1.GetOptions{})
-		Expect(err).To(Succeed())
-
-		serviceImport = &mcsv1a1.ServiceImport{}
-		Expect(scheme.Scheme.Convert(obj, serviceImport, nil)).To(Succeed())
+		serviceImport = findServiceImport(client, service.Namespace, service.Name)
+		Expect(serviceImport).ToNot(BeNil())
 
 		if serviceIP == "" {
 			return len(serviceImport.Spec.IPs) == 0, nil
@@ -436,20 +448,44 @@ func (c *cluster) awaitUpdatedServiceImport(service *corev1.Service, serviceIP s
 	awaitUpdatedServiceImport(c.localServiceImportClient, service, serviceIP)
 }
 
-func awaitEndpointSlice(endpointSliceClient dynamic.ResourceInterface, endpoints *corev1.Endpoints,
-	service *corev1.Service, namespace string, globalIPs []string,
+func findEndpointSlice(client dynamic.ResourceInterface, namespace, name string) *discovery.EndpointSlice {
+	list, err := client.List(context.TODO(), metav1.ListOptions{})
+	Expect(err).To(Succeed())
+
+	for i := range list.Items {
+		if list.Items[i].GetLabels()[constants.MCSLabelServiceName] == name &&
+			list.Items[i].GetLabels()[constants.LabelSourceNamespace] == namespace {
+			endpointSlice := &discovery.EndpointSlice{}
+			Expect(scheme.Scheme.Convert(&list.Items[i], endpointSlice, nil)).To(Succeed())
+
+			return endpointSlice
+		}
+	}
+
+	return nil
+}
+
+func awaitEndpointSlice(client dynamic.ResourceInterface, namespace, name string) *discovery.EndpointSlice {
+	var endpointSlice *discovery.EndpointSlice
+
+	Eventually(func() *discovery.EndpointSlice {
+		endpointSlice = findEndpointSlice(client, namespace, name)
+		return endpointSlice
+	}, 5*time.Second).ShouldNot(BeNil(), "EndpointSlice not found")
+
+	return endpointSlice
+}
+
+func awaitAndVerifyEndpointSlice(endpointSliceClient dynamic.ResourceInterface, endpoints *corev1.Endpoints,
+	namespace string, globalIPs []string,
 ) *discovery.EndpointSlice {
-	obj := test.AwaitResource(endpointSliceClient, endpoints.Name+"-"+clusterID1)
+	endpointSlice := awaitEndpointSlice(endpointSliceClient, endpoints.Namespace, endpoints.Name)
 
-	endpointSlice := &discovery.EndpointSlice{}
-	Expect(scheme.Scheme.Convert(obj, endpointSlice, nil)).To(Succeed())
 	Expect(endpointSlice.Namespace).To(Equal(namespace))
 
 	labels := endpointSlice.GetLabels()
 	Expect(labels).To(HaveKeyWithValue(discovery.LabelManagedBy, constants.LabelValueManagedBy))
-	Expect(labels).To(HaveKeyWithValue(constants.LabelSourceNamespace, service.Namespace))
 	Expect(labels).To(HaveKeyWithValue(constants.MCSLabelSourceCluster, clusterID1))
-	Expect(labels).To(HaveKeyWithValue(constants.MCSLabelServiceName, service.Name))
 
 	Expect(endpointSlice.AddressType).To(Equal(discovery.AddressTypeIPv4))
 
@@ -495,22 +531,17 @@ func awaitEndpointSlice(endpointSliceClient dynamic.ResourceInterface, endpoints
 }
 
 func (c *cluster) awaitEndpointSlice(t *testDriver) *discovery.EndpointSlice {
-	return awaitEndpointSlice(c.localEndpointSliceClient, t.endpoints, t.service, t.service.Namespace, t.endpointGlobalIPs)
+	return awaitAndVerifyEndpointSlice(c.localEndpointSliceClient, t.endpoints, t.service.Namespace, t.endpointGlobalIPs)
 }
 
 func awaitUpdatedEndpointSlice(endpointSliceClient dynamic.ResourceInterface, endpoints *corev1.Endpoints, expectedIPs []string) {
-	name := endpoints.Name + "-" + clusterID1
-
 	sort.Strings(expectedIPs)
 
 	var actualIPs []string
 
 	err := wait.PollImmediate(50*time.Millisecond, 5*time.Second, func() (bool, error) {
-		obj, err := endpointSliceClient.Get(context.TODO(), name, metav1.GetOptions{})
-		Expect(err).To(Succeed())
-
-		endpointSlice := &discovery.EndpointSlice{}
-		Expect(scheme.Scheme.Convert(obj, endpointSlice, nil)).To(Succeed())
+		endpointSlice := findEndpointSlice(endpointSliceClient, endpoints.Namespace, endpoints.Name)
+		Expect(endpointSlice).ToNot(BeNil())
 
 		actualIPs = nil
 		for _, ep := range endpointSlice.Endpoints {
@@ -542,7 +573,7 @@ func (t *testDriver) awaitBrokerServiceImport(sType mcsv1a1.ServiceImportType, s
 }
 
 func (t *testDriver) awaitBrokerEndpointSlice() *discovery.EndpointSlice {
-	return awaitEndpointSlice(t.brokerEndpointSliceClient, t.endpoints, t.service, test.RemoteNamespace, t.endpointGlobalIPs)
+	return awaitAndVerifyEndpointSlice(t.brokerEndpointSliceClient, t.endpoints, test.RemoteNamespace, t.endpointGlobalIPs)
 }
 
 func (t *testDriver) awaitUpdatedServiceImport(serviceIP string) {
@@ -592,7 +623,23 @@ func (t *testDriver) updateEndpoints() {
 }
 
 func (t *testDriver) dynamicEndpointsClient() dynamic.ResourceInterface {
-	return t.cluster1.localDynClient.Resource(schema.GroupVersionResource{Version: "v1", Resource: "endpoints"}).Namespace(t.service.Namespace)
+	return dynamicEndpointsClient(t.cluster1.localDynClient, t.service.Namespace)
+}
+
+func dynamicEndpointsClient(client dynamic.Interface, namespace string) dynamic.ResourceInterface {
+	return client.Resource(corev1.SchemeGroupVersion.WithResource("endpoints")).Namespace(namespace)
+}
+
+func serviceExportClient(client dynamic.Interface, namespace string) dynamic.ResourceInterface {
+	return client.Resource(schema.GroupVersionResource{
+		Group:    mcsv1a1.GroupVersion.Group,
+		Version:  mcsv1a1.GroupVersion.Version,
+		Resource: "serviceexports",
+	}).Namespace(namespace)
+}
+
+func endpointSliceClient(client dynamic.Interface, namespace string) dynamic.ResourceInterface {
+	return client.Resource(discovery.SchemeGroupVersion.WithResource("endpointslices")).Namespace(namespace)
 }
 
 func (t *testDriver) createServiceExport() {
diff --git a/pkg/agent/controller/reconciliation_test.go b/pkg/agent/controller/reconciliation_test.go
index 9aa8600..f65d7d4 100644
--- a/pkg/agent/controller/reconciliation_test.go
+++ b/pkg/agent/controller/reconciliation_test.go
@@ -21,7 +21,10 @@ package controller_test
 import (
 	. "github.com/onsi/ginkgo"
 	"github.com/submariner-io/admiral/pkg/syncer/test"
+	"github.com/submariner-io/lighthouse/pkg/constants"
 	corev1 "k8s.io/api/core/v1"
+	discovery "k8s.io/api/discovery/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
@@ -156,4 +159,32 @@ var _ = Describe("Reconciliation", func() {
 			t.awaitNoEndpointSlice(t.cluster2.localEndpointSliceClient)
 		})
 	})
+
+	When("a local EndpointSlice with the old naming convention sans namespace exists on startup", func() {
+		epsName := "nginx-" + clusterID1
+
+		JustBeforeEach(func() {
+			eps := &discovery.EndpointSlice{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      epsName,
+					Namespace: serviceNamespace,
+					Labels: map[string]string{
+						discovery.LabelManagedBy:        constants.LabelValueManagedBy,
+						constants.MCSLabelSourceCluster: clusterID1,
+						constants.MCSLabelServiceName:   "nginx",
+					},
+				},
+			}
+
+			test.CreateResource(t.cluster1.localEndpointSliceClient, eps)
+
+			eps.Namespace = test.RemoteNamespace
+			test.CreateResource(t.brokerEndpointSliceClient, test.SetClusterIDLabel(eps, clusterID1))
+		})
+
+		It("should delete it", func() {
+			test.AwaitNoResource(t.cluster1.localEndpointSliceClient, epsName)
+			test.AwaitNoResource(t.brokerEndpointSliceClient, epsName)
+		})
+	})
 })
diff --git a/pkg/agent/controller/service_import_sync_test.go b/pkg/agent/controller/service_import_sync_test.go
index 5b55b1c..b1b661f 100644
--- a/pkg/agent/controller/service_import_sync_test.go
+++ b/pkg/agent/controller/service_import_sync_test.go
@@ -20,8 +20,11 @@ package controller_test
 
 import (
 	. "github.com/onsi/ginkgo"
+	"github.com/submariner-io/admiral/pkg/syncer/test"
 	"github.com/submariner-io/lighthouse/pkg/constants"
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
 var _ = Describe("ServiceImport syncing", func() {
@@ -168,4 +171,55 @@ var _ = Describe("ServiceImport syncing", func() {
 			t.awaitServiceExported(t.service.Spec.ClusterIP)
 		})
 	})
+
+	When("two Services with the same name in different namespaces are exported", func() {
+		It("should sync ServiceImport and EndpointSlice resources for each", func() {
+			t.createEndpoints()
+			t.createService()
+			t.createServiceExport()
+			t.awaitServiceExported(t.service.Spec.ClusterIP)
+			t.awaitEndpointSlice()
+
+			service := &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      t.service.Name,
+					Namespace: "other-service-ns",
+				},
+				Spec: corev1.ServiceSpec{
+					ClusterIP: "10.253.9.2",
+				},
+			}
+
+			serviceExport := &mcsv1a1.ServiceExport{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      service.Name,
+					Namespace: service.Namespace,
+				},
+			}
+
+			endpoints := &corev1.Endpoints{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      service.Name,
+					Namespace: service.Namespace,
+				},
+			}
+
+			test.CreateResource(dynamicEndpointsClient(t.cluster1.localDynClient, endpoints.Namespace), endpoints)
+			test.CreateResource(t.cluster1.dynamicServiceClient().Namespace(service.Namespace), service)
+			test.CreateResource(serviceExportClient(t.cluster1.localDynClient, service.Namespace), serviceExport)
+
+			t.cluster1.awaitServiceImport(service, mcsv1a1.ClusterSetIP, service.Spec.ClusterIP)
+			awaitServiceImport(t.brokerServiceImportClient, service, mcsv1a1.ClusterSetIP, service.Spec.ClusterIP)
+			t.cluster2.awaitServiceImport(service, mcsv1a1.ClusterSetIP, service.Spec.ClusterIP)
+
+			awaitEndpointSlice(endpointSliceClient(t.cluster1.localDynClient, service.Namespace), endpoints.Namespace, endpoints.Name)
+			awaitEndpointSlice(t.brokerEndpointSliceClient, endpoints.Namespace, endpoints.Name)
+			awaitEndpointSlice(endpointSliceClient(t.cluster2.localDynClient, service.Namespace), endpoints.Namespace, endpoints.Name)
+
+			// Ensure the resources for the first Service weren't overwritten
+			t.cluster1.awaitServiceImport(t.service, mcsv1a1.ClusterSetIP, t.service.Spec.ClusterIP)
+			t.awaitBrokerServiceImport(mcsv1a1.ClusterSetIP, t.service.Spec.ClusterIP)
+			t.awaitEndpointSlice()
+		})
+	})
 })