Handle multicluster cleanup (#58093)

* Handle multicluster cleanup

Signed-off-by: Keith Mattix II <[email protected]>

* More cleanup to stop leak

Signed-off-by: Keith Mattix II <[email protected]>

* Address PR comments

Signed-off-by: Keith Mattix II <[email protected]>

---------

Signed-off-by: Keith Mattix II <[email protected]>

Author: keithmattix

pilot/pkg/config/kube/crdclient/client.go

@@ -176,6 +176,12 @@ func (cl *Client) Run(stop <-chan struct{}) {
 	}
 	<-stop
 	close(cl.stop)
+	// Cleanup handlers
+	for _, h := range cl.allKinds() {
+		for _, reg := range h.handlers {
+			reg.UnregisterHandler()
+		}
+	}
 	cl.logger.Infof("controller terminated")
 }
 

pilot/pkg/serviceregistry/kube/controller/controller.go

@@ -408,6 +408,9 @@ func (c *Controller) Cleanup() error {
 		c.opts.MeshNetworksWatcher.DeleteNetworksHandler(c.networksHandlerRegistration)
 	}
 
+	// Shutdown all the informer handlers
+	c.shutdownInformerHandlers()
+
 	return nil
 }
 
@@ -649,6 +652,14 @@ func (c *Controller) HasSynced() bool {
 	return c.queue.HasSynced()
 }
 
+func (c *Controller) shutdownInformerHandlers() {
+	c.namespaces.ShutdownHandlers()
+	c.services.ShutdownHandlers()
+	c.endpoints.slices.ShutdownHandlers()
+	c.pods.pods.ShutdownHandlers()
+	c.nodes.ShutdownHandlers()
+}
+
 func (c *Controller) informersSynced() bool {
 	return c.namespaces.HasSynced() &&
 		c.services.HasSynced() &&

pkg/kube/kclient/crdwatcher.go

@@ -123,6 +123,7 @@ func (c *crdWatcher) Run(stop <-chan struct{}) {
 	c.mutex.Unlock()
 	kube.WaitForCacheSync("crd watcher", stop, c.crds.HasSynced)
 	c.queue.Run(stop)
+	log.Info("Stopping CRD watcher")
 	c.crds.ShutdownHandlers()
 }
 

pkg/kube/multicluster/cluster.go

@@ -79,6 +79,11 @@ func (c *Cluster) Run(mesh mesh.Watcher, handlers []handler, action ACTION) {
 	// Build a namespace watcher. This must have no filter, since this is our input to the filter itself.
 	// This must be done before we build components, so they can access the filter.
 	namespaces := kclient.New[*corev1.Namespace](c.Client)
+	// When this cluster stops, clean up the namespace watcher
+	go func() {
+		<-c.stop
+		namespaces.ShutdownHandlers()
+	}()
 	// This will start a namespace informer and wait for it to be ready. So we must start it in a go routine to avoid blocking.
 	filter := filter.NewDiscoveryNamespacesFilter(namespaces, mesh, c.stop)
 	kube.SetObjectFilter(c.Client, filter)

pkg/kube/multicluster/secretcontroller.go

@@ -290,8 +290,7 @@ func (c *Controller) addSecret(name types.NamespacedName, s *corev1.Secret) erro
 			}
 			// stop previous remote cluster
 			prev.Stop()
-			// TODO(keithmattix): Is it safe to shutdown the kubeclient here as well? If we don't
-			// the goroutine will continue to run and the client will leak.
+			prev.Client.Shutdown() // Shutdown all of the informers so that the goroutines won't leak
 		} else if c.cs.Contains(cluster.ID(clusterID)) {
 			// if the cluster has been registered before by another secret, ignore the new one.
 			logger.Warnf("cluster has already been registered")

pkg/kube/namespace/filter.go

@@ -54,10 +54,16 @@ func NewDiscoveryNamespacesFilter(
 		namespaces:          namespaces,
 		discoveryNamespaces: sets.New[string](),
 	}
-	mesh.AddMeshHandler(func() {
+	reg := mesh.AddMeshHandler(func() {
 		f.selectorsChanged(mesh.Mesh().GetDiscoverySelectors(), true)
 	})
 
+	// Clean up mesh handler on stop
+	go func() {
+		<-stop
+		mesh.DeleteMeshHandler(reg)
+	}()
+
 	namespaces.AddEventHandler(controllers.EventHandler[*corev1.Namespace]{
 		AddFunc: func(ns *corev1.Namespace) {
 			f.lock.Lock()