Repository: submariner-io/lighthouse
Pull request #682 (2022-02-09 17:24:05)

Clean up EndpointSlices and ServiceImports on uninstall
Delete all from the local datastore and delete local objects from the broker.

Depends on https://github.com/submariner-io/admiral/pull/318
Depends on https://github.com/submariner-io/admiral/pull/319

github-actions[bot]: :hourglass_flowing_sand: Alright! Looks like we need to wait for some *dependencies*:

* submariner-io/admiral#318

: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#318~~
* ~~submariner-io/admiral#319~~

: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 f8fcb9b..39a901e 100644
--- a/go.mod
+++ b/go.mod
@@ -14,7 +14,7 @@ require (
 	github.com/onsi/gomega v1.18.1
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.12.1
-	github.com/submariner-io/admiral v0.12.0-m3
+	github.com/submariner-io/admiral v0.12.0-m3.0.20220211050139-69a40598bdd6
 	github.com/submariner-io/shipyard v0.12.0-m3
 	github.com/uw-labs/lichen v0.1.4
 	k8s.io/api v0.21.0
diff --git a/go.sum b/go.sum
index 3780bdf..16eeea4 100644
--- a/go.sum
+++ b/go.sum
@@ -748,6 +748,8 @@ 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.12.0-m3 h1:ohJK0FmJHzDIy9f/FJGR47ANESbStpPVhdN4UYbvtys=
 github.com/submariner-io/admiral v0.12.0-m3/go.mod h1:UBcJ547DlOWcX13HtWBS83tmd+DcWDu9FDbmI0iDSU4=
+github.com/submariner-io/admiral v0.12.0-m3.0.20220211050139-69a40598bdd6 h1:RMgMytzHKdn4giyMDnu9oFmT40ea0NX75HcttQCoL/U=
+github.com/submariner-io/admiral v0.12.0-m3.0.20220211050139-69a40598bdd6/go.mod h1:hFPcxH0gAr559/qK7ri10Yh1wyeK6yzow1PIqtBk+30=
 github.com/submariner-io/shipyard v0.12.0-m3 h1:kMVYrEPLEH9KyAHp49bmVEVsukfrIWWIqgsqgNII4nc=
 github.com/submariner-io/shipyard v0.12.0-m3/go.mod h1:el/lH6ed/1BN1FvEXlptiib7qT2cTwEQ4AMceH3js9g=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
diff --git a/pkg/agent/controller/cleanup.go b/pkg/agent/controller/cleanup.go
new file mode 100644
index 0000000..857b881
--- /dev/null
+++ b/pkg/agent/controller/cleanup.go
@@ -0,0 +1,90 @@
+/*
+SPDX-License-Identifier: Apache-2.0
+
+Copyright Contributors to the Submariner project.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+	lhconstants "github.com/submariner-io/lighthouse/pkg/constants"
+	discovery "k8s.io/api/discovery/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
+)
+
+var (
+	serviceImportGVR = schema.GroupVersionResource{
+		Group:    mcsv1a1.GroupName,
+		Version:  mcsv1a1.SchemeGroupVersion.Version,
+		Resource: "serviceimports",
+	}
+
+	endpointSliceGVR = schema.GroupVersionResource{
+		Group:    discovery.GroupName,
+		Version:  discovery.SchemeGroupVersion.Version,
+		Resource: "endpointslices",
+	}
+)
+
+func (a *Controller) Cleanup() error {
+	// Delete all ServiceImports from the local cluster skipping those in the broker namespace if the broker is on the
+	// local cluster.
+	err := a.serviceImportSyncer.GetLocalClient().Resource(serviceImportGVR).Namespace(metav1.NamespaceAll).DeleteCollection(context.TODO(),
+		metav1.DeleteOptions{},
+		metav1.ListOptions{
+			FieldSelector: fields.OneTermNotEqualSelector("metadata.namespace", a.serviceImportSyncer.GetBrokerNamespace()).String(),
+		})
+	if err != nil {
+		return errors.Wrap(err, "error deleting local ServiceImports")
+	}
+
+	// Delete all local ServiceImports from the broker.
+	err = a.serviceImportSyncer.GetBrokerClient().Resource(serviceImportGVR).Namespace(a.serviceImportSyncer.GetBrokerNamespace()).
+		DeleteCollection(context.TODO(), metav1.DeleteOptions{},
+			metav1.ListOptions{
+				LabelSelector: labels.Set(map[string]string{lhconstants.LighthouseLabelSourceCluster: a.clusterID}).String(),
+			})
+	if err != nil {
+		return errors.Wrap(err, "error deleting remote EndpointSlices")
+	}
+
+	// Delete all EndpointSlices from the local cluster skipping those in the broker namespace if the broker is on the
+	// local cluster.
+	err = a.endpointSliceSyncer.GetLocalClient().Resource(endpointSliceGVR).Namespace(metav1.NamespaceAll).DeleteCollection(context.TODO(),
+		metav1.DeleteOptions{},
+		metav1.ListOptions{
+			FieldSelector: fields.OneTermNotEqualSelector("metadata.namespace", a.serviceImportSyncer.GetBrokerNamespace()).String(),
+			LabelSelector: labels.Set(map[string]string{discovery.LabelManagedBy: lhconstants.LabelValueManagedBy}).String(),
+		})
+	if err != nil {
+		return errors.Wrap(err, "error deleting local EndpointSlices")
+	}
+
+	// Delete all local EndpointSlices from the broker.
+	err = a.endpointSliceSyncer.GetBrokerClient().Resource(endpointSliceGVR).Namespace(a.endpointSliceSyncer.GetBrokerNamespace()).
+		DeleteCollection(context.TODO(), metav1.DeleteOptions{},
+			metav1.ListOptions{
+				LabelSelector: labels.Set(map[string]string{lhconstants.MCSLabelSourceCluster: a.clusterID}).String(),
+			})
+
+	return errors.Wrap(err, "error deleting remote EndpointSlices")
+}
diff --git a/pkg/agent/controller/types.go b/pkg/agent/controller/types.go
index 3fc7088..7c612fa 100644
--- a/pkg/agent/controller/types.go
+++ b/pkg/agent/controller/types.go
@@ -47,6 +47,7 @@ type AgentSpecification struct {
 	ClusterID        string
 	Namespace        string
 	GlobalnetEnabled bool `split_words:"true"`
+	Uninstall        bool
 }
 
 // The ServiceImportController listens for ServiceImport resources created in the target namespace
diff --git a/pkg/agent/main.go b/pkg/agent/main.go
index 7a017dc..2df3e46 100644
--- a/pkg/agent/main.go
+++ b/pkg/agent/main.go
@@ -100,8 +100,6 @@ func main() {
 	// set up signals so we handle the first shutdown signal gracefully
 	ctx := signals.SetupSignalHandler()
 
-	httpServer := startHTTPServer()
-
 	lightHouseAgent, err := controller.New(&agentSpec, broker.SyncerConfig{
 		LocalRestConfig: cfg,
 		LocalClient:     localClient,
@@ -116,10 +114,23 @@ func main() {
 		klog.Fatalf("Failed to create lighthouse agent: %v", err)
 	}
 
+	if agentSpec.Uninstall {
+		klog.Info("Uninstalling lighthouse")
+
+		err := lightHouseAgent.Cleanup()
+		if err != nil {
+			klog.Fatalf("Error cleaning up the lighthouse agent controller: %+v", err)
+		}
+
+		return
+	}
+
 	if err := lightHouseAgent.Start(ctx.Done()); err != nil {
 		klog.Fatalf("Failed to start lighthouse agent: %v", err)
 	}
 
+	httpServer := startHTTPServer()
+
 	<-ctx.Done()
 
 	klog.Info("All controllers stopped or exited. Stopping main loop")


Test patch:
diff --git a/pkg/agent/controller/cleanup_test.go b/pkg/agent/controller/cleanup_test.go
new file mode 100644
index 0000000..a369a0f
--- /dev/null
+++ b/pkg/agent/controller/cleanup_test.go
@@ -0,0 +1,172 @@
+/*
+SPDX-License-Identifier: Apache-2.0
+
+Copyright Contributors to the Submariner project.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller_test
+
+import (
+	"time"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"github.com/submariner-io/admiral/pkg/syncer/test"
+	lhconstants "github.com/submariner-io/lighthouse/pkg/constants"
+	discovery "k8s.io/api/discovery/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/dynamic"
+	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
+)
+
+var _ = Describe("Cleanup", func() {
+	var (
+		t                                    *testDriver
+		existingLocalServiceImport           *mcsv1a1.ServiceImport
+		existingRemoteServiceImport          *mcsv1a1.ServiceImport
+		existingLocalServiceImportInRemoteNS *mcsv1a1.ServiceImport
+		existingLocalEndpointSlice           *discovery.EndpointSlice
+		existingLocalEndpointSliceInRemoteNS *discovery.EndpointSlice
+		existingRemoteEndpointSlice          *discovery.EndpointSlice
+		nonLHEndpointSlice                   *discovery.EndpointSlice
+		remoteNSServiceImportClient          dynamic.ResourceInterface
+		remoteNSEndpointSliceClient          dynamic.ResourceInterface
+	)
+
+	BeforeEach(func() {
+		t = newTestDiver()
+		t.doStart = false
+	})
+
+	JustBeforeEach(func() {
+		t.justBeforeEach()
+
+		existingLocalServiceImport = &mcsv1a1.ServiceImport{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "nginx-" + serviceNamespace + "-" + clusterID1,
+				Labels: map[string]string{
+					lhconstants.LighthouseLabelSourceCluster: clusterID1,
+				},
+			},
+		}
+
+		test.CreateResource(t.cluster1.localServiceImportClient, existingLocalServiceImport)
+		test.CreateResource(t.brokerServiceImportClient, test.SetClusterIDLabel(existingLocalServiceImport, clusterID1))
+
+		remoteNSServiceImportClient = t.cluster1.localDynClient.Resource(*test.GetGroupVersionResourceFor(t.syncerConfig.RestMapper,
+			&mcsv1a1.ServiceImport{})).Namespace(test.RemoteNamespace)
+
+		existingLocalServiceImportInRemoteNS = &mcsv1a1.ServiceImport{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "other-" + serviceNamespace + "-" + clusterID2,
+				Labels: map[string]string{
+					lhconstants.LighthouseLabelSourceCluster: clusterID2,
+				},
+			},
+		}
+
+		test.CreateResource(remoteNSServiceImportClient, test.SetClusterIDLabel(existingLocalServiceImportInRemoteNS, clusterID2))
+
+		existingRemoteServiceImport = &mcsv1a1.ServiceImport{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "nginx2-" + serviceNamespace + "-" + clusterID2,
+				Labels: map[string]string{
+					lhconstants.LighthouseLabelSourceCluster: clusterID2,
+				},
+			},
+		}
+
+		test.CreateResource(t.cluster1.localServiceImportClient, test.SetClusterIDLabel(existingRemoteServiceImport, clusterID2))
+		test.CreateResource(t.brokerServiceImportClient, test.SetClusterIDLabel(existingRemoteServiceImport, clusterID2))
+
+		existingLocalEndpointSlice = &discovery.EndpointSlice{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "nginx-" + clusterID1,
+				Labels: map[string]string{
+					lhconstants.MCSLabelSourceCluster: clusterID1,
+					discovery.LabelManagedBy:          lhconstants.LabelValueManagedBy,
+				},
+			},
+		}
+
+		test.CreateResource(t.cluster1.localEndpointSliceClient, existingLocalEndpointSlice)
+		test.CreateResource(t.brokerEndpointSliceClient, test.SetClusterIDLabel(existingLocalEndpointSlice, clusterID1))
+
+		remoteNSEndpointSliceClient = t.cluster1.localDynClient.Resource(*test.GetGroupVersionResourceFor(t.syncerConfig.RestMapper,
+			&discovery.EndpointSlice{})).Namespace(test.RemoteNamespace)
+
+		existingLocalEndpointSliceInRemoteNS = &discovery.EndpointSlice{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "other-" + clusterID2,
+				Labels: map[string]string{
+					lhconstants.MCSLabelSourceCluster: clusterID2,
+					discovery.LabelManagedBy:          lhconstants.LabelValueManagedBy,
+				},
+			},
+		}
+
+		test.CreateResource(remoteNSEndpointSliceClient, test.SetClusterIDLabel(existingLocalEndpointSliceInRemoteNS, clusterID2))
+
+		existingRemoteEndpointSlice = &discovery.EndpointSlice{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "nginx2-" + clusterID2,
+				Labels: map[string]string{
+					lhconstants.MCSLabelSourceCluster: clusterID2,
+					discovery.LabelManagedBy:          lhconstants.LabelValueManagedBy,
+				},
+			},
+		}
+
+		test.CreateResource(t.cluster1.localEndpointSliceClient, test.SetClusterIDLabel(existingRemoteEndpointSlice, clusterID2))
+		test.CreateResource(t.brokerEndpointSliceClient, test.SetClusterIDLabel(existingRemoteEndpointSlice, clusterID2))
+
+		nonLHEndpointSlice = &discovery.EndpointSlice{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "other",
+			},
+		}
+
+		test.CreateResource(t.cluster1.localEndpointSliceClient, nonLHEndpointSlice)
+	})
+
+	AfterEach(func() {
+		t.afterEach()
+	})
+
+	It("should remove local LH ServiceImports and EndpointSlices from the remote datastore", func() {
+		Expect(t.cluster1.agentController.Cleanup()).To(Succeed())
+
+		test.AwaitNoResource(t.brokerServiceImportClient, existingLocalServiceImport.GetName())
+		test.AwaitNoResource(t.brokerEndpointSliceClient, existingLocalEndpointSlice.GetName())
+
+		time.Sleep(300 * time.Millisecond)
+		test.AwaitResource(t.brokerServiceImportClient, existingRemoteServiceImport.GetName())
+		test.AwaitResource(t.brokerEndpointSliceClient, existingRemoteEndpointSlice.GetName())
+	})
+
+	It("should remove all LH ServiceImports and EndpointSlices from the local datastore", func() {
+		Expect(t.cluster1.agentController.Cleanup()).To(Succeed())
+
+		test.AwaitNoResource(t.cluster1.localServiceImportClient, existingLocalServiceImport.GetName())
+		test.AwaitNoResource(t.cluster1.localServiceImportClient, existingRemoteServiceImport.GetName())
+		test.AwaitNoResource(t.cluster1.localEndpointSliceClient, existingLocalEndpointSlice.GetName())
+		test.AwaitNoResource(t.cluster1.localEndpointSliceClient, existingRemoteEndpointSlice.GetName())
+
+		time.Sleep(300 * time.Millisecond)
+		test.AwaitResource(t.cluster1.localEndpointSliceClient, nonLHEndpointSlice.GetName())
+		test.AwaitResource(remoteNSServiceImportClient, existingLocalServiceImportInRemoteNS.GetName())
+		test.AwaitResource(remoteNSEndpointSliceClient, existingLocalEndpointSliceInRemoteNS.GetName())
+	})
+})
diff --git a/pkg/agent/controller/controller_suite_test.go b/pkg/agent/controller/controller_suite_test.go
index cb199ab..f16c8fa 100644
--- a/pkg/agent/controller/controller_suite_test.go
+++ b/pkg/agent/controller/controller_suite_test.go
@@ -91,6 +91,7 @@ type cluster struct {
 	localEndpointSliceClient dynamic.ResourceInterface
 	localKubeClient          kubernetes.Interface
 	endpointsReactor         *fake.FailingReactor
+	agentController          *controller.Controller
 }
 
 type testDriver struct {
@@ -104,6 +105,7 @@ type testDriver struct {
 	stopCh                    chan struct{}
 	syncerConfig              *broker.SyncerConfig
 	endpointGlobalIPs         []string
+	doStart                   bool
 }
 
 func newTestDiver() *testDriver {
@@ -144,7 +146,8 @@ func newTestDiver() *testDriver {
 			BrokerClient: fake.NewDynamicClient(syncerScheme),
 			Scheme:       syncerScheme,
 		},
-		stopCh: make(chan struct{}),
+		stopCh:  make(chan struct{}),
+		doStart: true,
 	}
 
 	t.serviceExport = &mcsv1a1.ServiceExport{
@@ -277,14 +280,17 @@ func (c *cluster) start(t *testDriver, syncerConfig broker.SyncerConfig) {
 
 	serviceExportCounterName := "submariner_service_export" + bigint.String()
 
-	agentController, err := controller.New(&c.agentSpec, syncerConfig, c.localKubeClient,
+	c.agentController, err = controller.New(&c.agentSpec, syncerConfig, c.localKubeClient,
 		controller.AgentConfig{
 			ServiceImportCounterName: serviceImportCounterName,
 			ServiceExportCounterName: serviceExportCounterName,
 		})
 
 	Expect(err).To(Succeed())
-	Expect(agentController.Start(t.stopCh)).To(Succeed())
+
+	if t.doStart {
+		Expect(c.agentController.Start(t.stopCh)).To(Succeed())
+	}
 }
 
 func awaitServiceImport(client dynamic.ResourceInterface, service *corev1.Service, sType mcsv1a1.ServiceImportType,
@@ -457,6 +463,10 @@ func (c *cluster) awaitUpdatedEndpointSlice(endpoints *corev1.Endpoints, expecte
 	awaitUpdatedEndpointSlice(c.localEndpointSliceClient, endpoints, expectedIPs)
 }
 
+func (c *cluster) dynamicServiceClient() dynamic.NamespaceableResourceInterface {
+	return c.localDynClient.Resource(schema.GroupVersionResource{Version: "v1", Resource: "services"})
+}
+
 func (t *testDriver) awaitBrokerServiceImport(sType mcsv1a1.ServiceImportType, serviceIP string) *mcsv1a1.ServiceImport {
 	return awaitServiceImport(t.brokerServiceImportClient, t.service, sType, serviceIP)
 }
@@ -487,7 +497,7 @@ func (t *testDriver) createService() {
 	_, err := t.cluster1.localKubeClient.CoreV1().Services(t.service.Namespace).Create(context.TODO(), t.service, metav1.CreateOptions{})
 	Expect(err).To(Succeed())
 
-	test.CreateResource(t.dynamicServiceClient(), t.service)
+	test.CreateResource(t.cluster1.dynamicServiceClient().Namespace(t.service.Namespace), t.service)
 }
 
 func (t *testDriver) createEndpoints() {
@@ -517,7 +527,8 @@ func (t *testDriver) deleteServiceExport() {
 }
 
 func (t *testDriver) deleteService() {
-	Expect(t.dynamicServiceClient().Delete(context.TODO(), t.service.Name, metav1.DeleteOptions{})).To(Succeed())
+	Expect(t.cluster1.dynamicServiceClient().Namespace(t.service.Namespace).Delete(context.TODO(), t.service.Name,
+		metav1.DeleteOptions{})).To(Succeed())
 
 	Expect(t.cluster1.localKubeClient.CoreV1().Services(t.service.Namespace).Delete(context.TODO(), t.service.Name,
 		metav1.DeleteOptions{})).To(Succeed())
@@ -534,10 +545,6 @@ func (t *testDriver) createEndpointIngressIPs() {
 	t.createGlobalIngressIP(t.newHeadlessGlobalIngressIP("not-ready", globalIP3))
 }
 
-func (t *testDriver) dynamicServiceClient() dynamic.ResourceInterface {
-	return t.cluster1.localDynClient.Resource(schema.GroupVersionResource{Version: "v1", Resource: "services"}).Namespace(t.service.Namespace)
-}
-
 func (t *testDriver) awaitNoServiceImport(client dynamic.ResourceInterface) {
 	test.AwaitNoResource(client, t.service.Name+"-"+t.service.Namespace+"-"+clusterID1)
 }