Repository: skupperproject/skupper
Pull request #1245 (2023-10-02 15:35:20)

unexpose command: Fix error message
> The skupper unexpose command shows text for the expose command.

```
# skupper unexpose
Error: expose target and name must be specified (e.g. 'skupper expose deployment <name>')
Usage:
  skupper unexpose [flags]Flags:
      --address string   Skupper address the target was exposed as
  -h, --help             help for unexposeGlobal Flags:
  -c, --context string      The kubeconfig context to use
      --kubeconfig string   Path to the kubeconfig file to use
  -n, --namespace string    The Kubernetes namespace to use
      --platform string     The platform type to use [kubernetes, podman]# skupper version
client version                 1.4.1 
```


Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index dba8ceb8..092a0174 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -104,7 +104,7 @@ commands:
           command: sudo rm -rf /usr/local/go
       - docker/install-docker
       - go/install:
-          version: "1.19"
+          version: "1.20"
       - kube-orb/install-kubectl
       - run: make
 
@@ -412,7 +412,7 @@ jobs:
   test:
     executor:
       name: go/default
-      tag: "1.19"
+      tag: "1.20"
     steps:
       - checkout
       - run:
@@ -458,7 +458,7 @@ jobs:
   build-all:
     executor:
       name: go/default
-      tag: "1.19"
+      tag: "1.20"
     steps:
       - checkout
       - go/mod-download-cached
@@ -542,7 +542,7 @@ jobs:
   generate-manifest:
     executor:
       name: go/default
-      tag: "1.19"
+      tag: "1.20"
     steps:
       - checkout
       - go/mod-download-cached
@@ -582,7 +582,7 @@ jobs:
   publish-github-release-images:
     executor:
       name: go/default
-      tag: "1.19"
+      tag: "1.20"
     steps:
       - checkout
       - setup_remote_docker
@@ -603,7 +603,7 @@ jobs:
   publish-github-main-images:
     executor:
       name: go/default
-      tag: "1.19"
+      tag: "1.20"
     steps:
       - checkout
       - setup_remote_docker
diff --git a/Dockerfile.config-sync b/Dockerfile.config-sync
index 62c74581..19331e4b 100644
--- a/Dockerfile.config-sync
+++ b/Dockerfile.config-sync
@@ -1,4 +1,4 @@
-FROM golang:1.19 AS builder
+FROM golang:1.20 AS builder
 
 WORKDIR /go/src/app
 
diff --git a/Dockerfile.controller-podman b/Dockerfile.controller-podman
index 9d959e27..6e687450 100644
--- a/Dockerfile.controller-podman
+++ b/Dockerfile.controller-podman
@@ -1,4 +1,4 @@
-FROM golang:1.19 AS builder
+FROM golang:1.20 AS builder
 
 WORKDIR /go/src/app
 COPY go.mod .
diff --git a/Dockerfile.flow-collector b/Dockerfile.flow-collector
index 7730f22c..5d93660d 100644
--- a/Dockerfile.flow-collector
+++ b/Dockerfile.flow-collector
@@ -1,4 +1,4 @@
-FROM golang:1.19 AS builder
+FROM golang:1.20 AS builder
 
 WORKDIR /go/src/app
 
@@ -10,7 +10,7 @@ COPY . .
 
 RUN make
 
-FROM node:16.17 AS console-builder
+FROM node:18.18.2 AS console-builder
 
 WORKDIR /skupper-console/
 ADD https://github.com/skupperproject/skupper-console/archive/main.tar.gz .
diff --git a/Dockerfile.service-controller b/Dockerfile.service-controller
index f4b42946..45c31d31 100644
--- a/Dockerfile.service-controller
+++ b/Dockerfile.service-controller
@@ -1,4 +1,4 @@
-FROM golang:1.19 AS builder
+FROM golang:1.20 AS builder
 
 WORKDIR /go/src/app
 COPY go.mod .
diff --git a/Dockerfile.site-controller b/Dockerfile.site-controller
index 7a54054f..9fb0186f 100644
--- a/Dockerfile.site-controller
+++ b/Dockerfile.site-controller
@@ -1,4 +1,4 @@
-FROM golang:1.19 AS builder
+FROM golang:1.20 AS builder
 
 WORKDIR /go/src/app
 
diff --git a/Makefile b/Makefile
index 8b11d8ba..0d934291 100644
--- a/Makefile
+++ b/Makefile
@@ -39,7 +39,7 @@ build-flow-collector:
 	go build -ldflags="${LDFLAGS}"  -o flow-collector cmd/flow-collector/main.go cmd/flow-collector/controller.go cmd/flow-collector/handlers.go
 
 build-config-sync:
-	go build -ldflags="${LDFLAGS}"  -o config-sync cmd/config-sync/main.go cmd/config-sync/config_sync.go
+	go build -ldflags="${LDFLAGS}"  -o config-sync cmd/config-sync/main.go cmd/config-sync/config_sync.go cmd/config-sync/collector.go
 
 build-controllers: build-site-controller build-service-controller build-controller-podman build-flow-collector
 
diff --git a/api/types/client.go b/api/types/client.go
index b44c4918..c8a69d60 100644
--- a/api/types/client.go
+++ b/api/types/client.go
@@ -17,7 +17,7 @@ type ConnectorCreateOptions struct {
 	SkupperNamespace string
 	Name             string
 	Cost             int32
-	Yaml             []byte
+	Secret           *corev1.Secret
 }
 
 type ConnectorRemoveOptions struct {
diff --git a/api/types/types.go b/api/types/types.go
index a926ddcc..9404d762 100644
--- a/api/types/types.go
+++ b/api/types/types.go
@@ -28,11 +28,13 @@ import (
 
 const (
 	// NamespaceDefault means the VAN is in the  skupper namespace which is applied when not specified by clients
-	NamespaceDefault    string = "skupper"
-	DefaultVanName      string = "skupper"
-	DefaultSiteName     string = "skupper-site"
-	ClusterLocalPostfix string = ".svc.cluster.local"
-	SiteConfigMapName   string = "skupper-site"
+	NamespaceDefault           string = "skupper"
+	DefaultVanName             string = "skupper"
+	DefaultSiteName            string = "skupper-site"
+	ClusterLocalPostfix        string = ".svc.cluster.local"
+	SiteConfigMapName          string = "skupper-site"
+	NetworkStatusConfigMapName string = "skupper-network-status"
+	SiteLeaderLockName         string = "skupper-site-leader"
 )
 
 const DefaultTimeoutDuration = time.Second * 120
@@ -71,7 +73,13 @@ var TransportPolicyRule = []rbacv1.PolicyRule{
 	{
 		Verbs:     []string{"get", "list", "watch"},
 		APIGroups: []string{""},
-		Resources: []string{"secrets", "pods", "configmaps"},
+		Resources: []string{"secrets", "pods"},
+	},
+	//needed for collector routine
+	{
+		Verbs:     []string{"get", "list", "watch", "create", "update", "delete"},
+		APIGroups: []string{""},
+		Resources: []string{"configmaps"},
 	},
 	//needed for redeeming token claims
 	{
@@ -85,6 +93,11 @@ var TransportPolicyRule = []rbacv1.PolicyRule{
 		APIGroups: []string{""},
 		Resources: []string{"services"},
 	},
+	{
+		Verbs:     []string{"get"},
+		APIGroups: []string{"apps"},
+		Resources: []string{"deployments"},
+	},
 	{
 		Verbs:     []string{"get", "list", "watch"},
 		APIGroups: []string{"route.openshift.io"},
@@ -534,10 +547,12 @@ type Credential struct {
 	Post        bool
 	Data        map[string][]byte
 	Simple      bool `default:"false"`
+	Labels      map[string]string
 }
 
 type CertAuthority struct {
-	Name string
+	Name   string
+	Labels map[string]string
 }
 
 type CredentialHandler interface {
diff --git a/client/client_debug.go b/client/client_debug.go
index b7e28dfc..77afb916 100644
--- a/client/client_debug.go
+++ b/client/client_debug.go
@@ -86,7 +86,7 @@ func (cli *VanClient) writeConfigMap(name string, tw *tar.Writer) error {
 }
 
 func (cli *VanClient) SkupperDump(ctx context.Context, tarName string, version string, kubeConfigPath string, kubeConfigContext string) (string, error) {
-	configMaps := []string{types.SiteConfigMapName, types.ServiceInterfaceConfigMap, types.TransportConfigMapName, "skupper-sasl-config"}
+	configMaps := []string{types.SiteConfigMapName, types.ServiceInterfaceConfigMap, types.TransportConfigMapName, "skupper-sasl-config", types.NetworkStatusConfigMapName, types.SiteLeaderLockName}
 	deployments := []string{"skupper-site-controller", "skupper-router", "skupper-service-controller"}
 	qdstatFlags := []string{"-g", "-c", "-l", "-n", "-e", "-a", "-m", "-p"}
 
diff --git a/client/connector_create.go b/client/connector_create.go
index 1382211a..fde22f53 100644
--- a/client/connector_create.go
+++ b/client/connector_create.go
@@ -29,7 +29,13 @@ func (cli *VanClient) ConnectorCreateFromFile(ctx context.Context, secretFile st
 		fmt.Println("Could not read connection token", err.Error())
 		return nil, err
 	}
-	options.Yaml = yaml
+	ys := json.NewYAMLSerializer(json.DefaultMetaFactory, scheme.Scheme,
+		scheme.Scheme)
+	options.Secret = &corev1.Secret{}
+	_, _, err = ys.Decode(yaml, nil, options.Secret)
+	if err != nil {
+		return nil, fmt.Errorf("Could not parse connection token: %w", err)
+	}
 	secret, err := cli.ConnectorCreateSecretFromData(ctx, options)
 	if err != nil {
 		return nil, err
@@ -48,84 +54,81 @@ func (cli *VanClient) ConnectorCreateFromFile(ctx context.Context, secretFile st
 func (cli *VanClient) ConnectorCreateSecretFromData(ctx context.Context, options types.ConnectorCreateOptions) (*corev1.Secret, error) {
 	current, err := kube.GetDeployment(types.TransportDeploymentName, options.SkupperNamespace, cli.KubeClient)
 	if err == nil {
-		s := json.NewYAMLSerializer(json.DefaultMetaFactory, scheme.Scheme,
-			scheme.Scheme)
-		var secret corev1.Secret
-		_, _, err = s.Decode(options.Yaml, nil, &secret)
-		if err != nil {
+		if options.Secret == nil {
 			return nil, fmt.Errorf("Could not parse connection token: %w", err)
-		} else {
-			// Validating destination host
-			siteConfig, err := cli.SiteConfigInspect(context.Background(), nil)
-			if err != nil {
-				return nil, err
-			}
-			hostname := ""
-			if secret.ObjectMeta.Labels[types.SkupperTypeQualifier] == types.TypeToken {
-				if siteConfig.Spec.RouterMode == string(types.TransportModeEdge) {
-					hostname = secret.ObjectMeta.Annotations["edge-host"]
-				} else {
-					hostname = secret.ObjectMeta.Annotations["inter-router-host"]
-				}
-			} else {
-				destUrl, err := url.Parse(secret.ObjectMeta.Annotations[types.ClaimUrlAnnotationKey])
-				if err != nil {
-					return nil, fmt.Errorf("Invalid URL defined in token: %s", err)
-				}
-				hostname = destUrl.Hostname()
-			}
-			policy := NewClusterPolicyValidator(cli)
-			res := policy.ValidateOutgoingLink(hostname)
-			if !res.Allowed() {
-				return nil, fmt.Errorf("outgoing link to %s is not allowed", hostname)
-			}
+		}
+		var secret = options.Secret
 
-			cfg, err := cli.getRouterConfig(ctx, cli.Namespace)
-			if err != nil {
-				return nil, fmt.Errorf("error reading router config: %w", err)
-			}
-			linkHandler := domainkube.NewLinkHandlerKube(options.SkupperNamespace, siteConfig, cfg, cli.KubeClient, cli.RestConfig)
-			if options.Name == "" {
-				options.Name = domain.GenerateLinkName(linkHandler)
-			}
-			secret.ObjectMeta.Name = options.Name
-			err = domain.VerifyToken(&secret)
-			if err != nil {
-				return nil, err
+		// Validating destination host
+		siteConfig, err := cli.SiteConfigInspect(context.Background(), nil)
+		if err != nil {
+			return nil, err
+		}
+		hostname := ""
+		if secret.ObjectMeta.Labels[types.SkupperTypeQualifier] == types.TypeToken {
+			if siteConfig.Spec.RouterMode == string(types.TransportModeEdge) {
+				hostname = secret.ObjectMeta.Annotations["edge-host"]
+			} else {
+				hostname = secret.ObjectMeta.Annotations["inter-router-host"]
 			}
-			// Verify if site link can be created
-			err = domain.VerifyNotSelfOrDuplicate(secret, siteConfig.Reference.UID, linkHandler)
+		} else {
+			destUrl, err := url.Parse(secret.ObjectMeta.Annotations[types.ClaimUrlAnnotationKey])
 			if err != nil {
-				return nil, err
+				return nil, fmt.Errorf("Invalid URL defined in token: %s", err)
 			}
-			err = cli.VerifySecretCompatibility(secret)
+			hostname = destUrl.Hostname()
+		}
+		policy := NewClusterPolicyValidator(cli)
+		res := policy.ValidateOutgoingLink(hostname)
+		if !res.Allowed() {
+			return nil, fmt.Errorf("outgoing link to %s is not allowed", hostname)
+		}
+
+		cfg, err := cli.getRouterConfig(ctx, cli.Namespace)
+		if err != nil {
+			return nil, fmt.Errorf("error reading router config: %w", err)
+		}
+		linkHandler := domainkube.NewLinkHandlerKube(options.SkupperNamespace, siteConfig, cfg, cli.KubeClient, cli.RestConfig)
+		if options.Name == "" {
+			options.Name = domain.GenerateLinkName(linkHandler)
+		}
+		secret.ObjectMeta.Name = options.Name
+		err = domain.VerifyToken(secret)
+		if err != nil {
+			return nil, err
+		}
+		// Verify if site link can be created
+		err = domain.VerifyNotSelfOrDuplicate(*secret, siteConfig.Reference.UID, linkHandler)
+		if err != nil {
+			return nil, err
+		}
+		err = cli.VerifySecretCompatibility(*secret)
+		if err != nil {
+			return nil, err
+		}
+		if secret.ObjectMeta.Labels[types.SkupperTypeQualifier] == types.TypeClaimRequest {
+			// can site handle claims?
+			err := cli.requireSiteVersion(ctx, options.SkupperNamespace, "0.7.0")
 			if err != nil {
-				return nil, err
-			}
-			if secret.ObjectMeta.Labels[types.SkupperTypeQualifier] == types.TypeClaimRequest {
-				// can site handle claims?
-				err := cli.requireSiteVersion(ctx, options.SkupperNamespace, "0.7.0")
-				if err != nil {
-					return nil, fmt.Errorf("Claims not supported. %s", err)
-				}
-			}
-			secret.ObjectMeta.SetOwnerReferences([]metav1.OwnerReference{
-				kube.GetDeploymentOwnerReference(current),
-			})
-			if options.Cost != 0 {
-				if secret.ObjectMeta.Annotations == nil {
-					secret.ObjectMeta.Annotations = map[string]string{}
-				}
-				secret.ObjectMeta.Annotations[types.TokenCost] = strconv.Itoa(int(options.Cost))
+				return nil, fmt.Errorf("Claims not supported. %s", err)
 			}
-			_, err = cli.KubeClient.CoreV1().Secrets(options.SkupperNamespace).Create(context.TODO(), &secret, metav1.CreateOptions{})
-			if err == nil {
-				return &secret, nil
-			} else if errors.IsAlreadyExists(err) {
-				return &secret, fmt.Errorf("The connector secret \"%s\"already exists, please choose a different name", secret.ObjectMeta.Name)
-			} else {
-				return nil, fmt.Errorf("Failed to create connector secret: %w", err)
+		}
+		secret.ObjectMeta.SetOwnerReferences([]metav1.OwnerReference{
+			kube.GetDeploymentOwnerReference(current),
+		})
+		if options.Cost != 0 {
+			if secret.ObjectMeta.Annotations == nil {
+				secret.ObjectMeta.Annotations = map[string]string{}
 			}
+			secret.ObjectMeta.Annotations[types.TokenCost] = strconv.Itoa(int(options.Cost))
+		}
+		_, err = cli.KubeClient.CoreV1().Secrets(options.SkupperNamespace).Create(context.TODO(), secret, metav1.CreateOptions{})
+		if err == nil {
+			return secret, nil
+		} else if errors.IsAlreadyExists(err) {
+			return secret, fmt.Errorf("The connector secret \"%s\" already exists, please choose a different name", secret.ObjectMeta.Name)
+		} else {
+			return nil, fmt.Errorf("Failed to create connector secret: %w", err)
 		}
 	} else {
 		return nil, fmt.Errorf("Failed to retrieve router deployment: %w", err)
diff --git a/client/container/client.go b/client/container/client.go
index b0dcb56c..334f8aae 100644
--- a/client/container/client.go
+++ b/client/container/client.go
@@ -21,6 +21,8 @@ type Client interface {
 	ContainerList() ([]*Container, error)
 	ContainerInspect(id string) (*Container, error)
 	ContainerCreate(container *Container) error
+	ContainerUpdate(name string, fn func(newContainer *Container)) (*Container, error)
+	ContainerRename(currentName, newName string) error
 	ContainerRemove(id string) error
 	ContainerExec(id string, command []string) (string, error)
 	ContainerLogs(id string) (string, error)
diff --git a/client/podman/container.go b/client/podman/container.go
index c881b68f..957d21a6 100644
--- a/client/podman/container.go
+++ b/client/podman/container.go
@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/go-openapi/runtime"
+	runtimeclient "github.com/go-openapi/runtime/client"
 	"github.com/skupperproject/skupper/api/types"
 	"github.com/skupperproject/skupper/client/container"
 	"github.com/skupperproject/skupper/client/generated/libpod/client/containers"
@@ -132,6 +133,140 @@ func (p *PodmanRestClient) ContainerCreate(container *container.Container) error
 	return nil
 }
 
+// ContainerUpdate replaces the container (by name) with an identical copy
+// with an applied Customization (required).
+// To achieve that, it follows this procedure:
+// - Create a new container
+// - Stop current container
+// - Rename current container
+// - Rename the new container (replacing the original one)
+// - Starts the new container
+// - Removes the original container
+//
+// In case of failures during this process, the original container is restored.
+func (p *PodmanRestClient) ContainerUpdate(name string, fn func(newContainer *container.Container)) (*container.Container, error) {
+	if fn == nil {
+		return nil, fmt.Errorf("at least one customization is needed")
+	}
+
+	datetime := time.Now().Format("20060102150405")
+	c, err := p.ContainerInspect(name)
+	if err != nil {
+		return nil, err
+	}
+
+	newContainerName := fmt.Sprintf("%s-new-%s", c.Name, datetime)
+	cc := &container.Container{
+		Name:          newContainerName,
+		Image:         c.Image,
+		Env:           c.Env,
+		Labels:        c.Labels,
+		Annotations:   c.Annotations,
+		Networks:      c.Networks,
+		Mounts:        c.Mounts,
+		FileMounts:    c.FileMounts,
+		Ports:         c.Ports,
+		EntryPoint:    c.EntryPoint,
+		Command:       c.Command,
+		RestartPolicy: c.RestartPolicy,
+	}
+
+	// apply new container customization
+	fn(cc)
+	if cc.Name != newContainerName {
+		return nil, fmt.Errorf("container name cannot be changed")
+	}
+
+	// creating a new container
+	err = p.ContainerCreate(cc)
+	if err != nil {
+		return cc, err
+	}
+
+	// rollback this one in case of failures below
+	defer func() {
+		if err != nil {
+			_ = p.ContainerStop(cc.Name)
+			_ = p.ContainerRemove(cc.Name)
+		}
+	}()
+
+	// stopping current container
+	err = p.ContainerStop(name)
+	if err != nil {
+		return cc, err
+	}
+
+	// restarting current container
+	defer func() {
+		if err != nil {
+			_ = p.ContainerStart(name)
+		}
+	}()
+
+	// renaming current container to a backup name
+	backupName := fmt.Sprintf("%s-%s", name, datetime)
+	err = p.ContainerRename(name, backupName)
+	if err != nil {
+		return cc, err
+	}
+
+	// restoring original container
+	defer func() {
+		if err != nil {
+			_ = p.ContainerRename(backupName, name)
+		}
+	}()
+
+	// renaming new container to current name
+	err = p.ContainerRename(cc.Name, name)
+	if err != nil {
+		return cc, err
+	}
+
+	defer func() {
+		if err != nil {
+			_ = p.ContainerRename(name, newContainerName)
+		}
+	}()
+
+	// starting new container
+	err = p.ContainerStart(name)
+	if err != nil {
+		return cc, err
+	} else {
+		if removeErr := p.ContainerRemove(backupName); removeErr != nil {
+			fmt.Printf("Unable to remove backup container: %s - %s", backupName, err)
+			fmt.Println()
+		}
+	}
+
+	return cc, nil
+}
+
+// ContainerUpdateImage updates the image used by the given container (name).
+func (p *PodmanRestClient) ContainerUpdateImage(name string, newImage string) (*container.Container, error) {
+	err := p.ImagePull(newImage)
+	if err != nil {
+		return nil, fmt.Errorf("error pulling image %q: %s", newImage, err)
+	}
+	return p.ContainerUpdate(name, func(newContainer *container.Container) {
+		newContainer.Image = newImage
+	})
+}
+
+func (p *PodmanRestClient) ContainerRename(currentName, newName string) error {
+	cli := containers.New(p.RestClient, formats)
+	params := containers.NewContainerRenameLibpodParams()
+	params.PathName = currentName
+	params.QueryName = newName
+	_, err := cli.ContainerRenameLibpod(params)
+	if err != nil {
+		return fmt.Errorf("error renaming container %s to %s: %v", currentName, newName, err)
+	}
+	return nil
+}
+
 func (p *PodmanRestClient) ContainerRemove(name string) error {
 	cli := containers.New(p.RestClient, formats)
 	params := containers.NewContainerDeleteLibpodParams()
@@ -229,7 +364,10 @@ func (p *PodmanRestClient) ContainerExec(id string, command []string) (string, e
 		Client:             params.HTTPClient,
 	}
 
-	p.RestClient.Consumers["*/*"] = &responseReaderBody{}
+	restClient, ok := p.RestClient.(*runtimeclient.Runtime)
+	if ok {
+		restClient.Consumers["*/*"] = &responseReaderBody{}
+	}
 	result, err = p.RestClient.Submit(startOp)
 	if err != nil {
 		return "", fmt.Errorf("error starting execution: %v", err)
diff --git a/client/podman/container_mock.go b/client/podman/container_mock.go
new file mode 100644
index 00000000..6ad4c2e8
--- /dev/null
+++ b/client/podman/container_mock.go
@@ -0,0 +1,476 @@
+package podman
+
+import (
+	"fmt"
+	"os"
+	"path"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/go-openapi/runtime"
+	"github.com/go-openapi/strfmt"
+	"github.com/google/uuid"
+	"github.com/skupperproject/skupper/client/container"
+	"github.com/skupperproject/skupper/client/generated/libpod/client/containers"
+	"github.com/skupperproject/skupper/client/generated/libpod/client/volumes"
+	"github.com/skupperproject/skupper/client/generated/libpod/models"
+)
+
+func NewPodmanClientMock(containers []*container.Container) *PodmanRestClient {
+	return &PodmanRestClient{
+		RestClient: &RestClientMock{
+			Containers: containers,
+		},
+		endpoint: "",
+	}
+}
+
+// RestClientMock maintains (in memory) a set of containers
+// and volumes that are manipulated through libpod client.
+// It also maintains a set of volume files (in disk) to mock
+// podman volumes.
+// The optional ErrorHook can be set by tests to modify the
+// behavior of certain operations before they are actually
+// performed by the mock implementation.
+type RestClientMock struct {
+	Containers   []*container.Container
+	volumes      map[string]*container.Volume
+	volumesFiles map[string]map[string]string
+	volumesDir   string
+	ErrorHook    func(operation *runtime.ClientOperation) error
+}
+
+func (r *RestClientMock) MockVolumeFiles(volumes map[string]*container.Volume, volumesFiles map[string]map[string]string) error {
+	var err error
+	r.volumesDir, err = os.MkdirTemp("", "skupper-mock-")
+	if err != nil {
+		return err
+	}
+	r.volumes = volumes
+	r.volumesFiles = volumesFiles
+
+	prepareVolumeDir := func(volumeName string, volume *container.Volume) error {
+		volumeDir := path.Join(r.volumesDir, volumeName)
+		if err = os.Mkdir(volumeDir, 0755); err != nil {
+			return err
+		}
+		volume.Source = volumeDir
+		return nil
+	}
+	// creating temp directory for volume
+	for volumeName, volume := range r.volumes {
+		if err = prepareVolumeDir(volumeName, volume); err != nil {
+			return err
+		}
+	}
+	// creating volume files
+	for volumeName, volumeFiles := range r.volumesFiles {
+		volume, ok := r.volumes[volumeName]
+		// if volume not defined earlier, create it
+		if !ok {
+			v := &container.Volume{
+				Name: volumeName,
+			}
+			r.volumes[volumeName] = v
+			if err = prepareVolumeDir(volumeName, v); err != nil {
+				return err
+			}
+
+		}
+		if _, err = volume.CreateFiles(volumeFiles, true); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (r *RestClientMock) CleanupMockVolumeDir() error {
+	if r.volumesDir != "" {
+		return os.RemoveAll(r.volumesDir)
+	}
+	return nil
+}
+
+func (r *RestClientMock) Submit(operation *runtime.ClientOperation) (interface{}, error) {
+	var res interface{}
+	var err error = nil
+
+	switch operation.ID {
+	case "ContainerListLibpod":
+		res, err = r.HandleContainerList(operation, r.ErrorHook)
+	case "ContainerInspectLibpod":
+		res, err = r.HandleContainerInspect(operation, r.ErrorHook)
+	case "ContainerCreateLibpod":
+		res, err = r.HandleContainerCreate(operation, r.ErrorHook)
+	case "ContainerDeleteLibpod":
+		res, err = r.HandleContainerDelete(operation, r.ErrorHook)
+	case "ContainerRenameLibpod":
+		res, err = r.HandleContainerRename(operation, r.ErrorHook)
+	case "ContainerStartLibpod":
+		res, err = r.HandleContainerStart(operation, r.ErrorHook)
+	case "ContainerStopLibpod":
+		res, err = r.HandleContainerStop(operation, r.ErrorHook)
+	case "VolumeInspectLibpod":
+		res, err = r.HandleVolumeInspect(operation, r.ErrorHook)
+	case "VolumeListLibpod":
+		res, err = r.HandleVolumeList(operation, r.ErrorHook)
+	}
+	return res, err
+}
+
+func (r *RestClientMock) HandleContainerList(operation *runtime.ClientOperation, errorHook func(operation *runtime.ClientOperation) error) (*containers.ContainerListLibpodOK, error) {
+	res := &containers.ContainerListLibpodOK{}
+	if len(r.Containers) > 0 {
+		res.Payload = []*models.ListContainer{}
+		for _, c := range r.Containers {
+			state := "running"
+			if !c.Running {
+				state = "exited"
+			}
+			res.Payload = append(res.Payload, &models.ListContainer{
+				Command:   c.Command,
+				Created:   strfmt.DateTime(c.CreatedAt),
+				CreatedAt: c.CreatedAt.String(),
+				ExitCode:  int32(c.ExitCode),
+				Exited:    !c.ExitedAt.Equal(time.Time{}),
+				ExitedAt:  c.ExitedAt.Unix(),
+				ID:        c.ID,
+				Image:     c.Image,
+				ImageID:   c.Image,
+				Labels:    c.Labels,
+				Mounts: func() []string {
+					var mounts []string
+					for _, m := range c.Mounts {
+						mounts = append(mounts, m.Destination)
+					}
+					return mounts
+				}(),
+				Names: []string{c.Name},
+				Networks: func() []string {
+					var networks []string
+					for _, n := range c.Networks {
+						networks = append(networks, n.ID)
+					}
+					return networks
+				}(),
+				Pod: c.Pod,
+				Ports: func() []*models.PortMapping {
+					var pm []*models.PortMapping
+					for _, port := range c.Ports {
+						containerPort, _ := strconv.ParseUint(port.Target, 10, 16)
+						hostPort, _ := strconv.ParseUint(port.Host, 10, 16)
+						pm = append(pm, &models.PortMapping{
+							ContainerPort: uint16(containerPort),
+							HostIP:        port.HostIP,
+							HostPort:      uint16(hostPort),
+							Protocol:      port.Protocol,
+						})
+					}
+					return pm
+				}(),
+				StartedAt: c.StartedAt.Unix(),
+				State:     state,
+			})
+		}
+	}
+	if errorHook != nil {
+		return res, errorHook(operation)
+	}
+	return res, nil
+}
+
+func (r *RestClientMock) HandleContainerInspect(operation *runtime.ClientOperation, hook func(operation *runtime.ClientOperation) error) (*containers.ContainerInspectLibpodOK, error) {
+	res := &containers.ContainerInspectLibpodOK{}
+	if hook != nil {
+		if err := hook(operation); err != nil {
+			return res, err
+		}
+	}
+	params := operation.Params.(*containers.ContainerInspectLibpodParams)
+	var c *container.Container
+	for _, cc := range r.Containers {
+		if cc.Name == params.Name || cc.ID == params.Name {
+			c = cc
+			break
+		}
+	}
+	if c == nil {
+		return nil, fmt.Errorf("container not found")
+	}
+	res.Payload = &containers.ContainerInspectLibpodOKBody{
+		Config: &models.InspectContainerConfig{
+			Annotations: c.Annotations,
+			Cmd:         c.Command,
+			Entrypoint:  strings.Join(c.EntryPoint, " "),
+			Env:         c.EnvSlice(),
+			Image:       c.Image,
+			Labels:      c.Labels,
+			User:        ToSpecGenerator(c).User,
+		},
+		Created:   strfmt.DateTime(c.CreatedAt),
+		ID:        c.ID,
+		Image:     c.Image,
+		ImageName: c.Image,
+		Mounts: func() []*models.InspectMount {
+			var m []*models.InspectMount
+			for _, mount := range c.Mounts {
+				m = append(m, &models.InspectMount{
+					Name:        mount.Name,
+					Source:      mount.Source,
+					Destination: mount.Destination,
+					Type:        "volume",
+					Mode:        mount.Mode,
+					RW:          mount.RW,
+				})
+			}
+			for _, fm := range c.FileMounts {
+				m = append(m, &models.InspectMount{
+					Source:      fm.Source,
+					Destination: fm.Destination,
+					Options:     fm.Options,
+					Type:        "bind",
+				})
+			}
+			return m
+		}(),
+		Name: c.Name,
+		NetworkSettings: func() *models.InspectNetworkSettings {
+			if len(c.Networks) == 0 && len(c.Ports) == 0 {
+				return nil
+			}
+			settings := &models.InspectNetworkSettings{
+				Networks: map[string]models.InspectAdditionalNetwork{},
+				Ports:    map[string][]models.InspectHostPort{},
+			}
+			for k, v := range c.Networks {
+				settings.Networks[k] = models.InspectAdditionalNetwork{
+					NetworkID:   "",
+					IPAddress:   "",
+					IPPrefixLen: 0,
+					MacAddress:  "",
+					Gateway:     "",
+					Aliases:     v.Aliases,
+				}
+			}
+			for _, port := range c.Ports {
+				portKey := fmt.Sprintf("%s/%s", port.Target, port.Protocol)
+				settings.Ports[portKey] = []models.InspectHostPort{
+					{HostIP: port.HostIP, HostPort: port.Host},
+				}
+			}
+			return settings
+		}(),
+		Pod: c.Pod,
+		State: &models.InspectContainerState{
+			ExitCode:   int32(c.ExitCode),
+			FinishedAt: strfmt.DateTime(c.ExitedAt),
+			Running:    c.Running,
+			StartedAt:  strfmt.DateTime(c.StartedAt),
+		},
+	}
+	return res, nil
+}
+
+func (r *RestClientMock) HandleContainerCreate(operation *runtime.ClientOperation, hook func(operation *runtime.ClientOperation) error) (*containers.ContainerCreateLibpodCreated, error) {
+	res := &containers.ContainerCreateLibpodCreated{}
+	if hook != nil {
+		if err := hook(operation); err != nil {
+			return res, err
+		}
+	}
+	params := operation.Params.(*containers.ContainerCreateLibpodParams)
+	spec := params.Create
+
+	for _, c := range r.Containers {
+		if c.Name == spec.Name {
+			return res, fmt.Errorf("container already exists")
+		}
+	}
+	c := &container.Container{
+		ID:          strings.Replace(uuid.New().String(), "-", "", -1),
+		Name:        spec.Name,
+		Pod:         spec.Pod,
+		Image:       spec.Image,
+		Env:         spec.Env,
+		Labels:      spec.Labels,
+		Annotations: spec.Annotations,
+		Networks: func() map[string]container.ContainerNetworkInfo {
+			res := map[string]container.ContainerNetworkInfo{}
+			for name, perNetworkOpts := range spec.Networks {
+				res[name] = container.ContainerNetworkInfo{
+					Aliases: perNetworkOpts.Aliases,
+				}
+			}
+			return res
+		}(),
+		Mounts: func() []container.Volume {
+			var res []container.Volume
+			for _, v := range spec.Volumes {
+				res = append(res, container.Volume{
+					Name:        v.Name,
+					Destination: v.Dest,
+				})
+			}
+			return res
+		}(),
+		FileMounts: func() []container.FileMount {
+			var res []container.FileMount
+			for _, fm := range spec.Mounts {
+				res = append(res, container.FileMount{
+					Source:      fm.Source,
+					Destination: fm.Destination,
+					Options:     fm.Options,
+				})
+			}
+			return nil
+		}(),
+		Ports: func() []container.Port {
+			var res []container.Port
+			// Port mapping
+			for _, pm := range spec.PortMappings {
+				res = append(res, container.Port{
+					Host:     strconv.Itoa(int(pm.HostPort)),
+					HostIP:   pm.HostIP,
+					Target:   strconv.Itoa(int(pm.ContainerPort)),
+					Protocol: pm.Protocol,
+				})
+			}
+			return res
+		}(),
+		EntryPoint:    spec.Entrypoint,
+		Command:       spec.Command,
+		RestartPolicy: spec.RestartPolicy,
+		RestartCount:  int(spec.RestartRetries),
+		CreatedAt:     time.Now(),
+	}
+	r.Containers = append(r.Containers, c)
+	return res, nil
+}
+
+func (r *RestClientMock) HandleContainerRename(operation *runtime.ClientOperation, hook func(operation *runtime.ClientOperation) error) (*containers.ContainerRenameLibpodNoContent, error) {
+	res := &containers.ContainerRenameLibpodNoContent{}
+	if hook != nil {
+		if err := hook(operation); err != nil {
+			return res, err
+		}
+	}
+	params := operation.Params.(*containers.ContainerRenameLibpodParams)
+	curName := params.PathName
+	newName := params.QueryName
+
+	// current name must exist
+	var cc *container.Container
+
+	for _, c := range r.Containers {
+		if c.Name == newName {
+			return res, fmt.Errorf("a container named %q already exists", newName)
+		} else if c.Name == curName {
+			cc = c
+		}
+	}
+	cc.Name = newName
+	return res, nil
+}
+
+func (r *RestClientMock) HandleContainerStart(operation *runtime.ClientOperation, hook func(operation *runtime.ClientOperation) error) (interface{}, error) {
+	res := &containers.ContainerStartLibpodNoContent{}
+	if hook != nil {
+		if err := hook(operation); err != nil {
+			return res, err
+		}
+	}
+	params := operation.Params.(*containers.ContainerStartLibpodParams)
+	for _, c := range r.Containers {
+		if c.Name == params.Name || c.ID == params.Name {
+			if !c.Running {
+				c.Running = true
+				c.StartedAt = time.Now()
+				c.ExitCode = 0
+				c.ExitedAt = time.Time{}
+			}
+			return res, nil
+		}
+	}
+	return res, fmt.Errorf("no container with name or ID %q", params.Name)
+}
+
+func (r *RestClientMock) HandleContainerStop(operation *runtime.ClientOperation, hook func(operation *runtime.ClientOperation) error) (interface{}, error) {
+	var res interface{}
+	if hook != nil {
+		if err := hook(operation); err != nil {
+			return res, err
+		}
+	}
+	params := operation.Params.(*containers.ContainerStopLibpodParams)
+	for _, c := range r.Containers {
+		if c.Name == params.Name || c.ID == params.Name {
+			res = &containers.ContainerStopLibpodNoContent{}
+			if c.Running {
+				c.Running = false
+				c.ExitedAt = time.Now()
+				c.ExitCode = 0
+			}
+			return res, nil
+		}
+	}
+	res = &containers.ContainerStopLibpodNotFound{}
+	return res, fmt.Errorf("no container with name or ID %q", params.Name)
+}
+
+func (r *RestClientMock) HandleContainerDelete(operation *runtime.ClientOperation, hook func(operation *runtime.ClientOperation) error) (interface{}, error) {
+	var res interface{}
+	if hook != nil {
+		if err := hook(operation); err != nil {
+			res = &containers.ContainerDeleteLibpodNoContent{}
+			return res, err
+		}
+	}
+	params := operation.Params.(*containers.ContainerDeleteLibpodParams)
+	for i, c := range r.Containers {
+		if c.Name == params.Name || c.ID == params.Name {
+			r.Containers = append(r.Containers[:i], r.Containers[i+1:]...)
+			res = &containers.ContainerDeleteLibpodOK{}
+			return res, nil
+		}
+	}
+	res = &containers.ContainerDeleteLibpodNoContent{}
+	return res, fmt.Errorf("no container with name or ID %q", params.Name)
+}
+
+func (r *RestClientMock) HandleVolumeInspect(operation *runtime.ClientOperation, hook func(operation *runtime.ClientOperation) error) (interface{}, error) {
+	var res = &volumes.VolumeInspectLibpodOK{}
+	if hook != nil {
+		if err := hook(operation); err != nil {
+			return res, err
+		}
+	}
+	params := operation.Params.(*volumes.VolumeInspectLibpodParams)
+	v, ok := r.volumes[params.Name]
+	if !ok {
+		return res, fmt.Errorf("no volume with name %q", params.Name)
+	}
+	res.Payload = &volumes.VolumeInspectLibpodOKBody{
+		Labels:     v.Labels,
+		Mountpoint: v.Source,
+		Name:       v.Name,
+	}
+	return res, nil
+}
+
+func (r *RestClientMock) HandleVolumeList(operation *runtime.ClientOperation, hook func(operation *runtime.ClientOperation) error) (interface{}, error) {
+	var res = &volumes.VolumeListLibpodOK{}
+	if hook != nil {
+		if err := hook(operation); err != nil {
+			return res, err
+		}
+	}
+	for vName, v := range r.volumes {
+		res.Payload = append(res.Payload, &models.VolumeConfigResponse{
+			Labels:     v.Labels,
+			Mountpoint: v.Source,
+			Name:       vName,
+		})
+	}
+	return res, nil
+}
diff --git a/client/podman/image.go b/client/podman/image.go
index 9683b4eb..59ea9450 100644
--- a/client/podman/image.go
+++ b/client/podman/image.go
@@ -1,7 +1,10 @@
 package podman
 
 import (
+	"encoding/base64"
+	"encoding/json"
 	"fmt"
+	"os"
 	"strings"
 
 	"github.com/go-openapi/runtime"
@@ -67,6 +70,7 @@ func (p *PodmanRestClient) ImagePull(id string) error {
 	params.OS = stringP("")
 	params.Variant = stringP("")
 	params.Policy = stringP("always")
+	params.XRegistryAuth = getXRegistryAuth(id)
 
 	// Need to do that as the default response reader is being closed too soon
 	op := &runtime.ClientOperation{
@@ -81,9 +85,79 @@ func (p *PodmanRestClient) ImagePull(id string) error {
 		Context:            params.Context,
 		Client:             params.HTTPClient,
 	}
-	_, err := p.RestClient.Submit(op)
+	res, err := p.RestClient.Submit(op)
 	if err != nil {
 		return fmt.Errorf("error pulling image %s: %v", id, err)
 	}
+	// eventually err is nil but auth problems are reported as json after a string msg
+	if resStr, ok := res.(string); ok {
+		resStrClean := strings.TrimLeftFunc(resStr, func(r rune) bool {
+			if r == '{' {
+				return false
+			}
+			return true
+		})
+		var jsonRes map[string]interface{}
+		if err = json.Unmarshal([]byte(resStrClean), &jsonRes); err == nil {
+			if errMsg, ok := jsonRes["error"]; ok && errMsg != "" {
+				return fmt.Errorf("unable to pull image %s: %v", id, errMsg)
+			}
+		}
+	}
+	return nil
+}
+
+func getXRegistryAuth(image string) *string {
+	authFile := os.Getenv("REGISTRY_AUTH_FILE")
+	// use the default
+	if authFile == "" {
+		return nil
+	}
+	data, err := os.ReadFile(authFile)
+	if err != nil {
+		fmt.Printf("Unable to read REGISTRY_AUTH_FILE - %s", err)
+		fmt.Println()
+		return nil
+	}
+	var jsonData map[string]interface{}
+	if err = json.Unmarshal(data, &jsonData); err != nil {
+		fmt.Printf("Unable to parse REGISTRY_AUTH_FILE - %s", err)
+		fmt.Println()
+		return nil
+	}
+	if auths, ok := jsonData["auths"]; ok {
+		authsMap, ok := auths.(map[string]interface{})
+		if !ok {
+			return nil
+		}
+		imageServer := strings.Split(image, "/")[0]
+		if serverData, ok := authsMap[imageServer]; ok {
+			if serverDataMap, ok := serverData.(map[string]interface{}); ok {
+				authInfo, ok := serverDataMap["auth"]
+				if !ok {
+					return nil
+				}
+				authInfoStr := authInfo.(string)
+				credentialsBytes, err := base64.StdEncoding.DecodeString(authInfoStr)
+				if err != nil {
+					fmt.Printf("Unable to decode base64 auth info for %s - %s", imageServer, err)
+					fmt.Println()
+					return nil
+				}
+				credentials := strings.Split(string(credentialsBytes), ":")
+				if len(credentials) != 2 {
+					return nil
+				}
+				registryAuthMap := map[string]map[string]string{}
+				registryAuthMap[imageServer] = map[string]string{
+					"username": credentials[0],
+					"password": credentials[1],
+				}
+				registryAuthMapJson, _ := json.Marshal(registryAuthMap)
+				registryAuthMapB64 := base64.StdEncoding.EncodeToString(registryAuthMapJson)
+				return &registryAuthMapB64
+			}
+		}
+	}
 	return nil
 }
diff --git a/client/podman/rest.go b/client/podman/rest.go
index 0395d093..b79a412e 100644
--- a/client/podman/rest.go
+++ b/client/podman/rest.go
@@ -14,8 +14,8 @@ import (
 	"strings"
 	"time"
 
-	runtime2 "github.com/go-openapi/runtime"
-	runtime "github.com/go-openapi/runtime/client"
+	"github.com/go-openapi/runtime"
+	runtimeclient "github.com/go-openapi/runtime/client"
 	"github.com/go-openapi/strfmt"
 	"github.com/skupperproject/skupper/client/generated/libpod/models"
 	"github.com/skupperproject/skupper/pkg/config"
@@ -34,7 +34,7 @@ var (
 )
 
 type PodmanRestClient struct {
-	RestClient *runtime.Runtime
+	RestClient runtime.ClientTransport
 	endpoint   string
 }
 
@@ -93,7 +93,7 @@ func NewPodmanClient(endpoint, basePath string) (*PodmanRestClient, error) {
 	if basePath == "" {
 		basePath = DEFAULT_BASE_PATH
 	}
-	c := runtime.New(hostPort, basePath, []string{u.Scheme})
+	c := runtimeclient.New(hostPort, basePath, []string{u.Scheme})
 	// Initializing transport like the http.DefaultTransport
 	// to avoid modifying it directly, as Runtime.Transport is
 	// set to http.DefaultTransport (variable)
@@ -170,7 +170,7 @@ func stringP(val string) *string {
 type responseReaderID struct {
 }
 
-func (r *responseReaderID) ReadResponse(response runtime2.ClientResponse, consumer runtime2.Consumer) (interface{}, error) {
+func (r *responseReaderID) ReadResponse(response runtime.ClientResponse, consumer runtime.Consumer) (interface{}, error) {
 	switch response.Code() {
 	case 200, 201:
 		resp := &models.IDResponse{}
@@ -225,7 +225,7 @@ func (r *responseReaderBody) Consume(reader io.Reader, i interface{}) error {
 	return nil
 }
 
-func (r *responseReaderBody) ReadResponse(response runtime2.ClientResponse, consumer runtime2.Consumer) (interface{}, error) {
+func (r *responseReaderBody) ReadResponse(response runtime.ClientResponse, consumer runtime.Consumer) (interface{}, error) {
 	switch response.Code() {
 	case 200, 201:
 		bodyStr := ""
@@ -245,7 +245,7 @@ func (r *responseReaderBody) ReadResponse(response runtime2.ClientResponse, cons
 	}
 }
 
-func (p *PodmanRestClient) ResponseIDReader(httpClient *runtime2.ClientOperation) {
+func (p *PodmanRestClient) ResponseIDReader(httpClient *runtime.ClientOperation) {
 	httpClient.Reader = &responseReaderID{}
 }
 
diff --git a/client/policy_validator.go b/client/policy_validator.go
index dd5e60e6..3b24ff72 100644
--- a/client/policy_validator.go
+++ b/client/policy_validator.go
@@ -130,11 +130,13 @@ func (p *ClusterPolicyValidator) AppliesToNS(policyName string) bool {
 	skupperPolicy, err := p.getSkupperPolicy()
 	// If unable to determine, revalidate
 	if err != nil {
+		recordPolicyValidatorEvent("unable to get policy client instance: %s - revalidating all policies", err)
 		return true
 	}
 	pol, err := skupperPolicy.Get(context.TODO(), policyName, v1.GetOptions{})
 	// If policy not found, revalidate
 	if err != nil {
+		recordPolicyValidatorEvent("unable to get policy %s: %s - revalidating all policies", policyName, err)
 		return true
 	}
 	namespace, _ := p.cli.KubeClient.CoreV1().Namespaces().Get(context.TODO(), p.cli.Namespace, v1.GetOptions{})
@@ -421,9 +423,7 @@ func (p *PolicyAPIClient) execGet(args ...string) (*PolicyAPIResult, error) {
 		} else if err != notEnabledErr {
 			err = fmt.Errorf("Unable to communicate with the API: %v", err)
 		}
-		if event.DefaultStore != nil {
-			event.Recordf("PolicyAPIError", err.Error())
-		}
+		recordPolicyValidatorEvent(err.Error())
 		return &PolicyAPIResult{
 			Allowed: false,
 			Enabled: false,
@@ -494,3 +494,9 @@ func (p *PolicyAPIClient) IncomingLink() (*PolicyAPIResult, error) {
 func (p *PolicyAPIClient) OutgoingLink(hostname string) (*PolicyAPIResult, error) {
 	return p.execGet("outgoinglink", hostname)
 }
+
+func recordPolicyValidatorEvent(format string, args ...interface{}) {
+	if event.DefaultStore != nil {
+		event.Recordf("PolicyAPIError", format, args...)
+	}
+}
diff --git a/client/revoke_all.go b/client/revoke_all.go
index 33e985fc..15d9d900 100644
--- a/client/revoke_all.go
+++ b/client/revoke_all.go
@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	"fmt"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -35,18 +36,21 @@ func (cli *VanClient) appendControllerIngressHost(cred *types.Credential) bool {
 	return false
 }
 
-func (cli *VanClient) regenerateSiteSecret(ctx context.Context, ca *corev1.Secret) error {
+func (cli *VanClient) regenerateSiteSecret(ctx context.Context, ca *corev1.Secret, namespace string) error {
 	siteServerSecret := types.Credential{
 		Name:    types.SiteServerSecret,
 		Subject: types.TransportServiceName,
-		Hosts:   []string{types.TransportServiceName + "." + cli.Namespace},
+		Hosts:   []string{types.TransportServiceName + "." + namespace},
 	}
 
-	siteconfig, err := cli.SiteConfigInspect(ctx, nil)
+	siteconfig, err := cli.SiteConfigInspectInNamespace(ctx, nil, namespace)
 	if err != nil {
 		return err
 	}
-	rslvr, err := resolver.NewResolver(cli, cli.Namespace, &siteconfig.Spec)
+	if siteconfig == nil {
+		return fmt.Errorf("No site found in %s", namespace)
+	}
+	rslvr, err := resolver.NewResolver(cli, namespace, &siteconfig.Spec)
 	if err != nil {
 		return err
 	}
@@ -55,11 +59,11 @@ func (cli *VanClient) regenerateSiteSecret(ctx context.Context, ca *corev1.Secre
 		return err
 	}
 	siteServerSecret.Hosts = append(siteServerSecret.Hosts, hosts...)
-	_, err = kube.RegenerateCredentials(siteServerSecret, cli.Namespace, ca, cli.KubeClient)
+	_, err = kube.RegenerateCredentials(siteServerSecret, namespace, ca, cli.KubeClient)
 	if err != nil {
 		return err
 	}
-	return cli.restartRouter(cli.Namespace)
+	return cli.restartRouter(namespace)
 }
 
 func (cli *VanClient) RevokeAccess(ctx context.Context) error {
@@ -78,5 +82,5 @@ func (cli *VanClient) RevokeAccess(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	return cli.regenerateSiteSecret(ctx, ca)
+	return cli.regenerateSiteSecret(ctx, ca, cli.Namespace)
 }
diff --git a/client/router_create.go b/client/router_create.go
index e303da71..106e14d2 100644
--- a/client/router_create.go
+++ b/client/router_create.go
@@ -1,20 +1,18 @@
 package client
 
 import (
-	"bytes"
 	"context"
 	"fmt"
 	"log"
 	"net"
 	"strconv"
 	"strings"
-	"text/template"
 	"time"
 
 	routev1 "github.com/openshift/api/route/v1"
+	"github.com/skupperproject/skupper/pkg/config"
 	"github.com/skupperproject/skupper/pkg/images"
 	"github.com/skupperproject/skupper/pkg/version"
-	"golang.org/x/crypto/bcrypt"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
@@ -103,10 +101,6 @@ func removeRules(original []rbacv1.PolicyRule, apigroups []string) []rbacv1.Poli
 	return rules
 }
 
-func hashPrometheusPassword(password string) ([]byte, error) {
-	return bcrypt.GenerateFromPassword([]byte(password), 14)
-}
-
 func (cli *VanClient) GetVanPrometheusServerSpec(options types.SiteConfigSpec, van *types.RouterSpec) {
 	// prometheus-server container index
 	const (
@@ -163,6 +157,7 @@ func (cli *VanClient) GetVanPrometheusServerSpec(options types.SiteConfigSpec, v
 		ObjectMeta: metav1.ObjectMeta{
 			Name:        types.PrometheusServiceAccountName,
 			Annotations: annotation,
+			Labels:      options.Labels,
 		},
 	})
 	van.PrometheusServer.ServiceAccounts = serviceAccounts
@@ -176,7 +171,8 @@ func (cli *VanClient) GetVanPrometheusServerSpec(options types.SiteConfigSpec, v
 			Kind:       "Role",
 		},
 		ObjectMeta: metav1.ObjectMeta{
-			Name: types.PrometheusRoleName,
+			Name:   types.PrometheusRoleName,
+			Labels: options.Labels,
 		},
 		Rules: cli.getControllerRules(),
 	})
@@ -189,7 +185,8 @@ func (cli *VanClient) GetVanPrometheusServerSpec(options types.SiteConfigSpec, v
 			Kind:       "RoleBinding",
 		},
 		ObjectMeta: metav1.ObjectMeta{
-			Name: types.PrometheusRoleBindingName,
+			Name:   types.PrometheusRoleBindingName,
+			Labels: options.Labels,
 		},
 		Subjects: []rbacv1.Subject{{
 			Kind: "ServiceAccount",
@@ -223,6 +220,7 @@ func (cli *VanClient) GetVanPrometheusServerSpec(options types.SiteConfigSpec, v
 			ObjectMeta: metav1.ObjectMeta{
 				Name:        types.PrometheusServiceName,
 				Annotations: annotations,
+				Labels:      options.Labels,
 			},
 			Spec: corev1.ServiceSpec{
 				Selector: van.PrometheusServer.Labels,
@@ -373,6 +371,7 @@ func (cli *VanClient) GetVanControllerSpec(options types.SiteConfigSpec, van *ty
 		ObjectMeta: metav1.ObjectMeta{
 			Name:        types.ControllerServiceAccountName,
 			Annotations: annotation,
+			Labels:      options.Labels,
 		},
 	})
 	van.Controller.ServiceAccounts = serviceAccounts
@@ -384,7 +383,8 @@ func (cli *VanClient) GetVanControllerSpec(options types.SiteConfigSpec, van *ty
 			Kind:       "Role",
 		},
 		ObjectMeta: metav1.ObjectMeta{
-			Name: types.ControllerRoleName,
+			Name:   types.ControllerRoleName,
+			Labels: options.Labels,
 		},
 		Rules: cli.getControllerRules(),
 	})
@@ -397,7 +397,8 @@ func (cli *VanClient) GetVanControllerSpec(options types.SiteConfigSpec, van *ty
 			Kind:       "RoleBinding",
 		},
 		ObjectMeta: metav1.ObjectMeta{
-			Name: types.ControllerRoleBindingName,
+			Name:   types.ControllerRoleBindingName,
+			Labels: options.Labels,
 		},
 		Subjects: []rbacv1.Subject{{
 			Kind: "ServiceAccount",
@@ -450,7 +451,8 @@ func (cli *VanClient) GetVanControllerSpec(options types.SiteConfigSpec, van *ty
 					Kind:       "Route",
 				},
 				ObjectMeta: metav1.ObjectMeta{
-					Name: types.ConsoleRouteName,
+					Name:   types.ConsoleRouteName,
+					Labels: options.Labels,
 				},
 				Spec: routev1.RouteSpec{
 					Path: "",
@@ -498,6 +500,7 @@ func (cli *VanClient) GetVanControllerSpec(options types.SiteConfigSpec, van *ty
 				Hosts:       controllerHosts,
 				ConnectJson: false,
 				Post:        post,
+				Labels:      options.Labels,
 			})
 		}
 		controllerPorts = append(controllerPorts, metricsPort)
@@ -512,6 +515,7 @@ func (cli *VanClient) GetVanControllerSpec(options types.SiteConfigSpec, van *ty
 				Hosts:       hosts,
 				ConnectJson: false,
 				Post:        false,
+				Labels:      options.Labels,
 			})
 		}
 		controllerPorts = append(controllerPorts, corev1.ServicePort{
@@ -536,6 +540,7 @@ func (cli *VanClient) GetVanControllerSpec(options types.SiteConfigSpec, van *ty
 			ObjectMeta: metav1.ObjectMeta{
 				Name:        types.ControllerServiceName,
 				Annotations: annotations,
+				Labels:      options.Labels,
 			},
 			Spec: corev1.ServiceSpec{
 				Selector: van.Controller.Labels,
@@ -807,7 +812,8 @@ func (cli *VanClient) GetRouterSpecFromOpts(options types.SiteConfigSpec, siteId
 			Kind:       "Role",
 		},
 		ObjectMeta: metav1.ObjectMeta{
-			Name: types.TransportRoleName,
+			Name:   types.TransportRoleName,
+			Labels: options.Labels,
 		},
 		Rules: cli.adjustRules(types.TransportPolicyRule),
 	})
@@ -820,7 +826,8 @@ func (cli *VanClient) GetRouterSpecFromOpts(options types.SiteConfigSpec, siteId
 			Kind:       "RoleBinding",
 		},
 		ObjectMeta: metav1.ObjectMeta{
-			Name: types.TransportRoleBindingName,
+			Name:   types.TransportRoleBindingName,
+			Labels: options.Labels,
 		},
 		Subjects: []rbacv1.Subject{{
 			Kind: "ServiceAccount",
@@ -848,21 +855,27 @@ func (cli *VanClient) GetRouterSpecFromOpts(options types.SiteConfigSpec, siteId
 		ObjectMeta: metav1.ObjectMeta{
 			Name:        types.TransportServiceAccountName,
 			Annotations: annotation,
+			Labels:      options.Labels,
 		},
 	})
 	van.Transport.ServiceAccounts = serviceAccounts
 
 	cas := []types.CertAuthority{}
 	cas = append(cas, types.CertAuthority{
-		Name: types.LocalCaSecret,
+		Name:   types.LocalCaSecret,
+		Labels: options.Labels,
 	})
 	if !isEdge {
 		cas = append(cas, types.CertAuthority{
-			Name: types.SiteCaSecret,
+			Name:   types.SiteCaSecret,
+			Labels: options.Labels,
 		})
 	}
 
-	cas = append(cas, types.CertAuthority{Name: types.ServiceCaSecret})
+	cas = append(cas, types.CertAuthority{
+		Name:   types.ServiceCaSecret,
+		Labels: options.Labels,
+	})
 
 	van.CertAuthoritys = cas
 
@@ -874,6 +887,7 @@ func (cli *VanClient) GetRouterSpecFromOpts(options types.SiteConfigSpec, siteId
 		Hosts:       []string{types.LocalTransportServiceName, types.LocalTransportServiceName + "." + van.Namespace + ".svc.cluster.local"},
 		ConnectJson: false,
 		Post:        false,
+		Labels:      options.Labels,
 	})
 	credentials = append(credentials, types.Credential{
 		CA:          types.LocalCaSecret,
@@ -882,6 +896,7 @@ func (cli *VanClient) GetRouterSpecFromOpts(options types.SiteConfigSpec, siteId
 		Hosts:       []string{},
 		ConnectJson: true,
 		Post:        false,
+		Labels:      options.Labels,
 	})
 
 	credentials = append(credentials, types.Credential{
@@ -891,6 +906,7 @@ func (cli *VanClient) GetRouterSpecFromOpts(options types.SiteConfigSpec, siteId
 		ConnectJson: false,
 		Post:        false,
 		Simple:      true,
+		Labels:      options.Labels,
 	})
 
 	if !isEdge {
@@ -901,6 +917,7 @@ func (cli *VanClient) GetRouterSpecFromOpts(options types.SiteConfigSpec, siteId
 			Hosts:       []string{types.TransportServiceName + "." + van.Namespace, types.TransportServiceName + "." + van.Namespace + ".svc.cluster.local"},
 			ConnectJson: false,
 			Post:        true,
+			Labels:      options.Labels,
 		})
 	}
 	if options.AuthMode == string(types.ConsoleAuthModeInternal) && (options.EnableFlowCollector || options.EnableRestAPI) {
@@ -915,6 +932,7 @@ func (cli *VanClient) GetRouterSpecFromOpts(options types.SiteConfigSpec, siteId
 			ConnectJson: false,
 			Data:        userData,
 			Post:        false,
+			Labels:      options.Labels,
 		})
 	}
 	van.TransportCredentials = credentials
@@ -929,6 +947,7 @@ func (cli *VanClient) GetRouterSpecFromOpts(options types.SiteConfigSpec, siteId
 		ObjectMeta: metav1.ObjectMeta{
 			Name:        types.LocalTransportServiceName,
 			Annotations: map[string]string{},
+			Labels:      options.Labels,
 		},
 		Spec: corev1.ServiceSpec{
 			Selector: van.Transport.Labels,
@@ -959,6 +978,7 @@ func (cli *VanClient) GetRouterSpecFromOpts(options types.SiteConfigSpec, siteId
 			ObjectMeta: metav1.ObjectMeta{
 				Name:        types.TransportServiceName,
 				Annotations: options.Router.ServiceAnnotations,
+				Labels:      options.Labels,
 			},
 			Spec: corev1.ServiceSpec{
 				Selector: van.Transport.Labels,
@@ -1012,6 +1032,7 @@ func (cli *VanClient) GetRouterSpecFromOpts(options types.SiteConfigSpec, siteId
 			ObjectMeta: metav1.ObjectMeta{
 				Name:        types.InterRouterRouteName,
 				Annotations: options.IngressAnnotations,
+				Labels:      options.Labels,
 			},
 			Spec: routev1.RouteSpec{
 				Path: "",
@@ -1037,6 +1058,7 @@ func (cli *VanClient) GetRouterSpecFromOpts(options types.SiteConfigSpec, siteId
 			ObjectMeta: metav1.ObjectMeta{
 				Name:        types.EdgeRouteName,
 				Annotations: options.IngressAnnotations,
+				Labels:      options.Labels,
 			},
 			Spec: routev1.RouteSpec{
 				Path: "",
@@ -1062,6 +1084,7 @@ func (cli *VanClient) GetRouterSpecFromOpts(options types.SiteConfigSpec, siteId
 			ObjectMeta: metav1.ObjectMeta{
 				Name:        types.ClaimRedemptionRouteName,
 				Annotations: options.IngressAnnotations,
+				Labels:      options.Labels,
 			},
 			Spec: routev1.RouteSpec{
 				Path: "",
@@ -1193,7 +1216,7 @@ sasldb_path: /tmp/skrouterd.sasldb
 		saslData := &map[string]string{
 			"skrouterd.conf": config,
 		}
-		kube.NewConfigMap("skupper-sasl-config", saslData, nil, nil, siteOwnerRef, van.Namespace, cli.KubeClient)
+		kube.NewConfigMap("skupper-sasl-config", saslData, &options.Spec.Labels, nil, siteOwnerRef, van.Namespace, cli.KubeClient)
 	}
 	for _, sa := range van.Transport.ServiceAccounts {
 		sa.ObjectMeta.OwnerReferences = ownerRefs
@@ -1252,9 +1275,9 @@ sasldb_path: /tmp/skrouterd.sasldb
 		return err
 	}
 
-	kube.NewConfigMap(types.ServiceInterfaceConfigMap, nil, nil, nil, siteOwnerRef, van.Namespace, cli.KubeClient)
+	kube.NewConfigMap(types.ServiceInterfaceConfigMap, nil, &options.Spec.Labels, nil, siteOwnerRef, van.Namespace, cli.KubeClient)
 	initialConfig := qdr.AsConfigMapData(van.RouterConfig)
-	kube.NewConfigMap(types.TransportConfigMapName, &initialConfig, nil, nil, siteOwnerRef, van.Namespace, cli.KubeClient)
+	kube.NewConfigMap(types.TransportConfigMapName, &initialConfig, &options.Spec.Labels, nil, siteOwnerRef, van.Namespace, cli.KubeClient)
 
 	currentContext, cn := getCurrentContextOrDefault(ctx)
 	if cn != nil {
@@ -1561,7 +1584,7 @@ func (cli *VanClient) createIngress(site types.SiteConfig) error {
 		})
 	}
 
-	return kube.CreateIngress(types.IngressName, routes, site.Spec.IsIngressNginxIngress(), true, asOwnerReferences(site.Reference), namespace, site.Spec.IngressAnnotations, cli)
+	return kube.CreateIngress(types.IngressName, routes, site.Spec.IsIngressNginxIngress(), true, asOwnerReferences(site.Reference), namespace, site.Spec.IngressAnnotations, site.Spec.Labels, cli)
 }
 
 func (cli *VanClient) createContourProxies(site types.SiteConfig) error {
@@ -1599,7 +1622,7 @@ func (cli *VanClient) createContourProxies(site types.SiteConfig) error {
 			ServicePort: int(types.ClaimRedemptionPort),
 		})
 	}
-	return kube.CreateContourProxies(routes, asOwnerReference(site.Reference), cli.DynamicClient, namespace)
+	return kube.CreateContourProxies(routes, site.Spec.Labels, asOwnerReference(site.Reference), cli.DynamicClient, namespace)
 }
 
 func asOwnerReference(ref types.SiteConfigReference) *metav1.OwnerReference {
@@ -1643,69 +1666,8 @@ func asOwnerReferences(in types.SiteConfigReference) []metav1.OwnerReference {
 	return []metav1.OwnerReference{*ref}
 }
 
-type PrometheusInfo struct {
-	BasicAuth   bool
-	TlsAuth     bool
-	ServiceName string
-	Namespace   string
-	Port        string
-	User        string
-	Password    string
-	Hash        string
-}
-
-func scrapeConfigForPrometheus(info PrometheusInfo) string {
-	const prometheusConfig = `
-global:
-  scrape_interval:     15s
-  evaluation_interval: 15s
-alerting:
-  alertmanagers:
-  - static_configs:
-    - targets:
-rule_files:
-  # - "example-file.yml"
-scrape_configs:
-  - job_name: 'prometheus'
-    metrics_path: "/api/v1alpha1/metrics"
-    scheme: "https"
-    tls_config:
-      insecure_skip_verify: true
-    static_configs:
-    - targets: ["{{.ServiceName}}.{{.Namespace}}.svc.cluster.local:{{.Port}}"]
-`
-	var buf bytes.Buffer
-	promConfig := template.Must(template.New("promConfig").Parse(prometheusConfig))
-	promConfig.Execute(&buf, info)
-
-	return buf.String()
-}
-
-func webConfigForPrometheus(info PrometheusInfo) string {
-	const webConfig = `
-# TLS configuration.
-#{{- if .TlsAuth }}
-#tls_server_config:
-#  cert_file: /etc/tls/certs/tls.crt
-#  key_file: /etc/tls/certs/tls.key
-#{{- end}}
-#
-# Usernames and passwords required to connect to Prometheus.
-# Passwords are hashed with bcrypt: https://github.com/prometheus/exporter-toolkit/blob/master/docs/web-configuration.md#about-bcrypt
-#basic_auth_users:
-#{{- if .BasicAuth }}
-#  {{.User}}: {{.Hash}}
-#{{- end}}
-`
-	var buf bytes.Buffer
-	promConfig := template.Must(template.New("prmConfig").Parse(webConfig))
-	promConfig.Execute(&buf, info)
-
-	return buf.String()
-}
-
 func (cli *VanClient) createPrometheus(ctx context.Context, siteConfig *types.SiteConfig, van types.RouterSpec) error {
-	promInfo := PrometheusInfo{
+	promInfo := config.PrometheusInfo{
 		BasicAuth:   false,
 		TlsAuth:     false,
 		ServiceName: types.ControllerServiceName,
@@ -1723,14 +1685,14 @@ func (cli *VanClient) createPrometheus(ctx context.Context, siteConfig *types.Si
 		if siteConfig.Spec.PrometheusServer.Password != "" {
 			promInfo.Password = siteConfig.Spec.PrometheusServer.Password
 		}
-		hash, _ := hashPrometheusPassword(promInfo.Password)
+		hash, _ := config.HashPrometheusPassword(promInfo.Password)
 		promInfo.Hash = string(hash)
 	} else if siteConfig.Spec.PrometheusServer.AuthMode == string(types.PrometheusAuthModeTls) {
 		promInfo.TlsAuth = true
 	}
 	prometheusData := &map[string]string{
-		"prometheus.yml": scrapeConfigForPrometheus(promInfo),
-		"web-config.yml": webConfigForPrometheus(promInfo),
+		"prometheus.yml": config.ScrapeConfigForPrometheus(promInfo),
+		"web-config.yml": config.ScrapeWebConfigForPrometheus(promInfo),
 	}
 	siteOwnerRef := asOwnerReference(siteConfig.Reference)
 	var ownerRefs []metav1.OwnerReference
@@ -1738,7 +1700,7 @@ func (cli *VanClient) createPrometheus(ctx context.Context, siteConfig *types.Si
 		ownerRefs = []metav1.OwnerReference{*siteOwnerRef}
 	}
 
-	kube.NewConfigMap("prometheus-server-config", prometheusData, nil, nil, siteOwnerRef, van.Namespace, cli.KubeClient)
+	kube.NewConfigMap("prometheus-server-config", prometheusData, &siteConfig.Spec.Labels, nil, siteOwnerRef, van.Namespace, cli.KubeClient)
 
 	for _, sa := range van.PrometheusServer.ServiceAccounts {
 		sa.ObjectMeta.OwnerReferences = ownerRefs
diff --git a/client/router_update.go b/client/router_update.go
index 48f1c9f9..2fa6df1c 100644
--- a/client/router_update.go
+++ b/client/router_update.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/skupperproject/skupper/pkg/utils/configs"
 	"io/ioutil"
 	"log"
 	"net"
@@ -94,6 +95,7 @@ func (cli *VanClient) RouterUpdateVersionInNamespace(ctx context.Context, hup bo
 		return false, fmt.Errorf("Site (%s) is newer than library (%s); cannot update", site.Version, version.Version)
 	}
 	renameFor050 := false
+	updateSecretsFor102 := false
 	addClaimsSupport := false
 	addMultiportServices := false
 	addClusterPolicy := false
@@ -116,6 +118,7 @@ func (cli *VanClient) RouterUpdateVersionInNamespace(ctx context.Context, hup bo
 		substituteFlowCollector = utils.LessRecentThanVersion(originalVersion, "1.3.0")
 		addPrometheusServer = utils.LessRecentThanVersion(originalVersion, "1.4.0")
 		moveClaims = utils.LessRecentThanVersion(originalVersion, "1.5.0")
+		updateSecretsFor102 = utils.LessRecentThanVersion(originalVersion, "1.0.2")
 	} else {
 		originalVersion = site.Version
 	}
@@ -407,6 +410,43 @@ func (cli *VanClient) RouterUpdateVersionInNamespace(ctx context.Context, hup bo
 		return false, err
 	}
 	updateRouter := false
+
+	if updateSecretsFor102 {
+
+		secret, err := cli.KubeClient.CoreV1().Secrets(namespace).Get(context.TODO(), types.LocalClientSecret, metav1.GetOptions{})
+		if err != nil {
+			return false, err
+		}
+
+		if strings.EqualFold(string(secret.Data["connect.json"]), configs.ConnectJson(types.QualifiedServiceName(types.LocalClientSecret, namespace))) {
+
+			err = kube.DeleteSecret(types.LocalClientSecret, namespace, cli.KubeClient)
+			if err != nil {
+				return false, err
+			}
+
+			credential := types.Credential{
+				CA:          types.LocalCaSecret,
+				Name:        types.LocalClientSecret,
+				Subject:     types.LocalTransportServiceName,
+				Hosts:       []string{},
+				ConnectJson: true,
+			}
+
+			var owner *metav1.OwnerReference
+			if len(configmap.ObjectMeta.OwnerReferences) > 0 {
+				owner = &configmap.ObjectMeta.OwnerReferences[0]
+			}
+
+			_, err = kube.NewSecret(credential, owner, namespace, cli.KubeClient)
+			if err != nil {
+				return false, err
+			}
+
+			updateRouter = true
+		}
+	}
+
 	if renameFor050 {
 		// update deployment
 		// - serviceaccount
@@ -464,10 +504,6 @@ func (cli *VanClient) RouterUpdateVersionInNamespace(ctx context.Context, hup bo
 		updateRouter = true
 	}
 	if addClaimsSupport {
-		err = kube.UpdateRole(namespace, types.TransportRoleName, types.TransportPolicyRule, cli.KubeClient)
-		if err != nil {
-			return false, err
-		}
 		if !config.IsEdge() {
 			if usingRoutes {
 				err = cli.createClaimsRedemptionRoute(ctx, namespace)
@@ -478,6 +514,10 @@ func (cli *VanClient) RouterUpdateVersionInNamespace(ctx context.Context, hup bo
 		}
 	}
 	if moveClaims {
+		err = kube.UpdateRole(namespace, types.TransportRoleName, types.TransportPolicyRule, cli.KubeClient)
+		if err != nil {
+			return false, err
+		}
 		if err := cli.moveClaimsToRouterService(ctx, namespace); err != nil {
 			return false, err
 		}
@@ -1299,7 +1339,7 @@ func (cli *VanClient) moveClaimsToRouterService(ctx context.Context, namespace s
 	if err != nil {
 		return err
 	}
-	err = cli.regenerateSiteSecret(ctx, ca)
+	err = cli.regenerateSiteSecret(ctx, ca, namespace)
 	if err != nil {
 		return err
 	}
diff --git a/cmd/config-sync/collector.go b/cmd/config-sync/collector.go
new file mode 100644
index 00000000..9143bc6c
--- /dev/null
+++ b/cmd/config-sync/collector.go
@@ -0,0 +1,118 @@
+package main
+
+import (
+	"context"
+	"log"
+	"os"
+	"time"
+
+	"github.com/skupperproject/skupper/api/types"
+	"github.com/skupperproject/skupper/client"
+	"github.com/skupperproject/skupper/pkg/config"
+	"github.com/skupperproject/skupper/pkg/flow"
+	"github.com/skupperproject/skupper/pkg/kube"
+	"github.com/skupperproject/skupper/pkg/qdr"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+)
+
+func updateLockOwner(lockname, namespace string, owner *metav1.OwnerReference, cli *client.VanClient) error {
+	current, err := cli.KubeClient.CoreV1().ConfigMaps(namespace).Get(context.TODO(), lockname, metav1.GetOptions{})
+	if err != nil {
+		return err
+	}
+	if owner != nil {
+		current.ObjectMeta.OwnerReferences = []metav1.OwnerReference{
+			*owner,
+		}
+	}
+	_, err = cli.KubeClient.CoreV1().ConfigMaps(namespace).Update(context.TODO(), current, metav1.UpdateOptions{})
+	return err
+}
+
+func siteCollector(stopCh <-chan struct{}, cli *client.VanClient) *flow.FlowCollector {
+	var fc *flow.FlowCollector
+	siteData := map[string]string{}
+	platform := config.GetPlatform()
+	if platform == "" || platform == types.PlatformKubernetes {
+		current, err := kube.GetDeployment(types.TransportDeploymentName, cli.Namespace, cli.KubeClient)
+		if err != nil {
+			log.Fatal("Failed to get transport deployment", err.Error())
+		}
+		owner := kube.GetDeploymentOwnerReference(current)
+
+		existing, err := kube.NewConfigMap(types.NetworkStatusConfigMapName, &siteData, nil, nil, &owner, cli.Namespace, cli.KubeClient)
+		if err != nil && existing == nil {
+			log.Fatal("Failed to create site status config map ", err.Error())
+		}
+
+		err = updateLockOwner(types.SiteLeaderLockName, cli.Namespace, &owner, cli)
+		if err != nil {
+			log.Println("Update lock error", err.Error())
+		}
+
+		fc = flow.NewFlowCollector(flow.FlowCollectorSpec{
+			Mode:              flow.RecordStatus,
+			Namespace:         cli.Namespace,
+			Origin:            os.Getenv("SKUPPER_SITE_ID"),
+			PromReg:           nil,
+			ConnectionFactory: qdr.NewConnectionFactory("amqp://localhost:5672", nil),
+			FlowRecordTtl:     time.Minute * 15})
+
+		fc.Start(stopCh)
+	}
+	return fc
+}
+
+func runLeaderElection(lock *resourcelock.ConfigMapLock, ctx context.Context, id string, cli *client.VanClient) {
+	var stopCh chan struct{}
+	podname, _ := os.Hostname()
+	leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
+		Lock:            lock,
+		ReleaseOnCancel: true,
+		LeaseDuration:   15 * time.Second,
+		RenewDeadline:   10 * time.Second,
+		RetryPeriod:     2 * time.Second,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(c context.Context) {
+				log.Printf("COLLECTOR: Leader %s starting site collection \n", podname)
+				stopCh = make(chan struct{})
+				siteCollector(stopCh, cli)
+			},
+			OnStoppedLeading: func() {
+				// No longer the leader, transition to inactive
+				close(stopCh)
+			},
+			OnNewLeader: func(current_id string) {
+				if current_id == id {
+					// Remain as the leader
+					return
+				}
+				log.Printf("COLLECTOR: New leader for site collection is %s\n", current_id)
+			},
+		},
+	})
+}
+
+func startCollector(cli *client.VanClient) {
+	lockname := types.SiteLeaderLockName
+	namespace := cli.Namespace
+	podname, _ := os.Hostname()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	cmLock := &resourcelock.ConfigMapLock{
+		ConfigMapMeta: metav1.ObjectMeta{
+			Name:      lockname,
+			Namespace: namespace,
+		},
+		Client: cli.KubeClient.CoreV1(),
+		LockConfig: resourcelock.ResourceLockConfig{
+			Identity: podname,
+		},
+	}
+
+	runLeaderElection(cmLock, ctx, podname, cli)
+}
diff --git a/cmd/config-sync/main.go b/cmd/config-sync/main.go
index f83c520c..a370968f 100644
--- a/cmd/config-sync/main.go
+++ b/cmd/config-sync/main.go
@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/skupperproject/skupper/pkg/version"
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	corev1informer "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/informers/internalinterfaces"
@@ -50,8 +51,7 @@ func main() {
 	}
 
 	// Startup message
-	log.Printf("Config Sync")
-	log.Printf("Version: %s", version.Version)
+	log.Printf("CONFIG_SYNC: Version: %s", version.Version)
 
 	// set up signals so we handle the first shutdown signal gracefully
 	stopCh := SetupSignalHandler()
@@ -61,18 +61,25 @@ func main() {
 		log.Fatal("Error determining namespace: ", err.Error())
 	}
 
-	log.Printf("Creating kube client...")
 	cli, err := client.NewClient(namespace, "", "")
 	if err != nil {
 		log.Fatal("Error getting van client: ", err.Error())
 	}
+
+	log.Println("CONFIG_SYNC: Waiting for Skupper router to be ready")
+	_, err = kube.WaitForPodsSelectorStatus(namespace, cli.KubeClient, "skupper.io/component=router", corev1.PodRunning, time.Second*180, time.Second*5)
+	if err != nil {
+		log.Fatal("Error waiting for router pods to be ready ", err.Error())
+	}
+
 	event.StartDefaultEventStore(stopCh)
 	if claims.StartClaimVerifier(cli.KubeClient, cli.Namespace, cli, cli) {
-		log.Println("Claim verifier started")
+		log.Println("CONFIG_SYNC: Claim verifier started")
 	} else {
-		log.Println("Claim verifier not enabled")
+		log.Println("CONFIG_SYNC: Claim verifier not enabled")
 	}
-	log.Printf("Creating informer...")
+
+	log.Printf("CONFIG_SYNC: Creating ConfigMap informer...")
 	informer := corev1informer.NewFilteredConfigMapInformer(
 		cli.KubeClient,
 		cli.Namespace,
@@ -82,15 +89,18 @@ func main() {
 			options.FieldSelector = "metadata.name=skupper-internal"
 		}))
 	go informer.Run(stopCh)
-	log.Printf("Waiting for informer to sync...")
+	log.Printf("CONFIG_SYNC: Waiting for informer to sync...")
 	if ok := cache.WaitForCacheSync(stopCh, informer.HasSynced); !ok {
 		log.Fatal("Failed to wait for caches to sync")
 	}
+
 	configSync := newConfigSync(informer, cli)
-	log.Println("Starting sync loop...")
+	log.Println("CONFIG_SYNC: Starting sync controller loop...")
 	configSync.start(stopCh)
+
+	go startCollector(cli)
+
 	<-stopCh
-	log.Println("Shutting down...")
+	log.Println("CONFIG_SYNC: Shutting down...")
 	configSync.stop()
-
 }
diff --git a/cmd/flow-collector/controller.go b/cmd/flow-collector/controller.go
index 85524887..bcdb8316 100644
--- a/cmd/flow-collector/controller.go
+++ b/cmd/flow-collector/controller.go
@@ -6,7 +6,6 @@ import (
 
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 
-	"github.com/interconnectedcloud/go-amqp"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/skupperproject/skupper/pkg/certs"
 	"github.com/skupperproject/skupper/pkg/flow"
@@ -14,17 +13,19 @@ import (
 )
 
 type Controller struct {
-	origin        string
-	amqpClient    *amqp.Client
-	amqpSession   *amqp.Session
 	FlowCollector *flow.FlowCollector
 }
 
 func NewController(origin string, reg prometheus.Registerer, scheme string, host string, port string, tlsConfig *certs.TlsConfigRetriever, recordTtl time.Duration) (*Controller, error) {
 
 	controller := &Controller{
-		origin:        origin,
-		FlowCollector: flow.NewFlowCollector(origin, reg, qdr.NewConnectionFactory(scheme+"://"+host+":"+port, tlsConfig), recordTtl),
+		FlowCollector: flow.NewFlowCollector(flow.FlowCollectorSpec{
+			Mode:              flow.RecordMetrics,
+			Origin:            origin,
+			PromReg:           reg,
+			ConnectionFactory: qdr.NewConnectionFactory(scheme+"://"+host+":"+port, tlsConfig),
+			FlowRecordTtl:     recordTtl,
+		}),
 	}
 
 	return controller, nil
diff --git a/cmd/flow-collector/handlers.go b/cmd/flow-collector/handlers.go
index 61138c49..0ebad711 100644
--- a/cmd/flow-collector/handlers.go
+++ b/cmd/flow-collector/handlers.go
@@ -180,10 +180,10 @@ func (c *Controller) promqueryHandler(w http.ResponseWriter, r *http.Request) {
 
 	w.Header().Set("Content-Type", "application/json")
 	proxyResp, err := client.Do(proxyReq)
-	w.WriteHeader(proxyResp.StatusCode)
 	if err != nil {
 		log.Printf("COLLECTOR: Prometheus query error: %s\n", err.Error())
 	} else {
+		w.WriteHeader(proxyResp.StatusCode)
 		data, _ := ioutil.ReadAll(proxyResp.Body)
 		proxyResp.Body.Close()
 		fmt.Fprintf(w, "%s\n", data)
diff --git a/cmd/flow-collector/main.go b/cmd/flow-collector/main.go
index 808026c9..610ec700 100644
--- a/cmd/flow-collector/main.go
+++ b/cmd/flow-collector/main.go
@@ -205,7 +205,7 @@ func main() {
 		}
 		flowRecordTtl, _ = time.ParseDuration(os.Getenv("FLOW_RECORD_TTL"))
 		enableConsole, _ = strconv.ParseBool(os.Getenv("ENABLE_CONSOLE"))
-		prometheusUrl = os.Getenv("PROMETHEUS_URL")
+		prometheusUrl = "http://skupper-prometheus:9090/api/v1/"
 	}
 
 	tlsConfig := certs.GetTlsConfigRetriever(true, types.ControllerConfigPath+"tls.crt", types.ControllerConfigPath+"tls.key", types.ControllerConfigPath+"ca.crt")
diff --git a/cmd/service-controller/definition_monitor.go b/cmd/service-controller/definition_monitor.go
index 683ab927..eaf6d079 100644
--- a/cmd/service-controller/definition_monitor.go
+++ b/cmd/service-controller/definition_monitor.go
@@ -5,7 +5,6 @@ import (
 	jsonencoding "encoding/json"
 	"fmt"
 	"reflect"
-	"strconv"
 	"strings"
 	"time"
 
@@ -768,9 +767,26 @@ func restoreServiceDefinitions(cli *client.VanClient, name string) error {
 	}
 	if hasOriginalTargetPort(*service) {
 		updated = true
-		originalTargetPort, _ := strconv.Atoi(service.ObjectMeta.Annotations[types.OriginalTargetPortQualifier])
+		originalTargetPortMap := kube.GetOriginalTargetPorts(service)
+		restoredPorts := make([]corev1.ServicePort, 0, len(originalTargetPortMap))
+		for _, port := range service.Spec.Ports {
+			originalTargetPort, exists := originalTargetPortMap[int(port.Port)]
+			if exists {
+				port.TargetPort = intstr.FromInt(originalTargetPort)
+				restoredPorts = append(restoredPorts, port)
+				delete(originalTargetPortMap, int(port.Port))
+			}
+		}
+		for iPort, tPort := range originalTargetPortMap {
+			restoredPorts = append(restoredPorts, corev1.ServicePort{
+				Name:       fmt.Sprintf("port%d", iPort),
+				Protocol:   "TCP",
+				Port:       int32(iPort),
+				TargetPort: intstr.FromInt(tPort),
+			})
+		}
 		delete(service.ObjectMeta.Annotations, types.OriginalTargetPortQualifier)
-		service.Spec.Ports[0].TargetPort = intstr.FromInt(originalTargetPort)
+		service.Spec.Ports = restoredPorts
 	}
 	if hasOriginalAssigned(*service) {
 		updated = true
diff --git a/cmd/service-controller/links.go b/cmd/service-controller/links.go
index 4a44e293..47ef6ae4 100644
--- a/cmd/service-controller/links.go
+++ b/cmd/service-controller/links.go
@@ -12,6 +12,8 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	serializerjson "k8s.io/apimachinery/pkg/runtime/serializer/json"
+	"k8s.io/client-go/kubernetes/scheme"
 
 	"github.com/gorilla/mux"
 
@@ -171,7 +173,15 @@ func (m *LinkManager) deleteLink(name string) (bool, error) {
 }
 
 func (m *LinkManager) createLink(cost int, token []byte) error {
-	secret, err := m.cli.ConnectorCreateSecretFromData(context.Background(), types.ConnectorCreateOptions{Cost: int32(cost), SkupperNamespace: m.cli.Namespace, Yaml: token})
+	ys := serializerjson.NewYAMLSerializer(serializerjson.DefaultMetaFactory, scheme.Scheme,
+		scheme.Scheme)
+	var secret = &corev1.Secret{}
+	_, _, err := ys.Decode(token, nil, secret)
+	if err != nil {
+		return fmt.Errorf("Could not parse connection token: %w", err)
+	}
+
+	secret, err = m.cli.ConnectorCreateSecretFromData(context.Background(), types.ConnectorCreateOptions{Cost: int32(cost), SkupperNamespace: m.cli.Namespace, Secret: secret})
 	if err != nil {
 		return err
 	}
diff --git a/cmd/service-controller/policy_controller.go b/cmd/service-controller/policy_controller.go
index e17e8bab..f7f103c1 100644
--- a/cmd/service-controller/policy_controller.go
+++ b/cmd/service-controller/policy_controller.go
@@ -80,6 +80,11 @@ func (c *PolicyController) OnDelete(obj interface{}) {
 	c.enqueue(obj)
 }
 
+func (c *PolicyController) logRecord(msg string) {
+	log.Println(msg)
+	event.Record(c.name, msg)
+}
+
 func (c *PolicyController) start(stopCh <-chan struct{}) error {
 	go func() {
 		period := time.NewTicker(time.Second)
@@ -92,7 +97,7 @@ func (c *PolicyController) start(stopCh <-chan struct{}) error {
 			c.createInformer()
 			go c.informer.Run(crdCh)
 			if ok := cache.WaitForCacheSync(crdCh, c.informer.HasSynced); !ok {
-				event.Recordf(c.name, "Error waiting for cache to sync")
+				c.logRecord("Error waiting for cache to sync")
 				close(crdCh)
 				return false
 			}
@@ -101,9 +106,9 @@ func (c *PolicyController) start(stopCh <-chan struct{}) error {
 		}
 		for {
 			if !enabled && c.validator.Enabled() {
-				log.Println("Skupper policy is enabled")
+				c.logRecord("Skupper policy is enabled")
 				if !c.validator.HasPermission() {
-					log.Printf("-> No permission to read SkupperClusterPolicies")
+					c.logRecord("-> No permission to read SkupperClusterPolicies")
 				} else {
 					if informerRunning = startInformer(); !informerRunning {
 						continue
@@ -114,7 +119,7 @@ func (c *PolicyController) start(stopCh <-chan struct{}) error {
 				enabled = true
 			} else if !c.validator.Enabled() && !disabledReported {
 				disabledReported = true
-				log.Printf("Skupper policy is disabled")
+				c.logRecord("Skupper policy is disabled")
 			}
 
 			select {
@@ -123,20 +128,20 @@ func (c *PolicyController) start(stopCh <-chan struct{}) error {
 					if informerRunning {
 						close(crdCh)
 					}
-					log.Println("Skupper policy has been disabled")
+					c.logRecord("Skupper policy has been disabled")
 					// reverts what has been denied by policies
 					c.validateStateChanged()
 					enabled = false
 				} else if enabled && !informerRunning && c.validator.HasPermission() {
 					// permission has been granted, running informer
 					informerRunning = startInformer()
-					log.Println("Permission to read SkupperClusterPolicies has been granted")
+					c.logRecord("Permission to read SkupperClusterPolicies has been granted")
 				} else if enabled && informerRunning && !c.validator.HasPermission() {
 					// permission revoked, stopping informer
 					close(crdCh)
 					informerRunning = false
 					c.validateStateChanged()
-					log.Println("Permission to read SkupperClusterPolicies has been revoked")
+					c.logRecord("Permission to read SkupperClusterPolicies has been revoked")
 				}
 			case <-stopCh:
 				if enabled {
@@ -216,13 +221,6 @@ func (c *PolicyController) validateIncomingLinkStateChanged() {
 		return
 	}
 
-	// Changed to allowed
-	if allowed {
-		event.Recordf(c.name, "[%s] allowing links", source)
-	} else {
-		event.Recordf(c.name, "[%s] blocking links", source)
-	}
-
 	siteConfig, err := c.cli.SiteConfigInspect(context.Background(), nil)
 	if err != nil {
 		event.Recordf(c.name, "[%s] error retrieving site config: %v", source, err)
@@ -240,8 +238,10 @@ func (c *PolicyController) validateIncomingLinkStateChanged() {
 
 		// Changed to allowed
 		if allowed {
+			event.Recordf(c.name, "[%s] allowing links", source)
 			current.AddListener(listenerFn(siteConfig.Spec.Router))
 		} else {
+			event.Recordf(c.name, "[%s] blocking links", source)
 			delete(current.Listeners, listenerName)
 		}
 	}
@@ -380,7 +380,9 @@ func (c *PolicyController) validateServiceStateChanged() {
 		}
 		if !res.Allowed() {
 			err = c.cli.ServiceInterfaceRemove(context.Background(), service.Address)
-			if err != nil {
+			if err == nil {
+				event.Recordf(c.name, "[validateServiceStateChanged] removing service definition %s", service.Address)
+			} else if !strings.Contains(err.Error(), "not defined") {
 				event.Recordf(c.name, "[validateServiceStateChanged] error removing service definition %s: %v", service.Address, err)
 				return
 			}
diff --git a/cmd/skupper/skupper.go b/cmd/skupper/skupper.go
index 1d35ab31..16fe4ddf 100644
--- a/cmd/skupper/skupper.go
+++ b/cmd/skupper/skupper.go
@@ -61,6 +61,7 @@ type SkupperServiceClient interface {
 	UnbindFlags(cmd *cobra.Command)
 	Unexpose(cmd *cobra.Command, args []string) error
 	UnexposeFlags(cmd *cobra.Command) error
+	UnexposeArgs(cmd *cobra.Command, args []string) error
 }
 
 type SkupperDebugClient interface {
@@ -126,6 +127,8 @@ type BindOptions struct {
 	PublishNotReadyAddresses bool
 	tlsCertAuthority         string
 	Namespace                string
+	Headless                 bool
+	ProxyTuning              types.Tuning
 }
 
 func SkupperNotInstalledError(namespace string) error {
@@ -458,10 +461,25 @@ installation that can then be connected to other skupper installations`,
 	cmd.Flags().StringVarP(&routerCreateOpts.Ingress, "ingress", "", "", "Setup Skupper ingress to one of: ["+strings.Join(types.ValidIngressOptions(platform), "|")+"].")
 	cmd.Flags().StringVarP(&initFlags.routerMode, "router-mode", "", string(types.TransportModeInterior), "Skupper router-mode")
 
-	cmd.Flags().StringSliceVar(&initFlags.labels, "labels", []string{}, "Labels to add to skupper pods")
+	cmd.Flags().StringSliceVar(&initFlags.labels, "labels", []string{}, "Labels to add to resources created by skupper")
 	cmd.Flags().StringVarP(&routerLogging, "router-logging", "", "", "Logging settings for router. 'trace', 'debug', 'info' (default), 'notice', 'warning', and 'error' are valid values.")
 	cmd.Flags().StringVarP(&routerCreateOpts.Router.DebugMode, "router-debug-mode", "", "", "Enable debug mode for router ('asan' or 'gdb' are valid values)")
 
+	cmd.Flags().StringVarP(&routerCreateOpts.PrometheusServer.ExternalServer, "external-prometheus-server", "", "", "External prometheus server for metric aggregation. Valid only when --enable-flow-collector")
+	cmd.Flags().StringVarP(&routerCreateOpts.PrometheusServer.AuthMode, "prometheus-auth", "", "", "Authentication mode for skupper prometheus server. One of: 'tls', 'basic', 'unsecured'")
+	cmd.Flags().StringVarP(&routerCreateOpts.PrometheusServer.User, "prometheus-user", "", "", "Skupper prometheus user. Valid only when --prometheus-auth=basic")
+	cmd.Flags().StringVarP(&routerCreateOpts.PrometheusServer.Password, "prometheus-password", "", "", "Skupper prometheus user. Valid only when --prometheus-auth=basic")
+
+	// hide prometheus flags until external server supported
+	f := cmd.Flag("external-prometheus-server")
+	f.Hidden = true
+	f = cmd.Flag("prometheus-auth")
+	f.Hidden = true
+	f = cmd.Flag("prometheus-user")
+	f.Hidden = true
+	f = cmd.Flag("prometheus-password")
+	f.Hidden = true
+
 	cmd.Flags().SortFlags = false
 
 	// platform specific flags
@@ -498,7 +516,7 @@ func NewCmdUpdate(skupperCli SkupperSiteClient) *cobra.Command {
 		PreRun: skupperCli.NewClient,
 		RunE:   skupperCli.Update,
 	}
-	cmd.Flags().BoolVarP(&forceHup, "force-restart", "", false, "Restart skupper daemons even if image tag is not updated")
+	skupperCli.UpdateFlags(cmd)
 	return cmd
 }
 
@@ -553,7 +571,7 @@ func NewCmdUnexpose(skupperCli SkupperServiceClient) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:    "unexpose",
 		Short:  "Unexpose a set of pods previously exposed through a Skupper address",
-		Args:   skupperCli.ExposeArgs,
+		Args:   skupperCli.UnexposeArgs,
 		PreRun: skupperCli.NewClient,
 		RunE:   skupperCli.Unexpose,
 	}
diff --git a/cmd/skupper/skupper_kube_expose.go b/cmd/skupper/skupper_kube_expose.go
index 2bd7fe9b..ee71cb44 100644
--- a/cmd/skupper/skupper_kube_expose.go
+++ b/cmd/skupper/skupper_kube_expose.go
@@ -96,16 +96,7 @@ func (s *SkupperKubeService) Expose(cmd *cobra.Command, args []string) error {
 }
 
 func (s *SkupperKubeService) ExposeArgs(cmd *cobra.Command, args []string) error {
-	if len(args) < 1 || (!strings.Contains(args[0], "/") && len(args) < 2) {
-		return fmt.Errorf("expose target and name must be specified (e.g. 'skupper expose deployment <name>')")
-	}
-	if len(args) > 2 {
-		return fmt.Errorf("illegal argument: %s", args[2])
-	}
-	if len(args) > 1 && strings.Contains(args[0], "/") {
-		return fmt.Errorf("extra argument: %s", args[1])
-	}
-	return s.verifyTargetTypeFromArgs(args)
+	return s.checkCommonExposeArgs(cmd.Name(), args)
 }
 
 func (s *SkupperKubeService) ExposeFlags(cmd *cobra.Command) {
@@ -149,3 +140,20 @@ func (s *SkupperKubeService) UnexposeFlags(cmd *cobra.Command) error {
 	cmd.Use = "unexpose [deployment <name>|pods <selector>|statefulset <statefulsetname>|service <name>|deploymentconfig <name>]"
 	return nil
 }
+
+func (s *SkupperKubeService) UnexposeArgs(cmd *cobra.Command, args []string) error {
+	return s.checkCommonExposeArgs(cmd.Name(), args)
+}
+
+func (s *SkupperKubeService) checkCommonExposeArgs(command string, args []string) error {
+	if len(args) < 1 || (!strings.Contains(args[0], "/") && len(args) < 2) {
+		return fmt.Errorf("%s target and name must be specified (e.g. 'skupper %s deployment <name>')", command, command)
+	}
+	if len(args) > 2 {
+		return fmt.Errorf("illegal argument: %s", args[2])
+	}
+	if len(args) > 1 && strings.Contains(args[0], "/") {
+		return fmt.Errorf("extra argument: %s", args[1])
+	}
+	return s.verifyTargetTypeFromArgs(args)
+}
diff --git a/cmd/skupper/skupper_kube_service.go b/cmd/skupper/skupper_kube_service.go
index 2147b61a..cee22d02 100644
--- a/cmd/skupper/skupper_kube_service.go
+++ b/cmd/skupper/skupper_kube_service.go
@@ -158,6 +158,34 @@ func (s *SkupperKubeService) Bind(cmd *cobra.Command, args []string) error {
 		return targetTypeServiceTargetNamespaceError()
 	}
 
+	if bindOptions.Headless && targetType != "statefulset" {
+		return fmt.Errorf("--headless option is only valid for statefulsets")
+	}
+
+	if !bindOptions.Headless {
+		if bindOptions.ProxyTuning.Cpu != "" {
+			return fmt.Errorf("--proxy-cpu option is only valid for binding statefulsets using headless services")
+		}
+		if bindOptions.ProxyTuning.Memory != "" {
+			return fmt.Errorf("--proxy-memory option is only valid for binding statefulsets using headless services")
+		}
+		if bindOptions.ProxyTuning.CpuLimit != "" {
+			return fmt.Errorf("--proxy-cpu-limit option is only valid for binding statefulsets using headless services")
+		}
+		if bindOptions.ProxyTuning.MemoryLimit != "" {
+			return fmt.Errorf("--proxy-memory-limit option is only valid for binding statefulsets using headless services")
+		}
+		if bindOptions.ProxyTuning.Affinity != "" {
+			return fmt.Errorf("--proxy-pod-affinity option is only valid for binding statefulsets using headless services")
+		}
+		if bindOptions.ProxyTuning.AntiAffinity != "" {
+			return fmt.Errorf("--proxy-pod-antiaffinity option is only valid for binding statefulsets using headless services")
+		}
+		if bindOptions.ProxyTuning.NodeSelector != "" {
+			return fmt.Errorf("--proxy-node-selector option is only valid for binding statefulsets using headless services")
+		}
+	}
+
 	service, err := s.kube.Cli.ServiceInterfaceInspect(context.Background(), args[0])
 
 	if err != nil {
@@ -166,6 +194,26 @@ func (s *SkupperKubeService) Bind(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("Service %s not found", args[0])
 	}
 
+	if targetType == "statefulset" {
+
+		if bindOptions.Headless {
+			err := s.kube.Cli.ServiceInterfaceRemove(context.Background(), service.Address)
+			if err != nil {
+				return err
+			}
+
+			service, err = s.kube.Cli.GetHeadlessServiceConfiguration(targetName, service.Protocol, service.Address, service.Ports, bindOptions.PublishNotReadyAddresses, bindOptions.Namespace)
+			if err != nil {
+				return err
+			}
+
+			err = configureHeadlessProxy(service.Headless, &bindOptions.ProxyTuning)
+
+			return s.kube.Cli.ServiceInterfaceUpdate(context.Background(), service)
+		}
+
+	}
+
 	// validating ports
 	portMapping, err := parsePortMapping(service, bindOptions.TargetPorts)
 	if err != nil {
@@ -191,6 +239,14 @@ func (s *SkupperKubeService) BindArgs(cmd *cobra.Command, args []string) error {
 func (s *SkupperKubeService) BindFlags(cmd *cobra.Command) {
 	cmd.Flags().BoolVar(&bindOptions.PublishNotReadyAddresses, "publish-not-ready-addresses", false, "If specified, skupper will not wait for pods to be ready")
 	cmd.Flags().StringVar(&bindOptions.Namespace, "target-namespace", "", "Expose resources from a specific namespace")
+	cmd.Flags().BoolVar(&bindOptions.Headless, "headless", false, "Bind through a headless service (valid only for a statefulset target)")
+	cmd.Flags().StringVar(&bindOptions.ProxyTuning.Cpu, "proxy-cpu", "", "CPU request for router pods")
+	cmd.Flags().StringVar(&bindOptions.ProxyTuning.Memory, "proxy-memory", "", "Memory request for router pods")
+	cmd.Flags().StringVar(&bindOptions.ProxyTuning.CpuLimit, "proxy-cpu-limit", "", "CPU limit for router pods")
+	cmd.Flags().StringVar(&bindOptions.ProxyTuning.MemoryLimit, "proxy-memory-limit", "", "Memory limit for router pods")
+	cmd.Flags().StringVar(&bindOptions.ProxyTuning.NodeSelector, "proxy-node-selector", "", "Node selector to control placement of router pods")
+	cmd.Flags().StringVar(&bindOptions.ProxyTuning.Affinity, "proxy-pod-affinity", "", "Pod affinity label matches to control placement of router pods")
+	cmd.Flags().StringVar(&bindOptions.ProxyTuning.AntiAffinity, "proxy-pod-antiaffinity", "", "Pod antiaffinity label matches to control placement of router pods")
 }
 
 func (s *SkupperKubeService) Unbind(cmd *cobra.Command, args []string) error {
diff --git a/cmd/skupper/skupper_kube_site.go b/cmd/skupper/skupper_kube_site.go
index 35aa343d..af6528a3 100644
--- a/cmd/skupper/skupper_kube_site.go
+++ b/cmd/skupper/skupper_kube_site.go
@@ -173,10 +173,6 @@ func (s *SkupperKubeSite) CreateFlags(cmd *cobra.Command) {
 	cmd.Flags().StringVar(&routerCreateOpts.FlowCollector.CpuLimit, "flow-collector-cpu-limit", "", "CPU limit for flow collector pods")
 	cmd.Flags().StringVar(&routerCreateOpts.FlowCollector.MemoryLimit, "flow-collector-memory-limit", "", "Memory limit for flow collector pods")
 
-	cmd.Flags().StringVarP(&routerCreateOpts.PrometheusServer.ExternalServer, "external-prometheus-server", "", "", "External prometheus server for metric aggregation. Valid only when --enable-flow-collector")
-	cmd.Flags().StringVarP(&routerCreateOpts.PrometheusServer.AuthMode, "prometheus-auth", "", "", "Authentication mode for skupper prometheus server. One of: 'tls', 'basic', 'unsecured'")
-	cmd.Flags().StringVarP(&routerCreateOpts.PrometheusServer.User, "prometheus-user", "", "", "Skupper prometheus user. Valid only when --prometheus-auth=basic")
-	cmd.Flags().StringVarP(&routerCreateOpts.PrometheusServer.Password, "prometheus-password", "", "", "Skupper prometheus user. Valid only when --prometheus-auth=basic")
 	cmd.Flags().StringVar(&routerCreateOpts.PrometheusServer.Cpu, "prometheus-cpu", "", "CPU request for prometheus pods")
 	cmd.Flags().StringVar(&routerCreateOpts.PrometheusServer.Memory, "prometheus-memory", "", "Memory request for prometheus pods")
 	cmd.Flags().StringVar(&routerCreateOpts.PrometheusServer.CpuLimit, "prometheus-cpu-limit", "", "CPU limit for prometheus pods")
@@ -192,15 +188,6 @@ func (s *SkupperKubeSite) CreateFlags(cmd *cobra.Command) {
 	f.Hidden = true
 	f = cmd.Flag("router-disable-mutual-tls")
 	f.Hidden = true
-	// hide prometheus flags until external server supported
-	f = cmd.Flag("external-prometheus-server")
-	f.Hidden = true
-	f = cmd.Flag("prometheus-auth")
-	f.Hidden = true
-	f = cmd.Flag("prometheus-user")
-	f.Hidden = true
-	f = cmd.Flag("prometheus-password")
-	f.Hidden = true
 }
 
 func (s *SkupperKubeSite) Delete(cmd *cobra.Command, args []string) error {
@@ -319,7 +306,9 @@ func (s *SkupperKubeSite) Update(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
-func (s *SkupperKubeSite) UpdateFlags(cmd *cobra.Command) {}
+func (s *SkupperKubeSite) UpdateFlags(cmd *cobra.Command) {
+	cmd.Flags().BoolVarP(&forceHup, "force-restart", "", false, "Restart skupper daemons even if image tag is not updated")
+}
 
 func (s *SkupperKubeSite) Version(cmd *cobra.Command, args []string) error {
 	cli := s.kube.Cli
diff --git a/cmd/skupper/skupper_link.go b/cmd/skupper/skupper_link.go
index 7fd527dd..1b4d4eda 100644
--- a/cmd/skupper/skupper_link.go
+++ b/cmd/skupper/skupper_link.go
@@ -4,12 +4,15 @@ import (
 	"context"
 	"fmt"
 	"io/ioutil"
-
+	"strconv"
 	"time"
 
 	"github.com/skupperproject/skupper/pkg/utils"
 	"github.com/skupperproject/skupper/pkg/utils/formatter"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime/serializer/json"
+	"k8s.io/client-go/kubernetes/scheme"
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
 
 	"github.com/spf13/cobra"
@@ -44,7 +47,23 @@ func NewCmdLinkCreate(skupperClient SkupperLinkClient, flag string) *cobra.Comma
 			if err != nil {
 				return fmt.Errorf("Could not read connection token: %s", err.Error())
 			}
-			connectorCreateOpts.Yaml = yaml
+			costFlag := cmd.Flag("cost")
+			ys := json.NewYAMLSerializer(json.DefaultMetaFactory, scheme.Scheme,
+				scheme.Scheme)
+			var secret = &corev1.Secret{}
+			_, _, err = ys.Decode(yaml, nil, secret)
+			if err != nil {
+				return fmt.Errorf("Could not parse connection token: %w", err)
+			}
+			connectorCreateOpts.Secret = secret
+			if secret.ObjectMeta.Annotations != nil && !costFlag.Changed {
+				if costStr, ok := secret.ObjectMeta.Annotations[types.TokenCost]; ok {
+					if cost, err := strconv.Atoi(costStr); err == nil {
+						connectorCreateOpts.Cost = int32(cost)
+					}
+				}
+			}
+
 			return skupperClient.Create(cmd, args)
 		},
 	}
diff --git a/cmd/skupper/skupper_podman.go b/cmd/skupper/skupper_podman.go
index 40fd7d5f..3e799928 100644
--- a/cmd/skupper/skupper_podman.go
+++ b/cmd/skupper/skupper_podman.go
@@ -14,7 +14,7 @@ var notImplementedErr = fmt.Errorf("Not implemented")
 
 var SkupperPodmanCommands = []string{
 	"switch", "init", "delete", "status", "version", "token", "link",
-	"service", "expose", "unexpose", "revoke-access",
+	"service", "expose", "unexpose", "revoke-access", "update",
 }
 
 type SkupperPodman struct {
diff --git a/cmd/skupper/skupper_podman_link.go b/cmd/skupper/skupper_podman_link.go
index 5ac28484..1e5daef5 100644
--- a/cmd/skupper/skupper_podman_link.go
+++ b/cmd/skupper/skupper_podman_link.go
@@ -1,15 +1,10 @@
 package main
 
 import (
-	"fmt"
-
 	"github.com/skupperproject/skupper/api/types"
 	"github.com/skupperproject/skupper/pkg/domain"
 	"github.com/skupperproject/skupper/pkg/domain/podman"
 	"github.com/spf13/cobra"
-	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/runtime/serializer/json"
-	"k8s.io/client-go/kubernetes/scheme"
 )
 
 type SkupperPodmanLink struct {
@@ -19,15 +14,8 @@ type SkupperPodmanLink struct {
 
 func (s *SkupperPodmanLink) Create(cmd *cobra.Command, args []string) error {
 	// reading secret from file
-	var secret corev1.Secret
-	serializer := json.NewSerializerWithOptions(json.DefaultMetaFactory, scheme.Scheme, scheme.Scheme, json.SerializerOptions{Yaml: true})
-	_, _, err := serializer.Decode(connectorCreateOpts.Yaml, nil, &secret)
-	if err != nil {
-		return fmt.Errorf("error decoding token - %w", err)
-	}
-
 	linkHandler := podman.NewLinkHandlerPodman(s.podman.currentSite, s.podman.cli)
-	return linkHandler.Create(&secret, connectorCreateOpts.Name, int(connectorCreateOpts.Cost))
+	return linkHandler.Create(connectorCreateOpts.Secret, connectorCreateOpts.Name, int(connectorCreateOpts.Cost))
 }
 
 func (s *SkupperPodmanLink) CreateFlags(cmd *cobra.Command) {}
diff --git a/cmd/skupper/skupper_podman_service.go b/cmd/skupper/skupper_podman_service.go
index a1ffc633..20bfd14a 100644
--- a/cmd/skupper/skupper_podman_service.go
+++ b/cmd/skupper/skupper_podman_service.go
@@ -346,36 +346,7 @@ func (s *SkupperPodmanService) Expose(cmd *cobra.Command, args []string) error {
 }
 
 func (s *SkupperPodmanService) ExposeArgs(cmd *cobra.Command, args []string) error {
-	address, err := cmd.Flags().GetString("address")
-	if err != nil || address == "" {
-		return fmt.Errorf("--address is required")
-	}
-	if exposeOpts.Address != "" && len(exposeOpts.Ports) == 0 {
-		return fmt.Errorf("--port is required")
-	}
-	if len(args) < 2 {
-		return fmt.Errorf("Target type and target value must all be specified (e.g. 'skupper expose <target-type> <target-value>')")
-	}
-	if len(args) > 2 {
-		return fmt.Errorf("illegal argument: %s", args[2])
-	}
-	if !utils.StringSliceContains(ValidBindTypes, args[0]) {
-		return fmt.Errorf("invalid target type: %s - valid target types are: %s", args[0], ValidBindTypes)
-	}
-	switch args[0] {
-	case BindTypeHost:
-		host := args[1]
-		if args[1] == "" {
-			return fmt.Errorf("a hostname or IP is required")
-		}
-		if net.ParseIP(host) == nil {
-			parsedUrl, err := url.Parse("http://" + host)
-			if err != nil || parsedUrl.Hostname() != host {
-				return fmt.Errorf("invalid hostname or ip")
-			}
-		}
-	}
-	return nil
+	return s.checkCommonExposeArgs(cmd, args)
 }
 
 func (s *SkupperPodmanService) ExposeFlags(cmd *cobra.Command) {
@@ -423,3 +394,40 @@ func (s *SkupperPodmanService) UnexposeFlags(cmd *cobra.Command) error {
 	cmd.Use = "unexpose [host hostOrIP]"
 	return nil
 }
+
+func (s *SkupperPodmanService) UnexposeArgs(cmd *cobra.Command, args []string) error {
+	return s.checkCommonExposeArgs(cmd, args)
+}
+
+func (s *SkupperPodmanService) checkCommonExposeArgs(cmd *cobra.Command, args []string) error {
+	address, err := cmd.Flags().GetString("address")
+	if err != nil || address == "" {
+		return fmt.Errorf("--address is required")
+	}
+	if exposeOpts.Address != "" && len(exposeOpts.Ports) == 0 {
+		return fmt.Errorf("--port is required")
+	}
+	if len(args) < 2 {
+		return fmt.Errorf("Target type and target value must all be specified (e.g. 'skupper %s <target-type> <target-value>')", cmd.Name())
+	}
+	if len(args) > 2 {
+		return fmt.Errorf("illegal argument: %s", args[2])
+	}
+	if !utils.StringSliceContains(ValidBindTypes, args[0]) {
+		return fmt.Errorf("invalid target type: %s - valid target types are: %s", args[0], ValidBindTypes)
+	}
+	switch args[0] {
+	case BindTypeHost:
+		host := args[1]
+		if args[1] == "" {
+			return fmt.Errorf("a hostname or IP is required")
+		}
+		if net.ParseIP(host) == nil {
+			parsedUrl, err := url.Parse("http://" + host)
+			if err != nil || parsedUrl.Hostname() != host {
+				return fmt.Errorf("invalid hostname or ip")
+			}
+		}
+	}
+	return nil
+}
diff --git a/cmd/skupper/skupper_podman_site.go b/cmd/skupper/skupper_podman_site.go
index 78b697a9..b1820925 100644
--- a/cmd/skupper/skupper_podman_site.go
+++ b/cmd/skupper/skupper_podman_site.go
@@ -2,6 +2,7 @@ package main
 
 import (
 	"fmt"
+	"net"
 	"os"
 	"strings"
 
@@ -9,13 +10,16 @@ import (
 	"github.com/skupperproject/skupper/client/container"
 	"github.com/skupperproject/skupper/pkg/domain"
 	podman "github.com/skupperproject/skupper/pkg/domain/podman"
+	"github.com/skupperproject/skupper/pkg/domain/podman/update"
 	"github.com/skupperproject/skupper/pkg/qdr"
+	"github.com/skupperproject/skupper/pkg/utils"
 	"github.com/spf13/cobra"
 )
 
 type SkupperPodmanSite struct {
 	podman *SkupperPodman
 	flags  PodmanInitFlags
+	up     *domain.UpdateProcessor
 }
 
 type PodmanInitFlags struct {
@@ -67,6 +71,7 @@ func (s *SkupperPodmanSite) Create(cmd *cobra.Command, args []string) error {
 		ConsolePassword:              routerCreateOpts.Password,
 		FlowCollectorRecordTtl:       routerCreateOpts.FlowCollector.FlowRecordTtl,
 		RouterOpts:                   routerCreateOpts.Router,
+		PrometheusOpts:               routerCreateOpts.PrometheusServer,
 	}
 
 	siteHandler, err := podman.NewSitePodmanHandler(site.PodmanEndpoint)
@@ -78,7 +83,36 @@ func (s *SkupperPodmanSite) Create(cmd *cobra.Command, args []string) error {
 	if routerCreateOpts.Ingress != types.IngressNoneString {
 		// Validating ingress hosts (required as certificates must have valid hosts)
 		if len(site.IngressHosts) == 0 {
-			return fmt.Errorf("At least one ingress host is required")
+			var ingressHosts []string
+			// Get hostname of the machine
+			hostname, hostnameErr := os.Hostname()
+			if hostnameErr == nil {
+				ingressHosts = append(ingressHosts, hostname)
+			}
+			// Get all system's unicast interface addresses
+			addresses, addressesErr := net.InterfaceAddrs()
+			if addressesErr == nil {
+				for _, address := range addresses {
+					ipnet, ok := address.(*net.IPNet)
+					if ok && !ipnet.IP.IsLoopback() && ipnet.IP.To4() != nil {
+						ipv4ValidAddress := ipnet.IP.String()
+						ingressHosts = append(ingressHosts, ipv4ValidAddress)
+						// Try a reverse lookup of a valid IPv4 address
+						fqdns, err := net.LookupAddr(ipv4ValidAddress)
+						if err == nil {
+							for _, fqdn := range fqdns {
+								if !utils.StringSliceContains(ingressHosts, fqdn) {
+									ingressHosts = append(ingressHosts, fqdn)
+								}
+							}
+						}
+					}
+				}
+			}
+			if addressesErr != nil && hostnameErr != nil {
+				return fmt.Errorf("Cannot get a default ingress host")
+			}
+			site.IngressHosts = append(site.IngressHosts, ingressHosts...)
 		}
 	} else {
 		// If none is set, do not allow any ingress host (ignore those provided via CLI)
@@ -239,10 +273,22 @@ func (s *SkupperPodmanSite) Platform() types.Platform {
 }
 
 func (s *SkupperPodmanSite) Update(cmd *cobra.Command, args []string) error {
-	return notImplementedErr
+	/*
+	  Registering tasks to be analyzed against current site version
+	*/
+	// updates images for all skupper containers
+	s.up.RegisterTasks(update.NewContainerImagesTask(s.podman.cli))
+	// updates site version number (always the last one)
+	s.up.RegisterTasks(update.NewVersionUpdateTask(s.podman.cli))
+	// process eligible tasks
+	return s.up.Process(s.podman.currentSite.Version)
 }
 
-func (s *SkupperPodmanSite) UpdateFlags(cmd *cobra.Command) {}
+func (s *SkupperPodmanSite) UpdateFlags(cmd *cobra.Command) {
+	s.up = &domain.UpdateProcessor{}
+	cmd.Flags().BoolVar(&s.up.DryRun, "dry-run", false, "only prints the tasks to be performed, but does not run any action")
+	cmd.Flags().BoolVar(&s.up.Verbose, "verbose", false, "displays tasks and post tasks being executed")
+}
 
 func (s *SkupperPodmanSite) Version(cmd *cobra.Command, args []string) error {
 	site := s.podman.currentSite
diff --git a/go.mod b/go.mod
index d909d45e..6e3582e3 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,6 @@
 module github.com/skupperproject/skupper
 
-go 1.19
+go 1.20
 
 require (
 	github.com/davecgh/go-spew v1.1.1
diff --git a/go.sum b/go.sum
index 9fe5e6ef..21c288cb 100644
--- a/go.sum
+++ b/go.sum
@@ -127,7 +127,6 @@ github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoD
 github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=
 github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
-github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
 github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY=
 github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw=
diff --git a/pkg/config/prometheus-web-config.yml.template b/pkg/config/prometheus-web-config.yml.template
new file mode 100644
index 00000000..6c93e9e1
--- /dev/null
+++ b/pkg/config/prometheus-web-config.yml.template
@@ -0,0 +1,13 @@
+# TLS configuration.
+#{{- if .TlsAuth }}
+#tls_server_config:
+#  cert_file: /etc/tls/certs/tls.crt
+#  key_file: /etc/tls/certs/tls.key
+#{{- end}}
+#
+# Usernames and passwords required to connect to Prometheus.
+# Passwords are hashed with bcrypt: https://github.com/prometheus/exporter-toolkit/blob/master/docs/web-configuration.md#about-bcrypt
+#basic_auth_users:
+{{- if .BasicAuth}}
+#  {{.User}}: {{.Hash}}
+{{- end}}
diff --git a/pkg/config/prometheus.go b/pkg/config/prometheus.go
new file mode 100644
index 00000000..1d1bac87
--- /dev/null
+++ b/pkg/config/prometheus.go
@@ -0,0 +1,48 @@
+package config
+
+import (
+	"bytes"
+	_ "embed"
+	"text/template"
+
+	"golang.org/x/crypto/bcrypt"
+)
+
+var (
+	//go:embed prometheus.yml.template
+	PrometheusConfig string
+
+	//go:embed prometheus-web-config.yml.template
+	WebConfigForPrometheus string
+)
+
+type PrometheusInfo struct {
+	BasicAuth   bool
+	TlsAuth     bool
+	ServiceName string
+	Namespace   string
+	Port        string
+	User        string
+	Password    string
+	Hash        string
+}
+
+func ScrapeConfigForPrometheus(info PrometheusInfo) string {
+	var buf bytes.Buffer
+	promConfig := template.Must(template.New("promConfig").Parse(PrometheusConfig))
+	promConfig.Execute(&buf, info)
+
+	return buf.String()
+}
+
+func ScrapeWebConfigForPrometheus(info PrometheusInfo) string {
+	var buf bytes.Buffer
+	promConfig := template.Must(template.New("prmConfig").Parse(WebConfigForPrometheus))
+	promConfig.Execute(&buf, info)
+
+	return buf.String()
+}
+
+func HashPrometheusPassword(password string) ([]byte, error) {
+	return bcrypt.GenerateFromPassword([]byte(password), 14)
+}
diff --git a/pkg/config/prometheus.yml.template b/pkg/config/prometheus.yml.template
new file mode 100644
index 00000000..9d45e2f9
--- /dev/null
+++ b/pkg/config/prometheus.yml.template
@@ -0,0 +1,21 @@
+global:
+  scrape_interval:     15s
+  evaluation_interval: 15s
+alerting:
+  alertmanagers:
+    - static_configs:
+        - targets:
+rule_files:
+# - "example-file.yml"
+scrape_configs:
+  - job_name: 'prometheus'
+    metrics_path: "/api/v1alpha1/metrics"
+    scheme: "https"
+    tls_config:
+      insecure_skip_verify: true
+    static_configs:
+      {{- if .Namespace}}
+      - targets: ["{{.ServiceName}}.{{.Namespace}}.svc.cluster.local:{{.Port}}"]
+      {{- else}}
+      - targets: ["{{.ServiceName}}:{{.Port}}"]
+      {{- end}}
diff --git a/pkg/domain/link.go b/pkg/domain/link.go
index cc85726f..ec5fa0af 100644
--- a/pkg/domain/link.go
+++ b/pkg/domain/link.go
@@ -110,10 +110,14 @@ func (c *ClaimRedeemer) RedeemClaim(claim *corev1.Secret) error {
 	request.URL.RawQuery = query.Encode()
 	response, err := client.Do(request)
 	if err != nil {
+		fmt.Printf("Claim request failed on invoke: %s", err)
+		fmt.Println()
 		return c.handleError(claim, err.Error(), false)
 	}
 	body, err := ioutil.ReadAll(response.Body)
 	if err != nil {
+		fmt.Printf("Claim request failed on reading body: %s", err)
+		fmt.Println()
 		return c.handleError(claim, err.Error(), false)
 	}
 	if response.StatusCode != http.StatusOK {
diff --git a/pkg/domain/podman/common.go b/pkg/domain/podman/common.go
index 8c04edfc..35e915c6 100644
--- a/pkg/domain/podman/common.go
+++ b/pkg/domain/podman/common.go
@@ -2,9 +2,6 @@ package podman
 
 import (
 	"fmt"
-	"os"
-	"os/user"
-	"strings"
 
 	"github.com/skupperproject/skupper/pkg/utils"
 
@@ -16,9 +13,9 @@ const (
 )
 
 var (
-	Username                = readUsername()
+	Username                = utils.ReadUsername()
 	SkupperContainerVolumes = []string{"skupper-services", "skupper-local-server", "skupper-internal", "skupper-site-server", SharedTlsCertificates,
-		types.ConsoleServerSecret, types.ConsoleUsersSecret}
+		types.ConsoleServerSecret, types.ConsoleUsersSecret, "prometheus-server-config", "prometheus-storage-volume"}
 )
 
 func OwnedBySkupper(resource string, labels map[string]string) error {
@@ -31,11 +28,3 @@ func OwnedBySkupper(resource string, labels map[string]string) error {
 	}
 	return nil
 }
-
-func readUsername() string {
-	u, err := user.Current()
-	if err != nil {
-		return utils.DefaultStr(os.Getenv("USER"), os.Getenv("USERNAME"))
-	}
-	return strings.Join(strings.Fields(u.Username), "")
-}
diff --git a/pkg/domain/podman/component.go b/pkg/domain/podman/component.go
index 0d721349..177594be 100644
--- a/pkg/domain/podman/component.go
+++ b/pkg/domain/podman/component.go
@@ -67,6 +67,13 @@ func (s *SkupperComponentHandler) Get(name string) (domain.SkupperComponent, err
 			Labels:        c.Labels,
 			SiteIngresses: siteIngresses,
 		}
+	case types.PrometheusDeploymentName:
+		component = &domain.Prometheus{
+			Image:         c.Image,
+			Env:           c.Env,
+			Labels:        c.Labels,
+			SiteIngresses: siteIngresses,
+		}
 	default:
 		return nil, fmt.Errorf("invalid component: %s", componentName)
 	}
diff --git a/pkg/domain/podman/controller/site_beacons.go b/pkg/domain/podman/controller/site_beacons.go
index d71ef713..12b99905 100644
--- a/pkg/domain/podman/controller/site_beacons.go
+++ b/pkg/domain/podman/controller/site_beacons.go
@@ -270,7 +270,8 @@ func (p *ContainerProcessInformer) containerToProcess(cc *container.Container) *
 	if ipAddress != "" {
 		pp.SourceHost = &ipAddress
 	}
-	if container.IsOwnedBySkupper(cc.Labels) {
+	_, isServiceContainer := cc.Labels[types.AddressQualifier]
+	if container.IsOwnedBySkupper(cc.Labels) && !isServiceContainer {
 		pp.ProcessRole = &flow.Internal
 	} else {
 		pp.ProcessRole = &flow.External
diff --git a/pkg/domain/podman/site.go b/pkg/domain/podman/site.go
index 70411257..499dd3d0 100644
--- a/pkg/domain/podman/site.go
+++ b/pkg/domain/podman/site.go
@@ -16,6 +16,7 @@ import (
 	"github.com/skupperproject/skupper/pkg/qdr"
 	"github.com/skupperproject/skupper/pkg/utils"
 	"github.com/skupperproject/skupper/pkg/version"
+	yaml "gopkg.in/yaml.v3"
 	"k8s.io/apimachinery/pkg/util/rand"
 )
 
@@ -36,6 +37,7 @@ type Site struct {
 	ConsolePassword              string
 	FlowCollectorRecordTtl       time.Duration
 	RouterOpts                   types.RouterOptions
+	PrometheusOpts               types.PrometheusServerOptions
 }
 
 func (s *Site) GetPlatform() string {
@@ -63,6 +65,13 @@ type SiteHandler struct {
 	endpoint string
 }
 
+func NewSitePodmanHandlerFromCli(cli *podman.PodmanRestClient) *SiteHandler {
+	return &SiteHandler{
+		cli:      cli,
+		endpoint: cli.GetEndpoint(),
+	}
+}
+
 func NewSitePodmanHandler(endpoint string) (*SiteHandler, error) {
 	if endpoint == "" {
 		podmanCfg, err := NewPodmanConfigFileHandler().GetConfig()
@@ -120,6 +129,7 @@ func (s *SiteHandler) ConfigurePodmanDeployments(site *Site) {
 	site.Deployments = append(site.Deployments, s.prepareControllerDeployment(site))
 	if site.EnableFlowCollector {
 		site.Deployments = append(site.Deployments, s.prepareFlowCollectorDeployment(site))
+		site.Deployments = append(site.Deployments, s.preparePrometheusDeployment(site))
 	}
 }
 
@@ -293,6 +303,11 @@ func (s *SiteHandler) Create(site domain.Site) error {
 		return err
 	}
 
+	// Create prometheus config
+	if err = s.createPrometheusConfigFiles(podmanSite); err != nil {
+		return err
+	}
+
 	// Deploy container(s)
 	deployHandler := NewSkupperDeploymentHandlerPodman(s.cli)
 	for _, depl := range podmanSite.GetDeployments() {
@@ -333,7 +348,6 @@ func (s *SiteHandler) canCreate(site *Site) error {
 	if s.cli == nil {
 		cli, err := podman.NewPodmanClient(site.PodmanEndpoint, "")
 		if err != nil {
-			// TODO try to start podman's user service instance?
 			return fmt.Errorf("unable to communicate with podman service through %s - %v", site.PodmanEndpoint, err)
 		}
 		s.cli = cli
@@ -514,11 +528,16 @@ func (s *SiteHandler) Get() (domain.Site, error) {
 				site.FlowCollectorRecordTtl, _ = time.ParseDuration(c.Env["FLOW_RECORD_TTL"])
 				user, password, err := s.getConsoleUserPass()
 				if err != nil {
-					fmt.Println("error retrieving console user and password - %w", err)
+					fmt.Println("error retrieving console user and password -", err)
 				}
 				site.ConsoleUser = user
 				site.ConsolePassword = password
 			case *domain.Controller:
+			case *domain.Prometheus:
+				site.PrometheusOpts, err = s.getPrometheusServerOptions()
+				if err != nil {
+					fmt.Println("error retrieving prometheus options -", err)
+				}
 			}
 		}
 	}
@@ -787,3 +806,101 @@ func (s *SiteHandler) prepareControllerDeployment(site *Site) *SkupperDeployment
 
 	return ctrlDeployment
 }
+
+func (s *SiteHandler) preparePrometheusDeployment(site *Site) domain.SkupperDeployment {
+	// Prometheus Server Deployment
+	volumeMounts := map[string]string{
+		"prometheus-server-config":  "/etc/prometheus",
+		"prometheus-storage-volume": "/prometheus",
+	}
+	prometheusComponent := &domain.Prometheus{
+		// TODO ADD Labels
+		Labels: map[string]string{},
+	}
+	prometheusDeployment := &SkupperDeployment{
+		Name: types.PrometheusDeploymentName,
+		SkupperDeploymentCommon: &domain.SkupperDeploymentCommon{
+			Components: []domain.SkupperComponent{
+				prometheusComponent,
+			},
+		},
+		Aliases:        []string{types.PrometheusDeploymentName},
+		VolumeMounts:   volumeMounts,
+		Networks:       []string{site.ContainerNetwork},
+		SELinuxDisable: true,
+	}
+	return prometheusDeployment
+}
+
+func (s *SiteHandler) getPrometheusServerOptions() (types.PrometheusServerOptions, error) {
+	var prometheusConfig types.PrometheusServerOptions
+	v, err := s.cli.VolumeInspect("prometheus-server-config")
+	if err != nil {
+		return prometheusConfig, err
+	}
+	prometheusConfigStr, err := v.ReadFile("skupper-prometheus.yml")
+	if err != nil {
+		return prometheusConfig, fmt.Errorf("error reading skupper-prometheus.yml - %s", err)
+	}
+	err = yaml.Unmarshal([]byte(prometheusConfigStr), &prometheusConfig)
+	if err != nil {
+		return prometheusConfig, fmt.Errorf("error parsing prometheus options - %s", err)
+	}
+	return prometheusConfig, nil
+}
+
+func (s *SiteHandler) savePrometheusServerConfigFile(name string, data interface{}) error {
+	v, err := s.cli.VolumeInspect("prometheus-server-config")
+	if err != nil {
+		return err
+	}
+	var dataStr string
+	var ok bool
+	if dataStr, ok = data.(string); !ok {
+		var yamlData []byte
+		yamlData, err = yaml.Marshal(data)
+		if err != nil {
+			return fmt.Errorf("error serializing prometheus options - %s", err)
+		}
+		dataStr = string(yamlData)
+	}
+	_, err = v.CreateFile(name, []byte(dataStr), true)
+	return err
+}
+
+func (s *SiteHandler) createPrometheusConfigFiles(site *Site) error {
+	promInfo := config.PrometheusInfo{
+		BasicAuth:   false,
+		TlsAuth:     false,
+		ServiceName: types.FlowCollectorContainerName,
+		Port:        strconv.Itoa(int(types.FlowCollectorDefaultServicePort)),
+		User:        utils.DefaultStr(site.PrometheusOpts.User, "admin"),
+		Password:    utils.DefaultStr(site.PrometheusOpts.Password, "admin"),
+		Hash:        "",
+	}
+
+	if site.PrometheusOpts.AuthMode == string(types.PrometheusAuthModeBasic) {
+		promInfo.BasicAuth = true
+		promInfo.User = site.PrometheusOpts.User
+		promInfo.Password = site.PrometheusOpts.Password
+		hash, _ := config.HashPrometheusPassword(promInfo.Password)
+		promInfo.Hash = string(hash)
+	} else if site.PrometheusOpts.AuthMode == string(types.PrometheusAuthModeTls) {
+		promInfo.TlsAuth = true
+	}
+
+	prometheusConfigFiles := map[string]interface{}{
+		"skupper-prometheus.yml": site.PrometheusOpts,
+		"prometheus.yml":         config.ScrapeConfigForPrometheus(promInfo),
+		"web-config.yml":         config.ScrapeWebConfigForPrometheus(promInfo),
+	}
+
+	var err error
+	for name, data := range prometheusConfigFiles {
+		err = s.savePrometheusServerConfigFile(name, data)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/pkg/domain/podman/update/common.go b/pkg/domain/podman/update/common.go
new file mode 100644
index 00000000..ced175d6
--- /dev/null
+++ b/pkg/domain/podman/update/common.go
@@ -0,0 +1,165 @@
+package update
+
+import (
+	"fmt"
+
+	"github.com/skupperproject/skupper/api/types"
+	clientpodman "github.com/skupperproject/skupper/client/podman"
+	"github.com/skupperproject/skupper/pkg/domain"
+	"github.com/skupperproject/skupper/pkg/domain/podman"
+	"github.com/skupperproject/skupper/pkg/images"
+	"github.com/skupperproject/skupper/pkg/utils"
+	"github.com/skupperproject/skupper/pkg/version"
+)
+
+func NewVersionUpdateTask(cli *clientpodman.PodmanRestClient) *VersionUpdateTask {
+	return &VersionUpdateTask{
+		cli:     cli,
+		version: version.Version,
+	}
+}
+
+type VersionUpdateTask struct {
+	cli     *clientpodman.PodmanRestClient
+	version string
+}
+
+func (v *VersionUpdateTask) Info() string {
+	return "Updates current version number"
+}
+
+func (v *VersionUpdateTask) AppliesTo(siteVersion string) bool {
+	curVersion := utils.ParseVersion(siteVersion)
+	return !(&curVersion).IsUndefined() && utils.LessRecentThanVersion(siteVersion, v.version)
+}
+
+func (v *VersionUpdateTask) Version() string {
+	return "*"
+}
+
+func (v *VersionUpdateTask) Priority() domain.UpdatePriority {
+	return domain.PriorityLast
+}
+
+func (v *VersionUpdateTask) Run() *domain.UpdateResult {
+	var res = &domain.UpdateResult{}
+	ch := podman.NewRouterConfigHandlerPodman(v.cli)
+	cfg, err := ch.GetRouterConfig()
+	if err != nil {
+		res.AddErrors(fmt.Errorf("error retrieving router config: %s", err))
+		return res
+	}
+	siteMeta := cfg.GetSiteMetadata()
+	siteMeta.Version = v.version
+	cfg.SetSiteMetadata(&siteMeta)
+	if err = ch.SaveRouterConfig(cfg); err != nil {
+		res.AddErrors(fmt.Errorf("error saving router config: %s", err))
+		return res
+	}
+	res.AddChange(fmt.Sprintf("updated site version to %s", v.version))
+	return res
+}
+
+func NewContainerImagesTask(cli *clientpodman.PodmanRestClient) *ContainerImagesTask {
+	return &ContainerImagesTask{
+		cli:     cli,
+		version: version.Version,
+	}
+}
+
+type ContainerImagesTask struct {
+	cli     *clientpodman.PodmanRestClient
+	version string
+}
+
+func (u *ContainerImagesTask) Info() string {
+	return "Updates skupper podman container images"
+}
+
+func (u *ContainerImagesTask) AppliesTo(siteVersion string) bool {
+	curVersion := utils.ParseVersion(siteVersion)
+	return !(&curVersion).IsUndefined() && utils.LessRecentThanVersion(siteVersion, u.version)
+}
+
+func (u *ContainerImagesTask) Version() string {
+	return "*"
+}
+
+func (u *ContainerImagesTask) Priority() domain.UpdatePriority {
+	return domain.PriorityNormal
+}
+
+func (u *ContainerImagesTask) Run() *domain.UpdateResult {
+	var result = &domain.UpdateResult{}
+	sh := podman.NewSitePodmanHandlerFromCli(u.cli)
+	site, err := sh.Get()
+	if err != nil {
+		result.AddErrors(fmt.Errorf("error retrieving site info: %s", err))
+		return result
+	}
+	// updating images for all running components
+	for _, dep := range site.GetDeployments() {
+		for _, cmp := range dep.GetComponents() {
+			var image string
+			switch cmp.Name() {
+			case types.TransportDeploymentName:
+				image = images.GetRouterImageName()
+			case types.FlowCollectorContainerName:
+				image = images.GetFlowCollectorImageName()
+			case types.ControllerPodmanContainerName:
+				image = images.GetControllerPodmanImageName()
+			case types.PrometheusDeploymentName:
+				image = images.GetPrometheusServerImageName()
+			}
+			if image != cmp.GetImage() {
+				_, err = u.cli.ContainerUpdateImage(cmp.Name(), image)
+				if err != nil {
+					result.AddErrors(fmt.Errorf("error updating container: %s - image: %s - %s",
+						cmp.Name(), image, err))
+					return result
+				}
+				result.AddChange(fmt.Sprintf("container updated: %s - image: %s", cmp.Name(), image))
+			}
+		}
+	}
+	// updating service containers
+	updSvcResult := u.updateServiceContainers()
+	if len(updSvcResult.Errors) > 0 {
+		result.AddErrors(updSvcResult.Errors...)
+		return result
+	}
+	if updSvcResult.Changed() {
+		result.AddChange(updSvcResult.GetChanges()...)
+	}
+	return result
+}
+
+func (u *ContainerImagesTask) updateServiceContainers() domain.UpdateResult {
+	var result domain.UpdateResult
+	sh := podman.NewServiceHandlerPodman(u.cli)
+	services, err := sh.List()
+	if err != nil {
+		result.AddErrors(fmt.Errorf("error listing services: %s", err))
+		return result
+	}
+	for _, svc := range services {
+		svcPodman := svc.(*podman.Service)
+		c, err := u.cli.ContainerInspect(svcPodman.ContainerName)
+		if err != nil {
+			result.AddErrors(fmt.Errorf("error retrieving container info for %s: %s",
+				svcPodman.ContainerName, err))
+			return result
+		}
+		if c.Image != images.GetRouterImageName() {
+			_, err = u.cli.ContainerUpdateImage(svcPodman.ContainerName, images.GetRouterImageName())
+			if err != nil {
+				result.AddErrors(fmt.Errorf("error updating service container image for %s: %s",
+					svcPodman.ContainerName, err))
+				return result
+			}
+			result.AddChange(fmt.Sprintf("service container updated: %s - image: %s",
+				svcPodman.ContainerName, images.GetRouterImageName()))
+		}
+	}
+	return result
+}
diff --git a/pkg/domain/podman/update/skrouterd.json b/pkg/domain/podman/update/skrouterd.json
new file mode 100644
index 00000000..89cc965d
--- /dev/null
+++ b/pkg/domain/podman/update/skrouterd.json
@@ -0,0 +1,99 @@
+[
+  [
+    "router",
+    {
+      "id": "fedora-myuser",
+      "mode": "interior",
+      "helloMaxAgeSeconds": "3",
+      "metadata": "{\"id\":\"fd70682b-c6fc-472d-8b6d-f187a167cd4e\",\"version\":\"1.4.2\",\"platform\":\"podman\"}"
+    }
+  ],
+  [
+    "sslProfile",
+    {
+      "name": "skupper-amqps",
+      "certFile": "/etc/skupper-router-certs/skupper-amqps/tls.crt",
+      "privateKeyFile": "/etc/skupper-router-certs/skupper-amqps/tls.key",
+      "caCertFile": "/etc/skupper-router-certs/skupper-amqps/ca.crt"
+    }
+  ],
+  [
+    "sslProfile",
+    {
+      "name": "skupper-service-client",
+      "caCertFile": "/etc/skupper-router-certs/skupper-service-client/ca.crt"
+    }
+  ],
+  [
+    "sslProfile",
+    {
+      "name": "skupper-internal",
+      "certFile": "/etc/skupper-router-certs/skupper-internal/tls.crt",
+      "privateKeyFile": "/etc/skupper-router-certs/skupper-internal/tls.key",
+      "caCertFile": "/etc/skupper-router-certs/skupper-internal/ca.crt"
+    }
+  ],
+  [
+    "listener",
+    {
+      "name": "edge-listener",
+      "role": "edge",
+      "port": 45671,
+      "sslProfile": "skupper-internal",
+      "saslMechanisms": "EXTERNAL",
+      "authenticatePeer": true,
+      "maxFrameSize": 16384,
+      "maxSessionFrames": 640
+    }
+  ],
+  [
+    "listener",
+    {
+      "name": "@9090",
+      "role": "normal",
+      "port": 9090,
+      "http": true,
+      "httpRootDir": "disabled",
+      "healthz": true,
+      "metrics": true
+    }
+  ],
+  [
+    "listener",
+    {
+      "name": "amqp",
+      "host": "localhost",
+      "port": 5672
+    }
+  ],
+  [
+    "listener",
+    {
+      "name": "amqps",
+      "port": 5671,
+      "sslProfile": "skupper-amqps",
+      "saslMechanisms": "EXTERNAL",
+      "authenticatePeer": true
+    }
+  ],
+  [
+    "listener",
+    {
+      "name": "interior-listener",
+      "role": "inter-router",
+      "port": 55671,
+      "sslProfile": "skupper-internal",
+      "saslMechanisms": "EXTERNAL",
+      "authenticatePeer": true,
+      "maxFrameSize": 16384,
+      "maxSessionFrames": 640
+    }
+  ],
+  [
+    "address",
+    {
+      "prefix": "mc",
+      "distribution": "multicast"
+    }
+  ]
+]
diff --git a/pkg/domain/podman/update/skupper-services.json b/pkg/domain/podman/update/skupper-services.json
new file mode 100644
index 00000000..db56b574
--- /dev/null
+++ b/pkg/domain/podman/update/skupper-services.json
@@ -0,0 +1,1 @@
+{"nginx":{"address":"nginx","protocol":"tcp","ports":[8080],"exposeIngress":"","targets":[{"name":"*domain.EgressResolverHost={\"host\":\"nginx\",\"ports\":{\"8080\":8080}}","targetPorts":{"8080":8080},"service":"nginx"}]}}
\ No newline at end of file
diff --git a/pkg/domain/skupper_component.go b/pkg/domain/skupper_component.go
index 5075756d..82c70962 100644
--- a/pkg/domain/skupper_component.go
+++ b/pkg/domain/skupper_component.go
@@ -3,6 +3,7 @@ package domain
 import (
 	"github.com/skupperproject/skupper/api/types"
 	"github.com/skupperproject/skupper/pkg/images"
+	"github.com/skupperproject/skupper/test/utils"
 )
 
 type SkupperComponent interface {
@@ -31,7 +32,7 @@ func (r *Router) Name() string {
 }
 
 func (r *Router) GetImage() string {
-	return images.GetRouterImageName()
+	return utils.StrDefault(images.GetRouterImageName(), r.Image)
 }
 
 func (r *Router) SetImage(image string) {
@@ -68,7 +69,7 @@ func (r *FlowCollector) Name() string {
 }
 
 func (r *FlowCollector) GetImage() string {
-	return images.GetFlowCollectorImageName()
+	return utils.StrDefault(images.GetFlowCollectorImageName(), r.Image)
 }
 
 func (r *FlowCollector) SetImage(image string) {
@@ -105,7 +106,7 @@ func (s *Controller) Name() string {
 }
 
 func (s *Controller) GetImage() string {
-	return images.GetControllerPodmanImageName()
+	return utils.StrDefault(images.GetControllerPodmanImageName(), s.Image)
 }
 
 func (s *Controller) SetImage(image string) {
@@ -129,3 +130,40 @@ func (s *Controller) GetLabels() map[string]string {
 func (s *Controller) GetSiteIngresses() []SiteIngress {
 	return s.SiteIngresses
 }
+
+type Prometheus struct {
+	Image         string
+	Env           map[string]string
+	Labels        map[string]string
+	SiteIngresses []SiteIngress
+}
+
+func (s *Prometheus) Name() string {
+	return types.PrometheusDeploymentName
+}
+
+func (s *Prometheus) GetImage() string {
+	return images.GetPrometheusServerImageName()
+}
+
+func (s *Prometheus) SetImage(image string) {
+	s.Image = image
+}
+
+func (s *Prometheus) GetEnv() map[string]string {
+	if s.Env == nil {
+		s.Env = map[string]string{}
+	}
+	return s.Env
+}
+
+func (s *Prometheus) GetLabels() map[string]string {
+	if s.Labels == nil {
+		s.Labels = map[string]string{}
+	}
+	return s.Labels
+}
+
+func (s *Prometheus) GetSiteIngresses() []SiteIngress {
+	return s.SiteIngresses
+}
diff --git a/pkg/domain/update.go b/pkg/domain/update.go
new file mode 100644
index 00000000..d65974c6
--- /dev/null
+++ b/pkg/domain/update.go
@@ -0,0 +1,208 @@
+package domain
+
+import (
+	"fmt"
+	"reflect"
+	"sort"
+
+	"github.com/skupperproject/skupper/pkg/utils"
+)
+
+type UpdatePriority int
+
+const (
+	PriorityFirst UpdatePriority = iota
+	PriorityNormal
+	PriorityLast
+)
+
+// UpdateResult contains errors, warnings and changes applied by tasks.
+// It also contains a set of PostTasks that can be populated by tasks.
+// The PostTasks will only be evaluated after all main tasks complete.
+type UpdateResult struct {
+	Errors    []error
+	warnings  []string
+	changes   []string
+	PostTasks []UpdateTask
+}
+
+func (u *UpdateResult) AddErrors(err ...error) {
+	u.Errors = append(u.Errors, err...)
+}
+
+func (u *UpdateResult) AddWarnings(msg ...string) {
+	u.warnings = append(u.warnings, msg...)
+}
+
+func (u *UpdateResult) GetWarnings() []string {
+	return u.warnings
+}
+
+func (u *UpdateResult) Changed() bool {
+	return len(u.changes) > 0
+}
+
+func (u *UpdateResult) AddChange(change ...string) {
+	u.changes = append(u.changes, change...)
+}
+
+func (u *UpdateResult) GetChanges() []string {
+	return u.changes
+}
+
+type UpdateTask interface {
+	// Info describes the update task
+	Info() string
+	// AppliesTo determines whether it should be executed or not
+	AppliesTo(siteVersion string) bool
+	// Version returns the version it has been introduced (use * when not constrained to a given version)
+	Version() string
+	// Priority determines how tasks will be sorted by the UpdateProcessor
+	Priority() UpdatePriority
+	// Run method is where the update task is done
+	Run() *UpdateResult
+}
+
+type UpdateProcessor struct {
+	tasks     []UpdateTask
+	postTasks []UpdateTask
+	changes   []string
+	Verbose   bool
+	DryRun    bool
+}
+
+func (p *UpdateProcessor) Println(a ...any) {
+	if p.Verbose {
+		fmt.Println(a...)
+	}
+}
+
+func (p *UpdateProcessor) Printf(format string, a ...any) {
+	if p.Verbose {
+		fmt.Printf(format, a...)
+	}
+}
+
+// RegisterTasks registers tasks to be evaluated and processed
+func (p *UpdateProcessor) RegisterTasks(task ...UpdateTask) {
+	p.tasks = append(p.tasks, task...)
+}
+
+// registerPostTasks register post tasks to be evaluated and processed
+// after all tasks have completed.
+// The Post Tasks are processed only once.
+func (p *UpdateProcessor) registerPostTasks(tasks ...UpdateTask) {
+	for _, task := range tasks {
+		add := true
+		for _, postTask := range p.postTasks {
+			if task == postTask {
+				add = false
+				break
+			}
+		}
+		if add {
+			p.postTasks = append(p.postTasks, task)
+		}
+	}
+}
+
+func (p *UpdateProcessor) Process(siteVersion string) error {
+	err := p.process(siteVersion, p.tasks)
+	if err != nil {
+		return err
+	}
+	if len(p.postTasks) > 0 {
+		p.Println("Post update tasks")
+		return p.process(siteVersion, p.postTasks)
+	}
+	if len(p.changes) > 0 {
+		fmt.Println("Skupper is now updated for '" + utils.ReadUsername() + "'.")
+	} else {
+		fmt.Println("No update required for '" + utils.ReadUsername() + "'.")
+	}
+	return nil
+}
+
+func (p *UpdateProcessor) process(siteVersion string, tasks []UpdateTask) error {
+	var validTasks []UpdateTask
+	title := "Task"
+	postTasks := false
+	if reflect.DeepEqual(tasks, p.postTasks) {
+		title = "Post task"
+		postTasks = true
+	}
+	for _, task := range tasks {
+		if task.AppliesTo(siteVersion) {
+			validTasks = append(validTasks, task)
+		}
+	}
+	totalTasks := len(validTasks)
+
+	if totalTasks > 0 {
+		// sorting update tasks by version and priority
+		sort.SliceStable(validTasks, func(i, j int) bool {
+			t1 := validTasks[i]
+			t2 := validTasks[j]
+			sameVersion := utils.EquivalentVersion(t1.Version(), t2.Version())
+			if !sameVersion && (t1.Version() != "*" && t2.Version() != "*") {
+				return utils.LessRecentThanVersion(t1.Version(), t2.Version())
+			} else if t1.Version() == "*" {
+				return false
+			} else if t2.Version() == "*" {
+				return true
+			}
+			return t1.Priority() < t2.Priority()
+		})
+	} else {
+		p.Println("Nothing to do")
+		return nil
+	}
+
+	// running update tasks
+	var errors []error
+	var warnings []string
+	for i, task := range validTasks {
+		version := task.Version()
+		if version == "*" {
+			version = "all"
+		}
+		p.Printf("%s: %d/%d - version: %s - priority: %d - %s\n",
+			title, i+1, totalTasks, version, task.Priority(), task.Info())
+		var result *UpdateResult
+		if !p.DryRun {
+			result = task.Run()
+			if result == nil {
+				result = &UpdateResult{}
+			}
+		} else {
+			result = &UpdateResult{}
+		}
+		if len(result.warnings) > 0 {
+			warnings = append(warnings, result.warnings...)
+			p.Printf("  -> WARNING: %v\n", result.warnings)
+		}
+		if len(result.Errors) > 0 {
+			errors = append(errors, result.Errors...)
+			p.Printf("  -> ERROR: %v\n", result.Errors)
+			p.Println("     unable to proceed")
+			return fmt.Errorf("unable to proceed due to the following errors (%d): %v", len(errors), errors)
+		}
+		// post tasks are only processed once, for main tasks
+		if !postTasks && len(result.PostTasks) > 0 {
+			p.registerPostTasks(result.PostTasks...)
+		}
+		if len(result.GetChanges()) > 0 {
+			p.changes = append(p.changes, result.GetChanges()...)
+		}
+		p.Println()
+	}
+
+	// warnings are considered as informative only and will not
+	// cause the update command to fail (rc != 0)
+	if !postTasks && len(warnings) > 0 {
+		errMsg := fmt.Sprintf("Update completed with (%d) warnings: %v", len(warnings), warnings)
+		p.Println(errMsg)
+	}
+
+	return nil
+}
diff --git a/pkg/flow/collector.go b/pkg/flow/collector.go
index 14c704f1..c52280c0 100644
--- a/pkg/flow/collector.go
+++ b/pkg/flow/collector.go
@@ -3,7 +3,6 @@ package flow
 import (
 	"bytes"
 	"encoding/gob"
-	"encoding/json"
 	"log"
 	"net/http"
 	"time"
@@ -35,9 +34,8 @@ type ApiResponse struct {
 
 type eventSource struct {
 	EventSourceRecord
-	nonFlowReceiver *receiver
-	flowReceiver    *receiver
-	send            *senderDirect
+	receivers []*receiver
+	send      *senderDirect
 }
 
 type collectorMetrics struct {
@@ -146,8 +144,26 @@ type FlowToPairRecord struct {
 	created   uint64
 }
 
+type CollectorMode int
+
+const (
+	RecordStatus CollectorMode = iota
+	RecordMetrics
+)
+
+type FlowCollectorSpec struct {
+	Mode              CollectorMode
+	Namespace         string
+	Origin            string
+	PromReg           prometheus.Registerer
+	ConnectionFactory messaging.ConnectionFactory
+	FlowRecordTtl     time.Duration
+}
+
 type FlowCollector struct {
+	mode                    CollectorMode
 	origin                  string
+	namespace               string
 	startTime               uint64
 	Collector               CollectorRecord
 	connectionFactory       messaging.ConnectionFactory
@@ -155,12 +171,12 @@ type FlowCollector struct {
 	prometheusReg           prometheus.Registerer
 	metrics                 *collectorMetrics
 	beaconsIncoming         chan []interface{}
+	heartbeatsIncoming      chan []interface{}
 	recordsIncoming         chan []interface{}
 	Request                 chan ApiRequest
 	Response                chan ApiResponse
 	eventSources            map[string]*eventSource
-	receivers               map[string]*receiver
-	senders                 map[string]*senderDirect
+	beaconReceiver          *receiver
 	pendingFlush            map[string]*senderDirect
 	Beacons                 map[string]*BeaconRecord
 	Sites                   map[string]*SiteRecord
@@ -193,20 +209,21 @@ func getTtl(ttl time.Duration) time.Duration {
 	return ttl
 }
 
-func NewFlowCollector(origin string, reg prometheus.Registerer, connectionFactory messaging.ConnectionFactory, recordTtl time.Duration) *FlowCollector {
+func NewFlowCollector(spec FlowCollectorSpec) *FlowCollector {
 	fc := &FlowCollector{
-		origin:                  origin,
+		mode:                    spec.Mode,
+		namespace:               spec.Namespace,
+		origin:                  spec.Origin,
 		startTime:               uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
-		connectionFactory:       connectionFactory,
-		recordTtl:               getTtl(recordTtl),
-		prometheusReg:           reg,
-		beaconsIncoming:         make(chan []interface{}),
-		recordsIncoming:         make(chan []interface{}),
+		connectionFactory:       spec.ConnectionFactory,
+		recordTtl:               getTtl(spec.FlowRecordTtl),
+		prometheusReg:           spec.PromReg,
+		beaconsIncoming:         make(chan []interface{}, 10),
+		heartbeatsIncoming:      make(chan []interface{}, 10),
+		recordsIncoming:         make(chan []interface{}, 10),
 		Request:                 make(chan ApiRequest),
 		Response:                make(chan ApiResponse),
 		eventSources:            make(map[string]*eventSource),
-		receivers:               make(map[string]*receiver),
-		senders:                 make(map[string]*senderDirect),
 		pendingFlush:            make(map[string]*senderDirect),
 		Beacons:                 make(map[string]*BeaconRecord),
 		Sites:                   make(map[string]*SiteRecord),
@@ -232,31 +249,13 @@ func NewFlowCollector(origin string, reg prometheus.Registerer, connectionFactor
 		Base: Base{
 			RecType:   recordNames[Collector],
 			Identity:  uuid.New().String(),
-			Parent:    origin,
+			Parent:    spec.Origin,
 			StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
 		},
 	}
 	return fc
 }
 
-func listToJSON(list []interface{}) *string {
-	data, err := json.MarshalIndent(list, "", " ")
-	if err == nil {
-		sd := string(data)
-		return &sd
-	}
-	return nil
-}
-
-func itemToJSON(item interface{}) *string {
-	data, err := json.Marshal(item)
-	if err == nil {
-		sd := string(data)
-		return &sd
-	}
-	return nil
-}
-
 func (fc *FlowCollector) serveRecords(request ApiRequest) ApiResponse {
 	request.HandlerName = mux.CurrentRoute(request.Request).GetName()
 	response := ApiResponse{
@@ -280,12 +279,7 @@ func getRealSizeOf(v interface{}) (int, error) {
 	return b.Len(), nil
 }
 
-func (c *FlowCollector) updates(stopCh <-chan struct{}) {
-	tickerFlush := time.NewTicker(1 * time.Second)
-	defer tickerFlush.Stop()
-	tickerReconcile := time.NewTicker(2 * time.Second)
-	defer tickerReconcile.Stop()
-
+func (c *FlowCollector) beaconUpdates(stopCh <-chan struct{}) {
 	for {
 		select {
 		case beaconUpdates := <-c.beaconsIncoming:
@@ -295,17 +289,24 @@ func (c *FlowCollector) updates(stopCh <-chan struct{}) {
 					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)
-						nonFlowReceiver := newReceiver(c.connectionFactory, beacon.Address, c.recordsIncoming)
-						nonFlowReceiver.start()
-						var flowReceiver *receiver = nil
+						receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address, c.recordsIncoming))
 						if beacon.SourceType == recordNames[Router] {
-							flowReceiver = newReceiver(c.connectionFactory, beacon.Address+".flows", c.recordsIncoming)
-							flowReceiver.start()
+							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, outgoing)
-						s.start()
+						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{
@@ -319,14 +320,18 @@ func (c *FlowCollector) updates(stopCh <-chan struct{}) {
 								LastHeard: now,
 								Beacons:   1,
 							},
-							nonFlowReceiver: nonFlowReceiver,
-							flowReceiver:    flowReceiver,
+							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)
@@ -334,10 +339,40 @@ func (c *FlowCollector) updates(stopCh <-chan struct{}) {
 					}
 				}
 			}
+		case <-stopCh:
+			return
+		}
+	}
+}
+
+func (c *FlowCollector) recordUpdates(stopCh <-chan struct{}) {
+	tickerFlush := time.NewTicker(1 * time.Second)
+	defer tickerFlush.Stop()
+	tickerReconcile := time.NewTicker(2 * time.Second)
+	defer tickerReconcile.Stop()
+	tickerAge := time.NewTicker(5 * time.Second)
+	defer tickerAge.Stop()
+
+	for {
+		select {
+		case heartbeatUpdates := <-c.heartbeatsIncoming:
+			for _, heartbeatUpdate := range heartbeatUpdates {
+				heartbeat, ok := heartbeatUpdate.(HeartbeatRecord)
+				if !ok {
+					log.Println("COLLECTOR: Unable to convert interface to heartbeat")
+				} else {
+					err := c.updateRecord(heartbeat)
+					if err != nil {
+						log.Println("COLLECTOR: heartbeat record error", err.Error())
+					}
+				}
+			}
 		case recordUpdates := <-c.recordsIncoming:
 			for _, update := range recordUpdates {
 				size, _ := getRealSizeOf(update)
-				c.metrics.collectorOctets.Add(float64(size))
+				if c.mode == RecordMetrics {
+					c.metrics.collectorOctets.Add(float64(size))
+				}
 				err := c.updateRecord(update)
 				if err != nil {
 					log.Println("COLLECTOR: Update record error", err.Error())
@@ -346,8 +381,6 @@ func (c *FlowCollector) updates(stopCh <-chan struct{}) {
 		case request := <-c.Request:
 			response := c.serveRecords(request)
 			c.Response <- response
-		case <-tickerReconcile.C:
-			c.reconcileRecords()
 		case <-tickerFlush.C:
 			for address, sender := range c.pendingFlush {
 				if sender.heartbeat {
@@ -356,6 +389,13 @@ func (c *FlowCollector) updates(stopCh <-chan struct{}) {
 					delete(c.pendingFlush, address)
 				}
 			}
+		case <-tickerReconcile.C:
+			if c.mode == RecordMetrics {
+				c.reconcileFlowRecords()
+			}
+			c.reconcileConnectorRecords()
+		case <-tickerAge.C:
+			c.ageAndPurgeRecords()
 		case <-stopCh:
 			return
 		}
@@ -367,19 +407,21 @@ func (c *FlowCollector) Start(stopCh <-chan struct{}) {
 }
 
 func (c *FlowCollector) run(stopCh <-chan struct{}) {
-	c.metrics = c.NewMetrics(c.prometheusReg)
-	c.metrics.info.With(prometheus.Labels{"version": version.Version}).Set(1)
-	r := newReceiver(c.connectionFactory, BeaconAddress, c.beaconsIncoming)
-	c.receivers[BeaconAddress] = r
-	r.start()
-	c.updates(stopCh)
+	if c.mode == RecordMetrics {
+		c.metrics = c.NewMetrics(c.prometheusReg)
+		c.metrics.info.With(prometheus.Labels{"version": version.Version}).Set(1)
+	}
+	c.beaconReceiver = newReceiver(c.connectionFactory, BeaconAddress, c.beaconsIncoming)
+	c.beaconReceiver.start()
+
+	go c.beaconUpdates(stopCh)
+	go c.recordUpdates(stopCh)
 	<-stopCh
 	for _, eventsource := range c.eventSources {
-		log.Println("COLLECTOR: Stopping record processing for ", eventsource.Identity)
-		eventsource.nonFlowReceiver.stop()
-		if eventsource.flowReceiver != nil {
-			eventsource.flowReceiver.stop()
+		for _, receiver := range eventsource.receivers {
+			receiver.stop()
 		}
 		eventsource.send.sender.stop()
 	}
+	c.beaconReceiver.stop()
 }
diff --git a/pkg/flow/controller.go b/pkg/flow/controller.go
index f5ec0001..47978e0f 100644
--- a/pkg/flow/controller.go
+++ b/pkg/flow/controller.go
@@ -6,6 +6,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/skupperproject/skupper/api/types"
+	"github.com/skupperproject/skupper/client"
 	"github.com/skupperproject/skupper/pkg/config"
 	"github.com/skupperproject/skupper/pkg/messaging"
 )
@@ -20,6 +22,7 @@ type FlowController struct {
 	version           string
 	connectionFactory messaging.ConnectionFactory
 	beaconOutgoing    chan interface{}
+	heartbeatOutgoing chan interface{}
 	recordOutgoing    chan interface{}
 	flushIncoming     chan []interface{}
 	processOutgoing   chan *ProcessRecord
@@ -35,13 +38,14 @@ func NewFlowController(origin string, version string, creationTime uint64, conne
 		creationTime:      creationTime,
 		version:           version,
 		connectionFactory: connectionFactory,
-		beaconOutgoing:    make(chan interface{}),
-		recordOutgoing:    make(chan interface{}),
-		flushIncoming:     make(chan []interface{}),
-		processOutgoing:   make(chan *ProcessRecord),
+		beaconOutgoing:    make(chan interface{}, 10),
+		heartbeatOutgoing: make(chan interface{}, 10),
+		recordOutgoing:    make(chan interface{}, 10),
+		flushIncoming:     make(chan []interface{}, 10),
+		processOutgoing:   make(chan *ProcessRecord, 10),
 		processRecords:    make(map[string]*ProcessRecord),
 		hostRecords:       make(map[string]*HostRecord),
-		hostOutgoing:      make(chan *HostRecord),
+		hostOutgoing:      make(chan *HostRecord, 10),
 		startTime:         time.Now().Unix(),
 	}
 	return fc
@@ -84,15 +88,18 @@ func UpdateHost(c *FlowController, deleted bool, name string, host *HostRecord)
 		if _, ok := c.hostRecords[host.Identity]; !ok {
 			c.hostRecords[host.Identity] = host
 		}
-	}
-	if host != nil {
 		c.hostOutgoing <- host
+	} else {
+		if existing, ok := c.hostRecords[host.Identity]; ok {
+			existing.EndTime = uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
+		}
+		delete(c.hostRecords, host.Identity)
 	}
 
 	return nil
 }
 
-func (c *FlowController) updates(stopCh <-chan struct{}) {
+func (c *FlowController) updateBeaconAndHeartbeats(stopCh <-chan struct{}) {
 	tickerAge := time.NewTicker(10 * time.Minute)
 	defer tickerAge.Stop()
 
@@ -118,9 +125,46 @@ func (c *FlowController) updates(stopCh <-chan struct{}) {
 		Source:   "sfe." + c.origin,
 	}
 
+	c.beaconOutgoing <- beacon
+	heartbeat.Now = uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
+	c.heartbeatOutgoing <- heartbeat
+
+	for {
+		select {
+		case <-beaconTimer.C:
+			c.beaconOutgoing <- beacon
+		case <-heartbeatTimer.C:
+			heartbeat.Now = uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
+			c.heartbeatOutgoing <- heartbeat
+		case <-tickerAge.C:
+		case <-stopCh:
+			return
+
+		}
+	}
+}
+
+func (c *FlowController) updateRecords(stopCh <-chan struct{}) {
+	tickerAge := time.NewTicker(10 * time.Minute)
+	defer tickerAge.Stop()
+
 	name := os.Getenv("SKUPPER_SITE_NAME")
 	nameSpace := os.Getenv("SKUPPER_NAMESPACE")
-	platform := string(config.GetPlatform())
+	policy := Disabled
+	var platformStr string
+	platform := config.GetPlatform()
+	if platform == "" || platform == types.PlatformKubernetes {
+		platformStr = string(types.PlatformKubernetes)
+		cli, err := client.NewClient(nameSpace, "", "")
+		if err == nil {
+			cpv := client.NewClusterPolicyValidator(cli)
+			if cpv.Enabled() {
+				policy = Enabled
+			}
+		}
+	} else if platform == types.PlatformPodman {
+		platformStr = string(types.PlatformPodman)
+	}
 	site := &SiteRecord{
 		Base: Base{
 			RecType:   recordNames[Site],
@@ -129,22 +173,15 @@ func (c *FlowController) updates(stopCh <-chan struct{}) {
 		},
 		Name:      &name,
 		NameSpace: &nameSpace,
+		Platform:  &platformStr,
 		Version:   &c.version,
-		Platform:  &platform,
+		Policy:    &policy,
 	}
 
-	c.beaconOutgoing <- beacon
-	heartbeat.Now = uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
-	c.recordOutgoing <- heartbeat
 	c.recordOutgoing <- site
 
 	for {
 		select {
-		case <-beaconTimer.C:
-			c.beaconOutgoing <- beacon
-		case <-heartbeatTimer.C:
-			heartbeat.Now = uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
-			c.recordOutgoing <- heartbeat
 		case process := <-c.processOutgoing:
 			c.recordOutgoing <- process
 		case host := <-c.hostOutgoing:
@@ -176,18 +213,22 @@ func (c *FlowController) Start(stopCh <-chan struct{}) {
 }
 
 func (c *FlowController) run(stopCh <-chan struct{}) {
-	beaconSender := newSender(c.connectionFactory, BeaconAddress, c.beaconOutgoing)
-	recordSender := newSender(c.connectionFactory, RecordPrefix+c.origin, c.recordOutgoing)
+	beaconSender := newSender(c.connectionFactory, BeaconAddress, true, c.beaconOutgoing)
+	heartbeatSender := newSender(c.connectionFactory, RecordPrefix+c.origin+".heartbeats", true, c.heartbeatOutgoing)
+	recordSender := newSender(c.connectionFactory, RecordPrefix+c.origin, false, c.recordOutgoing)
 	flushReceiver := newReceiver(c.connectionFactory, DirectPrefix+c.origin, c.flushIncoming)
 
 	beaconSender.start()
+	heartbeatSender.start()
 	recordSender.start()
 	flushReceiver.start()
 
-	go c.updates(stopCh)
+	go c.updateBeaconAndHeartbeats(stopCh)
+	go c.updateRecords(stopCh)
 	<-stopCh
 
 	beaconSender.stop()
+	heartbeatSender.start()
 	recordSender.stop()
 	flushReceiver.stop()
 }
diff --git a/pkg/flow/encoding.go b/pkg/flow/encoding.go
index 6c5d5a64..8fc83ea0 100644
--- a/pkg/flow/encoding.go
+++ b/pkg/flow/encoding.go
@@ -100,6 +100,7 @@ func encodeSite(site *SiteRecord) (*amqp.Message, error) {
 	var request amqp.Message
 	var properties amqp.MessageProperties
 	properties.Subject = "RECORD"
+	properties.To = RecordPrefix + site.Identity
 	request.Properties = &properties
 
 	m := make(map[interface{}]interface{})
@@ -112,13 +113,15 @@ func encodeSite(site *SiteRecord) (*amqp.Message, error) {
 	if site.NameSpace != nil {
 		m[uint32(Namespace)] = *site.NameSpace
 	}
+	if site.Platform != nil {
+		m[uint32(Platform)] = *site.Platform
+	}
 	if site.Version != nil {
 		m[uint32(Version)] = *site.Version
 	}
-	if site.Platform != nil {
-		m[uint32(Platform)] = *site.Platform
+	if site.Policy != nil {
+		m[uint32(Policy)] = *site.Policy
 	}
-
 	record = append(record, m)
 
 	request.Value = record
@@ -131,6 +134,7 @@ func encodeProcess(process *ProcessRecord) (*amqp.Message, error) {
 	var request amqp.Message
 	var properties amqp.MessageProperties
 	properties.Subject = "RECORD"
+	properties.To = RecordPrefix + process.Parent
 	request.Properties = &properties
 
 	m := make(map[interface{}]interface{})
@@ -172,6 +176,7 @@ func encodeHost(host *HostRecord) (*amqp.Message, error) {
 	var request amqp.Message
 	var properties amqp.MessageProperties
 	properties.Subject = "RECORD"
+	properties.To = RecordPrefix + host.Parent
 	request.Properties = &properties
 
 	m := make(map[interface{}]interface{})
@@ -199,6 +204,8 @@ func encodeHost(host *HostRecord) (*amqp.Message, error) {
 func decode(msg *amqp.Message) []interface{} {
 	var result []interface{}
 
+	source := strings.TrimPrefix(msg.Properties.To, RecordPrefix)
+
 	switch msg.Properties.Subject {
 	case "BEACON":
 		result = append(result, asBeaconMessage(msg))
@@ -227,6 +234,7 @@ func decode(msg *amqp.Message) []interface{} {
 				}
 				base := Base{
 					RecType: recordNames[rt],
+					Source:  source,
 				}
 				if v, ok := m["Identity"].(string); ok {
 					base.Identity = v
@@ -264,6 +272,9 @@ func decode(msg *amqp.Message) []interface{} {
 					if v, ok := m["Version"].(string); ok {
 						site.Version = &v
 					}
+					if v, ok := m["Policy"].(string); ok {
+						site.Policy = &v
+					}
 					result = append(result, site)
 				case Host:
 					host := HostRecord{
@@ -380,7 +391,27 @@ func decode(msg *amqp.Message) []interface{} {
 					if v, ok := m["FlowRateL7"].(uint64); ok {
 						connector.FlowRateL7 = &v
 					}
+					if v, ok := m["Target"].(string); ok {
+						connector.Target = &v
+					}
 					result = append(result, connector)
+				case LogEvent:
+					logEvent := LogEventRecord{
+						Base: base,
+					}
+					if v, ok := m["LogSeverity"].(uint64); ok {
+						logEvent.LogSeverity = &v
+					}
+					if v, ok := m["LogText"].(string); ok {
+						logEvent.LogText = &v
+					}
+					if v, ok := m["SourceFile"].(string); ok {
+						logEvent.SourceFile = &v
+					}
+					if v, ok := m["SourceLine"].(uint64); ok {
+						logEvent.SourceLine = &v
+					}
+					result = append(result, logEvent)
 				case Flow:
 					flow := FlowRecord{
 						Base: base,
diff --git a/pkg/flow/flow_mem_driver.go b/pkg/flow/flow_mem_driver.go
index e67eb1c8..6a460e41 100644
--- a/pkg/flow/flow_mem_driver.go
+++ b/pkg/flow/flow_mem_driver.go
@@ -1,6 +1,7 @@
 package flow
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"log"
@@ -11,6 +12,13 @@ import (
 	"github.com/google/uuid"
 	"github.com/gorilla/mux"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/skupperproject/skupper/api/types"
+	"github.com/skupperproject/skupper/client"
+	"github.com/skupperproject/skupper/pkg/config"
+	"github.com/skupperproject/skupper/pkg/kube"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/util/retry"
 )
 
 func (fc *FlowCollector) inferGatewaySite(siteId string) error {
@@ -175,6 +183,72 @@ func (c *FlowCollector) getRecordSiteId(record interface{}) string {
 	return ""
 }
 
+func (fc *FlowCollector) getRoutersForSite(site *SiteRecord) []*RouterRecord {
+	var routers []*RouterRecord
+	for _, router := range fc.Routers {
+		if router.Parent == site.Identity {
+			routers = append(routers, router)
+		}
+	}
+	return routers
+}
+
+func (fc *FlowCollector) getLinksForRouter(router *RouterRecord) []*LinkRecord {
+	var links []*LinkRecord
+	for _, link := range fc.Links {
+		if link.Parent == router.Identity {
+			links = append(links, link)
+		}
+	}
+	return links
+}
+
+func (fc *FlowCollector) getListenersForRouter(router *RouterRecord) []*ListenerRecord {
+	var listeners []*ListenerRecord
+	for _, listener := range fc.Listeners {
+		if listener.Parent == router.Identity {
+			listeners = append(listeners, listener)
+		}
+	}
+	return listeners
+}
+
+func (fc *FlowCollector) getConnectorsForRouter(router *RouterRecord) []*ConnectorRecord {
+	var connectors []*ConnectorRecord
+	for _, connector := range fc.Connectors {
+		if connector.Parent == router.Identity {
+			connectors = append(connectors, connector)
+		}
+	}
+	return connectors
+}
+
+func (fc *FlowCollector) getRouterForFlow(flow *FlowRecord) *RouterRecord {
+	if connector, ok := fc.Connectors[flow.Parent]; ok {
+		if router, ok := fc.Routers[connector.Parent]; ok {
+			return router
+		}
+	}
+	if listener, ok := fc.Listeners[flow.Parent]; ok {
+		if router, ok := fc.Routers[listener.Parent]; ok {
+			return router
+		}
+	}
+	if l4Flow, ok := fc.Flows[flow.Parent]; ok {
+		if connector, ok := fc.Connectors[l4Flow.Parent]; ok {
+			if router, ok := fc.Routers[connector.Parent]; ok {
+				return router
+			}
+		}
+		if listener, ok := fc.Listeners[l4Flow.Parent]; ok {
+			if router, ok := fc.Routers[listener.Parent]; ok {
+				return router
+			}
+		}
+	}
+	return nil
+}
+
 func (fc *FlowCollector) getFlowProtocol(flow *FlowRecord) *string {
 	if listener, ok := fc.Listeners[flow.Parent]; ok {
 		return listener.Protocol
@@ -405,6 +479,75 @@ func prettyPrint(i interface{}) string {
 	return string(s)
 }
 
+var defaultRetry = wait.Backoff{
+	Steps:    100,
+	Duration: 10 * time.Millisecond,
+	Factor:   1.0,
+	Jitter:   0.1,
+}
+
+func (fc *FlowCollector) updateNetworkStatus() error {
+	var err error
+	networkData := map[string]string{}
+	platform := config.GetPlatform()
+	if platform == "" || platform == types.PlatformKubernetes {
+		cli, err := client.NewClient(fc.namespace, "", "")
+		if err != nil {
+			return err
+		}
+
+		var sites []*SiteStatus
+		var addresses []*VanAddressRecord
+
+		for _, address := range fc.VanAddresses {
+			fc.getAddressAdaptorCounts(address)
+			addresses = append(addresses, address)
+		}
+		for _, site := range fc.Sites {
+			var routerStatus []RouterStatus
+			routers := fc.getRoutersForSite(site)
+			for _, router := range routers {
+				links := fc.getLinksForRouter(router)
+				listeners := fc.getListenersForRouter(router)
+				connectors := fc.getConnectorsForRouter(router)
+				routerStatus = append(routerStatus, RouterStatus{
+					Router:     router,
+					Links:      links,
+					Listeners:  listeners,
+					Connectors: connectors,
+				})
+			}
+			siteStatus := &SiteStatus{
+				Site:         site,
+				RouterStatus: routerStatus,
+			}
+			sites = append(sites, siteStatus)
+		}
+		networkStatus := NetworkStatus{
+			Addresses: addresses,
+			Sites:     sites,
+		}
+		networkData["NetworkStatus"] = prettyPrint(networkStatus)
+
+		err = retry.RetryOnConflict(defaultRetry, func() error {
+			configMap, err := kube.GetConfigMap(types.NetworkStatusConfigMapName, cli.Namespace, cli.KubeClient)
+			if err != nil {
+				return err
+			}
+
+			configMap.Data = networkData
+
+			_, err = cli.KubeClient.CoreV1().ConfigMaps(cli.Namespace).Update(context.TODO(), configMap, metav1.UpdateOptions{})
+			if err != nil {
+				return err
+			} else {
+				return nil
+			}
+		})
+	}
+	return err
+}
+
 func (fc *FlowCollector) addRecord(record interface{}) error {
 	if record == nil {
 		return fmt.Errorf("No record to add")
@@ -463,6 +606,9 @@ func (fc *FlowCollector) addRecord(record interface{}) error {
 	default:
 		return fmt.Errorf("Unknown record type to add")
 	}
+	if fc.mode == RecordStatus {
+		fc.updateNetworkStatus()
+	}
 	return nil
 }
 
@@ -525,6 +671,17 @@ func (fc *FlowCollector) deleteRecord(record interface{}) error {
 	default:
 		return fmt.Errorf("Unknown record type to delete")
 	}
+	if fc.mode == RecordStatus {
+		fc.updateNetworkStatus()
+	}
+	return nil
+}
+
+func (fc *FlowCollector) updateLastHeard(source string) error {
+	if eventsource, ok := fc.eventSources[source]; ok {
+		eventsource.LastHeard = uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
+		eventsource.Messages++
+	}
 	return nil
 }
 
@@ -533,8 +690,9 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 	case HeartbeatRecord:
 		if heartbeat, ok := record.(HeartbeatRecord); ok {
 			if eventsource, ok := fc.eventSources[heartbeat.Identity]; ok {
-				eventsource.LastHeard = heartbeat.Now
+				eventsource.LastHeard = uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
 				eventsource.Heartbeats++
+				eventsource.Messages++
 			}
 			if pending, ok := fc.pendingFlush[heartbeat.Source]; ok {
 				pending.heartbeat = true
@@ -562,6 +720,7 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 					*current = site
 				}
 			}
+			fc.updateLastHeard(site.Source)
 		}
 	case HostRecord:
 		if host, ok := record.(HostRecord); ok {
@@ -577,6 +736,7 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 					*current = host
 				}
 			}
+			fc.updateLastHeard(host.Source)
 		}
 	case RouterRecord:
 		if router, ok := record.(RouterRecord); ok {
@@ -598,6 +758,11 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 					}
 				}
 			}
+			fc.updateLastHeard(router.Source)
+		}
+	case LogEventRecord:
+		if logEvent, ok := record.(LogEventRecord); ok {
+			log.Printf("LOG_EVENT: %s \n", prettyPrint(logEvent))
 		}
 	case LinkRecord:
 		if link, ok := record.(LinkRecord); ok {
@@ -611,6 +776,7 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 					fc.deleteRecord(current)
 				}
 			}
+			fc.updateLastHeard(link.Source)
 		}
 	case ListenerRecord:
 		if listener, ok := record.(ListenerRecord); ok {
@@ -710,6 +876,7 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 					}
 				}
 			}
+			fc.updateLastHeard(listener.Source)
 		}
 	case ConnectorRecord:
 		if connector, ok := record.(ConnectorRecord); ok {
@@ -820,6 +987,7 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 					}
 				}
 			}
+			fc.updateLastHeard(connector.Source)
 		}
 	case FlowRecord:
 		if flow, ok := record.(FlowRecord); ok {
@@ -982,6 +1150,7 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 					fc.deleteRecord(current)
 				}
 			}
+			fc.updateLastHeard(process.Source)
 		}
 	case ProcessGroupRecord:
 		if processGroup, ok := record.(ProcessGroupRecord); ok {
@@ -1004,6 +1173,7 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 					fc.deleteRecord(current)
 				}
 			}
+			fc.updateLastHeard(processGroup.Source)
 		}
 	default:
 		return fmt.Errorf("Unrecognized record type %T", record)
@@ -1353,7 +1523,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 			addresses := []VanAddressRecord{}
 			for _, address := range fc.VanAddresses {
 				if filterRecord(*address, queryParams) {
-					fc.getAddressMetrics(address)
+					fc.getAddressAdaptorCounts(address)
 					addresses = append(addresses, *address)
 				}
 			}
@@ -1362,7 +1532,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "item":
 			if id, ok := vars["id"]; ok {
 				if address, ok := fc.VanAddresses[id]; ok {
-					fc.getAddressMetrics(address)
+					fc.getAddressAdaptorCounts(address)
 					p.Count = 1
 					p.Results = address
 				}
@@ -1522,6 +1692,15 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 			processes := []ProcessRecord{}
 			for _, process := range fc.Processes {
 				if filterRecord(*process, queryParams) && process.Base.TimeRangeValid(queryParams) {
+					if process.connector != nil {
+						process.Addresses = nil
+						if connector, ok := fc.Connectors[*process.connector]; ok {
+							if connector.Address != nil && connector.AddressId != nil {
+								addrDetails := *connector.Address + "@" + *connector.AddressId
+								process.Addresses = append(process.Addresses, &addrDetails)
+							}
+						}
+					}
 					processes = append(processes, *process)
 				}
 			}
@@ -1556,7 +1735,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 							for _, address := range fc.VanAddresses {
 								if *connector.Address == address.Name {
 									if filterRecord(*address, queryParams) {
-										fc.getAddressMetrics(address)
+										fc.getAddressAdaptorCounts(address)
 										addresses = append(addresses, *address)
 									}
 								}
@@ -1912,7 +2091,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 			retrieveError = sortAndSlice(flows, &p, queryParams)
 		case "pair-to-aggregate":
 			flowPairs := []string{}
-			for flowPairId, _ := range fc.aggregatesToReconcile {
+			for flowPairId := range fc.aggregatesToReconcile {
 				if _, ok := fc.FlowPairs[flowPairId]; ok {
 					p.TotalCount++
 					flowPairs = append(flowPairs, flowPairId)
@@ -2034,15 +2213,20 @@ func (fc *FlowCollector) setupFlowMetrics(va *VanAddressRecord, flow *FlowRecord
 	}
 
 	if direction, ok := metricLabel["direction"]; ok {
-		if direction == Incoming && flow.Method != nil {
-			metricLabel["method"] = *flow.Method
-			httpReqsMethod, err := fc.metrics.httpReqsMethod.GetMetricWith(metricLabel)
-			if err != nil {
-				return err
-			} else {
-				httpReqsMethod.Inc()
+		if direction == Incoming {
+			if flow.Method != nil {
+				metricLabel["method"] = *flow.Method
+				httpReqsMethod, err := fc.metrics.httpReqsMethod.GetMetricWith(metricLabel)
+				if err != nil {
+					return err
+				} else {
+					httpReqsMethod.Inc()
+				}
+				delete(metricLabel, "method")
 			}
-		} else if direction == Outgoing && flow.Result != nil {
+		}
+		// note: observed code to be present in both forward in reverse flows (e.g. directions)
+		if flow.Result != nil {
 			metricLabel["code"] = *flow.Result
 			httpReqsResult, err := fc.metrics.httpReqsResult.GetMetricWith(metricLabel)
 			if err != nil {
@@ -2050,12 +2234,13 @@ func (fc *FlowCollector) setupFlowMetrics(va *VanAddressRecord, flow *FlowRecord
 			} else {
 				httpReqsResult.Inc()
 			}
+			delete(metricLabel, "code")
 		}
 	}
 	return nil
 }
 
-func (fc *FlowCollector) reconcileRecords() error {
+func (fc *FlowCollector) reconcileFlowRecords() error {
 	age := uint64(time.Now().UnixNano())/uint64(time.Microsecond) - uint64(fc.recordTtl.Microseconds())
 
 	m, err := fc.metrics.activeReconcile.GetMetricWith(prometheus.Labels{"reconcileTask": "flowsToProcess"})
@@ -2134,68 +2319,6 @@ func (fc *FlowCollector) reconcileRecords() error {
 			}
 		}
 	}
-	m, err = fc.metrics.activeReconcile.GetMetricWith(prometheus.Labels{"reconcileTask": "connectorToProcess"})
-	if err == nil {
-		m.Set(float64(len(fc.connectorsToReconcile)))
-	}
-	for _, connId := range fc.connectorsToReconcile {
-		t := time.Now()
-		if connector, ok := fc.Connectors[connId]; ok {
-			if connector.EndTime > 0 {
-				delete(fc.connectorsToReconcile, connId)
-			} else if connector.DestHost != nil {
-				siteId := fc.getRecordSiteId(*connector)
-				var matchHost *string
-				found := false
-				if net.ParseIP(*connector.DestHost) == nil {
-					addrs, err := net.LookupHost(*connector.DestHost)
-					if err == nil && len(addrs) > 0 {
-						matchHost = &addrs[0]
-					}
-				} else {
-					matchHost = connector.DestHost
-				}
-				for _, process := range fc.Processes {
-					if siteId == process.Parent {
-						if process.SourceHost != nil {
-							if *matchHost == *process.SourceHost {
-								connector.ProcessId = &process.Identity
-								process.connector = &connector.Identity
-								process.ProcessBinding = &Bound
-								log.Printf("COLLECTOR: Connector %s/%s associated to process %s\n", connector.Identity, *connector.Address, *process.Name)
-								delete(fc.connectorsToReconcile, connId)
-								found = true
-								break
-							}
-						}
-					}
-				}
-				if !found {
-					parts := strings.Split(siteId, "-")
-					processName := "site-servers-" + parts[0]
-					diffTime := connector.StartTime
-					wait := 10 * oneSecond
-					if fc.startTime > connector.StartTime {
-						diffTime = fc.startTime
-						wait = 60 * oneSecond
-					}
-					diff := uint64(t.UnixNano())/uint64(time.Microsecond) - diffTime
-					if diff > wait {
-						for _, process := range fc.Processes {
-							if process.Name != nil && *process.Name == processName {
-								log.Printf("COLLECTOR: Associating connector %s to external process %s\n", connector.Identity, processName)
-								connector.ProcessId = &process.Identity
-								delete(fc.connectorsToReconcile, connId)
-								break
-							}
-						}
-					}
-				}
-			}
-		} else {
-			delete(fc.connectorsToReconcile, connId)
-		}
-	}
 	m, err = fc.metrics.activeReconcile.GetMetricWith(prometheus.Labels{"reconcileTask": "pairToAggregate"})
 	if err == nil {
 		m.Set(float64(len(fc.aggregatesToReconcile)))
@@ -2241,11 +2364,21 @@ func (fc *FlowCollector) reconcileRecords() error {
 					}
 					if sourceProcess, ok := fc.Processes[ffp.Identity]; ok {
 						pfa.SourceName = sourceProcess.Name
+						sourceSiteId := fc.getRecordSiteId(*sourceProcess)
+						if sourceSite, ok := fc.Sites[sourceSiteId]; ok {
+							pfa.SourceSiteId = &sourceSiteId
+							pfa.SourceSiteName = sourceSite.Name
+						}
 					}
 					if destinationProcess, ok := fc.Processes[cfp.Identity]; ok {
 						if destinationProcess.Name != nil {
 							pfa.DestinationName = destinationProcess.Name
 						}
+						destinationSiteId := fc.getRecordSiteId(*destinationProcess)
+						if destinationSite, ok := fc.Sites[destinationSiteId]; ok {
+							pfa.DestinationSiteId = &destinationSiteId
+							pfa.DestinationSiteName = destinationSite.Name
+						}
 					}
 					fc.FlowAggregates[processAggregateId] = pfa
 				}
@@ -2276,9 +2409,79 @@ func (fc *FlowCollector) reconcileRecords() error {
 			}
 		}
 	}
+	return nil
+}
+
+func (fc *FlowCollector) reconcileConnectorRecords() error {
+	for _, connId := range fc.connectorsToReconcile {
+		t := time.Now()
+		if connector, ok := fc.Connectors[connId]; ok {
+			if connector.EndTime > 0 {
+				delete(fc.connectorsToReconcile, connId)
+			} else if connector.DestHost != nil {
+				siteId := fc.getRecordSiteId(*connector)
+				var matchHost *string
+				found := false
+				if net.ParseIP(*connector.DestHost) == nil {
+					addrs, err := net.LookupHost(*connector.DestHost)
+					if err == nil && len(addrs) > 0 {
+						matchHost = &addrs[0]
+					}
+				} else {
+					matchHost = connector.DestHost
+				}
+				for _, process := range fc.Processes {
+					if siteId == process.Parent {
+						if process.SourceHost != nil {
+							if *matchHost == *process.SourceHost {
+								connector.ProcessId = &process.Identity
+								connector.Target = process.Name
+								process.connector = &connector.Identity
+								process.ProcessBinding = &Bound
+								fc.updateNetworkStatus()
+								log.Printf("COLLECTOR: Connector %s/%s associated to process %s\n", connector.Identity, *connector.Address, *process.Name)
+								delete(fc.connectorsToReconcile, connId)
+								found = true
+								break
+							}
+						}
+					}
+				}
+				if !found {
+					parts := strings.Split(siteId, "-")
+					processName := "site-servers-" + parts[0]
+					diffTime := connector.StartTime
+					wait := 30 * oneSecond
+					if fc.startTime > connector.StartTime {
+						diffTime = fc.startTime
+						wait = 120 * oneSecond
+					}
+					diff := uint64(t.UnixNano())/uint64(time.Microsecond) - diffTime
+					if diff > wait {
+						for _, process := range fc.Processes {
+							if process.Name != nil && *process.Name == processName {
+								log.Printf("COLLECTOR: Associating connector %s to external process %s\n", connector.Identity, processName)
+								connector.ProcessId = &process.Identity
+								delete(fc.connectorsToReconcile, connId)
+								break
+							}
+						}
+					}
+				}
+			}
+		} else {
+			delete(fc.connectorsToReconcile, connId)
+		}
+	}
+	return nil
+}
+
+func (fc *FlowCollector) ageAndPurgeRecords() error {
+	age := uint64(time.Now().UnixNano())/uint64(time.Microsecond) - uint64(fc.recordTtl.Microseconds())
 
 	for flowId, flow := range fc.Flows {
-		if flow.EndTime != 0 && age > flow.EndTime {
+		router := fc.getRouterForFlow(flow)
+		if flow.EndTime != 0 && age > flow.EndTime || router == nil {
 			fc.deleteRecord(flow)
 			if flowPair, ok := fc.FlowPairs["fp-"+flowId]; ok {
 				fc.deleteRecord(flowPair)
@@ -2289,7 +2492,7 @@ func (fc *FlowCollector) reconcileRecords() error {
 	t := time.Now()
 	for _, source := range fc.eventSources {
 		diff := uint64(t.UnixNano())/uint64(time.Microsecond) - source.EventSourceRecord.LastHeard
-		if diff > 30*oneSecond {
+		if diff > 60*oneSecond {
 			log.Printf("COLLECTOR: Purging event source %s of type %s \n", source.Beacon.Identity, source.Beacon.SourceType)
 			fc.purgeEventSource(source.EventSourceRecord)
 		}
@@ -2306,7 +2509,7 @@ func (fc *FlowCollector) reconcileRecords() error {
 	return nil
 }
 
-func (fc *FlowCollector) getAddressMetrics(addr *VanAddressRecord) error {
+func (fc *FlowCollector) getAddressAdaptorCounts(addr *VanAddressRecord) error {
 	listenerCount := 0
 	connectorCount := 0
 	for _, listener := range fc.Listeners {
@@ -2395,7 +2598,16 @@ func (fc *FlowCollector) purgeEventSource(eventSource EventSourceRecord) error {
 			}
 		}
 	}
+	for id, es := range fc.eventSources {
+		if id == eventSource.Identity {
+			for _, receiver := range es.receivers {
+				receiver.stop()
+			}
+			es.send.sender.stop()
+		}
+	}
 	eventSource.Purged = true
+	eventSource.EndTime = now
 	log.Printf("COLLECTOR: %s \n", prettyPrint(eventSource))
 	delete(fc.eventSources, eventSource.Identity)
 
@@ -2420,6 +2632,9 @@ func (fc *FlowCollector) createSiteProcess(name string, site SiteRecord) Process
 }
 
 func (fc *FlowCollector) needForSiteProcess(flow *FlowRecord, siteId string, startTime uint64, client bool) bool {
+	if fc.getFlowPlace(flow) == unknown {
+		return false
+	}
 	parts := strings.Split(siteId, "-")
 	name := "site-servers"
 	if client {
@@ -2427,10 +2642,10 @@ func (fc *FlowCollector) needForSiteProcess(flow *FlowRecord, siteId string, sta
 	}
 	processName := name + "-" + parts[0]
 	diffTime := startTime
-	wait := 10 * oneSecond
+	wait := 120 * oneSecond
 	if fc.startTime > startTime {
 		diffTime = fc.startTime
-		wait = 60 * oneSecond
+		wait = 120 * oneSecond
 	}
 	diff := uint64(time.Now().UnixNano())/uint64(time.Microsecond) - diffTime
 	found := false
diff --git a/pkg/flow/messaging.go b/pkg/flow/messaging.go
index 00cbd63f..32ed3a37 100644
--- a/pkg/flow/messaging.go
+++ b/pkg/flow/messaging.go
@@ -33,8 +33,9 @@ func (c *base) stop() {
 
 type sender struct {
 	base
-	ticker  *time.Ticker
-	request *amqp.Message
+	sendSettled bool
+	ticker      *time.Ticker
+	request     *amqp.Message
 }
 
 func (c *sender) start() {
@@ -42,13 +43,14 @@ func (c *sender) start() {
 	go c.send()
 }
 
-func newSender(connectionFactory messaging.ConnectionFactory, address string, update chan interface{}) *sender {
+func newSender(connectionFactory messaging.ConnectionFactory, address string, sendSettled bool, update chan interface{}) *sender {
 	return &sender{
 		base: base{
 			connectionFactory: connectionFactory,
 			outgoing:          update,
 			address:           address,
 		},
+		sendSettled: sendSettled,
 	}
 }
 
@@ -70,6 +72,7 @@ func (c *sender) _send() error {
 		return err
 	}
 	c.client = client
+	log.Printf("COLLECTOR: Connection for sender %s to %s established\n", c.address, c.connectionFactory.Url())
 	defer client.Close()
 
 	sender, err := client.Sender(c.address)
@@ -133,6 +136,7 @@ func (c *sender) _send() error {
 		case <-c.ticker.C:
 		}
 		if c.request != nil {
+			c.request.SendSettled = c.sendSettled
 			err = sender.Send(c.request)
 			if err != nil {
 				return err
@@ -167,7 +171,7 @@ func (r *receiver) receive() {
 	for !r.closed {
 		err := r._receive()
 		if err != nil {
-			log.Println("COLLECTOR: Error receiving message ", err.Error())
+			log.Printf("COLLECTOR: Receiver %s %s\n", r.address, err.Error())
 		}
 	}
 }
@@ -178,9 +182,10 @@ func (r *receiver) _receive() error {
 		return err
 	}
 	r.client = client
+	log.Printf("COLLECTOR: Connection for receiver %s to %s established\n", r.address, r.connectionFactory.Url())
 	defer client.Close()
 
-	receiver, err := client.Receiver(r.address, 10)
+	receiver, err := client.Receiver(r.address, 250)
 	if err != nil {
 		return err
 	}
@@ -189,7 +194,6 @@ func (r *receiver) _receive() error {
 	for {
 		msg, err := receiver.Receive()
 		if err != nil {
-			log.Println("COLLECTOR: Receiver error ", err.Error())
 			return err
 		}
 		receiver.Accept(msg)
diff --git a/pkg/flow/record.go b/pkg/flow/record.go
index 6ac887b2..afe1e786 100644
--- a/pkg/flow/record.go
+++ b/pkg/flow/record.go
@@ -30,13 +30,14 @@ const (
 	Collector               // 11
 	ProcessGroup            // 12
 	Host                    // 13
-	FlowPair                // 14 (generated)
-	FlowAggregate           // 15
-	EventSource             // 16
-	SitePair                // 17
-	ProcessGroupPair        // 18
-	ProcessPair             // 19
-	Address                 // 20
+	LogEvent                // 14
+	FlowPair                // 15 (generated)
+	FlowAggregate           // 16
+	EventSource             // 17
+	SitePair                // 18
+	ProcessGroupPair        // 19
+	ProcessPair             // 20
+	Address                 // 21
 )
 
 var recordNames = []string{
@@ -52,6 +53,7 @@ var recordNames = []string{
 	"INGRESS",
 	"EGRESS",
 	"COLLECTOR",
+	"LOGEVENT",
 	"PROCESS_GROUP",
 	"HOST",
 	"FLOWPAIR",
@@ -113,7 +115,13 @@ const (
 	ImageAttr              // 45
 	Group                  // 46
 	StreamIdentity         // 47
-	Version                //48
+	LogSeverity            // 48
+	LogText                // 49
+	SourceFile             // 50
+	SourceLine             // 51
+	Version                // 52
+	Policy                 // 53
+	Target                 // 54
 )
 
 var attributeNames = []string{
@@ -165,7 +173,13 @@ var attributeNames = []string{
 	"Image",           // 45
 	"Group",           // 46
 	"StreamIdentity",  // 47
-	"Version",         // 48
+	"LogSeverity",     // 48
+	"LogText",         // 49
+	"SourceFile",      // 50
+	"SourceLine",      // 51
+	"Version",         // 52
+	"Policy",          // 53
+	"Target",          // 54
 }
 
 var Internal string = "internal"
@@ -175,6 +189,8 @@ var Incoming string = "incoming"
 var Outgoing string = "outgoing"
 var Bound string = "bound"
 var Unbound string = "unbound"
+var Enabled string = "enabled"
+var Disabled string = "disabled"
 
 type Base struct {
 	RecType   string `json:"recType,omitempty"`
@@ -183,6 +199,7 @@ type Base struct {
 	StartTime uint64 `json:"startTime"`
 	EndTime   uint64 `json:"endTime"`
 	Purged    bool   `json:"purged,omitempty"`
+	Source    string `json:"source,omitempty"`
 }
 
 type BeaconRecord struct {
@@ -211,6 +228,7 @@ type EventSourceRecord struct {
 	LastHeard  uint64        `json:"lastHeard,omitempty"`
 	Heartbeats int           `json:"heartbeats,omitempty"`
 	Beacons    int           `json:"beacons,omitempty"`
+	Messages   int           `json:"messages,omitempty"`
 }
 
 type SiteRecord struct {
@@ -221,6 +239,7 @@ type SiteRecord struct {
 	Name      *string `json:"name,omitempty"`
 	NameSpace *string `json:"nameSpace,omitempty"`
 	Version   *string `json:"siteVersion,omitempty"`
+	Policy    *string `json:"policy,omitempty"`
 }
 
 type HostRecord struct {
@@ -283,10 +302,19 @@ type ConnectorRecord struct {
 	FlowRateL4  *uint64 `json:"flowRateL4,omitempty"`
 	FlowCountL7 *uint64 `json:"flowCountL7,omitempty"`
 	FlowRateL7  *uint64 `json:"flowRateL7,omitempty"`
+	Target      *string `json:"target,omitempty"`
 	ProcessId   *string `json:"processId,omitempty"`
 	AddressId   *string `json:"addressId,omitempty"`
 }
 
+type LogEventRecord struct {
+	Base
+	LogSeverity *uint64 `json:"severity,omitempty"`
+	LogText     *string `json:"logText,omitempty"`
+	SourceFile  *string `json:"fileName,omitempty"`
+	SourceLine  *uint64 `json:"lineNumber,omitempty"`
+}
+
 type metricKey struct {
 	sourceSite    string
 	sourceProcess string
@@ -310,16 +338,17 @@ type VanAddressRecord struct {
 
 type ProcessRecord struct {
 	Base
-	Name           *string `json:"name,omitempty"`
-	ParentName     *string `json:"parentName,omitempty"`
-	ImageName      *string `json:"imageName,omitempty"`
-	Image          *string `json:"image,omitempty"`
-	GroupName      *string `json:"groupName,omitempty"`
-	GroupIdentity  *string `json:"groupIdentity,omitempty"`
-	HostName       *string `json:"hostName,omitempty"`
-	SourceHost     *string `json:"sourceHost,omitempty"`
-	ProcessRole    *string `json:"processRole,omitempty"`
-	ProcessBinding *string `json:"processBinding,omitempty"`
+	Name           *string   `json:"name,omitempty"`
+	ParentName     *string   `json:"parentName,omitempty"`
+	ImageName      *string   `json:"imageName,omitempty"`
+	Image          *string   `json:"image,omitempty"`
+	GroupName      *string   `json:"groupName,omitempty"`
+	GroupIdentity  *string   `json:"groupIdentity,omitempty"`
+	HostName       *string   `json:"hostName,omitempty"`
+	SourceHost     *string   `json:"sourceHost,omitempty"`
+	ProcessRole    *string   `json:"processRole,omitempty"`
+	ProcessBinding *string   `json:"processBinding,omitempty"`
+	Addresses      []*string `json:"addresses,omitempty"`
 	connector      *string
 }
 
@@ -383,13 +412,17 @@ type FlowPairRecord struct {
 
 type FlowAggregateRecord struct {
 	Base
-	PairType        string  `json:"pairType,omitempty"`
-	RecordCount     uint64  `json:"recordCount,omitempty"`
-	SourceId        *string `json:"sourceId,omitempty"`
-	SourceName      *string `json:"sourceName,omitempty"`
-	DestinationId   *string `json:"destinationId,omitempty"`
-	DestinationName *string `json:"destinationName,omitempty"`
-	Protocol        *string `json:"protocol,omitempty"`
+	PairType            string  `json:"pairType,omitempty"`
+	RecordCount         uint64  `json:"recordCount,omitempty"`
+	SourceId            *string `json:"sourceId,omitempty"`
+	SourceName          *string `json:"sourceName,omitempty"`
+	DestinationId       *string `json:"destinationId,omitempty"`
+	DestinationName     *string `json:"destinationName,omitempty"`
+	SourceSiteId        *string `json:"sourceSiteId,omitempty"`
+	SourceSiteName      *string `json:"sourceSiteName,omitempty"`
+	DestinationSiteId   *string `json:"destinationSiteId,omitempty"`
+	DestinationSiteName *string `json:"destinationSiteName,omitempty"`
+	Protocol            *string `json:"protocol,omitempty"`
 }
 
 type ControllerRecord struct {
@@ -408,17 +441,26 @@ type ImageRecord struct {
 	// signature, url/rep, id??
 }
 
-type IngressRecord struct {
+type CollectorRecord struct {
 	Base
+	PrometheusUrl string
 }
 
-type EgressRecord struct {
-	Base
+type NetworkStatus struct {
+	Addresses []*VanAddressRecord `json:"addresses"`
+	Sites     []*SiteStatus       `json:"siteStatus"`
 }
 
-type CollectorRecord struct {
-	Base
-	PrometheusUrl string
+type SiteStatus struct {
+	Site         *SiteRecord    `json:"site"`
+	RouterStatus []RouterStatus `json:"routerStatus"`
+}
+
+type RouterStatus struct {
+	Router     *RouterRecord      `json:"router"`
+	Links      []*LinkRecord      `json:"links"`
+	Listeners  []*ListenerRecord  `json:"listeners"`
+	Connectors []*ConnectorRecord `json:"connectors"`
 }
 
 type Payload struct {
diff --git a/pkg/kube/claims/claim_verifier.go b/pkg/kube/claims/claim_verifier.go
index fa189012..54e379ff 100644
--- a/pkg/kube/claims/claim_verifier.go
+++ b/pkg/kube/claims/claim_verifier.go
@@ -21,7 +21,6 @@ import (
 	"k8s.io/client-go/util/retry"
 
 	"github.com/skupperproject/skupper/api/types"
-	"github.com/skupperproject/skupper/pkg/event"
 )
 
 const (
@@ -53,10 +52,12 @@ func newClaimVerifier(client kubernetes.Interface, namespace string, generator T
 }
 
 func (server *ClaimVerifier) checkAndUpdateClaim(name string, data []byte) (string, int) {
+	log.Printf("Checking claim %s", name)
 	claim, err := server.client.CoreV1().Secrets(server.namespace).Get(context.TODO(), name, metav1.GetOptions{})
 	if errors.IsNotFound(err) {
 		return "No such claim", http.StatusNotFound
 	} else if err != nil {
+		log.Printf("Error retrieving claim: %s", err)
 		return err.Error(), http.StatusInternalServerError
 	}
 	if claim.ObjectMeta.Labels == nil || claim.ObjectMeta.Labels[types.SkupperTypeQualifier] != types.TypeClaimRecord {
@@ -66,10 +67,10 @@ func (server *ClaimVerifier) checkAndUpdateClaim(name string, data []byte) (stri
 		if expirationString, ok := claim.ObjectMeta.Annotations[types.ClaimExpiration]; ok {
 			expiration, err := time.Parse(time.RFC3339, expirationString)
 			if err != nil {
-				event.Recordf(TokenClaimVerification, "Cannot determine expiration: %s", err)
+				log.Printf("Cannot determine expiration: %s", err)
 				return "Corrupted claim", http.StatusInternalServerError
 			} else if expiration.Before(time.Now()) {
-				event.Recordf(TokenClaimVerification, "Claim %s expired", name)
+				log.Printf("Claim %s expired", name)
 				return "No such claim", http.StatusNotFound
 			}
 		}
@@ -83,11 +84,11 @@ func (server *ClaimVerifier) checkAndUpdateClaim(name string, data []byte) (stri
 	if uses, ok := claim.ObjectMeta.Annotations[types.ClaimsRemaining]; ok {
 		remainingUses, err := strconv.Atoi(uses)
 		if err != nil {
-			event.Recordf(TokenClaimVerification, "Cannot determine remaining uses: %s", err)
+			log.Printf("Cannot determine remaining uses: %s", err)
 			return "Corrupted claim", http.StatusInternalServerError
 		}
 		if remainingUses == 0 {
-			event.Recordf(TokenClaimVerification, "Claim %s already used", name)
+			log.Printf("Claim %s already used", name)
 			return "No such claim", http.StatusNotFound
 		}
 		remainingUses -= 1
@@ -96,7 +97,7 @@ func (server *ClaimVerifier) checkAndUpdateClaim(name string, data []byte) (stri
 	if value, ok := claim.ObjectMeta.Annotations[types.ClaimsMade]; ok {
 		made, err := strconv.Atoi(value)
 		if err != nil {
-			event.Recordf(TokenClaimVerification, "Cannot determine claims made: %s", err)
+			log.Printf("Cannot determine claims made: %s", err)
 			return "Corrupted claim", http.StatusInternalServerError
 		}
 		made += 1
@@ -106,7 +107,7 @@ func (server *ClaimVerifier) checkAndUpdateClaim(name string, data []byte) (stri
 	}
 	_, err = server.client.CoreV1().Secrets(server.namespace).Update(context.TODO(), claim, metav1.UpdateOptions{})
 	if err != nil {
-		event.Recordf(TokenClaimVerification, "Error updating remaining uses: %s", err)
+		log.Printf("Error updating remaining uses: %s", err)
 		return "Internal error", http.StatusServiceUnavailable
 	}
 	return "ok", http.StatusOK
@@ -137,14 +138,14 @@ func (server *ClaimVerifier) redeemClaim(name string, subject string, data []byt
 
 func (server *ClaimVerifier) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
-		event.Recordf(TokenClaimVerification, "Bad method %s", r.Method)
+		log.Printf("Bad method %s", r.Method)
 		http.Error(w, "Only POST is supported", http.StatusMethodNotAllowed)
 		return
 	}
 	name := strings.Join(strings.Split(r.URL.Path, "/"), "")
 	body, err := ioutil.ReadAll(r.Body)
 	if err != nil {
-		event.Record(TokenClaimVerification, err.Error())
+		log.Printf("Error reading body: %s", err.Error())
 		http.Error(w, "Request body not valid", http.StatusBadRequest)
 		return
 	}
@@ -158,24 +159,24 @@ func (server *ClaimVerifier) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		if remoteSiteVersion == "" {
 			remoteSiteVersion = "undefined"
 		}
-		event.Recordf(TokenClaimVerification, "%s - remote site version is %s", err.Error(), remoteSiteVersion)
+		log.Printf("%s - remote site version is %s", err.Error(), remoteSiteVersion)
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
 	token, text, code := server.redeemClaim(name, subject, body, server.generator)
 	if token == nil {
-		event.Recordf(TokenClaimVerification, "Claim request for %s failed: %s", name, text)
+		log.Printf("Claim request for %s failed: %s", name, text)
 		http.Error(w, text, code)
 		return
 	}
 	s := json.NewYAMLSerializer(json.DefaultMetaFactory, scheme.Scheme, scheme.Scheme)
 	err = s.Encode(token, w)
 	if err != nil {
-		event.Record(TokenClaimVerification, err.Error())
+		log.Printf("Error encoding token: %s", err.Error())
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
-	event.Recordf(TokenClaimVerification, "Claim for %s succeeded", name)
+	log.Printf("Claim for %s succeeded", name)
 }
 
 const (
diff --git a/pkg/kube/contour.go b/pkg/kube/contour.go
index d0c5701f..4f266c06 100644
--- a/pkg/kube/contour.go
+++ b/pkg/kube/contour.go
@@ -86,11 +86,12 @@ func GetContourProxy(client dynamic.Interface, namespace string, name string) (*
 	return &proxy, nil
 }
 
-func CreateContourProxy(client dynamic.Interface, namespace string, name string, def IngressRoute, owner *metav1.OwnerReference) error {
+func CreateContourProxy(client dynamic.Interface, namespace string, name string, def IngressRoute, labels map[string]string, owner *metav1.OwnerReference) error {
 	obj := unstructured.Unstructured{}
 	obj.SetGroupVersionKind(httpProxyGVK)
 	obj.SetName(name)
 	obj.SetOwnerReferences([]metav1.OwnerReference{*owner})
+	obj.SetLabels(labels)
 	err := def.writeToContourProxy(&obj)
 	if err != nil {
 		return err
@@ -102,9 +103,9 @@ func CreateContourProxy(client dynamic.Interface, namespace string, name string,
 	return nil
 }
 
-func CreateContourProxies(routes []IngressRoute, owner *metav1.OwnerReference, client dynamic.Interface, namespace string) error {
+func CreateContourProxies(routes []IngressRoute, labels map[string]string, owner *metav1.OwnerReference, client dynamic.Interface, namespace string) error {
 	for _, route := range routes {
-		err := CreateContourProxy(client, namespace, route.Name, route, owner)
+		err := CreateContourProxy(client, namespace, route.Name, route, labels, owner)
 		if err != nil && !errors.IsAlreadyExists(err) {
 			return err
 		}
diff --git a/pkg/kube/ingress.go b/pkg/kube/ingress.go
index 36fbaa7c..99278550 100644
--- a/pkg/kube/ingress.go
+++ b/pkg/kube/ingress.go
@@ -111,7 +111,7 @@ func UpdateIngressRuleServiceName(ingressName string, hostPrefix string, service
 	}
 }
 
-func CreateIngress(name string, routes []IngressRoute, isNginx bool, sslPassthrough bool, ownerRefs []metav1.OwnerReference, namespace string, annotations map[string]string, clients Clients) error {
+func CreateIngress(name string, routes []IngressRoute, isNginx bool, sslPassthrough bool, ownerRefs []metav1.OwnerReference, namespace string, annotations map[string]string, labels map[string]string, clients Clients) error {
 	if useV1API(clients.GetDiscoveryClient()) {
 		if isNginx {
 			if annotations == nil {
@@ -125,13 +125,14 @@ func CreateIngress(name string, routes []IngressRoute, isNginx bool, sslPassthro
 				resolve = true
 			}
 		}
-		return ensureIngressRoutesV1(clients.GetDynamicClient(), namespace, name, routes, annotations, ownerRefs, resolve)
+		return ensureIngressRoutesV1(clients.GetDynamicClient(), namespace, name, routes, annotations, labels, ownerRefs, resolve)
 	}
 	cli := clients.GetKubeClient()
 	ingress := networkingv1.Ingress{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:        name,
 			Annotations: map[string]string{},
+			Labels:      labels,
 		},
 		Spec: networkingv1.IngressSpec{
 			Rules: []networkingv1.IngressRule{},
@@ -243,7 +244,7 @@ func appendIngressHostSuffix(desired []IngressRoute, suffix string) []IngressRou
 	return updated
 }
 
-func ensureIngressRoutesV1(client dynamic.Interface, namespace string, name string, routes []IngressRoute, annotations map[string]string, ownerRefs []metav1.OwnerReference, resolve bool) error {
+func ensureIngressRoutesV1(client dynamic.Interface, namespace string, name string, routes []IngressRoute, annotations map[string]string, labels map[string]string, ownerRefs []metav1.OwnerReference, resolve bool) error {
 	obj, err := client.Resource(ingressResource).Namespace(namespace).Get(context.TODO(), name, metav1.GetOptions{})
 	if errors.IsNotFound(err) {
 		obj = &unstructured.Unstructured{}
@@ -255,6 +256,7 @@ func ensureIngressRoutesV1(client dynamic.Interface, namespace string, name stri
 		obj.SetName(name)
 		obj.SetOwnerReferences(ownerRefs)
 		obj.SetAnnotations(annotations)
+		obj.SetLabels(labels)
 		err := writeIngressRules(routes, obj)
 		if err != nil {
 			return err
diff --git a/pkg/kube/pods.go b/pkg/kube/pods.go
index 38bd40c3..82f5c11d 100644
--- a/pkg/kube/pods.go
+++ b/pkg/kube/pods.go
@@ -121,6 +121,29 @@ func WaitForPodStatus(namespace string, clientset kubernetes.Interface, podName
 
 	return pod, err
 }
+func WaitForPodsSelectorStatus(namespace string, clientset kubernetes.Interface, selector string, status corev1.PodPhase, timeout time.Duration, interval time.Duration) ([]corev1.Pod, error) {
+	var pods []corev1.Pod
+	var pod corev1.Pod
+	var err error
+
+	ctx, cancel := context.WithTimeout(context.TODO(), timeout)
+	defer cancel()
+	err = utils.RetryWithContext(ctx, interval, func() (bool, error) {
+		pods, err = GetPods(selector, namespace, clientset)
+		if err != nil {
+			// pod does not exist yet
+			return false, nil
+		}
+		for _, pod = range pods {
+			if pod.Status.Phase != status {
+				return false, nil
+			}
+		}
+		return true, nil
+	})
+
+	return pods, err
+}
 
 func WaitForPodsStatus(namespace string, clientset kubernetes.Interface, selector string, status corev1.PodPhase, timeout time.Duration, interval time.Duration) error {
 	var err error
diff --git a/pkg/kube/qdr/tls_handler.go b/pkg/kube/qdr/tls_handler.go
index 4a047d24..5ddff82c 100644
--- a/pkg/kube/qdr/tls_handler.go
+++ b/pkg/kube/qdr/tls_handler.go
@@ -9,6 +9,7 @@ import (
 	"github.com/skupperproject/skupper/pkg/kube"
 	"github.com/skupperproject/skupper/pkg/qdr"
 	"github.com/skupperproject/skupper/pkg/service"
+	"github.com/skupperproject/skupper/pkg/site"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
@@ -144,6 +145,12 @@ func generateNewSecret(support TlsServiceSupport, tlsManager *TlsManager) (*core
 		return nil, err
 	}
 
+	siteConfig, err := tlsManager.getConfigMap(types.SiteConfigMapName)
+	if err != nil {
+		return nil, err
+	}
+	_, labels := site.GetSiteAnnotationsAndLabels(siteConfig)
+
 	serviceCredential := types.Credential{
 		CA:          types.ServiceCaSecret,
 		Name:        support.Credentials,
@@ -151,6 +158,7 @@ func generateNewSecret(support TlsServiceSupport, tlsManager *TlsManager) (*core
 		Hosts:       []string{support.Address},
 		ConnectJson: false,
 		Post:        false,
+		Labels:      labels,
 	}
 
 	ownerReference := metav1.OwnerReference{
@@ -182,8 +190,11 @@ func (manager *TlsManager) GetSecret(name string) (*corev1.Secret, error) {
 }
 
 func (manager *TlsManager) GetConfigMap() (*corev1.ConfigMap, error) {
+	return manager.getConfigMap(types.TransportConfigMapName)
+}
 
-	result, err := manager.KubeClient.CoreV1().ConfigMaps(manager.Namespace).Get(context.TODO(), types.TransportConfigMapName, metav1.GetOptions{})
+func (manager *TlsManager) getConfigMap(name string) (*corev1.ConfigMap, error) {
+	result, err := manager.KubeClient.CoreV1().ConfigMaps(manager.Namespace).Get(context.TODO(), name, metav1.GetOptions{})
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/kube/secrets.go b/pkg/kube/secrets.go
index 70af890f..c7102fed 100644
--- a/pkg/kube/secrets.go
+++ b/pkg/kube/secrets.go
@@ -22,6 +22,7 @@ func NewCertAuthority(ca types.CertAuthority, owner *metav1.OwnerReference, name
 		return existing, nil
 	} else if errors.IsNotFound(err) {
 		newCA := certs.GenerateCASecret(ca.Name, ca.Name)
+		newCA.Labels = ca.Labels
 		if owner != nil {
 			newCA.ObjectMeta.OwnerReferences = []metav1.OwnerReference{
 				*owner,
@@ -80,6 +81,7 @@ func NewSecret(cred types.Credential, owner *metav1.OwnerReference, namespace st
 	}
 
 	secret = PrepareNewSecret(cred, caSecret, connectJsonHost)
+	secret.ObjectMeta.Labels = cred.Labels
 
 	if owner != nil {
 		secret.ObjectMeta.OwnerReferences = []metav1.OwnerReference{
diff --git a/pkg/service_sync/service_sync.go b/pkg/service_sync/service_sync.go
index 2abfbc85..a08aa1e7 100644
--- a/pkg/service_sync/service_sync.go
+++ b/pkg/service_sync/service_sync.go
@@ -267,7 +267,7 @@ func getAddresses(services []types.ServiceInterface) []string {
 }
 
 func equivalentServiceDefinition(a *types.ServiceInterface, b *types.ServiceInterface) bool {
-	if a.Protocol != b.Protocol || !reflect.DeepEqual(a.Ports, b.Ports) || a.EventChannel != b.EventChannel || a.Aggregate != b.Aggregate || !reflect.DeepEqual(a.Labels, b.Labels) || !reflect.DeepEqual(a.Annotations, b.Annotations) {
+	if a.Protocol != b.Protocol || !reflect.DeepEqual(a.Ports, b.Ports) || a.EventChannel != b.EventChannel || a.Aggregate != b.Aggregate || !reflect.DeepEqual(a.Labels, b.Labels) || !reflect.DeepEqual(a.Annotations, b.Annotations) || a.TlsCredentials != b.TlsCredentials || a.TlsCertAuthority != b.TlsCertAuthority || a.PublishNotReadyAddresses != b.PublishNotReadyAddresses {
 		return false
 	}
 	if a.Headless == nil && b.Headless == nil {
diff --git a/pkg/site/site.go b/pkg/site/site.go
index 8e99bcb3..904ddb54 100644
--- a/pkg/site/site.go
+++ b/pkg/site/site.go
@@ -780,6 +780,16 @@ func ReadSiteConfig(siteConfig *corev1.ConfigMap, namespace string, defaultIngre
 		result.Spec.PrometheusServer.PodAnnotations = asMap(splitWithEscaping(prometheusPodAnnotations, ',', '\\'))
 	}
 
+	annotations, labels := GetSiteAnnotationsAndLabels(siteConfig)
+	result.Spec.Annotations = annotations
+	result.Spec.Labels = labels
+	if len(errs) > 0 {
+		return &result, fmt.Errorf(strings.Join(errs, ", "))
+	}
+	return &result, nil
+}
+
+func GetSiteAnnotationsAndLabels(siteConfig *corev1.ConfigMap) (map[string]string, map[string]string) {
 	annotationExclusions := []string{}
 	labelExclusions := []string{}
 	annotations := map[string]string{}
@@ -795,7 +805,6 @@ func ReadSiteConfig(siteConfig *corev1.ConfigMap, namespace string, defaultIngre
 	for _, key := range annotationExclusions {
 		delete(annotations, key)
 	}
-	result.Spec.Annotations = annotations
 	labels := map[string]string{}
 	for key, value := range siteConfig.ObjectMeta.Labels {
 		if key != types.SiteControllerIgnore {
@@ -805,11 +814,7 @@ func ReadSiteConfig(siteConfig *corev1.ConfigMap, namespace string, defaultIngre
 	for _, key := range labelExclusions {
 		delete(labels, key)
 	}
-	result.Spec.Labels = labels
-	if len(errs) > 0 {
-		return &result, fmt.Errorf(strings.Join(errs, ", "))
-	}
-	return &result, nil
+	return annotations, labels
 }
 
 func UpdateLogging(config types.SiteConfigSpec, configmap *corev1.ConfigMap) bool {
diff --git a/pkg/utils/utils.go b/pkg/utils/utils.go
index 8b5b8bfb..bc7658f8 100644
--- a/pkg/utils/utils.go
+++ b/pkg/utils/utils.go
@@ -18,6 +18,7 @@ import (
 	"crypto/rand"
 	"io"
 	"os"
+	"os/user"
 	"regexp"
 	"strings"
 )
@@ -162,3 +163,11 @@ func DefaultNumber[T Number](values ...T) T {
 	}
 	return DefaultNumber(values[1:]...)
 }
+
+func ReadUsername() string {
+	u, err := user.Current()
+	if err != nil {
+		return DefaultStr(os.Getenv("USER"), os.Getenv("USERNAME"))
+	}
+	return strings.Join(strings.Fields(u.Username), "")
+}


Test patch:
diff --git a/Dockerfile.ci-test b/Dockerfile.ci-test
index eb134b6a..fb2e64f2 100644
--- a/Dockerfile.ci-test
+++ b/Dockerfile.ci-test
@@ -1,4 +1,4 @@
-FROM golang:1.19 AS builder
+FROM golang:1.20 AS builder
 
 WORKDIR /go/src/app
 
diff --git a/client/podman/container_test.go b/client/podman/container_test.go
index 65540dc1..14c077ae 100644
--- a/client/podman/container_test.go
+++ b/client/podman/container_test.go
@@ -2,6 +2,7 @@ package podman
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"path"
 	"strings"
@@ -9,8 +10,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/go-openapi/runtime"
 	"github.com/skupperproject/skupper/api/types"
 	"github.com/skupperproject/skupper/client/container"
+	"github.com/skupperproject/skupper/client/generated/libpod/client/containers"
 	"github.com/skupperproject/skupper/pkg/images"
 	"github.com/skupperproject/skupper/pkg/utils"
 	"gotest.tools/assert"
@@ -139,7 +142,7 @@ func TestContainer(t *testing.T) {
 	})
 
 	// Listing containers
-	t.Run("container-inspect", func(t *testing.T) {
+	containerInspectTest := func(t *testing.T) {
 		c, err := cli.ContainerInspect(name)
 		assert.Assert(t, err, "error inspecting container")
 
@@ -161,7 +164,8 @@ func TestContainer(t *testing.T) {
 		assert.Equal(t, sourcePort, p.Host)
 		assert.Equal(t, hostIP, p.HostIP)
 		assert.Equal(t, targetPort, p.Target)
-	})
+	}
+	t.Run("container-inspect", containerInspectTest)
 
 	t.Run("container-exec", func(t *testing.T) {
 		out, err := cli.ContainerExec(name, strings.Split("ls -1 /app", " "))
@@ -186,6 +190,16 @@ func TestContainer(t *testing.T) {
 		assert.Equal(t, name, c.NetworkNames()[0])
 	})
 
+	// Updating container image
+	image = strings.ReplaceAll(image, ":main", ":latest")
+	t.Run("container-update-image", func(t *testing.T) {
+		c, err := cli.ContainerUpdateImage(name, image)
+		assert.Assert(t, err)
+		assert.Equal(t, c.Image, image)
+		containerInspectTest(t)
+	})
+	t.Run("container-inspect-after-image-update", containerInspectTest)
+
 	// Cleaning up
 	t.Run("cleanup", func(t *testing.T) {
 		var cleanupErrors []error
@@ -230,3 +244,144 @@ func ValidateStrings(t *testing.T, original []string, final []string) {
 		assert.Assert(t, utils.StringSliceContains(final, v), "string not found %s", v)
 	}
 }
+
+func TestContainerUpdateMock(t *testing.T) {
+	image := images.GetServiceControllerImageName()
+	cli := NewPodmanClientMock(mockContainers(image))
+
+	// starting the container
+	assert.Assert(t, cli.ContainerStart("my-container"))
+
+	// retrieving container
+	cc, err := cli.ContainerInspect("my-container")
+	assert.Assert(t, err)
+	startedAt := cc.StartedAt
+
+	newImage := strings.Replace(image, ":main", ":updated", -1)
+	_, err = cli.ContainerUpdateImage("my-container", newImage)
+	assert.Assert(t, err)
+
+	// validating that the container has the new image and has been restarted
+	cc, err = cli.ContainerInspect("my-container")
+	assert.Assert(t, err)
+	assert.Assert(t, time.Now().After(startedAt))
+	assert.Equal(t, cc.Image, newImage)
+
+	// assert that there is no other container left
+	cl, err := cli.ContainerList()
+	assert.Assert(t, err)
+	assert.Equal(t, len(cl), 1)
+}
+
+func TestContainerUpdateErrorMock(t *testing.T) {
+	image := images.GetServiceControllerImageName()
+	newImage := strings.Replace(image, ":main", ":updated", -1)
+	cli := NewPodmanClientMock(mockContainers(image))
+	mock := cli.RestClient.(*RestClientMock)
+
+	tests := []struct {
+		name      string
+		errorHook func(operation *runtime.ClientOperation) error
+	}{{
+		// must stop and remove the new container
+		name: "error-stopping-old-container",
+		errorHook: func(operation *runtime.ClientOperation) error {
+			if operation.ID == "ContainerStopLibpod" {
+				params := operation.Params.(*containers.ContainerStopLibpodParams)
+				// returns error when stopping the current container
+				if params.Name == "my-container" {
+					return fmt.Errorf("error stopping %q", params.Name)
+				}
+			}
+			return nil
+		},
+	}, {
+		// must start the old container, stop and remove the new one
+		name: "error-renaming-old-container",
+		errorHook: func(operation *runtime.ClientOperation) error {
+			if operation.ID == "ContainerRenameLibpod" {
+				params := operation.Params.(*containers.ContainerRenameLibpodParams)
+				curName := params.PathName
+				newName := params.QueryName
+				// returns error when renaming current container to a backup name
+				if curName == "my-container" {
+					return fmt.Errorf("error renaming container %q to %q", curName, newName)
+				}
+			}
+			return nil
+		},
+	}, {
+		// must rename old container back, start the old container, stop and remove the new one
+		name: "error-renaming-new-container-as-old-container",
+		errorHook: func(operation *runtime.ClientOperation) error {
+			if operation.ID == "ContainerRenameLibpod" {
+				params := operation.Params.(*containers.ContainerRenameLibpodParams)
+				curName := params.PathName
+				newName := params.QueryName
+				// return error when renaming new container to original name
+				if strings.Contains(curName, "-new-") && newName == "my-container" {
+					return fmt.Errorf("error renaming container %q to %q", curName, newName)
+				}
+			}
+			return nil
+		},
+	}, {
+		// must rename old container back, start the old container, stop and remove the new one
+		name: "error-starting-new-container",
+		errorHook: func(operation *runtime.ClientOperation) error {
+			if operation.ID == "ContainerStartLibpod" {
+				params := operation.Params.(*containers.ContainerStartLibpodParams)
+				hasBackupContainer := true
+				for _, c := range mock.Containers {
+					if strings.Contains(c.Name, "-new-") {
+						hasBackupContainer = false
+						break
+					}
+				}
+				// only returns error when starting the updated container
+				if len(mock.Containers) == 2 && hasBackupContainer && params.Name == "my-container" {
+					return fmt.Errorf("error starting %q", params.Name)
+				}
+			}
+			return nil
+		},
+	}}
+
+	// iterate through test scenarios
+	for _, test := range tests {
+		mock.Containers = mockContainers(image)
+		mock.ErrorHook = test.errorHook
+		t.Run(test.name, func(t *testing.T) {
+			_, err := cli.ContainerUpdate("my-container", func(newContainer *container.Container) {
+				newContainer.Image = newImage
+			})
+			assert.Assert(t, err != nil)
+			assert.Equal(t, 1, len(mock.Containers))
+			assert.Equal(t, "my-container", mock.Containers[0].Name)
+			assert.Equal(t, true, mock.Containers[0].Running)
+		})
+	}
+}
+
+func mockContainers(image string) []*container.Container {
+	return []*container.Container{{
+		ID:    "abcd",
+		Name:  "my-container",
+		Image: image,
+		Env:   map[string]string{"var1": "val1", "var2": "val2"},
+		Networks: map[string]container.ContainerNetworkInfo{
+			"skupper": container.ContainerNetworkInfo{
+				ID:        "skupper",
+				IPAddress: "172.17.0.10",
+				Gateway:   "172.17.0.1",
+				Aliases:   []string{"skupper", "service-controller"},
+			},
+		},
+		Ports: []container.Port{
+			{Host: "8888", HostIP: "10.0.0.1", Target: "8080", Protocol: "tcp"},
+		},
+		Running:   true,
+		CreatedAt: time.Now(),
+		StartedAt: time.Now(),
+	}}
+}
diff --git a/client/router_create_test.go b/client/router_create_test.go
index f88f41b3..b609e51c 100644
--- a/client/router_create_test.go
+++ b/client/router_create_test.go
@@ -42,6 +42,7 @@ func TestRouterCreateDefaults(t *testing.T) {
 		clusterLocal         bool
 		depsExpected         []string
 		cmsExpected          []string
+		cmsExtraExpected     []string
 		rolesExpected        []string
 		roleBindingsExpected []string
 		clusterRolesExpected []string
@@ -67,6 +68,7 @@ func TestRouterCreateDefaults(t *testing.T) {
 			clusterLocal:         true,
 			depsExpected:         []string{"skupper-service-controller", "skupper-router"},
 			cmsExpected:          []string{types.ServiceInterfaceConfigMap, types.TransportConfigMapName},
+			cmsExtraExpected:     []string{types.NetworkStatusConfigMapName, types.SiteLeaderLockName},
 			rolesExpected:        []string{types.TransportRoleName, types.ControllerRoleName},
 			clusterRolesExpected: []string{types.ControllerClusterRoleName},
 			clusterRoleResources: sets.NewString("skupperclusterpolicies", "nodes", "namespaces"),
@@ -103,6 +105,7 @@ func TestRouterCreateDefaults(t *testing.T) {
 			clusterLocal:         false,
 			depsExpected:         []string{"skupper-service-controller", "skupper-router", "skupper-prometheus"},
 			cmsExpected:          []string{types.ServiceInterfaceConfigMap, types.TransportConfigMapName},
+			cmsExtraExpected:     []string{types.NetworkStatusConfigMapName, types.SiteLeaderLockName},
 			rolesExpected:        []string{types.TransportRoleName, types.ControllerRoleName, types.PrometheusRoleName},
 			clusterRolesExpected: []string{types.ControllerClusterRoleName},
 			clusterRoleResources: sets.NewString("skupperclusterpolicies", "nodes", "namespaces"),
@@ -140,6 +143,7 @@ func TestRouterCreateDefaults(t *testing.T) {
 			clusterLocal:         false,
 			depsExpected:         []string{"skupper-service-controller", "skupper-router", "skupper-prometheus"},
 			cmsExpected:          []string{types.ServiceInterfaceConfigMap, types.TransportConfigMapName, "skupper-sasl-config"},
+			cmsExtraExpected:     []string{types.NetworkStatusConfigMapName, types.SiteLeaderLockName},
 			rolesExpected:        []string{types.TransportRoleName, types.ControllerRoleName, types.PrometheusRoleName},
 			clusterRolesExpected: []string{types.ControllerClusterRoleName},
 			clusterRoleResources: sets.NewString("skupperclusterpolicies", "nodes", "namespaces"),
@@ -178,6 +182,7 @@ func TestRouterCreateDefaults(t *testing.T) {
 			clusterLocal:         false,
 			depsExpected:         []string{"skupper-service-controller", "skupper-router", "skupper-prometheus"},
 			cmsExpected:          []string{types.ServiceInterfaceConfigMap, types.TransportConfigMapName},
+			cmsExtraExpected:     []string{types.NetworkStatusConfigMapName, types.SiteLeaderLockName},
 			rolesExpected:        []string{types.TransportRoleName, types.ControllerRoleName, types.PrometheusRoleName},
 			clusterRolesExpected: []string{types.ControllerClusterRoleName},
 			clusterRoleResources: sets.NewString("skupperclusterpolicies", "nodes", "namespaces"),
@@ -215,6 +220,7 @@ func TestRouterCreateDefaults(t *testing.T) {
 			clusterLocal:         false,
 			depsExpected:         []string{"skupper-service-controller", "skupper-router", "skupper-prometheus"},
 			cmsExpected:          []string{types.ServiceInterfaceConfigMap, types.TransportConfigMapName},
+			cmsExtraExpected:     []string{types.NetworkStatusConfigMapName, types.SiteLeaderLockName},
 			rolesExpected:        []string{types.TransportRoleName, types.ControllerRoleName, types.PrometheusRoleName},
 			clusterRolesExpected: []string{types.ControllerClusterRoleName},
 			clusterRoleResources: sets.NewString("skupperclusterpolicies", "nodes", "namespaces"),
@@ -250,6 +256,7 @@ func TestRouterCreateDefaults(t *testing.T) {
 			clusterLocal:         true,
 			depsExpected:         []string{"skupper-service-controller", "skupper-router"},
 			cmsExpected:          []string{types.ServiceInterfaceConfigMap, types.TransportConfigMapName},
+			cmsExtraExpected:     []string{types.NetworkStatusConfigMapName, types.SiteLeaderLockName},
 			rolesExpected:        []string{types.TransportRoleName, types.ControllerRoleName},
 			clusterRolesExpected: []string{types.ControllerClusterRoleName},
 			clusterRoleResources: sets.NewString("ingresses", "skupperclusterpolicies", "namespaces", "services", "configmaps", "pods", "nodes", "secrets", "deployments", "statefulsets", "daemonsets"),
@@ -406,9 +413,14 @@ func TestRouterCreateDefaults(t *testing.T) {
 			},
 		})
 
-		// TODO: make more deterministic
-		time.Sleep(time.Second * 1)
+		// TODO: make more deterministic, allow for leader election
+		time.Sleep(time.Second * 10)
 		assert.Check(t, err, c.doc)
+		if isCluster {
+			for _, cm := range c.cmsExtraExpected {
+				c.cmsExpected = append(c.cmsExpected, cm)
+			}
+		}
 		if diff := cmp.Diff(c.depsExpected, depsFound, c.opts...); diff != "" {
 			t.Errorf("TestRouterCreateDefaults "+c.doc+" deployments mismatch (-want +got):\n%s", diff)
 		}
diff --git a/client/serviceinterface_create_test.go b/client/serviceinterface_create_test.go
index 4739c868..8bf70bac 100644
--- a/client/serviceinterface_create_test.go
+++ b/client/serviceinterface_create_test.go
@@ -85,6 +85,7 @@ func TestServiceInterfaceCreate(t *testing.T) {
 		user             string
 		depsExpected     []string
 		cmsExpected      []string
+		cmsExtraExpected []string
 		rolesExpected    []string
 		svcsExpected     []string
 		realSvcsExpected []string
@@ -138,16 +139,17 @@ func TestServiceInterfaceCreate(t *testing.T) {
 		// of VAN service interface creation.
 		// I.e., certain deployments and services should be created.
 		{
-			namespace:     "vsic-5",
-			doc:           "Check basic deployments.",
-			init:          true,
-			addr:          "vsic-5-addr",
-			proto:         "tcp",
-			ports:         []int{1999, 2000},
-			expectedErr:   "",
-			depsExpected:  []string{"skupper-router", "skupper-service-controller"},
-			cmsExpected:   []string{types.TransportConfigMapName, types.ServiceInterfaceConfigMap},
-			rolesExpected: []string{types.ControllerRoleName, types.TransportRoleName},
+			namespace:        "vsic-5",
+			doc:              "Check basic deployments.",
+			init:             true,
+			addr:             "vsic-5-addr",
+			proto:            "tcp",
+			ports:            []int{1999, 2000},
+			expectedErr:      "",
+			depsExpected:     []string{"skupper-router", "skupper-service-controller"},
+			cmsExpected:      []string{types.TransportConfigMapName, types.ServiceInterfaceConfigMap},
+			cmsExtraExpected: []string{types.NetworkStatusConfigMapName, types.SiteLeaderLockName},
+			rolesExpected:    []string{types.ControllerRoleName, types.TransportRoleName},
 			// The list of expected services is slightly different in
 			// the mock environment vs. a real cluster.
 			// It usually takes 10 or 12 seconds for the address service to
@@ -168,6 +170,7 @@ func TestServiceInterfaceCreate(t *testing.T) {
 			expectedErr:      "",
 			depsExpected:     []string{"skupper-router", "skupper-service-controller"},
 			cmsExpected:      []string{types.TransportConfigMapName, types.ServiceInterfaceConfigMap},
+			cmsExtraExpected: []string{types.NetworkStatusConfigMapName, types.SiteLeaderLockName},
 			rolesExpected:    []string{types.ControllerRoleName, types.TransportRoleName},
 			svcsExpected:     []string{types.LocalTransportServiceName, types.TransportServiceName},
 			realSvcsExpected: []string{types.LocalTransportServiceName, types.TransportServiceName, "vsic-6-addr"},
@@ -314,6 +317,11 @@ func TestServiceInterfaceCreate(t *testing.T) {
 		}
 
 		// Check all the lists of expected entities.
+		if isCluster {
+			for _, cm := range testcase.cmsExtraExpected {
+				testcase.cmsExpected = append(testcase.cmsExpected, cm)
+			}
+		}
 		check_result(t, testcase.namespace, testcase.timeout, "dependencies", testcase.depsExpected, &depsFound, testcase.doc)
 		check_result(t, testcase.namespace, testcase.timeout, "config maps", testcase.cmsExpected, &cmsFound, testcase.doc)
 		check_result(t, testcase.namespace, testcase.timeout, "roles", testcase.rolesExpected, &rolesFound, testcase.doc)
diff --git a/cmd/skupper/skupper_cluster_test.go b/cmd/skupper/skupper_cluster_test.go
index 0f50b8e5..f880128b 100644
--- a/cmd/skupper/skupper_cluster_test.go
+++ b/cmd/skupper/skupper_cluster_test.go
@@ -1039,7 +1039,7 @@ func TestUnexposeWithCluster(t *testing.T) {
 			args:            []string{},
 			expectedCapture: "",
 			expectedOutput:  "",
-			expectedError:   "expose target and name must be specified (e.g. 'skupper expose deployment <name>')",
+			expectedError:   "target and name must be specified (e.g. 'skupper unexpose deployment <name>')",
 			realCluster:     false,
 		},
 		{
@@ -1047,7 +1047,7 @@ func TestUnexposeWithCluster(t *testing.T) {
 			args:            []string{"deployment"},
 			expectedCapture: "",
 			expectedOutput:  "",
-			expectedError:   "expose target and name must be specified (e.g. 'skupper expose deployment <name>')",
+			expectedError:   "target and name must be specified (e.g. 'skupper unexpose deployment <name>')",
 			realCluster:     false,
 		},
 		{
@@ -1354,13 +1354,77 @@ func TestBindWithCluster(t *testing.T) {
 			realCluster:     false,
 		},
 		{
-			doc:             "bind-test6",
+			doc:             "bind-test5",
 			args:            []string{"tcp-go-echo", "deployment", "tcp-go-echo", "--target-namespace", anotherNs},
 			expectedCapture: "",
 			expectedOutput:  "",
 			expectedError:   "",
 			realCluster:     true,
 		},
+		{
+			doc:             "bind-test6",
+			args:            []string{"web", "service", "web", "--headless"},
+			expectedCapture: "",
+			expectedOutput:  "",
+			expectedError:   "--headless option is only valid for statefulsets",
+			realCluster:     true,
+		},
+		{
+			doc:             "bind-test7",
+			args:            []string{"test", "statefulset", "test", "--proxy-cpu", "2"},
+			expectedCapture: "",
+			expectedOutput:  "",
+			expectedError:   "--proxy-cpu option is only valid for binding statefulsets using headless services",
+			realCluster:     true,
+		},
+		{
+			doc:             "bind-test8",
+			args:            []string{"test", "service", "test", "--proxy-memory", "2G"},
+			expectedCapture: "",
+			expectedOutput:  "",
+			expectedError:   "--proxy-memory option is only valid for binding statefulsets using headless services",
+			realCluster:     true,
+		},
+		{
+			doc:             "bind-test9",
+			args:            []string{"test", "statefulset", "test", "--proxy-cpu-limit", "2"},
+			expectedCapture: "",
+			expectedOutput:  "",
+			expectedError:   "--proxy-cpu-limit option is only valid for binding statefulsets using headless services",
+			realCluster:     true,
+		},
+		{
+			doc:             "bind-test10",
+			args:            []string{"test", "statefulset", "test", "--proxy-memory-limit", "2G"},
+			expectedCapture: "",
+			expectedOutput:  "",
+			expectedError:   "--proxy-memory-limit option is only valid for binding statefulsets using headless services",
+			realCluster:     true,
+		},
+		{
+			doc:             "bind-test11",
+			args:            []string{"test", "statefulset", "test", "--proxy-pod-affinity", "test"},
+			expectedCapture: "",
+			expectedOutput:  "",
+			expectedError:   "--proxy-pod-affinity option is only valid for binding statefulsets using headless services",
+			realCluster:     true,
+		},
+		{
+			doc:             "bind-test12",
+			args:            []string{"test", "statefulset", "test", "--proxy-pod-antiaffinity", "test"},
+			expectedCapture: "",
+			expectedOutput:  "",
+			expectedError:   "--proxy-pod-antiaffinity option is only valid for binding statefulsets using headless services",
+			realCluster:     true,
+		},
+		{
+			doc:             "bind-test13",
+			args:            []string{"test", "statefulset", "test", "--proxy-node-selector", "test"},
+			expectedCapture: "",
+			expectedOutput:  "",
+			expectedError:   "--proxy-node-selector option is only valid for binding statefulsets using headless services",
+			realCluster:     true,
+		},
 	}
 
 	namespace := "cmd-bind-cluster-test-" + strings.ToLower(utils.RandomId(4))
@@ -1604,68 +1668,3 @@ func TestDebugDumpWithCluster(t *testing.T) {
 		testCommand(t, cmd, tc.doc, tc.expectedError, tc.expectedCapture, tc.expectedOutput, tc.outputRegExp, tc.args...)
 	}
 }
-
-func TestNetworkStatusWithCluster(t *testing.T) {
-	if os.Getenv("USER") == "circleci" {
-		t.Skipf("Test is temporarily disabled")
-	}
-	testcases := []testCase{
-		{
-			doc:             "network-status-test1",
-			args:            []string{"--help"},
-			expectedCapture: "",
-			expectedOutput:  "Shows information about the current site, and connected sites.",
-			expectedError:   "",
-			realCluster:     true,
-		},
-		{
-			doc:             "network-status-test2",
-			args:            []string{"status"},
-			expectedCapture: "",
-			expectedOutput:  "",
-			expectedError:   "",
-			outputRegExp:    "^Sites\\:\\n╰─\\s\\[local\\]\\s[0-9a-f-]*\\s-\\s.*\\n\\s{3}URL:\\s\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\n\\s{3}mode:\\s.*\\n\\s{3}name:\\s.*\\n\\s{3}namespace:\\s.*\\n\\s{3}version:\\s.*\\n\\s{3}╰─\\sServices\\:\\n\\s{6}╰─\\sname:\\s.*\\n\\s{9}address:\\s.*\\:\\s\\d{4}\\n\\s{9}protocol:\\s.*\\n\\s{9}╰─\\sTargets\\:\\n\\s{12}╰─\\sname:\\s.*",
-			realCluster:     true,
-		},
-	}
-
-	namespace := "cmd-debug-network-status-cluster-test-" + strings.ToLower(utils.RandomId(4))
-
-	if !*clusterRun {
-		lightRed := "\033[1;31m"
-		resetColor := "\033[0m"
-		t.Skip(fmt.Sprintf("%sSkipping: This test only works in real clusters.%s", string(lightRed), string(resetColor)))
-	}
-
-	testClient = &SkupperTestClient{
-		SkupperKube: &SkupperKube{
-			Namespace: namespace,
-		},
-	}
-	testClient.NewClient(nil, nil)
-	cli := testClient.Cli
-
-	if c, ok := cli.(*client.VanClient); ok {
-		_, err := kube.NewNamespace(namespace, c.KubeClient)
-		assert.Check(t, err)
-		defer kube.DeleteNamespace(namespace, c.KubeClient)
-	}
-	skupperInit(t, []string{"--console-ingress=none"}...)
-	skupperExpose(t, []string{"service", "tcp-go-echo", "--port", "9090", "--address", "tcp-go-echo-dup"}...)
-
-	// wait for the service controller to be available
-	time.Sleep(15 * time.Second)
-
-	testPath := "./tmp/"
-	os.Mkdir(testPath, 0755)
-	defer os.RemoveAll(testPath)
-
-	for _, tc := range testcases {
-		if tc.realCluster && !*clusterRun {
-			continue
-		}
-		cmd := NewCmdNetworkStatus(testClient.Network())
-		silenceCobra(cmd)
-		testCommand(t, cmd, tc.doc, tc.expectedError, tc.expectedCapture, tc.expectedOutput, tc.outputRegExp, tc.args...)
-	}
-}
diff --git a/cmd/skupper/skupper_test.go b/cmd/skupper/skupper_test.go
index 3d288120..b3ac795b 100644
--- a/cmd/skupper/skupper_test.go
+++ b/cmd/skupper/skupper_test.go
@@ -2,6 +2,7 @@ package main
 
 import (
 	"flag"
+	"github.com/spf13/cobra"
 	"os"
 	"testing"
 	"time"
@@ -93,8 +94,11 @@ func TestExposeTargetArgs(t *testing.T) {
 	genericError := "expose target and name must be specified (e.g. 'skupper expose deployment <name>')"
 	targetError := "target type must be one of: [deployment, statefulset, pods, service, deploymentconfig]"
 
+	cmd := cobra.Command{
+		Use: "expose",
+	}
 	e := func(args []string) error {
-		return s.ExposeArgs(nil, args)
+		return s.ExposeArgs(&cmd, args)
 	}
 
 	assert.Error(t, e([]string{}), genericError)
diff --git a/pkg/domain/podman/site_test.go b/pkg/domain/podman/site_test.go
index 8dc72696..61e61c8a 100644
--- a/pkg/domain/podman/site_test.go
+++ b/pkg/domain/podman/site_test.go
@@ -6,6 +6,7 @@ package podman
 import (
 	"testing"
 
+	"github.com/skupperproject/skupper/api/types"
 	"github.com/skupperproject/skupper/client/container"
 	"github.com/skupperproject/skupper/pkg/domain"
 	"github.com/skupperproject/skupper/pkg/utils"
@@ -41,6 +42,12 @@ func TestSiteHandler(t *testing.T) {
 			AuthMode:            "internal",
 			ConsoleUser:         "internal",
 			ConsolePassword:     "internal",
+			PrometheusOpts: types.PrometheusServerOptions{
+				ExternalServer: "http://10.0.0.1:8080/v1",
+				AuthMode:       "internal",
+				User:           "admin",
+				Password:       "admin",
+			},
 		},
 	}, {
 		name: "flow-collector-internal-auth-ingress-none",
@@ -87,7 +94,7 @@ func TestSiteHandler(t *testing.T) {
 			expIngHosts := 1 + len(scenarioSite.IngressHosts)
 			expDeployments := 2
 			if scenarioSite.EnableFlowCollector {
-				expDeployments += 1
+				expDeployments += 2
 			}
 			assert.Assert(t, len(podmanSite.IngressHosts) == expIngHosts)
 			assert.Equal(t, len(podmanSite.Deployments), expDeployments)
@@ -101,6 +108,10 @@ func TestSiteHandler(t *testing.T) {
 			}
 			assert.Assert(t, len(podmanSite.Credentials) > 0)
 			assert.Assert(t, len(podmanSite.CertAuthorities) > 0)
+			assert.Equal(t, scenarioSite.PrometheusOpts.ExternalServer, podmanSite.PrometheusOpts.ExternalServer)
+			assert.Equal(t, scenarioSite.PrometheusOpts.AuthMode, podmanSite.PrometheusOpts.AuthMode)
+			assert.Equal(t, scenarioSite.PrometheusOpts.User, podmanSite.PrometheusOpts.User)
+			assert.Equal(t, scenarioSite.PrometheusOpts.Password, podmanSite.PrometheusOpts.Password)
 		})
 	}
 }
diff --git a/pkg/domain/podman/update/common_test.go b/pkg/domain/podman/update/common_test.go
new file mode 100644
index 00000000..b7503fd9
--- /dev/null
+++ b/pkg/domain/podman/update/common_test.go
@@ -0,0 +1,305 @@
+//go:build podman
+// +build podman
+
+package update
+
+import (
+	_ "embed"
+	"flag"
+	"os"
+	"regexp"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/skupperproject/skupper/client/container"
+	"github.com/skupperproject/skupper/client/podman"
+	domainpodman "github.com/skupperproject/skupper/pkg/domain/podman"
+	"github.com/skupperproject/skupper/pkg/images"
+	"gotest.tools/assert"
+)
+
+// some podman tests require a cluster as well
+var clusterRun = flag.Bool("use-cluster", false, "run tests against a configured cluster")
+
+func TestMain(m *testing.M) {
+	flag.Parse()
+	os.Exit(m.Run())
+}
+
+var (
+	//go:embed skrouterd.json
+	skrouterdJson string
+	//go:embed skupper-services.json
+	skupperServicesJson string
+)
+
+func TestContainerImagesTask(t *testing.T) {
+	cli := podman.NewPodmanClientMock(mockContainers())
+	mock := cli.RestClient.(*podman.RestClientMock)
+	assert.Assert(t, mock.MockVolumeFiles(mockVolumes()))
+	defer func() {
+		_ = mock.CleanupMockVolumeDir()
+	}()
+	sh := domainpodman.NewSitePodmanHandlerFromCli(cli)
+	site, err := sh.Get()
+	assert.Assert(t, err)
+	imagesTask := NewContainerImagesTask(cli)
+
+	tests := []struct {
+		name    string
+		version string
+		changed bool
+	}{
+		{
+			name:    "newer-version",
+			version: "1.5.0",
+			changed: true,
+		},
+		{
+			name:    "older-version",
+			version: "1.4.0",
+			changed: false,
+		},
+		{
+			name:    "invalid-version",
+			version: "invalid-version",
+			changed: false,
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			imagesTask.version = test.version
+			t.Logf("New version: %s - current mock site version: %s", test.version, site.GetVersion())
+			assert.Assert(t, imagesTask.AppliesTo(site.GetVersion()) == test.changed)
+			res := imagesTask.Run()
+			assert.Assert(t, len(res.Errors) == 0)
+			assert.Assert(t, res.Changed() == test.changed)
+			expectedChanges := 4
+			if !test.changed {
+				expectedChanges = 0
+			}
+			assert.Equal(t, expectedChanges, len(res.GetChanges()), "Changes performed: ", res.GetChanges())
+		})
+	}
+}
+
+func TestVersionUpdateTask(t *testing.T) {
+	cli := podman.NewPodmanClientMock(mockContainers())
+	mock := cli.RestClient.(*podman.RestClientMock)
+	assert.Assert(t, mock.MockVolumeFiles(mockVolumes()))
+	defer func() {
+		_ = mock.CleanupMockVolumeDir()
+	}()
+	ch := domainpodman.NewRouterConfigHandlerPodman(cli)
+
+	versionUpdTask := NewVersionUpdateTask(cli)
+	tests := []struct {
+		name       string
+		curVersion string
+		version    string
+		changed    bool
+	}{
+		{
+			name:       "newer-version",
+			curVersion: "1.4.2",
+			version:    "1.5.0",
+			changed:    true,
+		},
+		{
+			name:       "older-version",
+			curVersion: "1.4.2",
+			version:    "1.4.0",
+			changed:    false,
+		},
+		{
+			name:       "invalid-old-version",
+			curVersion: "invalid-old-version",
+			version:    "1.5.0",
+			changed:    false,
+		},
+		{
+			name:       "invalid-new-version",
+			curVersion: "1.4.2",
+			version:    "invalid-new-version",
+			changed:    false,
+		},
+		{
+			name:       "invalid-versions",
+			curVersion: "invalid-old-version",
+			version:    "invalid-new-version",
+			changed:    false,
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// updating site version to curVersion
+			cfg, err := ch.GetRouterConfig()
+			assert.Assert(t, err)
+			siteMeta := cfg.GetSiteMetadata()
+			siteMeta.Version = test.curVersion
+			cfg.SetSiteMetadata(&siteMeta)
+			assert.Assert(t, ch.SaveRouterConfig(cfg))
+			// simulating new version
+			versionUpdTask.version = test.version
+			t.Logf("New version: %s - current mock site version: %s", test.version, test.curVersion)
+
+			// validating if task has to be executed or not
+			assert.Assert(t, versionUpdTask.AppliesTo(test.curVersion) == test.changed)
+
+			// mocking task execution
+			if test.changed {
+				res := versionUpdTask.Run()
+				assert.Assert(t, len(res.Errors) == 0)
+				assert.Assert(t, res.Changed() == test.changed)
+				assert.Assert(t, 1 == len(res.GetChanges()))
+			}
+		})
+	}
+}
+
+func mockContainers() []*container.Container {
+	imageNoTag := regexp.MustCompile(`(.*):.*`)
+	return []*container.Container{
+		{
+			ID:    strings.Replace(uuid.New().String(), "-", "", -1),
+			Name:  "skupper-router",
+			Image: imageNoTag.ReplaceAllString(images.GetRouterImageName(), "$1:old"),
+			Labels: map[string]string{
+				"application":          "skupper",
+				"skupper.io/component": "skupper-router",
+			},
+			Networks: map[string]container.ContainerNetworkInfo{
+				"skupper": {
+					ID:        "skupper",
+					IPAddress: "172.17.0.10",
+					Gateway:   "172.17.0.1",
+					Aliases:   []string{"skupper-router"},
+					//Aliases:   []string{"skupper", "service-controller"},
+				},
+			},
+			Ports: []container.Port{
+				{Host: "55671", HostIP: "", Target: "55671", Protocol: "tcp"},
+				{Host: "45671", HostIP: "", Target: "45671", Protocol: "tcp"},
+			},
+			Running:   true,
+			CreatedAt: time.Now(),
+			StartedAt: time.Now(),
+		},
+		{
+			ID:    strings.Replace(uuid.New().String(), "-", "", -1),
+			Name:  "skupper-controller-podman",
+			Image: imageNoTag.ReplaceAllString(images.GetControllerPodmanImageName(), "$1:old"),
+			Labels: map[string]string{
+				"application":          "skupper",
+				"skupper.io/component": "skupper-controller-podman",
+			},
+			Networks: map[string]container.ContainerNetworkInfo{
+				"skupper": {
+					ID:        "skupper",
+					IPAddress: "172.17.0.11",
+					Gateway:   "172.17.0.1",
+					Aliases:   []string{"skupper", "service-controller-podman"},
+				},
+			},
+			Running:   true,
+			CreatedAt: time.Now(),
+			StartedAt: time.Now(),
+		}, {
+			ID:    strings.Replace(uuid.New().String(), "-", "", -1),
+			Name:  "flow-collector",
+			Image: imageNoTag.ReplaceAllString(images.GetFlowCollectorImageName(), "$1:old"),
+			Labels: map[string]string{
+				"application":          "skupper",
+				"skupper.io/component": "flow-collector",
+			},
+			Networks: map[string]container.ContainerNetworkInfo{
+				"skupper": {
+					ID:        "skupper",
+					IPAddress: "172.17.0.12",
+					Gateway:   "172.17.0.1",
+					Aliases:   []string{"flow-collector"},
+				},
+			},
+			Ports: []container.Port{
+				{Host: "8010", HostIP: "", Target: "8010", Protocol: "tcp"},
+			},
+			Running:   true,
+			CreatedAt: time.Now(),
+			StartedAt: time.Now(),
+		}, {
+			ID:    strings.Replace(uuid.New().String(), "-", "", -1),
+			Name:  "nginx-service",
+			Image: imageNoTag.ReplaceAllString(images.GetRouterImageName(), "$1:old"),
+			Labels: map[string]string{
+				"application":        "skupper",
+				"skupper.io/address": "nginx",
+			},
+			Networks: map[string]container.ContainerNetworkInfo{
+				"skupper": {
+					ID:        "skupper",
+					IPAddress: "172.17.0.13",
+					Gateway:   "172.17.0.1",
+					Aliases:   []string{"nginx-service"},
+				},
+			},
+			Ports: []container.Port{
+				{Host: "8080", HostIP: "", Target: "8080", Protocol: "tcp"},
+			},
+			Running:   true,
+			CreatedAt: time.Now(),
+			StartedAt: time.Now(),
+		}, {
+			ID:     strings.Replace(uuid.New().String(), "-", "", -1),
+			Name:   "nginx",
+			Image:  "docker.io/nginxinc/nginx-unprivileged:stable-alpine",
+			Labels: map[string]string{},
+			Networks: map[string]container.ContainerNetworkInfo{
+				"skupper": {
+					ID:        "skupper",
+					IPAddress: "172.17.0.14",
+					Gateway:   "172.17.0.1",
+					Aliases:   []string{"nginx"},
+				},
+			},
+			Running:   true,
+			CreatedAt: time.Now(),
+			StartedAt: time.Now(),
+		},
+	}
+}
+
+func mockVolumes() (map[string]*container.Volume, map[string]map[string]string) {
+	var volumes = map[string]*container.Volume{}
+	var volumesFiles = map[string]map[string]string{}
+	addSkupperVolume := func(name string) {
+		volumes[name] = &container.Volume{Name: name, Labels: map[string]string{"application": "skupper"}}
+	}
+	addSkupperVolume("skupper-console-certs")
+	addSkupperVolume("skupper-console-users")
+	addSkupperVolume("skupper-internal")
+	addSkupperVolume("skupper-local-ca")
+	addSkupperVolume("skupper-local-client")
+	addSkupperVolume("skupper-local-server")
+	addSkupperVolume("skupper-router-certs")
+	addSkupperVolume("skupper-service-ca")
+	addSkupperVolume("skupper-service-client")
+	addSkupperVolume("skupper-services")
+	addSkupperVolume("skupper-site-ca")
+	addSkupperVolume("skupper-site-server")
+
+	// volumes content
+	volumesFiles["skupper-internal"] = map[string]string{
+		"skrouterd.json": skrouterdJson,
+	}
+	volumesFiles["skupper-console-users"] = map[string]string{
+		"admin": "admin",
+	}
+	volumesFiles["skupper-services"] = map[string]string{
+		"skupper-services.json": skupperServicesJson,
+	}
+
+	// defining skupper-services configmap
+	return volumes, volumesFiles
+}
diff --git a/pkg/domain/update_test.go b/pkg/domain/update_test.go
new file mode 100644
index 00000000..0a7f5dbf
--- /dev/null
+++ b/pkg/domain/update_test.go
@@ -0,0 +1,452 @@
+package domain
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/skupperproject/skupper/pkg/utils"
+	"gotest.tools/assert"
+)
+
+func TestUpdateProcessor(t *testing.T) {
+	tests := []struct {
+		name            string
+		curVersion      string
+		newVersion      string
+		tasks           []UpdateTask
+		expectedChanges []string
+		expectError     bool
+	}{
+		{
+			name:       "all-tasks-and-post-tasks",
+			curVersion: "1.4.2",
+			newVersion: "1.5.0",
+			tasks: []UpdateTask{
+				NewMockUpdateTask(
+					"create new container",
+					"1.5.0",
+					func(siteVersion string) bool {
+						return utils.LessRecentThanVersion(siteVersion, "1.5.0")
+					},
+					PriorityNormal,
+					&UpdateResult{
+						changes: []string{
+							"created new container",
+						},
+						PostTasks: []UpdateTask{
+							NewMockUpdateTask(
+								"start new container",
+								"1.5.0",
+								func(siteVersion string) bool {
+									return true
+								},
+								PriorityLast,
+								&UpdateResult{
+									changes: []string{
+										"started new container",
+									},
+								},
+							),
+							NewMockUpdateTask(
+								"create config file",
+								"1.5.0",
+								func(siteVersion string) bool {
+									return true
+								},
+								PriorityFirst,
+								&UpdateResult{
+									changes: []string{
+										"created config file",
+									},
+								},
+							),
+							NewMockUpdateTask(
+								"modify config file",
+								"1.5.0",
+								func(siteVersion string) bool {
+									return true
+								},
+								PriorityFirst,
+								&UpdateResult{
+									changes: []string{
+										"modified config file",
+									},
+								},
+							),
+						},
+					},
+				),
+				NewMockUpdateTask(
+					"create new volume",
+					"1.5.0",
+					func(siteVersion string) bool {
+						return utils.LessRecentThanVersion(siteVersion, "1.5.0")
+					},
+					PriorityFirst,
+					&UpdateResult{
+						changes: []string{
+							"created new volume",
+						},
+						PostTasks: []UpdateTask{},
+					},
+				),
+				NewMockUpdateTask(
+					"this task should not be executed",
+					"1.4.0",
+					func(siteVersion string) bool {
+						return utils.LessRecentThanVersion(siteVersion, "1.4.0")
+					},
+					PriorityFirst,
+					&UpdateResult{
+						changes: []string{
+							"this task was incorrectly executed",
+						},
+						PostTasks: []UpdateTask{
+							NewMockUpdateTask(
+								"this post task should not be executed",
+								"1.4.0",
+								func(siteVersion string) bool {
+									return false
+								},
+								PriorityFirst,
+								&UpdateResult{
+									changes: []string{
+										"this post task was incorrectly executed",
+									},
+								},
+							),
+						},
+					},
+				),
+			},
+			expectedChanges: []string{
+				"created new volume",
+				"created new container",
+				"created config file",
+				"modified config file",
+				"started new container",
+			},
+			expectError: false,
+		},
+		{
+			name:       "second-task-with-warning",
+			curVersion: "1.4.2",
+			newVersion: "1.5.0",
+			tasks: []UpdateTask{
+				NewMockUpdateTask(
+					"create new container",
+					"1.5.0",
+					func(siteVersion string) bool {
+						return utils.LessRecentThanVersion(siteVersion, "1.5.0")
+					},
+					PriorityNormal,
+					&UpdateResult{
+						warnings: []string{
+							"container already exists, ignoring",
+						},
+						changes: []string{
+							"created new container",
+						},
+						PostTasks: []UpdateTask{
+							NewMockUpdateTask(
+								"start new container",
+								"1.5.0",
+								func(siteVersion string) bool {
+									return true
+								},
+								PriorityLast,
+								&UpdateResult{
+									changes: []string{
+										"started new container",
+									},
+								},
+							),
+						},
+					},
+				),
+				NewMockUpdateTask(
+					"create new volume",
+					"1.5.0",
+					func(siteVersion string) bool {
+						return utils.LessRecentThanVersion(siteVersion, "1.5.0")
+					},
+					PriorityFirst,
+					&UpdateResult{
+						changes: []string{
+							"created new volume",
+						},
+						PostTasks: []UpdateTask{},
+					},
+				),
+			},
+			expectedChanges: []string{
+				"created new volume",
+				"created new container",
+				"started new container",
+			},
+			expectError: false,
+		},
+		{
+			name:       "second-task-with-error",
+			curVersion: "1.4.2",
+			newVersion: "1.5.0",
+			tasks: []UpdateTask{
+				NewMockUpdateTask(
+					"create new container",
+					"1.5.0",
+					func(siteVersion string) bool {
+						return utils.LessRecentThanVersion(siteVersion, "1.5.0")
+					},
+					PriorityNormal,
+					&UpdateResult{
+						Errors: []error{
+							fmt.Errorf("container already exists"),
+						},
+						PostTasks: []UpdateTask{
+							NewMockUpdateTask(
+								"start new container",
+								"1.5.0",
+								func(siteVersion string) bool {
+									return true
+								},
+								PriorityLast,
+								&UpdateResult{
+									changes: []string{
+										"should not be executed - started new container",
+									},
+								},
+							),
+						},
+					},
+				),
+				NewMockUpdateTask(
+					"create new volume",
+					"1.5.0",
+					func(siteVersion string) bool {
+						return utils.LessRecentThanVersion(siteVersion, "1.5.0")
+					},
+					PriorityFirst,
+					&UpdateResult{
+						changes: []string{
+							"created new volume",
+						},
+						PostTasks: []UpdateTask{},
+					},
+				),
+			},
+			expectedChanges: []string{
+				"created new volume",
+			},
+			expectError: true,
+		},
+		{
+			name:       "second-post-task-with-error",
+			curVersion: "1.4.2",
+			newVersion: "1.5.0",
+			tasks: []UpdateTask{
+				NewMockUpdateTask(
+					"create new container",
+					"1.5.0",
+					func(siteVersion string) bool {
+						return utils.LessRecentThanVersion(siteVersion, "1.5.0")
+					},
+					PriorityNormal,
+					&UpdateResult{
+						changes: []string{
+							"created new container",
+						},
+						PostTasks: []UpdateTask{
+							NewMockUpdateTask(
+								"start new container",
+								"1.5.0",
+								func(siteVersion string) bool {
+									return true
+								},
+								PriorityLast,
+								&UpdateResult{
+									changes: []string{
+										"started new container",
+									},
+								},
+							),
+							NewMockUpdateTask(
+								"create config file",
+								"1.5.0",
+								func(siteVersion string) bool {
+									return true
+								},
+								PriorityFirst,
+								&UpdateResult{
+									changes: []string{
+										"created config file",
+									},
+								},
+							),
+							NewMockUpdateTask(
+								"modify config file",
+								"1.5.0",
+								func(siteVersion string) bool {
+									return true
+								},
+								PriorityFirst,
+								&UpdateResult{
+									Errors: []error{
+										fmt.Errorf("error modifying file"),
+									},
+								},
+							),
+						},
+					},
+				),
+				NewMockUpdateTask(
+					"create new volume",
+					"1.5.0",
+					func(siteVersion string) bool {
+						return utils.LessRecentThanVersion(siteVersion, "1.5.0")
+					},
+					PriorityFirst,
+					&UpdateResult{
+						changes: []string{
+							"created new volume",
+						},
+						PostTasks: []UpdateTask{},
+					},
+				),
+				NewMockUpdateTask(
+					"this task should not be executed",
+					"1.4.0",
+					func(siteVersion string) bool {
+						return utils.LessRecentThanVersion(siteVersion, "1.4.0")
+					},
+					PriorityFirst,
+					&UpdateResult{
+						changes: []string{
+							"this task was incorrectly executed",
+						},
+						PostTasks: []UpdateTask{
+							NewMockUpdateTask(
+								"this post task should not be executed",
+								"1.4.0",
+								func(siteVersion string) bool {
+									return false
+								},
+								PriorityFirst,
+								&UpdateResult{
+									changes: []string{
+										"this post task was incorrectly executed",
+									},
+								},
+							),
+						},
+					},
+				),
+			},
+			expectedChanges: []string{
+				"created new volume",
+				"created new container",
+				"created config file",
+			},
+			expectError: true,
+		},
+		{
+			name:       "no-updated-needed",
+			curVersion: "1.5.0",
+			newVersion: "1.5.0",
+			tasks: []UpdateTask{
+				NewMockUpdateTask(
+					"create new container",
+					"1.5.0",
+					func(siteVersion string) bool {
+						return utils.LessRecentThanVersion(siteVersion, "1.5.0")
+					},
+					PriorityNormal,
+					&UpdateResult{
+						changes: []string{
+							"created new container",
+						},
+						PostTasks: []UpdateTask{
+							NewMockUpdateTask(
+								"start new container",
+								"1.5.0",
+								func(siteVersion string) bool {
+									return true
+								},
+								PriorityLast,
+								&UpdateResult{
+									changes: []string{
+										"started new container",
+									},
+								},
+							),
+						},
+					},
+				),
+				NewMockUpdateTask(
+					"create new volume",
+					"1.5.0",
+					func(siteVersion string) bool {
+						return utils.LessRecentThanVersion(siteVersion, "1.5.0")
+					},
+					PriorityFirst,
+					&UpdateResult{
+						changes: []string{
+							"created new volume",
+						},
+						PostTasks: []UpdateTask{},
+					},
+				),
+			},
+			expectError: false,
+		},
+	}
+
+	for _, scenario := range tests {
+		t.Run(scenario.name, func(t *testing.T) {
+			up := &UpdateProcessor{
+				Verbose: true,
+			}
+			up.RegisterTasks(scenario.tasks...)
+			err := up.Process(scenario.curVersion)
+			assert.Equal(t, err != nil, scenario.expectError, fmt.Sprintf("error expected == %v but got err == %v", scenario.expectError, err))
+			assert.DeepEqual(t, scenario.expectedChanges, up.changes)
+		})
+	}
+}
+
+func NewMockUpdateTask(info, version string, appliesTo func(siteVersion string) bool, priority UpdatePriority, result *UpdateResult) UpdateTask {
+	return &mockUpdateTask{
+		info:      info,
+		version:   version,
+		appliesTo: appliesTo,
+		priority:  priority,
+		result:    result,
+	}
+}
+
+type mockUpdateTask struct {
+	info      string
+	version   string
+	appliesTo func(siteVersion string) bool
+	priority  UpdatePriority
+	result    *UpdateResult
+}
+
+func (u *mockUpdateTask) Info() string {
+	return u.info
+}
+
+func (u *mockUpdateTask) AppliesTo(siteVersion string) bool {
+	return u.appliesTo(siteVersion)
+}
+
+func (u *mockUpdateTask) Version() string {
+	return u.version
+}
+
+func (u *mockUpdateTask) Priority() UpdatePriority {
+	return u.priority
+}
+
+func (u *mockUpdateTask) Run() *UpdateResult {
+	return u.result
+}
diff --git a/pkg/flow/collector_test.go b/pkg/flow/collector_test.go
new file mode 100644
index 00000000..3a7ce08a
--- /dev/null
+++ b/pkg/flow/collector_test.go
@@ -0,0 +1,112 @@
+package flow
+
+import (
+	"reflect"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"gotest.tools/assert"
+)
+
+func TestUpdates(t *testing.T) {
+	scenarios := []struct {
+		name       string
+		subject    string
+		to         string
+		properties map[string]string
+		stype      reflect.Type
+		fields     map[int]interface{}
+	}{
+		{
+			name:    "router-beacon",
+			subject: "BEACON",
+			to:      "mc/sfe.all",
+			stype:   reflect.TypeOf(BeaconRecord{}),
+			properties: map[string]string{
+				"Version":    "1",
+				"SourceType": "ROUTER",
+				"Address":    "sfe.gnjdr:0",
+			},
+		},
+		{
+			name:    "controller-beacon",
+			subject: "BEACON",
+			to:      "mc/sfe.all",
+			stype:   reflect.TypeOf(BeaconRecord{}),
+			properties: map[string]string{
+				"Version":    "1",
+				"SourceType": "CONTROLLER",
+				"Address":    "sfe.gnjdr:0",
+			},
+		},
+	}
+	for _, s := range scenarios {
+		switch s.stype {
+		case reflect.TypeOf(BeaconRecord{}):
+			scenarioBeacon := &BeaconRecord{}
+			version, _ := strconv.Atoi(s.properties["Version"])
+			scenarioBeacon.Version = uint32(version)
+			scenarioBeacon.SourceType = s.properties["SourceType"]
+			scenarioBeacon.Address = s.properties["Address"]
+
+			msg, err := encodeBeacon(scenarioBeacon)
+			assert.Assert(t, err)
+			assert.Equal(t, msg.Properties.Subject, s.subject)
+		}
+	}
+}
+
+func TestRecordUpdates(t *testing.T) {
+	reg := prometheus.NewRegistry()
+	u, _ := time.ParseDuration("5m")
+	fc := NewFlowCollector(FlowCollectorSpec{
+		Mode:              RecordMetrics,
+		Origin:            "origin",
+		PromReg:           reg,
+		ConnectionFactory: nil,
+		FlowRecordTtl:     u,
+	})
+	fc.metrics = fc.NewMetrics(fc.prometheusReg)
+
+	assert.Assert(t, fc != nil)
+	stopCh := make(chan struct{})
+	go fc.beaconUpdates(stopCh)
+	go fc.recordUpdates(stopCh)
+
+	beacon := BeaconRecord{
+		Version:    1,
+		SourceType: "CONTROLLER",
+		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)
+
+	heartbeat := HeartbeatRecord{
+		Version:  1,
+		Identity: "1234",
+		Source:   "sfe.1234",
+	}
+
+	var heartbeats []interface{}
+	heartbeats = append(heartbeats, heartbeat)
+	fc.heartbeatsIncoming <- heartbeats
+	time.Sleep(1 * time.Second)
+	assert.Equal(t, eventSource.Heartbeats, 1)
+
+	time.Sleep(1 * time.Second)
+
+	close(stopCh)
+}
diff --git a/pkg/flow/controller_test.go b/pkg/flow/controller_test.go
new file mode 100644
index 00000000..f0badd1e
--- /dev/null
+++ b/pkg/flow/controller_test.go
@@ -0,0 +1,149 @@
+package flow
+
+import (
+	"os"
+	"testing"
+	"time"
+
+	"gotest.tools/assert"
+)
+
+func TestUpdateProcess(t *testing.T) {
+	fc := NewFlowController("mysite", "X.Y.Z", uint64(time.Now().UnixNano())/uint64(time.Microsecond), nil)
+	assert.Assert(t, fc != nil)
+	procName := "tcp-go-echo"
+	sourceIP := "10.0.0.1"
+	process := &ProcessRecord{
+		Base: Base{
+			Identity:  "myprocess",
+			Parent:    "mysite",
+			StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+		},
+		Name:       &procName,
+		SourceHost: &sourceIP,
+	}
+	err := UpdateProcess(fc, false, "tcp-go-echo", process)
+	assert.Assert(t, err)
+	assert.Equal(t, len(fc.processRecords), 1)
+	processUpdate := <-fc.processOutgoing
+	assert.Assert(t, processUpdate != nil)
+	assert.Equal(t, *processUpdate.SourceHost, "10.0.0.1")
+	err = UpdateProcess(fc, true, "ns/tcp-go-echo", process)
+	assert.Assert(t, err)
+	assert.Equal(t, len(fc.processRecords), 0)
+	processUpdate = <-fc.processOutgoing
+	assert.Assert(t, processUpdate != nil)
+	assert.Assert(t, processUpdate.EndTime != 0)
+}
+
+func TestUpdateHost(t *testing.T) {
+	fc := NewFlowController("mysite", "X.Y.Z", uint64(time.Now().UnixNano())/uint64(time.Microsecond), nil)
+	assert.Assert(t, fc != nil)
+	hostName := "bastion-server"
+	host := &HostRecord{
+		Base: Base{
+			Identity:  "myhost",
+			Parent:    "mysite",
+			StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+		},
+		Name: &hostName,
+	}
+	err := UpdateHost(fc, false, "bastion-server", host)
+	assert.Assert(t, err)
+	assert.Equal(t, len(fc.hostRecords), 1)
+	hostUpdate := <-fc.hostOutgoing
+	assert.Assert(t, hostUpdate != nil)
+	assert.Equal(t, hostUpdate.Parent, "mysite")
+	err = UpdateHost(fc, true, "bastion-server", host)
+	assert.Assert(t, err)
+	assert.Equal(t, len(fc.hostRecords), 0)
+}
+
+func TestUpdateBeaconAndHeartbeats(t *testing.T) {
+	_ = os.Setenv("SKUPPER_SITE_ID", "mySite")
+	fc := NewFlowController("mySite", "X.Y.Z", uint64(time.Now().UnixNano())/uint64(time.Microsecond), nil)
+	assert.Assert(t, fc != nil)
+	stopCh := make(chan struct{})
+	go fc.updateBeaconAndHeartbeats(stopCh)
+	beaconUpdate := <-fc.beaconOutgoing
+	assert.Assert(t, beaconUpdate != nil)
+	beacon, ok := beaconUpdate.(*BeaconRecord)
+	assert.Assert(t, ok)
+	assert.Equal(t, beacon.Version, uint32(1))
+	assert.Equal(t, beacon.SourceType, "CONTROLLER")
+	heartbeatUpdate := <-fc.heartbeatOutgoing
+	assert.Assert(t, heartbeatUpdate != nil)
+	heartbeat, ok := heartbeatUpdate.(*HeartbeatRecord)
+	assert.Assert(t, ok)
+	assert.Equal(t, heartbeat.Source, "sfe.mySite")
+	time.Sleep(1 * time.Second)
+	beaconUpdate = <-fc.beaconOutgoing
+	assert.Assert(t, beaconUpdate != nil)
+	heartbeatUpdate = <-fc.heartbeatOutgoing
+	assert.Assert(t, heartbeatUpdate != nil)
+	heartbeat, ok = heartbeatUpdate.(*HeartbeatRecord)
+	assert.Assert(t, ok)
+	close(stopCh)
+}
+
+func TestUpdateRecords(t *testing.T) {
+	_ = os.Setenv("SKUPPER_SITE_ID", "mySite")
+	_ = os.Setenv("SKUPPER_NAMESPACE", "myNamespace")
+	fc := NewFlowController("mySite", "X.Y.Z", uint64(time.Now().UnixNano())/uint64(time.Microsecond), nil)
+	assert.Assert(t, fc != nil)
+	stopCh := make(chan struct{})
+	go fc.updateRecords(stopCh)
+
+	recordUpdate := <-fc.recordOutgoing
+	assert.Assert(t, recordUpdate != nil)
+	siteUpdate, ok := recordUpdate.(*SiteRecord)
+	assert.Assert(t, ok)
+	assert.Equal(t, siteUpdate.Identity, "mySite")
+
+	procName := "tcp-go-echo"
+	sourceIP := "10.0.0.1"
+	process := &ProcessRecord{
+		Base: Base{
+			Identity:  "myprocess",
+			Parent:    "mysite",
+			StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+		},
+		Name:       &procName,
+		SourceHost: &sourceIP,
+	}
+	fc.processOutgoing <- process
+
+	recordUpdate = <-fc.recordOutgoing
+	assert.Assert(t, recordUpdate != nil)
+	processUpdate, ok := recordUpdate.(*ProcessRecord)
+	assert.Assert(t, ok)
+	assert.Equal(t, *processUpdate.Name, "tcp-go-echo")
+
+	hostName := "bastion-server"
+	host := &HostRecord{
+		Base: Base{
+			Identity:  "myhost",
+			Parent:    "mysite",
+			StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+		},
+		Name: &hostName,
+	}
+	fc.hostOutgoing <- host
+
+	recordUpdate = <-fc.recordOutgoing
+	assert.Assert(t, recordUpdate != nil)
+	hostUpdate, ok := recordUpdate.(*HostRecord)
+	assert.Assert(t, ok)
+	assert.Equal(t, *hostUpdate.Name, hostName)
+
+	flush := FlushRecord{
+		Address: "sfe.1234",
+	}
+	var flushes []interface{}
+	flushes = append(flushes, flush)
+	fc.flushIncoming <- flushes
+
+	time.Sleep(5 * time.Second)
+	close(stopCh)
+
+}
diff --git a/pkg/flow/encoding_test.go b/pkg/flow/encoding_test.go
index 9b04ca57..548f4197 100644
--- a/pkg/flow/encoding_test.go
+++ b/pkg/flow/encoding_test.go
@@ -24,6 +24,9 @@ func TestRecordDecoding(t *testing.T) {
 				Identity:     uuid.New().String(),
 				Name:         "skupper-site",
 				Namespace:    "van-namespace",
+				Platform:     "kubernetes",
+				Version:      "1.0.0",
+				Policy:       "Disabled",
 			},
 		},
 		{
@@ -70,7 +73,19 @@ func TestRecordDecoding(t *testing.T) {
 				DestHost:     "172.21.122.169",
 				DestPort:     "9090",
 				Protocol:     "tcp",
-				VanAddress:   " tcp-go-echo",
+				VanAddress:   "tcp-go-echo",
+				Target:       "some-container-name",
+			},
+		},
+		{
+			name:  "logevent",
+			stype: reflect.TypeOf(LogEventRecord{}),
+			fields: map[int]interface{}{
+				TypeOfRecord: uint32(LogEvent),
+				LogSeverity:  uint64(99),
+				LogText:      "Something to know",
+				SourceFile:   "client.c",
+				SourceLine:   uint64(1010),
 			},
 		},
 		{
@@ -137,6 +152,9 @@ func TestRecordDecoding(t *testing.T) {
 				assert.Equal(t, site.Identity, s.fields[Identity])
 				assert.Equal(t, *site.Name, s.fields[Name])
 				assert.Equal(t, *site.NameSpace, s.fields[Namespace])
+				assert.Equal(t, *site.Platform, s.fields[Platform])
+				assert.Equal(t, *site.Version, s.fields[Version])
+				assert.Equal(t, *site.Policy, s.fields[Policy])
 			case RouterRecord:
 				router, ok := record.(RouterRecord)
 				assert.Assert(t, ok)
@@ -168,6 +186,14 @@ func TestRecordDecoding(t *testing.T) {
 				assert.Equal(t, *connector.DestPort, s.fields[DestPort])
 				assert.Equal(t, *connector.Protocol, s.fields[Protocol])
 				assert.Equal(t, *connector.Address, s.fields[VanAddress])
+				assert.Equal(t, *connector.Target, s.fields[Target])
+			case LogEventRecord:
+				logevent, ok := record.(LogEventRecord)
+				assert.Assert(t, ok)
+				assert.Equal(t, *logevent.LogSeverity, s.fields[LogSeverity])
+				assert.Equal(t, *logevent.LogText, s.fields[LogText])
+				assert.Equal(t, *logevent.SourceFile, s.fields[SourceFile])
+				assert.Equal(t, *logevent.SourceLine, s.fields[SourceLine])
 			case FlowRecord:
 				flow, ok := record.(FlowRecord)
 				assert.Assert(t, ok)
diff --git a/pkg/flow/flow_mem_driver_test.go b/pkg/flow/flow_mem_driver_test.go
index 1a1381e1..a6eb2f6c 100644
--- a/pkg/flow/flow_mem_driver_test.go
+++ b/pkg/flow/flow_mem_driver_test.go
@@ -11,7 +11,7 @@ import (
 	"gotest.tools/assert"
 )
 
-func TestRecordGraph(t *testing.T) {
+func TestRecordGraphWithMetrics(t *testing.T) {
 	name := "skupper-site"
 	namespace := "skupper-public"
 	provider := "aws"
@@ -285,7 +285,13 @@ func TestRecordGraph(t *testing.T) {
 
 	reg := prometheus.NewRegistry()
 	u, _ := time.ParseDuration("5m")
-	fc := NewFlowCollector("origin", reg, nil, u)
+	fc := NewFlowCollector(FlowCollectorSpec{
+		Mode:              RecordMetrics,
+		Origin:            "origin",
+		PromReg:           reg,
+		ConnectionFactory: nil,
+		FlowRecordTtl:     u,
+	})
 	fc.metrics = fc.NewMetrics(fc.prometheusReg)
 	for _, s := range sites {
 		err := fc.updateRecord(s)
@@ -374,16 +380,9 @@ func TestRecordGraph(t *testing.T) {
 			assert.Equal(t, *trace, "skupper-site@skupper-site")
 		}
 	}
-	//	for _, f := range fc.Flows {
-	//		if f.CounterFlow != nil {
-	//			// note until we call reconcile
-	//			fp, ok := fc.linkFlowPair(f)
-	//			assert.Assert(t, ok)
-	//			fc.FlowPairs[fp.Identity] = fp
-	//		}
-	//	}
 
-	fc.reconcileRecords()
+	fc.reconcileConnectorRecords()
+	fc.reconcileFlowRecords()
 
 	type test struct {
 		recordType   int
@@ -765,6 +764,15 @@ func TestRecordGraph(t *testing.T) {
 			name:         "item",
 			responseSize: 1,
 		},
+		{
+			recordType:   Collector,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{},
+			name:         "list",
+			responseSize: 1,
+		},
 	}
 
 	var payload Payload
@@ -782,4 +790,319 @@ func TestRecordGraph(t *testing.T) {
 		assert.Assert(t, err)
 		assert.Equal(t, test.responseSize, payload.Count)
 	}
+
+	endTime := uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
+	for _, c := range connectors {
+		c.EndTime = endTime
+		err := fc.updateRecord(c)
+		assert.Assert(t, err)
+	}
+	for _, l := range listeners {
+		l.EndTime = endTime
+		err := fc.updateRecord(l)
+		assert.Assert(t, err)
+	}
+	for _, p := range processes {
+		p.EndTime = endTime
+		err := fc.updateRecord(p)
+		assert.Assert(t, err)
+	}
+	for _, l := range links {
+		l.EndTime = endTime
+		err := fc.updateRecord(l)
+		assert.Assert(t, err)
+	}
+	for _, h := range hosts {
+		h.EndTime = endTime
+		err := fc.updateRecord(h)
+		assert.Assert(t, err)
+	}
+	for _, r := range routers {
+		r.EndTime = endTime
+		err := fc.updateRecord(r)
+		assert.Assert(t, err)
+	}
+	for _, s := range sites {
+		s.EndTime = endTime
+		err := fc.updateRecord(s)
+		assert.Assert(t, err)
+	}
+}
+
+// TODO: address dependence on cli
+func xTestRecordGraphNetworkStatus(t *testing.T) {
+	name := "skupper-site"
+	namespace := "skupper-public"
+	provider := "aws"
+	address := "tcp-go-echo"
+	address2 := "redis-cart"
+	address3 := "mongo"
+	address4 := "cartservice"
+	protocol := "tcp"
+	processName1 := "checkout-1234"
+	processName2 := "payment-1234"
+	groupName1 := "online-store"
+	connectorProcess1 := "process:1"
+	connectorProcess2 := "process:2"
+	processConnector := "connector:0"
+
+	sites := []SiteRecord{
+		{
+			Base: Base{
+				RecType:   recordNames[Site],
+				Identity:  "site:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			Name:      &name,
+			NameSpace: &namespace,
+			Provider:  &provider,
+		},
+	}
+	routers := []RouterRecord{
+		{
+			Base: Base{
+				RecType:   recordNames[Router],
+				Identity:  "router:0",
+				Parent:    "site:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			Name: &name,
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Router],
+				Identity:  "router:1",
+				Parent:    "site:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			Name: &name,
+		},
+	}
+	hosts := []HostRecord{
+		{
+			Base: Base{
+				RecType:   recordNames[Host],
+				Identity:  "host:0",
+				Parent:    "site:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Host],
+				Identity:  "host:1",
+				Parent:    "site:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Host],
+				Identity:  "host:2",
+				Parent:    "site:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+		},
+	}
+	links := []LinkRecord{
+		{
+			Base: Base{
+				RecType:   recordNames[Link],
+				Identity:  "link:0",
+				Parent:    "router:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Link],
+				Identity:  "link:1",
+				Parent:    "router:1",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+		},
+	}
+	processes := []ProcessRecord{
+		{
+			Base: Base{
+				RecType:   recordNames[Process],
+				Identity:  "process:0",
+				Parent:    "site:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			Name:      &processName1,
+			GroupName: &groupName1,
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Process],
+				Identity:  "process:1",
+				Parent:    "site:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			Name:      &processName2,
+			GroupName: &groupName1,
+			connector: &processConnector,
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Process],
+				Identity:  "process:2",
+				Parent:    "site:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			Name:      &processName1,
+			GroupName: &groupName1,
+		},
+	}
+	listeners := []ListenerRecord{
+		{
+			Base: Base{
+				RecType:   recordNames[Listener],
+				Identity:  "listener:0",
+				Parent:    "router:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			Address:  &address,
+			Protocol: &protocol,
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Listener],
+				Identity:  "listener:1",
+				Parent:    "router:1",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			Protocol: &protocol,
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Listener],
+				Identity:  "listener:2",
+				Parent:    "router:1",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			Protocol: &protocol,
+			Address:  &address2,
+		},
+	}
+	connectors := []ConnectorRecord{
+		{
+			Base: Base{
+				RecType:   recordNames[Connector],
+				Identity:  "connector:0",
+				Parent:    "router:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			Address:   &address3,
+			Protocol:  &protocol,
+			ProcessId: &connectorProcess1,
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Connector],
+				Identity:  "connector:1",
+				Parent:    "router:1",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			Protocol: &protocol,
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Connector],
+				Identity:  "connector:1",
+				Parent:    "router:1",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			Protocol:  &protocol,
+			Address:   &address4,
+			ProcessId: &connectorProcess2,
+		},
+	}
+
+	u, _ := time.ParseDuration("5m")
+	fc := NewFlowCollector(FlowCollectorSpec{
+		Mode:              RecordStatus,
+		Namespace:         "default",
+		Origin:            "origin",
+		PromReg:           nil,
+		ConnectionFactory: nil,
+		FlowRecordTtl:     u,
+	})
+	for _, s := range sites {
+		err := fc.updateRecord(s)
+		assert.Assert(t, err)
+		err = fc.updateRecord(s)
+		assert.Assert(t, err)
+	}
+	for _, r := range routers {
+		err := fc.updateRecord(r)
+		assert.Assert(t, err)
+		err = fc.updateRecord(r)
+		assert.Assert(t, err)
+	}
+	for _, h := range hosts {
+		err := fc.updateRecord(h)
+		assert.Assert(t, err)
+		err = fc.updateRecord(h)
+		assert.Assert(t, err)
+	}
+	for _, l := range links {
+		err := fc.updateRecord(l)
+		assert.Assert(t, err)
+		err = fc.updateRecord(l)
+		assert.Assert(t, err)
+	}
+	for _, p := range processes {
+		err := fc.updateRecord(p)
+		assert.Assert(t, err)
+		err = fc.updateRecord(p)
+		assert.Assert(t, err)
+	}
+	for _, l := range listeners {
+		err := fc.updateRecord(l)
+		assert.Assert(t, err)
+		err = fc.updateRecord(l)
+		assert.Assert(t, err)
+	}
+	for _, c := range connectors {
+		err := fc.updateRecord(c)
+		assert.Assert(t, err)
+		err = fc.updateRecord(c)
+		assert.Assert(t, err)
+	}
+
+	for _, s := range sites {
+		id := fc.getRecordSiteId(s)
+		assert.Equal(t, id, s.Identity)
+	}
+	for _, r := range routers {
+		id := fc.getRecordSiteId(r)
+		assert.Equal(t, id, r.Parent)
+	}
+	for _, h := range hosts {
+		id := fc.getRecordSiteId(h)
+		assert.Equal(t, id, h.Parent)
+	}
+	for _, l := range links {
+		id := fc.getRecordSiteId(l)
+		assert.Equal(t, id, "site:0")
+	}
+	for _, p := range processes {
+		id := fc.getRecordSiteId(p)
+		assert.Equal(t, id, p.Parent)
+	}
+	for _, l := range listeners {
+		id := fc.getRecordSiteId(l)
+		assert.Equal(t, id, "site:0")
+	}
+	for _, l := range connectors {
+		id := fc.getRecordSiteId(l)
+		assert.Equal(t, id, "site:0")
+	}
+
+	fc.reconcileConnectorRecords()
+
+	err := fc.updateNetworkStatus()
+	assert.Assert(t, err)
 }
diff --git a/test/utils/skupper/cli/init.go b/test/utils/skupper/cli/init.go
index 577b7e78..f67fbc82 100644
--- a/test/utils/skupper/cli/init.go
+++ b/test/utils/skupper/cli/init.go
@@ -7,6 +7,7 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/skupperproject/skupper/api/types"
 	"github.com/skupperproject/skupper/client/container"
@@ -19,7 +20,7 @@ import (
 	"github.com/skupperproject/skupper/test/utils/base"
 	"github.com/skupperproject/skupper/test/utils/constants"
 	"github.com/skupperproject/skupper/test/utils/skupper/console"
-	v12 "k8s.io/api/core/v1"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -359,9 +360,9 @@ func (s *InitTester) validateIngressFor(cluster *base.ClusterContext, ingress st
 		return fmt.Errorf("could not find service: %s - %v", service, err)
 	} else if err == nil {
 		// validating service type
-		var expectedType v12.ServiceType = v12.ServiceTypeClusterIP
+		var expectedType corev1.ServiceType = corev1.ServiceTypeClusterIP
 		if ingress == "loadbalancer" {
-			expectedType = v12.ServiceTypeLoadBalancer
+			expectedType = corev1.ServiceTypeLoadBalancer
 		}
 		if svc.Spec.Type != expectedType {
 			return fmt.Errorf("invalid service type for service: %s - expected: %s - found: %s",
@@ -544,37 +545,48 @@ func (s *InitTester) validateRouterCPUMemory(cluster *base.ClusterContext) error
 
 	// retrieving the router pods
 	routerSelector := fmt.Sprintf("%s=%s", types.ComponentAnnotation, types.TransportComponentName)
-	pods, err := kube.GetPods(routerSelector, cluster.Namespace, cluster.VanClient.KubeClient)
-	if err != nil {
-		return err
-	}
 
-	// looping through pods
-	for _, pod := range pods {
-		for _, container := range pod.Spec.Containers {
-			if container.Name == "router" {
-				if !expectResources {
-					// If not expecting requests, but something is defined throw an error
-					if len(container.Resources.Requests) > 0 {
-						return fmt.Errorf("resources not requested but defined in the router (pod: %s, container: %s) - requests: %v",
-							pod.Name, container.Name, container.Resources.Requests)
-					}
-				} else {
-					// If requests have been specified, assert the correct values have been defined
-					cpu := container.Resources.Requests.Cpu().String()
-					mem := container.Resources.Requests.Memory().String()
+	ctx, cn := context.WithTimeout(context.Background(), time.Minute*5)
+	defer cn()
+	err := utils.RetryWithContext(ctx, time.Second, func() (bool, error) {
+		pods, err := kube.WaitForPodsSelectorStatus(cluster.Namespace, cluster.VanClient.KubeClient, routerSelector, corev1.PodRunning, time.Second*180, time.Second*5)
+		if err != nil {
+			return false, nil
+		}
 
-					if s.RouterCPU != cpu {
-						return fmt.Errorf("--router-cpu defined as: [%s] but container has: [%s]", s.RouterCPU, cpu)
-					}
-					if s.RouterMemory != mem {
-						return fmt.Errorf("--router-memory defined as: [%s] but container has: [%s]", s.RouterMemory, mem)
+		// looping through pods
+		for _, pod := range pods {
+			for _, cs := range pod.Status.ContainerStatuses {
+				if !cs.Ready {
+					return false, nil
+				}
+			}
+			for _, container := range pod.Spec.Containers {
+				if container.Name == "router" {
+					if !expectResources {
+						// If not expecting requests, but something is defined throw an error
+						if len(container.Resources.Requests) > 0 {
+							return true, fmt.Errorf("resources not requested but defined in the router (pod: %s, container: %s) - requests: %v",
+								pod.Name, container.Name, container.Resources.Requests)
+						}
+					} else {
+						// If requests have been specified, assert the correct values have been defined
+						cpu := container.Resources.Requests.Cpu().String()
+						mem := container.Resources.Requests.Memory().String()
+
+						if s.RouterCPU != cpu {
+							return true, fmt.Errorf("--router-cpu defined as: [%s] but container has: [%s] - pod: %s.%s", s.RouterCPU, cpu, pod.Namespace, pod.Name)
+						}
+						if s.RouterMemory != mem {
+							return true, fmt.Errorf("--router-memory defined as: [%s] but container has: [%s] - pod: %s.%s", s.RouterMemory, mem, pod.Namespace, pod.Name)
+						}
 					}
 				}
 			}
 		}
-	}
-	return nil
+		return true, nil
+	})
+	return err
 }
 
 func (s *InitTester) validateControllerCPUMemory(cluster *base.ClusterContext) error {