switch to tracking services

Author: ajssmith

internal/kube/controller/controller.go

@@ -40,6 +40,7 @@ type Controller struct {
 	connectorWatcher     *watchers.ConnectorWatcher
 	linkAccessWatcher    *watchers.RouterAccessWatcher
 	grantWatcher         *watchers.AccessGrantWatcher
+	serviceWatcher       *watchers.ServiceWatcher
 	sites                map[string]*site.Site
 	startGrantServer     func()
 	accessMgr            *securedaccess.SecuredAccessManager
@@ -52,6 +53,7 @@ type Controller struct {
 	attachableConnectors map[string]*skupperv2alpha1.AttachedConnector
 	log                  *slog.Logger
 	namespaces           *NamespaceConfig
+	observedServices     map[string]string
 }
 
 func skupperRouterConfig() internalinterfaces.TweakListOptionsFunc {
@@ -71,6 +73,12 @@ func listenerServices() internalinterfaces.TweakListOptionsFunc {
 	}
 }
 
+func sansSkupperListenerServices() internalinterfaces.TweakListOptionsFunc {
+	return func(options *metav1.ListOptions) {
+		options.LabelSelector = "!internal.skupper.io/listener"
+	}
+}
+
 func skupperSiteSizingConfig() internalinterfaces.TweakListOptionsFunc {
 	return func(options *metav1.ListOptions) {
 		options.LabelSelector = sizing.SiteSizingLabel
@@ -91,6 +99,7 @@ func NewController(cli internalclient.Clients, config *Config, options ...watche
 		labelling:            labels.NewLabelsAndAnnotations(config.Namespace),
 		attachableConnectors: map[string]*skupperv2alpha1.AttachedConnector{},
 		log:                  slog.New(slog.Default().Handler()).With(slog.String("component", "kube.controller")),
+		observedServices:     map[string]string{},
 	}
 
 	hostname := os.Getenv("HOSTNAME")
@@ -120,6 +129,7 @@ func NewController(cli internalclient.Clients, config *Config, options ...watche
 	controller.siteWatcher = controller.eventProcessor.WatchSites(config.WatchNamespace, filter(controller, controller.checkSite))
 	controller.listenerWatcher = controller.eventProcessor.WatchListeners(config.WatchNamespace, filter(controller, controller.checkListener))
 	controller.eventProcessor.WatchServices(listenerServices(), config.WatchNamespace, filter(controller, controller.checkListenerService))
+	controller.serviceWatcher = controller.eventProcessor.WatchServices(sansSkupperListenerServices(), config.WatchNamespace, filter(controller, controller.checkObservedService))
 	controller.connectorWatcher = controller.eventProcessor.WatchConnectors(config.WatchNamespace, filter(controller, controller.checkConnector))
 	controller.linkAccessWatcher = controller.eventProcessor.WatchRouterAccesses(config.WatchNamespace, filter(controller, controller.checkRouterAccess))
 	controller.eventProcessor.WatchAttachedConnectors(config.WatchNamespace, filter(controller, controller.checkAttachedConnector))
@@ -220,6 +230,10 @@ func (c *Controller) init(stopCh <-chan struct{}) error {
 		)
 		c.labelling.Update(config.Namespace+"/"+config.Name, config)
 	}
+	// get observed services prior to restoring listeners
+	for _, svc := range c.serviceWatcher.List() {
+		c.observedServices[svc.Namespace+"/"+svc.ObjectMeta.Name] = svc.ObjectMeta.Name
+	}
 	//recover existing sites & bindings
 	siteRecovery := site.NewSiteRecovery(c.eventProcessor.GetKubeClient())
 	for _, site := range c.siteWatcher.List() {
@@ -266,7 +280,8 @@ func (c *Controller) init(stopCh <-chan struct{}) error {
 			slog.String("name", listener.Name),
 			slog.String("namespace", listener.Namespace),
 		)
-		site.CheckListener(listener.ObjectMeta.Name, listener)
+		_, svcExists := c.observedServices[listener.ObjectMeta.Namespace+"/"+listener.Spec.Host]
+		site.CheckListener(listener.ObjectMeta.Name, listener, svcExists)
 	}
 	for _, la := range c.linkAccessWatcher.List() {
 		if !c.namespaces.isControlled(la.Namespace) {
@@ -367,7 +382,11 @@ func (c *Controller) checkListener(key string, listener *skupperv2alpha1.Listene
 	if err != nil {
 		return err
 	}
-	return c.getSite(namespace).CheckListener(name, listener)
+	svcExists := false
+	if listener != nil {
+		_, svcExists = c.observedServices[namespace+"/"+listener.Spec.Host]
+	}
+	return c.getSite(namespace).CheckListener(name, listener, svcExists)
 }
 
 func (c *Controller) checkListenerService(key string, svc *corev1.Service) error {
@@ -378,6 +397,17 @@ func (c *Controller) checkListenerService(key string, svc *corev1.Service) error
 	return c.getSite(svc.Namespace).CheckListenerService(svc)
 }
 
+func (c *Controller) checkObservedService(key string, svc *corev1.Service) error {
+	c.log.Debug("checkObservedService", slog.String("key", key))
+
+	if svc == nil {
+		delete(c.observedServices, key)
+	} else {
+		c.observedServices[key] = svc.ObjectMeta.Name
+	}
+	return nil
+}
+
 func (c *Controller) checkLink(key string, linkconfig *skupperv2alpha1.Link) error {
 	namespace, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {

internal/kube/site/per_target_listener.go

@@ -31,7 +31,7 @@ func (p *PerTargetListener) updateListener(l *skupperv2alpha1.Listener) bool {
 }
 
 func (p *PerTargetListener) extractTargets(network []skupperv2alpha1.SiteRecord, mapping *qdr.PortMapping, exposedPorts ExposedPorts, context BindingContext) (bool, error) {
-	p.logger.Info("Extracting targets for listener",
+	p.logger.Debug("Extracting targets for listener",
 		slog.String("namespace", p.definition.Namespace),
 		slog.String("listener", p.definition.Name))
 	targets := extractTargets(p.address(""), network)

internal/kube/site/site.go

@@ -920,21 +920,15 @@ func (s *Site) CheckListenerService(svc *corev1.Service) error {
 	return nil
 }
 
-func (s *Site) CheckListener(name string, listener *skupperv2alpha1.Listener) error {
+func (s *Site) CheckListener(name string, listener *skupperv2alpha1.Listener, svcExists bool) error {
 	if s.site == nil {
 		if listener == nil {
 			return nil
 		}
 		return s.updateListenerStatus(listener, stderrors.New("No active site in namespace"))
 	}
-	if listener != nil {
-		ctxt := context.TODO()
-		current, err := s.clients.GetKubeClient().CoreV1().Services(s.namespace).Get(ctxt, listener.Spec.Host, metav1.GetOptions{})
-		if current != nil && err == nil {
-			if !isOwned(current) {
-				return s.updateListenerStatus(listener, fmt.Errorf("Service to expose %s already exists in namespace", listener.Spec.Host))
-			}
-		}
+	if listener != nil && svcExists {
+		return s.updateListenerStatus(listener, fmt.Errorf("Service %s already exists in namespace", listener.Spec.Host))
 	}
 
 	update, err1 := s.bindings.UpdateListener(name, listener)

internal/kube/site/site_test.go

@@ -280,8 +280,9 @@ func TestSite_ExposeUnexpose(t *testing.T) {
 }
 func TestSite_CheckListener(t *testing.T) {
 	type args struct {
-		name     string
-		listener *skupperv2alpha1.Listener
+		name      string
+		listener  *skupperv2alpha1.Listener
+		svcExists bool
 	}
 	tests := []struct {
 		name                string
@@ -317,6 +318,7 @@ func TestSite_CheckListener(t *testing.T) {
 						Host:       "1.2.3.4",
 					},
 				},
+				svcExists: false,
 			},
 			skupperObjects: []runtime.Object{
 				&skupperv2alpha1.Listener{
@@ -347,6 +349,7 @@ func TestSite_CheckListener(t *testing.T) {
 						Host:       "backend",
 					},
 				},
+				svcExists: true,
 			},
 			skupperObjects: []runtime.Object{
 				&skupperv2alpha1.Listener{
@@ -418,7 +421,7 @@ func TestSite_CheckListener(t *testing.T) {
 				assert.Assert(t, err)
 			}
 
-			if err := s.CheckListener(tt.args.name, tt.args.listener); (err != nil) != tt.wantErr {
+			if err := s.CheckListener(tt.args.name, tt.args.listener, tt.args.svcExists); (err != nil) != tt.wantErr {
 				t.Errorf("Site.CheckListener() error = %v", err)
 			}