Repository: skupperproject/skupper
Pull request #1076 (2023-03-29 19:03:52)

add prometheus metrics

chassing: Closes #951
ajssmith: @grs presently, all other metrics are derived from traffic (e.g. flow) records. Are you generating any adaptor traffic?
grs: > @grs presently, all other metrics are derived from traffic (e.g. flow) records. Are you generating any adaptor traffic?

Yes, I've made some requests on a skupper service.

Patch:
diff --git a/api/types/client.go b/api/types/client.go
index ac6e26e3..18c03838 100644
--- a/api/types/client.go
+++ b/api/types/client.go
@@ -84,6 +84,14 @@ type FlowCollectorOptions struct {
 	FlowRecordTtl time.Duration
 }
 
+type PrometheusServerOptions struct {
+	Tuning
+	ExternalServer string
+	AuthMode       string
+	User           string
+	Password       string
+}
+
 type SiteConfigSpec struct {
 	SkupperName              string
 	SkupperNamespace         string
@@ -111,6 +119,7 @@ type SiteConfigSpec struct {
 	Controller               ControllerOptions
 	ConfigSync               ConfigSyncOptions
 	FlowCollector            FlowCollectorOptions
+	PrometheusServer         PrometheusServerOptions
 	Platform                 Platform
 	RunAsUser                int64
 	RunAsGroup               int64
diff --git a/api/types/types.go b/api/types/types.go
index 9d45b0d8..18ec6600 100644
--- a/api/types/types.go
+++ b/api/types/types.go
@@ -80,7 +80,7 @@ var TransportPrometheusAnnotations = map[string]string{
 	"prometheus.io/scrape": "true",
 }
 
-// Controller constants
+// Controller and Collector constants
 const (
 	ControllerDeploymentName             string = "skupper-service-controller"
 	ControllerComponentName              string = "service-controller"
@@ -93,6 +93,13 @@ const (
 	ControllerConfigPath                 string = "/etc/messaging/"
 	ControllerServiceName                string = "skupper"
 	FlowCollectorContainerName           string = "flow-collector"
+	PrometheusDeploymentName             string = "skupper-prometheus"
+	PrometheusComponentName              string = "prometheus"
+	PrometheusContainerName              string = "prometheus-server"
+	PrometheusServiceAccountName         string = "skupper-prometheus"
+	PrometheusServiceName                string = "skupper-prometheus"
+	PrometheusRoleBindingName            string = "skupper-prometheus"
+	PrometheusRoleName                   string = "skupper-prometheus"
 )
 
 var ControllerPolicyRule = []rbacv1.PolicyRule{
@@ -156,6 +163,7 @@ const (
 	SiteCaSecret             string = "skupper-site-ca"
 	ConsoleServerSecret      string = "skupper-console-certs"
 	ConsoleUsersSecret       string = "skupper-console-users"
+	PrometheusServerSecret   string = "skupper-prometheus-certs"
 	OauthRouterConsoleSecret string = "skupper-router-console-certs"
 	ServiceCaSecret          string = "skupper-service-ca"
 	ServiceClientSecret      string = "skupper-service-client" // Secret that is used in sslProfiles for all http2 connectors with tls enabled
@@ -253,6 +261,8 @@ const (
 	RouterConsoleServiceName               string = "skupper-router-console"
 )
 
+const DefaultFlowTimeoutDuration = time.Minute * 15
+
 type Platform string
 
 const (
@@ -278,6 +288,22 @@ const (
 	ClaimRedemptionRouteName string = "claims"
 )
 
+type PrometheusAuthMode string
+
+const (
+	PrometheusAuthModeTls       PrometheusAuthMode = "tls"
+	PrometheusAuthModeBasic                        = "basic"
+	PrometheusAuthModeUnsecured                    = "unsecured"
+)
+
+// Prometheus server constants (note: use console auth mode)
+const (
+	PrometheusPortName                       string = "prometheus"
+	PrometheusServerDefaultServicePort       int32  = 9090
+	PrometheusServerDefaultServiceTargetPort int32  = 9090
+	PrometheusRouteName                      string = "skupper-prometheus"
+)
+
 // Assembly constants
 const (
 	AmqpDefaultPort         int32  = 5672
@@ -310,11 +336,13 @@ type RouterSpec struct {
 	ConfigSync            DeploymentSpec  `json:"configSync,omitempty"`
 	Controller            DeploymentSpec  `json:"controller,omitempty"`
 	Collector             DeploymentSpec  `json:"collector,omitempty"`
+	PrometheusServer      DeploymentSpec  `json:"prometheusServer,omitempty"`
 	RouterConfig          string          `json:"routerConfig,omitempty"`
 	Users                 []User          `json:"users,omitempty"`
 	CertAuthoritys        []CertAuthority `json:"certAuthoritys,omitempty"`
 	TransportCredentials  []Credential    `json:"transportCredentials,omitempty"`
 	ControllerCredentials []Credential    `json:"controllerCredentials,omitempty"`
+	PrometheusCredentials []Credential    `json:"prometheusCredentials,omitempty"`
 }
 
 type ImageDetails struct {
@@ -341,6 +369,7 @@ type DeploymentSpec struct {
 	ServiceAccounts     []*corev1.ServiceAccount     `json:"serviceAccounts,omitempty"`
 	Services            []*corev1.Service            `json:"services,omitempty"`
 	Sidecars            []*corev1.Container          `json:"sidecars,omitempty"`
+	SecurityContext     *corev1.SecurityContext      `json:"securityContext,omitempty"`
 	Affinity            map[string]string            `json:"affinity,omitempty"`
 	AntiAffinity        map[string]string            `json:"antiAffinity,omitempty"`
 	NodeSelector        map[string]string            `json:"nodeSelector,omitempty"`
@@ -458,6 +487,7 @@ const (
 	ClaimsIngressPrefix      = "skupper-claims"
 	InterRouterIngressPrefix = "skupper-inter-router"
 	EdgeIngressPrefix        = "skupper-edge"
+	PrometheusIngressPrefix  = "skupper-prometheus"
 )
 
 type Connector struct {
diff --git a/client/router_create.go b/client/router_create.go
index bf239818..47c2397c 100644
--- a/client/router_create.go
+++ b/client/router_create.go
@@ -1,16 +1,19 @@
 package client
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"log"
 	"strconv"
 	"strings"
+	"text/template"
 	"time"
 
 	routev1 "github.com/openshift/api/route/v1"
 	"github.com/skupperproject/skupper/pkg/images"
 	"github.com/skupperproject/skupper/pkg/version"
+	"golang.org/x/crypto/bcrypt"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
@@ -78,6 +81,220 @@ func (cli *VanClient) getControllerRules() []rbacv1.PolicyRule {
 	return rules
 }
 
+func hashPrometheusPassword(password string) ([]byte, error) {
+	return bcrypt.GenerateFromPassword([]byte(password), 14)
+}
+
+func (cli *VanClient) GetVanPrometheusServerSpec(options types.SiteConfigSpec, van *types.RouterSpec, transport *appsv1.Deployment, siteId string) {
+	// prometheus-server container index
+	const (
+		prometheusServer = iota
+	)
+
+	van.PrometheusServer.Image = images.GetPrometheusServerImageDetails()
+	van.PrometheusServer.Replicas = 1
+	van.PrometheusServer.LabelSelector = map[string]string{
+		types.ComponentAnnotation: types.PrometheusComponentName,
+	}
+	van.PrometheusServer.Labels = map[string]string{
+		types.AppLabel:    types.PrometheusDeploymentName,
+		types.PartOfLabel: types.AppName,
+	}
+	for key, value := range van.PrometheusServer.LabelSelector {
+		van.PrometheusServer.Labels[key] = value
+	}
+	van.PrometheusServer.Annotations = options.Annotations
+	for key, value := range options.Labels {
+		van.PrometheusServer.Labels[key] = value
+	}
+
+	envVars := []corev1.EnvVar{}
+
+	sidecars := []*corev1.Container{}
+	volumes := []corev1.Volume{}
+	mounts := make([][]corev1.VolumeMount, 1)
+	err := configureDeployment(&van.PrometheusServer, &options.PrometheusServer.Tuning)
+	if err != nil {
+		fmt.Println("Error configuring prometheus server deployment:", err)
+	}
+	kube.AppendConfigVolume(&volumes, &mounts[prometheusServer], "prometheus-config", "prometheus-server-config", "/etc/prometheus")
+	kube.AppendSharedVolume(&volumes, []*[]corev1.VolumeMount{&mounts[prometheusServer]}, "prometheus-storage-volume", "/prometheus")
+	if options.PrometheusServer.AuthMode == string(types.PrometheusAuthModeTls) {
+		kube.AppendSecretVolume(&volumes, &mounts[prometheusServer], types.PrometheusServerSecret, "/etc/tls/certs/")
+	}
+
+	van.PrometheusServer.EnvVar = envVars
+	van.PrometheusServer.Volumes = volumes
+	van.PrometheusServer.VolumeMounts = mounts
+	van.PrometheusServer.Sidecars = sidecars
+
+	serviceAccounts := []*corev1.ServiceAccount{}
+	annotation := map[string]string{}
+	serviceAccounts = append(serviceAccounts, &corev1.ServiceAccount{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "v1",
+			Kind:       "ServiceAccount",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        types.PrometheusServiceAccountName,
+			Annotations: annotation,
+		},
+	})
+	van.PrometheusServer.ServiceAccounts = serviceAccounts
+
+	van.PrometheusCredentials = []types.Credential{}
+
+	roles := []*rbacv1.Role{}
+	roles = append(roles, &rbacv1.Role{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "rbac.authorization.k8s.io/v1",
+			Kind:       "Role",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: types.PrometheusRoleName,
+		},
+		Rules: cli.getControllerRules(),
+	})
+	van.PrometheusServer.Roles = roles
+
+	roleBindings := []*rbacv1.RoleBinding{}
+	roleBindings = append(roleBindings, &rbacv1.RoleBinding{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "rbac.authorization.k8s.io/v1",
+			Kind:       "RoleBinding",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: types.PrometheusRoleBindingName,
+		},
+		Subjects: []rbacv1.Subject{{
+			Kind: "ServiceAccount",
+			Name: types.PrometheusServiceAccountName,
+		}},
+		RoleRef: rbacv1.RoleRef{
+			Kind: "Role",
+			Name: types.PrometheusRoleName,
+		},
+	})
+	van.PrometheusServer.RoleBindings = roleBindings
+
+	svctype := corev1.ServiceTypeClusterIP
+	if options.IsConsoleIngressLoadBalancer() {
+		svctype = corev1.ServiceTypeLoadBalancer
+	} else if options.IsConsoleIngressNodePort() {
+		svctype = corev1.ServiceTypeNodePort
+	}
+	annotations := map[string]string{}
+	prometheusPorts := []corev1.ServicePort{}
+	routes := []*routev1.Route{}
+	prometheusPort := corev1.ServicePort{
+		Name:       "prometheus",
+		Protocol:   "TCP",
+		Port:       types.PrometheusServerDefaultServicePort,
+		TargetPort: intstr.FromInt(int(types.PrometheusServerDefaultServiceTargetPort)),
+	}
+
+	if options.IsConsoleIngressRoute() {
+		annotations = map[string]string{"service.alpha.openshift.io/serving-cert-secret-name": types.PrometheusServerSecret}
+		host := options.GetControllerIngressHost()
+		if host != "" {
+			host = types.PrometheusRouteName + "-" + van.Namespace + "." + host
+		}
+
+		route := routev1.Route{
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: "v1",
+				Kind:       "Route",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name: types.PrometheusRouteName,
+			},
+			Spec: routev1.RouteSpec{
+				Path: "",
+				Host: host,
+				Port: &routev1.RoutePort{
+					TargetPort: intstr.FromString("prometheus"),
+				},
+				To: routev1.RouteTargetReference{
+					Kind: "Service",
+					Name: types.PrometheusServiceName,
+				},
+			},
+		}
+		if options.PrometheusServer.AuthMode == string(types.PrometheusAuthModeTls) {
+			route.Spec.TLS = &routev1.TLSConfig{
+				Termination:                   routev1.TLSTerminationReencrypt,
+				InsecureEdgeTerminationPolicy: routev1.InsecureEdgeTerminationPolicyRedirect,
+			}
+		}
+		routes = append(routes, &route)
+	} else {
+		// note: using controller ingress
+		prometheusHosts := []string{types.PrometheusServiceName + "." + van.Namespace}
+		prometheusIngressHost := options.GetControllerIngressHost()
+		post := false // indicates whether credentials need to be revised after creating appropriate ingress resources
+		if options.IsIngressNginxIngress() {
+			if prometheusIngressHost != "" {
+				prometheusHosts = append(prometheusHosts, strings.Join([]string{"prometheus", van.Namespace, prometheusIngressHost}, "."))
+			} else {
+				post = true
+			}
+		} else if options.IsIngressContourHttpProxy() {
+			if prometheusIngressHost != "" {
+				prometheusHosts = append(prometheusHosts, strings.Join([]string{"prometheus", van.Namespace, prometheusIngressHost}, "."))
+			}
+		} else if options.IsIngressLoadBalancer() {
+			post = true
+		} else {
+			if prometheusIngressHost != "" {
+				prometheusHosts = append(prometheusHosts, prometheusIngressHost)
+			}
+		}
+		van.PrometheusCredentials = append(van.PrometheusCredentials, types.Credential{
+			CA:          types.LocalCaSecret,
+			Name:        types.PrometheusServerSecret,
+			Subject:     types.PrometheusServiceName,
+			Hosts:       prometheusHosts,
+			ConnectJson: false,
+			Post:        post,
+		})
+	}
+	prometheusPorts = append(prometheusPorts, prometheusPort)
+
+	svcs := []*corev1.Service{}
+	if len(prometheusPorts) > 0 {
+		svc := &corev1.Service{
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: "v1",
+				Kind:       "Service",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:        types.PrometheusServiceName,
+				Annotations: annotations,
+			},
+			Spec: corev1.ServiceSpec{
+				Selector: van.PrometheusServer.Labels,
+				Ports:    prometheusPorts,
+				Type:     svctype,
+			},
+		}
+
+		if options.Controller.LoadBalancerIp != "" && svctype == corev1.ServiceTypeLoadBalancer {
+			svc.Spec.LoadBalancerIP = options.Controller.LoadBalancerIp
+		}
+
+		svcs = append(svcs, svc)
+		for _, p := range prometheusPorts {
+			van.PrometheusServer.Ports = append(van.PrometheusServer.Ports, corev1.ContainerPort{
+				Name:          p.Name,
+				ContainerPort: p.Port,
+			})
+		}
+	}
+
+	van.PrometheusServer.Services = svcs
+	van.PrometheusServer.Routes = routes
+}
+
 func (cli *VanClient) GetVanControllerSpec(options types.SiteConfigSpec, van *types.RouterSpec, transport *appsv1.Deployment, siteId string) {
 	// service-controller container index
 	const (
@@ -102,6 +319,16 @@ func (cli *VanClient) GetVanControllerSpec(options types.SiteConfigSpec, van *ty
 	for key, value := range options.Labels {
 		van.Controller.Labels[key] = value
 	}
+	runAsNonRoot := true
+	van.Controller.SecurityContext = &corev1.SecurityContext{
+		RunAsNonRoot: &runAsNonRoot,
+	}
+	if options.RunAsUser > 0 {
+		van.Controller.SecurityContext.RunAsUser = &options.RunAsUser
+	}
+	if options.RunAsGroup > 0 {
+		van.Controller.SecurityContext.RunAsGroup = &options.RunAsGroup
+	}
 
 	envVars := []corev1.EnvVar{}
 	envVars = append(envVars, corev1.EnvVar{Name: "SKUPPER_NAMESPACE", Value: van.Namespace})
@@ -557,6 +784,16 @@ func (cli *VanClient) GetRouterSpecFromOpts(options types.SiteConfigSpec, siteId
 	for key, value := range options.Annotations {
 		van.Transport.Annotations[key] = value
 	}
+	runAsNonRoot := true
+	van.Transport.SecurityContext = &corev1.SecurityContext{
+		RunAsNonRoot: &runAsNonRoot,
+	}
+	if options.RunAsUser > 0 {
+		van.Transport.SecurityContext.RunAsUser = &options.RunAsUser
+	}
+	if options.RunAsGroup > 0 {
+		van.Transport.SecurityContext.RunAsGroup = &options.RunAsGroup
+	}
 	err := configureDeployment(&van.Transport, &options.Router.Tuning)
 	if err != nil {
 		fmt.Println("Error configuring router:", err)
@@ -1039,18 +1276,7 @@ sasldb_path: /tmp/skrouterd.sasldb
 		}
 	}
 
-	runAsNonRoot := true
-	securityContext := corev1.SecurityContext{
-		RunAsNonRoot: &runAsNonRoot,
-	}
-	if options.Spec.RunAsUser > 0 {
-		securityContext.RunAsUser = &options.Spec.RunAsUser
-	}
-	if options.Spec.RunAsGroup > 0 {
-		securityContext.RunAsGroup = &options.Spec.RunAsGroup
-	}
-
-	dep, err := kube.NewTransportDeployment(van, siteOwnerRef, &securityContext, cli.KubeClient)
+	dep, err := kube.NewTransportDeployment(van, siteOwnerRef, cli.KubeClient)
 	if err != nil && !errors.IsAlreadyExists(err) {
 		return err
 	}
@@ -1216,7 +1442,103 @@ sasldb_path: /tmp/skrouterd.sasldb
 				return err
 			}
 		}
-		_, err = kube.NewControllerDeployment(van, siteOwnerRef, &securityContext, cli.KubeClient)
+		_, err = kube.NewControllerDeployment(van, siteOwnerRef, cli.KubeClient)
+		if err != nil && !errors.IsAlreadyExists(err) {
+			return err
+		}
+	}
+	if options.Spec.EnableFlowCollector && options.Spec.PrometheusServer.ExternalServer == "" {
+		//	Stand up local prometheus server for metric aggregation
+		cli.GetVanPrometheusServerSpec(options.Spec, van, dep, siteId)
+		err := configureDeployment(&van.PrometheusServer, &options.Spec.Controller.Tuning)
+		if err != nil {
+			fmt.Println("Error configuring controller:", err)
+		}
+
+		promInfo := PrometheusInfo{
+			BasicAuth:   false,
+			TlsAuth:     false,
+			ServiceName: types.ControllerServiceName,
+			Namespace:   van.Namespace,
+			Port:        strconv.Itoa(int(types.FlowCollectorDefaultServicePort)),
+			User:        "admin",
+			Password:    "admin",
+			Hash:        "",
+		}
+		if options.Spec.PrometheusServer.AuthMode == string(types.PrometheusAuthModeBasic) {
+			promInfo.BasicAuth = true
+			if options.Spec.PrometheusServer.User != "" {
+				promInfo.User = options.Spec.PrometheusServer.User
+			}
+			if options.Spec.PrometheusServer.Password != "" {
+				promInfo.Password = options.Spec.PrometheusServer.Password
+			}
+			hash, _ := hashPrometheusPassword(promInfo.Password)
+			promInfo.Hash = string(hash)
+		} else if options.Spec.PrometheusServer.AuthMode == string(types.PrometheusAuthModeTls) {
+			promInfo.TlsAuth = true
+		}
+		prometheusData := &map[string]string{
+			"prometheus.yml": scrapeConfigForPrometheus(promInfo),
+			"web-config.yml": webConfigForPrometheus(promInfo),
+		}
+		kube.NewConfigMap("prometheus-server-config", prometheusData, nil, nil, siteOwnerRef, van.Namespace, cli.KubeClient)
+
+		for _, sa := range van.PrometheusServer.ServiceAccounts {
+			sa.ObjectMeta.OwnerReferences = ownerRefs
+			_, err = kube.CreateServiceAccount(van.Namespace, sa, cli.KubeClient)
+			if err != nil && !errors.IsAlreadyExists(err) {
+				return err
+			}
+		}
+		for _, role := range van.PrometheusServer.Roles {
+			role.ObjectMeta.OwnerReferences = ownerRefs
+			_, err = kube.CreateRole(van.Namespace, role, cli.KubeClient)
+			if err != nil && !errors.IsAlreadyExists(err) {
+				return err
+			}
+		}
+		for _, roleBinding := range van.PrometheusServer.RoleBindings {
+			roleBinding.ObjectMeta.OwnerReferences = ownerRefs
+			_, err = kube.CreateRoleBinding(van.Namespace, roleBinding, cli.KubeClient)
+			if err != nil && !errors.IsAlreadyExists(err) {
+				return err
+			}
+		}
+		for _, svc := range van.PrometheusServer.Services {
+			svc.ObjectMeta.OwnerReferences = ownerRefs
+			_, err = kube.CreateService(svc, van.Namespace, cli.KubeClient)
+			if err != nil && !errors.IsAlreadyExists(err) {
+				return err
+			}
+		}
+		if options.Spec.IsIngressRoute() {
+			for _, rte := range van.PrometheusServer.Routes {
+				rte.ObjectMeta.OwnerReferences = ownerRefs
+				_, err = kube.CreateRoute(rte, van.Namespace, cli.RouteClient)
+				if err != nil && !errors.IsAlreadyExists(err) {
+					return err
+				}
+			}
+		}
+		for _, cred := range van.PrometheusCredentials {
+			if options.Spec.IsIngressLoadBalancer() {
+				err = cli.appendLoadBalancerHostOrIp(currentContext, types.PrometheusServiceName, van.Namespace, &cred)
+				if err != nil {
+					return err
+				}
+			} else if options.Spec.IsIngressNginxIngress() && cred.Post {
+				err = cli.appendIngressHost([]string{"prometheus"}, van.Namespace, &cred)
+				if err != nil {
+					return err
+				}
+			}
+			_, err = kube.NewSecret(cred, siteOwnerRef, van.Namespace, cli.KubeClient)
+			if err != nil && !errors.IsAlreadyExists(err) {
+				return err
+			}
+		}
+		_, err = kube.NewPrometheusServerDeployment(van, siteOwnerRef, cli.KubeClient)
 		if err != nil && !errors.IsAlreadyExists(err) {
 			return err
 		}
@@ -1441,3 +1763,64 @@ func asOwnerReferences(in types.SiteConfigReference) []metav1.OwnerReference {
 	}
 	return []metav1.OwnerReference{*ref}
 }
+
+type PrometheusInfo struct {
+	BasicAuth   bool
+	TlsAuth     bool
+	ServiceName string
+	Namespace   string
+	Port        string
+	User        string
+	Password    string
+	Hash        string
+}
+
+func scrapeConfigForPrometheus(info PrometheusInfo) string {
+	const prometheusConfig = `
+global:
+  scrape_interval:     15s
+  evaluation_interval: 15s
+alerting:
+  alertmanagers:
+  - static_configs:
+    - targets:
+rule_files:
+  # - "example-file.yml"
+scrape_configs:
+  - job_name: 'prometheus'
+    metrics_path: "/api/v1alpha1/metrics"
+    scheme: "https"
+    tls_config:
+      insecure_skip_verify: true
+    static_configs:
+    - targets: ["{{.ServiceName}}.{{.Namespace}}.svc.cluster.local:{{.Port}}"]
+`
+	var buf bytes.Buffer
+	promConfig := template.Must(template.New("promConfig").Parse(prometheusConfig))
+	promConfig.Execute(&buf, info)
+
+	return buf.String()
+}
+
+func webConfigForPrometheus(info PrometheusInfo) string {
+	const webConfig = `
+# TLS configuration.
+{{- if .TlsAuth }}
+tls_server_config:
+  cert_file: /etc/tls/certs/tls.crt
+  key_file: /etc/tls/certs/tls.key
+{{- end}}
+#
+# Usernames and passwords required to connect to Prometheus.
+# Passwords are hashed with bcrypt: https://github.com/prometheus/exporter-toolkit/blob/master/docs/web-configuration.md#about-bcrypt
+basic_auth_users:
+{{- if .BasicAuth }}
+  {{.User}}: {{.Hash}}
+{{- end}}
+`
+	var buf bytes.Buffer
+	promConfig := template.Must(template.New("prmConfig").Parse(webConfig))
+	promConfig.Execute(&buf, info)
+
+	return buf.String()
+}
diff --git a/client/site_config_create.go b/client/site_config_create.go
index 7e9d9113..2f69f9f4 100644
--- a/client/site_config_create.go
+++ b/client/site_config_create.go
@@ -44,6 +44,16 @@ const (
 	SiteConfigFlowCollectorCpuLimitKey    string = "flow-collector-cpu-limit"
 	SiteConfigFlowCollectorMemoryLimitKey string = "flow-collector-memory-limit"
 
+	// prometheus server options
+	SiteConfigPrometheusExternalServerKey       string = "prometheus-external-server"
+	SiteConfigPrometheusServerAuthenticationKey string = "prometheus-server-authentication"
+	SiteConfigPrometheusServerUserKey           string = "prometheus-server-user"
+	SiteConfigPrometheusServerPasswordKey       string = "prometheus-server-password"
+	SiteConfigPrometheusServerCpuKey            string = "prometheus-server-cpu"
+	SiteConfigPrometheusServerMemoryKey         string = "prometheus-server-memory"
+	SiteConfigPrometheusServerCpuLimitKey       string = "prometheus-server-cpu-limit"
+	SiteConfigPrometheusServerMemoryLimitKey    string = "prometheus-server-memory-limit"
+
 	// router options
 	SiteConfigRouterConsoleKey            string = "router-console"
 	SiteConfigRouterLoggingKey            string = "router-logging"
@@ -102,8 +112,8 @@ func (cli *VanClient) SiteConfigCreate(ctx context.Context, spec types.SiteConfi
 			SiteConfigRouterModeKey:            string(types.TransportModeInterior),
 			SiteConfigServiceControllerKey:     "true",
 			SiteConfigServiceSyncKey:           "true",
-			SiteConfigConsoleKey:               "true",
-			SiteConfigFlowCollectorKey:         "true",
+			SiteConfigConsoleKey:               "false",
+			SiteConfigFlowCollectorKey:         "false",
 			SiteConfigClusterPermissionsKey:    "false",
 			SiteConfigRouterConsoleKey:         "false",
 			SiteConfigRouterLoggingKey:         "",
@@ -131,14 +141,14 @@ func (cli *VanClient) SiteConfigCreate(ctx context.Context, spec types.SiteConfi
 	if spec.SiteTtl != 0 {
 		siteConfig.Data[SiteConfigServiceSyncSiteTtlKey] = spec.SiteTtl.String()
 	}
-	if !spec.EnableConsole {
-		siteConfig.Data[SiteConfigConsoleKey] = "false"
+	if spec.EnableConsole {
+		siteConfig.Data[SiteConfigConsoleKey] = "true"
 	}
 	if spec.EnableRestAPI {
 		siteConfig.Data[SiteConfigRestAPIKey] = "true"
 	}
-	if !spec.EnableFlowCollector {
-		siteConfig.Data[SiteConfigFlowCollectorKey] = "false"
+	if spec.EnableFlowCollector {
+		siteConfig.Data[SiteConfigFlowCollectorKey] = "true"
 	}
 	if spec.AuthMode != "" {
 		siteConfig.Data[SiteConfigConsoleAuthenticationKey] = spec.AuthMode
@@ -344,6 +354,43 @@ func (cli *VanClient) SiteConfigCreate(ctx context.Context, spec types.SiteConfi
 		siteConfig.Data[SiteConfigEnableSkupperEventsKey] = "false"
 	}
 
+	if spec.PrometheusServer.ExternalServer != "" {
+		siteConfig.Data[SiteConfigPrometheusExternalServerKey] = spec.PrometheusServer.ExternalServer
+	}
+	if spec.PrometheusServer.AuthMode != "" {
+		siteConfig.Data[SiteConfigPrometheusServerAuthenticationKey] = spec.PrometheusServer.AuthMode
+	}
+	if spec.PrometheusServer.User != "" {
+		siteConfig.Data[SiteConfigPrometheusServerUserKey] = spec.PrometheusServer.User
+	}
+	if spec.PrometheusServer.Password != "" {
+		siteConfig.Data[SiteConfigPrometheusServerPasswordKey] = spec.PrometheusServer.Password
+	}
+	if spec.PrometheusServer.Cpu != "" {
+		if _, err := resource.ParseQuantity(spec.PrometheusServer.Cpu); err != nil {
+			return nil, fmt.Errorf("Invalid value for %s %q: %s", SiteConfigPrometheusServerCpuKey, spec.PrometheusServer.Cpu, err)
+		}
+		siteConfig.Data[SiteConfigPrometheusServerCpuKey] = spec.PrometheusServer.Cpu
+	}
+	if spec.PrometheusServer.Memory != "" {
+		if _, err := resource.ParseQuantity(spec.PrometheusServer.Memory); err != nil {
+			return nil, fmt.Errorf("Invalid value for %s %q: %s", SiteConfigPrometheusServerMemoryKey, spec.PrometheusServer.Memory, err)
+		}
+		siteConfig.Data[SiteConfigPrometheusServerMemoryKey] = spec.PrometheusServer.Memory
+	}
+	if spec.PrometheusServer.CpuLimit != "" {
+		if _, err := resource.ParseQuantity(spec.PrometheusServer.CpuLimit); err != nil {
+			return nil, fmt.Errorf("Invalid value for %s %q: %s", SiteConfigPrometheusServerCpuLimitKey, spec.PrometheusServer.CpuLimit, err)
+		}
+		siteConfig.Data[SiteConfigPrometheusServerCpuLimitKey] = spec.PrometheusServer.CpuLimit
+	}
+	if spec.PrometheusServer.MemoryLimit != "" {
+		if _, err := resource.ParseQuantity(spec.PrometheusServer.MemoryLimit); err != nil {
+			return nil, fmt.Errorf("Invalid value for %s %q: %s", SiteConfigPrometheusServerMemoryLimitKey, spec.PrometheusServer.MemoryLimit, err)
+		}
+		siteConfig.Data[SiteConfigPrometheusServerMemoryLimitKey] = spec.PrometheusServer.MemoryLimit
+	}
+
 	// TODO: allow Replicas to be set through skupper-site configmap?
 	if !spec.SiteControlled {
 		if siteConfig.ObjectMeta.Labels == nil {
diff --git a/client/site_config_inspect.go b/client/site_config_inspect.go
index b32f4d1b..366f24f1 100644
--- a/client/site_config_inspect.go
+++ b/client/site_config_inspect.go
@@ -94,6 +94,8 @@ func (cli *VanClient) SiteConfigInspectInNamespace(ctx context.Context, input *c
 		if err == nil {
 			result.Spec.FlowCollector.FlowRecordTtl = ttl
 		}
+	} else {
+		result.Spec.FlowCollector.FlowRecordTtl = types.DefaultFlowTimeoutDuration
 	}
 	if value, ok := siteConfig.Data[SiteConfigRestAPIKey]; ok {
 		result.Spec.EnableRestAPI, _ = strconv.ParseBool(value)
@@ -176,6 +178,12 @@ func (cli *VanClient) SiteConfigInspectInNamespace(ctx context.Context, input *c
 			return &result, err
 		}
 		result.Spec.Router.Logging = logConf
+	} else {
+		logConf, err := qdr.ParseRouterLogConfig("ROUTER_CORE:error+")
+		if err != nil {
+			return &result, err
+		}
+		result.Spec.Router.Logging = logConf
 	}
 	if routerCpu, ok := siteConfig.Data[SiteConfigRouterCpuKey]; ok && routerCpu != "" {
 		result.Spec.Router.Cpu = routerCpu
@@ -279,6 +287,51 @@ func (cli *VanClient) SiteConfigInspectInNamespace(ctx context.Context, input *c
 		result.Spec.EnableSkupperEvents, _ = strconv.ParseBool(value)
 	}
 
+	if flowCollectorCpu, ok := siteConfig.Data[SiteConfigFlowCollectorCpuKey]; ok && flowCollectorCpu != "" {
+		result.Spec.FlowCollector.Cpu = flowCollectorCpu
+	}
+	if flowCollectorMemory, ok := siteConfig.Data[SiteConfigFlowCollectorMemoryKey]; ok && flowCollectorMemory != "" {
+		result.Spec.FlowCollector.Memory = flowCollectorMemory
+	}
+	if flowCollectorCpuLimit, ok := siteConfig.Data[SiteConfigFlowCollectorCpuLimitKey]; ok && flowCollectorCpuLimit != "" {
+		result.Spec.FlowCollector.CpuLimit = flowCollectorCpuLimit
+	}
+	if flowCollectorMemoryLimit, ok := siteConfig.Data[SiteConfigFlowCollectorMemoryLimitKey]; ok && flowCollectorMemoryLimit != "" {
+		result.Spec.FlowCollector.MemoryLimit = flowCollectorMemoryLimit
+	}
+
+	if externalServer, ok := siteConfig.Data[SiteConfigPrometheusExternalServerKey]; ok {
+		result.Spec.PrometheusServer.ExternalServer = externalServer
+	} else {
+		result.Spec.PrometheusServer.ExternalServer = ""
+	}
+	if authMode, ok := siteConfig.Data[SiteConfigPrometheusServerAuthenticationKey]; ok {
+		result.Spec.PrometheusServer.AuthMode = authMode
+	} else {
+		result.Spec.PrometheusServer.AuthMode = string(types.PrometheusAuthModeTls)
+	}
+	if user, ok := siteConfig.Data[SiteConfigPrometheusServerUserKey]; ok {
+		result.Spec.PrometheusServer.User = user
+	} else {
+		result.Spec.PrometheusServer.User = ""
+	}
+	if password, ok := siteConfig.Data[SiteConfigPrometheusServerPasswordKey]; ok {
+		result.Spec.PrometheusServer.Password = password
+	} else {
+		result.Spec.PrometheusServer.Password = ""
+	}
+	if prometheusCpu, ok := siteConfig.Data[SiteConfigPrometheusServerCpuKey]; ok && prometheusCpu != "" {
+		result.Spec.PrometheusServer.Cpu = prometheusCpu
+	}
+	if prometheusMemory, ok := siteConfig.Data[SiteConfigPrometheusServerMemoryKey]; ok && prometheusMemory != "" {
+		result.Spec.PrometheusServer.Memory = prometheusMemory
+	}
+	if prometheusCpuLimit, ok := siteConfig.Data[SiteConfigPrometheusServerCpuLimitKey]; ok && prometheusCpuLimit != "" {
+		result.Spec.PrometheusServer.CpuLimit = prometheusCpuLimit
+	}
+	if prometheusMemoryLimit, ok := siteConfig.Data[SiteConfigPrometheusServerMemoryLimitKey]; ok && prometheusMemoryLimit != "" {
+		result.Spec.PrometheusServer.MemoryLimit = prometheusMemoryLimit
+	}
 	annotationExclusions := []string{}
 	labelExclusions := []string{}
 	annotations := map[string]string{}
diff --git a/cmd/flow-collector/controller.go b/cmd/flow-collector/controller.go
index cefc9d26..64ad374e 100644
--- a/cmd/flow-collector/controller.go
+++ b/cmd/flow-collector/controller.go
@@ -8,6 +8,7 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 
 	"github.com/interconnectedcloud/go-amqp"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/skupperproject/skupper/pkg/flow"
 	"github.com/skupperproject/skupper/pkg/qdr"
 )
@@ -16,14 +17,14 @@ type Controller struct {
 	origin        string
 	amqpClient    *amqp.Client
 	amqpSession   *amqp.Session
-	flowCollector *flow.FlowCollector
+	FlowCollector *flow.FlowCollector
 }
 
-func NewController(origin string, scheme string, host string, port string, tlsConfig *tls.Config, recordTtl time.Duration) (*Controller, error) {
+func NewController(origin string, reg prometheus.Registerer, scheme string, host string, port string, tlsConfig *tls.Config, recordTtl time.Duration) (*Controller, error) {
 
 	controller := &Controller{
 		origin:        origin,
-		flowCollector: flow.NewFlowCollector(origin, qdr.NewConnectionFactory(scheme+"://"+host+":"+port, tlsConfig), recordTtl),
+		FlowCollector: flow.NewFlowCollector(origin, reg, qdr.NewConnectionFactory(scheme+"://"+host+":"+port, tlsConfig), recordTtl),
 	}
 
 	return controller, nil
@@ -34,7 +35,7 @@ func (c *Controller) Run(stopCh <-chan struct{}) error {
 
 	log.Println("Starting the Skupper flow collector")
 
-	c.flowCollector.Start(stopCh)
+	c.FlowCollector.Start(stopCh)
 
 	<-stopCh
 	log.Println("Shutting down the Skupper flow collector")
diff --git a/cmd/flow-collector/deploy.yaml b/cmd/flow-collector/deploy.yaml
index e8b98658..fe8d5099 100644
--- a/cmd/flow-collector/deploy.yaml
+++ b/cmd/flow-collector/deploy.yaml
@@ -13,6 +13,12 @@ metadata:
     application: skupper-flow-collector
   name: skupper-flow-collector
 rules:
+- apiGroups:
+  - ""
+  resources:
+  - configmaps
+  verbs:
+  - get
 - apiGroups:
   - apps
   resources:
@@ -56,11 +62,11 @@ spec:
       # do NOT support this field by default
       securityContext:
         runAsNonRoot: true
-        seccompProfile:
-          type: RuntimeDefault
+        #seccompProfile:
+          #type: RuntimeDefault
       containers:
       - name: skupper-flow-collector
-        image: quay.io/skupper/flow-collector:latest
+        image: quay.io/skupper/flow-collector:master
         imagePullPolicy: Always
         securityContext:
           capabilities:
diff --git a/cmd/flow-collector/handlers.go b/cmd/flow-collector/handlers.go
index ec6a164b..932cd641 100644
--- a/cmd/flow-collector/handlers.go
+++ b/cmd/flow-collector/handlers.go
@@ -9,8 +9,8 @@ import (
 
 func (c *Controller) eventsourceHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
-	c.flowCollector.Request <- flow.ApiRequest{RecordType: flow.EventSource, Request: r}
-	response := <-c.flowCollector.Response
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.EventSource, Request: r}
+	response := <-c.FlowCollector.Response
 	w.WriteHeader(response.Status)
 	if response.Body != nil {
 		fmt.Fprintf(w, "%s", *response.Body)
@@ -19,8 +19,8 @@ func (c *Controller) eventsourceHandler(w http.ResponseWriter, r *http.Request)
 
 func (c *Controller) siteHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
-	c.flowCollector.Request <- flow.ApiRequest{RecordType: flow.Site, Request: r}
-	response := <-c.flowCollector.Response
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.Site, Request: r}
+	response := <-c.FlowCollector.Response
 	w.WriteHeader(response.Status)
 	if response.Body != nil {
 		fmt.Fprintf(w, "%s", *response.Body)
@@ -29,8 +29,8 @@ func (c *Controller) siteHandler(w http.ResponseWriter, r *http.Request) {
 
 func (c *Controller) hostHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
-	c.flowCollector.Request <- flow.ApiRequest{RecordType: flow.Host, Request: r}
-	response := <-c.flowCollector.Response
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.Host, Request: r}
+	response := <-c.FlowCollector.Response
 	w.WriteHeader(response.Status)
 	if response.Body != nil {
 		fmt.Fprintf(w, "%s", *response.Body)
@@ -39,8 +39,8 @@ func (c *Controller) hostHandler(w http.ResponseWriter, r *http.Request) {
 
 func (c *Controller) routerHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
-	c.flowCollector.Request <- flow.ApiRequest{RecordType: flow.Router, Request: r}
-	response := <-c.flowCollector.Response
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.Router, Request: r}
+	response := <-c.FlowCollector.Response
 	w.WriteHeader(response.Status)
 	if response.Body != nil {
 		fmt.Fprintf(w, "%s", *response.Body)
@@ -49,8 +49,8 @@ func (c *Controller) routerHandler(w http.ResponseWriter, r *http.Request) {
 
 func (c *Controller) linkHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
-	c.flowCollector.Request <- flow.ApiRequest{RecordType: flow.Link, Request: r}
-	response := <-c.flowCollector.Response
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.Link, Request: r}
+	response := <-c.FlowCollector.Response
 	w.WriteHeader(response.Status)
 	if response.Body != nil {
 		fmt.Fprintf(w, "%s", *response.Body)
@@ -59,8 +59,8 @@ func (c *Controller) linkHandler(w http.ResponseWriter, r *http.Request) {
 
 func (c *Controller) listenerHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
-	c.flowCollector.Request <- flow.ApiRequest{RecordType: flow.Listener, Request: r}
-	response := <-c.flowCollector.Response
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.Listener, Request: r}
+	response := <-c.FlowCollector.Response
 	w.WriteHeader(response.Status)
 	if response.Body != nil {
 		fmt.Fprintf(w, "%s", *response.Body)
@@ -69,8 +69,8 @@ func (c *Controller) listenerHandler(w http.ResponseWriter, r *http.Request) {
 
 func (c *Controller) connectorHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
-	c.flowCollector.Request <- flow.ApiRequest{RecordType: flow.Connector, Request: r}
-	response := <-c.flowCollector.Response
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.Connector, Request: r}
+	response := <-c.FlowCollector.Response
 	w.WriteHeader(response.Status)
 	if response.Body != nil {
 		fmt.Fprintf(w, "%s", *response.Body)
@@ -79,8 +79,8 @@ func (c *Controller) connectorHandler(w http.ResponseWriter, r *http.Request) {
 
 func (c *Controller) addressHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
-	c.flowCollector.Request <- flow.ApiRequest{RecordType: flow.Address, Request: r}
-	response := <-c.flowCollector.Response
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.Address, Request: r}
+	response := <-c.FlowCollector.Response
 	w.WriteHeader(response.Status)
 	if response.Body != nil {
 		fmt.Fprintf(w, "%s", *response.Body)
@@ -89,8 +89,8 @@ func (c *Controller) addressHandler(w http.ResponseWriter, r *http.Request) {
 
 func (c *Controller) processHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
-	c.flowCollector.Request <- flow.ApiRequest{RecordType: flow.Process, Request: r}
-	response := <-c.flowCollector.Response
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.Process, Request: r}
+	response := <-c.FlowCollector.Response
 	w.WriteHeader(response.Status)
 	if response.Body != nil {
 		fmt.Fprintf(w, "%s", *response.Body)
@@ -99,8 +99,8 @@ func (c *Controller) processHandler(w http.ResponseWriter, r *http.Request) {
 
 func (c *Controller) processGroupHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
-	c.flowCollector.Request <- flow.ApiRequest{RecordType: flow.ProcessGroup, Request: r}
-	response := <-c.flowCollector.Response
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.ProcessGroup, Request: r}
+	response := <-c.FlowCollector.Response
 	w.WriteHeader(response.Status)
 	if response.Body != nil {
 		fmt.Fprintf(w, "%s", *response.Body)
@@ -109,8 +109,8 @@ func (c *Controller) processGroupHandler(w http.ResponseWriter, r *http.Request)
 
 func (c *Controller) flowHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
-	c.flowCollector.Request <- flow.ApiRequest{RecordType: flow.Flow, Request: r}
-	response := <-c.flowCollector.Response
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.Flow, Request: r}
+	response := <-c.FlowCollector.Response
 	w.WriteHeader(response.Status)
 	if response.Body != nil {
 		fmt.Fprintf(w, "%s", *response.Body)
@@ -119,8 +119,8 @@ func (c *Controller) flowHandler(w http.ResponseWriter, r *http.Request) {
 
 func (c *Controller) flowPairHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
-	c.flowCollector.Request <- flow.ApiRequest{RecordType: flow.FlowPair, Request: r}
-	response := <-c.flowCollector.Response
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.FlowPair, Request: r}
+	response := <-c.FlowCollector.Response
 	w.WriteHeader(response.Status)
 	if response.Body != nil {
 		fmt.Fprintf(w, "%s", *response.Body)
@@ -129,8 +129,8 @@ func (c *Controller) flowPairHandler(w http.ResponseWriter, r *http.Request) {
 
 func (c *Controller) sitePairHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
-	c.flowCollector.Request <- flow.ApiRequest{RecordType: flow.SitePair, Request: r}
-	response := <-c.flowCollector.Response
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.SitePair, Request: r}
+	response := <-c.FlowCollector.Response
 	w.WriteHeader(response.Status)
 	if response.Body != nil {
 		fmt.Fprintf(w, "%s", *response.Body)
@@ -139,8 +139,8 @@ func (c *Controller) sitePairHandler(w http.ResponseWriter, r *http.Request) {
 
 func (c *Controller) processGroupPairHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
-	c.flowCollector.Request <- flow.ApiRequest{RecordType: flow.ProcessGroupPair, Request: r}
-	response := <-c.flowCollector.Response
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.ProcessGroupPair, Request: r}
+	response := <-c.FlowCollector.Response
 	w.WriteHeader(response.Status)
 	if response.Body != nil {
 		fmt.Fprintf(w, "%s", *response.Body)
@@ -149,8 +149,18 @@ func (c *Controller) processGroupPairHandler(w http.ResponseWriter, r *http.Requ
 
 func (c *Controller) processPairHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
-	c.flowCollector.Request <- flow.ApiRequest{RecordType: flow.ProcessPair, Request: r}
-	response := <-c.flowCollector.Response
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.ProcessPair, Request: r}
+	response := <-c.FlowCollector.Response
+	w.WriteHeader(response.Status)
+	if response.Body != nil {
+		fmt.Fprintf(w, "%s", *response.Body)
+	}
+}
+
+func (c *Controller) collectorHandler(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Content-Type", "application/json")
+	c.FlowCollector.Request <- flow.ApiRequest{RecordType: flow.Collector, Request: r}
+	response := <-c.FlowCollector.Response
 	w.WriteHeader(response.Status)
 	if response.Body != nil {
 		fmt.Fprintf(w, "%s", *response.Body)
diff --git a/cmd/flow-collector/main.go b/cmd/flow-collector/main.go
index ff8866bc..f445510c 100644
--- a/cmd/flow-collector/main.go
+++ b/cmd/flow-collector/main.go
@@ -17,6 +17,8 @@ import (
 	"time"
 
 	"github.com/gorilla/mux"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	clientpodman "github.com/skupperproject/skupper/client/podman"
 	"github.com/skupperproject/skupper/pkg/config"
 	"github.com/skupperproject/skupper/pkg/domain/podman"
@@ -153,6 +155,10 @@ func main() {
 	platform := config.GetPlatform()
 	var flowRecordTtl time.Duration
 	var enableConsole bool
+	var prometheusHost string
+	var prometheusAuthMethod string
+	var prometheusUser string
+	var prometheusPassword string
 
 	// waiting on skupper-router to be available
 	if platform == "" || platform == types.PlatformKubernetes {
@@ -174,6 +180,36 @@ func main() {
 
 		flowRecordTtl = siteConfig.Spec.FlowCollector.FlowRecordTtl
 		enableConsole = siteConfig.Spec.EnableConsole
+		prometheusAuthMethod = siteConfig.Spec.PrometheusServer.AuthMode
+		prometheusUser = siteConfig.Spec.PrometheusServer.User
+		prometheusPassword = siteConfig.Spec.PrometheusServer.Password
+		promProtocol := "https://"
+		if prometheusAuthMethod == "basic" {
+			promProtocol = "http://"
+		}
+		if siteConfig.Spec.PrometheusServer.ExternalServer != "" {
+			prometheusHost = promProtocol + siteConfig.Spec.PrometheusServer.ExternalServer + ":9090" + "/api/v1"
+			//	} else if siteConfig.Spec.Ingress == types.IngressRouteString {
+		} else if siteConfig.Spec.IsIngressRoute() {
+			route, err := kube.GetRoute(types.PrometheusRouteName, namespace, cli.RouteClient)
+			if err != nil {
+				log.Fatal("Error getting prometheus host", err.Error())
+			}
+			prometheusHost = promProtocol + route.Spec.Host + "/api/v1"
+		} else if siteConfig.Spec.IsIngressLoadBalancer() {
+			service, err := kube.GetService(types.PrometheusServiceName, namespace, cli.KubeClient)
+			if err == nil {
+				host := kube.GetLoadBalancerHostOrIP(service)
+				prometheusHost = promProtocol + host + ":9090" + "/api/v1"
+			}
+		} else if siteConfig.Spec.IsIngressNginxIngress() || siteConfig.Spec.IsIngressKubernetes() {
+			routes, err := kube.GetIngressRoutes("skupper-prometheus", namespace, cli)
+			if err == nil {
+				host := routes[0].Host
+				port := strconv.Itoa(routes[0].ServicePort)
+				prometheusHost = promProtocol + host + ":" + port + "/api/v1"
+			}
+		}
 	} else {
 		cfg, err := podman.NewPodmanConfigFileHandler().GetConfig()
 		if err != nil {
@@ -198,6 +234,10 @@ func main() {
 		}
 		flowRecordTtl, _ = time.ParseDuration(os.Getenv("FLOW_RECORD_TTL"))
 		enableConsole, _ = strconv.ParseBool(os.Getenv("ENABLE_CONSOLE"))
+		prometheusHost = os.Getenv("PROMETHEUS_HOST")
+		prometheusAuthMethod = os.Getenv("PROMETHEUS_AUTH_METHOD")
+		prometheusUser = os.Getenv("PROMETHEUS_USER")
+		prometheusPassword = os.Getenv("PROMETHEUS_PASSWORD")
 	}
 
 	tlsConfig, err := certs.GetTlsConfig(true, types.ControllerConfigPath+"tls.crt", types.ControllerConfigPath+"tls.key", types.ControllerConfigPath+"ca.crt")
@@ -210,11 +250,17 @@ func main() {
 		log.Fatal("Error getting connect.json", err.Error())
 	}
 
-	c, err := NewController(origin, conn.Scheme, conn.Host, conn.Port, tlsConfig, flowRecordTtl)
+	reg := prometheus.NewRegistry()
+	c, err := NewController(origin, reg, conn.Scheme, conn.Host, conn.Port, tlsConfig, flowRecordTtl)
 	if err != nil {
 		log.Fatal("Error getting new flow collector ", err.Error())
 	}
 
+	c.FlowCollector.Collector.PrometheusHost = prometheusHost
+	c.FlowCollector.Collector.PrometheusAuthMethod = prometheusAuthMethod
+	c.FlowCollector.Collector.PrometheusUser = prometheusUser
+	c.FlowCollector.Collector.PrometheusPassword = prometheusPassword
+
 	var mux = mux.NewRouter().StrictSlash(true)
 
 	var api = mux.PathPrefix("/api").Subrouter()
@@ -239,6 +285,10 @@ func main() {
 		})
 	}
 
+	var metricsApi = api1.PathPrefix("/metrics").Subrouter()
+	metricsApi.StrictSlash(true)
+	metricsApi.Handle("/", promhttp.HandlerFor(reg, promhttp.HandlerOpts{Registry: reg}))
+
 	var eventsourceApi = api1.PathPrefix("/eventsources").Subrouter()
 	eventsourceApi.StrictSlash(true)
 	eventsourceApi.HandleFunc("/", http.HandlerFunc(c.eventsourceHandler)).Name("list")
@@ -386,6 +436,14 @@ func main() {
 		log.Println("Skupper console is disabled")
 	}
 
+	var collectorApi = api1.PathPrefix("/collectors").Subrouter()
+	collectorApi.StrictSlash(true)
+	collectorApi.HandleFunc("/", authenticated(http.HandlerFunc(c.collectorHandler))).Name("list")
+	collectorApi.HandleFunc("/{id}", authenticated(http.HandlerFunc(c.collectorHandler))).Name("item")
+	collectorApi.NotFoundHandler = authenticated(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusNotFound)
+	}))
+
 	addr := ":8010"
 	if os.Getenv("FLOW_PORT") != "" {
 		addr = ":" + os.Getenv("FLOW_PORT")
diff --git a/cmd/skupper/skupper_kube_site.go b/cmd/skupper/skupper_kube_site.go
index f8c1117a..f7352fe7 100644
--- a/cmd/skupper/skupper_kube_site.go
+++ b/cmd/skupper/skupper_kube_site.go
@@ -157,12 +157,21 @@ func (s *SkupperKubeSite) CreateFlags(cmd *cobra.Command) {
 	cmd.Flags().StringVar(&routerCreateOpts.ConfigSync.MemoryLimit, "config-sync-memory-limit", "", "Memory limit for config-sync pods")
 	cmd.Flags().BoolVarP(&routerCreateOpts.EnableClusterPermissions, "enable-cluster-permissions", "", false, "Enable cluster wide permissions in order to expose deployments/statefulsets in other namespaces")
 
-	cmd.Flags().DurationVar(&routerCreateOpts.FlowCollector.FlowRecordTtl, "flow-collector-record-ttl", 0, "Time after which terminated flow records are deleted, i.e. those flow records that have an end time set. Default is 30 minutes.")
+	cmd.Flags().DurationVar(&routerCreateOpts.FlowCollector.FlowRecordTtl, "flow-collector-record-ttl", 0, "Time after which terminated flow records are deleted, i.e. those flow records that have an end time set. Default is 15 minutes.")
 	cmd.Flags().StringVar(&routerCreateOpts.FlowCollector.Cpu, "flow-collector-cpu", "", "CPU request for flow collector pods")
 	cmd.Flags().StringVar(&routerCreateOpts.FlowCollector.Memory, "flow-collector-memory", "", "Memory request for flow collector pods")
 	cmd.Flags().StringVar(&routerCreateOpts.FlowCollector.CpuLimit, "flow-collector-cpu-limit", "", "CPU limit for flow collector pods")
 	cmd.Flags().StringVar(&routerCreateOpts.FlowCollector.MemoryLimit, "flow-collector-memory-limit", "", "Memory limit for flow collector pods")
 
+	cmd.Flags().StringVarP(&routerCreateOpts.PrometheusServer.ExternalServer, "external-prometheus-server", "", "", "External prometheus server for metric aggregation. Valid only when --enable-flow-collector")
+	cmd.Flags().StringVarP(&routerCreateOpts.PrometheusServer.AuthMode, "prometheus-auth", "", "", "Authentication mode for skupper prometheus server. One of: 'tls', 'basic', 'unsecured'")
+	cmd.Flags().StringVarP(&routerCreateOpts.PrometheusServer.User, "prometheus-user", "", "", "Skupper prometheus user. Valid only when --prometheus-auth=basic")
+	cmd.Flags().StringVarP(&routerCreateOpts.PrometheusServer.Password, "prometheus-password", "", "", "Skupper prometheus user. Valid only when --prometheus-auth=basic")
+	cmd.Flags().StringVar(&routerCreateOpts.PrometheusServer.Cpu, "prometheus-cpu", "", "CPU request for prometheus pods")
+	cmd.Flags().StringVar(&routerCreateOpts.PrometheusServer.Memory, "prometheus-memory", "", "Memory request for prometheus pods")
+	cmd.Flags().StringVar(&routerCreateOpts.PrometheusServer.CpuLimit, "prometheus-cpu-limit", "", "CPU limit for prometheus pods")
+	cmd.Flags().StringVar(&routerCreateOpts.PrometheusServer.MemoryLimit, "prometheus-memory-limit", "", "Memory limit for prometheus pods")
+
 	cmd.Flags().DurationVar(&LoadBalancerTimeout, "timeout", types.DefaultTimeoutDuration, "Configurable timeout for the ingress loadbalancer option.")
 	cmd.Flags().BoolVar(&routerCreateOpts.EnableSkupperEvents, "enable-skupper-events", true, "Enable sending Skupper events to Kubernetes")
 
diff --git a/go.mod b/go.mod
index 88b8b252..30d1badd 100644
--- a/go.mod
+++ b/go.mod
@@ -10,13 +10,13 @@ require (
 	github.com/go-openapi/swag v0.21.1
 	github.com/go-openapi/validate v0.22.0
 	github.com/golang/glog v1.0.0
-	github.com/google/go-cmp v0.5.2
+	github.com/google/go-cmp v0.5.8
 	github.com/google/uuid v1.1.2
 	github.com/gorilla/mux v1.8.0
 	github.com/interconnectedcloud/go-amqp v0.12.6-0.20200506124159-f51e540008b5
 	github.com/openshift/api v0.0.0-20210428205234-a8389931bee7
 	github.com/openshift/client-go v0.0.0-20210112165513-ebc401615f47
-	github.com/prometheus/common v0.4.0
+	github.com/prometheus/client_golang v1.14.0
 	github.com/rogpeppe/go-internal v1.3.0
 	github.com/skupperproject/skupper-example-tcp-echo v0.0.0-20210727195922-db4a7dc7b35d
 	github.com/spf13/cobra v0.0.6
@@ -25,7 +25,7 @@ require (
 	golang.org/x/net v0.8.0
 	golang.org/x/text v0.8.0
 	google.golang.org/grpc v1.35.0
-	google.golang.org/protobuf v1.25.0
+	google.golang.org/protobuf v1.28.1
 	gopkg.in/yaml.v3 v3.0.1
 	gotest.tools v2.2.0+incompatible
 	k8s.io/api v0.20.0
@@ -36,7 +36,7 @@ require (
 replace github.com/gogo/protobuf => github.com/gogo/protobuf v1.3.2
 
 require (
-	cloud.google.com/go v0.54.0 // indirect
+	cloud.google.com/go v0.65.0 // indirect
 	github.com/Azure/azure-sdk-for-go v46.0.0+incompatible // indirect
 	github.com/Azure/go-autorest v14.2.0+incompatible // indirect
 	github.com/Azure/go-autorest/autorest v0.11.1 // indirect
@@ -48,9 +48,9 @@ require (
 	github.com/Azure/go-autorest/tracing v0.6.0 // indirect
 	github.com/PuerkitoBio/purell v1.1.1 // indirect
 	github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect
-	github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc // indirect
-	github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf // indirect
 	github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.0 // indirect
 	github.com/docker/spdystream v0.0.0-20160310174837-449fdfce4d96 // indirect
 	github.com/evanphx/json-patch v4.9.0+incompatible // indirect
@@ -64,7 +64,7 @@ require (
 	github.com/go-openapi/spec v0.20.4 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e // indirect
-	github.com/golang/protobuf v1.4.3 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/golang/snappy v0.0.1 // indirect
 	github.com/google/gofuzz v1.1.0 // indirect
 	github.com/googleapis/gnostic v0.4.1 // indirect
@@ -73,10 +73,10 @@ require (
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect
 	github.com/influxdata/tdigest v0.0.0-20180711151920-a7d76c6f093a // indirect
 	github.com/josharian/intern v1.0.0 // indirect
-	github.com/json-iterator/go v1.1.10 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/compress v1.13.6 // indirect
-	github.com/konsorten/go-windows-terminal-sequences v1.0.2 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/mitchellh/mapstructure v1.4.3 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
@@ -84,24 +84,24 @@ require (
 	github.com/oklog/ulid v1.3.1 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
+	github.com/prometheus/client_model v0.3.0 // indirect
+	github.com/prometheus/common v0.37.0 // indirect
+	github.com/prometheus/procfs v0.8.0 // indirect
 	github.com/russross/blackfriday/v2 v2.0.1 // indirect
 	github.com/shurcooL/sanitized_anchor_name v1.0.0 // indirect
-	github.com/sirupsen/logrus v1.4.2 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/scram v1.1.1 // indirect
 	github.com/xdg-go/stringprep v1.0.3 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect
 	golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d // indirect
-	golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d // indirect
-	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
+	golang.org/x/oauth2 v0.0.0-20220223155221-ee480838109b // indirect
+	golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f // indirect
 	golang.org/x/sys v0.6.0 // indirect
 	golang.org/x/term v0.6.0 // indirect
 	golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e // indirect
-	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
-	google.golang.org/appengine v1.6.5 // indirect
+	google.golang.org/appengine v1.6.6 // indirect
 	google.golang.org/genproto v0.0.0-20210106152847-07624b53cd92 // indirect
-	gopkg.in/alecthomas/kingpin.v2 v2.2.6 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	k8s.io/klog/v2 v2.4.0 // indirect
diff --git a/go.sum b/go.sum
index a50b710f..631c5c39 100644
--- a/go.sum
+++ b/go.sum
@@ -8,19 +8,29 @@ cloud.google.com/go v0.46.3/go.mod h1:a6bKKbmY7er1mI7TEI4lsAkts/mkhTSZK8w33B4RAg
 cloud.google.com/go v0.50.0/go.mod h1:r9sluTvynVuxRIOHXQEHMFffphuXHOMZMycpNR5e6To=
 cloud.google.com/go v0.52.0/go.mod h1:pXajvRH/6o3+F9jDHZWQ5PbGhn+o8w9qiu/CffaVdO4=
 cloud.google.com/go v0.53.0/go.mod h1:fp/UouUEsRkN6ryDKNW/Upv/JBKnv6WDthjR6+vze6M=
-cloud.google.com/go v0.54.0 h1:3ithwDMr7/3vpAMXiH+ZQnYbuIsh+OPhUPMFC9enmn0=
 cloud.google.com/go v0.54.0/go.mod h1:1rq2OEkV3YMf6n/9ZvGWI3GWw0VoqH/1x2nd8Is/bPc=
+cloud.google.com/go v0.56.0/go.mod h1:jr7tqZxxKOVYizybht9+26Z/gUq7tiRzu+ACVAMbKVk=
+cloud.google.com/go v0.57.0/go.mod h1:oXiQ6Rzq3RAkkY7N6t3TcE6jE+CIBBbA36lwQ1JyzZs=
+cloud.google.com/go v0.62.0/go.mod h1:jmCYTdRCQuc1PHIIJ/maLInMho30T/Y0M4hTdTShOYc=
+cloud.google.com/go v0.65.0 h1:Dg9iHVQfrhq82rUNu9ZxUDrJLaxFUe/HlCVaLyRruq8=
+cloud.google.com/go v0.65.0/go.mod h1:O5N8zS7uWy9vkA9vayVHs65eM1ubvY4h553ofrNHObY=
 cloud.google.com/go/bigquery v1.0.1/go.mod h1:i/xbL2UlR5RvWAURpBYZTtm/cXjCha9lbfbpx4poX+o=
 cloud.google.com/go/bigquery v1.3.0/go.mod h1:PjpwJnslEMmckchkHFfq+HTD2DmtT67aNFKH1/VBDHE=
 cloud.google.com/go/bigquery v1.4.0/go.mod h1:S8dzgnTigyfTmLBfrtrhyYhwRxG72rYxvftPBK2Dvzc=
+cloud.google.com/go/bigquery v1.5.0/go.mod h1:snEHRnqQbz117VIFhE8bmtwIDY80NLUZUMb4Nv6dBIg=
+cloud.google.com/go/bigquery v1.7.0/go.mod h1://okPTzCYNXSlb24MZs83e2Do+h+VXtc4gLoIoXIAPc=
+cloud.google.com/go/bigquery v1.8.0/go.mod h1:J5hqkt3O0uAFnINi6JXValWIb1v0goeZM77hZzJN/fQ=
 cloud.google.com/go/datastore v1.0.0/go.mod h1:LXYbyblFSglQ5pkeyhO+Qmw7ukd3C+pD7TKLgZqpHYE=
 cloud.google.com/go/datastore v1.1.0/go.mod h1:umbIZjpQpHh4hmRpGhH4tLFup+FVzqBi1b3c64qFpCk=
 cloud.google.com/go/pubsub v1.0.1/go.mod h1:R0Gpsv3s54REJCy4fxDixWD93lHJMoZTyQ2kNxGRt3I=
 cloud.google.com/go/pubsub v1.1.0/go.mod h1:EwwdRX2sKPjnvnqCa270oGRyludottCI76h+R3AArQw=
 cloud.google.com/go/pubsub v1.2.0/go.mod h1:jhfEVHT8odbXTkndysNHCcx0awwzvfOlguIAii9o8iA=
+cloud.google.com/go/pubsub v1.3.1/go.mod h1:i+ucay31+CNRpDW4Lu78I4xXG+O1r/MAHgjpRVR+TSU=
 cloud.google.com/go/storage v1.0.0/go.mod h1:IhtSnM/ZTZV8YYJWCY8RULGVqBDmpoyjwiyrjsg+URw=
 cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0ZeosJ0Rtdos=
 cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=
+cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RXyy7KQOVs=
+cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/Azure/azure-sdk-for-go v46.0.0+incompatible h1:4qlEOCDcDQZTGczYGzbGYCdJfVpZLIs8AEo5+MoXBPw=
 github.com/Azure/azure-sdk-for-go v46.0.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
@@ -53,10 +63,11 @@ github.com/PuerkitoBio/purell v1.1.1/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbt
 github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 h1:d+Bc7a5rLufV/sSk/8dngufqelfh6jnri85riMAaF/M=
 github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE=
 github.com/alecthomas/jsonschema v0.0.0-20180308105923-f2c93856175a/go.mod h1:qpebaTNSsyUn5rPSJMsfqEtDw71TTggXM6stUDI16HA=
-github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc h1:cAKDfWh5VpdgMhJosfJnn5/FoN2SRZ4p7fJNX58YPaU=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
-github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf h1:qet1QNfXsQxTZqLG4oE62mJzwPIB8+Tee4RNCL9ulrY=
+github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
+github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
+github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
 github.com/asaskevich/govalidator v0.0.0-20190424111038-f61b66f89f4a/go.mod h1:lB+ZfQJz7igIIfQNfa7Ml4HSf2uFQQRzpGGRXenZAgY=
 github.com/asaskevich/govalidator v0.0.0-20200907205600-7a23bdc65eef/go.mod h1:WaHUgvxTVq04UNunO+XhnAqY/wQc+bxr74GqbsZ/Jqw=
@@ -64,15 +75,21 @@ github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d h1:Byv0BzEl
 github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d/go.mod h1:WaHUgvxTVq04UNunO+XhnAqY/wQc+bxr74GqbsZ/Jqw=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
+github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
+github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b h1:AP/Y7sqYicnjGDfD5VcY4CIfh1hRXBUavxrvELjTiOE=
 github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b/go.mod h1:ac9efd0D1fsDb3EJvhqgXRbFx7bs2wqZ10HQPeU8U/Q=
 github.com/c2h5oh/datasize v0.0.0-20171227191756-4eba002a5eae/go.mod h1:S/7n9copUssQ56c7aAgHqftWO4LTf4xY6CGWt8Bc+3M=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
+github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
+github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
 github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
+github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
@@ -100,6 +117,7 @@ github.com/emicklei/go-restful v0.0.0-20170410110728-ff4f55a20633/go.mod h1:otzb
 github.com/emicklei/go-restful v2.9.5+incompatible/go.mod h1:otzb+WCGbkyDHkqmQmT5YD2WR4BBwUdeQoFo8l/7tVs=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
+github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/evanphx/json-patch v4.9.0+incompatible h1:kLcOMZeuLAJvL2BPWLMIj5oaZQobrkAqrL+WFZwQses=
@@ -116,8 +134,13 @@ github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
 github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
+github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
+github.com/go-kit/log v0.1.0/go.mod h1:zbhenjAZHb184qTLMA9ZjW7ThYL0H2mk7Q6pNt4vbaY=
+github.com/go-kit/log v0.2.0/go.mod h1:NwTd00d/i8cPZ3xOwwiv2PO5MOcx78fFErGNcVmBjv0=
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
+github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A=
+github.com/go-logfmt/logfmt v0.5.1/go.mod h1:WYhtIu8zTZfxdn5+rREduYbwxfcBr/Vr6KEVveWlfTs=
 github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7sIas=
 github.com/go-logr/logr v0.2.0 h1:QvGt2nLcHH0WK9orKa+ppBPAxREcH364nPUedEpK0TY=
 github.com/go-logr/logr v0.2.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU=
@@ -197,11 +220,14 @@ github.com/golang/mock v1.2.0/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfb
 github.com/golang/mock v1.3.1/go.mod h1:sBzyDLLjw3U8JLTeZvSv8jJB+tU5PVekmnlKIyFUx0Y=
 github.com/golang/mock v1.4.0/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
 github.com/golang/mock v1.4.1/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
+github.com/golang/mock v1.4.3/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
+github.com/golang/mock v1.4.4/go.mod h1:l3mdAwkq5BuhzHwde/uurv3sEJeZMXNpwsxVWU71h+4=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.3/go.mod h1:vzj43D7+SQXF/4pzW/hwtAqwc6iTitCiVSaWz5lYuqw=
 github.com/golang/protobuf v1.3.4/go.mod h1:vzj43D7+SQXF/4pzW/hwtAqwc6iTitCiVSaWz5lYuqw=
+github.com/golang/protobuf v1.3.5/go.mod h1:6O5/vntMXwX2lRkT1hjjk0nAC1IDOTvTlVgjlRvqsdk=
 github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8=
 github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA=
 github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs=
@@ -209,8 +235,10 @@ github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:W
 github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0=
 github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QDs8UjoX8=
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
-github.com/golang/protobuf v1.4.3 h1:JjCZWpVbqXDqFVmTfYWEVTMIYrL/NPdPSCHPJ0T/raM=
 github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
+github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw=
+github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
 github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/gonum/blas v0.0.0-20181208220705-f22b278b28ac h1:Q0Jsdxl5jbxouNs1TQYt0gxesYMU4VXRbsTlgDloZ50=
@@ -235,18 +263,26 @@ github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5a
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.4.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/google/go-cmp v0.5.2 h1:X2ev0eStA3AbceY54o37/0PQ/UWqKEiiO2dKL5OPaFM=
+github.com/google/go-cmp v0.5.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.8 h1:e6P7q2lk1O+qJJb4BtCQXlK8vWEO8V1ZeuEdJNOqZyg=
+github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/gofuzz v1.1.0 h1:Hsa8mG0dQ46ij8Sl2AYJDUv1oA9/d6Vk+3LG99Oe02g=
 github.com/google/gofuzz v1.1.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
+github.com/google/martian/v3 v3.0.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0=
 github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
 github.com/google/pprof v0.0.0-20190515194954-54271f7e092f/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
 github.com/google/pprof v0.0.0-20191218002539-d4f498aebedc/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
 github.com/google/pprof v0.0.0-20200212024743-f11f1df84d12/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
 github.com/google/pprof v0.0.0-20200229191704-1ebb73c60ed3/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
+github.com/google/pprof v0.0.0-20200430221834-fc25d7d30c6d/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
+github.com/google/pprof v0.0.0-20200708004538-1a94d8640e99/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.2 h1:EVhdT+1Kseyi1/pUmXKaFxYsDNy9RQYkMWRH68J/W7Y=
@@ -282,12 +318,16 @@ github.com/joho/godotenv v1.3.0/go.mod h1:7hK45KPybAkOC6peb+G5yklZfMxEjkZhHbwpqx
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
+github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
-github.com/json-iterator/go v1.1.10 h1:Kz6Cvnvv2wGdaG/V8yMvfkmNiXq9Ya2KUv4rouJJr68=
 github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/json-iterator/go v1.1.11/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
+github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
 github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/XSXhF0NWZEnDohbsk=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
+github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
 github.com/karrick/godirwalk v1.8.0/go.mod h1:H5KPZjojv4lE+QYImBI8xVtrBRgYrIVsaRPx4tDPEn4=
 github.com/karrick/godirwalk v1.10.3/go.mod h1:RoGL9dQei4vP9ilrpETWE8CLOZ1kiN0LhBygSwrAsHA=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
@@ -295,8 +335,8 @@ github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+o
 github.com/klauspost/compress v1.13.6 h1:P76CopJELS0TiO2mebmnzgWaajssP/EszplttgQxcgc=
 github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
-github.com/konsorten/go-windows-terminal-sequences v1.0.2 h1:DB17ag19krx9CFsz4o3enTrPXyIXCl+2iCXH/aMAp9s=
 github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
+github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
@@ -314,6 +354,7 @@ github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0
 github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
 github.com/markbates/oncer v0.0.0-20181203154359-bf2de49a0be2/go.mod h1:Ld9puTsIW75CHf65OeIOkyKbteujpZVXDpWK6YGZbxE=
 github.com/markbates/safe v1.0.1/go.mod h1:nAqgmRi7cY2nqMc92/bSEeQA+R4OheNU2T1kNSCBdG0=
+github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/miekg/dns v1.1.17/go.mod h1:WgzbA6oji13JREwiNsRDNfl7jYdPnmz+VEuLrA+/48M=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
@@ -333,6 +374,7 @@ github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe h1:iruDEfMl2E6f
 github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc=
 github.com/munnerz/goautoneg v0.0.0-20120707110453-a547fc61f48d/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
+github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f/go.mod h1:ZdcZmHo+o7JKHSa8/e818NopupXU1YMK5fe1lsApnBw=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
@@ -364,14 +406,34 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDft0ttaMvbicHlPoso=
+github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
+github.com/prometheus/client_golang v1.7.1/go.mod h1:PY5Wy2awLA44sXw4AOSfFBetzPP4j5+D6mVACh+pe2M=
+github.com/prometheus/client_golang v1.11.0/go.mod h1:Z6t4BnS23TR94PD6BsDNk8yVqroYurpAkEiz0P2BEV0=
+github.com/prometheus/client_golang v1.12.1/go.mod h1:3Z9XVyYiZYEO+YQWt3RD2R3jrbd179Rt297l4aS6nDY=
+github.com/prometheus/client_golang v1.14.0 h1:nJdhIvne2eSX/XRAFV9PcvFFRbrjbcTUj0VP62TMhnw=
+github.com/prometheus/client_golang v1.14.0/go.mod h1:8vpkKitgIVNcqrRBWh1C4TIUQgYNtG/XQE4E/Zae36Y=
 github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
+github.com/prometheus/client_model v0.2.0/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
+github.com/prometheus/client_model v0.3.0 h1:UBgGFHqYdG/TPFD1B1ogZywDqEkwp3fBMvqdiQ7Xew4=
+github.com/prometheus/client_model v0.3.0/go.mod h1:LDGWKZIo7rky3hgvBe+caln+Dr3dPggB5dvjtD7w9+w=
 github.com/prometheus/common v0.0.0-20181113130724-41aa239b4cce/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
-github.com/prometheus/common v0.4.0 h1:7etb9YClo3a6HjLzfl6rIQaU+FDfi0VSX39io3aQ+DM=
 github.com/prometheus/common v0.4.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
+github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
+github.com/prometheus/common v0.10.0/go.mod h1:Tlit/dnDKsSWFlCLTWaA1cyBgKHSMdTB80sz/V91rCo=
+github.com/prometheus/common v0.26.0/go.mod h1:M7rCNAaPfAosfx8veZJCuw84e35h3Cfd9VFqTh1DIvc=
+github.com/prometheus/common v0.32.1/go.mod h1:vu+V0TpY+O6vW9J44gczi3Ap/oXXR10b+M/gUGO4Hls=
+github.com/prometheus/common v0.37.0 h1:ccBbHCgIiT9uSoFY0vX8H3zsNR5eLt17/RQLUvn8pXE=
+github.com/prometheus/common v0.37.0/go.mod h1:phzohg0JFMnBEFGxTDbfu3QyL5GI8gTQJFhYO5B3mfA=
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
 github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
+github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
+github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
+github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
+github.com/prometheus/procfs v0.7.3/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
+github.com/prometheus/procfs v0.8.0 h1:ODq8ZFEaYeCaZOJlZZdJA2AbQR98dSHSM1KW/You5mo=
+github.com/prometheus/procfs v0.8.0/go.mod h1:z7EfXMXOkbkqb9IINtpCn86r/to3BnA0uaxHdg830/4=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/go-internal v1.1.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
@@ -385,8 +447,8 @@ github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeV
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMBDgk/93Q=
-github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
+github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
 github.com/skupperproject/skupper-example-tcp-echo v0.0.0-20210727195922-db4a7dc7b35d h1:XmA9DC9Ffu1gnNAJZyjjoNImRs+iI5XBf/30YbqXVcs=
 github.com/skupperproject/skupper-example-tcp-echo v0.0.0-20210727195922-db4a7dc7b35d/go.mod h1:+Hm/U62PKiUAFpD3RTrjogZnPA0W2ZIDRG9+Cf7qjSU=
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=
@@ -437,7 +499,9 @@ github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q
 github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
 github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d h1:splanxYIlg+5LfHAM6xpdFEAYOk8iySO56hMFq6uLyA=
 github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA=
+github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.mongodb.org/mongo-driver v1.7.3/go.mod h1:NqaYOwnXWr5Pm7AOpO5QFxKJ503nbMse/R79oO62zWg=
@@ -449,6 +513,7 @@ go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
 go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
 go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
+go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
 go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
@@ -511,6 +576,7 @@ golang.org/x/net v0.0.0-20190522155817-f3200d17e092/go.mod h1:HSz+uSET+XFnRR8LxR
 golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
 golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20190628185345-da137c7871d7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190724013045-ca1201d0de80/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190827160401-ba9fcec4b297/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191209160850-c0dbc17a3553/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
@@ -520,19 +586,30 @@ golang.org/x/net v0.0.0-20200222125558-5a598a2470a0/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200301022130-244492dfa37a/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200324143707-d3edc9973b7e/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
+golang.org/x/net v0.0.0-20200501053045-e0ff5e5a1de5/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
+golang.org/x/net v0.0.0-20200506145744-7e3656a0809f/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
+golang.org/x/net v0.0.0-20200513185701-a91f0712d120/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
+golang.org/x/net v0.0.0-20200520182314-0ba52f642ac2/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
+golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
+golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
+golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20210421230115-4e50805a0758/go.mod h1:72T/g9IO56b78aLF+1Kcs5dz7/ng1VjMUvfKvpfy+jM=
+golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
+golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
 golang.org/x/net v0.8.0 h1:Zrh2ngAOFYneWTAIAPethzeaQLuHwhuBkuV6ZiRnUaQ=
 golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20191202225959-858c2ad4c8b6/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
-golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d h1:TzXSXBo42m9gQenoE3b9BGiEpg5IG2JkU5FkPIawgtw=
 golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
+golang.org/x/oauth2 v0.0.0-20210514164344-f6687ab2804c/go.mod h1:KelEdhl1UZF7XfJ4dDtk6s++YSgaE7mD/BuKKDLBl4A=
+golang.org/x/oauth2 v0.0.0-20220223155221-ee480838109b h1:clP8eMhB30EHdc0bd2Twtq6kgU7yl5ub2cQLSdrv1Dg=
+golang.org/x/oauth2 v0.0.0-20220223155221-ee480838109b/go.mod h1:DAh4E804XQdzx2j+YRIaUnCqCV2RuMz24cGBJ5QYIrc=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -540,9 +617,13 @@ golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190412183630-56d357773e84/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
+golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f h1:Ax0t5p6N38Ga0dThY21weqDEyz2oklo4IvDkpigvkD8=
+golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -567,6 +648,7 @@ golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191228213918-04cbcbbfeed8/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200106162015-b016eb3dc98e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200113162924-86b910548bc1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200122134326-e047566fdf82/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200202164722-d101bd2416d5/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -574,13 +656,24 @@ golang.org/x/sys v0.0.0-20200212091648-12a6c2dcc1e4/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200331124033-c3d80250170d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200501052902-10377860bb8e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200511232937-7e40ca221e25/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200515095857-1151b9dac4a9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200523222454-059865788121/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200615200032-f1bc736245b1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200803210538-64077c9b5642/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201112073958-5cba982894dd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210420072515-93ed5bcd2bfe/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0 h1:MVltZSvRTcU2ljQOhs94SXPftV6DCNnZViHeQps87pQ=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
@@ -639,15 +732,24 @@ golang.org/x/tools v0.0.0-20200204074204-1cc6d1ef6c74/go.mod h1:TB2adYChydJhpapK
 golang.org/x/tools v0.0.0-20200207183749-b753a1ba74fa/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/tools v0.0.0-20200212150539-ea181f53ac56/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/tools v0.0.0-20200224181240-023911ca70b2/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
+golang.org/x/tools v0.0.0-20200227222343-706bc42d1f0d/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/tools v0.0.0-20200304193943-95d2e580d8eb/go.mod h1:o4KQGtdN14AW+yjsvvwRTJJuXz8XRtIHtEnmAXLyFUw=
+golang.org/x/tools v0.0.0-20200312045724-11d5b4c81c7d/go.mod h1:o4KQGtdN14AW+yjsvvwRTJJuXz8XRtIHtEnmAXLyFUw=
+golang.org/x/tools v0.0.0-20200331025713-a30bf2db82d4/go.mod h1:Sl4aGygMT6LrqrWclx+PTx3U+LnKx/seiNR+3G19Ar8=
+golang.org/x/tools v0.0.0-20200501065659-ab2804fb9c9d/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200505023115-26f46d2f7ef8/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20200512131952-2bc93b1c0c88/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20200515010526-7d3b6ebf133d/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200616133436-c1934b75d054/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20200618134242-20370b0cb4b2/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20200729194436-6467de6f59a7/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
+golang.org/x/tools v0.0.0-20200804011535-6c149bb5ef0d/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
+golang.org/x/tools v0.0.0-20200825202427-b303f430e36d/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/api v0.4.0/go.mod h1:8k5glujaEP+g9n7WNsDg8QP6cUVNI86fCNMcbazEtwE=
 google.golang.org/api v0.7.0/go.mod h1:WtwebWUNSVBH/HAw79HIFXZNqEvBhG+Ra+ax0hx3E3M=
@@ -658,13 +760,20 @@ google.golang.org/api v0.14.0/go.mod h1:iLdEw5Ide6rF15KTC1Kkl0iskquN2gFfn9o9XIsb
 google.golang.org/api v0.15.0/go.mod h1:iLdEw5Ide6rF15KTC1Kkl0iskquN2gFfn9o9XIsbkAI=
 google.golang.org/api v0.17.0/go.mod h1:BwFmGc8tA3vsd7r/7kR8DY7iEEGSU04BFxCo5jP/sfE=
 google.golang.org/api v0.18.0/go.mod h1:BwFmGc8tA3vsd7r/7kR8DY7iEEGSU04BFxCo5jP/sfE=
+google.golang.org/api v0.19.0/go.mod h1:BwFmGc8tA3vsd7r/7kR8DY7iEEGSU04BFxCo5jP/sfE=
 google.golang.org/api v0.20.0/go.mod h1:BwFmGc8tA3vsd7r/7kR8DY7iEEGSU04BFxCo5jP/sfE=
+google.golang.org/api v0.22.0/go.mod h1:BwFmGc8tA3vsd7r/7kR8DY7iEEGSU04BFxCo5jP/sfE=
+google.golang.org/api v0.24.0/go.mod h1:lIXQywCXRcnZPGlsd8NbLnOjtAoL6em04bJ9+z0MncE=
+google.golang.org/api v0.28.0/go.mod h1:lIXQywCXRcnZPGlsd8NbLnOjtAoL6em04bJ9+z0MncE=
+google.golang.org/api v0.29.0/go.mod h1:Lcubydp8VUV7KeIHD9z2Bys/sm/vGKnG1UHuDBSrHWM=
+google.golang.org/api v0.30.0/go.mod h1:QGmEvQ87FHZNiUVJkT14jQNYJ4ZJjdRF23ZXz5138Fc=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/appengine v1.5.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/appengine v1.6.1/go.mod h1:i06prIuMbXzDqacNJfV5OdTW448YApPu5ww/cMBSeb0=
-google.golang.org/appengine v1.6.5 h1:tycE03LOZYQNhDpS27tcQdAzLCVMaj7QT2SXxebnpCM=
 google.golang.org/appengine v1.6.5/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=
+google.golang.org/appengine v1.6.6 h1:lMO5rYAqUxkmaj76jAkRUvt5JZgFymx/+Q5Mzfivuhc=
+google.golang.org/appengine v1.6.6/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=
 google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
 google.golang.org/genproto v0.0.0-20190307195333-5fe7a883aa19/go.mod h1:VzzqZJRnGkLBvHegQrXjBqPurQTc5/KpmUdxsrq26oE=
 google.golang.org/genproto v0.0.0-20190418145605-e7d98fc518a7/go.mod h1:VzzqZJRnGkLBvHegQrXjBqPurQTc5/KpmUdxsrq26oE=
@@ -682,8 +791,18 @@ google.golang.org/genproto v0.0.0-20200122232147-0452cf42e150/go.mod h1:n3cpQtvx
 google.golang.org/genproto v0.0.0-20200204135345-fa8e72b47b90/go.mod h1:GmwEX6Z4W5gMy59cAlVYjN9JhxgbQH6Gn+gFDQe2lzA=
 google.golang.org/genproto v0.0.0-20200212174721-66ed5ce911ce/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
 google.golang.org/genproto v0.0.0-20200224152610-e50cd9704f63/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
+google.golang.org/genproto v0.0.0-20200228133532-8c2c7df3a383/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
 google.golang.org/genproto v0.0.0-20200305110556-506484158171/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
+google.golang.org/genproto v0.0.0-20200312145019-da6875a35672/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
+google.golang.org/genproto v0.0.0-20200331122359-1ee6d9798940/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
+google.golang.org/genproto v0.0.0-20200430143042-b979b6f78d84/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
+google.golang.org/genproto v0.0.0-20200511104702-f5ebc3bea380/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
+google.golang.org/genproto v0.0.0-20200515170657-fc4c6c6a6587/go.mod h1:YsZOwe1myG/8QRHRsmBRE1LrgQY60beZKjly0O1fX9U=
 google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo=
+google.golang.org/genproto v0.0.0-20200618031413-b414f8b61790/go.mod h1:jDfRM7FcilCzHH/e9qn6dsT145K34l5v+OpcnNgKAAA=
+google.golang.org/genproto v0.0.0-20200729003335-053ba62fc06f/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
+google.golang.org/genproto v0.0.0-20200804131852-c06518451d9c/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
+google.golang.org/genproto v0.0.0-20200825200019-8632dd797987/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20210106152847-07624b53cd92 h1:jOTk2Z6KYaWoptUFqZ167cS8peoUPjFEXrsqfVkkCGc=
 google.golang.org/genproto v0.0.0-20210106152847-07624b53cd92/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
@@ -695,6 +814,10 @@ google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQ
 google.golang.org/grpc v1.26.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
 google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
 google.golang.org/grpc v1.27.1/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
+google.golang.org/grpc v1.28.0/go.mod h1:rpkK4SK4GF4Ach/+MFLZUBavHOvF2JJB5uozKKal+60=
+google.golang.org/grpc v1.29.1/go.mod h1:itym6AZVZYACWQqET3MqgPpjcuV5QH3BxFS3IjizoKk=
+google.golang.org/grpc v1.30.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
+google.golang.org/grpc v1.31.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
 google.golang.org/grpc v1.35.0 h1:TwIQcH3es+MojMVojxxfQ3l3OF2KzlRxML2xZq0kRo8=
 google.golang.org/grpc v1.35.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
@@ -706,9 +829,11 @@ google.golang.org/protobuf v1.22.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2
 google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 google.golang.org/protobuf v1.24.0/go.mod h1:r/3tXBNzIEhYS9I1OUVjXDlt8tc493IdKGjtUeSXeh4=
-google.golang.org/protobuf v1.25.0 h1:Ejskq+SyPohKW+1uil0JJMtmHCgJPJ/qWTxr8qp+R4c=
 google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
-gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc=
+google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
+google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
+google.golang.org/protobuf v1.28.1 h1:d0NfwRgPtno5B1Wa6L2DAG+KivqkdutMf1UhdNx175w=
+google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
@@ -726,6 +851,8 @@ gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWD
 gopkg.in/yaml.v2 v2.0.0-20170812160011-eb3733d160e7/go.mod h1:JAlM8MvJe8wmxCU4Bli9HhUf9+ttbYbLASfIpnQbh74=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.5/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
@@ -744,6 +871,7 @@ honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWh
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
 honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
+honnef.co/go/tools v0.0.1-2020.1.4/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
 k8s.io/api v0.20.0 h1:WwrYoZNM1W1aQEbyl8HNG+oWGzLpZQBlcerS9BQw9yI=
 k8s.io/api v0.20.0/go.mod h1:HyLC5l5eoS/ygQYl1BXBgFzWNlkHiAuyNAbevIn+FKg=
 k8s.io/apimachinery v0.20.0 h1:jjzbTJRXk0unNS71L7h3lxGDH/2HPxMPaQY+MjECKL8=
diff --git a/pkg/flow/collector.go b/pkg/flow/collector.go
index fa0a8b56..5a9389aa 100644
--- a/pkg/flow/collector.go
+++ b/pkg/flow/collector.go
@@ -1,13 +1,19 @@
 package flow
 
 import (
+	"bytes"
+	"encoding/gob"
 	"encoding/json"
 	"log"
 	"net/http"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/gorilla/mux"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/skupperproject/skupper/api/types"
 	"github.com/skupperproject/skupper/pkg/messaging"
+	"github.com/skupperproject/skupper/pkg/version"
 )
 
 type senderDirect struct {
@@ -34,10 +40,104 @@ type eventSource struct {
 	send            *senderDirect
 }
 
+type collectorMetrics struct {
+	info            *prometheus.GaugeVec
+	collectorOctets prometheus.Counter
+	flows           *prometheus.CounterVec
+	octets          *prometheus.CounterVec
+	httpReqsMethod  *prometheus.CounterVec
+	httpReqsResult  *prometheus.CounterVec
+	activeFlows     *prometheus.GaugeVec
+	lastAccessed    *prometheus.GaugeVec
+	flowLatency     *prometheus.HistogramVec
+	activeReconcile *prometheus.GaugeVec
+}
+
+func (fc *FlowCollector) NewMetrics(reg prometheus.Registerer) *collectorMetrics {
+	m := &collectorMetrics{
+		info: prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Name: "skupper_info",
+				Help: "Skupper deployment information",
+			},
+			[]string{"version"}),
+		collectorOctets: prometheus.NewCounter(
+			prometheus.CounterOpts{
+				Name: "collector_octets_total",
+				Help: "The total number of record octets received by collector",
+			}),
+		flows: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Name: "flows_total",
+				Help: "Total Flows",
+			},
+			[]string{"sourceSite", "destSite", "address", "protocol", "direction", "sourceProcess", "destProcess"}),
+		octets: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Name: "octets_total",
+				Help: "Total Octets",
+			},
+			[]string{"sourceSite", "destSite", "address", "protocol", "direction", "sourceProcess", "destProcess"}),
+		httpReqsMethod: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Name: "http_requests_method_total",
+				Help: "How many HTTP requests processed, partitioned by method",
+			},
+			[]string{"sourceSite", "destSite", "address", "protocol", "direction", "sourceProcess", "destProcess", "method"}),
+		httpReqsResult: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Name: "http_requests_result_total",
+				Help: "How many HTTP requests processed, partitioned by result code",
+			},
+			[]string{"sourceSite", "destSite", "address", "protocol", "direction", "sourceProcess", "destProcess", "code"}),
+		activeFlows: prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Name: "active_flows",
+				Help: "Number of flows that are currently active, partititioned by source and destination",
+			},
+			[]string{"sourceSite", "destSite", "address", "protocol", "direction", "sourceProcess", "destProcess"}),
+		lastAccessed: prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Name: "address_last_time_seconds",
+				Help: "The last time the address was served",
+			},
+			[]string{"sourceSite", "destSite", "address", "protocol", "direction", "sourceProcess", "destProcess"}),
+		flowLatency: prometheus.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Name: "flow_latency_microseconds",
+				Help: "The measure latency for the direction of flow",
+				//                 1ms,  2 ms, 5ms,  10ms,  100ms,  1s,      10s
+				Buckets: []float64{1000, 2000, 5000, 10000, 100000, 1000000, 10000000},
+			},
+			[]string{"sourceSite", "destSite", "address", "protocol", "direction", "sourceProcess", "destProcess"}),
+		activeReconcile: prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Name: "active_reconciles",
+				Help: "Number of active reconcile tasks, partitione by type",
+			},
+			[]string{"reconcileTask"}),
+	}
+	reg.MustRegister(m.info)
+	reg.MustRegister(m.collectorOctets)
+	reg.MustRegister(m.flows)
+	reg.MustRegister(m.octets)
+	reg.MustRegister(m.httpReqsMethod)
+	reg.MustRegister(m.httpReqsResult)
+	reg.MustRegister(m.activeFlows)
+	reg.MustRegister(m.lastAccessed)
+	reg.MustRegister(m.flowLatency)
+	reg.MustRegister(m.activeReconcile)
+	return m
+
+}
+
 type FlowCollector struct {
 	origin                  string
+	Collector               CollectorRecord
 	connectionFactory       messaging.ConnectionFactory
 	recordTtl               time.Duration
+	prometheusReg           prometheus.Registerer
+	metrics                 *collectorMetrics
 	beaconsIncoming         chan []interface{}
 	recordsIncoming         chan []interface{}
 	Request                 chan ApiRequest
@@ -69,7 +169,7 @@ type FlowCollector struct {
 
 func getTtl(ttl time.Duration) time.Duration {
 	if ttl == 0 {
-		return 30 * time.Minute
+		return types.DefaultFlowTimeoutDuration
 	}
 	if ttl < time.Minute {
 		return time.Minute
@@ -77,11 +177,12 @@ func getTtl(ttl time.Duration) time.Duration {
 	return ttl
 }
 
-func NewFlowCollector(origin string, connectionFactory messaging.ConnectionFactory, recordTtl time.Duration) *FlowCollector {
+func NewFlowCollector(origin string, reg prometheus.Registerer, connectionFactory messaging.ConnectionFactory, recordTtl time.Duration) *FlowCollector {
 	fc := &FlowCollector{
 		origin:                  origin,
 		connectionFactory:       connectionFactory,
 		recordTtl:               getTtl(recordTtl),
+		prometheusReg:           reg,
 		beaconsIncoming:         make(chan []interface{}),
 		recordsIncoming:         make(chan []interface{}),
 		Request:                 make(chan ApiRequest),
@@ -110,6 +211,14 @@ func NewFlowCollector(origin string, connectionFactory messaging.ConnectionFacto
 		processesToReconcile:    make(map[string]*ProcessRecord),
 		aggregatesToReconcile:   make(map[string]*FlowPairRecord),
 	}
+	fc.Collector = CollectorRecord{
+		Base: Base{
+			RecType:   recordNames[Collector],
+			Identity:  uuid.New().String(),
+			Parent:    origin,
+			StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+		},
+	}
 	return fc
 }
 
@@ -137,7 +246,6 @@ func (fc *FlowCollector) serveRecords(request ApiRequest) ApiResponse {
 		Body:   nil,
 		Status: http.StatusOK,
 	}
-
 	result, err := fc.retrieve(request)
 	if err == nil {
 		response.Body = result
@@ -147,6 +255,14 @@ func (fc *FlowCollector) serveRecords(request ApiRequest) ApiResponse {
 	return response
 }
 
+func getRealSizeOf(v interface{}) (int, error) {
+	b := new(bytes.Buffer)
+	if err := gob.NewEncoder(b).Encode(v); err != nil {
+		return 0, err
+	}
+	return b.Len(), nil
+}
+
 func (c *FlowCollector) updates(stopCh <-chan struct{}) {
 	tickerAge := time.NewTicker(10 * time.Second)
 	defer tickerAge.Stop()
@@ -203,6 +319,8 @@ func (c *FlowCollector) updates(stopCh <-chan struct{}) {
 			}
 		case recordUpdates := <-c.recordsIncoming:
 			for _, update := range recordUpdates {
+				size, _ := getRealSizeOf(update)
+				c.metrics.collectorOctets.Add(float64(size))
 				err := c.updateRecord(update)
 				if err != nil {
 					log.Println("Update record error", err.Error())
@@ -232,6 +350,8 @@ 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()
diff --git a/pkg/flow/flow_mem_driver.go b/pkg/flow/flow_mem_driver.go
index 20e37c69..128437e0 100644
--- a/pkg/flow/flow_mem_driver.go
+++ b/pkg/flow/flow_mem_driver.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/gorilla/mux"
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 func (c *FlowCollector) inferGatewayProcess(siteId string, flow FlowRecord, connector bool) error {
@@ -18,7 +19,6 @@ func (c *FlowCollector) inferGatewayProcess(siteId string, flow FlowRecord, conn
 			sourceHost = connector.DestHost
 		}
 	}
-
 	if site, ok := c.Sites[siteId]; ok {
 		groupName := *site.NameSpace + "-" + *sourceHost
 		groupIdentity := ""
@@ -67,7 +67,6 @@ func (c *FlowCollector) inferGatewayProcess(siteId string, flow FlowRecord, conn
 			}
 		}
 	}
-
 	return nil
 }
 
@@ -86,7 +85,6 @@ func (c *FlowCollector) isGatewaySite(siteId string) bool {
 }
 
 func (c *FlowCollector) getRecordSiteId(record interface{}) string {
-
 	if record == nil {
 		return ""
 	}
@@ -171,6 +169,52 @@ func (fc *FlowCollector) getFlowProtocol(flow *FlowRecord) *string {
 	return nil
 }
 
+func (fc *FlowCollector) getFlowLabels(flow *FlowRecord) map[string]string {
+	labels := make(map[string]string)
+	if flow.ProcessName != nil {
+		labels["process"] = *flow.ProcessName
+	}
+	if listener, ok := fc.Listeners[flow.Parent]; ok {
+		labels["direction"] = Incoming
+		if listener.AddressId != nil {
+			labels["addressId"] = *listener.AddressId
+		}
+		if listener.Protocol != nil {
+			labels["protocol"] = *listener.Protocol
+		}
+		return labels
+	} else if connector, ok := fc.Connectors[flow.Parent]; ok {
+		labels["direction"] = Outgoing
+		if connector.AddressId != nil {
+			labels["addressId"] = *connector.AddressId
+		}
+		if connector.Protocol != nil {
+			labels["protocol"] = *connector.Protocol
+		}
+		return labels
+	} else if l4Flow, ok := fc.Flows[flow.Parent]; ok {
+		if listener, ok := fc.Listeners[l4Flow.Parent]; ok {
+			labels["direction"] = Incoming
+			if listener.AddressId != nil {
+				labels["addressId"] = *listener.AddressId
+			}
+			if listener.Protocol != nil {
+				labels["protocol"] = *listener.Protocol
+			}
+		} else if connector, ok := fc.Connectors[l4Flow.Parent]; ok {
+			labels["direction"] = Outgoing
+			if connector.AddressId != nil {
+				labels["addressId"] = *connector.AddressId
+			}
+			if connector.Protocol != nil {
+				labels["protocol"] = *connector.Protocol
+			}
+			return labels
+		}
+	}
+	return labels
+}
+
 func (fc *FlowCollector) getFlowPlace(flow *FlowRecord) FlowPlace {
 	if _, ok := fc.Listeners[flow.Parent]; ok {
 		return clientSide
@@ -258,8 +302,26 @@ func (fc *FlowCollector) linkFlowPair(flow *FlowRecord) (*FlowPairRecord, bool)
 	} else {
 		return nil, false
 	}
+
+	fwdLabels := fc.getFlowLabels(sourceFlow)
+	revLabels := fc.getFlowLabels(destFlow)
+	if len(fwdLabels) != 4 || len(revLabels) != 4 {
+		return nil, false
+	}
+
 	sourceSiteId = fc.getRecordSiteId(*sourceFlow)
 	destSiteId = fc.getRecordSiteId(*destFlow)
+	fwdLabels["sourceSite"] = sourceSiteId
+	fwdLabels["destSite"] = destSiteId
+	fwdLabels["sourceProcess"] = *sourceFlow.ProcessName
+	fwdLabels["destProcess"] = *destFlow.ProcessName
+	delete(fwdLabels, "process")
+	revLabels["sourceSite"] = destSiteId
+	revLabels["destSite"] = sourceSiteId
+	revLabels["sourceProcess"] = *destFlow.ProcessName
+	revLabels["destProcess"] = *sourceFlow.ProcessName
+	delete(revLabels, "process")
+
 	fp := &FlowPairRecord{
 		Base: Base{
 			RecType:   recordNames[FlowPair],
@@ -283,9 +345,87 @@ func (fc *FlowCollector) linkFlowPair(flow *FlowRecord) (*FlowPairRecord, bool)
 		return fp, ok
 	}
 	fp.FlowTrace = fc.annotateFlowTrace(destFlow)
+
+	// setup flow metrics inc flow count, set octets, assign metric to flow, etc.
+	addressId := fwdLabels["addressId"]
+	if va, ok := fc.VanAddresses[addressId]; ok {
+		fwdLabels["address"] = va.Name
+		delete(fwdLabels, "addressId")
+		revLabels["address"] = va.Name
+		delete(revLabels, "addressId")
+		err := fc.setupFlowMetrics(va, sourceFlow, fwdLabels)
+		if err != nil {
+			log.Println("setup metric error", err.Error())
+		}
+		err = fc.setupFlowMetrics(va, destFlow, revLabels)
+		if err != nil {
+			log.Println("setup metric error", err.Error())
+		}
+	}
+
 	return fp, ok
 }
 
+func (fc *FlowCollector) deleteRecord(record interface{}) error {
+	// TODO: log the record
+	if record == nil {
+		return fmt.Errorf("No record to delete")
+	}
+	switch record.(type) {
+	case SiteRecord:
+		if site, ok := record.(SiteRecord); ok {
+			delete(fc.Sites, site.Identity)
+		}
+	case HostRecord:
+		if host, ok := record.(HostRecord); ok {
+			delete(fc.Hosts, host.Identity)
+		}
+	case RouterRecord:
+		if router, ok := record.(RouterRecord); ok {
+			delete(fc.Routers, router.Identity)
+		}
+	case LinkRecord:
+		if link, ok := record.(LinkRecord); ok {
+			delete(fc.Links, link.Identity)
+		}
+	case ListenerRecord:
+		if listener, ok := record.(ListenerRecord); ok {
+			delete(fc.Listeners, listener.Identity)
+		}
+	case ConnectorRecord:
+		if connector, ok := record.(ConnectorRecord); ok {
+			delete(fc.Connectors, connector.Identity)
+		}
+	case FlowRecord:
+		if flow, ok := record.(FlowRecord); ok {
+			delete(fc.Flows, flow.Identity)
+		}
+	case FlowPairRecord:
+		if flowPair, ok := record.(FlowPairRecord); ok {
+			delete(fc.FlowPairs, flowPair.Identity)
+		}
+	case ProcessRecord:
+		if process, ok := record.(ProcessRecord); ok {
+			delete(fc.Processes, process.Identity)
+		}
+	case ProcessGroupRecord:
+		if processGroup, ok := record.(ProcessGroupRecord); ok {
+			delete(fc.ProcessGroups, processGroup.Identity)
+		}
+	case FlowAggregateRecord:
+		if aggregate, ok := record.(FlowAggregateRecord); ok {
+			delete(fc.FlowAggregates, aggregate.Identity)
+		}
+	case VanAddressRecord:
+		if va, ok := record.(VanAddressRecord); ok {
+			delete(fc.VanAddresses, va.Identity)
+		}
+	default:
+		return fmt.Errorf("Unknown record type to delete")
+	}
+	return nil
+}
+
 func (fc *FlowCollector) updateRecord(record interface{}) error {
 	switch record.(type) {
 	case HeartbeatRecord:
@@ -300,9 +440,22 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 		}
 	case SiteRecord:
 		if site, ok := record.(SiteRecord); ok {
-			if site.StartTime > 0 && site.EndTime == 0 {
-				if current, ok := fc.Sites[site.Identity]; !ok {
+			if current, ok := fc.Sites[site.Identity]; !ok {
+				if site.StartTime > 0 && site.EndTime == 0 {
 					fc.Sites[site.Identity] = &site
+				}
+			} else {
+				if site.EndTime > 0 {
+					current.EndTime = site.EndTime
+					for _, aggregate := range fc.FlowAggregates {
+						if aggregate.PairType == recordNames[Site] {
+							if current.Identity == *aggregate.SourceId || current.Identity == *aggregate.DestinationId {
+								aggregate.EndTime = current.EndTime
+								fc.deleteRecord(*aggregate)
+							}
+						}
+					}
+					fc.deleteRecord(*current)
 				} else {
 					*current = site
 				}
@@ -320,8 +473,8 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 		}
 	case RouterRecord:
 		if router, ok := record.(RouterRecord); ok {
-			if router.StartTime > 0 && router.EndTime == 0 {
-				if _, ok := fc.Routers[router.Identity]; !ok {
+			if current, ok := fc.Routers[router.Identity]; !ok {
+				if router.StartTime > 0 && router.EndTime == 0 {
 					fc.Routers[router.Identity] = &router
 					if router.Parent != "" {
 						if _, ok := fc.Sites[router.Parent]; !ok {
@@ -329,15 +482,28 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 						}
 					}
 				}
+			} else {
+				if router.EndTime > 0 {
+					current.EndTime = router.EndTime
+					fc.deleteRecord(*current)
+				} else if router.Parent != "" && current.Parent == "" {
+					current.Parent = router.Parent
+					if _, ok := fc.Sites[current.Parent]; !ok {
+						fc.routersToSiteReconcile[current.Identity] = current.Parent
+					}
+				}
 			}
 		}
 	case LinkRecord:
 		if link, ok := record.(LinkRecord); ok {
 			if current, ok := fc.Links[link.Identity]; !ok {
-				fc.Links[link.Identity] = &link
+				if link.StartTime > 0 && link.EndTime == 0 {
+					fc.Links[link.Identity] = &link
+				}
 			} else {
 				if link.EndTime > 0 {
 					current.EndTime = link.EndTime
+					fc.deleteRecord(*current)
 				}
 			}
 		}
@@ -347,15 +513,15 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 				if listener.StartTime > 0 && listener.EndTime == 0 {
 					fc.Listeners[listener.Identity] = &listener
 					if listener.Address != nil {
-						var addr *VanAddressRecord
+						var va *VanAddressRecord
 						for _, y := range fc.VanAddresses {
 							if y.Name == *listener.Address {
-								addr = y
+								va = y
 								break
 							}
 						}
-						if addr == nil {
-							va := &VanAddressRecord{
+						if va == nil {
+							va = &VanAddressRecord{
 								Base: Base{
 									RecType:   attributeNames[VanAddress],
 									Identity:  uuid.New().String(),
@@ -364,13 +530,34 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 								Name:     *listener.Address,
 								Protocol: *listener.Protocol,
 							}
+							va.flowCount = make(map[metricKey]prometheus.Counter)
+							va.activeFlowCount = make(map[metricKey]prometheus.Gauge)
+							va.octetCount = make(map[metricKey]prometheus.Counter)
+							va.lastAccessed = make(map[metricKey]prometheus.Gauge)
+							va.flowLatency = make(map[metricKey]prometheus.Observer)
 							fc.VanAddresses[va.Identity] = va
 						}
+						listener.AddressId = &va.Identity
 					}
 				}
 			} else {
 				if current.EndTime == 0 && listener.EndTime > 0 {
 					current.EndTime = listener.EndTime
+					if current.AddressId != nil {
+						count := 0
+						for id, l := range fc.Listeners {
+							if id != current.Identity && l.AddressId != nil && *l.AddressId == *current.AddressId {
+								count++
+							}
+						}
+						if count == 0 {
+							if va, ok := fc.VanAddresses[*current.AddressId]; ok {
+								va.EndTime = listener.EndTime
+								fc.deleteRecord(*va)
+							}
+						}
+					}
+					fc.deleteRecord(*current)
 				} else {
 					if current.Parent == "" && listener.Parent != "" {
 						current.Parent = listener.Parent
@@ -389,16 +576,16 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 					}
 					if current.Address == nil && listener.Address != nil {
 						current.Address = listener.Address
-						var addr *VanAddressRecord
+						var va *VanAddressRecord
 						for _, y := range fc.VanAddresses {
 							if y.Name == *listener.Address {
-								addr = y
+								va = y
 								break
 							}
 						}
-						if addr == nil {
+						if va == nil {
 							t := time.Now()
-							va := &VanAddressRecord{
+							va = &VanAddressRecord{
 								Base: Base{
 									RecType:   attributeNames[VanAddress],
 									Identity:  uuid.New().String(),
@@ -407,8 +594,14 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 								Name:     *listener.Address,
 								Protocol: *listener.Protocol,
 							}
+							va.flowCount = make(map[metricKey]prometheus.Counter)
+							va.activeFlowCount = make(map[metricKey]prometheus.Gauge)
+							va.octetCount = make(map[metricKey]prometheus.Counter)
+							va.lastAccessed = make(map[metricKey]prometheus.Gauge)
+							va.flowLatency = make(map[metricKey]prometheus.Observer)
 							fc.VanAddresses[va.Identity] = va
 						}
+						listener.AddressId = &va.Identity
 					}
 				}
 			}
@@ -420,16 +613,16 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 					fc.Connectors[connector.Identity] = &connector
 					if connector.Parent != "" {
 						if connector.Address != nil {
-							var addr *VanAddressRecord
+							var va *VanAddressRecord
 							for _, y := range fc.VanAddresses {
 								if y.Name == *connector.Address {
-									addr = y
+									va = y
 									break
 								}
 							}
-							if addr == nil {
+							if va == nil {
 								t := time.Now()
-								va := &VanAddressRecord{
+								va = &VanAddressRecord{
 									Base: Base{
 										RecType:   attributeNames[VanAddress],
 										Identity:  uuid.New().String(),
@@ -438,8 +631,14 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 									Name:     *connector.Address,
 									Protocol: *connector.Protocol,
 								}
+								va.flowCount = make(map[metricKey]prometheus.Counter)
+								va.activeFlowCount = make(map[metricKey]prometheus.Gauge)
+								va.octetCount = make(map[metricKey]prometheus.Counter)
+								va.lastAccessed = make(map[metricKey]prometheus.Gauge)
+								va.flowLatency = make(map[metricKey]prometheus.Observer)
 								fc.VanAddresses[va.Identity] = va
 							}
+							connector.AddressId = &va.Identity
 						}
 					}
 					fc.connectorsToReconcile[connector.Identity] = connector.Identity
@@ -452,6 +651,9 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 							process.connector = nil
 						}
 					}
+					// Note a new connector can create an address but does not delete it
+					// removal of last listener will delete the address
+					fc.deleteRecord(*current)
 				} else {
 					if current.Parent == "" && connector.Parent != "" {
 						current.Parent = connector.Parent
@@ -473,16 +675,16 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 					}
 					if current.Address == nil && connector.Address != nil {
 						current.Address = connector.Address
-						var addr *VanAddressRecord
+						var va *VanAddressRecord
 						for _, y := range fc.VanAddresses {
 							if y.Name == *connector.Address {
-								addr = y
+								va = y
 								break
 							}
 						}
-						if addr == nil {
+						if va == nil {
 							t := time.Now()
-							va := &VanAddressRecord{
+							va = &VanAddressRecord{
 								Base: Base{
 									RecType:   attributeNames[VanAddress],
 									Identity:  uuid.New().String(),
@@ -491,8 +693,14 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 								Name:     *connector.Address,
 								Protocol: *connector.Protocol,
 							}
+							va.flowCount = make(map[metricKey]prometheus.Counter)
+							va.activeFlowCount = make(map[metricKey]prometheus.Gauge)
+							va.octetCount = make(map[metricKey]prometheus.Counter)
+							va.lastAccessed = make(map[metricKey]prometheus.Gauge)
+							va.flowLatency = make(map[metricKey]prometheus.Observer)
 							fc.VanAddresses[va.Identity] = va
 						}
+						connector.AddressId = &va.Identity
 					}
 				}
 			}
@@ -511,14 +719,14 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 						flow.Protocol = fc.getFlowProtocol(&flow)
 						flow.Place = fc.getFlowPlace(&flow)
 						if listener, ok := fc.Listeners[flow.Parent]; ok {
-							// TODO: workaround for gateway
 							siteId := fc.getRecordSiteId(*listener)
+							// TODO: workaround for gateway
 							if fc.isGatewaySite(siteId) {
 								fc.inferGatewayProcess(siteId, flow, false)
 							}
 						} else if connector, ok := fc.Connectors[flow.Parent]; ok {
-							// TODO: workaround for gateway
 							siteId := fc.getRecordSiteId(*connector)
+							// TODO: workaround for gateway
 							if fc.isGatewaySite(siteId) {
 								fc.inferGatewayProcess(siteId, flow, true)
 								fc.connectorsToReconcile[connector.Identity] = connector.Identity
@@ -558,6 +766,10 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 				}
 				if flow.Octets != nil {
 					current.Octets = flow.Octets
+					if current.octetMetric != nil {
+						current.octetMetric.Add(float64(*current.Octets - current.lastOctets))
+					}
+					current.lastOctets = *current.Octets
 				}
 				if flow.OctetRate != nil {
 					current.OctetRate = flow.OctetRate
@@ -596,8 +808,11 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 					// this should trigger pairing
 					current.CounterFlow = flow.CounterFlow
 				}
-				if flow.EndTime > 0 {
+				if flow.EndTime > 0 && current.EndTime == 0 {
 					current.EndTime = flow.EndTime
+					if current.activeFlowMetric != nil {
+						current.activeFlowMetric.Dec()
+					}
 					if fc.getFlowPlace(current) == clientSide {
 						if flowpair, ok := fc.FlowPairs["fp-"+current.Identity]; ok {
 							flowpair.EndTime = current.EndTime
@@ -618,7 +833,7 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 					fc.Processes[process.Identity] = &process
 				}
 				for _, pg := range fc.ProcessGroups {
-					if *process.GroupName == *pg.Name {
+					if pg.EndTime == 0 && *process.GroupName == *pg.Name {
 						process.GroupIdentity = &pg.Identity
 						break
 					}
@@ -639,15 +854,54 @@ func (fc *FlowCollector) updateRecord(record interface{}) error {
 			} else {
 				if process.EndTime > 0 {
 					current.EndTime = process.EndTime
+					// check if there are any process pairs active
+					for _, aggregate := range fc.FlowAggregates {
+						if aggregate.PairType == recordNames[Process] {
+							if current.Identity == *aggregate.SourceId || current.Identity == *aggregate.DestinationId {
+								aggregate.EndTime = current.EndTime
+								fc.deleteRecord(*aggregate)
+							}
+						}
+					}
+
+					if current.GroupIdentity != nil {
+						count := 0
+						for id, p := range fc.Processes {
+							if id != current.Identity && *p.GroupIdentity == *current.GroupIdentity {
+								count++
+							}
+						}
+						if count == 0 {
+							if processGroup, ok := fc.ProcessGroups[*current.GroupIdentity]; ok {
+								processGroup.EndTime = process.EndTime
+								fc.updateRecord(*processGroup)
+							}
+						}
+					}
+					fc.deleteRecord(*current)
 				}
 			}
 		}
 	case ProcessGroupRecord:
 		if processGroup, ok := record.(ProcessGroupRecord); ok {
-			if processGroup.StartTime > 0 && processGroup.EndTime == 0 {
-				if _, ok := fc.ProcessGroups[processGroup.Identity]; !ok {
+			if current, ok := fc.ProcessGroups[processGroup.Identity]; !ok {
+				if processGroup.StartTime > 0 && processGroup.EndTime == 0 {
 					fc.ProcessGroups[processGroup.Identity] = &processGroup
 				}
+			} else {
+				if processGroup.EndTime > 0 {
+					current.EndTime = processGroup.EndTime
+					// check if there are an processgroup pairs active
+					for _, aggregate := range fc.FlowAggregates {
+						if aggregate.PairType == recordNames[ProcessGroup] {
+							if current.Identity == *aggregate.SourceId || current.Identity == *aggregate.DestinationId {
+								aggregate.EndTime = current.EndTime
+								fc.deleteRecord(*aggregate)
+							}
+						}
+					}
+					fc.deleteRecord(*current)
+				}
 			}
 		}
 	default:
@@ -680,7 +934,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 			sites := []SiteRecord{}
 			for _, site := range fc.Sites {
 				if filterRecord(*site, p.QueryParams.Filter) && site.Base.TimeRangeValid(queryParams) {
-					fc.getSiteMetrics(site)
 					sites = append(sites, *site)
 				}
 			}
@@ -689,7 +942,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "item":
 			if id, ok := vars["id"]; ok {
 				if site, ok := fc.Sites[id]; ok {
-					fc.getSiteMetrics(site)
 					p.Count = 1
 					p.Results = site
 				}
@@ -702,7 +954,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 						if process.Parent == site.Identity {
 							p.TotalCount++
 							if filterRecord(*process, p.QueryParams.Filter) && process.Base.TimeRangeValid(queryParams) {
-								fc.getProcessMetrics(process)
 								processes = append(processes, *process)
 							}
 						}
@@ -989,7 +1240,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 				if connector, ok := fc.Connectors[id]; ok {
 					if connector.process != nil {
 						if process, ok := fc.Processes[*connector.process]; ok {
-							fc.getProcessMetrics(process)
 							p.Count = 1
 							p.Results = *process
 						}
@@ -1108,7 +1358,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 						}
 					}
 					for _, process := range unique {
-						fc.getProcessMetrics(process)
 						processes = append(processes, *process)
 					}
 				}
@@ -1152,7 +1401,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 			processes := []ProcessRecord{}
 			for _, process := range fc.Processes {
 				if filterRecord(*process, p.QueryParams.Filter) && process.Base.TimeRangeValid(queryParams) {
-					fc.getProcessMetrics(process)
 					processes = append(processes, *process)
 				}
 			}
@@ -1161,7 +1409,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "item":
 			if id, ok := vars["id"]; ok {
 				if process, ok := fc.Processes[id]; ok {
-					fc.getProcessMetrics(process)
 					p.Count = 1
 					p.Results = process
 				}
@@ -1218,7 +1465,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 			for _, processGroup := range fc.ProcessGroups {
 				p.TotalCount++
 				if filterRecord(*processGroup, p.QueryParams.Filter) && processGroup.Base.TimeRangeValid(queryParams) {
-					fc.getProcessGroupMetrics(processGroup)
 					processGroups = append(processGroups, *processGroup)
 				}
 			}
@@ -1226,7 +1472,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "item":
 			if id, ok := vars["id"]; ok {
 				if processGroup, ok := fc.ProcessGroups[id]; ok {
-					fc.getProcessGroupMetrics(processGroup)
 					p.Count = 1
 					p.Results = processGroup
 				}
@@ -1239,7 +1484,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 						if *process.GroupIdentity == processGroup.Identity {
 							p.TotalCount++
 							if filterRecord(*process, p.QueryParams.Filter) && process.Base.TimeRangeValid(queryParams) {
-								fc.getProcessMetrics(process)
 								processes = append(processes, *process)
 							}
 						}
@@ -1271,7 +1515,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 				if flow, ok := fc.Flows[id]; ok {
 					if flow.Process != nil {
 						if process, ok := fc.Processes[*flow.Process]; ok {
-							fc.getProcessMetrics(process)
 							p.Count = 1
 							p.Results = process
 						}
@@ -1312,7 +1555,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 						sourceId == "" && destinationId == *aggregate.DestinationId ||
 						sourceId == *aggregate.SourceId && destinationId == *aggregate.DestinationId {
 						if filterRecord(*aggregate, p.QueryParams.Filter) {
-							fc.getFlowAggregateMetrics(Site, aggregate.Identity)
 							aggregates = append(aggregates, *aggregate)
 						}
 					}
@@ -1343,7 +1585,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 						sourceId == "" && destinationId == *aggregate.DestinationId ||
 						sourceId == *aggregate.SourceId && destinationId == *aggregate.DestinationId {
 						if filterRecord(*aggregate, p.QueryParams.Filter) {
-							fc.getFlowAggregateMetrics(ProcessGroup, aggregate.Identity)
 							aggregates = append(aggregates, *aggregate)
 						}
 					}
@@ -1374,7 +1615,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 						sourceId == "" && destinationId == *aggregate.DestinationId ||
 						sourceId == *aggregate.SourceId && destinationId == *aggregate.DestinationId {
 						if filterRecord(*aggregate, p.QueryParams.Filter) {
-							fc.getFlowAggregateMetrics(Process, aggregate.Identity)
 							aggregates = append(aggregates, *aggregate)
 						}
 					}
@@ -1404,7 +1644,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 						sourceId == *aggregate.SourceId && destinationId == "" ||
 						sourceId == "" && destinationId == *aggregate.DestinationId ||
 						sourceId == *aggregate.SourceId && destinationId == *aggregate.DestinationId {
-						fc.getFlowAggregateMetrics(Site, aggregate.Identity)
 						aggregates = append(aggregates, *aggregate)
 					}
 				}
@@ -1428,7 +1667,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 						sourceId == *aggregate.SourceId && destinationId == "" ||
 						sourceId == "" && destinationId == *aggregate.DestinationId ||
 						sourceId == *aggregate.SourceId && destinationId == *aggregate.DestinationId {
-						fc.getFlowAggregateMetrics(Process, aggregate.Identity)
 						aggregates = append(aggregates, *aggregate)
 					}
 				}
@@ -1452,7 +1690,6 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 						sourceId == *aggregate.SourceId && destinationId == "" ||
 						sourceId == "" && destinationId == *aggregate.DestinationId ||
 						sourceId == *aggregate.SourceId && destinationId == *aggregate.DestinationId {
-						fc.getFlowAggregateMetrics(ProcessGroup, aggregate.Identity)
 						aggregates = append(aggregates, *aggregate)
 					}
 				}
@@ -1487,6 +1724,18 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 				}
 			}
 		}
+	case Collector:
+		// TODO: emit and collect Collector records
+		switch request.HandlerName {
+		case "list":
+			collectors := []CollectorRecord{}
+			collectors = append(collectors, fc.Collector)
+			p.TotalCount = len(collectors)
+			retrieveError = sortAndSlice(collectors, &p)
+		case "item":
+			p.Count = 1
+			p.Results = &fc.Collector
+		}
 	default:
 		log.Println("Unrecognize record request", request.RecordType)
 	}
@@ -1514,7 +1763,115 @@ func (fc *FlowCollector) getFlowProcess(id string) (ProcessRecord, bool) {
 	return ProcessRecord{}, false
 }
 
+func (fc *FlowCollector) setupFlowMetrics(va *VanAddressRecord, flow *FlowRecord, metricLabel prometheus.Labels) error {
+	var flowMetric prometheus.Counter
+	var octetMetric prometheus.Counter
+	var flowLatencyMetric prometheus.Observer
+	var lastAccessedMetric prometheus.Gauge
+	var activeFlowMetric prometheus.Gauge
+	var err error
+	var ok bool
+
+	key := metricKey{}
+	if key.sourceSite, ok = metricLabel["sourceSite"]; !ok {
+		return fmt.Errorf("Metric label missing source site key")
+	}
+	if key.sourceProcess, ok = metricLabel["sourceProcess"]; !ok {
+		return fmt.Errorf("Metric label soruce process key")
+	}
+	if key.destSite, ok = metricLabel["destSite"]; !ok {
+		return fmt.Errorf("Metric label missing dest site key")
+	}
+	if key.destProcess, ok = metricLabel["destProcess"]; !ok {
+		return fmt.Errorf("Metric label missing dest process key")
+	}
+
+	if flowMetric, ok = va.flowCount[key]; !ok {
+		flowMetric, err = fc.metrics.flows.GetMetricWith(metricLabel)
+		if err != nil {
+			return err
+		} else {
+			va.flowCount[key] = flowMetric
+		}
+	}
+	flowMetric.Inc()
+
+	if octetMetric, ok = va.octetCount[key]; !ok {
+		octetMetric, err = fc.metrics.octets.GetMetricWith(metricLabel)
+		if err != nil {
+			return err
+		} else {
+			va.octetCount[key] = octetMetric
+		}
+	}
+	flow.octetMetric = octetMetric
+	if flow.Octets != nil {
+		octetMetric.Add(float64(*flow.Octets))
+		flow.lastOctets = *flow.Octets
+	}
+
+	if flowLatencyMetric, ok = va.flowLatency[key]; !ok {
+		flowLatencyMetric, err = fc.metrics.flowLatency.GetMetricWith(metricLabel)
+		if err != nil {
+			return err
+		} else {
+			va.flowLatency[key] = flowLatencyMetric
+		}
+	}
+	if flow.Latency != nil {
+		flowLatencyMetric.Observe(float64(*flow.Latency))
+	}
+
+	if lastAccessedMetric, ok = va.lastAccessed[key]; !ok {
+		lastAccessedMetric, err = fc.metrics.lastAccessed.GetMetricWith(metricLabel)
+		if err != nil {
+			return err
+		} else {
+			va.lastAccessed[key] = lastAccessedMetric
+		}
+	}
+	lastAccessedMetric.Set(float64((flow.StartTime / uint64(time.Microsecond))))
+
+	if activeFlowMetric, ok = va.activeFlowCount[key]; !ok {
+		activeFlowMetric, err = fc.metrics.activeFlows.GetMetricWith(metricLabel)
+		if err != nil {
+			return err
+		} else {
+			va.activeFlowCount[key] = activeFlowMetric
+		}
+	}
+	flow.activeFlowMetric = activeFlowMetric
+	if flow.EndTime == 0 {
+		activeFlowMetric.Inc()
+	}
+
+	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()
+			}
+		} else if direction == Outgoing && flow.Result != nil {
+			metricLabel["code"] = *flow.Result
+			httpReqsResult, err := fc.metrics.httpReqsResult.GetMetricWith(metricLabel)
+			if err != nil {
+				return err
+			} else {
+				httpReqsResult.Inc()
+			}
+		}
+	}
+	return nil
+}
+
 func (fc *FlowCollector) reconcileRecords() error {
+	m, err := fc.metrics.activeReconcile.GetMetricWith(prometheus.Labels{"reconcileTask": "routersToSite"})
+	if err == nil {
+		m.Set(float64(len(fc.routersToSiteReconcile)))
+	}
 	for routerId, siteId := range fc.routersToSiteReconcile {
 		// TODO: This is temporary workaround until gateway or CE site emits events
 		parts := strings.Split(siteId, "_")
@@ -1537,25 +1894,34 @@ func (fc *FlowCollector) reconcileRecords() error {
 		}
 		delete(fc.routersToSiteReconcile, routerId)
 	}
+	m, err = fc.metrics.activeReconcile.GetMetricWith(prometheus.Labels{"reconcileTask": "flowsToProcess"})
+	if err == nil {
+		m.Set(float64(len(fc.flowsToProcessReconcile)))
+	}
 	for _, flowId := range fc.flowsToProcessReconcile {
 		if flow, ok := fc.Flows[flowId]; ok {
+			siteId := fc.getRecordSiteId(*flow)
+			if siteId == "" {
+				continue
+			}
 			if flow.SourceHost != nil {
 				if connector, ok := fc.Connectors[flow.Parent]; ok {
-					if connector.process != nil {
+					if connector.process != nil && connector.AddressId != nil {
 						flow.Process = connector.process
 						if process, ok := fc.Processes[*flow.Process]; ok {
 							flow.ProcessName = process.Name
 						}
 						delete(fc.flowsToProcessReconcile, flowId)
 					}
-				} else if _, ok := fc.Listeners[flow.Parent]; ok {
-					siteId := fc.getRecordSiteId(*flow)
-					for _, process := range fc.Processes {
-						if siteId == process.Parent && process.SourceHost != nil {
-							if *flow.SourceHost == *process.SourceHost {
-								flow.Process = &process.Identity
-								flow.ProcessName = process.Name
-								delete(fc.flowsToProcessReconcile, flowId)
+				} else if listener, ok := fc.Listeners[flow.Parent]; ok {
+					if listener.AddressId != nil {
+						for _, process := range fc.Processes {
+							if siteId == process.Parent && process.SourceHost != nil {
+								if *flow.SourceHost == *process.SourceHost {
+									flow.Process = &process.Identity
+									flow.ProcessName = process.Name
+									delete(fc.flowsToProcessReconcile, flowId)
+								}
 							}
 						}
 					}
@@ -1571,6 +1937,10 @@ func (fc *FlowCollector) reconcileRecords() error {
 			delete(fc.flowsToProcessReconcile, flowId)
 		}
 	}
+	m, err = fc.metrics.activeReconcile.GetMetricWith(prometheus.Labels{"reconcileTask": "flowsToPair"})
+	if err == nil {
+		m.Set(float64(len(fc.flowsToPairReconcile)))
+	}
 	for reverseId, forwardId := range fc.flowsToPairReconcile {
 		if reverseFlow, ok := fc.Flows[reverseId]; ok {
 			if forwardFlow, ok := fc.Flows[forwardId]; ok {
@@ -1585,6 +1955,10 @@ 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 {
 		if connector, ok := fc.Connectors[connId]; ok {
 			if connector.EndTime > 0 {
@@ -1622,6 +1996,10 @@ func (fc *FlowCollector) reconcileRecords() error {
 			delete(fc.connectorsToReconcile, connId)
 		}
 	}
+	m, err = fc.metrics.activeReconcile.GetMetricWith(prometheus.Labels{"reconcileTask": "pairToAggregate"})
+	if err == nil {
+		m.Set(float64(len(fc.aggregatesToReconcile)))
+	}
 	for flowPairId, flowPair := range fc.aggregatesToReconcile {
 		if flowPair.ForwardFlow != nil && flowPair.CounterFlow != nil {
 			ffp, ffpOk := fc.getFlowProcess(flowPair.ForwardFlow.Identity)
@@ -1700,55 +2078,12 @@ func (fc *FlowCollector) reconcileRecords() 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 {
-			delete(fc.Flows, flowId)
-			delete(fc.FlowPairs, "fp-"+flowId)
-		}
-	}
-
-	return nil
-}
-
-func (fc *FlowCollector) getSiteMetrics(site *SiteRecord) error {
-	//TODO: active flows, flows since, flows in interval
-
-	octetsSent := uint64(0)
-	octetsSentRate := uint64(0)
-	octetsReceived := uint64(0)
-	octetsReceivedRate := uint64(0)
-	for _, fp := range fc.FlowPairs {
-
-		if site.Identity == fp.SourceSiteId {
-			if fp.ForwardFlow.Octets != nil {
-				octetsSent += *fp.ForwardFlow.Octets
-			}
-			if fp.ForwardFlow.OctetRate != nil {
-				octetsSentRate = octetsSentRate + *fp.ForwardFlow.OctetRate
-			}
-			if fp.CounterFlow.Octets != nil {
-				octetsReceived += *fp.CounterFlow.Octets
-			}
-			if fp.CounterFlow.OctetRate != nil {
-				octetsReceivedRate = octetsReceivedRate + *fp.CounterFlow.OctetRate
-			}
-		} else if site.Identity == fp.DestinationSiteId {
-			if fp.ForwardFlow.Octets != nil {
-				octetsReceived += *fp.ForwardFlow.Octets
-			}
-			if fp.ForwardFlow.OctetRate != nil {
-				octetsReceivedRate = octetsSentRate + *fp.ForwardFlow.OctetRate
-			}
-			if fp.CounterFlow.Octets != nil {
-				octetsSent += *fp.CounterFlow.Octets
-			}
-			if fp.CounterFlow.OctetRate != nil {
-				octetsSentRate = octetsReceivedRate + *fp.CounterFlow.OctetRate
+			fc.deleteRecord(*flow)
+			if flowPair, ok := fc.FlowPairs["fp-"+flowId]; ok {
+				fc.deleteRecord(*flowPair)
 			}
 		}
 	}
-	site.OctetsSent = octetsSent
-	site.OctetsSentRate = octetsSentRate
-	site.OctetsReceived = octetsReceived
-	site.OctetsReceivedRate = octetsReceivedRate
 
 	return nil
 }
@@ -1756,245 +2091,18 @@ func (fc *FlowCollector) getSiteMetrics(site *SiteRecord) error {
 func (fc *FlowCollector) getAddressMetrics(addr *VanAddressRecord) error {
 	listenerCount := 0
 	connectorCount := 0
-	totalFlows := 0
-	currentFlows := 0
-	sourceOctets := uint64(0)
-	destOctets := uint64(0)
-	for listenerId, listener := range fc.Listeners {
-		if *listener.Address == addr.Name {
+	for _, listener := range fc.Listeners {
+		if listener.Address != nil && *listener.Address == addr.Name {
 			listenerCount++
-			for _, flow := range fc.Flows {
-				if flow.Parent == listenerId && *listener.Protocol == "tcp" && flow.StartTime != 0 {
-					totalFlows++
-					if flow.EndTime == 0 {
-						currentFlows++
-					}
-					if flow.Octets != nil {
-						sourceOctets += *flow.Octets
-					}
-				} else if l4Flow, ok := fc.Flows[flow.Parent]; ok {
-					if l4Flow.Parent == listenerId && flow.StartTime != 0 {
-						totalFlows++
-						if flow.EndTime == 0 {
-							currentFlows++
-						}
-						if flow.Octets != nil {
-							sourceOctets += *flow.Octets
-						}
-					}
-				}
-			}
 		}
 	}
-	for connectorId, connector := range fc.Connectors {
-		if *connector.Address == addr.Name {
+	for _, connector := range fc.Connectors {
+		if connector.Address != nil && *connector.Address == addr.Name {
 			connectorCount++
-			for _, flow := range fc.Flows {
-				if flow.Parent == connectorId && *connector.Protocol == "tcp" && flow.StartTime != 0 {
-					totalFlows++
-					if flow.EndTime == 0 {
-						currentFlows++
-					}
-					if flow.Octets != nil {
-						destOctets += *flow.Octets
-					}
-				} else if l4Flow, ok := fc.Flows[flow.Parent]; ok {
-					if l4Flow.Parent == connectorId && flow.StartTime != 0 {
-						totalFlows++
-						if flow.EndTime == 0 {
-							currentFlows++
-						}
-						if flow.Octets != nil {
-							destOctets += *flow.Octets
-						}
-					}
-				}
-			}
 		}
 	}
 	addr.ListenerCount = listenerCount
 	addr.ConnectorCount = connectorCount
-	addr.TotalFlows = totalFlows
-	addr.CurrentFlows = currentFlows
-	addr.SourceOctets = sourceOctets
-	addr.DestOctets = destOctets
-
-	return nil
-}
-
-func (fc *FlowCollector) getProcessGroupMetrics(pg *ProcessGroupRecord) error {
-	octetsSent := uint64(0)
-	octetsSentRate := uint64(0)
-	octetsReceived := uint64(0)
-	octetsReceivedRate := uint64(0)
-	for _, fp := range fc.FlowPairs {
-		if fp.ProcessGroupAggregateId != nil {
-			parts := strings.Split(*fp.ProcessGroupAggregateId, "-to-")
-			if len(parts) == 2 {
-				if pg.Identity == parts[0] {
-					if fp.ForwardFlow.Octets != nil {
-						octetsSent += *fp.ForwardFlow.Octets
-					}
-					if fp.ForwardFlow.OctetRate != nil {
-						octetsSentRate = octetsSentRate + *fp.ForwardFlow.OctetRate
-					}
-					if fp.CounterFlow.Octets != nil {
-						octetsReceived += *fp.CounterFlow.Octets
-					}
-					if fp.CounterFlow.OctetRate != nil {
-						octetsReceivedRate = octetsReceivedRate + *fp.CounterFlow.OctetRate
-					}
-				} else if pg.Identity == parts[1] {
-					if fp.ForwardFlow.Octets != nil {
-						octetsReceived += *fp.ForwardFlow.Octets
-					}
-					if fp.ForwardFlow.OctetRate != nil {
-						octetsReceivedRate = octetsReceivedRate + *fp.ForwardFlow.OctetRate
-					}
-					if fp.CounterFlow.Octets != nil {
-						octetsSent += *fp.CounterFlow.Octets
-					}
-					if fp.CounterFlow.OctetRate != nil {
-						octetsSentRate = octetsSentRate + *fp.CounterFlow.OctetRate
-					}
-				}
-			}
-		}
-	}
-	pg.OctetsSent = octetsSent
-	pg.OctetsSentRate = octetsSentRate
-	pg.OctetsReceived = octetsReceived
-	pg.OctetsReceivedRate = octetsReceivedRate
-
-	return nil
-}
-
-func (fc *FlowCollector) getProcessMetrics(proc *ProcessRecord) error {
-	//TODO: active flows, flows in time interval, flows since time
-
-	octetsSent := uint64(0)
-	octetsSentRate := uint64(0)
-	octetsReceived := uint64(0)
-	octetsReceivedRate := uint64(0)
-	for _, fp := range fc.FlowPairs {
-		if fp.ProcessAggregateId != nil {
-			parts := strings.Split(*fp.ProcessAggregateId, "-to-")
-			if len(parts) == 2 {
-				if proc.Identity == parts[0] {
-					if fp.ForwardFlow.Octets != nil {
-						octetsSent += *fp.ForwardFlow.Octets
-					}
-					if fp.ForwardFlow.OctetRate != nil {
-						octetsSentRate = octetsSentRate + *fp.ForwardFlow.OctetRate
-					}
-					if fp.CounterFlow.Octets != nil {
-						octetsReceived += *fp.CounterFlow.Octets
-					}
-					if fp.CounterFlow.OctetRate != nil {
-						octetsReceivedRate = octetsReceivedRate + *fp.CounterFlow.OctetRate
-					}
-				} else if proc.Identity == parts[1] {
-					if fp.ForwardFlow.Octets != nil {
-						octetsReceived += *fp.ForwardFlow.Octets
-					}
-					if fp.ForwardFlow.OctetRate != nil {
-						octetsReceivedRate = octetsReceivedRate + *fp.ForwardFlow.OctetRate
-					}
-					if fp.CounterFlow.Octets != nil {
-						octetsSent += *fp.CounterFlow.Octets
-					}
-					if fp.CounterFlow.OctetRate != nil {
-						octetsSentRate = octetsSentRate + *fp.CounterFlow.OctetRate
-					}
-				}
-			}
-		}
-	}
-	proc.OctetsSent = octetsSent
-	proc.OctetsSentRate = octetsSentRate
-	proc.OctetsReceived = octetsReceived
-	proc.OctetsReceivedRate = octetsReceivedRate
 
 	return nil
 }
-
-func (fc *FlowCollector) getFlowAggregateMetrics(itemType int, identity string) (*FlowAggregateRecord, error) {
-	if aggregate, ok := fc.FlowAggregates[identity]; ok {
-		// todo determine way to prime latency calcs
-		for _, flowPair := range fc.FlowPairs {
-			if flowPair.ForwardFlow.Latency != nil && flowPair.CounterFlow.Latency != nil {
-				aggregate.SourceMaxLatency = *flowPair.ForwardFlow.Latency
-				aggregate.SourceMinLatency = *flowPair.ForwardFlow.Latency
-				aggregate.DestinationMaxLatency = *flowPair.CounterFlow.Latency
-				aggregate.DestinationMinLatency = *flowPair.CounterFlow.Latency
-				break
-			}
-		}
-		sumSourceLatency := uint64(0)
-		sumDestinationLatency := uint64(0)
-		aggregate.RecordCount = uint64(0)
-		sourceOctets := uint64(0)
-		sourceOctetRate := uint64(0)
-		destinationOctets := uint64(0)
-		destinationOctetRate := uint64(0)
-
-		for _, flowPair := range fc.FlowPairs {
-			found := false
-			switch itemType {
-			case Site:
-				if aggregate.PairType == recordNames[Site] {
-					if flowPair.SiteAggregateId != nil && *flowPair.SiteAggregateId == aggregate.Identity {
-						found = true
-					}
-				}
-			case ProcessGroup:
-				if aggregate.PairType == recordNames[ProcessGroup] {
-					if flowPair.ProcessGroupAggregateId != nil && *flowPair.ProcessGroupAggregateId == aggregate.Identity {
-						found = true
-					}
-				}
-			case Process:
-				if aggregate.PairType == recordNames[Process] {
-					if flowPair.ProcessAggregateId != nil && *flowPair.ProcessAggregateId == aggregate.Identity {
-						found = true
-					}
-				}
-			}
-			if found {
-				aggregate.RecordCount++
-				if flowPair.ForwardFlow.Octets != nil {
-					sourceOctets += *flowPair.ForwardFlow.Octets
-				}
-				if flowPair.ForwardFlow.OctetRate != nil {
-					sourceOctetRate += *flowPair.ForwardFlow.OctetRate
-				}
-				if flowPair.ForwardFlow.Latency != nil {
-					aggregate.SourceMinLatency = min(aggregate.SourceMinLatency, *flowPair.ForwardFlow.Latency)
-					aggregate.SourceMaxLatency = max(aggregate.SourceMaxLatency, *flowPair.ForwardFlow.Latency)
-					sumSourceLatency += *flowPair.ForwardFlow.Latency
-				}
-				if flowPair.CounterFlow.Octets != nil {
-					destinationOctets += *flowPair.CounterFlow.Octets
-				}
-				if flowPair.CounterFlow.OctetRate != nil {
-					destinationOctetRate += *flowPair.CounterFlow.OctetRate
-				}
-				if flowPair.CounterFlow.Latency != nil {
-					aggregate.DestinationMinLatency = min(aggregate.DestinationMinLatency, *flowPair.CounterFlow.Latency)
-					aggregate.DestinationMaxLatency = max(aggregate.DestinationMaxLatency, *flowPair.CounterFlow.Latency)
-					sumDestinationLatency += *flowPair.CounterFlow.Latency
-				}
-			}
-		}
-		aggregate.SourceOctets = sourceOctets
-		aggregate.SourceOctetRate = sourceOctetRate
-		aggregate.DestinationOctets = destinationOctets
-		aggregate.DestinationOctetRate = destinationOctetRate
-		if aggregate.RecordCount > 0 {
-			aggregate.SourceAverageLatency = sumSourceLatency / aggregate.RecordCount
-			aggregate.DestinationAverageLatency = sumDestinationLatency / aggregate.RecordCount
-		}
-		return aggregate, nil
-	}
-	return nil, fmt.Errorf("Aggregate not found %s", identity)
-}
diff --git a/pkg/flow/record.go b/pkg/flow/record.go
index b65e7af6..bdd36741 100644
--- a/pkg/flow/record.go
+++ b/pkg/flow/record.go
@@ -9,6 +9,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"golang.org/x/text/cases"
 	"golang.org/x/text/language"
 )
@@ -165,6 +166,11 @@ var attributeNames = []string{
 	"StreamIdentity",  // 47
 }
 
+var Internal string = "internal"
+var External string = "external"
+var Incoming string = "incoming"
+var Outgoing string = "outgoing"
+
 type Base struct {
 	RecType   string `json:"recType,omitempty"`
 	Identity  string `json:"identity,omitempty"`
@@ -203,15 +209,11 @@ type EventSourceRecord struct {
 
 type SiteRecord struct {
 	Base
-	Location           *string `json:"location,omitempty"`
-	Provider           *string `json:"provider,omitempty"`
-	Platform           *string `json:"platform,omitempty"`
-	Name               *string `json:"name,omitempty"`
-	NameSpace          *string `json:"nameSpace,omitempty"`
-	OctetsSent         uint64  `json:"octetsSent"`
-	OctetsSentRate     uint64  `json:"octetSentRate"`
-	OctetsReceived     uint64  `json:"octetsReceived"`
-	OctetsReceivedRate uint64  `json:"octetReceivedRate"`
+	Location  *string `json:"location,omitempty"`
+	Provider  *string `json:"provider,omitempty"`
+	Platform  *string `json:"platform,omitempty"`
+	Name      *string `json:"name,omitempty"`
+	NameSpace *string `json:"nameSpace,omitempty"`
 }
 
 type HostRecord struct {
@@ -261,6 +263,7 @@ type ListenerRecord struct {
 	FlowRateL4  *uint64 `json:"flowRateL4,omitempty"`
 	FlowCountL7 *uint64 `json:"flowCountL7,omitempty"`
 	FlowRateL7  *uint64 `json:"flowRateL7,omitempty"`
+	AddressId   *string `json:"addressId,omitempty"`
 }
 
 type ConnectorRecord struct {
@@ -274,50 +277,48 @@ type ConnectorRecord struct {
 	FlowCountL7 *uint64 `json:"flowCountL7,omitempty"`
 	FlowRateL7  *uint64 `json:"flowRateL7,omitempty"`
 	process     *string
+	AddressId   *string `json:"addressId,omitempty"`
+}
+
+type metricKey struct {
+	sourceSite    string
+	sourceProcess string
+	destSite      string
+	destProcess   string
 }
 
 // Van Address represents a service that is attached to the application network
 type VanAddressRecord struct {
 	Base
-	Name           string `json:"name,omitempty"`
-	Protocol       string `json:"protocol,omitempty"`
-	ListenerCount  int    `json:"listenerCount"`
-	ConnectorCount int    `json:"connectorCount"`
-	TotalFlows     int    `json:"totalFlows"`
-	CurrentFlows   int    `json:"currentFlows"`
-	SourceOctets   uint64 `json:"sourceOctets"`
-	DestOctets     uint64 `json:"destOctets"`
+	Name            string `json:"name,omitempty"`
+	Protocol        string `json:"protocol,omitempty"`
+	ListenerCount   int    `json:"listenerCount"`
+	ConnectorCount  int    `json:"connectorCount"`
+	flowCount       map[metricKey]prometheus.Counter
+	octetCount      map[metricKey]prometheus.Counter
+	lastAccessed    map[metricKey]prometheus.Gauge
+	flowLatency     map[metricKey]prometheus.Observer
+	activeFlowCount map[metricKey]prometheus.Gauge
 }
 
-var Internal string = "internal"
-var External string = "external"
-
 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"`
-	OctetsSent         uint64  `json:"octetsSent"`
-	OctetsSentRate     uint64  `json:"octetSentRate"`
-	OctetsReceived     uint64  `json:"octetsReceived"`
-	OctetsReceivedRate uint64  `json:"octetReceivedRate"`
-	connector          *string
+	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"`
+	connector     *string
 }
 
 type ProcessGroupRecord struct {
 	Base
-	Name               *string `json:"name,omitempty"`
-	ProcessGroupRole   *string `json:"processGroupRole,omitempty"`
-	OctetsSent         uint64  `json:"octetsSent"`
-	OctetsSentRate     uint64  `json:"octetSentRate"`
-	OctetsReceived     uint64  `json:"octetsReceived"`
-	OctetsReceivedRate uint64  `json:"octetReceivedRate"`
+	Name             *string `json:"name,omitempty"`
+	ProcessGroupRole *string `json:"processGroupRole,omitempty"`
 }
 
 type FlowPlace int
@@ -330,25 +331,29 @@ const (
 
 type FlowRecord struct {
 	Base
-	SourceHost     *string   `json:"sourceHost,omitempty"`
-	SourcePort     *string   `json:"sourcePort,omitempty"`
-	CounterFlow    *string   `json:"counterFlow,omitempty"`
-	Trace          *string   `json:"trace,omitempty"`
-	Latency        *uint64   `json:"latency,omitempty"`
-	Octets         *uint64   `json:"octets"`
-	OctetRate      *uint64   `json:"octetRate"`
-	OctetsOut      *uint64   `json:"octetsOut,omitempty"`
-	OctetsUnacked  *uint64   `json:"octetsUnacked,omitempty"`
-	WindowClosures *uint64   `json:"windowClosures,omitempty"`
-	WindowSize     *uint64   `json:"windowSize,omitempty"`
-	Reason         *string   `json:"reason,omitempty"`
-	Method         *string   `json:"method,omitempty"`
-	Result         *string   `json:"result,omitempty"`
-	StreamIdentity *uint64   `json:"streamIdentity,omitempty"`
-	Process        *string   `json:"process,omitempty"`
-	ProcessName    *string   `json:"processName,omitempty"`
-	Protocol       *string   `json:"protocol,omitempty"`
-	Place          FlowPlace `json:"place"`
+	SourceHost       *string   `json:"sourceHost,omitempty"`
+	SourcePort       *string   `json:"sourcePort,omitempty"`
+	CounterFlow      *string   `json:"counterFlow,omitempty"`
+	Trace            *string   `json:"trace,omitempty"`
+	Latency          *uint64   `json:"latency,omitempty"`
+	Octets           *uint64   `json:"octets"`
+	OctetRate        *uint64   `json:"octetRate"`
+	OctetsOut        *uint64   `json:"octetsOut,omitempty"`
+	OctetsUnacked    *uint64   `json:"octetsUnacked,omitempty"`
+	WindowClosures   *uint64   `json:"windowClosures,omitempty"`
+	WindowSize       *uint64   `json:"windowSize,omitempty"`
+	Reason           *string   `json:"reason,omitempty"`
+	Method           *string   `json:"method,omitempty"`
+	Result           *string   `json:"result,omitempty"`
+	StreamIdentity   *uint64   `json:"streamIdentity,omitempty"`
+	Process          *string   `json:"process,omitempty"`
+	ProcessName      *string   `json:"processName,omitempty"`
+	Protocol         *string   `json:"protocol,omitempty"`
+	Place            FlowPlace `json:"place"`
+	lastOctets       uint64
+	octetMetric      prometheus.Counter
+	activeFlowMetric prometheus.Gauge
+	httpReqsMetric   prometheus.Counter
 }
 
 // Note a flowpair does not have a defined parent relationship through Base
@@ -369,22 +374,12 @@ 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"`
-	SourceOctets              uint64  `json:"sourceOctets,omitempty"`
-	SourceOctetRate           uint64  `json:"sourceOctetRate,omitempty"`
-	SourceMinLatency          uint64  `json:"sourceMinLatency,omitempty"`
-	SourceMaxLatency          uint64  `json:"sourceMaxLatency,omitempty"`
-	SourceAverageLatency      uint64  `json:"sourceAverageLatency,omitempty"`
-	DestinationOctets         uint64  `json:"destinationOctets,omitempty"`
-	DestinationOctetRate      uint64  `json:"destinationOctetRate,omitempty"`
-	DestinationMinLatency     uint64  `json:"destinationMinLatency,omitempty"`
-	DestinationMaxLatency     uint64  `json:"destinationMaxLatency,omitempty"`
-	DestinationAverageLatency uint64  `json:"destinationAverageLatency,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"`
 }
 
 type ControllerRecord struct {
@@ -413,7 +408,10 @@ type EgressRecord struct {
 
 type CollectorRecord struct {
 	Base
-	//	name, kind, process
+	PrometheusHost       string
+	PrometheusAuthMethod string
+	PrometheusUser       string
+	PrometheusPassword   string
 }
 
 type Payload struct {
diff --git a/pkg/images/image_utils.go b/pkg/images/image_utils.go
index b89fdcca..2ea02d4f 100644
--- a/pkg/images/image_utils.go
+++ b/pkg/images/image_utils.go
@@ -13,11 +13,14 @@ const (
 	ServiceControllerImageEnvKey      string = "SKUPPER_SERVICE_CONTROLLER_IMAGE"
 	ConfigSyncImageEnvKey             string = "SKUPPER_CONFIG_SYNC_IMAGE"
 	FlowCollectorImageEnvKey          string = "SKUPPER_FLOW_COLLECTOR_IMAGE"
+	PrometheusServerImageEnvKey       string = "PROMETHEUS_SERVER_IMAGE"
 	RouterPullPolicyEnvKey            string = "QDROUTERD_IMAGE_PULL_POLICY"
 	ServiceControllerPullPolicyEnvKey string = "SKUPPER_SERVICE_CONTROLLER_IMAGE_PULL_POLICY"
 	ConfigSyncPullPolicyEnvKey        string = "SKUPPER_CONFIG_SYNC_IMAGE_PULL_POLICY"
 	FlowCollectorPullPolicyEnvKey     string = "SKUPPER_FLOW_COLLECTOR_IMAGE_PULL_POLICY"
+	PrometheusServerPullPolicyEnvKey  string = "PROMETHEUS_SERVER_IMAGE_PULL_POLICY"
 	SkupperImageRegistryEnvKey        string = "SKUPPER_IMAGE_REGISTRY"
+	PrometheusImageRegistryEnvKey     string = "PROMETHEUS_IMAGE_REGISTRY"
 )
 
 func getPullPolicy(key string) string {
@@ -126,6 +129,27 @@ func GetFlowCollectorImageDetails() types.ImageDetails {
 	}
 }
 
+func GetPrometheusServerImageName() string {
+	image := os.Getenv(PrometheusServerImageEnvKey)
+	if image == "" {
+		imageRegistry := GetPrometheusImageRegistry()
+		return strings.Join([]string{imageRegistry, PrometheusServerImageName}, "/")
+	} else {
+		return image
+	}
+}
+
+func GetPrometheusServerImagePullPolicy() string {
+	return getPullPolicy(PrometheusServerPullPolicyEnvKey)
+}
+
+func GetPrometheusServerImageDetails() types.ImageDetails {
+	return types.ImageDetails{
+		Name:       GetPrometheusServerImageName(),
+		PullPolicy: GetPrometheusServerImagePullPolicy(),
+	}
+}
+
 func GetImageRegistry() string {
 	imageRegistry := os.Getenv(SkupperImageRegistryEnvKey)
 	if imageRegistry == "" {
@@ -133,3 +157,11 @@ func GetImageRegistry() string {
 	}
 	return imageRegistry
 }
+
+func GetPrometheusImageRegistry() string {
+	imageRegistry := os.Getenv(PrometheusImageRegistryEnvKey)
+	if imageRegistry == "" {
+		return PrometheusImageRegistry
+	}
+	return imageRegistry
+}
diff --git a/pkg/images/images.go b/pkg/images/images.go
index 7b95f474..a296da6b 100644
--- a/pkg/images/images.go
+++ b/pkg/images/images.go
@@ -6,4 +6,6 @@ const (
 	ServiceControllerImageName string = "service-controller:master"
 	ConfigSyncImageName        string = "config-sync:master"
 	FlowCollectorImageName     string = "flow-collector:master"
+	PrometheusImageRegistry    string = "quay.io/prometheus"
+	PrometheusServerImageName  string = "prometheus:v2.42.0"
 )
diff --git a/pkg/kube/containers.go b/pkg/kube/containers.go
index fb88e5b2..31ac48e4 100644
--- a/pkg/kube/containers.go
+++ b/pkg/kube/containers.go
@@ -193,3 +193,18 @@ func ContainerForFlowCollector(ds types.DeploymentSpec) *corev1.Container {
 	setResourceRequests(&container, &ds)
 	return &container
 }
+
+func ContainerForPrometheusServer(ds types.DeploymentSpec) corev1.Container {
+	// --web.config.file=/path_to/web-config.yaml for tls server config and basic user auth
+
+	container := corev1.Container{
+		Name:            types.PrometheusContainerName,
+		Image:           ds.Image.Name,
+		Args:            []string{"--config.file=/etc/prometheus/prometheus.yml", "--storage.tsdb.path=/prometheus/", "--web.config.file=/etc/prometheus/web-config.yml"},
+		Env:             ds.EnvVar,
+		VolumeMounts:    []corev1.VolumeMount{},
+		ImagePullPolicy: GetPullPolicy(ds.Image.PullPolicy),
+	}
+	setResourceRequests(&container, &ds)
+	return container
+}
diff --git a/pkg/kube/deployments.go b/pkg/kube/deployments.go
index d0b078ea..90cfdcf8 100644
--- a/pkg/kube/deployments.go
+++ b/pkg/kube/deployments.go
@@ -258,7 +258,7 @@ func NewProxyStatefulSet(image types.ImageDetails, serviceInterface types.Servic
 
 }
 
-func NewControllerDeployment(van *types.RouterSpec, ownerRef *metav1.OwnerReference, securityContext *corev1.SecurityContext, cli kubernetes.Interface) (*appsv1.Deployment, error) {
+func NewControllerDeployment(van *types.RouterSpec, ownerRef *metav1.OwnerReference, cli kubernetes.Interface) (*appsv1.Deployment, error) {
 	deployments := cli.AppsV1().Deployments(van.Namespace)
 	existing, err := deployments.Get(context.TODO(), types.ControllerDeploymentName, metav1.GetOptions{})
 	if err == nil {
@@ -304,12 +304,12 @@ func NewControllerDeployment(van *types.RouterSpec, ownerRef *metav1.OwnerRefere
 			dep.Spec.Template.Spec.Containers[i].VolumeMounts = van.Controller.VolumeMounts[i]
 		}
 
-		if securityContext != nil {
+		if van.Controller.SecurityContext != nil {
 			dep.Spec.Template.Spec.SecurityContext = &corev1.PodSecurityContext{
-				RunAsNonRoot: securityContext.RunAsNonRoot,
+				RunAsNonRoot: van.Controller.SecurityContext.RunAsNonRoot,
 			}
 			for i, _ := range dep.Spec.Template.Spec.Containers {
-				dep.Spec.Template.Spec.Containers[i].SecurityContext = securityContext
+				dep.Spec.Template.Spec.Containers[i].SecurityContext = van.Controller.SecurityContext
 			}
 		}
 
@@ -326,6 +326,74 @@ func NewControllerDeployment(van *types.RouterSpec, ownerRef *metav1.OwnerRefere
 	}
 }
 
+func NewPrometheusServerDeployment(van *types.RouterSpec, ownerRef *metav1.OwnerReference, cli kubernetes.Interface) (*appsv1.Deployment, error) {
+	deployments := cli.AppsV1().Deployments(van.Namespace)
+	existing, err := deployments.Get(context.TODO(), types.PrometheusDeploymentName, metav1.GetOptions{})
+	if err == nil {
+		return existing, nil
+	} else if errors.IsNotFound(err) {
+		dep := &appsv1.Deployment{
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: "apps/v1",
+				Kind:       "Deployment",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      types.PrometheusDeploymentName,
+				Namespace: van.Namespace,
+			},
+			Spec: appsv1.DeploymentSpec{
+				Replicas: &van.PrometheusServer.Replicas,
+				Selector: &metav1.LabelSelector{
+					MatchLabels: van.PrometheusServer.LabelSelector,
+				},
+				Template: corev1.PodTemplateSpec{
+					ObjectMeta: metav1.ObjectMeta{
+						Labels:      van.PrometheusServer.Labels,
+						Annotations: van.PrometheusServer.Annotations,
+					},
+					Spec: corev1.PodSpec{
+						ServiceAccountName: types.PrometheusServiceAccountName,
+						Containers:         []corev1.Container{ContainerForPrometheusServer(van.PrometheusServer)},
+					},
+				},
+			},
+		}
+		if ownerRef != nil {
+			dep.ObjectMeta.OwnerReferences = []metav1.OwnerReference{*ownerRef}
+		}
+
+		setAffinity(&van.PrometheusServer, &dep.Spec.Template.Spec)
+		for _, sc := range van.PrometheusServer.Sidecars {
+			dep.Spec.Template.Spec.Containers = append(dep.Spec.Template.Spec.Containers, *sc)
+		}
+
+		dep.Spec.Template.Spec.Volumes = van.PrometheusServer.Volumes
+		for i, _ := range van.PrometheusServer.VolumeMounts {
+			dep.Spec.Template.Spec.Containers[i].VolumeMounts = van.PrometheusServer.VolumeMounts[i]
+		}
+
+		if van.PrometheusServer.SecurityContext != nil {
+			dep.Spec.Template.Spec.SecurityContext = &corev1.PodSecurityContext{
+				RunAsNonRoot: van.PrometheusServer.SecurityContext.RunAsNonRoot,
+			}
+			for i, _ := range dep.Spec.Template.Spec.Containers {
+				dep.Spec.Template.Spec.Containers[i].SecurityContext = van.PrometheusServer.SecurityContext
+			}
+		}
+
+		created, err := deployments.Create(context.TODO(), dep, metav1.CreateOptions{})
+		if err != nil {
+			return nil, fmt.Errorf("Failed to create prometheus server deployment: %w", err)
+		} else {
+			return created, nil
+		}
+
+	} else {
+		dep := &appsv1.Deployment{}
+		return dep, fmt.Errorf("Failed to check prometheus server deployment: %w", err)
+	}
+}
+
 func setAffinity(spec *types.DeploymentSpec, podspec *corev1.PodSpec) {
 	if len(spec.NodeSelector) > 0 {
 		podspec.NodeSelector = spec.NodeSelector
@@ -366,7 +434,7 @@ func setAffinity(spec *types.DeploymentSpec, podspec *corev1.PodSpec) {
 	}
 }
 
-func NewTransportDeployment(van *types.RouterSpec, ownerRef *metav1.OwnerReference, securityContext *corev1.SecurityContext, cli kubernetes.Interface) (*appsv1.Deployment, error) {
+func NewTransportDeployment(van *types.RouterSpec, ownerRef *metav1.OwnerReference, cli kubernetes.Interface) (*appsv1.Deployment, error) {
 	deployments := cli.AppsV1().Deployments(van.Namespace)
 	existing, err := deployments.Get(context.TODO(), types.TransportDeploymentName, metav1.GetOptions{})
 	if err == nil {
@@ -416,12 +484,12 @@ func NewTransportDeployment(van *types.RouterSpec, ownerRef *metav1.OwnerReferen
 			dep.Spec.Template.Spec.Containers[i].VolumeMounts = van.Transport.VolumeMounts[i]
 		}
 
-		if securityContext != nil {
+		if van.Transport.SecurityContext != nil {
 			dep.Spec.Template.Spec.SecurityContext = &corev1.PodSecurityContext{
-				RunAsNonRoot: securityContext.RunAsNonRoot,
+				RunAsNonRoot: van.Transport.SecurityContext.RunAsNonRoot,
 			}
 			for i, _ := range dep.Spec.Template.Spec.Containers {
-				dep.Spec.Template.Spec.Containers[i].SecurityContext = securityContext
+				dep.Spec.Template.Spec.Containers[i].SecurityContext = van.Transport.SecurityContext
 			}
 		}
 


Test patch:
diff --git a/client/router_create_test.go b/client/router_create_test.go
index 296c8a82..5c51d35c 100644
--- a/client/router_create_test.go
+++ b/client/router_create_test.go
@@ -102,12 +102,12 @@ func TestRouterCreateDefaults(t *testing.T) {
 			user:                 "",
 			password:             "",
 			clusterLocal:         false,
-			depsExpected:         []string{"skupper-service-controller", "skupper-router"},
+			depsExpected:         []string{"skupper-service-controller", "skupper-router", "skupper-prometheus"},
 			cmsExpected:          []string{types.ServiceInterfaceConfigMap, types.TransportConfigMapName},
-			rolesExpected:        []string{types.TransportRoleName, types.ControllerRoleName},
+			rolesExpected:        []string{types.TransportRoleName, types.ControllerRoleName, types.PrometheusRoleName},
 			clusterRolesExpected: []string{types.ControllerClusterRoleName},
 			clusterRoleResources: sets.NewString("skupperclusterpolicies", "nodes", "namespaces"),
-			roleBindingsExpected: []string{types.TransportRoleBindingName, types.ControllerRoleBindingName},
+			roleBindingsExpected: []string{types.TransportRoleBindingName, types.ControllerRoleBindingName, types.PrometheusRoleBindingName},
 			secretsExpected: []string{types.LocalCaSecret,
 				types.SiteCaSecret,
 				types.LocalServerSecret,
@@ -116,9 +116,10 @@ func TestRouterCreateDefaults(t *testing.T) {
 				types.ConsoleServerSecret,
 				types.SiteServerSecret,
 				types.ServiceCaSecret,
-				types.ServiceClientSecret},
-			svcsExpected:        []string{types.LocalTransportServiceName, types.TransportServiceName, types.ControllerServiceName},
-			svcAccountsExpected: []string{types.TransportServiceAccountName, types.ControllerServiceAccountName},
+				types.ServiceClientSecret,
+				types.PrometheusServerSecret},
+			svcsExpected:        []string{types.LocalTransportServiceName, types.TransportServiceName, types.ControllerServiceName, types.PrometheusServiceName},
+			svcAccountsExpected: []string{types.TransportServiceAccountName, types.ControllerServiceAccountName, types.PrometheusServiceAccountName},
 			opts: []cmp.Option{
 				trans,
 				cmpopts.IgnoreSliceElements(func(v string) bool { return !strings.HasPrefix(v, "skupper") }),
@@ -140,12 +141,12 @@ func TestRouterCreateDefaults(t *testing.T) {
 			user:                 "",
 			password:             "",
 			clusterLocal:         false,
-			depsExpected:         []string{"skupper-service-controller", "skupper-router"},
+			depsExpected:         []string{"skupper-service-controller", "skupper-router", "skupper-prometheus"},
 			cmsExpected:          []string{types.ServiceInterfaceConfigMap, types.TransportConfigMapName, "skupper-sasl-config"},
-			rolesExpected:        []string{types.TransportRoleName, types.ControllerRoleName},
+			rolesExpected:        []string{types.TransportRoleName, types.ControllerRoleName, types.PrometheusRoleName},
 			clusterRolesExpected: []string{types.ControllerClusterRoleName},
 			clusterRoleResources: sets.NewString("skupperclusterpolicies", "nodes", "namespaces"),
-			roleBindingsExpected: []string{types.TransportRoleBindingName, types.ControllerRoleBindingName},
+			roleBindingsExpected: []string{types.TransportRoleBindingName, types.ControllerRoleBindingName, types.PrometheusRoleBindingName},
 			secretsExpected: []string{types.LocalCaSecret,
 				types.SiteCaSecret,
 				types.LocalServerSecret,
@@ -155,9 +156,10 @@ func TestRouterCreateDefaults(t *testing.T) {
 				"skupper-console-users",
 				types.ConsoleServerSecret,
 				types.ServiceCaSecret,
-				types.ServiceClientSecret},
-			svcsExpected:        []string{types.LocalTransportServiceName, types.TransportServiceName, types.ControllerServiceName},
-			svcAccountsExpected: []string{types.TransportServiceAccountName, types.ControllerServiceAccountName},
+				types.ServiceClientSecret,
+				types.PrometheusServerSecret},
+			svcsExpected:        []string{types.LocalTransportServiceName, types.TransportServiceName, types.ControllerServiceName, types.PrometheusServiceName},
+			svcAccountsExpected: []string{types.TransportServiceAccountName, types.ControllerServiceAccountName, types.PrometheusServiceAccountName},
 			opts: []cmp.Option{
 				trans,
 				cmpopts.IgnoreSliceElements(func(v string) bool { return !strings.HasPrefix(v, "skupper") }),
@@ -179,12 +181,12 @@ func TestRouterCreateDefaults(t *testing.T) {
 			user:                 "",
 			password:             "",
 			clusterLocal:         false,
-			depsExpected:         []string{"skupper-service-controller", "skupper-router"},
+			depsExpected:         []string{"skupper-service-controller", "skupper-router", "skupper-prometheus"},
 			cmsExpected:          []string{types.ServiceInterfaceConfigMap, types.TransportConfigMapName},
-			rolesExpected:        []string{types.TransportRoleName, types.ControllerRoleName},
+			rolesExpected:        []string{types.TransportRoleName, types.ControllerRoleName, types.PrometheusRoleName},
 			clusterRolesExpected: []string{types.ControllerClusterRoleName},
 			clusterRoleResources: sets.NewString("skupperclusterpolicies", "nodes", "namespaces"),
-			roleBindingsExpected: []string{types.TransportRoleBindingName, types.ControllerRoleBindingName},
+			roleBindingsExpected: []string{types.TransportRoleBindingName, types.ControllerRoleBindingName, types.PrometheusRoleBindingName},
 			secretsExpected: []string{types.LocalCaSecret,
 				types.SiteCaSecret,
 				types.LocalServerSecret,
@@ -193,9 +195,10 @@ func TestRouterCreateDefaults(t *testing.T) {
 				types.SiteServerSecret,
 				types.ConsoleServerSecret,
 				types.ServiceCaSecret,
-				types.ServiceClientSecret},
-			svcsExpected:        []string{types.LocalTransportServiceName, types.TransportServiceName, types.ControllerServiceName},
-			svcAccountsExpected: []string{types.TransportServiceAccountName, types.ControllerServiceAccountName},
+				types.ServiceClientSecret,
+				types.PrometheusServerSecret},
+			svcsExpected:        []string{types.LocalTransportServiceName, types.TransportServiceName, types.ControllerServiceName, types.PrometheusServiceName},
+			svcAccountsExpected: []string{types.TransportServiceAccountName, types.ControllerServiceAccountName, types.PrometheusServiceAccountName},
 			opts: []cmp.Option{
 				trans,
 				cmpopts.IgnoreSliceElements(func(v string) bool { return !strings.HasPrefix(v, "skupper") }),
@@ -217,20 +220,21 @@ func TestRouterCreateDefaults(t *testing.T) {
 			user:                 "Barney",
 			password:             "Rubble",
 			clusterLocal:         false,
-			depsExpected:         []string{"skupper-service-controller", "skupper-router"},
+			depsExpected:         []string{"skupper-service-controller", "skupper-router", "skupper-prometheus"},
 			cmsExpected:          []string{types.ServiceInterfaceConfigMap, types.TransportConfigMapName},
-			rolesExpected:        []string{types.TransportRoleName, types.ControllerRoleName},
+			rolesExpected:        []string{types.TransportRoleName, types.ControllerRoleName, types.PrometheusRoleName},
 			clusterRolesExpected: []string{types.ControllerClusterRoleName},
 			clusterRoleResources: sets.NewString("skupperclusterpolicies", "nodes", "namespaces"),
-			roleBindingsExpected: []string{types.TransportRoleBindingName, types.ControllerRoleBindingName},
+			roleBindingsExpected: []string{types.TransportRoleBindingName, types.ControllerRoleBindingName, types.PrometheusRoleBindingName},
 			secretsExpected: []string{types.LocalCaSecret,
 				types.ConsoleServerSecret,
 				types.LocalServerSecret,
 				types.LocalClientSecret,
 				types.ServiceCaSecret,
-				types.ServiceClientSecret},
-			svcsExpected:        []string{types.LocalTransportServiceName, types.ControllerServiceName},
-			svcAccountsExpected: []string{types.TransportServiceAccountName, types.ControllerServiceAccountName},
+				types.ServiceClientSecret,
+				types.PrometheusServerSecret},
+			svcsExpected:        []string{types.LocalTransportServiceName, types.ControllerServiceName, types.PrometheusServiceName},
+			svcAccountsExpected: []string{types.TransportServiceAccountName, types.ControllerServiceAccountName, types.PrometheusServiceAccountName},
 			opts: []cmp.Option{
 				trans,
 				cmpopts.IgnoreSliceElements(func(v string) bool { return !strings.HasPrefix(v, "skupper") }),
diff --git a/client/site_config_test.go b/client/site_config_test.go
index 05a2a986..a1a0fe3a 100644
--- a/client/site_config_test.go
+++ b/client/site_config_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/skupperproject/skupper/api/types"
@@ -25,6 +26,9 @@ func TestSiteConfigRoundtrip(t *testing.T) {
 				AuthMode:         "internal",
 				Annotations:      map[string]string{},
 				Labels:           map[string]string{},
+				Router:           types.RouterOptions{Logging: []types.RouterLogConfig{{Module: "ROUTER_CORE", Level: "error+"}}},
+				FlowCollector:    types.FlowCollectorOptions{FlowRecordTtl: types.DefaultFlowTimeoutDuration},
+				PrometheusServer: types.PrometheusServerOptions{AuthMode: "tls"},
 			},
 		},
 		{
@@ -42,6 +46,9 @@ func TestSiteConfigRoundtrip(t *testing.T) {
 				AuthMode:         "none",
 				Annotations:      map[string]string{},
 				Labels:           map[string]string{},
+				Router:           types.RouterOptions{Logging: []types.RouterLogConfig{{Module: "ROUTER_CORE", Level: "error+"}}},
+				FlowCollector:    types.FlowCollectorOptions{FlowRecordTtl: types.DefaultFlowTimeoutDuration},
+				PrometheusServer: types.PrometheusServerOptions{AuthMode: "tls"},
 			},
 		},
 		{
@@ -63,6 +70,9 @@ func TestSiteConfigRoundtrip(t *testing.T) {
 				Password:         "secret",
 				Annotations:      map[string]string{},
 				Labels:           map[string]string{},
+				Router:           types.RouterOptions{Logging: []types.RouterLogConfig{{Module: "ROUTER_CORE", Level: "error+"}}},
+				FlowCollector:    types.FlowCollectorOptions{FlowRecordTtl: types.DefaultFlowTimeoutDuration},
+				PrometheusServer: types.PrometheusServerOptions{AuthMode: "tls"},
 			},
 		},
 		{
@@ -111,6 +121,8 @@ func TestSiteConfigRoundtrip(t *testing.T) {
 					MaxFrameSize:     1111,
 					MaxSessionFrames: 2222,
 				},
+				FlowCollector:    types.FlowCollectorOptions{FlowRecordTtl: types.DefaultFlowTimeoutDuration},
+				PrometheusServer: types.PrometheusServerOptions{AuthMode: "tls"},
 			},
 		},
 		{
@@ -134,6 +146,7 @@ func TestSiteConfigRoundtrip(t *testing.T) {
 				AuthMode:         "internal",
 				Annotations:      map[string]string{},
 				Labels:           map[string]string{},
+				Router:           types.RouterOptions{Logging: []types.RouterLogConfig{{Module: "ROUTER_CORE", Level: "error+"}}},
 				Controller: types.ControllerOptions{
 					Tuning: types.Tuning{
 						Cpu:          "100m",
@@ -143,6 +156,8 @@ func TestSiteConfigRoundtrip(t *testing.T) {
 						NodeSelector: "kubernetes.io/hostname=nodeX",
 					},
 				},
+				FlowCollector:    types.FlowCollectorOptions{FlowRecordTtl: types.DefaultFlowTimeoutDuration},
+				PrometheusServer: types.PrometheusServerOptions{AuthMode: "tls"},
 			},
 		},
 		{
@@ -162,7 +177,10 @@ func TestSiteConfigRoundtrip(t *testing.T) {
 				Labels:           map[string]string{},
 				Router: types.RouterOptions{
 					IngressHost: "foo.com",
+					Logging:     []types.RouterLogConfig{{Module: "ROUTER_CORE", Level: "error+"}},
 				},
+				FlowCollector:    types.FlowCollectorOptions{FlowRecordTtl: types.DefaultFlowTimeoutDuration},
+				PrometheusServer: types.PrometheusServerOptions{AuthMode: "tls"},
 			},
 		},
 		{
@@ -180,6 +198,51 @@ func TestSiteConfigRoundtrip(t *testing.T) {
 				AuthMode:         "internal",
 				Annotations:      map[string]string{},
 				Labels:           map[string]string{},
+				Router:           types.RouterOptions{Logging: []types.RouterLogConfig{{Module: "ROUTER_CORE", Level: "error+"}}},
+				FlowCollector:    types.FlowCollectorOptions{FlowRecordTtl: types.DefaultFlowTimeoutDuration},
+				PrometheusServer: types.PrometheusServerOptions{AuthMode: "tls"},
+			},
+		},
+		{
+			input: types.SiteConfigSpec{
+				Ingress: "none",
+				FlowCollector: types.FlowCollectorOptions{
+					Tuning: types.Tuning{
+						Cpu:    "1",
+						Memory: "2G",
+					},
+					FlowRecordTtl: time.Minute * 30,
+				},
+				PrometheusServer: types.PrometheusServerOptions{
+					Tuning: types.Tuning{
+						Cpu:    "1",
+						Memory: "2G",
+					},
+				},
+			},
+			expected: types.SiteConfigSpec{
+				SkupperName:      "site-config-roundtrip-8",
+				SkupperNamespace: "site-config-roundtrip-8",
+				Ingress:          "none",
+				RouterMode:       "interior",
+				AuthMode:         "internal",
+				Annotations:      map[string]string{},
+				Labels:           map[string]string{},
+				FlowCollector: types.FlowCollectorOptions{
+					Tuning: types.Tuning{
+						Cpu:    "1",
+						Memory: "2G",
+					},
+					FlowRecordTtl: time.Minute * 30,
+				},
+				Router: types.RouterOptions{Logging: []types.RouterLogConfig{{Module: "ROUTER_CORE", Level: "error+"}}},
+				PrometheusServer: types.PrometheusServerOptions{
+					Tuning: types.Tuning{
+						Cpu:    "1",
+						Memory: "2G",
+					},
+					AuthMode: "tls",
+				},
 			},
 		},
 	}
diff --git a/cmd/skupper/skupper_test.go b/cmd/skupper/skupper_test.go
index 97047167..3d288120 100644
--- a/cmd/skupper/skupper_test.go
+++ b/cmd/skupper/skupper_test.go
@@ -293,3 +293,23 @@ func TestSkupperInitRunAsParseArgs(t *testing.T) {
 	assert.Assert(t, cmd.ParseFlags(cmdArgs))
 	assert.Equal(t, routerCreateOpts.RunAsGroup, int64(2000))
 }
+
+func TestSkupperInitFlowCollectorParseArgs(t *testing.T) {
+	cmd := NewCmdInit(NewSkupperTestClient().Site())
+
+	assert.Assert(t, cmd.ParseFlags([]string{}))
+	assert.Equal(t, routerCreateOpts.FlowCollector.Cpu, "")
+	assert.Equal(t, routerCreateOpts.FlowCollector.CpuLimit, "")
+	assert.Equal(t, routerCreateOpts.FlowCollector.Memory, "")
+	assert.Equal(t, routerCreateOpts.FlowCollector.MemoryLimit, "")
+	assert.Equal(t, routerCreateOpts.FlowCollector.FlowRecordTtl, time.Second*0)
+
+	cmdArgs := []string{"--flow-collector-record-ttl", "15m", "--flow-collector-cpu", "1", "--flow-collector-memory", "2G", "--flow-collector-cpu-limit", "2", "--flow-collector-memory-limit", "3G"}
+
+	assert.Assert(t, cmd.ParseFlags(cmdArgs))
+	assert.Equal(t, routerCreateOpts.FlowCollector.Cpu, "1")
+	assert.Equal(t, routerCreateOpts.FlowCollector.CpuLimit, "2")
+	assert.Equal(t, routerCreateOpts.FlowCollector.Memory, "2G")
+	assert.Equal(t, routerCreateOpts.FlowCollector.MemoryLimit, "3G")
+	assert.Equal(t, routerCreateOpts.FlowCollector.FlowRecordTtl, time.Minute*15)
+}
diff --git a/pkg/flow/flow_mem_driver_test.go b/pkg/flow/flow_mem_driver_test.go
new file mode 100644
index 00000000..22d6e6b3
--- /dev/null
+++ b/pkg/flow/flow_mem_driver_test.go
@@ -0,0 +1,785 @@
+package flow
+
+import (
+	"encoding/json"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/prometheus/client_golang/prometheus"
+	"gotest.tools/assert"
+)
+
+func TestRecordGraph(t *testing.T) {
+	name := "skupper-site"
+	namespace := "skupper-public"
+	provider := "aws"
+	address := "tcp-go-echo"
+	address2 := "redis-cart"
+	address3 := "mongo"
+	address4 := "cartservice"
+	protocol := "tcp"
+	counterFlow1 := "flow:0"
+	counterFlow2 := "flow:2"
+	counterFlow4 := "flow:4"
+	flowTrace := "router:0|router:1"
+	processName1 := "checkout-1234"
+	processName2 := "payment-1234"
+	groupName1 := "online-store"
+	listenerProcess := "process:0"
+	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,
+			process:  &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,
+			process:  &connectorProcess2,
+		},
+	}
+	flows := []FlowRecord{
+		{
+			Base: Base{
+				RecType:   recordNames[Flow],
+				Identity:  "flow:0",
+				Parent:    "listener:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			ProcessName: &processName1,
+			Process:     &listenerProcess,
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Flow],
+				Identity:  "flow:1",
+				Parent:    "connector:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			CounterFlow: &counterFlow1,
+			Trace:       &flowTrace,
+			ProcessName: &processName2,
+			Process:     &connectorProcess1,
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Flow],
+				Identity:  "flow:2",
+				Parent:    "listener:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			ProcessName: &processName1,
+			Process:     &listenerProcess,
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Flow],
+				Identity:  "flow:3",
+				Parent:    "connector:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			CounterFlow: &counterFlow2,
+			ProcessName: &processName2,
+			Process:     &connectorProcess1,
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Flow],
+				Identity:  "flow:2",
+				Parent:    "listener:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+				EndTime:   uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			ProcessName: &processName1,
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Flow],
+				Identity:  "flow:4",
+				Parent:    "flow:0",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			ProcessName: &processName1,
+			Process:     &listenerProcess,
+		},
+		{
+			Base: Base{
+				RecType:   recordNames[Flow],
+				Identity:  "flow:5",
+				Parent:    "flow:1",
+				StartTime: uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+			},
+			CounterFlow: &counterFlow4,
+			ProcessName: &processName2,
+			Trace:       &flowTrace,
+			Process:     &connectorProcess1,
+		},
+	}
+
+	reg := prometheus.NewRegistry()
+	u, _ := time.ParseDuration("5m")
+	fc := NewFlowCollector("origin", reg, nil, u)
+	fc.metrics = fc.NewMetrics(fc.prometheusReg)
+	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 _, f := range flows {
+		err := fc.updateRecord(f)
+		assert.Assert(t, err)
+		err = fc.updateRecord(f)
+		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")
+	}
+	for _, f := range flows {
+		if f.Parent != "" {
+			id := fc.getRecordSiteId(f)
+			assert.Equal(t, id, "site:0")
+		}
+		if f.Trace != nil {
+			trace := fc.annotateFlowTrace(&f)
+			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()
+
+	type test struct {
+		recordType   int
+		method       string
+		url          string
+		params       map[string]string
+		vars         map[string]string
+		name         string
+		responseSize int
+	}
+
+	testTable := []test{
+		{
+			recordType:   Site,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{},
+			name:         "list",
+			responseSize: len(sites),
+		},
+		{
+			recordType:   Site,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "site:0"},
+			name:         "item",
+			responseSize: 1,
+		},
+		{
+			recordType:   Site,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "site:0"},
+			name:         "routers",
+			responseSize: len(routers),
+		},
+		{
+			recordType:   Site,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "site:0"},
+			name:         "links",
+			responseSize: len(links),
+		},
+		{
+			recordType:   Site,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "site:0"},
+			name:         "hosts",
+			responseSize: len(hosts),
+		},
+		{
+			recordType:   Site,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "site:0"},
+			name:         "processes",
+			responseSize: len(processes),
+		},
+		{
+			recordType:   Host,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{},
+			name:         "list",
+			responseSize: len(hosts),
+		},
+		{
+			recordType:   Host,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "host:0"},
+			name:         "item",
+			responseSize: 1,
+		},
+		{
+			recordType:   Router,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{},
+			name:         "list",
+			responseSize: len(routers),
+		},
+		{
+			recordType:   Router,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "router:0"},
+			name:         "item",
+			responseSize: 1,
+		},
+		{
+			recordType:   Router,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "router:0"},
+			name:         "flows",
+			responseSize: 6,
+		},
+		{
+			recordType:   Router,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "router:0"},
+			name:         "links",
+			responseSize: 1,
+		},
+		{
+			recordType:   Router,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "router:0"},
+			name:         "listeners",
+			responseSize: 1,
+		},
+		{
+			recordType:   Router,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "router:0"},
+			name:         "connectors",
+			responseSize: 1,
+		},
+		{
+			recordType:   Link,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{},
+			name:         "list",
+			responseSize: len(links),
+		},
+		{
+			recordType:   Link,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "link:0"},
+			name:         "item",
+			responseSize: 1,
+		},
+		{
+			recordType:   Listener,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{},
+			name:         "list",
+			responseSize: len(listeners),
+		},
+		{
+			recordType:   Listener,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "listener:0"},
+			name:         "item",
+			responseSize: 1,
+		},
+		{
+			recordType:   Listener,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "listener:0"},
+			name:         "flows",
+			responseSize: 3,
+		},
+		{
+			recordType:   Connector,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{},
+			name:         "list",
+			responseSize: 2,
+		},
+		{
+			recordType:   Connector,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "connector:0"},
+			name:         "item",
+			responseSize: 1,
+		},
+		{
+			recordType:   Connector,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "connector:0"},
+			name:         "flows",
+			responseSize: 3,
+		},
+		{
+			recordType:   Connector,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "connector:0"},
+			name:         "process",
+			responseSize: 1,
+		},
+		{
+			recordType:   Address,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{},
+			name:         "list",
+			responseSize: len(fc.VanAddresses),
+		},
+		{
+			recordType:   Connector,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "connector:0"},
+			name:         "item",
+			responseSize: 1,
+		},
+		{
+			recordType:   Flow,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{},
+			name:         "list",
+			responseSize: len(fc.Flows),
+		},
+		{
+			recordType:   Flow,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "flow:1"},
+			name:         "item",
+			responseSize: 1,
+		},
+		{
+			recordType:   Flow,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "flow:1"},
+			name:         "process",
+			responseSize: 1,
+		},
+		{
+			recordType:   FlowPair,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{},
+			name:         "list",
+			responseSize: len(fc.FlowPairs),
+		},
+		{
+			recordType:   FlowPair,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "fp-flow:0"},
+			name:         "item",
+			responseSize: 1,
+		},
+		{
+			recordType:   Process,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{},
+			name:         "list",
+			responseSize: len(processes),
+		},
+		{
+			recordType:   Process,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "process:1"},
+			name:         "item",
+			responseSize: 1,
+		},
+		{
+			recordType:   Process,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "process:0"},
+			name:         "flows",
+			responseSize: 3,
+		},
+		{
+			recordType:   Process,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "process:1"},
+			name:         "addresses",
+			responseSize: 1,
+		},
+		{
+			recordType:   Process,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "process:1"},
+			name:         "connector",
+			responseSize: 1,
+		},
+		{
+			recordType:   ProcessGroup,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{},
+			name:         "list",
+			responseSize: 1,
+		},
+		{
+			recordType:   SitePair,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{},
+			name:         "list",
+			responseSize: 1,
+		},
+		{
+			recordType:   SitePair,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "site:0-to-site:0"},
+			name:         "item",
+			responseSize: 1,
+		},
+		{
+			recordType:   ProcessGroupPair,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{},
+			name:         "list",
+			responseSize: 1,
+		},
+		{
+			recordType:   ProcessPair,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{},
+			name:         "list",
+			responseSize: 1,
+		},
+		{
+			recordType:   ProcessPair,
+			method:       "Get",
+			url:          "/",
+			params:       map[string]string{"sortBy": "identity.asc"},
+			vars:         map[string]string{"id": "process:0-to-process:1"},
+			name:         "item",
+			responseSize: 1,
+		},
+	}
+
+	var payload Payload
+	for _, test := range testTable {
+		req, _ := http.NewRequest(test.method, test.url, nil)
+		q := req.URL.Query()
+		for k, v := range test.params {
+			q.Add(k, v)
+		}
+		req = mux.SetURLVars(req, test.vars)
+		req.URL.RawQuery = q.Encode()
+		resp, err := fc.retrieve(ApiRequest{RecordType: test.recordType, HandlerName: test.name, Request: req})
+		assert.Assert(t, err)
+		err = json.Unmarshal([]byte(*resp), &payload)
+		assert.Assert(t, err)
+		assert.Equal(t, test.responseSize, payload.Count)
+	}
+}
diff --git a/pkg/flow/record_test.go b/pkg/flow/record_test.go
index e4c8fe03..27c701ce 100644
--- a/pkg/flow/record_test.go
+++ b/pkg/flow/record_test.go
@@ -278,32 +278,38 @@ func TestParameters(t *testing.T) {
 
 	testTable := []test{
 		{
-			values: map[string]string{},
+			values: map[string]string{"timeRangeStart": "1234", "timeRangeEnd": "0"},
 			queryParams: QueryParams{
 				Offset:             -1,
 				Limit:              -1,
 				SortBy:             "identity.asc",
 				Filter:             "",
+				TimeRangeStart:     uint64(1234),
+				TimeRangeEnd:       uint64(0),
 				TimeRangeOperation: intersects,
 			},
 		},
 		{
-			values: map[string]string{"timeRangeOperation": "contains"},
+			values: map[string]string{"timeRangeStart": "0", "timeRangeEnd": "0", "timeRangeOperation": "contains"},
 			queryParams: QueryParams{
 				Offset:             -1,
 				Limit:              -1,
 				SortBy:             "identity.asc",
 				Filter:             "",
+				TimeRangeStart:     uint64(0),
+				TimeRangeEnd:       uint64(0),
 				TimeRangeOperation: contains,
 			},
 		},
 		{
-			values: map[string]string{"timeRangeOperation": "within"},
+			values: map[string]string{"timeRangeStart": "1234", "timeRangeEnd": "0", "timeRangeOperation": "within"},
 			queryParams: QueryParams{
 				Offset:             -1,
 				Limit:              -1,
 				SortBy:             "identity.asc",
 				Filter:             "",
+				TimeRangeStart:     uint64(1234),
+				TimeRangeEnd:       uint64(0),
 				TimeRangeOperation: within,
 			},
 		},
@@ -313,6 +319,8 @@ func TestParameters(t *testing.T) {
 				"limit":              "10",
 				"sortBy":             "sourcePort.desc",
 				"filter":             "forwardFlow.protocol.tcp",
+				"timeRangeStart":     "0",
+				"timeRangeEnd":       "4567",
 				"timeRangeOperation": "intersects",
 			},
 			queryParams: QueryParams{
@@ -320,6 +328,8 @@ func TestParameters(t *testing.T) {
 				Limit:              10,
 				SortBy:             "sourcePort.desc",
 				Filter:             "forwardFlow.protocol.tcp",
+				TimeRangeStart:     uint64(0),
+				TimeRangeEnd:       uint64(4567),
 				TimeRangeOperation: intersects,
 			},
 		},
@@ -336,6 +346,8 @@ func TestParameters(t *testing.T) {
 		assert.Equal(t, qp.Offset, test.queryParams.Offset)
 		assert.Equal(t, qp.Limit, test.queryParams.Limit)
 		assert.Equal(t, qp.SortBy, test.queryParams.SortBy)
+		assert.Equal(t, qp.TimeRangeStart, test.queryParams.TimeRangeStart)
+		assert.Equal(t, qp.TimeRangeEnd, test.queryParams.TimeRangeEnd)
 		assert.Equal(t, qp.TimeRangeOperation, test.queryParams.TimeRangeOperation)
 	}
 }
diff --git a/test/integration/acceptance/console/console_data_endpoint.go b/test/integration/acceptance/console/console_data_endpoint.go
index 46eac930..b86a7553 100644
--- a/test/integration/acceptance/console/console_data_endpoint.go
+++ b/test/integration/acceptance/console/console_data_endpoint.go
@@ -2,8 +2,8 @@ package console
 
 import (
 	"context"
+	"log"
 
-	"github.com/prometheus/common/log"
 	"github.com/skupperproject/skupper/api/types"
 	"github.com/skupperproject/skupper/client"
 	"github.com/skupperproject/skupper/test/utils/base"
@@ -120,7 +120,7 @@ func CreateBackendDeployment(t *testing.T, cluster *client.VanClient) {
 }
 
 func Setup(ctx context.Context, t *testing.T, r base.ClusterTestRunner) {
-	log.Warn("Setting up deployments for console_data_endpoint test")
+	log.Println("Setting up deployments for console_data_endpoint test")
 	publicCluster, _ := r.GetPublicContext(1)
 	privateCluster, _ := r.GetPrivateContext(1)
 
diff --git a/test/integration/acceptance/custom/basic/basic.go b/test/integration/acceptance/custom/basic/basic.go
index 08c0ad3a..cda54b18 100644
--- a/test/integration/acceptance/custom/basic/basic.go
+++ b/test/integration/acceptance/custom/basic/basic.go
@@ -3,7 +3,11 @@ package basic
 import (
 	"context"
 	"encoding/json"
-	"github.com/prometheus/common/log"
+	"log"
+	"os"
+	"testing"
+	"time"
+
 	"github.com/skupperproject/skupper/api/types"
 	"github.com/skupperproject/skupper/pkg/kube"
 	"github.com/skupperproject/skupper/pkg/utils"
@@ -13,9 +17,6 @@ import (
 	"gotest.tools/assert"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"os"
-	"testing"
-	"time"
 )
 
 const (
@@ -170,12 +171,12 @@ func (r *BasicTestRunner) TearDown(ctx context.Context) {
 
 	pub, err := r.GetPublicContext(1)
 	if err != nil {
-		log.Warn(errMsg)
+		log.Println(errMsg)
 	}
 
 	priv, err := r.GetPrivateContext(1)
 	if err != nil {
-		log.Warn(errMsg)
+		log.Println(errMsg)
 	}
 
 	_ = pub.DeleteNamespace()
diff --git a/test/integration/acceptance/custom/edge_connectivity/edge_connectivity.go b/test/integration/acceptance/custom/edge_connectivity/edge_connectivity.go
index 6e2ebd5e..e3be71ad 100644
--- a/test/integration/acceptance/custom/edge_connectivity/edge_connectivity.go
+++ b/test/integration/acceptance/custom/edge_connectivity/edge_connectivity.go
@@ -3,10 +3,10 @@ package edgecon
 import (
 	"context"
 	"fmt"
+	"log"
 	"testing"
 	"time"
 
-	"github.com/prometheus/common/log"
 	"github.com/skupperproject/skupper/api/types"
 	"github.com/skupperproject/skupper/test/utils/base"
 	"github.com/skupperproject/skupper/test/utils/constants"
@@ -183,14 +183,14 @@ func (r *EdgeConnectivityTestRunner) TearDown(ctx context.Context, testcase *Tes
 	for i := 0; i < int(createOptsPublic.Replicas); i++ {
 		pub, err := r.GetPublicContext(i + 1)
 		if err != nil {
-			log.Warn(err.Error())
+			log.Println(err.Error())
 		}
 		pub.DeleteNamespace()
 	}
 
 	priv, err := r.GetPrivateContext(1) // There can be only one.
 	if err != nil {
-		log.Warn(err.Error())
+		log.Println(err.Error())
 	}
 	priv.DeleteNamespace()
 }