Repository: skupperproject/skupper
Pull request #1324 (2023-12-15 14:30:32)

fix psa warning
For k8s 1.24+


Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index c6a73b04..40b03b91 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -322,6 +322,10 @@ commands:
             sudo apt-get update -qq
             sudo apt-get -qq -y install podman
             podman version
+            
+            # temporary fix for https://github.com/containers/podman/issues/21024
+            wget https://download.opensuse.org/repositories/devel:/kubic:/libcontainers:/stable/xUbuntu_22.04/amd64/conmon_2.1.2~0_amd64.deb -O /tmp/conmon_2.1.2.deb
+            sudo apt install /tmp/conmon_2.1.2.deb
 
             # Bypassing CircleCI issue with user session (see: https://github.com/containers/podman/issues/16529)
             mkdir -p ~/.config/containers/containers.conf.d
diff --git a/api/types/client.go b/api/types/client.go
index 30cace52..6dfea411 100644
--- a/api/types/client.go
+++ b/api/types/client.go
@@ -350,6 +350,5 @@ type VanClientInterface interface {
 	GetIngressDefault() string
 	RevokeAccess(ctx context.Context) error
 	NetworkStatus(ctx context.Context) (*network.NetworkStatusInfo, error)
-	GetRemoteLinks(ctx context.Context, siteConfig *SiteConfig) ([]*network.RemoteLinkInfo, error)
 	GetConsoleUrl(namespace string) (string, error)
 }
diff --git a/client/network_status.go b/client/network_status.go
index cea132ea..b9ea34f0 100644
--- a/client/network_status.go
+++ b/client/network_status.go
@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"github.com/skupperproject/skupper/api/types"
-	"github.com/skupperproject/skupper/pkg/domain/kube"
 	k8s "github.com/skupperproject/skupper/pkg/kube"
 	"github.com/skupperproject/skupper/pkg/network"
 )
@@ -29,12 +28,3 @@ func (cli *VanClient) NetworkStatus(ctx context.Context) (*network.NetworkStatus
 
 	return vanInfo, nil
 }
-
-func (cli *VanClient) GetRemoteLinks(ctx context.Context, siteConfig *types.SiteConfig) ([]*network.RemoteLinkInfo, error) {
-	cfg, err := cli.getRouterConfig(ctx, cli.Namespace)
-	if err != nil {
-		return nil, err
-	}
-	linkHander := kube.NewLinkHandlerKube(cli.Namespace, siteConfig, cfg, cli.KubeClient, cli.RestConfig)
-	return linkHander.RemoteLinks(ctx)
-}
diff --git a/client/podman/rest.go b/client/podman/rest.go
index e73b595a..2328beef 100644
--- a/client/podman/rest.go
+++ b/client/podman/rest.go
@@ -37,6 +37,8 @@ type PodmanRestClient struct {
 	endpoint   string
 }
 
+type RestClientFactory func(endpoint, basePath string) (*PodmanRestClient, error)
+
 func NewPodmanClient(endpoint, basePath string) (*PodmanRestClient, error) {
 	var err error
 
diff --git a/client/router_create.go b/client/router_create.go
index 6d60bf0b..4dbd7091 100644
--- a/client/router_create.go
+++ b/client/router_create.go
@@ -3,13 +3,14 @@ package client
 import (
 	"context"
 	"fmt"
-	"github.com/briandowns/spinner"
 	"log"
 	"net"
 	"strconv"
 	"strings"
 	"time"
 
+	"github.com/briandowns/spinner"
+
 	routev1 "github.com/openshift/api/route/v1"
 	"github.com/skupperproject/skupper/pkg/config"
 	"github.com/skupperproject/skupper/pkg/images"
@@ -23,6 +24,7 @@ import (
 	kubetypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/util/validation"
+	apiversion "k8s.io/apimachinery/pkg/version"
 
 	"github.com/skupperproject/skupper/api/types"
 	"github.com/skupperproject/skupper/pkg/kube"
@@ -73,6 +75,15 @@ func ConfigSyncContainer() *corev1.Container {
 	}
 }
 
+func addPsa(vi *apiversion.Info) bool {
+	// for kubernetes versions 1.24+
+	if vi.Major == "1" && strings.Compare(vi.Minor, "24") >= 0 {
+		return true
+	} else {
+		return false
+	}
+}
+
 func (cli *VanClient) getControllerRules() []rbacv1.PolicyRule {
 	return cli.adjustRules(types.ControllerPolicyRule)
 }
@@ -285,7 +296,17 @@ func (cli *VanClient) GetVanControllerSpec(options types.SiteConfigSpec, van *ty
 	if options.RunAsGroup > 0 {
 		van.Controller.SecurityContext.RunAsGroup = &options.RunAsGroup
 	}
-
+	vi, err := cli.KubeClient.Discovery().ServerVersion()
+	if err == nil && addPsa(vi) {
+		ape := false
+		van.Controller.SecurityContext.AllowPrivilegeEscalation = &ape
+		van.Controller.SecurityContext.SeccompProfile = &corev1.SeccompProfile{
+			Type: corev1.SeccompProfileTypeRuntimeDefault,
+		}
+		van.Controller.SecurityContext.Capabilities = &corev1.Capabilities{
+			Drop: []corev1.Capability{"ALL"},
+		}
+	}
 	envVars := []corev1.EnvVar{}
 	envVars = append(envVars, corev1.EnvVar{Name: "SKUPPER_NAMESPACE", Value: van.Namespace})
 	envVars = append(envVars, corev1.EnvVar{Name: "SKUPPER_SITE_NAME", Value: van.Name})
@@ -747,7 +768,18 @@ func (cli *VanClient) GetRouterSpecFromOpts(options types.SiteConfigSpec, siteId
 	if options.RunAsGroup > 0 {
 		van.Transport.SecurityContext.RunAsGroup = &options.RunAsGroup
 	}
-	err := configureDeployment(&van.Transport, &options.Router.Tuning)
+	vi, err := cli.KubeClient.Discovery().ServerVersion()
+	if err == nil && addPsa(vi) {
+		ape := false
+		van.Transport.SecurityContext.AllowPrivilegeEscalation = &ape
+		van.Transport.SecurityContext.SeccompProfile = &corev1.SeccompProfile{
+			Type: corev1.SeccompProfileTypeRuntimeDefault,
+		}
+		van.Transport.SecurityContext.Capabilities = &corev1.Capabilities{
+			Drop: []corev1.Capability{"ALL"},
+		}
+	}
+	err = configureDeployment(&van.Transport, &options.Router.Tuning)
 	if err != nil {
 		fmt.Println("Error configuring router:", err)
 	}
diff --git a/client/router_inspect.go b/client/router_inspect.go
index 35193d96..a1291407 100644
--- a/client/router_inspect.go
+++ b/client/router_inspect.go
@@ -162,7 +162,7 @@ func getSelf(sites []data.Site, siteId string) *data.Site {
 }
 
 func (cli *VanClient) exec(command []string, namespace string) (*bytes.Buffer, error) {
-	pod, err := kube.GetReadyPod(namespace, cli.KubeClient, "service-controller")
+	pod, err := kube.GetReadyPod(namespace, cli.KubeClient, "service-controller", "")
 	if err != nil {
 		return nil, err
 	}
diff --git a/cmd/manifest/manifest.go b/cmd/manifest/manifest.go
index fb332039..1dafb566 100644
--- a/cmd/manifest/manifest.go
+++ b/cmd/manifest/manifest.go
@@ -1,89 +1,13 @@
 package main
 
 import (
-	"encoding/json"
 	"fmt"
+	"github.com/skupperproject/skupper/pkg/utils/configs"
 	"os"
 
-	"github.com/skupperproject/skupper/pkg/images"
 	"github.com/spf13/cobra"
 )
 
-type SkupperImage struct {
-	Name       string `yaml:"name"`
-	SHA        string `yaml:"sha"`
-	Repository string `yaml:"repository,omitempty"`
-}
-
-func generateManifestFile() error {
-	// Define a struct for the manifest file.
-	manifest := struct {
-		Images []SkupperImage `json:"images"`
-	}{
-		Images: []SkupperImage{
-			{
-				Name:       images.GetRouterImageName(),
-				SHA:        images.GetSha(images.GetRouterImageName()),
-				Repository: "https://github.com/skupperproject/skupper-router",
-			},
-			{
-				Name:       images.GetServiceControllerImageName(),
-				SHA:        images.GetSha(images.GetServiceControllerImageName()),
-				Repository: "https://github.com/skupperproject/skupper",
-			},
-			{
-				Name:       images.GetControllerPodmanImageName(),
-				SHA:        images.GetSha(images.GetControllerPodmanImageName()),
-				Repository: "https://github.com/skupperproject/skupper",
-			},
-			{
-				Name:       images.GetConfigSyncImageName(),
-				SHA:        images.GetSha(images.GetConfigSyncImageName()),
-				Repository: "https://github.com/skupperproject/skupper",
-			},
-			{
-				Name:       images.GetFlowCollectorImageName(),
-				SHA:        images.GetSha(images.GetFlowCollectorImageName()),
-				Repository: "https://github.com/skupperproject/skupper",
-			},
-			{
-				Name:       images.GetSiteControllerImageName(),
-				SHA:        images.GetSha(images.GetSiteControllerImageName()),
-				Repository: "https://github.com/skupperproject/skupper",
-			},
-			{
-				Name: images.GetPrometheusServerImageName(),
-				SHA:  images.GetSha(images.GetPrometheusServerImageName()),
-			},
-			{
-				Name: images.GetOauthProxyImageName(),
-				SHA:  images.GetSha(images.GetOauthProxyImageName()),
-			},
-		},
-	}
-
-	// Encode the manifest image list as JSON.
-	manifestListJSON, err := json.MarshalIndent(manifest, "", "   ")
-	if err != nil {
-		return fmt.Errorf("Error encoding manifest image list: %v\n", err)
-
-	}
-
-	// Create a new file.
-	file, err := os.Create("manifest.json")
-	if err != nil {
-		return fmt.Errorf("Error creating file: %v\n", err)
-	}
-
-	// Write the JSON data to the file.
-	_, err = file.Write(manifestListJSON)
-	if err != nil {
-		return fmt.Errorf("Error writing to file: %v\n", err)
-	}
-
-	return nil
-}
-
 func main() {
 
 	var command = &cobra.Command{
@@ -93,7 +17,9 @@ func main() {
 		Short: "generates a manifest.json file",
 		Args:  cobra.NoArgs,
 		RunE: func(cmd *cobra.Command, args []string) error {
-			return generateManifestFile()
+			manifestManager := configs.ManifestManager{EnableSHA: true}
+			return manifestManager.CreateFile(manifestManager.GetConfiguredManifest())
+
 		},
 	}
 
diff --git a/cmd/skupper/skupper.go b/cmd/skupper/skupper.go
index a12ff51c..ee834f17 100644
--- a/cmd/skupper/skupper.go
+++ b/cmd/skupper/skupper.go
@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"fmt"
+	"github.com/skupperproject/skupper/pkg/utils/configs"
 	"os"
 	"reflect"
 	"strconv"
@@ -866,6 +867,21 @@ func NewCmdVersion(skupperClient SkupperSiteClient) *cobra.Command {
 	return cmd
 }
 
+func NewCmdVersionManifest() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "manifest",
+		Short: "Report the version of the Skupper images by default and the value of the environment variables",
+		Args:  cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			silenceCobra(cmd)
+
+			manifestManager := configs.ManifestManager{}
+			return manifestManager.CreateFile(manifestManager.GetDefaultManifestWithEnv())
+		},
+	}
+	return cmd
+}
+
 func NewCmdDebug() *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "debug dump <file>, debug events or debug service <service-name>",
@@ -1040,7 +1056,10 @@ func init() {
 	cmdStatusService := NewCmdServiceStatus(skupperCli.Service())
 	cmdLabelsService := NewCmdServiceLabel(skupperCli.Service())
 
+	cmdVersionManifest := NewCmdVersionManifest()
 	cmdVersion := NewCmdVersion(skupperCli.Site())
+	cmdVersion.AddCommand(cmdVersionManifest)
+
 	cmdDebugDump := NewCmdDebugDump(skupperCli.Debug())
 	cmdDebugEvents := NewCmdDebugEvents(skupperCli.Debug())
 	cmdDebugService := NewCmdDebugService(skupperCli.Debug())
diff --git a/cmd/skupper/skupper_kube_network.go b/cmd/skupper/skupper_kube_network.go
index cc048a8a..cc1a1b06 100644
--- a/cmd/skupper/skupper_kube_network.go
+++ b/cmd/skupper/skupper_kube_network.go
@@ -73,8 +73,12 @@ func (s *SkupperKubeNetwork) Status(cmd *cobra.Command, args []string) error {
 
 				if len(siteStatus.RouterStatus) > 0 {
 
-					//to get the generic information about the links of a site, we can get the first router, because in case of multiple routers the information will be the same.
-					mapSiteLink := statusManager.GetSiteLinkMapPerRouter(&siteStatus.RouterStatus[0], &siteStatus.Site)
+					err, index := statusManager.GetRouterIndex(&siteStatus)
+					if err != nil {
+						return err
+					}
+
+					mapSiteLink := statusManager.GetSiteLinkMapPerRouter(&siteStatus.RouterStatus[index], &siteStatus.Site)
 
 					if len(mapSiteLink) > 0 {
 						siteLinks := siteLevel.NewChild("Linked sites:")
@@ -87,26 +91,28 @@ func (s *SkupperKubeNetwork) Status(cmd *cobra.Command, args []string) error {
 						routers := siteLevel.NewChild("Routers:")
 						for _, routerStatus := range siteStatus.RouterStatus {
 							routerId := strings.Split(routerStatus.Router.Name, "/")
-							routerItem := fmt.Sprintf("name: %s\n", routerId[1])
-							detailsRouter := map[string]string{"image name": routerStatus.Router.ImageName, "image version": routerStatus.Router.ImageVersion}
 
-							routerLevel := routers.NewChildWithDetail(routerItem, detailsRouter)
+							// skip routers that belong to headless services
+							if len(routerId) > 1 && strings.HasPrefix(routerId[1], siteStatus.Site.Name) {
+								routerItem := fmt.Sprintf("name: %s\n", routerId[1])
+								detailsRouter := map[string]string{"image name": routerStatus.Router.ImageName, "image version": routerStatus.Router.ImageVersion}
 
-							printableLinks := statusManager.RemoveLinksFromSameSite(routerStatus, siteStatus.Site)
+								routerLevel := routers.NewChildWithDetail(routerItem, detailsRouter)
 
-							if len(printableLinks) > 0 {
-								links := routerLevel.NewChild("Links:")
-								for _, link := range printableLinks {
-									linkItem := fmt.Sprintf("name:  %s\n", link.Name)
-									detailsLink := map[string]string{"direction": link.Direction}
-									if link.LinkCost > 0 {
-										detailsLink["cost"] = strconv.FormatUint(link.LinkCost, 10)
-									}
-									links.NewChildWithDetail(linkItem, detailsLink)
+								printableLinks := statusManager.RemoveLinksFromSameSite(routerStatus, siteStatus.Site)
 
+								if len(printableLinks) > 0 {
+									links := routerLevel.NewChild("Links:")
+									for _, link := range printableLinks {
+										linkItem := fmt.Sprintf("name:  %s\n", link.Name)
+										detailsLink := map[string]string{"direction": link.Direction}
+										if link.LinkCost > 0 {
+											detailsLink["cost"] = strconv.FormatUint(link.LinkCost, 10)
+										}
+										links.NewChildWithDetail(linkItem, detailsLink)
+									}
 								}
 							}
-
 						}
 					}
 				}
diff --git a/cmd/skupper/skupper_kube_service.go b/cmd/skupper/skupper_kube_service.go
index 33a4b0c8..7e7f892c 100644
--- a/cmd/skupper/skupper_kube_service.go
+++ b/cmd/skupper/skupper_kube_service.go
@@ -62,7 +62,7 @@ func (s *SkupperKubeService) Status(cmd *cobra.Command, args []string) error {
 	}
 
 	mapServiceSites := statusManager.GetServiceSitesMap()
-	mapSiteTargets := statusManager.GetSiteTargetsMap()
+	mapSiteTarget := statusManager.GetSiteTargetMap()
 
 	var mapServiceLabels map[string]map[string]string
 	if err == nil {
@@ -74,10 +74,6 @@ func (s *SkupperKubeService) Status(cmd *cobra.Command, args []string) error {
 	} else {
 		l := formatter.NewList()
 		l.Item("Services exposed through Skupper:")
-		var addresses []string
-		for _, si := range currentNetworkStatus.Addresses {
-			addresses = append(addresses, si.Name)
-		}
 
 		for _, si := range currentNetworkStatus.Addresses {
 			svc := l.NewChild(fmt.Sprintf("%s (%s)", si.Name, si.Protocol))
@@ -91,7 +87,7 @@ func (s *SkupperKubeService) Status(cmd *cobra.Command, args []string) error {
 						theSite := sites.NewChildWithDetail(item, map[string]string{"policy": site.Site.Policy})
 
 						if si.ConnectorCount > 0 {
-							t := mapSiteTargets[site.Site.Identity][si.Name]
+							t := mapSiteTarget[site.Site.Identity][si.Name]
 
 							if len(t.Address) > 0 {
 								targets := theSite.NewChild("Targets:")
diff --git a/cmd/skupper/skupper_kube_site.go b/cmd/skupper/skupper_kube_site.go
index 79a2e72b..dc2e68c6 100644
--- a/cmd/skupper/skupper_kube_site.go
+++ b/cmd/skupper/skupper_kube_site.go
@@ -268,7 +268,12 @@ func (s *SkupperKubeSite) Status(cmd *cobra.Command, args []string) error {
 				policies:  currentSite.Site.Policy,
 			}
 
-			mapSiteLink := statusManager.GetSiteLinkMapPerRouter(&currentSite.RouterStatus[0], &currentSite.Site)
+			err, index := statusManager.GetRouterIndex(currentSite)
+			if err != nil {
+				return err
+			}
+
+			mapSiteLink := statusManager.GetSiteLinkMapPerRouter(&currentSite.RouterStatus[index], &currentSite.Site)
 
 			totalSites := len(currentStatus.SiteStatus)
 			// the current site does not count as a connection
diff --git a/cmd/skupper/skupper_link.go b/cmd/skupper/skupper_link.go
index 6b0e967a..b19fa991 100644
--- a/cmd/skupper/skupper_link.go
+++ b/cmd/skupper/skupper_link.go
@@ -227,7 +227,7 @@ func NewCmdLinkStatus(skupperClient SkupperLinkClient) *cobra.Command {
 		},
 	}
 	cmd.Flags().IntVar(&waitFor, "wait", 0, "The number of seconds to wait for links to become connected")
-	cmd.Flags().BoolVar(&verboseLinkStatus, "verbose", false, "Show detailed information about a link")
+	cmd.Flags().BoolVarP(&verboseLinkStatus, "verbose", "v", false, "Show detailed information about a link")
 
 	return cmd
 
diff --git a/cmd/skupper/skupper_podman.go b/cmd/skupper/skupper_podman.go
index aefec2b3..46f8e72f 100644
--- a/cmd/skupper/skupper_podman.go
+++ b/cmd/skupper/skupper_podman.go
@@ -2,6 +2,7 @@ package main
 
 import (
 	"fmt"
+	"io"
 	"os"
 
 	"github.com/skupperproject/skupper/api/types"
@@ -19,14 +20,20 @@ var SkupperPodmanCommands = []string{
 }
 
 type SkupperPodman struct {
-	cli         *clientpodman.PodmanRestClient
-	currentSite *podman.Site
-	site        *SkupperPodmanSite
-	token       *SkupperPodmanToken
-	link        *SkupperPodmanLink
-	service     *SkupperPodmanService
+	cliFactory         clientpodman.RestClientFactory
+	cli                *clientpodman.PodmanRestClient
+	currentSite        *podman.Site
+	siteHandlerFactory podman.SiteHandlerFactory
+	site               *SkupperPodmanSite
+	token              *SkupperPodmanToken
+	link               *SkupperPodmanLink
+	service            *SkupperPodmanService
+	exit               exitHandler
+	output             io.Writer
 }
 
+type exitHandler func(code int)
+
 func (s *SkupperPodman) Site() SkupperSiteClient {
 	if s.site != nil {
 		return s.site
@@ -85,6 +92,10 @@ func (s *SkupperPodman) NewClient(cmd *cobra.Command, args []string) {
 	var endpoint string
 	var isInitCmd bool
 	exitOnError := true
+	if s.output == nil {
+		s.output = os.Stdout
+	}
+	out := s.output
 	switch cmd.Name() {
 	case "init":
 		// require site not present
@@ -97,19 +108,40 @@ func (s *SkupperPodman) NewClient(cmd *cobra.Command, args []string) {
 	default:
 		podmanCfg, err := podman.NewPodmanConfigFileHandler().GetConfig()
 		if err != nil {
-			fmt.Println("error reading current podman endpoint")
+			fmt.Fprintln(out, "error reading current podman endpoint")
 			return
 		}
 		endpoint = podmanCfg.Endpoint
 	}
-
-	c, err := clientpodman.NewPodmanClient(endpoint, "")
+	if s.cliFactory == nil {
+		s.cliFactory = clientpodman.NewPodmanClient
+	}
+	if s.exit == nil {
+		s.exit = os.Exit
+	}
+	c, err := s.cliFactory(endpoint, "")
 	if err != nil {
 		if exitOnError {
-			fmt.Printf("Podman endpoint is not available: %s",
+			fmt.Fprintf(out, "Podman endpoint is not available: %s",
 				utils.DefaultStr(endpoint, clientpodman.GetDefaultPodmanEndpoint()))
-			fmt.Println()
-			os.Exit(1)
+			fmt.Fprintln(out)
+			recommendation := `
+Recommendation:
+
+	Make sure you have an active podman endpoint available.
+	On most systems you can execute:
+
+		systemctl --user enable --now podman.socket
+
+	Alternatively you could also create your own service that runs:
+
+		podman system service --time=0 <URI>
+
+	You can get concrete examples through:
+
+		podman help system service`
+			fmt.Fprintln(out, recommendation)
+			s.exit(1)
 		}
 		return
 	}
@@ -117,32 +149,39 @@ func (s *SkupperPodman) NewClient(cmd *cobra.Command, args []string) {
 	s.cli = c
 
 	// Ensure that site does not exist on init, but exists for all other commands
-	siteHandler, err := podman.NewSitePodmanHandler(endpoint)
+	if s.siteHandlerFactory == nil {
+		s.siteHandlerFactory = podman.NewSiteHandler
+	}
+	siteHandler, err := s.siteHandlerFactory(endpoint)
 	if err != nil {
-		fmt.Printf("error verifying existing skupper site - %s", err)
-		fmt.Println()
-		os.Exit(1)
+		fmt.Fprintf(out, "error verifying existing skupper site - %s", err)
+		fmt.Fprintln(out)
+		s.exit(1)
+		return
 	}
 	currentSite, err := siteHandler.Get()
 	if isInitCmd {
 		// Validating if site is already initialized
 		if err == nil && currentSite != nil {
-			fmt.Printf("Skupper has already been initialized for user '" + podman.Username + "'.")
-			fmt.Println()
-			os.Exit(0)
+			fmt.Fprintf(out, "Skupper has already been initialized for user '"+podman.Username+"'.")
+			fmt.Fprintln(out)
+			s.exit(0)
+			return
 		}
 	} else if !utils.StringSliceContains([]string{"version", "delete"}, cmd.Name()) {
 		// All other commands, but version and delete, must stop now
 		if err != nil {
-			fmt.Printf("Skupper is not enabled for user '%s'", podman.Username)
-			fmt.Println()
-			if siteHandler.AnyResourceLeft() {
-				fmt.Println("Reason:", err)
-				fmt.Println()
-				fmt.Println("There are podman resources missing or left from an earlier installation")
-				fmt.Println("To clean them up, run: skupper delete")
+			fmt.Fprintf(out, "Skupper is not enabled for user '%s'", podman.Username)
+			fmt.Fprintln(out)
+			siteHandlerPodman, ok := siteHandler.(*podman.SiteHandler)
+			if ok && siteHandlerPodman.AnyResourceLeft() {
+				fmt.Fprintln(out, "Reason:", err)
+				fmt.Fprintln(out)
+				fmt.Fprintln(out, "There are podman resources missing or left from an earlier installation")
+				fmt.Fprintln(out, "To clean them up, run: skupper delete")
 			}
-			os.Exit(0)
+			s.exit(0)
+			return
 		}
 	}
 	if currentSite != nil {
diff --git a/cmd/skupper/skupper_podman_site.go b/cmd/skupper/skupper_podman_site.go
index 84ef49b5..d4730e5a 100644
--- a/cmd/skupper/skupper_podman_site.go
+++ b/cmd/skupper/skupper_podman_site.go
@@ -78,26 +78,10 @@ func (s *SkupperPodmanSite) Create(cmd *cobra.Command, args []string) error {
 	siteHandler, err := podman.NewSitePodmanHandler(site.PodmanEndpoint)
 	if err != nil {
 		initErr := fmt.Errorf("Unable to initialize Skupper - %w", err)
-		recommendation := `
-Recommendation:
-
-	Make sure you have an active podman endpoint available.
-	On most systems you can execute:
-
-		systemctl --user enable --now podman.socket
-
-	Alternatively you could also create your own service that runs:
-
-		podman system service --time=0 <URI>
-
-	You can get concrete examples through:
-
-		podman help system service`
 
 		cmd.SilenceUsage = true
 		cmd.SilenceErrors = true
 		fmt.Println("Error:", initErr)
-		fmt.Println(recommendation)
 		return initErr
 	}
 
diff --git a/pkg/domain/kube/link_kube.go b/pkg/domain/kube/link_kube.go
index c9c6cda2..839150ec 100644
--- a/pkg/domain/kube/link_kube.go
+++ b/pkg/domain/kube/link_kube.go
@@ -49,8 +49,9 @@ func (l *LinkHandlerKube) List() ([]*corev1.Secret, error) {
 		return nil, fmt.Errorf("Could not retrieve secrets: %w", err)
 	}
 	var secrets []*corev1.Secret
-	for _, s := range currentSecrets.Items {
-		secrets = append(secrets, &s)
+
+	for i, _ := range currentSecrets.Items {
+		secrets = append(secrets, &currentSecrets.Items[i])
 	}
 	return secrets, nil
 }
diff --git a/pkg/domain/link.go b/pkg/domain/link.go
index 5938692e..cadd3619 100644
--- a/pkg/domain/link.go
+++ b/pkg/domain/link.go
@@ -205,7 +205,6 @@ func GenerateLinkName(linkHandler LinkHandler) string {
 					max = v + 1
 				}
 			}
-
 		}
 	} else {
 		log.Fatal("Could not retrieve token secrets:", err)
diff --git a/pkg/domain/podman/site.go b/pkg/domain/podman/site.go
index e43e29d8..e4e04faf 100644
--- a/pkg/domain/podman/site.go
+++ b/pkg/domain/podman/site.go
@@ -74,6 +74,12 @@ func NewSitePodmanHandlerFromCli(cli *podman.PodmanRestClient) *SiteHandler {
 	}
 }
 
+type SiteHandlerFactory func(endpoint string) (domain.SiteHandler, error)
+
+func NewSiteHandler(endpoint string) (domain.SiteHandler, error) {
+	return NewSitePodmanHandler(endpoint)
+}
+
 func NewSitePodmanHandler(endpoint string) (*SiteHandler, error) {
 	if endpoint == "" {
 		podmanCfg, err := NewPodmanConfigFileHandler().GetConfig()
diff --git a/pkg/flow/collector.go b/pkg/flow/collector.go
index ea85d201..436c196c 100644
--- a/pkg/flow/collector.go
+++ b/pkg/flow/collector.go
@@ -279,69 +279,55 @@ func getRealSizeOf(v interface{}) (int, error) {
 	return b.Len(), nil
 }
 
-func (c *FlowCollector) beaconUpdates(stopCh <-chan struct{}) {
-	for {
-		select {
-		case beaconUpdates := <-c.beaconsIncoming:
-			for _, beaconUpdate := range beaconUpdates {
-				beacon, ok := beaconUpdate.(BeaconRecord)
-				if !ok {
-					log.Println("COLLECTOR: Unable to convert interface to beacon")
-				} else {
-					if source, ok := c.eventSources[beacon.Identity]; !ok {
-						var receivers []*receiver
-						log.Printf("COLLECTOR: Detected event source %s of type %s \n", beacon.Identity, beacon.SourceType)
-						receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address, c.recordsIncoming))
-						if beacon.SourceType == recordNames[Router] {
-							switch c.mode {
-							case RecordMetrics:
-								receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address+".flows", c.recordsIncoming))
-							case RecordStatus:
-								receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address+".logs", c.recordsIncoming))
-							}
-						} else if beacon.SourceType == recordNames[Controller] {
-							receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address+".heartbeats", c.heartbeatsIncoming))
-						}
-						outgoing := make(chan interface{})
-						s := newSender(c.connectionFactory, beacon.Direct, false, outgoing)
-						if c.connectionFactory != nil {
-							s.start()
-						}
-						now := uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
-						c.eventSources[beacon.Identity] = &eventSource{
-							EventSourceRecord: EventSourceRecord{
-								Base: Base{
-									RecType:   recordNames[EventSource],
-									Identity:  beacon.Identity,
-									StartTime: now,
-									EndTime:   0,
-								},
-								Beacon:    &beacon,
-								LastHeard: now,
-								Beacons:   1,
-							},
-							receivers: receivers,
-							send: &senderDirect{
-								sender:    s,
-								outgoing:  outgoing,
-								heartbeat: false,
-							},
-						}
-						if c.connectionFactory != nil {
-							for _, receiver := range receivers {
-								receiver.start()
-							}
-						}
-						c.pendingFlush[beacon.Direct] = c.eventSources[beacon.Identity].send
-					} else {
-						source.LastHeard = uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
-						source.Beacons++
-					}
-				}
+func (c *FlowCollector) beaconUpdate(beacon BeaconRecord) {
+	if source, ok := c.eventSources[beacon.Identity]; !ok {
+		var receivers []*receiver
+		log.Printf("COLLECTOR: Detected event source %s of type %s \n", beacon.Identity, beacon.SourceType)
+		receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address, c.recordsIncoming))
+		if beacon.SourceType == recordNames[Router] {
+			switch c.mode {
+			case RecordMetrics:
+				receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address+".flows", c.recordsIncoming))
+			case RecordStatus:
+				receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address+".logs", c.recordsIncoming))
 			}
-		case <-stopCh:
-			return
+		} else if beacon.SourceType == recordNames[Controller] {
+			receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address+".heartbeats", c.heartbeatsIncoming))
+		}
+		outgoing := make(chan interface{})
+		s := newSender(c.connectionFactory, beacon.Direct, false, outgoing)
+		if c.connectionFactory != nil {
+			s.start()
 		}
+		now := uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
+		c.eventSources[beacon.Identity] = &eventSource{
+			EventSourceRecord: EventSourceRecord{
+				Base: Base{
+					RecType:   recordNames[EventSource],
+					Identity:  beacon.Identity,
+					StartTime: now,
+					EndTime:   0,
+				},
+				Beacon:    &beacon,
+				LastHeard: now,
+				Beacons:   1,
+			},
+			receivers: receivers,
+			send: &senderDirect{
+				sender:    s,
+				outgoing:  outgoing,
+				heartbeat: false,
+			},
+		}
+		if c.connectionFactory != nil {
+			for _, receiver := range receivers {
+				receiver.start()
+			}
+		}
+		c.pendingFlush[beacon.Direct] = c.eventSources[beacon.Identity].send
+	} else {
+		source.LastHeard = uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
+		source.Beacons++
 	}
 }
 
@@ -355,6 +341,15 @@ func (c *FlowCollector) recordUpdates(stopCh <-chan struct{}) {
 
 	for {
 		select {
+		case beaconUpdates := <-c.beaconsIncoming:
+			for _, beaconUpdate := range beaconUpdates {
+				beacon, ok := beaconUpdate.(BeaconRecord)
+				if !ok {
+					log.Println("COLLECTOR: Unable to convert interface to beacon")
+				} else {
+					c.beaconUpdate(beacon)
+				}
+			}
 		case heartbeatUpdates := <-c.heartbeatsIncoming:
 			for _, heartbeatUpdate := range heartbeatUpdates {
 				heartbeat, ok := heartbeatUpdate.(HeartbeatRecord)
@@ -414,9 +409,13 @@ func (c *FlowCollector) run(stopCh <-chan struct{}) {
 	c.beaconReceiver = newReceiver(c.connectionFactory, BeaconAddress, c.beaconsIncoming)
 	c.beaconReceiver.start()
 
-	go c.beaconUpdates(stopCh)
-	go c.recordUpdates(stopCh)
-	<-stopCh
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		c.recordUpdates(stopCh)
+	}()
+	<-done
+	log.Println("COLLECTOR: Finished running. Shutting down")
 	for _, eventsource := range c.eventSources {
 		for _, receiver := range eventsource.receivers {
 			receiver.stop()
diff --git a/pkg/flow/messaging.go b/pkg/flow/messaging.go
index 32ed3a37..2bcdb3a9 100644
--- a/pkg/flow/messaging.go
+++ b/pkg/flow/messaging.go
@@ -2,7 +2,7 @@ package flow
 
 import (
 	"log"
-	"time"
+	"sync"
 
 	amqp "github.com/interconnectedcloud/go-amqp"
 	"github.com/skupperproject/skupper/pkg/event"
@@ -16,36 +16,31 @@ const (
 )
 
 type base struct {
+	stopOnce          sync.Once
+	done              chan struct{}
 	connectionFactory messaging.ConnectionFactory
-	closed            bool
-	client            messaging.Connection
 	incoming          chan []interface{}
 	outgoing          chan interface{}
 	address           string
 }
 
 func (c *base) stop() {
-	c.closed = true
-	if c.client != nil {
-		c.client.Close()
-	}
+	c.stopOnce.Do(func() { close(c.done) })
 }
 
 type sender struct {
 	base
 	sendSettled bool
-	ticker      *time.Ticker
-	request     *amqp.Message
 }
 
 func (c *sender) start() {
-	c.closed = false
 	go c.send()
 }
 
 func newSender(connectionFactory messaging.ConnectionFactory, address string, sendSettled bool, update chan interface{}) *sender {
 	return &sender{
 		base: base{
+			done:              make(chan struct{}),
 			connectionFactory: connectionFactory,
 			outgoing:          update,
 			address:           address,
@@ -55,15 +50,18 @@ func newSender(connectionFactory messaging.ConnectionFactory, address string, se
 }
 
 func (c *sender) send() {
-	c.ticker = time.NewTicker(5 * time.Second)
-	defer c.ticker.Stop()
-	for !c.closed {
-		err := c._send()
-		if err != nil {
-			log.Printf("COLLECTOR: Error sending out updates %s", err.Error())
+	for {
+		select {
+		case <-c.done:
+			log.Println("COLLECTOR: Flow process stopped sending")
+			return
+		default:
+			if err := c._send(); err != nil {
+				log.Printf("COLLECTOR: Error sending out updates %s", err.Error())
+			}
 		}
+
 	}
-	log.Println("COLLECTOR: Flow process stopped sending")
 }
 
 func (c *sender) _send() error {
@@ -71,7 +69,6 @@ func (c *sender) _send() error {
 	if err != nil {
 		return err
 	}
-	c.client = client
 	log.Printf("COLLECTOR: Connection for sender %s to %s established\n", c.address, c.connectionFactory.Url())
 	defer client.Close()
 
@@ -82,15 +79,18 @@ func (c *sender) _send() error {
 
 	defer sender.Close()
 
+	var request *amqp.Message
 	for {
 		select {
+		case <-c.done:
+			return nil
 		case update := <-c.outgoing:
 			if beacon, ok := update.(*BeaconRecord); ok {
 				msg, err := encodeBeacon(beacon)
 				if err != nil {
 					event.Recordf(FlowControllerEvent, "Failed to encode message for flow controller: %s", err.Error())
 				} else {
-					c.request = msg
+					request = msg
 				}
 			}
 			if heartbeat, ok := update.(*HeartbeatRecord); ok {
@@ -98,7 +98,7 @@ func (c *sender) _send() error {
 				if err != nil {
 					event.Recordf(FlowControllerEvent, "Failed to encode message for flow controller: %s", err.Error())
 				} else {
-					c.request = msg
+					request = msg
 				}
 			}
 			if fr, ok := update.(*FlushRecord); ok {
@@ -106,7 +106,7 @@ func (c *sender) _send() error {
 				if err != nil {
 					event.Recordf(FlowControllerEvent, "Failed to encode message for flow controller: %s", err.Error())
 				} else {
-					c.request = msg
+					request = msg
 				}
 			}
 			if site, ok := update.(*SiteRecord); ok {
@@ -114,7 +114,7 @@ func (c *sender) _send() error {
 				if err != nil {
 					event.Recordf(FlowControllerEvent, "Failed to encode message for flow controller: %s", err.Error())
 				} else {
-					c.request = msg
+					request = msg
 				}
 			}
 			if process, ok := update.(*ProcessRecord); ok {
@@ -122,7 +122,7 @@ func (c *sender) _send() error {
 				if err != nil {
 					event.Recordf(FlowControllerEvent, "Failed to encode message for flow controller: %s", err.Error())
 				} else {
-					c.request = msg
+					request = msg
 				}
 			}
 			if host, ok := update.(*HostRecord); ok {
@@ -130,22 +130,19 @@ func (c *sender) _send() error {
 				if err != nil {
 					event.Recordf(FlowControllerEvent, "Failed to encode message for flow controller: %s", err.Error())
 				} else {
-					c.request = msg
+					request = msg
 				}
 			}
-		case <-c.ticker.C:
 		}
-		if c.request != nil {
-			c.request.SendSettled = c.sendSettled
-			err = sender.Send(c.request)
+		if request != nil {
+			request.SendSettled = c.sendSettled
+			err = sender.Send(request)
 			if err != nil {
 				return err
-			} else {
-				c.request = nil
 			}
+			request = nil
 		}
 	}
-	return nil
 }
 
 type receiver struct {
@@ -155,6 +152,7 @@ type receiver struct {
 func newReceiver(connectionFactory messaging.ConnectionFactory, address string, updates chan []interface{}) *receiver {
 	return &receiver{
 		base: base{
+			done:              make(chan struct{}),
 			connectionFactory: connectionFactory,
 			incoming:          updates,
 			address:           address,
@@ -163,15 +161,18 @@ func newReceiver(connectionFactory messaging.ConnectionFactory, address string,
 }
 
 func (r *receiver) start() {
-	r.closed = false
 	go r.receive()
 }
 
 func (r *receiver) receive() {
-	for !r.closed {
-		err := r._receive()
-		if err != nil {
-			log.Printf("COLLECTOR: Receiver %s %s\n", r.address, err.Error())
+	for {
+		select {
+		case <-r.done:
+			return
+		default:
+			if err := r._receive(); err != nil {
+				log.Printf("COLLECTOR: Receiver %s %s\n", r.address, err.Error())
+			}
 		}
 	}
 }
@@ -181,7 +182,6 @@ func (r *receiver) _receive() error {
 	if err != nil {
 		return err
 	}
-	r.client = client
 	log.Printf("COLLECTOR: Connection for receiver %s to %s established\n", r.address, r.connectionFactory.Url())
 	defer client.Close()
 
@@ -192,6 +192,11 @@ func (r *receiver) _receive() error {
 	defer receiver.Close()
 
 	for {
+		select {
+		case <-r.done:
+			return nil
+		default:
+		}
 		msg, err := receiver.Receive()
 		if err != nil {
 			return err
@@ -200,5 +205,4 @@ func (r *receiver) _receive() error {
 		results := decode(msg)
 		r.incoming <- results
 	}
-	return nil
 }
diff --git a/pkg/kube/pods.go b/pkg/kube/pods.go
index 82f5c11d..bffc7447 100644
--- a/pkg/kube/pods.go
+++ b/pkg/kube/pods.go
@@ -20,6 +20,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"k8s.io/apimachinery/pkg/labels"
 	"strings"
 	"time"
 
@@ -65,9 +66,17 @@ func FirstReadyPod(list []corev1.Pod) *corev1.Pod {
 	return nil
 }
 
-func GetReadyPod(namespace string, clientset kubernetes.Interface, component string) (*corev1.Pod, error) {
-	selector := "skupper.io/component=" + component
-	pods, err := clientset.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{LabelSelector: selector})
+func GetReadyPod(namespace string, clientset kubernetes.Interface, component string, application string) (*corev1.Pod, error) {
+	matchLabels := map[string]string{"skupper.io/component": component}
+
+	if application != "" {
+		matchLabels["application"] = application
+	}
+
+	labelSelector := metav1.LabelSelector{MatchLabels: matchLabels}
+	listOptions := metav1.ListOptions{LabelSelector: labels.Set(labelSelector.MatchLabels).String()}
+
+	pods, err := clientset.CoreV1().Pods(namespace).List(context.TODO(), listOptions)
 	if err != nil {
 		return nil, err
 	} else if len(pods.Items) == 0 {
@@ -96,7 +105,11 @@ func GetImageVersion(pod *corev1.Pod, container string) string {
 }
 
 func GetComponentVersion(namespace string, clientset kubernetes.Interface, component string, container string) string {
-	pod, err := GetReadyPod(namespace, clientset, component)
+	application := ""
+	if component == "router" {
+		application = "skupper-router"
+	}
+	pod, err := GetReadyPod(namespace, clientset, component, application)
 	if err == nil {
 		return GetImageVersion(pod, container)
 	} else {
diff --git a/pkg/kube/qdr/mgmt.go b/pkg/kube/qdr/mgmt.go
index 95698269..a4902872 100644
--- a/pkg/kube/qdr/mgmt.go
+++ b/pkg/kube/qdr/mgmt.go
@@ -290,7 +290,7 @@ func getLocalRouterId(namespace string, clientset kubernetes.Interface, config *
 }
 
 func router_exec(command []string, namespace string, clientset kubernetes.Interface, config *restclient.Config) (*bytes.Buffer, error) {
-	pod, err := kube.GetReadyPod(namespace, clientset, "router")
+	pod, err := kube.GetReadyPod(namespace, clientset, "router", "skupper-router")
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/messaging/mock.go b/pkg/messaging/mock.go
new file mode 100644
index 00000000..e4f9fc7e
--- /dev/null
+++ b/pkg/messaging/mock.go
@@ -0,0 +1,189 @@
+package messaging
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	amqp "github.com/interconnectedcloud/go-amqp"
+)
+
+type MockConnectionFactory struct {
+	URL    string
+	Broker *Broker
+}
+
+// NewMockConnectionFactory creates an in-memory message brokering stub.
+// Messages will be sent multicast and are buffered for all unclosed receivers.
+func NewMockConnectionFactory(t *testing.T, url string) *MockConnectionFactory {
+	return &MockConnectionFactory{
+		URL:    url,
+		Broker: NewBroker(),
+	}
+}
+
+func (f *MockConnectionFactory) Connect() (Connection, error) {
+	c := &mockConnection{broker: f.Broker, done: make(chan struct{})}
+	return c, nil
+}
+
+func (f *MockConnectionFactory) Url() string {
+	return f.URL
+}
+
+type Broker struct {
+	mu     sync.Mutex
+	topics map[string]*multicast
+}
+
+func NewBroker() *Broker {
+	return &Broker{
+		topics: map[string]*multicast{},
+	}
+}
+func (b *Broker) AwaitReceivers(address string, n int) {
+	topic := b.get(address)
+	ct := func() int {
+		topic.mu.Lock()
+		defer topic.mu.Unlock()
+		return len(topic.receivers)
+	}
+	for n > ct() {
+		time.Sleep(time.Millisecond * 5)
+	}
+}
+
+func (b *Broker) subscribe(address string, r *mockReceiver) {
+	b.get(address).subscribe(r)
+}
+
+func (b *Broker) send(address string, msg *amqp.Message) {
+	b.get(address).send(msg)
+}
+
+func (b *Broker) get(address string) *multicast {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	t, ok := b.topics[address]
+	if !ok {
+		t = newMulticast()
+		b.topics[address] = t
+	}
+	return t
+}
+
+type multicast struct {
+	mu        sync.Mutex
+	receivers []*mockReceiver
+}
+
+func newMulticast() *multicast {
+	return &multicast{}
+}
+
+func (m *multicast) send(msg *amqp.Message) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for _, r := range m.receivers {
+		r.send(msg)
+	}
+}
+
+func (m *multicast) subscribe(r *mockReceiver) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.receivers = append(m.receivers, r)
+}
+
+type mockConnection struct {
+	broker    *Broker
+	closeOnce sync.Once
+	done      chan struct{}
+}
+
+func (c *mockConnection) Sender(address string) (Sender, error) {
+	return &mockSender{address: address, connection: c, done: make(chan struct{})}, nil
+}
+
+func (c *mockConnection) Receiver(address string, credit uint32) (Receiver, error) {
+	r := &mockReceiver{connection: c, channel: make(chan *amqp.Message, credit), done: make(chan struct{})}
+	c.broker.subscribe(address, r)
+	return r, nil
+}
+
+func (c *mockConnection) Close() {
+	c.closeOnce.Do(func() { close(c.done) })
+}
+
+func (c *mockConnection) send(address string, msg *amqp.Message) error {
+	select {
+	case <-c.done:
+		return fmt.Errorf("channel closed")
+	default:
+		c.broker.send(address, msg)
+	}
+	return nil
+}
+
+func (c *mockConnection) receive(r *mockReceiver) (*amqp.Message, error) {
+	select {
+	case msg := <-r.channel:
+		if msg == nil {
+			return nil, fmt.Errorf("Failed to receive")
+		}
+		return msg, nil
+	case <-c.done:
+		return nil, fmt.Errorf("connection closed")
+	}
+}
+
+type mockSender struct {
+	connection *mockConnection
+	address    string
+	closeOnce  sync.Once
+	done       chan struct{}
+}
+
+func (s *mockSender) Send(msg *amqp.Message) error {
+	select {
+	case <-s.done:
+		return fmt.Errorf("sender closed")
+	default:
+		return s.connection.send(s.address, msg)
+	}
+}
+
+func (s *mockSender) Close() error {
+	s.closeOnce.Do(func() { close(s.done) })
+	return nil
+}
+
+type mockReceiver struct {
+	connection *mockConnection
+	channel    chan *amqp.Message
+	closeOnce  sync.Once
+	done       chan struct{}
+}
+
+func (r *mockReceiver) send(msg *amqp.Message) {
+	select {
+	case <-r.done:
+		return
+	default:
+		r.channel <- msg
+	}
+}
+
+func (r *mockReceiver) Receive() (*amqp.Message, error) {
+	return r.connection.receive(r)
+}
+
+func (r *mockReceiver) Accept(msg *amqp.Message) error {
+	return nil
+}
+
+func (r *mockReceiver) Close() error {
+	r.closeOnce.Do(func() { close(r.done) })
+	return nil
+}
diff --git a/pkg/network/network.go b/pkg/network/network.go
index 07a230ae..0b32a563 100644
--- a/pkg/network/network.go
+++ b/pkg/network/network.go
@@ -17,39 +17,49 @@ func (s *SkupperStatus) GetServiceSitesMap() map[string][]SiteStatusInfo {
 	for _, site := range s.NetworkStatus.SiteStatus {
 
 		if len(site.RouterStatus) > 0 {
-			for _, listener := range site.RouterStatus[0].Listeners {
-				if mapServiceSites[listener.Name] != nil {
-					serviceSites := mapServiceSites[listener.Name]
-
-					serviceSites = append(serviceSites, site)
-					mapServiceSites[listener.Name] = serviceSites
-				} else {
-					mapServiceSites[listener.Name] = []SiteStatusInfo{site}
+			for _, router := range site.RouterStatus {
+				for _, listener := range router.Listeners {
+					if len(mapServiceSites[listener.Address]) == 0 || !sliceContainsSite(mapServiceSites[listener.Address], site) {
+						mapServiceSites[listener.Address] = append(mapServiceSites[listener.Address], site)
+					}
+				}
+
+				/* Checking for headless services:
+				   the service can be available in the same site where the headless service was exposed, but in that site there is no
+				   listeners for the statefulstet.
+				*/
+				for _, connector := range router.Connectors {
+					if len(mapServiceSites[connector.Address]) == 0 || !sliceContainsSite(mapServiceSites[connector.Address], site) {
+						mapServiceSites[connector.Address] = append(mapServiceSites[connector.Address], site)
+					}
 				}
 			}
+
 		}
 	}
 
 	return mapServiceSites
 }
 
-func (s *SkupperStatus) GetSiteTargetsMap() map[string]map[string]ConnectorInfo {
+func (s *SkupperStatus) GetSiteTargetMap() map[string]map[string]ConnectorInfo {
 
-	mapSiteTargets := make(map[string]map[string]ConnectorInfo)
+	mapSiteTarget := make(map[string]map[string]ConnectorInfo)
 
 	for _, site := range s.NetworkStatus.SiteStatus {
 
 		if len(site.RouterStatus) > 0 {
-			for _, connector := range site.RouterStatus[0].Connectors {
-				if mapSiteTargets[site.Site.Identity] == nil {
-					mapSiteTargets[site.Site.Identity] = make(map[string]ConnectorInfo)
+			for _, router := range site.RouterStatus {
+				for _, connector := range router.Connectors {
+					if mapSiteTarget[site.Site.Identity] == nil {
+						mapSiteTarget[site.Site.Identity] = make(map[string]ConnectorInfo)
+					}
+					mapSiteTarget[site.Site.Identity][connector.Address] = connector
 				}
-				mapSiteTargets[site.Site.Identity][connector.Address] = connector
 			}
 		}
 	}
 
-	return mapSiteTargets
+	return mapSiteTarget
 }
 
 func (s *SkupperStatus) GetRouterSiteMap() map[string]SiteStatusInfo {
@@ -59,7 +69,11 @@ func (s *SkupperStatus) GetRouterSiteMap() map[string]SiteStatusInfo {
 			for _, routerStatus := range siteStatus.RouterStatus {
 				// the name of the router has a "0/" as a prefix that it is needed to remove
 				routerName := strings.Split(routerStatus.Router.Name, "/")
-				mapRouterSite[routerName[1]] = siteStatus
+
+				// Remove routers that belong to statefulsets for headless services
+				if len(routerName) > 1 && strings.HasPrefix(routerName[1], siteStatus.Site.Name) {
+					mapRouterSite[routerName[1]] = siteStatus
+				}
 			}
 		}
 	}
@@ -79,7 +93,6 @@ func (s *SkupperStatus) GetSiteById(siteId string) *SiteStatusInfo {
 }
 
 func (s *SkupperStatus) GetSiteLinkMapPerRouter(router *RouterStatusInfo, site *SiteInfo) map[string]LinkInfo {
-
 	routerSiteMap := s.GetRouterSiteMap()
 	siteLinkMap := make(map[string]LinkInfo)
 	if len(router.Links) > 0 {
@@ -104,6 +117,21 @@ func (s *SkupperStatus) LinkBelongsToSameSite(linkName string, siteId string, ro
 
 }
 
+func (s *SkupperStatus) GetRouterIndex(site *SiteStatusInfo) (error, int) {
+
+	for index, router := range site.RouterStatus {
+		// Ignore routers that belong to statefulsets for headless services and any other router
+		routerId := strings.Split(router.Router.Name, "/")
+
+		if len(routerId) > 1 && strings.HasPrefix(routerId[1], site.Site.Name) {
+			return nil, index
+
+		}
+	}
+
+	return fmt.Errorf("not valid router found"), -1
+}
+
 func (s *SkupperStatus) RemoveLinksFromSameSite(router RouterStatusInfo, site SiteInfo) []LinkInfo {
 	routerSiteMap := s.GetRouterSiteMap()
 	var filteredLinks []LinkInfo
@@ -129,3 +157,13 @@ func UnmarshalSkupperStatus(data map[string]string) (*NetworkStatusInfo, error)
 
 	return networkStatusInfo, nil
 }
+
+func sliceContainsSite(sites []SiteStatusInfo, site SiteStatusInfo) bool {
+	for _, s := range sites {
+		if site.Site.Identity == s.Site.Identity {
+			return true
+		}
+	}
+
+	return false
+}
diff --git a/pkg/utils/configs/manifest.go b/pkg/utils/configs/manifest.go
new file mode 100644
index 00000000..47b510bc
--- /dev/null
+++ b/pkg/utils/configs/manifest.go
@@ -0,0 +1,208 @@
+package configs
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/skupperproject/skupper/pkg/images"
+	"os"
+	"strings"
+)
+
+type SkupperImage struct {
+	Name       string `json:"name"`
+	SHA        string `json:"sha,omitempty"`
+	Repository string `json:"repository,omitempty"`
+}
+
+type Manifest struct {
+	Images    []SkupperImage     `json:"images"`
+	Variables *map[string]string `json:"variables,omitempty"`
+}
+
+type ManifestManager struct {
+	EnableSHA bool
+}
+
+type ManifestGenerator interface {
+	GetConfiguredManifest() Manifest
+	GetDefaultManifestWithEnv() Manifest
+	CreateFile(m Manifest) error
+}
+
+func (manager *ManifestManager) GetConfiguredManifest() Manifest {
+	return Manifest{
+		Images: getSkupperConfiguredImages(manager.EnableSHA),
+	}
+}
+
+func (manager *ManifestManager) GetDefaultManifestWithEnv() Manifest {
+	return Manifest{
+		Images:    getSkupperDefaultImages(),
+		Variables: getEnvironmentVariableMap(),
+	}
+}
+
+func (manager *ManifestManager) CreateFile(m Manifest) error {
+	// Encode the manifest image list as JSON.
+	manifestListJSON, err := json.MarshalIndent(m, "", "   ")
+	if err != nil {
+		return fmt.Errorf("Error encoding manifest image list: %v\n", err)
+
+	}
+
+	// Create a new file.
+	file, err := os.Create("manifest.json")
+	if err != nil {
+		return fmt.Errorf("Error creating file: %v\n", err)
+	}
+
+	// Write the JSON data to the file.
+	_, err = file.Write(manifestListJSON)
+	if err != nil {
+		return fmt.Errorf("Error writing to file: %v\n", err)
+	}
+
+	return nil
+}
+
+func getSkupperConfiguredImages(enableSHA bool) []SkupperImage {
+	return []SkupperImage{
+		{
+			Name:       images.GetRouterImageName(),
+			SHA:        getSHAIfEnabled(enableSHA, images.GetRouterImageName()),
+			Repository: "https://github.com/skupperproject/skupper-router",
+		},
+		{
+			Name:       images.GetServiceControllerImageName(),
+			SHA:        getSHAIfEnabled(enableSHA, images.GetServiceControllerImageName()),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       images.GetControllerPodmanImageName(),
+			SHA:        getSHAIfEnabled(enableSHA, images.GetControllerPodmanImageName()),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       images.GetConfigSyncImageName(),
+			SHA:        getSHAIfEnabled(enableSHA, images.GetConfigSyncImageName()),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       images.GetFlowCollectorImageName(),
+			SHA:        getSHAIfEnabled(enableSHA, images.GetFlowCollectorImageName()),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       images.GetSiteControllerImageName(),
+			SHA:        getSHAIfEnabled(enableSHA, images.GetSiteControllerImageName()),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name: images.GetPrometheusServerImageName(),
+			SHA:  getSHAIfEnabled(enableSHA, images.GetPrometheusServerImageName()),
+		},
+		{
+			Name: images.GetOauthProxyImageName(),
+			SHA:  getSHAIfEnabled(enableSHA, images.GetOauthProxyImageName()),
+		},
+	}
+}
+
+func getSkupperDefaultImages() []SkupperImage {
+	return []SkupperImage{
+		{
+			Name:       strings.Join([]string{images.DefaultImageRegistry, images.RouterImageName}, "/"),
+			Repository: "https://github.com/skupperproject/skupper-router",
+		},
+		{
+			Name:       strings.Join([]string{images.DefaultImageRegistry, images.ServiceControllerImageName}, "/"),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       strings.Join([]string{images.DefaultImageRegistry, images.ControllerPodmanImageName}, "/"),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       strings.Join([]string{images.DefaultImageRegistry, images.ConfigSyncImageName}, "/"),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       strings.Join([]string{images.DefaultImageRegistry, images.FlowCollectorImageName}, "/"),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       strings.Join([]string{images.DefaultImageRegistry, images.SiteControllerImageName}, "/"),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name: strings.Join([]string{images.PrometheusImageRegistry, images.PrometheusServerImageName}, "/"),
+		},
+		{
+			Name: strings.Join([]string{images.OauthProxyImageRegistry, images.OauthProxyImageName}, "/"),
+		},
+	}
+}
+
+func getEnvironmentVariableMap() *map[string]string {
+
+	envVariables := make(map[string]string)
+
+	skupperImageRegistry := os.Getenv(images.SkupperImageRegistryEnvKey)
+	if skupperImageRegistry != "" {
+		envVariables[images.SkupperImageRegistryEnvKey] = skupperImageRegistry
+	}
+
+	prometheusImageRegistry := os.Getenv(images.PrometheusImageRegistryEnvKey)
+	if prometheusImageRegistry != "" {
+		envVariables[images.PrometheusImageRegistry] = prometheusImageRegistry
+	}
+
+	oauthImageRegistry := os.Getenv(images.OauthProxyImageRegistry)
+	if oauthImageRegistry != "" {
+		envVariables[images.OauthProxyImageRegistry] = oauthImageRegistry
+	}
+
+	routerImage := os.Getenv(images.RouterImageEnvKey)
+	if routerImage != "" {
+		envVariables[images.RouterImageEnvKey] = routerImage
+	}
+
+	serviceControllerImage := os.Getenv(images.ServiceControllerImageEnvKey)
+	if serviceControllerImage != "" {
+		envVariables[images.ServiceControllerImageEnvKey] = serviceControllerImage
+	}
+
+	controllerPodmanImage := os.Getenv(images.ControllerPodmanImageEnvKey)
+	if controllerPodmanImage != "" {
+		envVariables[images.ControllerPodmanImageEnvKey] = controllerPodmanImage
+	}
+
+	configSyncImage := os.Getenv(images.ConfigSyncImageEnvKey)
+	if configSyncImage != "" {
+		envVariables[images.ConfigSyncImageEnvKey] = configSyncImage
+	}
+
+	flowCollectorImage := os.Getenv(images.FlowCollectorImageEnvKey)
+	if flowCollectorImage != "" {
+		envVariables[images.FlowCollectorImageEnvKey] = flowCollectorImage
+	}
+
+	prometheusImage := os.Getenv(images.PrometheusServerImageEnvKey)
+	if prometheusImage != "" {
+		envVariables[images.PrometheusServerImageEnvKey] = prometheusImage
+	}
+
+	oauthImage := os.Getenv(images.OauthProxyImageEnvKey)
+	if oauthImage != "" {
+		envVariables[images.OauthProxyImageEnvKey] = oauthImage
+	}
+
+	return &envVariables
+}
+
+func getSHAIfEnabled(enableSHA bool, imageName string) string {
+	if !enableSHA {
+		return ""
+	}
+	return images.GetSha(imageName)
+}


Test patch:
diff --git a/client/router_create_test.go b/client/router_create_test.go
index f60fff07..62245eed 100644
--- a/client/router_create_test.go
+++ b/client/router_create_test.go
@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/version"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
@@ -19,6 +20,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	fakediscovery "k8s.io/client-go/discovery/fake"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/tools/cache"
 )
@@ -1143,3 +1145,50 @@ func TestLabelandAnnotationOptions(t *testing.T) {
 		}
 	}
 }
+
+func TestAddPsa(t *testing.T) {
+	testcases := []struct {
+		major    string
+		minor    string
+		expected bool
+	}{
+		{
+			major:    "1",
+			minor:    "11",
+			expected: false,
+		},
+		{
+			major:    "1",
+			minor:    "23",
+			expected: false,
+		},
+		{
+			major:    "1",
+			minor:    "24",
+			expected: true,
+		},
+		{
+			major:    "1",
+			minor:    "25",
+			expected: true,
+		},
+		{
+			major:    "2",
+			minor:    "11",
+			expected: false,
+		},
+	}
+
+	cli, err := newMockClient("test-psa", "", "")
+	assert.Check(t, err, "test-psa")
+
+	for _, c := range testcases {
+		cli.KubeClient.Discovery().(*fakediscovery.FakeDiscovery).FakedServerVersion = &version.Info{
+			Major: c.major,
+			Minor: c.minor,
+		}
+		sv, err := cli.KubeClient.Discovery().ServerVersion()
+		assert.Check(t, err)
+		assert.Equal(t, addPsa(sv), c.expected, c.major+"."+c.minor)
+	}
+}
diff --git a/cmd/skupper/skupper_mock_test.go b/cmd/skupper/skupper_mock_test.go
index fa688b2c..1f0918f6 100644
--- a/cmd/skupper/skupper_mock_test.go
+++ b/cmd/skupper/skupper_mock_test.go
@@ -319,10 +319,6 @@ func (v *vanClientMock) NetworkStatus(ctx context.Context) (*network.NetworkStat
 	return &result, nil
 }
 
-func (v *vanClientMock) GetRemoteLinks(ctx context.Context, siteConfig *types.SiteConfig) ([]*network.RemoteLinkInfo, error) {
-	return nil, nil
-}
-
 func (v *vanClientMock) GetConsoleUrl(namespace string) (string, error) {
 	return "", nil
 }
diff --git a/cmd/skupper/skupper_podman_test.go b/cmd/skupper/skupper_podman_test.go
new file mode 100644
index 00000000..eeefc078
--- /dev/null
+++ b/cmd/skupper/skupper_podman_test.go
@@ -0,0 +1,177 @@
+package main
+
+import (
+	"bytes"
+	"fmt"
+	"strings"
+	"testing"
+
+	clientpodman "github.com/skupperproject/skupper/client/podman"
+	"github.com/skupperproject/skupper/pkg/domain"
+	"github.com/skupperproject/skupper/pkg/domain/podman"
+	"github.com/spf13/cobra"
+	"gotest.tools/assert"
+)
+
+func TestNewClient(t *testing.T) {
+
+	const didNotExit = -99
+	tests := []struct {
+		scenario           string
+		command            string
+		args               []string
+		cliFactory         clientpodman.RestClientFactory
+		siteHandlerFactory podman.SiteHandlerFactory
+		currentSiteLoaded  bool
+		exitCode           int
+		stdoutContains     string
+	}{
+		{
+			scenario: "init-invalid-endpoint",
+			command:  "init",
+			args:     []string{"/invalid/podman/endpoint"},
+			cliFactory: func(endpoint, basePath string) (*clientpodman.PodmanRestClient, error) {
+				return nil, fmt.Errorf("Podman endpoint is not available: %s", endpoint)
+			},
+			exitCode:       1,
+			stdoutContains: "Recommendation:",
+		},
+		{
+			scenario: "init-sitehandler-factory-error",
+			command:  "init",
+			cliFactory: func(endpoint, basePath string) (*clientpodman.PodmanRestClient, error) {
+				return &clientpodman.PodmanRestClient{}, nil
+			},
+			siteHandlerFactory: func(endpoint string) (domain.SiteHandler, error) {
+				return nil, fmt.Errorf("unable to get sitehandler instance [mock]")
+			},
+			exitCode:       1,
+			stdoutContains: "error verifying existing skupper site",
+		},
+		{
+			scenario: "init-already-exists",
+			command:  "init",
+			cliFactory: func(endpoint, basePath string) (*clientpodman.PodmanRestClient, error) {
+				return &clientpodman.PodmanRestClient{}, nil
+			},
+			siteHandlerFactory: func(endpoint string) (domain.SiteHandler, error) {
+				siteHandler := &siteHandlerMock{
+					getHook: func() (domain.Site, error) {
+						return &podman.Site{}, nil
+					},
+				}
+				return siteHandler, nil
+			},
+			currentSiteLoaded: false,
+			exitCode:          0,
+			stdoutContains:    "Skupper has already been initialized for user",
+		},
+		{
+			scenario: "status-ok",
+			command:  "status",
+			cliFactory: func(endpoint, basePath string) (*clientpodman.PodmanRestClient, error) {
+				return &clientpodman.PodmanRestClient{}, nil
+			},
+			siteHandlerFactory: func(endpoint string) (domain.SiteHandler, error) {
+				siteHandler := &siteHandlerMock{
+					getHook: func() (domain.Site, error) {
+						return &podman.Site{}, nil
+					},
+				}
+				return siteHandler, nil
+			},
+			currentSiteLoaded: true,
+			exitCode:          didNotExit,
+		},
+		{
+			scenario: "status-not-enabled",
+			command:  "status",
+			cliFactory: func(endpoint, basePath string) (*clientpodman.PodmanRestClient, error) {
+				return &clientpodman.PodmanRestClient{}, nil
+			},
+			siteHandlerFactory: func(endpoint string) (domain.SiteHandler, error) {
+				siteHandler := &siteHandlerMock{
+					getHook: func() (domain.Site, error) {
+						return nil, fmt.Errorf("skupper is not enabled")
+					},
+				}
+				return siteHandler, nil
+			},
+			currentSiteLoaded: false,
+			stdoutContains:    "Skupper is not enabled for user",
+			exitCode:          0,
+		},
+		{
+			scenario: "version-not-enabled",
+			command:  "version",
+			cliFactory: func(endpoint, basePath string) (*clientpodman.PodmanRestClient, error) {
+				return &clientpodman.PodmanRestClient{}, nil
+			},
+			siteHandlerFactory: func(endpoint string) (domain.SiteHandler, error) {
+				siteHandler := &siteHandlerMock{
+					getHook: func() (domain.Site, error) {
+						return nil, fmt.Errorf("skupper is not enabled")
+					},
+				}
+				return siteHandler, nil
+			},
+			exitCode: didNotExit,
+		},
+	}
+
+	var s *SkupperPodman
+	var lastExitCode int
+	stdout := new(bytes.Buffer)
+	resetSkupperPodman := func() {
+		s = &SkupperPodman{}
+		s.exit = func(code int) {
+			lastExitCode = code
+		}
+		s.output = stdout
+		lastExitCode = didNotExit
+	}
+
+	for _, test := range tests {
+		t.Run(test.scenario, func(t *testing.T) {
+			// reset test status
+			resetSkupperPodman()
+			// set factories
+			s.cliFactory = test.cliFactory
+			s.siteHandlerFactory = test.siteHandlerFactory
+			s.NewClient(&cobra.Command{
+				Use: test.command,
+			}, test.args)
+			assert.Equal(t, lastExitCode, test.exitCode)
+			assert.Assert(t, strings.Contains(stdout.String(), test.stdoutContains))
+			siteLoaded := s.currentSite != nil
+			assert.Assert(t, siteLoaded == test.currentSiteLoaded, "expected site to be loaded? %v", test.currentSiteLoaded)
+		})
+	}
+}
+
+type siteHandlerMock struct {
+	getHook func() (domain.Site, error)
+}
+
+func (sh *siteHandlerMock) Create(s domain.Site) error {
+	return fmt.Errorf("not implemented")
+}
+
+func (sh *siteHandlerMock) Get() (domain.Site, error) {
+	if sh.getHook == nil {
+		return nil, fmt.Errorf("not implemented")
+	}
+	return sh.getHook()
+}
+
+func (sh *siteHandlerMock) Delete() error {
+	return fmt.Errorf("not implemented")
+}
+
+func (sh *siteHandlerMock) Update() error {
+	return fmt.Errorf("not implemented")
+}
+
+func (sh *siteHandlerMock) RevokeAccess() error {
+	return fmt.Errorf("not implemented")
+}
diff --git a/pkg/flow/collector_test.go b/pkg/flow/collector_test.go
index 3a7ce08a..8f024506 100644
--- a/pkg/flow/collector_test.go
+++ b/pkg/flow/collector_test.go
@@ -1,12 +1,15 @@
 package flow
 
 import (
+	"fmt"
 	"reflect"
 	"strconv"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/skupperproject/skupper/pkg/messaging"
 	"gotest.tools/assert"
 )
 
@@ -59,54 +62,197 @@ func TestUpdates(t *testing.T) {
 }
 
 func TestRecordUpdates(t *testing.T) {
-	reg := prometheus.NewRegistry()
-	u, _ := time.ParseDuration("5m")
+	factory := messaging.NewMockConnectionFactory(t, "mockamqp://local")
 	fc := NewFlowCollector(FlowCollectorSpec{
 		Mode:              RecordMetrics,
 		Origin:            "origin",
-		PromReg:           reg,
-		ConnectionFactory: nil,
-		FlowRecordTtl:     u,
+		PromReg:           prometheus.NewRegistry(),
+		ConnectionFactory: factory,
+		FlowRecordTtl:     time.Minute * 5,
 	})
-	fc.metrics = fc.NewMetrics(fc.prometheusReg)
 
-	assert.Assert(t, fc != nil)
 	stopCh := make(chan struct{})
-	go fc.beaconUpdates(stopCh)
-	go fc.recordUpdates(stopCh)
+	fc.Start(stopCh)
+	factory.Broker.AwaitReceivers(BeaconAddress, 1)
+
+	conn, err := factory.Connect()
+	assert.Assert(t, err)
+	beaconSender, err := conn.Sender(BeaconAddress)
+	assert.Assert(t, err)
+	flushReceiver, err := conn.Receiver("sfe.1234", 32)
+	assert.Assert(t, err)
 
 	beacon := BeaconRecord{
 		Version:    1,
-		SourceType: "CONTROLLER",
+		SourceType: "ROUTER",
 		Address:    RecordPrefix + "1234",
 		Direct:     "sfe.1234",
 		Identity:   "1234",
 	}
-
-	var beacons []interface{}
-	beacons = append(beacons, beacon)
-	fc.beaconsIncoming <- beacons
-	time.Sleep(1 * time.Second)
-	eventSource, ok := fc.eventSources["1234"]
-	assert.Assert(t, ok)
-	assert.Equal(t, eventSource.Beacons, 1)
-	fc.beaconsIncoming <- beacons
-	time.Sleep(1 * time.Second)
-	assert.Equal(t, eventSource.Beacons, 2)
+	msgBeacon, err := encodeBeacon(&beacon)
+	assert.Assert(t, err)
+	assert.Assert(t, beaconSender.Send(msgBeacon))
+	assert.Assert(t, beaconSender.Send(msgBeacon))
 
 	heartbeat := HeartbeatRecord{
 		Version:  1,
 		Identity: "1234",
 		Source:   "sfe.1234",
 	}
+	msgHB, err := encodeHeartbeat(&heartbeat)
+	assert.Assert(t, err)
 
-	var heartbeats []interface{}
-	heartbeats = append(heartbeats, heartbeat)
-	fc.heartbeatsIncoming <- heartbeats
-	time.Sleep(1 * time.Second)
-	assert.Equal(t, eventSource.Heartbeats, 1)
+	factory.Broker.AwaitReceivers("mc/sfe.1234", 1)
+	heartBeatSender, err := conn.Sender("mc/sfe.1234")
+	heartBeatSender.Send(msgHB)
 
-	time.Sleep(1 * time.Second)
+	// we don't have a safe way to access the internal state of the collector
+	// without stopping it (or standing up a stub http mux). Instead, we'll
+	// wait for a flush record so we know that the internal state is settled
+	// before stopping it and cracking things open.
+	msg, err := flushReceiver.Receive()
+	assert.Assert(t, err)
+	assert.Equal(t, msg.Properties.Subject, "FLUSH")
 
 	close(stopCh)
+	<-fc.beaconReceiver.done // only way to tell the flow collector is done
+
+	eventSource, ok := fc.eventSources["1234"]
+	t.Log(fc.eventSources)
+	assert.Assert(t, ok)
+	assert.Equal(t, eventSource.Beacons, 2)
+	assert.Equal(t, eventSource.Heartbeats, 1)
+}
+
+func TestStartupShutdown(t *testing.T) {
+	scenarios := []struct {
+		Name        string
+		Spec        FlowCollectorSpec
+		Controllers int
+	}{
+		{
+			Name:        "RecordMetrics basic",
+			Controllers: 0,
+			Spec: FlowCollectorSpec{
+				Mode:              RecordMetrics,
+				Origin:            "origin",
+				PromReg:           prometheus.NewRegistry(),
+				ConnectionFactory: messaging.NewMockConnectionFactory(t, "local://test"),
+				FlowRecordTtl:     5 * time.Minute,
+			},
+		}, {
+			Name:        "RecordMetrics four controllers",
+			Controllers: 4,
+			Spec: FlowCollectorSpec{
+				Mode:              RecordMetrics,
+				Origin:            "origin",
+				PromReg:           prometheus.NewRegistry(),
+				ConnectionFactory: messaging.NewMockConnectionFactory(t, "local://test"),
+				FlowRecordTtl:     5 * time.Minute,
+			},
+		}, {
+			Name:        "RecordStatus basic",
+			Controllers: 0,
+			Spec: FlowCollectorSpec{
+				Mode:              RecordStatus,
+				Origin:            "origin",
+				PromReg:           prometheus.NewRegistry(),
+				ConnectionFactory: messaging.NewMockConnectionFactory(t, "local://test"),
+				FlowRecordTtl:     5 * time.Minute,
+			},
+		}, {
+			Name:        "RecordStatus many controllers",
+			Controllers: 128,
+			Spec: FlowCollectorSpec{
+				Mode:              RecordStatus,
+				Origin:            "origin",
+				PromReg:           prometheus.NewRegistry(),
+				ConnectionFactory: messaging.NewMockConnectionFactory(t, "local://test"),
+				FlowRecordTtl:     5 * time.Minute,
+			},
+		},
+	}
+	for _, _testspec := range scenarios {
+		s := _testspec
+		t.Run(s.Name, func(t *testing.T) {
+			t.Parallel()
+			fc := NewFlowCollector(s.Spec)
+
+			done := make(chan struct{})
+
+			conn, err := s.Spec.ConnectionFactory.Connect()
+			assert.Assert(t, err)
+
+			flushReceivers := []messaging.Receiver{}
+			for site := 1; site < s.Controllers; site++ {
+
+				r, err := conn.Receiver(fmt.Sprintf("sfe.%d", site), 128)
+				assert.Assert(t, err)
+				flushReceivers = append(flushReceivers, r)
+
+				// simulated controller - beacons+heartbeats
+				go runMockCollector(t, strconv.Itoa(site), conn, done)
+			}
+			// start flow collector
+			fc.Start(done)
+			mockFactory := s.Spec.ConnectionFactory.(*messaging.MockConnectionFactory)
+			mockFactory.Broker.AwaitReceivers(BeaconAddress, 1)
+
+			// wait for flushes
+			var allFlushed sync.WaitGroup
+			for _, r := range flushReceivers {
+				allFlushed.Add(1)
+				go func(receiver messaging.Receiver) {
+					defer allFlushed.Done()
+					msg, err := receiver.Receive()
+					assert.Assert(t, err)
+					assert.Equal(t, msg.Properties.Subject, "FLUSH")
+				}(r)
+			}
+			allFlushed.Wait()
+			close(done)
+
+			// last thing FlowCollector does on stop is close the beaconReceiver
+			// make sure that is done
+			<-fc.beaconReceiver.done
+
+		})
+	}
+}
+func runMockCollector(t *testing.T, siteID string, conn messaging.Connection, done chan struct{}) {
+	ticker := time.NewTicker(time.Second / 15)
+	defer ticker.Stop()
+	mcAddr := fmt.Sprintf("mc/sfe.%s", siteID)
+	directAddr := fmt.Sprintf("sfe.%s", siteID)
+	heartSender, err := conn.Sender(mcAddr + ".heartbeats")
+	assert.Assert(t, err)
+	beaconSender, err := conn.Sender(BeaconAddress)
+	assert.Assert(t, err)
+	beacon := BeaconRecord{
+		Version:    1,
+		SourceType: "CONTROLLER",
+		Address:    mcAddr,
+		Direct:     directAddr,
+		Identity:   siteID,
+	}
+	msgBeacon, err := encodeBeacon(&beacon)
+	assert.Assert(t, err)
+	heartbeat := HeartbeatRecord{
+		Version:  1,
+		Identity: siteID,
+		Source:   directAddr,
+	}
+	assert.Assert(t, err)
+	for {
+		select {
+		case <-done:
+			return
+		case <-ticker.C:
+			assert.Assert(t, beaconSender.Send(msgBeacon))
+			heartbeat.Now = uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
+			msgHeart, err := encodeHeartbeat(&heartbeat)
+			assert.Assert(t, err)
+			assert.Assert(t, heartSender.Send(msgHeart))
+		}
+	}
 }
diff --git a/pkg/flow/controller_test.go b/pkg/flow/controller_test.go
index 90e136f5..27adcbaf 100644
--- a/pkg/flow/controller_test.go
+++ b/pkg/flow/controller_test.go
@@ -145,10 +145,7 @@ func TestUpdateRecords(t *testing.T) {
 	var flushes []interface{}
 	flushes = append(flushes, flush)
 	fc.flushIncoming <- flushes
-
-	time.Sleep(5 * time.Second)
 	close(stopCh)
-
 }
 
 func TestSiteController(t *testing.T) {
diff --git a/pkg/network/network_test.go b/pkg/network/network_test.go
index 93a443e1..1727f7be 100644
--- a/pkg/network/network_test.go
+++ b/pkg/network/network_test.go
@@ -53,7 +53,7 @@ func TestGetSiteTargetsMap(t *testing.T) {
 	expectedSite2 := "429c2780-003d-44cc-9a91-4139885c7d20"
 	expectedService := "backend:8080"
 
-	result := skupperStatus.GetSiteTargetsMap()
+	result := skupperStatus.GetSiteTargetMap()
 
 	assert.Check(t, result[expectedSite2] == nil)
 	assert.Check(t, result[expectedSite1][expectedService].Address == expectedService)
diff --git a/pkg/service_sync/messaging_test.go b/pkg/service_sync/messaging_test.go
index 4ab925c8..bab9c736 100644
--- a/pkg/service_sync/messaging_test.go
+++ b/pkg/service_sync/messaging_test.go
@@ -1,11 +1,9 @@
 package service_sync
 
 import (
-	"fmt"
 	"reflect"
 	"testing"
 
-	amqp "github.com/interconnectedcloud/go-amqp"
 	"gotest.tools/assert"
 
 	"github.com/skupperproject/skupper/api/types"
@@ -13,173 +11,6 @@ import (
 	"github.com/skupperproject/skupper/pkg/messaging"
 )
 
-type multicast struct {
-	receivers []*MockReceiver
-	channel   chan int
-}
-
-func newMulticast() *multicast {
-	return &multicast{
-		channel: make(chan int, 5),
-	}
-}
-
-func (m *multicast) send(msg *amqp.Message) {
-	for _, r := range m.receivers {
-		r.send(msg)
-	}
-}
-
-func (m *multicast) subscribe(r *MockReceiver) {
-	m.receivers = append(m.receivers, r)
-	m.channel <- len(m.receivers)
-}
-
-func (m *multicast) waitForReceivers(count int) {
-	for len(m.receivers) < count {
-		<-m.channel
-	}
-}
-
-type broker struct {
-	topics map[string]*multicast
-}
-
-func newBroker() *broker {
-	return &broker{
-		topics: map[string]*multicast{},
-	}
-}
-
-func (b *broker) newTopic(address string) *multicast {
-	t := newMulticast()
-	b.topics[address] = t
-	return t
-}
-
-func (b *broker) subscribe(address string, r *MockReceiver) {
-	t, ok := b.topics[address]
-	if !ok {
-		t = newMulticast()
-		b.topics[address] = t
-	}
-	t.subscribe(r)
-}
-
-func (b *broker) send(address string, msg *amqp.Message) {
-	if t, ok := b.topics[address]; ok {
-		t.send(msg)
-	}
-}
-
-type MockConnectionFactory struct {
-	url    string
-	topics *broker
-}
-
-func NewMockConnectionFactory(url string) *MockConnectionFactory {
-	return &MockConnectionFactory{
-		url:    url,
-		topics: newBroker(),
-	}
-}
-
-func (f *MockConnectionFactory) Connect() (messaging.Connection, error) {
-	c := &MockConnection{topics: f.topics, channel: make(chan bool)}
-	return c, nil
-}
-
-func (f *MockConnectionFactory) Url() string {
-	return f.url
-}
-
-type MockConnection struct {
-	topics  *broker
-	channel chan bool
-	closed  bool
-}
-
-func (c *MockConnection) Sender(address string) (messaging.Sender, error) {
-	return &MockSender{address: address, connection: c}, nil
-}
-
-func (c *MockConnection) Receiver(address string, credit uint32) (messaging.Receiver, error) {
-	r := &MockReceiver{connection: c, channel: make(chan *amqp.Message, credit)}
-	c.topics.subscribe(address, r)
-	return r, nil
-}
-
-func (c *MockConnection) Close() {
-	if !c.closed {
-		c.closed = true
-		close(c.channel)
-	}
-}
-
-func (c *MockConnection) send(address string, msg *amqp.Message) error {
-	if c.closed {
-		return fmt.Errorf("Channel closed")
-	}
-	c.topics.send(address, msg)
-	return nil
-}
-
-func (c *MockConnection) receive(r *MockReceiver) (*amqp.Message, error) {
-	if c.closed {
-		return nil, fmt.Errorf("Channel closed")
-	}
-	select {
-	case msg := <-r.channel:
-		if msg == nil {
-			return nil, fmt.Errorf("Failed to receive")
-		}
-		return msg, nil
-	case <-c.channel:
-		return nil, fmt.Errorf("Channel closed")
-	}
-}
-
-type MockSender struct {
-	connection *MockConnection
-	address    string
-}
-
-func (s *MockSender) Send(msg *amqp.Message) error {
-	return s.connection.send(s.address, msg)
-}
-
-func (s *MockSender) Close() error {
-	return nil
-}
-
-type MockReceiver struct {
-	connection *MockConnection
-	channel    chan *amqp.Message
-	closed     bool
-}
-
-func (r *MockReceiver) send(msg *amqp.Message) {
-	if !r.closed {
-		r.channel <- msg
-	}
-}
-
-func (r *MockReceiver) Receive() (*amqp.Message, error) {
-	return r.connection.receive(r)
-}
-
-func (r *MockReceiver) Accept(msg *amqp.Message) error {
-	return nil
-}
-
-func (r *MockReceiver) Close() error {
-	if !r.closed {
-		close(r.channel)
-		r.closed = true
-	}
-	return nil
-}
-
 func TestSender(t *testing.T) {
 	stopper := make(chan struct{})
 	event.StartDefaultEventStore(stopper)
@@ -234,7 +65,7 @@ func TestSender(t *testing.T) {
 
 	for _, s := range scenarios {
 		t.Run(s.name, func(t *testing.T) {
-			factory := NewMockConnectionFactory("test-channel")
+			factory := messaging.NewMockConnectionFactory(t, "test-channel")
 			outgoing := make(chan ServiceUpdate)
 			sender := newSender(factory, outgoing, event.NewDefaultEventLogger())
 			sender.start()
@@ -326,16 +157,21 @@ func TestReceiver(t *testing.T) {
 
 	for _, s := range scenarios {
 		t.Run(s.name, func(t *testing.T) {
-			factory := NewMockConnectionFactory("test-channel")
+			factory := messaging.NewMockConnectionFactory(t, "test-channel")
 			incoming := make(chan ServiceUpdate)
 			receiver := newReceiver(factory, incoming, event.NewDefaultEventLogger())
 			receiver.start()
-			factory.topics.newTopic(ServiceSyncAddress).waitForReceivers(1)
+			factory.Broker.AwaitReceivers(ServiceSyncAddress, 1)
+
+			tConn, err := factory.Connect()
+			assert.Assert(t, err)
+			serviceSyncSender, err := tConn.Sender(ServiceSyncAddress)
+			assert.Assert(t, err)
 
 			for _, update := range s.updates {
 				msg, err := encode(&update)
 				assert.Assert(t, err)
-				factory.topics.send(ServiceSyncAddress, msg)
+				serviceSyncSender.Send(msg)
 				received := <-incoming
 
 				assert.Assert(t, err)
diff --git a/pkg/service_sync/service_sync_test.go b/pkg/service_sync/service_sync_test.go
index 79445019..893269fd 100644
--- a/pkg/service_sync/service_sync_test.go
+++ b/pkg/service_sync/service_sync_test.go
@@ -1,17 +1,19 @@
 package service_sync
 
 import (
-	vanClient "github.com/skupperproject/skupper/client"
-	"k8s.io/client-go/kubernetes/fake"
 	"reflect"
 	"strings"
 	"testing"
 	"time"
 
+	vanClient "github.com/skupperproject/skupper/client"
+	"k8s.io/client-go/kubernetes/fake"
+
 	"gotest.tools/assert"
 
 	"github.com/skupperproject/skupper/api/types"
 	"github.com/skupperproject/skupper/pkg/event"
+	"github.com/skupperproject/skupper/pkg/messaging"
 )
 
 type updateRecord struct {
@@ -126,14 +128,14 @@ func TestServiceSync(t *testing.T) {
 		event.StartDefaultEventStore(stopper)
 		t.Run(s.name, func(t *testing.T) {
 			//hook up two instances of service sync
-			factory := NewMockConnectionFactory("test-channel")
+			factory := messaging.NewMockConnectionFactory(t, "test-channel")
 			updates1 := newUpdateChannel()
 			updates2 := newUpdateChannel()
 			site1 := NewServiceSync("foo", 0, "v1", factory, updates1.handler, event.NewDefaultEventLogger())
 			site2 := NewServiceSync("bar", 0, "v1", factory, updates2.handler, event.NewDefaultEventLogger())
 			site1.Start(stopper)
 			site2.Start(stopper)
-			factory.topics.newTopic(ServiceSyncAddress).waitForReceivers(2)
+			factory.Broker.AwaitReceivers(ServiceSyncAddress, 2)
 			site1.LocalDefinitionsUpdated(s.site1.definitions)
 			site2.LocalDefinitionsUpdated(s.site2.definitions)
 			site1Done := false
@@ -184,7 +186,7 @@ func TestRemoveStaleDefinitions(t *testing.T) {
 	event.StartDefaultEventStore(stopper)
 
 	updates := newUpdateCollector()
-	factory := NewMockConnectionFactory("test-channel")
+	factory := messaging.NewMockConnectionFactory(t, "test-channel")
 	site := NewServiceSync("foo", 0, "v1", factory, updates.handler, event.NewDefaultEventLogger())
 
 	defs := map[string]types.ServiceInterface{
@@ -231,7 +233,7 @@ func TestUpdateRemoteDefinitions(t *testing.T) {
 	event.StartDefaultEventStore(stopper)
 
 	updates := newUpdateCollector()
-	factory := NewMockConnectionFactory("test-channel")
+	factory := messaging.NewMockConnectionFactory(t, "test-channel")
 	site := NewServiceSync("foo", 0, "v1", factory, updates.handler, event.NewDefaultEventLogger())
 
 	defs := map[string]types.ServiceInterface{
@@ -295,7 +297,7 @@ func TestLocalDefinitionsUpdated(t *testing.T) {
 	event.StartDefaultEventStore(stopper)
 
 	updates := newUpdateCollector()
-	factory := NewMockConnectionFactory("test-channel")
+	factory := messaging.NewMockConnectionFactory(t, "test-channel")
 	site := NewServiceSync("foo", 0, "v1", factory, updates.handler, event.NewDefaultEventLogger())
 
 	defs := map[string]types.ServiceInterface{
diff --git a/pkg/utils/configs/manifest_test.go b/pkg/utils/configs/manifest_test.go
new file mode 100644
index 00000000..c1bed966
--- /dev/null
+++ b/pkg/utils/configs/manifest_test.go
@@ -0,0 +1,137 @@
+package configs
+
+import (
+	"fmt"
+	"github.com/skupperproject/skupper/pkg/images"
+	"gotest.tools/assert"
+	"os"
+	"strings"
+	"testing"
+)
+
+func TestManifestManager(t *testing.T) {
+	testcases := []struct {
+		title                          string
+		envVariablesWithValue          []string
+		expectedConfiguredManifest     Manifest
+		expectedDefaultManifestWithEnv Manifest
+	}{
+		{
+			title: "configured manifest has different images that the default manifest",
+			envVariablesWithValue: []string{
+				images.SkupperImageRegistryEnvKey,
+				images.ConfigSyncImageEnvKey,
+				images.RouterImageEnvKey},
+			expectedConfiguredManifest: Manifest{
+				Images: []SkupperImage{
+					{
+						Name: "SKUPPER_CONFIG_SYNC_IMAGE_TESTING",
+					},
+					{
+						Name: "QDROUTERD_IMAGE_TESTING"},
+				},
+			},
+			expectedDefaultManifestWithEnv: Manifest{
+				Images: []SkupperImage{
+					{
+						Name: strings.Join([]string{images.DefaultImageRegistry, images.ConfigSyncImageName}, "/"),
+					},
+					{
+						Name: strings.Join([]string{images.DefaultImageRegistry, images.RouterImageName}, "/"),
+					},
+				},
+				Variables: &map[string]string{
+					images.SkupperImageRegistryEnvKey: "SKUPPER_IMAGE_REGISTRY_TESTING",
+					images.ConfigSyncImageEnvKey:      "SKUPPER_CONFIG_SYNC_IMAGE_TESTING",
+					images.RouterImageEnvKey:          "QDROUTERD_IMAGE_TESTING",
+				},
+			},
+		},
+		{
+			title:                 "configured manifest the same images that the default manifest",
+			envVariablesWithValue: []string{},
+			expectedConfiguredManifest: Manifest{
+				Images: []SkupperImage{
+					{
+						Name: strings.Join([]string{images.DefaultImageRegistry, images.ConfigSyncImageName}, "/"),
+					},
+					{
+						Name: strings.Join([]string{images.DefaultImageRegistry, images.RouterImageName}, "/"),
+					},
+				},
+			},
+			expectedDefaultManifestWithEnv: Manifest{
+				Images: []SkupperImage{
+					{
+						Name: strings.Join([]string{images.DefaultImageRegistry, images.ConfigSyncImageName}, "/"),
+					},
+					{
+						Name: strings.Join([]string{images.DefaultImageRegistry, images.RouterImageName}, "/"),
+					},
+				},
+				Variables: &map[string]string{},
+			},
+		},
+	}
+
+	for _, c := range testcases {
+
+		setUpEnvVariables(c.envVariablesWithValue)
+
+		manifestManager := ManifestManager{EnableSHA: false}
+		configuredManifest := manifestManager.GetConfiguredManifest()
+		defaultManifest := manifestManager.GetDefaultManifestWithEnv()
+
+		for _, expectedImage := range c.expectedConfiguredManifest.Images {
+
+			configuredImage := getSkupperImageFromManifest(&configuredManifest, expectedImage.Name)
+
+			assert.Check(t, configuredImage != nil)
+		}
+
+		assert.Check(t, configuredManifest.Variables == nil)
+
+		for _, expectedImage := range c.expectedDefaultManifestWithEnv.Images {
+
+			defaultImage := getSkupperImageFromManifest(&defaultManifest, expectedImage.Name)
+
+			assert.Check(t, defaultImage != nil)
+		}
+
+		assert.DeepEqual(t, c.expectedDefaultManifestWithEnv.Variables, defaultManifest.Variables)
+
+		clearEnvVariables(c.envVariablesWithValue)
+	}
+}
+
+func setUpEnvVariables(variables []string) {
+	for _, variable := range variables {
+		err := os.Setenv(variable, strings.Join([]string{variable, "TESTING"}, "_"))
+		if err != nil {
+			fmt.Printf("error while setting %s", variable)
+			return
+		}
+	}
+}
+
+func clearEnvVariables(variables []string) {
+	for _, variable := range variables {
+		err := os.Unsetenv(variable)
+		if err != nil {
+			fmt.Printf("error while unssetting %s", variable)
+			return
+		}
+	}
+}
+
+func getSkupperImageFromManifest(m *Manifest, name string) *SkupperImage {
+
+	for _, skImage := range m.Images {
+
+		if skImage.Name == name {
+			return &skImage
+		}
+	}
+
+	return nil
+}
diff --git a/test/integration/acceptance/custom/basic/basic.go b/test/integration/acceptance/custom/basic/basic.go
index 1f240e0a..43824398 100644
--- a/test/integration/acceptance/custom/basic/basic.go
+++ b/test/integration/acceptance/custom/basic/basic.go
@@ -128,7 +128,7 @@ func (r *BasicTestRunner) Setup(ctx context.Context, createOptsPublic types.Site
 			"StartTimeAfter=", podStartTimeAfter, "Router component restarted - POD status:", string(podStatus))
 
 		// Check if the Volume is shared by both containers
-		podContainers, _ := kube.GetReadyPod(prv1Cluster.Namespace, prv1Cluster.VanClient.KubeClient, "router")
+		podContainers, _ := kube.GetReadyPod(prv1Cluster.Namespace, prv1Cluster.VanClient.KubeClient, "router", "skupper-router")
 		for _, container := range podContainers.Spec.Containers {
 			foundCertVol := false
 			for _, contVolume := range container.VolumeMounts {
diff --git a/test/integration/acceptance/custom/hello_policy/issues_test.go b/test/integration/acceptance/custom/hello_policy/issues_test.go
index 1920f5c8..c6523676 100644
--- a/test/integration/acceptance/custom/hello_policy/issues_test.go
+++ b/test/integration/acceptance/custom/hello_policy/issues_test.go
@@ -26,7 +26,7 @@ import (
 //
 // The pod name comes from kube.GetReadyPod for service-controller
 func seekServiceControllerAndDelete(ctx *base.ClusterContext) {
-	pod, err := kube.GetReadyPod(ctx.Namespace, ctx.VanClient.KubeClient, "service-controller")
+	pod, err := kube.GetReadyPod(ctx.Namespace, ctx.VanClient.KubeClient, "service-controller", "")
 	if err != nil {
 		log.Printf("Ignoring pod listing error '%v'", err)
 		return
diff --git a/test/integration/performance/common/setup.go b/test/integration/performance/common/setup.go
index 5d60977e..67817196 100644
--- a/test/integration/performance/common/setup.go
+++ b/test/integration/performance/common/setup.go
@@ -404,7 +404,7 @@ func tailRouterLogs(ctx context.Context, saveLogs *bool) *sync.WaitGroup {
 		var err error
 		log.Printf("Waiting for router pod to be running")
 		err = pkgutils.RetryWithContext(ctx, time.Second*5, func() (bool, error) {
-			pod, err = kube.GetReadyPod(cli.Namespace, cli.KubeClient, "router")
+			pod, err = kube.GetReadyPod(cli.Namespace, cli.KubeClient, "router", "skupper-router")
 			if pod != nil && err == nil {
 				return true, nil
 			}