Repository: skupperproject/skupper
Pull request #1196 (2023-08-07 22:03:06)

dp collector lite
collector for site, router records only for dp status

fgiorgetti: Other than this minor comment, it looks good to me.
The leader election seems to be working fine, the new configmap also looks right to me.
Tested on latest minikube and minishift (3.11).

Patch:
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/types.go b/api/types/types.go
index 6a28f4c4..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"},
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/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/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/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


Test patch:
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/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 {