Recreate listener Service when deleted (#2387)

* Recreate listener Service when deleted

Fixes #2383

* Added unit test

Author: AryanP123

internal/kube/controller/controller.go

@@ -407,7 +407,11 @@ func (c *Controller) checkListener(key string, listener *skupperv2alpha1.Listene
 	if listener != nil {
 		_, svcExists = c.observedServices[namespace+"/"+listener.Spec.Host]
 	}
-	return c.getSite(namespace).CheckListener(name, listener, svcExists)
+	err = c.getSite(namespace).CheckListener(name, listener, svcExists)
+	if err != nil {
+		return err
+	}
+	return nil
 }
 
 func (c *Controller) checkMultiKeyListener(key string, mkl *skupperv2alpha1.MultiKeyListener) error {
@@ -422,7 +426,14 @@ func (c *Controller) checkMultiKeyListener(key string, mkl *skupperv2alpha1.Mult
 func (c *Controller) checkListenerService(key string, svc *corev1.Service) error {
 	c.log.Debug("checkListenerService", slog.String("key", key))
 	if svc == nil {
-		return nil
+		namespace, serviceName, err := cache.SplitMetaNamespaceKey(key)
+		if err != nil {
+			return err
+		}
+		if !c.namespaces.isControlled(namespace) {
+			return nil
+		}
+		return c.getSite(namespace).HandleDeletedListenerService(serviceName)
 	}
 	return c.getSite(svc.Namespace).CheckListenerService(svc)
 }

internal/kube/controller/controller_test.go

@@ -35,6 +35,7 @@ import (
 	"github.com/skupperproject/skupper/internal/kube/resource"
 	"github.com/skupperproject/skupper/internal/network"
 	"github.com/skupperproject/skupper/internal/qdr"
+	"github.com/skupperproject/skupper/internal/utils"
 	"github.com/skupperproject/skupper/internal/version"
 	skupperv2alpha1 "github.com/skupperproject/skupper/pkg/apis/skupper/v2alpha1"
 )
@@ -663,6 +664,18 @@ func TestUpdate(t *testing.T) {
 				serviceCheck("adifferentsvc", "test").check,
 				negativeServiceCheck("mysvc", "test"),
 			},
+		}, {
+			name: "listener service recreated when deleted",
+			skupperObjects: []runtime.Object{
+				f.site("mysite", "test", "", false, false),
+				f.listener("mylistener", "test", "mysvc", 8080),
+			},
+			functions: []WaitFunction{
+				isListenerStatusConditionTrue("mylistener", "test", skupperv2alpha1.CONDITION_TYPE_CONFIGURED),
+				serviceCheck("mysvc", "test").check,
+				deleteListenerService("mysvc", "test"),
+				waitForService("mysvc", "test"),
+			},
 		}, {
 			name: "exposePodsByName handles pod delete",
 			k8sObjects: []runtime.Object{
@@ -840,6 +853,47 @@ func TestUpdate(t *testing.T) {
 	}
 }
 
+func TestListenerServiceRecreatedWhenDeleted(t *testing.T) {
+	runListenerServiceRecreatedWithFakeClient(t)
+}
+
+func runListenerServiceRecreatedWithFakeClient(t *testing.T) {
+	t.Helper()
+	flags := flag.NewFlagSet("", flag.ContinueOnError)
+	config, err := BoundConfig(flags)
+	assert.NilError(t, err)
+	clients, err := fakeclient.NewFakeClient("test", nil, []runtime.Object{
+		f.site("mysite", "test", "", false, false),
+		f.listener("mylistener", "test", "mysvc", 8080),
+	}, "")
+	assert.NilError(t, err)
+	enableSSA(clients.GetDynamicClient())
+	ctrl, err := NewController(clients, config, func(e *watchers.EventProcessor) { e.SetResyncShort(time.Second) })
+	assert.NilError(t, err)
+	stopCh := make(chan struct{})
+	err = ctrl.init(stopCh)
+	assert.NilError(t, err)
+	for i := 0; i < 2; i++ {
+		ctrl.eventProcessor.TestProcess()
+	}
+	err = utils.Retry(100*time.Millisecond, 50, func() (bool, error) {
+		ctrl.eventProcessor.TestProcess()
+		return isListenerStatusConditionTrue("mylistener", "test", skupperv2alpha1.CONDITION_TYPE_CONFIGURED)(t, clients), nil
+	})
+	assert.NilError(t, err)
+	err = utils.Retry(100*time.Millisecond, 50, func() (bool, error) {
+		ctrl.eventProcessor.TestProcess()
+		return serviceCheck("mysvc", "test").check(t, clients), nil
+	})
+	assert.NilError(t, err)
+	deleteListenerService("mysvc", "test")(t, clients)
+	err = utils.Retry(100*time.Millisecond, 50, func() (bool, error) {
+		ctrl.eventProcessor.TestProcess()
+		return waitForService("mysvc", "test")(t, clients), nil
+	})
+	assert.NilError(t, err)
+}
+
 func deleteAttachedConnector(name string, namespace string) WaitFunction {
 	return func(t *testing.T, clients internalclient.Clients) bool {
 		err := clients.GetSkupperClient().SkupperV2alpha1().AttachedConnectors(namespace).Delete(context.Background(), name, metav1.DeleteOptions{})
@@ -1973,6 +2027,17 @@ func (s *ServiceCheck) checkAbsent(t *testing.T, clients internalclient.Clients)
 	return false
 }
 
+func waitForService(name string, namespace string) WaitFunction {
+	return func(t *testing.T, clients internalclient.Clients) bool {
+		_, err := clients.GetKubeClient().CoreV1().Services(namespace).Get(context.Background(), name, metav1.GetOptions{})
+		if errors.IsNotFound(err) {
+			return false
+		}
+		assert.Assert(t, err)
+		return true
+	}
+}
+
 func updateListener(name string, namespace string, host string, port int) WaitFunction {
 	return func(t *testing.T, clients internalclient.Clients) bool {
 		ctxt := context.Background()
@@ -1995,6 +2060,14 @@ func negativeServiceCheck(name string, namespace string) WaitFunction {
 	}
 }
 
+func deleteListenerService(serviceName string, namespace string) WaitFunction {
+	return func(t *testing.T, clients internalclient.Clients) bool {
+		err := clients.GetKubeClient().CoreV1().Services(namespace).Delete(context.Background(), serviceName, metav1.DeleteOptions{})
+		assert.Assert(t, err)
+		return true
+	}
+}
+
 func deleteTargetPod(name string, namespace string) WaitFunction {
 	return func(t *testing.T, clients internalclient.Clients) bool {
 		ctxt := context.Background()

internal/kube/site/extended_bindings.go

@@ -109,28 +109,33 @@ func (a *ExtendedBindings) ListenerUpdated(listener *skupperv2alpha1.Listener) {
 			slog.String("name", listener.Name),
 			slog.Any("error", err),
 		)
-	} else {
-		port := Port{
-			Name:       listener.Name,
-			Port:       listener.Spec.Port,
-			TargetPort: allocatedRouterPort,
-			Protocol:   listener.Protocol(),
+		return
+	}
+	port := Port{
+		Name:       listener.Name,
+		Port:       listener.Spec.Port,
+		TargetPort: allocatedRouterPort,
+		Protocol:   listener.Protocol(),
+	}
+	if exposed := a.exposed.Expose(listener.Spec.Host, port); exposed != nil {
+		if err := a.context.Expose(exposed); err != nil {
+			bindings_logger.Error("Error exposing listener",
+				slog.String("namespace", listener.Namespace),
+				slog.String("name", listener.Name),
+				slog.Any("error", err))
+		} else {
+			bindings_logger.Info("Exposed listener",
+				slog.String("namespace", listener.Namespace),
+				slog.String("name", listener.Name))
 		}
-		if exposed := a.exposed.Expose(listener.Spec.Host, port); exposed != nil {
-			if err := a.context.Expose(exposed); err != nil {
-				//TODO: write error to listener status
-				bindings_logger.Error("Error exposing listener",
-					slog.String("namespace", listener.Namespace),
-					slog.String("name", listener.Name),
-					slog.Any("error", err))
-			} else {
-				bindings_logger.Info("Exposed listener",
-					slog.String("namespace", listener.Namespace),
-					slog.String("name", listener.Name))
+	}
+}
 
-			}
-		}
+func (a *ExtendedBindings) GetExposedPortSet(host string) *ExposedPortSet {
+	if existing, ok := a.exposed[host]; ok && !existing.empty() {
+		return existing
 	}
+	return nil
 }
 
 func (a *ExtendedBindings) ListenerDeleted(listener *skupperv2alpha1.Listener) {

internal/kube/site/site.go

@@ -936,6 +936,17 @@ func (s *Site) CheckListenerService(svc *corev1.Service) error {
 	return nil
 }
 
+func (s *Site) HandleDeletedListenerService(serviceName string) error {
+	if s.site == nil {
+		return nil
+	}
+	portSet := s.bindings.GetExposedPortSet(serviceName)
+	if portSet == nil {
+		return nil
+	}
+	return s.Expose(portSet)
+}
+
 func (s *Site) CheckListener(name string, listener *skupperv2alpha1.Listener, svcExists bool) error {
 	if s.site == nil {
 		if listener == nil {