Repository: linkerd/linkerd2
Pull request #4522 (2020-06-01 17:19:40)

Fetch logs/events when integration test fails, not only for install tests
## Motivation

Mainly to know what caused containers to not start (or to restart), like in #4285

## Implementation

Followup to #4410, where we fetched unexpected logs/events when a test failed in `test/install_test.go`; now we're expanding that behavior to every integration test.

For that, we replace in each `TestMain()`:

```go
os.Exit(m.Run())
```

with

```go
os.Exit(testutil.Run(m, TestHelper, true))

```

where `testutil.Run()` executes the tests and fetches the logs/events if the tests failed.

Also extracted the log/event fetching and matching into its own separate file.


Patch:
diff --git a/bin/fmt b/bin/fmt
index b5e15d9ff..af64e4653 100755
--- a/bin/fmt
+++ b/bin/fmt
@@ -1,17 +1,18 @@
-#!/bin/sh
+#!/bin/bash
 
 set -eu
 
 # go list will exclude the vendor directory.  We use grep to exclude generated
 # code.
-dirs=$(go list -f \{\{.Dir\}\} ./... | grep -v controller/gen)
+while IFS= read -r line; do dirs+=("$line"); done <<< "$(go list -f \{\{.Dir\}\} ./... | grep -v controller/gen)"
+
 # go list will list all subdirectories and goimports acts recursively.  This
 # results in certain files being reported multiple time.  Therefore, we must
 # dedup them.
-files=$(bin/goimports -l "$dirs" | sort -u)
+files=$(bin/goimports -l "${dirs[@]}" | sort -u)
 
 if [ -n "$files" ]; then
-  for file in $files; do
+  for file in "${files[@]}"; do
     bin/goimports -d "$@" "$file"
   done
   exit 64
diff --git a/charts/linkerd2-multicluster/templates/gateway.yaml b/charts/linkerd2-multicluster/templates/gateway.yaml
index c7b0ec7fd..c3800ad00 100644
--- a/charts/linkerd2-multicluster/templates/gateway.yaml
+++ b/charts/linkerd2-multicluster/templates/gateway.yaml
@@ -3,7 +3,7 @@
 apiVersion: v1
 kind: ConfigMap
 metadata:
-  name: nginx-configuration
+  name: linkerd-gateway
   annotations:
     {{.Values.createdByAnnotation}}: {{default (printf "linkerd/helm %s" .Values.linkerdVersion) .Values.cliVersion}}
   namespace: {{.Values.namespace}}
@@ -63,7 +63,7 @@ spec:
       volumes:
         - name: config
           configMap:
-            name: nginx-configuration   
+            name: linkerd-gateway   
       containers:
         - name: nginx
           readinessProbe:
@@ -96,7 +96,6 @@ metadata:
   namespace: {{.Values.namespace}}
   annotations: 
     mirror.linkerd.io/gateway-identity: {{.Values.gatewayName}}.{{.Values.namespace}}.serviceaccount.identity.{{.Values.linkerdNamespace}}.{{.Values.identityTrustDomain}}
-    mirror.linkerd.io/probe-port: "{{.Values.gatewayProbePort}}"
     mirror.linkerd.io/probe-period: "{{.Values.gatewayProbeSeconds}}"
     mirror.linkerd.io/probe-path: {{.Values.gatewayProbePath}}
     mirror.linkerd.io/multicluster-gateway: "true"
diff --git a/cli/cmd/check.go b/cli/cmd/check.go
index 9820bfc12..b2a4ab1ce 100644
--- a/cli/cmd/check.go
+++ b/cli/cmd/check.go
@@ -213,7 +213,7 @@ func configureAndRunChecks(wout io.Writer, werr io.Writer, stage string, options
 	success := runChecks(wout, werr, hc, options.output)
 
 	if !success {
-		os.Exit(2)
+		os.Exit(1)
 	}
 
 	return nil
diff --git a/cli/cmd/multicluster.go b/cli/cmd/multicluster.go
index 1fcb0cd6e..a801a2d7a 100644
--- a/cli/cmd/multicluster.go
+++ b/cli/cmd/multicluster.go
@@ -63,10 +63,10 @@ type (
 	}
 
 	linkOptions struct {
-		namespace           string
-		clusterName         string
-		apiServerAddress    string
-		serviceAccountName  string
+		namespace          string
+		clusterName        string
+		apiServerAddress   string
+		serviceAccountName string
 	}
 
 	exportServiceOptions struct {
@@ -294,6 +294,7 @@ func newGatewaysCommand() *cobra.Command {
 			if err != nil {
 				return err
 			}
+
 			renderGateways(resp.GetOk().GatewaysTable.Rows, stdout)
 			return nil
 		},
@@ -314,9 +315,9 @@ func newMulticlusterInstallCommand() *cobra.Command {
 	}
 
 	cmd := &cobra.Command{
-		Use:    "install",
-		Short:  "Output Kubernetes configs to install the Linkerd multicluster add-on",
-		Args:   cobra.NoArgs,
+		Use:   "install",
+		Short: "Output Kubernetes configs to install the Linkerd multicluster add-on",
+		Args:  cobra.NoArgs,
 		RunE: func(cmd *cobra.Command, args []string) error {
 
 			values, err := buildMulticlusterInstallValues(options)
@@ -389,9 +390,9 @@ func newLinkCommand() *cobra.Command {
 	opts := linkOptions{}
 
 	cmd := &cobra.Command{
-		Use:    "link",
-		Short:  "Outputs a Kubernetes secret that allows a service mirror component to connect to this cluster",
-		Args:   cobra.NoArgs,
+		Use:   "link",
+		Short: "Outputs a Kubernetes secret that allows a service mirror component to connect to this cluster",
+		Args:  cobra.NoArgs,
 		RunE: func(cmd *cobra.Command, args []string) error {
 
 			if opts.clusterName == "" {
@@ -764,7 +765,7 @@ components on a cluster, manage credentials and link clusters together.`,
   # Exporting all the resources inside a folder and its sub-folders.
   linkerd export-service  <folder> | kubectl apply -f -`,
 	}
-	
+
 	multiclusterCmd.AddCommand(newLinkCommand())
 	multiclusterCmd.AddCommand(newMulticlusterInstallCommand())
 	multiclusterCmd.AddCommand(newExportServiceCommand())
diff --git a/controller/api/public/gateways.go b/controller/api/public/gateways.go
index d836ab674..564a444b8 100644
--- a/controller/api/public/gateways.go
+++ b/controller/api/public/gateways.go
@@ -4,14 +4,15 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/prometheus/common/model"
-
 	pb "github.com/linkerd/linkerd2/controller/gen/public"
+	"github.com/linkerd/linkerd2/pkg/k8s"
+	"github.com/prometheus/common/model"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 const (
 	gatewayAliveQuery           = "sum(gateway_alive%s) by (%s)"
-	numMirroredServicesQuery    = "sum(num_mirrored_services%s) by (%s)"
 	gatewayLatencyQuantileQuery = "histogram_quantile(%s, sum(irate(gateway_probe_latency_ms_bucket%s[%s])) by (le, %s))"
 )
 
@@ -53,7 +54,30 @@ func buildGatewaysRequestLabels(req *pb.GatewaysRequest) (labels model.LabelSet,
 	return labels, groupBy
 }
 
-func processPrometheusResult(results []promResult) map[string]*pb.GatewaysTable_Row {
+// this function returns a map of gateways to the number of services using them
+func (s *grpcServer) getNumServicesMap() (map[string]uint64, error) {
+
+	results := make(map[string]uint64)
+	selector := fmt.Sprintf("%s,!%s", k8s.MirroredResourceLabel, k8s.MirroredGatewayLabel)
+	services, err := s.k8sAPI.Client.CoreV1().Services(corev1.NamespaceAll).List(metav1.ListOptions{LabelSelector: selector})
+	if err != nil {
+		return nil, err
+	}
+
+	for _, svc := range services.Items {
+		clusterName := svc.Labels[k8s.RemoteClusterNameLabel]
+		gatewayName := svc.Labels[k8s.RemoteGatewayNameLabel]
+		gatewayNs := svc.Labels[k8s.RemoteGatewayNsLabel]
+		key := fmt.Sprintf("%s-%s-%s", clusterName, gatewayName, gatewayNs)
+
+		results[key]++
+	}
+
+	return results, nil
+}
+
+func processPrometheusResult(results []promResult, numSvcMap map[string]uint64) map[string]*pb.GatewaysTable_Row {
+
 	rows := make(map[string]*pb.GatewaysTable_Row)
 
 	for _, result := range results {
@@ -62,6 +86,7 @@ func processPrometheusResult(results []promResult) map[string]*pb.GatewaysTable_
 			clusterName := sample.Metric[remoteClusterNameLabel]
 			gatewayName := sample.Metric[gatewayNameLabel]
 			gatewayNamespace := sample.Metric[gatewayNamespaceLabel]
+			numPairedSvc := numSvcMap[fmt.Sprintf("%s-%s-%s", clusterName, gatewayName, gatewayNamespace)]
 
 			key := fmt.Sprintf("%s-%s-%s", clusterName, gatewayNamespace, gatewayName)
 
@@ -71,7 +96,7 @@ func processPrometheusResult(results []promResult) map[string]*pb.GatewaysTable_
 					rows[key].ClusterName = string(clusterName)
 					rows[key].Name = string(gatewayName)
 					rows[key].Namespace = string(gatewayNamespace)
-
+					rows[key].PairedServices = numPairedSvc
 				}
 			}
 
@@ -80,15 +105,7 @@ func processPrometheusResult(results []promResult) map[string]*pb.GatewaysTable_
 			switch result.prom {
 			case promGatewayAlive:
 				addRow()
-				if value == 0 {
-					rows[key].Alive = false
-				} else {
-					rows[key].Alive = true
-				}
-
-			case promNumMirroredServices:
-				addRow()
-				rows[key].PairedServices = value
+				rows[key].Alive = value > 0
 			case promLatencyP50:
 				addRow()
 				rows[key].LatencyMsP50 = value
@@ -111,17 +128,21 @@ func (s *grpcServer) getGatewaysMetrics(ctx context.Context, req *pb.GatewaysReq
 	reqLabels := generateLabelStringWithExclusion(labels, string(gatewayNameLabel))
 
 	promQueries := map[promType]string{
-		promGatewayAlive:        gatewayAliveQuery,
-		promNumMirroredServices: numMirroredServicesQuery,
+		promGatewayAlive: gatewayAliveQuery,
 	}
 
 	metricsResp, err := s.getPrometheusMetrics(ctx, promQueries, gatewayLatencyQuantileQuery, reqLabels, timeWindow, groupBy.String())
 
+	if err != nil {
+		return nil, err
+	}
+	numSvcMap, err := s.getNumServicesMap()
+
 	if err != nil {
 		return nil, err
 	}
 
-	rowsMap := processPrometheusResult(metricsResp)
+	rowsMap := processPrometheusResult(metricsResp, numSvcMap)
 
 	return rowsMap, nil
 }
diff --git a/controller/cmd/service-mirror/cluster_watcher.go b/controller/cmd/service-mirror/cluster_watcher.go
index 85ebdaf98..dcc2807ee 100644
--- a/controller/cmd/service-mirror/cluster_watcher.go
+++ b/controller/cmd/service-mirror/cluster_watcher.go
@@ -36,7 +36,6 @@ type (
 		log                    *logging.Entry
 		eventsQueue            workqueue.RateLimitingInterface
 		requeueLimit           int
-		probeEventsSink        ProbeEventSink
 	}
 
 	// ProbeConfig describes the configured probe on particular gateway (if presents)
@@ -62,7 +61,7 @@ type (
 	// this event means that the service in question is not mirrored atm
 	RemoteServiceCreated struct {
 		service     *corev1.Service
-		gatewayData *gatewayMetadata
+		gatewayData gatewayMetadata
 	}
 
 	// RemoteServiceUpdated is generated when we see something about an already
@@ -79,34 +78,26 @@ type (
 	// RemoteServiceDeleted when a remote service is going away or it is not
 	// considered mirrored anymore
 	RemoteServiceDeleted struct {
-		Name        string
-		Namespace   string
-		GatewayData gatewayMetadata
+		Name      string
+		Namespace string
 	}
 
-	// RemoteGatewayDeleted is observed when a service that is a gateway to at least
-	// one already mirrored service is deleted
+	// RemoteGatewayDeleted is observed when a service that is a gateway is deleted
 	RemoteGatewayDeleted struct {
 		gatewayData gatewayMetadata
 	}
 
-	// RemoteGatewayUpdated happens when a service that is a gateway to at least
-	// one already mirrored service is updated. This might mean an IP change,
-	// incoming port change, etc...
+	// RemoteGatewayCreated is observed when a gateway service is created on the remote cluster
+	RemoteGatewayCreated struct {
+		gatewaySpec GatewaySpec
+	}
+
+	// RemoteGatewayUpdated happens when a service that is updated.
 	RemoteGatewayUpdated struct {
 		gatewaySpec      GatewaySpec
 		affectedServices []*corev1.Service
 	}
 
-	// ConsiderGatewayUpdateDispatch is issued when we are receiving an update for a
-	// service but we are not sure that this is a gateway. We need to hit the local
-	// API and make sure there are services that have this service as a gateway. Since
-	// this is an operation that might fail (a glitch in the local api connectivity)
-	// we want to represent that as a separate event that can be requeued
-	ConsiderGatewayUpdateDispatch struct {
-		maybeGateway *corev1.Service
-	}
-
 	// ClusterUnregistered is issued when the secret containing the remote cluster
 	// access information is deleted
 	ClusterUnregistered struct{}
@@ -181,7 +172,6 @@ func NewRemoteClusterServiceWatcher(
 	clusterName string,
 	requeueLimit int,
 	clusterDomain string,
-	probeEventsSink ProbeEventSink,
 ) (*RemoteClusterServiceWatcher, error) {
 	remoteAPI, err := k8s.InitializeAPIForConfig(cfg, false, k8s.Svc)
 	if err != nil {
@@ -199,9 +189,8 @@ func NewRemoteClusterServiceWatcher(
 			"cluster":    clusterName,
 			"apiAddress": cfg.Host,
 		}),
-		eventsQueue:     workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
-		requeueLimit:    requeueLimit,
-		probeEventsSink: probeEventsSink,
+		eventsQueue:  workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
+		requeueLimit: requeueLimit,
 	}, nil
 }
 
@@ -292,7 +281,7 @@ func (rcsw *RemoteClusterServiceWatcher) cleanupOrphanedServices() error {
 
 	var errors []error
 	for _, srv := range servicesOnLocalCluster {
-		remoteSvc, err := rcsw.remoteAPIClient.Svc().Lister().Services(srv.Namespace).Get(rcsw.originalResourceName(srv.Name))
+		_, err := rcsw.remoteAPIClient.Svc().Lister().Services(srv.Namespace).Get(rcsw.originalResourceName(srv.Name))
 		if err != nil {
 			if kerrors.IsNotFound(err) {
 				// service does not exist anymore. Need to delete
@@ -306,19 +295,6 @@ func (rcsw *RemoteClusterServiceWatcher) cleanupOrphanedServices() error {
 				// something went wrong getting the service, we can retry
 				errors = append(errors, err)
 			}
-		} else {
-			if gtwData := getGatewayMetadata(remoteSvc.Annotations); gtwData != nil {
-				gatewaySpec, err := rcsw.resolveGateway(gtwData)
-				if gatewaySpec != nil && err == nil {
-					rcsw.probeEventsSink.send(&MirroredServicePaired{
-						serviceName:      srv.Name,
-						serviceNamespace: srv.Namespace,
-						GatewaySpec:      *gatewaySpec,
-					})
-				} else {
-					rcsw.log.Debugf("Could not resolve gateway for %s/%s: %s", srv.Namespace, srv.Name, err)
-				}
-			}
 		}
 	}
 	if len(errors) > 0 {
@@ -407,13 +383,6 @@ func (rcsw *RemoteClusterServiceWatcher) handleRemoteServiceDeleted(ev *RemoteSe
 	}
 
 	rcsw.log.Debugf("Successfully deleted Service: %s/%s", ev.Namespace, localServiceName)
-	rcsw.probeEventsSink.send(&MirroredServiceUnpaired{
-		serviceName:      localServiceName,
-		serviceNamespace: ev.Namespace,
-		gatewayName:      ev.GatewayData.Name,
-		gatewayNs:        ev.GatewayData.Namespace,
-		clusterName:      rcsw.clusterName,
-	})
 	return nil
 }
 
@@ -423,18 +392,6 @@ func (rcsw *RemoteClusterServiceWatcher) handleRemoteServiceUpdated(ev *RemoteSe
 	serviceInfo := fmt.Sprintf("%s/%s", ev.remoteUpdate.Namespace, ev.remoteUpdate.Name)
 	rcsw.log.Debugf("Updating remote mirrored service %s/%s", ev.localService.Namespace, ev.localService.Name)
 
-	gatewayChanged := false
-	if ev.localEndpoints.Labels[consts.RemoteGatewayNameLabel] != ev.gatewayData.Name || ev.localEndpoints.Labels[consts.RemoteGatewayNsLabel] != ev.gatewayData.Namespace {
-		gatewayChanged = true
-		rcsw.probeEventsSink.send(&MirroredServiceUnpaired{
-			serviceName:      ev.localService.Name,
-			serviceNamespace: ev.localService.Namespace,
-			gatewayName:      ev.localEndpoints.Labels[consts.RemoteGatewayNameLabel],
-			gatewayNs:        ev.localEndpoints.Labels[consts.RemoteGatewayNsLabel],
-			clusterName:      rcsw.clusterName,
-		})
-	}
-
 	gatewaySpec, err := rcsw.resolveGateway(&ev.gatewayData)
 	copiedEndpoints := ev.localEndpoints.DeepCopy()
 	if err == nil {
@@ -451,14 +408,6 @@ func (rcsw *RemoteClusterServiceWatcher) handleRemoteServiceUpdated(ev *RemoteSe
 			delete(copiedEndpoints.Annotations, consts.RemoteGatewayIdentity)
 		}
 
-		if gatewayChanged {
-			rcsw.probeEventsSink.send(&MirroredServicePaired{
-				serviceName:      ev.localService.Name,
-				serviceNamespace: ev.localService.Namespace,
-				GatewaySpec:      *gatewaySpec,
-			})
-		}
-
 	} else {
 		rcsw.log.Warnf("Could not resolve gateway for %s: %s, nulling endpoints", serviceInfo, err)
 		copiedEndpoints.Subsets = nil
@@ -500,137 +449,6 @@ func remapRemoteServicePorts(ports []corev1.ServicePort) []corev1.ServicePort {
 	return newPorts
 }
 
-func (rcsw *RemoteClusterServiceWatcher) updateGatewayMirrorService(spec *GatewaySpec) error {
-	localServiceName := rcsw.mirroredResourceName(spec.gatewayName)
-	service, err := rcsw.localAPIClient.Svc().Lister().Services(rcsw.serviceMirrorNamespace).Get(localServiceName)
-	if err != nil {
-		return err
-	}
-
-	if service.Annotations != nil && service.Annotations[consts.RemoteGatewayResourceVersionAnnotation] != spec.resourceVersion {
-		updatedService := service.DeepCopy()
-		if updatedService.Annotations != nil {
-			updatedService.Annotations[consts.RemoteGatewayResourceVersionAnnotation] = spec.resourceVersion
-		}
-
-		endpoints, err := rcsw.localAPIClient.Endpoint().Lister().Endpoints(rcsw.serviceMirrorNamespace).Get(localServiceName)
-		if err != nil {
-			return err
-		}
-
-		updatedEndpoints := endpoints.DeepCopy()
-		updatedEndpoints.Subsets = []corev1.EndpointSubset{
-			{
-				Addresses: spec.addresses,
-				Ports: []corev1.EndpointPort{
-					{
-						Protocol: "TCP",
-						Port:     int32(spec.ProbeConfig.port),
-					},
-				},
-			},
-		}
-
-		_, err = rcsw.localAPIClient.Client.CoreV1().Services(rcsw.serviceMirrorNamespace).Update(updatedService)
-		if err != nil {
-			return err
-		}
-
-		_, err = rcsw.localAPIClient.Client.CoreV1().Endpoints(rcsw.serviceMirrorNamespace).Update(updatedEndpoints)
-		if err != nil {
-			rcsw.localAPIClient.Client.CoreV1().Services(rcsw.serviceMirrorNamespace).Delete(updatedService.Name, &metav1.DeleteOptions{})
-			return err
-		}
-		rcsw.log.Debugf("%s gateway mirror updated", localServiceName)
-	}
-
-	return nil
-}
-
-// the logic here creates a mirror service for the gateway. The only port exposed there is the
-// probes port. This enables us to discover the gateways probe endpoints through the dst service
-// and apply proper identity
-func (rcsw *RemoteClusterServiceWatcher) createGatewayMirrorService(spec *GatewaySpec) error {
-	localServiceName := rcsw.mirroredResourceName(spec.gatewayName)
-	if spec.ProbeConfig == nil {
-		rcsw.log.Debugf("Skipping creation of gateway mirror as gateway does not specify probe config")
-		return nil
-	}
-
-	_, err := rcsw.localAPIClient.Svc().Lister().Services(rcsw.serviceMirrorNamespace).Get(localServiceName)
-	if err != nil {
-		if kerrors.IsNotFound(err) {
-			serviceToCreate := &corev1.Service{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      localServiceName,
-					Namespace: rcsw.serviceMirrorNamespace,
-					Annotations: map[string]string{
-						consts.RemoteGatewayResourceVersionAnnotation: spec.resourceVersion,
-					},
-					Labels: map[string]string{
-						consts.MirroredResourceLabel:  "true",
-						consts.RemoteClusterNameLabel: rcsw.clusterName,
-					},
-				},
-				Spec: corev1.ServiceSpec{
-					Ports: []corev1.ServicePort{
-						{
-							Protocol: "TCP",
-							Port:     int32(spec.ProbeConfig.port),
-						},
-					},
-				},
-			}
-
-			endpointsToCreate := &corev1.Endpoints{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      localServiceName,
-					Namespace: rcsw.serviceMirrorNamespace,
-					Labels: map[string]string{
-						consts.MirroredResourceLabel:  "true",
-						consts.RemoteClusterNameLabel: rcsw.clusterName,
-					},
-					Annotations: map[string]string{
-						consts.RemoteGatewayIdentity: spec.identity,
-					},
-				},
-				Subsets: []corev1.EndpointSubset{
-					{
-						Addresses: spec.addresses,
-						Ports: []corev1.EndpointPort{
-							{
-								Protocol: "TCP",
-								Port:     int32(spec.ProbeConfig.port),
-							},
-						},
-					},
-				},
-			}
-
-			rcsw.log.Debugf("Creating a new gateway mirror Service for %s", localServiceName)
-			if _, err := rcsw.localAPIClient.Client.CoreV1().Services(rcsw.serviceMirrorNamespace).Create(serviceToCreate); err != nil {
-				if !kerrors.IsAlreadyExists(err) {
-					// we might have created it during earlier attempt, if that is not the case, we retry
-					return RetryableError{[]error{err}}
-				}
-			}
-
-			rcsw.log.Debugf("Creating a new gateway mirror Endpoints for %s", localServiceName)
-			if _, err := rcsw.localAPIClient.Client.CoreV1().Endpoints(rcsw.serviceMirrorNamespace).Create(endpointsToCreate); err != nil {
-				// we clean up after ourselves
-				rcsw.localAPIClient.Client.CoreV1().Services(rcsw.serviceMirrorNamespace).Delete(spec.gatewayName, &metav1.DeleteOptions{})
-				// and retry
-				return RetryableError{[]error{err}}
-			}
-
-			return nil
-		}
-		return err
-	}
-	rcsw.log.Debugf("Skipping creation of gateway mirror as it already exists")
-	return nil
-}
-
 func (rcsw *RemoteClusterServiceWatcher) handleRemoteServiceCreated(ev *RemoteServiceCreated) error {
 	remoteService := ev.service.DeepCopy()
 	serviceInfo := fmt.Sprintf("%s/%s", remoteService.Namespace, remoteService.Name)
@@ -645,7 +463,7 @@ func (rcsw *RemoteClusterServiceWatcher) handleRemoteServiceCreated(ev *RemoteSe
 			Name:        localServiceName,
 			Namespace:   remoteService.Namespace,
 			Annotations: rcsw.getMirroredServiceAnnotations(remoteService),
-			Labels:      rcsw.getMirroredServiceLabels(ev.gatewayData),
+			Labels:      rcsw.getMirroredServiceLabels(&ev.gatewayData),
 		},
 		Spec: corev1.ServiceSpec{
 			Ports: remapRemoteServicePorts(remoteService.Spec.Ports),
@@ -669,7 +487,7 @@ func (rcsw *RemoteClusterServiceWatcher) handleRemoteServiceCreated(ev *RemoteSe
 	}
 
 	// Now we try to resolve the remote gateway
-	gatewaySpec, err := rcsw.resolveGateway(ev.gatewayData)
+	gatewaySpec, err := rcsw.resolveGateway(&ev.gatewayData)
 	if err == nil {
 		// only if we resolve it, we are updating the endpoints addresses and ports
 		rcsw.log.Debugf("Resolved remote gateway [%v:%d] for %s", gatewaySpec.addresses, gatewaySpec.incomingPort, serviceInfo)
@@ -684,16 +502,6 @@ func (rcsw *RemoteClusterServiceWatcher) handleRemoteServiceCreated(ev *RemoteSe
 			endpointsToCreate.Annotations[consts.RemoteGatewayIdentity] = gatewaySpec.identity
 		}
 
-		if err := rcsw.createGatewayMirrorService(gatewaySpec); err != nil {
-			return err
-		}
-
-		rcsw.probeEventsSink.send(&MirroredServicePaired{
-			serviceName:      serviceToCreate.Name,
-			serviceNamespace: serviceToCreate.Namespace,
-			GatewaySpec:      *gatewaySpec,
-		})
-
 	} else {
 		rcsw.log.Warnf("Could not resolve gateway for %s: %s, skipping subsets", serviceInfo, err)
 		endpointsToCreate.Subsets = nil
@@ -718,11 +526,6 @@ func (rcsw *RemoteClusterServiceWatcher) handleRemoteServiceCreated(ev *RemoteSe
 }
 
 func (rcsw *RemoteClusterServiceWatcher) handleRemoteGatewayDeleted(ev *RemoteGatewayDeleted) error {
-	rcsw.probeEventsSink.send(&GatewayDeleted{
-		gatewayName: ev.gatewayData.Name,
-		gatewayNs:   ev.gatewayData.Namespace,
-		clusterName: rcsw.clusterName,
-	})
 
 	if err := rcsw.localAPIClient.Client.CoreV1().Services(rcsw.serviceMirrorNamespace).Delete(rcsw.mirroredResourceName(ev.gatewayData.Name), &metav1.DeleteOptions{}); err != nil {
 		rcsw.log.Errorf("Could not delete gateway mirror %s", err)
@@ -756,14 +559,98 @@ func (rcsw *RemoteClusterServiceWatcher) handleRemoteGatewayDeleted(ev *RemoteGa
 	return nil
 }
 
-func (rcsw *RemoteClusterServiceWatcher) handleRemoteGatewayUpdated(ev *RemoteGatewayUpdated) error {
-	rcsw.log.Debugf("Updating %d services due to remote gateway [%s/%s] update", len(ev.affectedServices), ev.gatewaySpec.gatewayNamespace, ev.gatewaySpec.gatewayName)
+// the logic here creates a mirror service for the gateway. The only port exposed there is the
+// probes port. This enables us to discover the gateways probe endpoints through the dst service
+// and apply proper identity
+func (rcsw *RemoteClusterServiceWatcher) handleRemoteGatewayCreated(event *RemoteGatewayCreated) error {
+	localServiceName := rcsw.mirroredResourceName(event.gatewaySpec.gatewayName)
+	if event.gatewaySpec.ProbeConfig == nil {
+		rcsw.log.Debugf("Skipping creation of gateway mirror as gateway does not specify probe config")
+		return nil
+	}
+	serviceToCreate := &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      localServiceName,
+			Namespace: rcsw.serviceMirrorNamespace,
+			Annotations: map[string]string{
+				consts.RemoteGatewayResourceVersionAnnotation: event.gatewaySpec.resourceVersion,
+				consts.MirroredGatewayRemoteName:              event.gatewaySpec.gatewayName,
+				consts.MirroredGatewayRemoteNameSpace:         event.gatewaySpec.gatewayNamespace,
+				consts.MirroredGatewayProbePath:               event.gatewaySpec.ProbeConfig.path,
+				consts.MirroredGatewayProbePeriod:             fmt.Sprint(event.gatewaySpec.ProbeConfig.periodInSeconds),
+			},
+			Labels: map[string]string{
+				consts.MirroredResourceLabel:  "true",
+				consts.RemoteClusterNameLabel: rcsw.clusterName,
+				consts.MirroredGatewayLabel:   "true",
+			},
+		},
+		Spec: corev1.ServiceSpec{
+			Ports: []corev1.ServicePort{
+				{
+					Name:     consts.ProbePortName,
+					Protocol: "TCP",
+					Port:     int32(event.gatewaySpec.ProbeConfig.port),
+				},
+			},
+		},
+	}
 
+	endpointsToCreate := &corev1.Endpoints{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      localServiceName,
+			Namespace: rcsw.serviceMirrorNamespace,
+			Labels: map[string]string{
+				consts.MirroredResourceLabel:  "true",
+				consts.RemoteClusterNameLabel: rcsw.clusterName,
+			},
+			Annotations: map[string]string{
+				consts.RemoteGatewayIdentity: event.gatewaySpec.identity,
+			},
+		},
+	}
+
+	if len(event.gatewaySpec.addresses) > 0 {
+		endpointsToCreate.Subsets = []corev1.EndpointSubset{
+			{
+				Addresses: event.gatewaySpec.addresses,
+				Ports: []corev1.EndpointPort{
+					{
+						Name:     consts.ProbePortName,
+						Protocol: "TCP",
+						Port:     int32(event.gatewaySpec.ProbeConfig.port),
+					},
+				},
+			},
+		}
+	}
+
+	rcsw.log.Debugf("Creating a new gateway mirror Service for %s", localServiceName)
+	if _, err := rcsw.localAPIClient.Client.CoreV1().Services(rcsw.serviceMirrorNamespace).Create(serviceToCreate); err != nil {
+		if !kerrors.IsAlreadyExists(err) {
+			// we might have created it during earlier attempt, if that is not the case, we retry
+			return RetryableError{[]error{err}}
+		}
+	}
+
+	rcsw.log.Debugf("Creating a new gateway mirror Endpoints for %s", localServiceName)
+	if _, err := rcsw.localAPIClient.Client.CoreV1().Endpoints(rcsw.serviceMirrorNamespace).Create(endpointsToCreate); err != nil {
+		// we clean up after ourselves
+		rcsw.localAPIClient.Client.CoreV1().Services(rcsw.serviceMirrorNamespace).Delete(event.gatewaySpec.gatewayName, &metav1.DeleteOptions{})
+		// and retry
+		return RetryableError{[]error{err}}
+	}
+
+	return nil
+}
+
+func (rcsw *RemoteClusterServiceWatcher) updateAffectedServices(gatewaySpec GatewaySpec, affectedServices []*corev1.Service) error {
+	rcsw.log.Debugf("Updating %d services due to remote gateway [%s/%s] update", len(affectedServices), gatewaySpec.gatewayNamespace, gatewaySpec.gatewayName)
 	var errors []error
-	for _, svc := range ev.affectedServices {
+	for _, svc := range affectedServices {
 		updatedService := svc.DeepCopy()
 		if updatedService.Annotations != nil {
-			updatedService.Annotations[consts.RemoteGatewayResourceVersionAnnotation] = ev.gatewaySpec.resourceVersion
+			updatedService.Annotations[consts.RemoteGatewayResourceVersionAnnotation] = gatewaySpec.resourceVersion
 		}
 		endpoints, err := rcsw.localAPIClient.Endpoint().Lister().Endpoints(svc.Namespace).Get(svc.Name)
 		if err != nil {
@@ -772,15 +659,19 @@ func (rcsw *RemoteClusterServiceWatcher) handleRemoteGatewayUpdated(ev *RemoteGa
 		}
 
 		updatedEndpoints := endpoints.DeepCopy()
-		updatedEndpoints.Subsets = []corev1.EndpointSubset{
-			{
-				Addresses: ev.gatewaySpec.addresses,
-				Ports:     rcsw.getEndpointsPorts(updatedService, int32(ev.gatewaySpec.incomingPort)),
-			},
+		if len(gatewaySpec.addresses) > 0 {
+			updatedEndpoints.Subsets = []corev1.EndpointSubset{
+				{
+					Addresses: gatewaySpec.addresses,
+					Ports:     rcsw.getEndpointsPorts(updatedService, int32(gatewaySpec.incomingPort)),
+				},
+			}
+		} else {
+			updatedEndpoints.Subsets = nil
 		}
 
-		if ev.gatewaySpec.identity != "" {
-			updatedEndpoints.Annotations[consts.RemoteGatewayIdentity] = ev.gatewaySpec.identity
+		if gatewaySpec.identity != "" {
+			updatedEndpoints.Annotations[consts.RemoteGatewayIdentity] = gatewaySpec.identity
 		} else {
 			delete(updatedEndpoints.Annotations, consts.RemoteGatewayIdentity)
 		}
@@ -793,30 +684,80 @@ func (rcsw *RemoteClusterServiceWatcher) handleRemoteGatewayUpdated(ev *RemoteGa
 
 		_, err = rcsw.localAPIClient.Client.CoreV1().Endpoints(updatedService.Namespace).Update(updatedEndpoints)
 		if err != nil {
-			rcsw.localAPIClient.Client.CoreV1().Services(updatedService.Namespace).Delete(updatedService.Name, &metav1.DeleteOptions{})
 			errors = append(errors, err)
 		}
+	}
 
-		// we need to repair as this gateway might have been deleted
-		// due to being updated to a state that is not valid anymore
-		rcsw.probeEventsSink.send(&MirroredServicePaired{
-			serviceName:      updatedService.Name,
-			serviceNamespace: updatedService.Namespace,
-			GatewaySpec:      ev.gatewaySpec,
-		})
+	if len(errors) > 0 {
+		return RetryableError{errors}
 	}
+	return nil
+}
 
-	if err := rcsw.updateGatewayMirrorService(&ev.gatewaySpec); err != nil {
-		rcsw.log.Errorf("Could not update %s gateway proxy service: %s", ev.gatewaySpec.gatewayName, err)
+func (rcsw *RemoteClusterServiceWatcher) updateGatewayMirrorService(spec *GatewaySpec) error {
+	localServiceName := rcsw.mirroredResourceName(spec.gatewayName)
+	service, err := rcsw.localAPIClient.Svc().Lister().Services(rcsw.serviceMirrorNamespace).Get(localServiceName)
+	if err != nil {
+		return err
 	}
 
-	rcsw.probeEventsSink.send(&GatewayUpdated{
-		GatewaySpec: ev.gatewaySpec,
-	})
+	if service.Annotations != nil && service.Annotations[consts.RemoteGatewayResourceVersionAnnotation] != spec.resourceVersion {
+		updatedService := service.DeepCopy()
+		if updatedService.Annotations != nil {
+			updatedService.Annotations[consts.RemoteGatewayResourceVersionAnnotation] = spec.resourceVersion
+			updatedService.Annotations[consts.MirroredGatewayProbePath] = spec.ProbeConfig.path
+			updatedService.Annotations[consts.MirroredGatewayProbePeriod] = fmt.Sprint(spec.ProbeConfig.periodInSeconds)
+		}
+
+		endpoints, err := rcsw.localAPIClient.Endpoint().Lister().Endpoints(rcsw.serviceMirrorNamespace).Get(localServiceName)
+		if err != nil {
+			return err
+		}
 
-	if len(errors) > 0 {
-		return RetryableError{errors}
+		updatedEndpoints := endpoints.DeepCopy()
+		if spec.addresses == nil {
+			updatedEndpoints.Subsets = nil
+		} else {
+			updatedEndpoints.Subsets = []corev1.EndpointSubset{
+				{
+					Addresses: spec.addresses,
+					Ports: []corev1.EndpointPort{
+						{
+							Name:     consts.ProbePortName,
+							Protocol: "TCP",
+							Port:     int32(spec.ProbeConfig.port),
+						},
+					},
+				},
+			}
+		}
+
+		endpoints.Annotations[consts.RemoteGatewayIdentity] = spec.identity
+
+		_, err = rcsw.localAPIClient.Client.CoreV1().Services(rcsw.serviceMirrorNamespace).Update(updatedService)
+		if err != nil {
+			return err
+		}
+
+		_, err = rcsw.localAPIClient.Client.CoreV1().Endpoints(rcsw.serviceMirrorNamespace).Update(updatedEndpoints)
+		if err != nil {
+			return err
+		}
+		rcsw.log.Debugf("%s gateway mirror updated", localServiceName)
 	}
+
+	return nil
+}
+
+func (rcsw *RemoteClusterServiceWatcher) handleRemoteGatewayUpdated(ev *RemoteGatewayUpdated) error {
+	if err := rcsw.updateAffectedServices(ev.gatewaySpec, ev.affectedServices); err != nil {
+		return err
+	}
+
+	if err := rcsw.updateGatewayMirrorService(&ev.gatewaySpec); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -842,52 +783,14 @@ func isGateway(annotations map[string]string) bool {
 	}
 	return false
 }
-func (rcsw *RemoteClusterServiceWatcher) handleConsiderGatewayUpdateDispatch(event *ConsiderGatewayUpdateDispatch) error {
-	gtwMetadata := &gatewayMetadata{
-		Name:      event.maybeGateway.Name,
-		Namespace: event.maybeGateway.Namespace,
-	}
-
-	services, err := rcsw.mirroredServicesForGateway(gtwMetadata)
-	if err != nil {
-		// we can fail and requeue here in case there is a problem obtaining these...
-		if kerrors.IsNotFound(err) {
-			return err
-		}
-		return RetryableError{[]error{err}}
-
-	}
-
-	if len(services) > 0 {
-		gatewayMeta := gatewayMetadata{
-			Name:      event.maybeGateway.Name,
-			Namespace: event.maybeGateway.Namespace,
-		}
-		if gatewaySpec, err := rcsw.extractGatewaySpec(event.maybeGateway); err != nil {
-			rcsw.log.Warnf("Gateway [%s/%s] is not a compliant gateway anymore, dispatching GatewayDeleted event: %s", event.maybeGateway.Namespace, event.maybeGateway.Name, err)
-			// in case something changed about this gateway and it is not really a gateway anymore,
-			// simply dispatch deletion event so all endpoints are nulled
-			rcsw.eventsQueue.AddRateLimited(&RemoteGatewayDeleted{gatewayMeta})
-		} else {
-			affectedServices, err := rcsw.affectedMirroredServicesForGatewayUpdate(gtwMetadata, event.maybeGateway.ResourceVersion)
-			if err != nil {
-				if kerrors.IsNotFound(err) {
-					return err
-				}
-				return RetryableError{[]error{err}}
-
-			}
 
-			if len(affectedServices) > 0 {
-				rcsw.eventsQueue.Add(&RemoteGatewayUpdated{
-					gatewaySpec:      *gatewaySpec,
-					affectedServices: affectedServices,
-				})
-			}
-
-		}
+func isMirroredService(annotations map[string]string) bool {
+	if annotations != nil {
+		_, hasGtwName := annotations[consts.GatewayNameAnnotation]
+		_, hasGtwNs := annotations[consts.GatewayNsAnnotation]
+		return hasGtwName && hasGtwNs
 	}
-	return nil
+	return false
 }
 
 // this method is common to both CREATE and UPDATE because if we have been
@@ -895,84 +798,86 @@ func (rcsw *RemoteClusterServiceWatcher) handleConsiderGatewayUpdateDispatch(eve
 // observed before is simply a case of UPDATE
 func (rcsw *RemoteClusterServiceWatcher) createOrUpdateService(service *corev1.Service) error {
 	localName := rcsw.mirroredResourceName(service.Name)
-	localService, err := rcsw.localAPIClient.Svc().Lister().Services(service.Namespace).Get(localName)
-	var localNotFound = false
-	if err != nil {
-		localNotFound = kerrors.IsNotFound(err)
-	}
+
 	if isGateway(service.Annotations) {
 		gatewaySpec, err := rcsw.extractGatewaySpec(service)
 		if err != nil {
 			return RetryableError{[]error{err}}
 		}
-		if localNotFound {
-			return rcsw.createGatewayMirrorService(gatewaySpec)
+
+		_, err = rcsw.localAPIClient.Svc().Lister().Services(rcsw.serviceMirrorNamespace).Get(localName)
+		if err != nil {
+			if kerrors.IsNotFound(err) {
+				rcsw.eventsQueue.Add(&RemoteGatewayCreated{
+					gatewaySpec: *gatewaySpec,
+				})
+				return nil
+			}
+			return RetryableError{[]error{err}}
 		}
-		return rcsw.updateGatewayMirrorService(gatewaySpec)
 
-	}
-	gtwData := getGatewayMetadata(service.Annotations)
-	if err != nil {
-		if localNotFound {
-			if gtwData != nil {
-				// at this point we know that this is a service that
-				// we are not mirroring but has gateway data, so we need
-				// to create it
+		affectedServices, err := rcsw.affectedMirroredServicesForGatewayUpdate(&gatewayMetadata{
+			Name:      service.Name,
+			Namespace: service.Namespace,
+		}, service.ResourceVersion)
+		if err != nil {
+			return RetryableError{[]error{err}}
+		}
+
+		rcsw.eventsQueue.Add(&RemoteGatewayUpdated{
+			affectedServices: affectedServices,
+			gatewaySpec:      *gatewaySpec,
+		})
+		return nil
+
+	} else if isMirroredService(service.Annotations) {
+		gatewayData := getGatewayMetadata(service.Annotations)
+		if gatewayData == nil {
+			return fmt.Errorf("got service in invalid state, no gateway metadata %s", service)
+		}
+		localService, err := rcsw.localAPIClient.Svc().Lister().Services(service.Namespace).Get(localName)
+		if err != nil {
+			if kerrors.IsNotFound(err) {
 				rcsw.eventsQueue.Add(&RemoteServiceCreated{
 					service:     service,
-					gatewayData: gtwData,
+					gatewayData: *gatewayData,
 				})
-			} else {
-				// at this point we know that we do not have such a service
-				// and the remote service does not have metadata. So we try to
-				// dispatch a gateway update as the remote service might be a
-				/// gateway for some of our already mirrored services
-				rcsw.eventsQueue.Add(&ConsiderGatewayUpdateDispatch{maybeGateway: service})
+				return nil
 			}
-		} else {
-			// we can retry the operation here
 			return RetryableError{[]error{err}}
 		}
+		// if we have the local service present, we need to issue an update
+		lastMirroredRemoteVersion, ok := localService.Annotations[consts.RemoteResourceVersionAnnotation]
+		if ok && lastMirroredRemoteVersion != service.ResourceVersion {
+			endpoints, err := rcsw.localAPIClient.Endpoint().Lister().Endpoints(service.Namespace).Get(localName)
+			if err == nil {
+				rcsw.eventsQueue.Add(&RemoteServiceUpdated{
+					localService:   localService,
+					localEndpoints: endpoints,
+					remoteUpdate:   service,
+					gatewayData:    *gatewayData,
+				})
+				return nil
+			}
+			return RetryableError{[]error{err}}
+		}
+		return nil
 	} else {
-		if gtwData != nil {
-			// at this point we know this is an update to a service that we already
-			// have locally, so we try and see whether the res version has changed
-			// and if so, dispatch an RemoteServiceUpdated event
-			lastMirroredRemoteVersion, ok := localService.Annotations[consts.RemoteResourceVersionAnnotation]
-			if ok && lastMirroredRemoteVersion != service.ResourceVersion {
-				endpoints, err := rcsw.localAPIClient.Endpoint().Lister().Endpoints(service.Namespace).Get(localName)
-				if err == nil {
-					rcsw.eventsQueue.Add(&RemoteServiceUpdated{
-						localService:   localService,
-						localEndpoints: endpoints,
-						remoteUpdate:   service,
-						gatewayData:    *gtwData,
+		localSvc, err := rcsw.localAPIClient.Svc().Lister().Services(service.Namespace).Get(localName)
+		if err == nil {
+			if localSvc.Labels != nil {
+				_, isMirroredRes := localSvc.Labels[consts.MirroredResourceLabel]
+				clusterName := localSvc.Labels[consts.RemoteClusterNameLabel]
+				if isMirroredRes && (clusterName == rcsw.clusterName) {
+					rcsw.eventsQueue.Add(&RemoteServiceDeleted{
+						Name:      service.Name,
+						Namespace: service.Namespace,
 					})
-				} else {
-					return RetryableError{[]error{err}}
 				}
 			}
-		} else {
-			// if this is missing gateway metadata, but we have the
-			// service we can dispatch a RemoteServiceDeleted event
-			// because at some point in time we mirrored this service,
-			// however it is not mirrorable anymore
-
-			// try to extract the gateway metadata, so we have a well
-			// formed event that we can use to dispatch a service unpaired
-			// event to the probe manager
-			rcsw.eventsQueue.Add(&RemoteServiceDeleted{
-				Name:      service.Name,
-				Namespace: service.Namespace,
-				GatewayData: gatewayMetadata{
-					Name:      localService.Labels[consts.RemoteGatewayNameLabel],
-					Namespace: localService.Labels[consts.RemoteGatewayNsLabel],
-				},
-			})
 		}
+		return nil
 	}
-
-	return nil
 }
 
 func (rcsw *RemoteClusterServiceWatcher) affectedMirroredServicesForGatewayUpdate(gatewayData *gatewayMetadata, latestResourceVersion string) ([]*corev1.Service, error) {
@@ -1022,18 +927,19 @@ func (rcsw *RemoteClusterServiceWatcher) endpointsForGateway(gatewayData *gatewa
 }
 
 func (rcsw *RemoteClusterServiceWatcher) handleOnDelete(service *corev1.Service) {
-	if gtwData := getGatewayMetadata(service.Annotations); gtwData != nil {
+	if isMirroredService(service.Annotations) {
 		rcsw.eventsQueue.Add(&RemoteServiceDeleted{
-			Name:        service.Name,
-			Namespace:   service.Namespace,
-			GatewayData: *gtwData,
+			Name:      service.Name,
+			Namespace: service.Namespace,
 		})
-	} else {
+	} else if isGateway(service.Annotations) {
 		rcsw.eventsQueue.Add(&RemoteGatewayDeleted{
 			gatewayData: gatewayMetadata{
 				Name:      service.Name,
 				Namespace: service.Namespace,
 			}})
+	} else {
+		rcsw.log.Debugf("Skipping OnDelete for service %s", service)
 	}
 }
 
@@ -1065,8 +971,8 @@ func (rcsw *RemoteClusterServiceWatcher) processNextEvent() (bool, interface{},
 		err = rcsw.handleRemoteGatewayUpdated(ev)
 	case *RemoteGatewayDeleted:
 		err = rcsw.handleRemoteGatewayDeleted(ev)
-	case *ConsiderGatewayUpdateDispatch:
-		err = rcsw.handleConsiderGatewayUpdateDispatch(ev)
+	case *RemoteGatewayCreated:
+		err = rcsw.handleRemoteGatewayCreated(ev)
 	case *ClusterUnregistered:
 		err = rcsw.cleanupMirroredResources()
 	case *OprhanedServicesGcTriggered:
@@ -1152,9 +1058,6 @@ func (rcsw *RemoteClusterServiceWatcher) Start() error {
 
 // Stop stops watching the cluster and cleans up all mirrored resources
 func (rcsw *RemoteClusterServiceWatcher) Stop(cleanupState bool) {
-	rcsw.probeEventsSink.send(&ClusterNotRegistered{
-		clusterName: rcsw.clusterName,
-	})
 	close(rcsw.stopper)
 	if cleanupState {
 		rcsw.eventsQueue.Add(&ClusterUnregistered{})
@@ -1162,14 +1065,25 @@ func (rcsw *RemoteClusterServiceWatcher) Stop(cleanupState bool) {
 	rcsw.eventsQueue.ShutDown()
 }
 
-func parseProbeConfig(data map[string]string) (*ProbeConfig, error) {
-	probePath := data[consts.GatewayProbePath]
-	probePort, err := strconv.ParseUint(data[consts.GatewayProbePort], 10, 32)
+func extractPort(port []corev1.ServicePort, portName string) (uint32, error) {
+	for _, p := range port {
+		if p.Name == portName {
+			return uint32(p.Port), nil
+		}
+	}
+	return 0, fmt.Errorf("could not find port with name %s", portName)
+}
+
+func extractProbeConfig(gateway *corev1.Service) (*ProbeConfig, error) {
+	probePath := gateway.Annotations[consts.GatewayProbePath]
+
+	probePort, err := extractPort(gateway.Spec.Ports, consts.ProbePortName)
+
 	if err != nil {
 		return nil, err
 	}
 
-	probePeriod, err := strconv.ParseUint(data[consts.GatewayProbePeriod], 10, 32)
+	probePeriod, err := strconv.ParseUint(gateway.Annotations[consts.GatewayProbePeriod], 10, 32)
 	if err != nil {
 		return nil, err
 	}
@@ -1180,28 +1094,16 @@ func parseProbeConfig(data map[string]string) (*ProbeConfig, error) {
 
 	return &ProbeConfig{
 		path:            probePath,
-		port:            uint32(probePort),
+		port:            probePort,
 		periodInSeconds: uint32(probePeriod),
 	}, nil
 }
 
 func (rcsw *RemoteClusterServiceWatcher) extractGatewaySpec(gateway *corev1.Service) (*GatewaySpec, error) {
-	if len(gateway.Status.LoadBalancer.Ingress) == 0 {
-		return nil, fmt.Errorf("expected gateway %s/%s to have at lest 1 external Ip address but it has none", gateway.Namespace, gateway.Name)
-	}
+	incomingPort, err := extractPort(gateway.Spec.Ports, consts.GatewayPortName)
 
-	var foundPort = false
-	var port uint32
-	for _, p := range gateway.Spec.Ports {
-		if p.Name == consts.GatewayPortName {
-			foundPort = true
-			port = uint32(p.Port)
-			break
-		}
-	}
-
-	if !foundPort {
-		return nil, fmt.Errorf("cannot find  port named %s on gateway", consts.GatewayPortName)
+	if err != nil {
+		return nil, err
 	}
 
 	var gatewayEndpoints []corev1.EndpointAddress
@@ -1213,10 +1115,9 @@ func (rcsw *RemoteClusterServiceWatcher) extractGatewaySpec(gateway *corev1.Serv
 	}
 
 	gatewayIdentity := gateway.Annotations[consts.GatewayIdentity]
-	probeConfig, err := parseProbeConfig(gateway.Annotations)
-
+	probeConfig, err := extractProbeConfig(gateway)
 	if err != nil {
-		rcsw.log.Debugf("could not parse probe config for gateway: %s/%s: %s", gateway.Namespace, gateway.Name, err)
+		return nil, fmt.Errorf("could not parse probe config for gateway: %s/%s: %s", gateway.Namespace, gateway.Name, err)
 	}
 
 	return &GatewaySpec{
@@ -1224,7 +1125,7 @@ func (rcsw *RemoteClusterServiceWatcher) extractGatewaySpec(gateway *corev1.Serv
 		gatewayName:      gateway.Name,
 		gatewayNamespace: gateway.Namespace,
 		addresses:        gatewayEndpoints,
-		incomingPort:     port,
+		incomingPort:     incomingPort,
 		resourceVersion:  gateway.ResourceVersion,
 		identity:         gatewayIdentity,
 		ProbeConfig:      probeConfig,
diff --git a/controller/cmd/service-mirror/config_watcher.go b/controller/cmd/service-mirror/config_watcher.go
index 46fc215e1..2aabb1d2d 100644
--- a/controller/cmd/service-mirror/config_watcher.go
+++ b/controller/cmd/service-mirror/config_watcher.go
@@ -14,11 +14,6 @@ import (
 	sm "github.com/linkerd/linkerd2/pkg/servicemirror"
 )
 
-// ProbeEventSink is an interface for a type that can send events to the probe manager
-type ProbeEventSink interface {
-	send(event interface{})
-}
-
 // RemoteClusterConfigWatcher watches for secrets of type MirrorSecretType
 // and upon the detection of such secret created starts a RemoteClusterServiceWatcher
 type RemoteClusterConfigWatcher struct {
@@ -27,17 +22,15 @@ type RemoteClusterConfigWatcher struct {
 	clusterWatchers        map[string]*RemoteClusterServiceWatcher
 	requeueLimit           int
 	sync.RWMutex
-	probeEventsSink ProbeEventSink
 }
 
 // NewRemoteClusterConfigWatcher Creates a new config watcher
-func NewRemoteClusterConfigWatcher(serviceMirrorNamespace string, secretsInformer cache.SharedIndexInformer, k8sAPI *k8s.API, requeueLimit int, probeEventsSink ProbeEventSink) *RemoteClusterConfigWatcher {
+func NewRemoteClusterConfigWatcher(serviceMirrorNamespace string, secretsInformer cache.SharedIndexInformer, k8sAPI *k8s.API, requeueLimit int) *RemoteClusterConfigWatcher {
 	rcw := &RemoteClusterConfigWatcher{
 		serviceMirrorNamespace: serviceMirrorNamespace,
 		k8sAPI:                 k8sAPI,
 		clusterWatchers:        map[string]*RemoteClusterServiceWatcher{},
 		requeueLimit:           requeueLimit,
-		probeEventsSink:        probeEventsSink,
 	}
 	secretsInformer.AddEventHandler(
 		cache.FilteringResourceEventHandler{
@@ -133,7 +126,7 @@ func (rcw *RemoteClusterConfigWatcher) registerRemoteCluster(secret *corev1.Secr
 		return fmt.Errorf("there is already a cluster with name %s being watcher. Please delete its config before attempting to register a new one", config.ClusterName)
 	}
 
-	watcher, err := NewRemoteClusterServiceWatcher(rcw.serviceMirrorNamespace, rcw.k8sAPI, clientConfig, config.ClusterName, rcw.requeueLimit, config.ClusterDomain, rcw.probeEventsSink)
+	watcher, err := NewRemoteClusterServiceWatcher(rcw.serviceMirrorNamespace, rcw.k8sAPI, clientConfig, config.ClusterName, rcw.requeueLimit, config.ClusterDomain)
 	if err != nil {
 		return err
 	}
diff --git a/controller/cmd/service-mirror/events_formatting.go b/controller/cmd/service-mirror/events_formatting.go
index 957fefc81..6a2cba1bf 100644
--- a/controller/cmd/service-mirror/events_formatting.go
+++ b/controller/cmd/service-mirror/events_formatting.go
@@ -73,13 +73,17 @@ func (rsu RemoteServiceUpdated) String() string {
 }
 
 func (rsd RemoteServiceDeleted) String() string {
-	return fmt.Sprintf("RemoteServiceDeleted: {name: %s, namespace: %s, gatewayData: %s}", rsd.Name, rsd.Namespace, rsd.GatewayData)
+	return fmt.Sprintf("RemoteServiceDeleted: {name: %s, namespace: %s }", rsd.Name, rsd.Namespace)
 }
 
-func (rgd *RemoteGatewayDeleted) String() string {
+func (rgd RemoteGatewayDeleted) String() string {
 	return fmt.Sprintf("RemoteGatewayDeleted: {gatewayData: %s}", rgd.gatewayData)
 }
 
+func (rgd *RemoteGatewayCreated) String() string {
+	return fmt.Sprintf("RemoteGatewayCreated: {gatewaySpec: %s}", rgd.gatewaySpec)
+}
+
 func (rgu RemoteGatewayUpdated) String() string {
 	var services []string
 
@@ -89,10 +93,6 @@ func (rgu RemoteGatewayUpdated) String() string {
 	return fmt.Sprintf("RemoteGatewayUpdated: {gatewaySpec: %s, affectedServices: [%s]}", rgu.gatewaySpec, strings.Join(services, ","))
 }
 
-func (cgu ConsiderGatewayUpdateDispatch) String() string {
-	return fmt.Sprintf("ConsiderGatewayUpdateDispatch: {maybeGateway: %s}", formatService(cgu.maybeGateway))
-}
-
 func (cgu ClusterUnregistered) String() string {
 	return "ClusterUnregistered: {}"
 }
@@ -115,22 +115,18 @@ func (od OnDeleteCalled) String() string {
 
 //Events for probe manager
 
-func (msu MirroredServiceUnpaired) String() string {
-	return fmt.Sprintf("MirroredServiceUnpaired: {serviceName: %s, serviceNamespace: %s, gatewayName: %s, gatewayNs: %s, clusterName: %s}", msu.serviceName, msu.serviceNamespace, msu.gatewayName, msu.gatewayNs, msu.clusterName)
-}
-
-func (msp MirroredServicePaired) String() string {
-	return fmt.Sprintf("MirroredServicePaired: {serviceName: %s, serviceNamespace: %s, gatewaySpec: %s}", msp.serviceName, msp.serviceNamespace, msp.GatewaySpec)
+func (ps probeSpec) String() string {
+	return fmt.Sprintf("ProbeSpec: {path: %s, port: %d, period: %d}", ps.path, ps.port, ps.periodInSeconds)
 }
 
-func (gtwd GatewayDeleted) String() string {
-	return fmt.Sprintf("GatewayDeleted: {gatewayName: %s, gatewayNs: %s, clusterName: %s}", gtwd.gatewayName, gtwd.gatewayNs, gtwd.clusterName)
+func (gmc GatewayMirrorCreated) String() string {
+	return fmt.Sprintf("GatewayMirrorCreated: {gatewayName: %s, gatewayNamespace: %s, clusterName: %s, probeSpec: %s}", gmc.gatewayName, gmc.gatewayNamespace, gmc.clusterName, gmc.probeSpec)
 }
 
-func (cnr ClusterNotRegistered) String() string {
-	return fmt.Sprintf("ClusterNotRegistered: {clusterName: %s}", cnr.clusterName)
+func (gmd GatewayMirrorDeleted) String() string {
+	return fmt.Sprintf("GatewayMirrorDeleted: {gatewayName: %s, gatewayNamespace: %s, clusterName: %s}", gmd.gatewayName, gmd.gatewayNamespace, gmd.clusterName)
 }
 
-func (gtwu GatewayUpdated) String() string {
-	return fmt.Sprintf("GatewayUpdated: {gatewaySpec: %s}", gtwu.GatewaySpec)
+func (gmu GatewayMirrorUpdated) String() string {
+	return fmt.Sprintf("GatewayMirrorUpdated: {gatewayName: %s, gatewayNamespace: %s, clusterName: %s, probeSpec: %s}", gmu.gatewayName, gmu.gatewayNamespace, gmu.clusterName, gmu.probeSpec)
 }
diff --git a/controller/cmd/service-mirror/main.go b/controller/cmd/service-mirror/main.go
index e59012592..831d8cf66 100644
--- a/controller/cmd/service-mirror/main.go
+++ b/controller/cmd/service-mirror/main.go
@@ -2,8 +2,8 @@ package servicemirror
 
 import (
 	"context"
-	"errors"
 	"flag"
+	"fmt"
 	"os"
 	"os/signal"
 	"syscall"
@@ -19,27 +19,31 @@ import (
 	"k8s.io/client-go/kubernetes"
 )
 
-type chanProbeEventSink struct{ sender func(event interface{}) }
+func initLocalResourceInformer(api kubernetes.Interface, namespace string, resource k8s.APIResource) (cache.SharedIndexInformer, error) {
+	sharedInformers := informers.NewSharedInformerFactoryWithOptions(api, 10*time.Minute, informers.WithNamespace(namespace))
 
-func (s *chanProbeEventSink) send(event interface{}) {
-	s.sender(event)
-}
+	var informer cache.SharedIndexInformer
 
-func initLocalSecretsInformer(api kubernetes.Interface, namespace string) (cache.SharedIndexInformer, error) {
-	sharedInformers := informers.NewSharedInformerFactoryWithOptions(api, 10*time.Minute, informers.WithNamespace(namespace))
+	switch resource {
+	case k8s.Svc:
+		informer = sharedInformers.Core().V1().Services().Informer()
+	case k8s.Secret:
+		informer = sharedInformers.Core().V1().Secrets().Informer()
+	default:
+		return nil, fmt.Errorf("cannot instantiate local informer for %v", resource)
 
-	informer := sharedInformers.Core().V1().Secrets().Informer()
+	}
 
 	sharedInformers.Start(nil)
 
 	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
 	defer cancel()
 
-	log.Infof("waiting for local namespaced secrets informer caches to sync")
+	log.Infof("waiting for local namespaced %v informer caches to sync", resource)
 	if !cache.WaitForCacheSync(ctx.Done(), informer.HasSynced) {
-		return nil, errors.New("failed to sync local namespaced secrets informer caches")
+		return nil, fmt.Errorf("failed to sync local namespaced %v informer caches", resource)
 	}
-	log.Infof("local namespaced secrets informer  caches synced")
+	log.Infof("local namespaced %v informer  caches synced", resource)
 	return informer, nil
 }
 
@@ -70,17 +74,21 @@ func Main(args []string) {
 		log.Fatalf("Failed to initialize K8s API: %s", err)
 	}
 
-	secretsInformer, err := initLocalSecretsInformer(k8sAPI.Client, *namespace)
+	secretsInformer, err := initLocalResourceInformer(k8sAPI.Client, *namespace, k8s.Secret)
+	if err != nil {
+		log.Fatalf("Failed to initialize secret informer: %s", err)
+	}
+	svcInformer, err := initLocalResourceInformer(k8sAPI.Client, *namespace, k8s.Svc)
 
 	if err != nil {
-		log.Fatalf("Failed to initialize secrets informer: %s", err)
+		log.Fatalf("Failed to initialize service informer: %s", err)
 	}
 
-	probeManager := NewProbeManager(k8sAPI)
+	probeManager := NewProbeManager(svcInformer)
 	probeManager.Start()
 
 	k8sAPI.Sync(nil)
-	watcher := NewRemoteClusterConfigWatcher(*namespace, secretsInformer, k8sAPI, *requeueLimit, &chanProbeEventSink{probeManager.enqueueEvent})
+	watcher := NewRemoteClusterConfigWatcher(*namespace, secretsInformer, k8sAPI, *requeueLimit)
 	log.Info("Started cluster config watcher")
 
 	go admin.StartServer(*metricsAddr)
diff --git a/controller/cmd/service-mirror/metrics.go b/controller/cmd/service-mirror/metrics.go
index bb7612603..f82ade2c3 100644
--- a/controller/cmd/service-mirror/metrics.go
+++ b/controller/cmd/service-mirror/metrics.go
@@ -15,7 +15,6 @@ const (
 )
 
 type probeMetricVecs struct {
-	services  *prometheus.GaugeVec
 	alive     *prometheus.GaugeVec
 	latencies *prometheus.HistogramVec
 	enqueues  *prometheus.CounterVec
@@ -24,7 +23,6 @@ type probeMetricVecs struct {
 }
 
 type probeMetrics struct {
-	services   prometheus.Gauge
 	alive      prometheus.Gauge
 	latencies  prometheus.Observer
 	probes     *prometheus.CounterVec
@@ -58,14 +56,6 @@ func newProbeMetricVecs() probeMetricVecs {
 		[]string{eventTypeLabelName},
 	)
 
-	services := promauto.NewGaugeVec(
-		prometheus.GaugeOpts{
-			Name: "num_mirrored_services",
-			Help: "A gauge for the current number of mirrored services associated with a gateway",
-		},
-		labelNames,
-	)
-
 	alive := promauto.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Name: "gateway_alive",
@@ -89,7 +79,6 @@ func newProbeMetricVecs() probeMetricVecs {
 		labelNames)
 
 	return probeMetricVecs{
-		services:  services,
 		alive:     alive,
 		latencies: latencies,
 		enqueues:  enqueues,
@@ -110,7 +99,6 @@ func (mv probeMetricVecs) newWorkerMetrics(gatewayNamespace, gatewayName, remote
 		return nil, err
 	}
 	return &probeMetrics{
-		services:  mv.services.With(labels),
 		alive:     mv.alive.With(labels),
 		latencies: mv.latencies.With(labels),
 		probes:    curriedProbes,
@@ -127,9 +115,6 @@ func (mv probeMetricVecs) unregister(gatewayNamespace, gatewayName, remoteCluste
 		gatewayClusterName:    remoteClusterName,
 	}
 
-	if !mv.services.Delete(labels) {
-		logging.Warnf("unable to delete num_mirrored_services metric with labels %s", labels)
-	}
 	if !mv.alive.Delete(labels) {
 		logging.Warnf("unable to delete gateway_alive metric with labels %s", labels)
 	}
diff --git a/controller/cmd/service-mirror/probe_manager.go b/controller/cmd/service-mirror/probe_manager.go
index 9bf08782d..c13842976 100644
--- a/controller/cmd/service-mirror/probe_manager.go
+++ b/controller/cmd/service-mirror/probe_manager.go
@@ -2,83 +2,69 @@ package servicemirror
 
 import (
 	"fmt"
-
 	"github.com/prometheus/client_golang/prometheus"
+	"k8s.io/client-go/tools/cache"
+	"strconv"
 
-	"github.com/linkerd/linkerd2/controller/k8s"
 	consts "github.com/linkerd/linkerd2/pkg/k8s"
 	log "github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/labels"
 )
 
 const probeChanBufferSize = 500
 
 // ProbeManager takes care of managing the lifecycle of probe workers
 type ProbeManager struct {
-	k8sAPI       *k8s.API
-	probeWorkers map[string]*ProbeWorker
-
-	events     chan interface{}
-	metricVecs *probeMetricVecs
-	done       chan struct{}
+	probeWorkers            map[string]*ProbeWorker
+	mirroredGatewayInformer cache.SharedIndexInformer
+	events                  chan interface{}
+	metricVecs              *probeMetricVecs
+	done                    chan struct{}
 }
 
-// MirroredServiceUnpaired is emitted when a service is no longer mirrored
-type MirroredServiceUnpaired struct {
-	serviceName      string
-	serviceNamespace string
+// GatewayMirrorCreated is observed when a mirror of a remote gateway is created locally
+type GatewayMirrorCreated struct {
 	gatewayName      string
-	gatewayNs        string
+	gatewayNamespace string
 	clusterName      string
+	probeSpec
 }
 
-// MirroredServicePaired is emitted when a new mirrored service is created
-type MirroredServicePaired struct {
-	serviceName      string
-	serviceNamespace string
-	GatewaySpec
-}
-
-// GatewayDeleted is emitted when a gateway is deleted
-type GatewayDeleted struct {
-	gatewayName string
-	gatewayNs   string
-	clusterName string
-}
-
-// ClusterNotRegistered is is emitted when the cluster is not monitored anymore
-type ClusterNotRegistered struct {
-	clusterName string
+// GatewayMirrorDeleted is emitted when a mirror of a remote gateway is deleted
+type GatewayMirrorDeleted struct {
+	gatewayName      string
+	gatewayNamespace string
+	clusterName      string
 }
 
-// GatewayUpdated is emitted when something about the gateway is updated (i.e. its external IP(s))
-type GatewayUpdated struct {
-	GatewaySpec
+// GatewayMirrorUpdated is emitted when the mirror of a remote gateway has changed
+type GatewayMirrorUpdated struct {
+	gatewayName      string
+	gatewayNamespace string
+	clusterName      string
+	probeSpec
 }
 
 // NewProbeManager creates a new probe manager
-func NewProbeManager(k8sAPI *k8s.API) *ProbeManager {
+func NewProbeManager(mirroredGatewayInformer cache.SharedIndexInformer) *ProbeManager {
 	metricVecs := newProbeMetricVecs()
 	return &ProbeManager{
-		k8sAPI:       k8sAPI,
-		probeWorkers: make(map[string]*ProbeWorker),
-		events:       make(chan interface{}, probeChanBufferSize),
-		metricVecs:   &metricVecs,
-		done:         make(chan struct{}),
+		mirroredGatewayInformer: mirroredGatewayInformer,
+		probeWorkers:            make(map[string]*ProbeWorker),
+		events:                  make(chan interface{}, probeChanBufferSize),
+		metricVecs:              &metricVecs,
+		done:                    make(chan struct{}),
 	}
 }
 
 func eventTypeString(ev interface{}) string {
 	switch ev.(type) {
-	case *MirroredServicePaired:
-		return "MirroredServicePaired"
-	case *MirroredServiceUnpaired:
-		return "MirroredServiceUnpaired"
-	case *GatewayUpdated:
-		return "GatewayUpdated"
-	case *ClusterNotRegistered:
-		return "ClusterNotRegistered"
+	case *GatewayMirrorCreated:
+		return "GatewayMirrorCreated"
+	case *GatewayMirrorDeleted:
+		return "GatewayMirrorDeleted"
+	case *GatewayMirrorUpdated:
+		return "GatewayMirrorUpdated"
 	default:
 		return "Unknown"
 	}
@@ -95,99 +81,51 @@ func probeKey(gatewayNamespace string, gatewayName string, clusterName string) s
 
 func (m *ProbeManager) handleEvent(ev interface{}) {
 	switch ev := ev.(type) {
-	case *MirroredServicePaired:
-		m.handleMirroredServicePaired(ev)
-	case *MirroredServiceUnpaired:
-		m.handleMirroredServiceUnpaired(ev)
-	case *GatewayUpdated:
-		m.handleGatewayUpdated(ev)
-	case *GatewayDeleted:
-		m.handleGatewayDeleted(ev)
-	case *ClusterNotRegistered:
-		m.handleClusterNotRegistered(ev)
+	case *GatewayMirrorCreated:
+		m.handleGatewayMirrorCreated(ev)
+	case *GatewayMirrorUpdated:
+		m.handleGatewayMirrorUpdated(ev)
+	case *GatewayMirrorDeleted:
+		m.handleGatewayMirrorDeleted(ev)
 	default:
 		log.Errorf("Received unknown event: %v", ev)
 	}
 }
 
-func endpointAddressesToIps(addrs []corev1.EndpointAddress) []string {
-	result := []string{}
-
-	for _, a := range addrs {
-
-		result = append(result, a.IP)
-	}
-
-	return result
-}
-
-func gatewayToProbeSpec(gatewaySpec GatewaySpec) *probeSpec {
-
-	if gatewaySpec.ProbeConfig == nil {
-		return nil
-	}
-	return &probeSpec{
-		ips:             endpointAddressesToIps(gatewaySpec.addresses),
-		path:            gatewaySpec.path,
-		port:            gatewaySpec.port,
-		periodInSeconds: gatewaySpec.periodInSeconds,
-		gatewayIdentity: gatewaySpec.identity,
-	}
+func (m *ProbeManager) handleGatewayMirrorDeleted(event *GatewayMirrorDeleted) {
+	probeKey := probeKey(event.gatewayNamespace, event.gatewayName, event.clusterName)
+	m.stopProbe(probeKey)
 }
 
-func (m *ProbeManager) handleMirroredServicePaired(event *MirroredServicePaired) {
+func (m *ProbeManager) handleGatewayMirrorCreated(event *GatewayMirrorCreated) {
 	probeKey := probeKey(event.gatewayNamespace, event.gatewayName, event.clusterName)
 	worker, ok := m.probeWorkers[probeKey]
 	if ok {
-		log.Debugf("Probe worker %s already exists", probeKey)
-		worker.PairService(event.serviceName, event.serviceNamespace)
+		log.Debugf("There is already a probe worker for %s. Updating instead of creating", probeKey)
+		worker.UpdateProbeSpec(&event.probeSpec)
 	} else {
 		log.Debugf("Creating probe worker %s", probeKey)
 		probeMetrics, err := m.metricVecs.newWorkerMetrics(event.gatewayNamespace, event.gatewayName, event.clusterName)
 		if err != nil {
 			log.Errorf("Could not crete probe metrics: %s", err)
 		} else {
-			probeSpec := gatewayToProbeSpec(event.GatewaySpec)
-			if probeSpec != nil {
-				localGatewayName := fmt.Sprintf("%s-%s", event.gatewayName, event.clusterName)
-				worker = NewProbeWorker(localGatewayName, probeSpec, probeMetrics, probeKey)
-				worker.PairService(event.serviceName, event.serviceNamespace)
-				m.probeWorkers[probeKey] = worker
-				worker.Start()
-			} else {
-				log.Debugf("No probe spec for: %s", probeKey)
-			}
-		}
-	}
-}
-
-func (m *ProbeManager) handleMirroredServiceUnpaired(event *MirroredServiceUnpaired) {
-	probeKey := probeKey(event.gatewayNs, event.gatewayName, event.clusterName)
-	worker, ok := m.probeWorkers[probeKey]
-	if ok {
-		worker.UnPairService(event.serviceName, event.serviceNamespace)
-		if worker.NumPairedServices() < 1 {
-			log.Debugf("Probe worker's %s associated services dropped to 0, cleaning up", probeKey)
-			worker.Stop()
-			delete(m.probeWorkers, probeKey)
+			localGatewayName := fmt.Sprintf("%s-%s", event.gatewayName, event.clusterName)
+			worker = NewProbeWorker(localGatewayName, &event.probeSpec, probeMetrics, probeKey)
+			m.probeWorkers[probeKey] = worker
+			worker.Start()
 		}
-	} else {
-		log.Debugf("Could not find a worker for %s while handling MirroredServiceUnpaired event", probeKey)
 	}
 }
 
-func (m *ProbeManager) handleGatewayUpdated(event *GatewayUpdated) {
+func (m *ProbeManager) handleGatewayMirrorUpdated(event *GatewayMirrorUpdated) {
 	probeKey := probeKey(event.gatewayNamespace, event.gatewayName, event.clusterName)
 	worker, ok := m.probeWorkers[probeKey]
 	if ok {
-		probeSpec := gatewayToProbeSpec(event.GatewaySpec)
-		if probeSpec != nil {
-			worker.UpdateProbeSpec(probeSpec)
-		} else {
-			log.Debugf("No probe spec for: %s", probeKey)
+		if worker.probeSpec.port != event.port || worker.probeSpec.periodInSeconds != event.periodInSeconds || worker.probeSpec.path != event.path {
+			worker.UpdateProbeSpec(&event.probeSpec)
 		}
 	} else {
-		log.Debugf("Could not find a worker for %s while handling GatewayUpdated event", probeKey)
+		log.Debugf("Could not find a worker for %s while handling GatewayMirrorUpdated event", probeKey)
 	}
 }
 
@@ -200,33 +138,6 @@ func (m *ProbeManager) stopProbe(key string) {
 	}
 }
 
-func (m *ProbeManager) handleGatewayDeleted(event *GatewayDeleted) {
-	probeKey := probeKey(event.gatewayNs, event.gatewayName, event.clusterName)
-	m.stopProbe(probeKey)
-}
-
-func (m *ProbeManager) handleClusterNotRegistered(event *ClusterNotRegistered) {
-	matchLabels := map[string]string{
-		consts.MirroredResourceLabel:  "true",
-		consts.RemoteClusterNameLabel: event.clusterName,
-	}
-
-	services, err := m.k8sAPI.Svc().Lister().List(labels.Set(matchLabels).AsSelector())
-	if err != nil {
-		log.Errorf("Was not able to unregister cluster %s, %s", event.clusterName, err)
-	}
-
-	stopped := make(map[string]bool)
-	for _, svc := range services {
-		probeKey := probeKey(svc.Labels[consts.RemoteGatewayNsLabel], svc.Labels[consts.RemoteGatewayNameLabel], event.clusterName)
-
-		if _, ok := stopped[probeKey]; !ok {
-			m.stopProbe(probeKey)
-			stopped[probeKey] = true
-		}
-	}
-}
-
 func (m *ProbeManager) run() {
 	for {
 		select {
@@ -244,8 +155,113 @@ func (m *ProbeManager) run() {
 	}
 }
 
+func extractProbeSpec(svc *corev1.Service) (*probeSpec, error) {
+	path, hasPath := svc.Annotations[consts.MirroredGatewayProbePath]
+	if !hasPath {
+		return nil, fmt.Errorf("mirrored Gateway service is missing %s annotation", consts.MirroredGatewayProbePath)
+	}
+
+	probePort, err := extractPort(svc.Spec.Ports, consts.ProbePortName)
+	if err != nil {
+		return nil, fmt.Errorf("%s: %s", svc.Name, err)
+	}
+
+	period, hasPeriod := svc.Annotations[consts.MirroredGatewayProbePeriod]
+	if !hasPeriod {
+		return nil, fmt.Errorf("mirrored Gateway service is missing %s annotation", consts.MirroredGatewayProbePeriod)
+	}
+
+	probePeriod, err := strconv.ParseUint(period, 10, 32)
+	if err != nil {
+		return nil, err
+	}
+
+	return &probeSpec{
+		path:            path,
+		port:            probePort,
+		periodInSeconds: uint32(probePeriod),
+	}, nil
+
+}
+
 // Start starts the probe manager
 func (m *ProbeManager) Start() {
+	m.mirroredGatewayInformer.AddEventHandler(
+		cache.FilteringResourceEventHandler{
+			FilterFunc: func(obj interface{}) bool {
+				switch object := obj.(type) {
+				case *corev1.Service:
+					_, isMirrorGateway := object.Labels[consts.MirroredGatewayLabel]
+					return isMirrorGateway
+
+				case cache.DeletedFinalStateUnknown:
+					if svc, ok := object.Obj.(*corev1.Service); ok {
+						_, isMirrorGateway := svc.Labels[consts.MirroredGatewayLabel]
+						return isMirrorGateway
+					}
+					return false
+				default:
+					return false
+				}
+			},
+
+			Handler: cache.ResourceEventHandlerFuncs{
+				AddFunc: func(obj interface{}) {
+					service := obj.(*corev1.Service)
+					spec, err := extractProbeSpec(service)
+					if err != nil {
+						log.Errorf("Could not parse probe spec %s", err)
+					} else {
+						m.enqueueEvent(&GatewayMirrorCreated{
+							gatewayName:      service.Annotations[consts.MirroredGatewayRemoteName],
+							gatewayNamespace: service.Annotations[consts.MirroredGatewayRemoteNameSpace],
+							clusterName:      service.Labels[consts.RemoteClusterNameLabel],
+							probeSpec:        *spec,
+						})
+					}
+				},
+				DeleteFunc: func(obj interface{}) {
+					service, ok := obj.(*corev1.Service)
+					if !ok {
+						tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+						if !ok {
+							log.Errorf("couldn't get object from DeletedFinalStateUnknown %#v", obj)
+							return
+						}
+						service, ok = tombstone.Obj.(*corev1.Service)
+						if !ok {
+							log.Errorf("DeletedFinalStateUnknown contained object that is not a Secret %#v", obj)
+							return
+						}
+					}
+
+					m.enqueueEvent(&GatewayMirrorDeleted{
+						gatewayName:      service.Annotations[consts.MirroredGatewayRemoteName],
+						gatewayNamespace: service.Annotations[consts.MirroredGatewayRemoteNameSpace],
+						clusterName:      service.Labels[consts.RemoteClusterNameLabel],
+					})
+				},
+				UpdateFunc: func(old, new interface{}) {
+					oldService := old.(*corev1.Service)
+					newService := new.(*corev1.Service)
+
+					if oldService.ResourceVersion != newService.ResourceVersion {
+						spec, err := extractProbeSpec(newService)
+						if err != nil {
+							log.Errorf("Could not parse probe spec %s", err)
+						} else {
+							m.enqueueEvent(&GatewayMirrorUpdated{
+								gatewayName:      newService.Annotations[consts.MirroredGatewayRemoteName],
+								gatewayNamespace: newService.Annotations[consts.MirroredGatewayRemoteNameSpace],
+								clusterName:      newService.Labels[consts.RemoteClusterNameLabel],
+								probeSpec:        *spec,
+							})
+						}
+					}
+				},
+			},
+		},
+	)
 	go m.run()
 }
 
diff --git a/controller/cmd/service-mirror/probe_worker.go b/controller/cmd/service-mirror/probe_worker.go
index 3886cbb6d..1253d2797 100644
--- a/controller/cmd/service-mirror/probe_worker.go
+++ b/controller/cmd/service-mirror/probe_worker.go
@@ -13,22 +13,19 @@ import (
 const httpGatewayTimeoutMillis = 50000
 
 type probeSpec struct {
-	ips             []string
 	path            string
 	port            uint32
 	periodInSeconds uint32
-	gatewayIdentity string
 }
 
 // ProbeWorker is responsible for monitoring gateways using a probe specification
 type ProbeWorker struct {
 	localGatewayName string
 	*sync.RWMutex
-	probeSpec      *probeSpec
-	pairedServices map[string]struct{}
-	stopCh         chan struct{}
-	metrics        *probeMetrics
-	log            *logging.Entry
+	probeSpec *probeSpec
+	stopCh    chan struct{}
+	metrics   *probeMetrics
+	log       *logging.Entry
 }
 
 // NewProbeWorker creates a new probe worker associated with a particular gateway
@@ -37,7 +34,6 @@ func NewProbeWorker(localGatewayName string, spec *probeSpec, metrics *probeMetr
 		localGatewayName: localGatewayName,
 		RWMutex:          &sync.RWMutex{},
 		probeSpec:        spec,
-		pairedServices:   make(map[string]struct{}),
 		stopCh:           make(chan struct{}),
 		metrics:          metrics,
 		log: logging.WithFields(logging.Fields{
@@ -46,29 +42,6 @@ func NewProbeWorker(localGatewayName string, spec *probeSpec, metrics *probeMetr
 	}
 }
 
-// NumPairedServices returns the number of paired services for this probe worker
-func (pw *ProbeWorker) NumPairedServices() int {
-	return len(pw.pairedServices)
-}
-
-// PairService increments the number of services that are routed by the gateway
-func (pw *ProbeWorker) PairService(serviceName, serviceNamespace string) {
-	svcKey := fmt.Sprintf("%s-%s", serviceNamespace, serviceName)
-	if _, ok := pw.pairedServices[svcKey]; !ok {
-		pw.pairedServices[svcKey] = struct{}{}
-		pw.metrics.services.Set(float64(len(pw.pairedServices)))
-	}
-}
-
-// UnPairService decrements the number of services that are routed by the gateway
-func (pw *ProbeWorker) UnPairService(serviceName, serviceNamespace string) {
-	svcKey := fmt.Sprintf("%s-%s", serviceNamespace, serviceName)
-	if _, ok := pw.pairedServices[svcKey]; ok {
-		delete(pw.pairedServices, svcKey)
-		pw.metrics.services.Set(float64(len(pw.pairedServices)))
-	}
-}
-
 // UpdateProbeSpec is used to update the probe specification when something about the gateway changes
 func (pw *ProbeWorker) UpdateProbeSpec(spec *probeSpec) {
 	pw.Lock()
diff --git a/pkg/healthcheck/healthcheck.go b/pkg/healthcheck/healthcheck.go
index 1bf12013d..3e4ebb665 100644
--- a/pkg/healthcheck/healthcheck.go
+++ b/pkg/healthcheck/healthcheck.go
@@ -261,6 +261,18 @@ func (e *SkipError) Error() string {
 	return e.Reason
 }
 
+// VerboseSuccess implements the error interface but represents a success with
+// a message.
+type VerboseSuccess struct {
+	Message string
+}
+
+// Error satisfies the error interface for VerboseSuccess.  Since VerboseSuccess
+// does not actually represent a failure, this returns the empty string.
+func (e *VerboseSuccess) Error() string {
+	return ""
+}
+
 type checker struct {
 	// description is the short description that's printed to the command line
 	// when the check is executed
@@ -1317,18 +1329,20 @@ func (hc *HealthChecker) runCheck(categoryID CategoryID, c *checker, observer Ch
 			log.Debugf("Skipping check: %s. Reason: %s", c.description, se.Reason)
 			return true
 		}
-		if err != nil {
-			err = &CategoryError{categoryID, err}
-		}
+
 		checkResult := &CheckResult{
 			Category:    categoryID,
 			Description: c.description,
 			HintAnchor:  c.hintAnchor,
 			Warning:     c.warning,
-			Err:         err,
+		}
+		if vs, ok := err.(*VerboseSuccess); ok {
+			checkResult.Description = fmt.Sprintf("%s\n%s", checkResult.Description, vs.Message)
+		} else if err != nil {
+			checkResult.Err = &CategoryError{categoryID, err}
 		}
 
-		if err != nil && time.Now().Before(c.retryDeadline) {
+		if checkResult.Err != nil && time.Now().Before(c.retryDeadline) {
 			checkResult.Retry = true
 			if !c.surfaceErrorOnRetry {
 				checkResult.Err = errors.New("waiting for check to complete")
@@ -1341,7 +1355,7 @@ func (hc *HealthChecker) runCheck(categoryID CategoryID, c *checker, observer Ch
 		}
 
 		observer(checkResult)
-		return err == nil
+		return checkResult.Err == nil
 	}
 }
 
diff --git a/pkg/healthcheck/healthcheck_multicluster.go b/pkg/healthcheck/healthcheck_multicluster.go
index 11f023859..cc208b322 100644
--- a/pkg/healthcheck/healthcheck_multicluster.go
+++ b/pkg/healthcheck/healthcheck_multicluster.go
@@ -10,6 +10,7 @@ import (
 
 	pb "github.com/linkerd/linkerd2/controller/gen/public"
 	sm "github.com/linkerd/linkerd2/pkg/servicemirror"
+	tsclient "github.com/servicemeshinterface/smi-sdk-go/pkg/gen/client/split/clientset/versioned"
 	corev1 "k8s.io/api/core/v1"
 
 	"github.com/linkerd/linkerd2/pkg/k8s"
@@ -143,6 +144,14 @@ func (hc *HealthChecker) multiClusterCategory() category {
 					return hc.checkRemoteClusterAnchors()
 				},
 			},
+			{
+				description: "multicluster daisy chaining is avoided",
+				hintAnchor:  "l5d-multicluster-daisy-chaining",
+				warning:     true,
+				check: func(ctx context.Context) error {
+					return hc.checkDaisyChains()
+				},
+			},
 		},
 	}
 }
@@ -321,6 +330,70 @@ func (hc *HealthChecker) checkRemoteClusterAnchors() error {
 	return nil
 }
 
+func serviceExported(svc corev1.Service) bool {
+	_, hasGtwName := svc.Annotations[k8s.GatewayNameAnnotation]
+	_, hasGtwNs := svc.Annotations[k8s.GatewayNsAnnotation]
+	return hasGtwName && hasGtwNs
+}
+
+func (hc *HealthChecker) checkDaisyChains() error {
+	if hc.Options.ShouldCheckMulticluster {
+		errs := []error{}
+
+		svcs, err := hc.kubeAPI.CoreV1().Services(metav1.NamespaceAll).List(metav1.ListOptions{})
+		if err != nil {
+			return err
+		}
+		for _, svc := range svcs.Items {
+			_, isMirror := svc.Labels[k8s.MirroredResourceLabel]
+			if isMirror && serviceExported(svc) {
+				errs = append(errs, fmt.Errorf("mirror service %s.%s is exported", svc.Name, svc.Namespace))
+			}
+		}
+
+		ts, err := tsclient.NewForConfig(hc.kubeAPI.Config)
+		if err != nil {
+			return err
+		}
+		splits, err := ts.SplitV1alpha1().TrafficSplits(metav1.NamespaceAll).List(metav1.ListOptions{})
+		if err != nil {
+			return err
+		}
+		for _, split := range splits.Items {
+			apex, err := hc.kubeAPI.CoreV1().Services(split.Namespace).Get(split.Spec.Service, metav1.GetOptions{})
+			if err != nil {
+				return err
+			}
+			if serviceExported(*apex) {
+				for _, backend := range split.Spec.Backends {
+					if backend.Weight.IsZero() {
+						continue
+					}
+					leaf, err := hc.kubeAPI.CoreV1().Services(split.Namespace).Get(backend.Service, metav1.GetOptions{})
+					if err != nil {
+						return err
+					}
+					_, isMirror := leaf.Labels[k8s.MirroredResourceLabel]
+					if isMirror {
+						errs = append(errs, fmt.Errorf("exported service %s.%s routes to mirror service %s.%s via traffic split %s.%s",
+							apex.Name, apex.Namespace, leaf.Name, leaf.Namespace, split.Name, split.Namespace,
+						))
+					}
+				}
+			}
+		}
+		if len(errs) > 0 {
+			messages := []string{}
+			for _, err := range errs {
+				messages = append(messages, fmt.Sprintf("* %s", err.Error()))
+			}
+			return errors.New(strings.Join(messages, "\n"))
+		}
+		return nil
+	}
+	return &SkipError{Reason: "not checking muticluster"}
+}
+
 func (hc *HealthChecker) checkRemoteClusterConnectivity() error {
 	if hc.Options.ShouldCheckMulticluster {
 		options := metav1.ListOptions{
@@ -399,11 +472,14 @@ func (hc *HealthChecker) checkRemoteClusterGatewaysHealth(ctx context.Context) e
 		}
 
 		var deadGateways []string
+		var aliveGateways []string
 		if len(rsp.GetOk().GatewaysTable.Rows) == 0 {
 			return &SkipError{Reason: "no remote gateways"}
 		}
 		for _, gtw := range rsp.GetOk().GatewaysTable.Rows {
-			if !gtw.Alive {
+			if gtw.Alive {
+				aliveGateways = append(aliveGateways, fmt.Sprintf("\t* cluster: [%s], gateway: [%s/%s]", gtw.ClusterName, gtw.Namespace, gtw.Name))
+			} else {
 				deadGateways = append(deadGateways, fmt.Sprintf("* cluster: [%s], gateway: [%s/%s]", gtw.ClusterName, gtw.Namespace, gtw.Name))
 			}
 		}
@@ -411,7 +487,7 @@ func (hc *HealthChecker) checkRemoteClusterGatewaysHealth(ctx context.Context) e
 		if len(deadGateways) > 0 {
 			return fmt.Errorf("Some gateways are not alive:\n\t%s", strings.Join(deadGateways, "\n\t"))
 		}
-		return nil
+		return &VerboseSuccess{Message: strings.Join(aliveGateways, "\n")}
 	}
 	return &SkipError{Reason: "not checking muticluster"}
 }
diff --git a/pkg/k8s/labels.go b/pkg/k8s/labels.go
index f8ab3b1ae..f743f1ebe 100644
--- a/pkg/k8s/labels.go
+++ b/pkg/k8s/labels.go
@@ -371,6 +371,21 @@ const (
 	// of a mirroring operation (can be a namespace or a service)
 	MirroredResourceLabel = SvcMirrorPrefix + "/mirrored-service"
 
+	// MirroredGatewayLabel indicates that this is a mirrored gateway
+	MirroredGatewayLabel = SvcMirrorPrefix + "/mirrored-gateway"
+
+	// MirroredGatewayProbePeriod specifies the probe period for the gateway mirror
+	MirroredGatewayProbePeriod = SvcMirrorPrefix + "/mirrored-gateway-probe-period"
+
+	// MirroredGatewayProbePath specifies the probe path for the gateway mirror
+	MirroredGatewayProbePath = SvcMirrorPrefix + "/mirrored-gateway-probe-path"
+
+	// MirroredGatewayRemoteName specifies the name of the remote gateway that has been mirrored
+	MirroredGatewayRemoteName = SvcMirrorPrefix + "/mirrored-gateway-remote-name"
+
+	// MirroredGatewayRemoteNameSpace specifies the namespace of the remote gateway that has been mirrored
+	MirroredGatewayRemoteNameSpace = SvcMirrorPrefix + "/mirrored-gateway-remote-namespace"
+
 	// MulticlusterGatewayAnnotation indicates that this service is a
 	// gateway
 	MulticlusterGatewayAnnotation = SvcMirrorPrefix + "/multicluster-gateway"
@@ -407,9 +422,6 @@ const (
 	// GatewayIdentity can be found on the remote gateway service
 	GatewayIdentity = SvcMirrorPrefix + "/gateway-identity"
 
-	// GatewayProbePort the port on which the gateway can be probed
-	GatewayProbePort = SvcMirrorPrefix + "/probe-port"
-
 	// GatewayProbePeriod the interval at which the health of the gateway should be probed
 	GatewayProbePeriod = SvcMirrorPrefix + "/probe-period"
 
@@ -423,6 +435,9 @@ const (
 	// GatewayPortName is the name of the incoming port of the gateway
 	GatewayPortName = "incoming-port"
 
+	// ProbePortName is the name of the probe port of the gateway
+	ProbePortName = "probe-port"
+
 	// ServiceMirrorLabel is the value used in the controller component label
 	ServiceMirrorLabel = "servicemirror"
 )


Test patch:
diff --git a/bin/_test-run.sh b/bin/_test-run.sh
index 972949dff..6c3f9e630 100755
--- a/bin/_test-run.sh
+++ b/bin/_test-run.sh
@@ -3,6 +3,10 @@
 # This file is a collection of helper functions for running integration tests.
 # It is used primarily by `bin/test-run` and ci.
 
+# Override CI's `set -e` default, so we can catch errors manually and display
+# proper messages
+set +e
+
 # Returns the latest stable verson
 latest_stable() {
   curl -s https://versioncheck.linkerd.io/version.json | grep -o "stable-[0-9]*.[0-9]*.[0-9]*"
@@ -46,7 +50,6 @@ upgrade_integration_tests() {
     # fail subsequent tests. `cleanup` is not called if this test failed so that
     # there is a chance to debug the problem
     run_upgrade_test "$linkerd_namespace"-upgrade
-    exit_on_err "can't upgrade to version $linkerd_version"
     cleanup
 }
 
@@ -56,9 +59,7 @@ helm_upgrade_integration_tests() {
     helm_release_name=$linkerd_namespace-test
 
     run_helm_upgrade_test
-    exit_on_err 'error testing Helm upgrade'
     helm_cleanup
-    exit_on_err 'error cleaning up Helm upgrade'
     # clean the data plane test resources
     cleanup
 }
@@ -69,40 +70,30 @@ helm_integration_tests() {
     helm_release_name=$linkerd_namespace-test
 
     run_helm_test
-    exit_on_err 'error testing Helm'
     helm_cleanup
-    exit_on_err 'error cleaning up Helm'
     # clean the data plane test resources
     cleanup
 }
 
 uninstall_integration_tests() {
     run_test "$test_directory/uninstall/uninstall_test.go" --linkerd-namespace=$linkerd_namespace --uninstall=true
-    exit_on_err 'error during uninstall tests'
     cleanup
 }
 
 deep_integration_tests() {
     run_test "$test_directory/install_test.go" --linkerd-namespace=$linkerd_namespace
-    exit_on_err 'error during install'
-
     run_test "$(go list $test_directory/.../...)" --linkerd-namespace=$linkerd_namespace
-    exit_on_err 'error during deep tests'
     cleanup
 }
 
 custom_domain_integration_tests() {
     run_test "$test_directory/install_test.go" --linkerd-namespace=$linkerd_namespace --cluster-domain="custom.domain"
-    exit_on_err 'error during install'
     cleanup
 }
 
 external_issuer_integration_tests() {
     run_test "$test_directory/install_test.go" --linkerd-namespace=$linkerd_namespace-external-issuer --external-issuer=true
-    exit_on_err 'error during install with --external-issuer=true'
-
     run_test "$test_directory/externalissuer/external_issuer_test.go" --linkerd-namespace=$linkerd_namespace-external-issuer --external-issuer=true
-    exit_on_err 'error during external issuer tests'
     cleanup
 }
 
@@ -160,24 +151,31 @@ run_test(){
 
     printf 'Test script: [%s] Params: [%s]\n' "$(basename $filename 2>/dev/null || echo $filename )" "$*"
     GO111MODULE=on go test --failfast --mod=readonly $filename --linkerd="$linkerd_path" --k8s-context="$k8s_context" --integration-tests "$@"
+    [ $? -ne 0 ] && exit 1
 }
 
 # Install the latest stable release.
 # $1 - namespace to use for the stable release
 install_stable() {
     tmp=$(mktemp -d -t l5dbin.XXX)
-    trap "rm -rf $tmp" RETURN
 
     curl -s https://run.linkerd.io/install | HOME=$tmp sh > /dev/null 2>&1
 
     local linkerd_path=$tmp/.linkerd2/bin/linkerd
     local stable_namespace=$1
     local test_app_namespace=$stable_namespace-upgrade-test
+
     (
         set -x
         "$linkerd_path" install --linkerd-namespace="$stable_namespace" | kubectl --context="$k8s_context" apply -f - 2>&1
+    )
+    exit_on_err 'install_stable() - installing stable failed'
+
+    (
+        set -x
         "$linkerd_path" check --linkerd-namespace="$stable_namespace" 2>&1
     )
+    exit_on_err 'install_stable() - linkerd check failed'
 
     #Now we need to install the app that will be used to verify that upgrade does not break anything
     kubectl --context=$k8s_context create namespace "$test_app_namespace" > /dev/null 2>&1
@@ -186,6 +184,7 @@ install_stable() {
         set -x
         "$linkerd_path" inject --linkerd-namespace="$stable_namespace" "$test_directory/testdata/upgrade_test.yaml" | kubectl --context="$k8s_context" apply --namespace="$test_app_namespace" -f - 2>&1
     )
+    exit_on_err 'install_stable() - linkerd inject failed'
 }
 
 # Run the upgrade test by upgrading the most-recent stable release to the HEAD of
@@ -200,7 +199,7 @@ run_upgrade_test() {
 }
 
 setup_helm() {
-      (
+    (
         set -e
         "$bindir"/helm-build
         "$helm_path" --kube-context=$k8s_context repo add linkerd https://helm.linkerd.io/stable
@@ -232,12 +231,21 @@ helm_cleanup() {
         # We don't have that problem with global resources, so no need to wait for them to be gone.
         kubectl wait --for=delete ns/$linkerd_namespace-helm --timeout=120s
     )
+    exit_on_err 'error cleaning up Helm'
 }
 
+# exit_on_err should be called right after a command to check the result status and eventually generate a Github error
+# annotation. Do not use after calls to `go test` as that generates its own annotations.
+# Note this should be called outside subshells in order for the script to terminate.
 exit_on_err() {
     exit_code=$?
     if [ $exit_code -ne 0 ]; then
-        printf '\n=== FAIL: %s\n' "$@"
+        export GH_ANNOTATION=${GH_ANNOTATION:-}
+        if [ -n "$GH_ANNOTATION" ]; then
+          printf '::error::%s\n' "$1"
+        else
+          printf '\n=== FAIL: %s\n' "$1"
+        fi
         exit $exit_code
     fi
 }
diff --git a/controller/cmd/service-mirror/cluster_watcher_mirroring_test.go b/controller/cmd/service-mirror/cluster_watcher_mirroring_test.go
index f95fd1d8f..468e18e0f 100644
--- a/controller/cmd/service-mirror/cluster_watcher_mirroring_test.go
+++ b/controller/cmd/service-mirror/cluster_watcher_mirroring_test.go
@@ -10,10 +10,6 @@ import (
 	"k8s.io/client-go/util/workqueue"
 )
 
-type NoOpProbeEventSink struct{}
-
-func (s *NoOpProbeEventSink) send(event interface{}) {}
-
 type mirroringTestCase struct {
 	description            string
 	environment            *testEnvironment
@@ -26,7 +22,7 @@ func (tc *mirroringTestCase) run(t *testing.T) {
 	t.Run(tc.description, func(t *testing.T) {
 
 		q := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
-		localAPI, err := tc.environment.runEnvironment(&NoOpProbeEventSink{}, q)
+		localAPI, err := tc.environment.runEnvironment(q)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -464,19 +460,12 @@ func onAddOrUpdateTestCases(isAdd bool) []mirroringTestCase {
 			environment: onAddOrUpdateExportedSvc(isAdd),
 			expectedEventsInQueue: []interface{}{&RemoteServiceCreated{
 				service: remoteService("test-service", "test-namespace", "gateway", "gateway-ns", "resVersion", nil),
-				gatewayData: &gatewayMetadata{
+				gatewayData: gatewayMetadata{
 					Name:      "gateway",
 					Namespace: "gateway-ns",
 				},
 			}},
 		},
-		{
-			description: fmt.Sprintf("enqueue a ConsiderGatewayUpdateDispatch event if this is clearly not a mirrorable service (%s)", testType),
-			environment: onAddOrUpdateNonExportedSvc(isAdd),
-			expectedEventsInQueue: []interface{}{&ConsiderGatewayUpdateDispatch{
-				maybeGateway: remoteService("test-service", "test-namespace", "", "", "resVersion", nil),
-			}},
-		},
 		{
 			description: fmt.Sprintf("enqueue a RemoteServiceUpdated event if this is a service that we have already mirrored and its res version is different (%s)", testType),
 			environment: onAddOrUpdateRemoteServiceUpdated(isAdd),
@@ -512,10 +501,6 @@ func onAddOrUpdateTestCases(isAdd bool) []mirroringTestCase {
 			expectedEventsInQueue: []interface{}{&RemoteServiceDeleted{
 				Name:      "test-service",
 				Namespace: "test-namespace",
-				GatewayData: gatewayMetadata{
-					Name:      "gateway",
-					Namespace: "gateway-ns",
-				},
 			}},
 
 			expectedLocalServices: []*corev1.Service{
@@ -551,24 +536,13 @@ func TestOnDelete(t *testing.T) {
 				&RemoteServiceDeleted{
 					Name:      "test-service",
 					Namespace: "test-namespace",
-					GatewayData: gatewayMetadata{
-						Name:      "gateway",
-						Namespace: "gateway-ns",
-					},
 				},
 			},
 		},
 		{
-			description: "enqueues a RemoteGatewayDeleted because there is no gateway metadata present on the service",
-			environment: onDeleteNoGatewayMetadata,
-			expectedEventsInQueue: []interface{}{
-				&RemoteGatewayDeleted{
-					gatewayData: gatewayMetadata{
-						Name:      "gateway",
-						Namespace: "test-namespace",
-					},
-				},
-			},
+			description:           "skips because there is no gateway metadata present on the service",
+			environment:           onDeleteNoGatewayMetadata,
+			expectedEventsInQueue: []interface{}{},
 		},
 	} {
 		tc := tt // pin
diff --git a/controller/cmd/service-mirror/cluster_watcher_probe_events_test.go b/controller/cmd/service-mirror/cluster_watcher_probe_events_test.go
deleted file mode 100644
index 296093063..000000000
--- a/controller/cmd/service-mirror/cluster_watcher_probe_events_test.go
+++ /dev/null
@@ -1,249 +0,0 @@
-package servicemirror
-
-import (
-	"fmt"
-	"reflect"
-	"testing"
-
-	v1 "k8s.io/api/core/v1"
-	"k8s.io/client-go/util/workqueue"
-)
-
-type BufferingProbeEventSink struct {
-	events []interface{}
-}
-
-func (s *BufferingProbeEventSink) send(event interface{}) {
-	s.events = append(s.events, event)
-}
-
-type probeEventsTestCase struct {
-	description                      string
-	environment                      *testEnvironment
-	expectedEventsSentToProbeManager []interface{}
-}
-
-func (tc *probeEventsTestCase) run(t *testing.T) {
-	t.Run(tc.description, func(t *testing.T) {
-		q := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
-		probeEventsSink := &BufferingProbeEventSink{}
-		_, err := tc.environment.runEnvironment(probeEventsSink, q)
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		expectedNumEvents := len(tc.expectedEventsSentToProbeManager)
-		actualNumEvents := len(probeEventsSink.events)
-
-		if expectedNumEvents != actualNumEvents {
-			t.Fatalf("Was expecting %d events but got %d", expectedNumEvents, actualNumEvents)
-		}
-
-		for i, ev := range tc.expectedEventsSentToProbeManager {
-			evInQueue := probeEventsSink.events[i]
-			if !reflect.DeepEqual(ev, evInQueue) {
-				t.Fatalf("was expecting to see event %T but got %T", ev, evInQueue)
-			}
-		}
-	})
-}
-
-func gatewaySpec(name, namespace, clustername, address, resVersion, identity string, incomingPort, probePort uint32, probePath string, probePeriod uint32) GatewaySpec {
-	return GatewaySpec{
-		gatewayName:      name,
-		gatewayNamespace: namespace,
-		clusterName:      clustername,
-		addresses: []v1.EndpointAddress{
-			{
-				IP: address,
-			},
-		},
-		incomingPort:    incomingPort,
-		resourceVersion: resVersion,
-		identity:        identity,
-		ProbeConfig: &ProbeConfig{
-			path:            probePath,
-			port:            probePort,
-			periodInSeconds: probePeriod,
-		},
-	}
-}
-
-func TestRemoteServiceCreatedProbeEvents(t *testing.T) {
-	for _, tt := range []probeEventsTestCase{
-		{
-			description:                      "do not send event if gateway cannot be resolved",
-			environment:                      serviceCreateWithMissingGateway,
-			expectedEventsSentToProbeManager: []interface{}{},
-		},
-		{
-			description:                      "do not send event if gateway has wrong spec",
-			environment:                      createServiceWrongGatewaySpec,
-			expectedEventsSentToProbeManager: []interface{}{},
-		},
-		{
-			description: "create service and endpoints when gateway can be resolved",
-			environment: createServiceOkeGatewaySpec,
-			expectedEventsSentToProbeManager: []interface{}{
-				&MirroredServicePaired{
-					serviceName:      fmt.Sprintf("service-one-%s", clusterName),
-					serviceNamespace: "ns1",
-					GatewaySpec:      gatewaySpec("existing-gateway", "existing-namespace", clusterName, "192.0.2.127", "222", "gateway-identity", 888, defaultProbePort, defaultProbePath, defaultProbePeriod),
-				},
-			},
-		},
-	} {
-		tc := tt // pin
-		tc.run(t)
-	}
-}
-
-func TestRemoteServiceDeletedProbeEvents(t *testing.T) {
-	for _, tt := range []probeEventsTestCase{
-		{
-			description: "send a service unpaired event",
-			environment: deleteMirroredService,
-			expectedEventsSentToProbeManager: []interface{}{
-				&MirroredServiceUnpaired{
-					serviceName:      fmt.Sprintf("test-service-remote-to-delete-%s", clusterName),
-					serviceNamespace: "test-namespace-to-delete",
-					gatewayName:      "gateway",
-					gatewayNs:        "gateway-ns",
-					clusterName:      clusterName,
-				},
-			},
-		},
-	} {
-		tc := tt // pin
-		tc.run(t)
-	}
-}
-
-func TestRemoteServiceUpdatedProbeEvents(t *testing.T) {
-	for _, tt := range []probeEventsTestCase{
-		{
-			description: "unpairs from old and pairs to new gateway",
-			environment: updateServiceToNewGateway,
-			expectedEventsSentToProbeManager: []interface{}{
-				&MirroredServiceUnpaired{
-					serviceName:      fmt.Sprintf("test-service-%s", clusterName),
-					serviceNamespace: "test-namespace",
-					gatewayName:      "gateway",
-					gatewayNs:        "gateway-ns",
-					clusterName:      clusterName,
-				},
-				&MirroredServicePaired{
-					serviceName:      fmt.Sprintf("test-service-%s", clusterName),
-					serviceNamespace: "test-namespace",
-					GatewaySpec:      gatewaySpec("gateway-new", "gateway-ns", clusterName, "0.0.0.0", "currentGatewayResVersion", "", 999, defaultProbePort, defaultProbePath, defaultProbePeriod),
-				},
-			},
-		},
-		{
-			description: "does not send event when gateway assignment does not change",
-			environment: updateServiceWithChangedPorts,
-		},
-	} {
-		tc := tt // pin
-		tc.run(t)
-	}
-}
-
-func TestRemoteGatewayUpdatedProbeEvents(t *testing.T) {
-	for _, tt := range []probeEventsTestCase{
-		{
-			description: "sends gateway updated when endpoints ports",
-			environment: remoteGatewayUpdated,
-			expectedEventsSentToProbeManager: []interface{}{
-				&MirroredServicePaired{
-					serviceName:      "test-service-1-remote",
-					serviceNamespace: "test-namespace",
-					GatewaySpec:      gatewaySpec("gateway", "gateway-ns", clusterName, "0.0.0.0", "currentGatewayResVersion", "", 999, defaultProbePort, defaultProbePath, defaultProbePeriod),
-				},
-				&MirroredServicePaired{
-					serviceName:      "test-service-2-remote",
-					serviceNamespace: "test-namespace",
-					GatewaySpec:      gatewaySpec("gateway", "gateway-ns", clusterName, "0.0.0.0", "currentGatewayResVersion", "", 999, defaultProbePort, defaultProbePath, defaultProbePeriod),
-				},
-				&GatewayUpdated{
-					GatewaySpec: gatewaySpec("gateway", "gateway-ns", clusterName, "0.0.0.0", "currentGatewayResVersion", "", 999, defaultProbePort, defaultProbePath, defaultProbePeriod),
-				},
-			},
-		},
-
-		{
-			description: "sends gateway updated when address changes",
-			environment: gatewayAddressChanged,
-			expectedEventsSentToProbeManager: []interface{}{
-				&MirroredServicePaired{
-					serviceName:      "test-service-1-remote",
-					serviceNamespace: "test-namespace",
-					GatewaySpec:      gatewaySpec("gateway", "gateway-ns", "some-cluster", "0.0.0.1", "currentGatewayResVersion", "", 888, 1, "/p", 222),
-				},
-				&MirroredServicePaired{
-					serviceName:      "test-service-2-remote",
-					serviceNamespace: "test-namespace",
-					GatewaySpec:      gatewaySpec("gateway", "gateway-ns", "some-cluster", "0.0.0.1", "currentGatewayResVersion", "", 888, 1, "/p", 222),
-				},
-				&GatewayUpdated{
-					GatewaySpec: gatewaySpec("gateway", "gateway-ns", "some-cluster", "0.0.0.1", "currentGatewayResVersion", "", 888, 1, "/p", 222),
-				},
-			},
-		},
-		{
-			description: "sends gateway updated when identity changes",
-			environment: gatewayIdentityChanged,
-			expectedEventsSentToProbeManager: []interface{}{
-				&MirroredServicePaired{
-					serviceName:      "test-service-1-remote",
-					serviceNamespace: "test-namespace",
-					GatewaySpec:      gatewaySpec("gateway", "gateway-ns", clusterName, "0.0.0.0", "currentGatewayResVersion", "new-identity", 888, defaultProbePort, defaultProbePath, defaultProbePeriod),
-				},
-				&MirroredServicePaired{
-					serviceName:      "test-service-2-remote",
-					serviceNamespace: "test-namespace",
-					GatewaySpec:      gatewaySpec("gateway", "gateway-ns", clusterName, "0.0.0.0", "currentGatewayResVersion", "new-identity", 888, defaultProbePort, defaultProbePath, defaultProbePeriod),
-				},
-				&GatewayUpdated{
-					GatewaySpec: gatewaySpec("gateway", "gateway-ns", clusterName, "0.0.0.0", "currentGatewayResVersion", "new-identity", 888, defaultProbePort, defaultProbePath, defaultProbePeriod),
-				},
-			},
-		},
-		{
-			description: "sends gateway updated when probe changes",
-			environment: gatewayProbeConfigChanged,
-			expectedEventsSentToProbeManager: []interface{}{
-				&MirroredServicePaired{
-					serviceName:      "test-service-1-remote",
-					serviceNamespace: "test-namespace",
-					GatewaySpec:      gatewaySpec("gateway", "gateway-ns", clusterName, "0.0.0.0", "currentGatewayResVersion", "identity", 888, defaultProbePort, "/new-path", defaultProbePeriod),
-				},
-				&GatewayUpdated{
-					GatewaySpec: gatewaySpec("gateway", "gateway-ns", clusterName, "0.0.0.0", "currentGatewayResVersion", "identity", 888, defaultProbePort, "/new-path", defaultProbePeriod),
-				},
-			},
-		},
-	} {
-		tc := tt // pin
-		tc.run(t)
-	}
-}
-
-func TestRemoteGatewayDeletedProbeEvents(t *testing.T) {
-	for _, tt := range []probeEventsTestCase{
-		{
-			description: "sends a gateway deleted event to the probe manager",
-			environment: gatewayDeleted,
-			expectedEventsSentToProbeManager: []interface{}{
-				&GatewayDeleted{
-					gatewayName: "gateway",
-					gatewayNs:   "gateway-ns",
-					clusterName: clusterName,
-				},
-			},
-		},
-	} {
-		tc := tt // pin
-		tc.run(t)
-	}
-}
diff --git a/controller/cmd/service-mirror/cluster_watcher_test_util.go b/controller/cmd/service-mirror/cluster_watcher_test_util.go
index 5c3ef5556..1cfe2994f 100644
--- a/controller/cmd/service-mirror/cluster_watcher_test_util.go
+++ b/controller/cmd/service-mirror/cluster_watcher_test_util.go
@@ -29,7 +29,7 @@ type testEnvironment struct {
 	localResources  []string
 }
 
-func (te *testEnvironment) runEnvironment(probeEventSync ProbeEventSink, watcherQueue workqueue.RateLimitingInterface) (*k8s.API, error) {
+func (te *testEnvironment) runEnvironment(watcherQueue workqueue.RateLimitingInterface) (*k8s.API, error) {
 	remoteAPI, err := k8s.NewFakeAPI(te.remoteResources...)
 	if err != nil {
 		return nil, err
@@ -52,7 +52,6 @@ func (te *testEnvironment) runEnvironment(probeEventSync ProbeEventSink, watcher
 		log:             logging.WithFields(logging.Fields{"cluster": clusterName}),
 		eventsQueue:     watcherQueue,
 		requeueLimit:    0,
-		probeEventsSink: probeEventSync,
 	}
 
 	for _, ev := range te.events {
@@ -73,7 +72,7 @@ var serviceCreateWithMissingGateway = &testEnvironment{
 	events: []interface{}{
 		&RemoteServiceCreated{
 			service: remoteService("service-one", "ns1", "missing-gateway", "missing-namespace", "111", nil),
-			gatewayData: &gatewayMetadata{
+			gatewayData: gatewayMetadata{
 				Name:      "missing-gateway",
 				Namespace: "missing-namespace",
 			},
@@ -98,7 +97,7 @@ var createServiceWrongGatewaySpec = &testEnvironment{
 					},
 				}),
 
-			gatewayData: &gatewayMetadata{
+			gatewayData: gatewayMetadata{
 				Name:      "existing-gateway",
 				Namespace: "existing-namespace",
 			},
@@ -124,7 +123,7 @@ var createServiceOkeGatewaySpec = &testEnvironment{
 					Port:     666,
 				},
 			}),
-			gatewayData: &gatewayMetadata{
+			gatewayData: gatewayMetadata{
 				Name:      "existing-gateway",
 				Namespace: "existing-namespace",
 			},
@@ -140,10 +139,6 @@ var deleteMirroredService = &testEnvironment{
 		&RemoteServiceDeleted{
 			Name:      "test-service-remote-to-delete",
 			Namespace: "test-namespace-to-delete",
-			GatewayData: gatewayMetadata{
-				Name:      "gateway",
-				Namespace: "gateway-ns",
-			},
 		},
 	},
 	localResources: []string{
@@ -520,54 +515,6 @@ var gatewayIdentityChanged = &testEnvironment{
 	},
 }
 
-var gatewayProbeConfigChanged = &testEnvironment{
-	events: []interface{}{
-		&RemoteGatewayUpdated{
-			gatewaySpec: GatewaySpec{
-				gatewayName:      "gateway",
-				gatewayNamespace: "gateway-ns",
-				clusterName:      clusterName,
-				addresses:        []corev1.EndpointAddress{{IP: "0.0.0.0"}},
-				incomingPort:     888,
-				resourceVersion:  "currentGatewayResVersion",
-				identity:         "identity",
-				ProbeConfig: &ProbeConfig{
-					path:            "/new-path",
-					port:            defaultProbePort,
-					periodInSeconds: defaultProbePeriod,
-				},
-			},
-			affectedServices: []*corev1.Service{
-				mirroredService("test-service-1-remote", "test-namespace", "gateway", "gateway-ns", "", "pastGatewayResVersion",
-					[]corev1.ServicePort{
-						{
-							Name:     "svc-1-port",
-							Protocol: "TCP",
-							Port:     8081,
-						},
-					}),
-			},
-		},
-	},
-	localResources: []string{
-		mirroredServiceAsYaml("test-service-1-remote", "test-namespace", "gateway", "gateway-ns", "", "pastGatewayResVersion",
-			[]corev1.ServicePort{
-				{
-					Name:     "svc-1-port",
-					Protocol: "TCP",
-					Port:     8081,
-				},
-			}),
-		endpointsAsYaml("test-service-1-remote", "test-namespace", "gateway", "gateway-ns", "0.0.0.0", "",
-			[]corev1.EndpointPort{
-				{
-					Name:     "svc-1-port",
-					Port:     888,
-					Protocol: "TCP",
-				}}),
-	},
-}
-
 var gatewayDeleted = &testEnvironment{
 	events: []interface{}{
 		&RemoteGatewayDeleted{
@@ -646,15 +593,6 @@ func onAddOrUpdateExportedSvc(isAdd bool) *testEnvironment {
 
 }
 
-func onAddOrUpdateNonExportedSvc(isAdd bool) *testEnvironment {
-	return &testEnvironment{
-		events: []interface{}{
-			onAddOrUpdateEvent(isAdd, remoteService("test-service", "test-namespace", "", "", "resVersion", nil)),
-		},
-	}
-
-}
-
 func onAddOrUpdateRemoteServiceUpdated(isAdd bool) *testEnvironment {
 	return &testEnvironment{
 		events: []interface{}{
@@ -836,7 +774,7 @@ func mirroredServiceAsYaml(name, namespace, gtwName, gtwNs, resourceVersion, gat
 	return string(bytes)
 }
 
-func gateway(name, namespace, resourceVersion, ip, portName string, port int32, identity string, probePort int, probePath string, probePeriod int) *corev1.Service {
+func gateway(name, namespace, resourceVersion, ip, portName string, port int32, identity string, probePort int32, probePath string, probePeriod int) *corev1.Service {
 	svc := corev1.Service{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "Service",
@@ -850,7 +788,6 @@ func gateway(name, namespace, resourceVersion, ip, portName string, port int32,
 				consts.GatewayIdentity:    identity,
 				consts.GatewayProbePath:   probePath,
 				consts.GatewayProbePeriod: fmt.Sprint(probePeriod),
-				consts.GatewayProbePort:   fmt.Sprint(probePort),
 			},
 		},
 		Spec: corev1.ServiceSpec{
@@ -860,6 +797,11 @@ func gateway(name, namespace, resourceVersion, ip, portName string, port int32,
 					Protocol: "TCP",
 					Port:     port,
 				},
+				{
+					Name:     consts.ProbePortName,
+					Protocol: "TCP",
+					Port:     probePort,
+				},
 			},
 		},
 	}
@@ -870,7 +812,7 @@ func gateway(name, namespace, resourceVersion, ip, portName string, port int32,
 	return &svc
 }
 
-func gatewayAsYaml(name, namespace, resourceVersion, ip, portName string, port int32, identity string, probePort int, probePath string, probePeriod int) string {
+func gatewayAsYaml(name, namespace, resourceVersion, ip, portName string, port int32, identity string, probePort int32, probePath string, probePeriod int) string {
 	gtw := gateway(name, namespace, resourceVersion, ip, portName, port, identity, probePort, probePath, probePeriod)
 
 	bytes, err := yaml.Marshal(gtw)
diff --git a/test/edges/edges_test.go b/test/edges/edges_test.go
index c7b7774f0..52ef88fe8 100644
--- a/test/edges/edges_test.go
+++ b/test/edges/edges_test.go
@@ -18,7 +18,7 @@ var TestHelper *testutil.TestHelper
 
 func TestMain(m *testing.M) {
 	TestHelper = testutil.NewTestHelper()
-	os.Exit(m.Run())
+	os.Exit(testutil.Run(m, TestHelper, true))
 }
 
 // TestEdges requires that there has been traffic recently between linkerd-web
diff --git a/test/egress/egress_test.go b/test/egress/egress_test.go
index e6404d1d1..2f5783da5 100644
--- a/test/egress/egress_test.go
+++ b/test/egress/egress_test.go
@@ -16,7 +16,7 @@ var TestHelper *testutil.TestHelper
 
 func TestMain(m *testing.M) {
 	TestHelper = testutil.NewTestHelper()
-	os.Exit(m.Run())
+	os.Exit(testutil.Run(m, TestHelper, true))
 }
 
 //////////////////////
diff --git a/test/endpoints/endpoints_test.go b/test/endpoints/endpoints_test.go
index 723256d06..6467ad3f9 100644
--- a/test/endpoints/endpoints_test.go
+++ b/test/endpoints/endpoints_test.go
@@ -16,7 +16,7 @@ var TestHelper *testutil.TestHelper
 
 func TestMain(m *testing.M) {
 	TestHelper = testutil.NewTestHelper()
-	os.Exit(m.Run())
+	os.Exit(testutil.Run(m, TestHelper, true))
 }
 
 func TestGoodEndpoints(t *testing.T) {
diff --git a/test/externalissuer/external_issuer_test.go b/test/externalissuer/external_issuer_test.go
index 0aaf613cc..daca55e5a 100644
--- a/test/externalissuer/external_issuer_test.go
+++ b/test/externalissuer/external_issuer_test.go
@@ -24,7 +24,7 @@ func TestMain(m *testing.M) {
 		fmt.Fprintln(os.Stdout, "Skiping as --external-issuer=false")
 		os.Exit(0)
 	}
-	os.Exit(m.Run())
+	os.Exit(testutil.Run(m, TestHelper, true))
 }
 
 func TestExternalIssuer(t *testing.T) {
diff --git a/test/get/get_test.go b/test/get/get_test.go
index 39466c175..c5cd49706 100644
--- a/test/get/get_test.go
+++ b/test/get/get_test.go
@@ -21,7 +21,7 @@ var TestHelper *testutil.TestHelper
 
 func TestMain(m *testing.M) {
 	TestHelper = testutil.NewTestHelper()
-	os.Exit(m.Run())
+	os.Exit(testutil.Run(m, TestHelper, true))
 }
 
 var (
diff --git a/test/inject/inject_test.go b/test/inject/inject_test.go
index cf949fffe..1f01eb173 100644
--- a/test/inject/inject_test.go
+++ b/test/inject/inject_test.go
@@ -24,7 +24,7 @@ var TestHelper *testutil.TestHelper
 
 func TestMain(m *testing.M) {
 	TestHelper = testutil.NewTestHelper()
-	os.Exit(m.Run())
+	os.Exit(testutil.Run(m, TestHelper, true))
 }
 
 //////////////////////
diff --git a/test/install_test.go b/test/install_test.go
index a642ba199..38261be44 100644
--- a/test/install_test.go
+++ b/test/install_test.go
@@ -1,10 +1,8 @@
 package test
 
 import (
-	"errors"
 	"fmt"
 	"os"
-	"regexp"
 	"strconv"
 	"strings"
 	"testing"
@@ -13,7 +11,6 @@ import (
 	"github.com/linkerd/linkerd2/pkg/k8s"
 	"github.com/linkerd/linkerd2/pkg/tls"
 	"github.com/linkerd/linkerd2/testutil"
-	corev1 "k8s.io/api/core/v1"
 )
 
 //////////////////////
@@ -30,22 +27,7 @@ var (
 
 func TestMain(m *testing.M) {
 	TestHelper = testutil.NewTestHelper()
-	code := m.Run()
-	// if the test failed, show logs and events that can potentially explain cause
-	if code != 0 && controlPlaneInstalled {
-		_, errs1 := fetchAndCheckLogs()
-		for _, err := range errs1 {
-			fmt.Println(err)
-		}
-		errs2 := fetchAndCheckEvents()
-		for _, err := range errs2 {
-			fmt.Println(err)
-		}
-		if len(errs1) == 0 && len(errs2) == 0 {
-			fmt.Println("No unexpected log entries or events found")
-		}
-	}
-	os.Exit(code)
+	os.Exit(testutil.Run(m, TestHelper, controlPlaneInstalled))
 }
 
 var (
@@ -63,44 +45,6 @@ var (
 		"linkerd-tap",
 	}
 
-	// Linkerd commonly logs these errors during testing, remove these once
-	// they're addressed: https://github.com/linkerd/linkerd2/issues/2453
-	knownControllerErrorsRegex = regexp.MustCompile(strings.Join([]string{
-		`.*linkerd-controller-.*-.* tap time=".*" level=error msg="\[.*\] encountered an error: rpc error: code = Canceled desc = context canceled"`,
-		`.*linkerd-web-.*-.* web time=".*" level=error msg="Post http://linkerd-controller-api\..*\.svc\.cluster\.local:8085/api/v1/Version: context canceled"`,
-		`.*linkerd-proxy-injector-.*-.* proxy-injector time=".*" level=warning msg="failed to retrieve replicaset from indexer .*-smoke-test.*/smoke-test-.*-.*: replicaset\.apps \\"smoke-test-.*-.*\\" not found"`,
-		`.*linkerd-destination-.* destination time=".*" level=warning msg="failed to retrieve replicaset from indexer .* not found"`,
-	}, "|"))
-
-	knownProxyErrorsRegex = regexp.MustCompile(strings.Join([]string{
-		// k8s hitting readiness endpoints before components are ready
-		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|sp-validator|web|tap)-.*-.* linkerd-proxy ERR! \[ +\d+.\d+s\] proxy={server=in listen=0\.0\.0\.0:4143 remote=.*} linkerd2_proxy::app::errors unexpected error: an IO error occurred: Connection reset by peer \(os error 104\)`,
-		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|sp-validator|web|tap)-.*-.* linkerd-proxy ERR! \[ *\d+.\d+s\] proxy={server=in listen=0\.0\.0\.0:4143 remote=.*} linkerd2_proxy::(proxy::http::router service|app::errors unexpected) error: an error occurred trying to connect: Connection refused \(os error 111\) \(address: 127\.0\.0\.1:.*\)`,
-		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|sp-validator|web|tap)-.*-.* linkerd-proxy ERR! \[ *\d+.\d+s\] proxy={server=out listen=127\.0\.0\.1:4140 remote=.*} linkerd2_proxy::(proxy::http::router service|app::errors unexpected) error: an error occurred trying to connect: Connection refused \(os error 111\) \(address: .*\)`,
-		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|sp-validator|web|tap)-.*-.* linkerd-proxy ERR! \[ *\d+.\d+s\] proxy={server=out listen=127\.0\.0\.1:4140 remote=.*} linkerd2_proxy::(proxy::http::router service|app::errors unexpected) error: an error occurred trying to connect: operation timed out after 1s`,
-		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|sp-validator|web|tap)-.*-.* linkerd-proxy WARN \[ *\d+.\d+s\] .* linkerd2_proxy::proxy::reconnect connect error to ControlAddr .*`,
-
-		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|sp-validator|web|tap)-.*-.* linkerd-proxy ERR! \[ *\d+.\d+s\] admin={server=metrics listen=0\.0\.0\.0:4191 remote=.*} linkerd2_proxy::control::serve_http error serving metrics: Error { kind: Shutdown, .* }`,
-		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|sp-validator|web|tap)-.*-.* linkerd-proxy ERR! \[ +\d+.\d+s\] admin={server=admin listen=127\.0\.0\.1:4191 remote=.*} linkerd2_proxy::control::serve_http error serving admin: Error { kind: Shutdown, cause: Os { code: 107, kind: NotConnected, message: "Transport endpoint is not connected" } }`,
-
-		`.* linkerd-web-.*-.* linkerd-proxy WARN trust_dns_proto::xfer::dns_exchange failed to associate send_message response to the sender`,
-		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|web|tap)-.*-.* linkerd-proxy WARN \[.*\] linkerd2_proxy::proxy::canonicalize failed to refine linkerd-.*\..*\.svc\.cluster\.local: deadline has elapsed; using original name`,
-
-		// prometheus scrape failures of control-plane
-		`.* linkerd-prometheus-.*-.* linkerd-proxy ERR! \[ +\d+.\d+s\] proxy={server=out listen=127\.0\.0\.1:4140 remote=.*} linkerd2_proxy::proxy::http::router service error: an error occurred trying to connect: .*`,
-	}, "|"))
-
-	knownEventWarningsRegex = regexp.MustCompile(strings.Join([]string{
-		`MountVolume.SetUp failed for volume .* : couldn't propagate object cache: timed out waiting for the condition`, // pre k8s 1.16
-		`MountVolume.SetUp failed for volume .* : failed to sync .* cache: timed out waiting for the condition`,         // post k8s 1.16
-		`(Liveness|Readiness) probe failed: HTTP probe failed with statuscode: 50(2|3)`,
-		`(Liveness|Readiness) probe failed: Get http://.*: dial tcp .*: connect: connection refused`,
-		`(Liveness|Readiness) probe failed: Get http://.*: read tcp .*: read: connection reset by peer`,
-		`(Liveness|Readiness) probe failed: Get http://.*: net/http: request canceled .*\(Client\.Timeout exceeded while awaiting headers\)`,
-		`Failed to update endpoint .*/linkerd-.*: Operation cannot be fulfilled on endpoints "linkerd-.*": the object has been modified; please apply your changes to the latest version and try again`,
-		`error killing pod: failed to "KillPodSandbox" for ".*" with KillPodSandboxError: "rpc error: code = Unknown desc`,
-	}, "|"))
-
 	injectionCases = []struct {
 		ns          string
 		annotations map[string]string
@@ -378,10 +322,12 @@ func helmOverridesEdge(root *tls.CA) []string {
 }
 
 func TestInstallHelm(t *testing.T) {
+	// at this point all the tests assume the control plane is present
+	controlPlaneInstalled = true
+
 	if TestHelper.GetHelmReleaseName() == "" {
 		return
 	}
-	controlPlaneInstalled = true
 
 	cn := fmt.Sprintf("identity.%s.cluster.local", TestHelper.GetLinkerdNamespace())
 	var err error
@@ -745,82 +691,8 @@ func TestCheckProxy(t *testing.T) {
 	}
 }
 
-func fetchAndCheckLogs() ([]string, []error) {
-	okMessages := []string{}
-	errs := []error{}
-
-	controllerRegex := regexp.MustCompile("level=(panic|fatal|error|warn)")
-	proxyRegex := regexp.MustCompile(fmt.Sprintf("%s (ERR|WARN)", k8s.ProxyContainerName))
-	clientGoRegex := regexp.MustCompile("client-go@")
-	hasClientGoLogs := false
-
-	for deploy, spec := range testutil.LinkerdDeployReplicas {
-		deploy := strings.TrimPrefix(deploy, "linkerd-")
-		containers := append(spec.Containers, k8s.ProxyContainerName)
-
-		for _, container := range containers {
-			container := container // pin
-			name := fmt.Sprintf("%s/%s", deploy, container)
-
-			proxy := false
-			errRegex := controllerRegex
-			knownErrorsRegex := knownControllerErrorsRegex
-			if container == k8s.ProxyContainerName {
-				proxy = true
-				errRegex = proxyRegex
-				knownErrorsRegex = knownProxyErrorsRegex
-			}
-
-			outputStream, err := TestHelper.LinkerdRunStream(
-				"logs", "--no-color",
-				"--control-plane-component", deploy,
-				"--container", container,
-			)
-			if err != nil {
-				errs = append(errs, fmt.Errorf("error running command:\n%s", err))
-				continue
-			}
-			defer outputStream.Stop()
-			// Ignore the error returned, since ReadUntil will return an error if it
-			// does not return 10,000 after 2 seconds. We don't need 10,000 log lines.
-			outputLines, _ := outputStream.ReadUntil(10000, 2*time.Second)
-			if len(outputLines) == 0 {
-				// Retry one time for 30 more seconds, in case the cluster is slow to
-				// produce log lines.
-				outputLines, _ = outputStream.ReadUntil(10000, 30*time.Second)
-				if len(outputLines) == 0 {
-					errs = append(errs, fmt.Errorf("no logs found for %s", name))
-				}
-			}
-
-			for _, line := range outputLines {
-				if errRegex.MatchString(line) {
-					if knownErrorsRegex.MatchString(line) {
-						// report all known logging errors in the output
-						okMessages = append(okMessages, fmt.Sprintf("found known error in %s log: %s", name, line))
-					} else {
-						if proxy {
-							okMessages = append(okMessages, fmt.Sprintf("found unexpected proxy error in %s log: %s", name, line))
-						} else {
-							errs = append(errs, fmt.Errorf("found unexpected controller error in %s log: %s", name, line))
-						}
-					}
-				}
-				if clientGoRegex.MatchString((line)) {
-					hasClientGoLogs = true
-				}
-			}
-		}
-	}
-	if !hasClientGoLogs {
-		errs = append(errs, errors.New("didn't find any client-go entries"))
-	}
-
-	return okMessages, errs
-}
-
 func TestLogs(t *testing.T) {
-	okMessages, errs := fetchAndCheckLogs()
+	okMessages, errs := testutil.FetchAndCheckLogs(TestHelper)
 	for msg := range okMessages {
 		t.Log(msg)
 	}
@@ -829,37 +701,8 @@ func TestLogs(t *testing.T) {
 	}
 }
 
-func fetchAndCheckEvents() []error {
-	out, err := TestHelper.Kubectl("",
-		"--namespace", TestHelper.GetLinkerdNamespace(),
-		"get", "events", "-ojson",
-	)
-	if err != nil {
-		return []error{fmt.Errorf("'kubectl get events' command failed with %s\n%s", err, out)}
-	}
-
-	events, err := testutil.ParseEvents(out)
-	if err != nil {
-		return []error{err}
-	}
-
-	errs := []error{}
-	for _, e := range events {
-		if e.Type == corev1.EventTypeNormal {
-			continue
-		}
-
-		evtStr := fmt.Errorf("found unexpected warning event: reason: [%s] Object: [%s] Message: [%s]", e.Reason, e.InvolvedObject.Name, e.Message)
-		if !knownEventWarningsRegex.MatchString(e.Message) {
-			errs = append(errs, evtStr)
-		}
-	}
-
-	return errs
-}
-
 func TestEvents(t *testing.T) {
-	for err := range fetchAndCheckEvents() {
+	for err := range testutil.FetchAndCheckEvents(TestHelper) {
 		testutil.AnnotatedError(t, "Error checking events", err)
 	}
 }
diff --git a/test/routes/routes_test.go b/test/routes/routes_test.go
index 69ad2f8c2..381879576 100644
--- a/test/routes/routes_test.go
+++ b/test/routes/routes_test.go
@@ -17,7 +17,7 @@ var TestHelper *testutil.TestHelper
 
 func TestMain(m *testing.M) {
 	TestHelper = testutil.NewTestHelper()
-	os.Exit(m.Run())
+	os.Exit(testutil.Run(m, TestHelper, true))
 }
 
 //////////////////////
diff --git a/test/serviceaccounts/serviceaccounts_test.go b/test/serviceaccounts/serviceaccounts_test.go
index 58f46bace..03eb36835 100644
--- a/test/serviceaccounts/serviceaccounts_test.go
+++ b/test/serviceaccounts/serviceaccounts_test.go
@@ -13,7 +13,7 @@ var TestHelper *testutil.TestHelper
 
 func TestMain(m *testing.M) {
 	TestHelper = testutil.NewTestHelper()
-	os.Exit(m.Run())
+	os.Exit(testutil.Run(m, TestHelper, true))
 }
 
 // namesMatch checks if all the expectedServiceAccountNames are present in the given list,
diff --git a/test/serviceprofiles/serviceprofiles_test.go b/test/serviceprofiles/serviceprofiles_test.go
index 356a7151a..8c5c922a9 100644
--- a/test/serviceprofiles/serviceprofiles_test.go
+++ b/test/serviceprofiles/serviceprofiles_test.go
@@ -27,7 +27,7 @@ type testCase struct {
 
 func TestMain(m *testing.M) {
 	TestHelper = testutil.NewTestHelper()
-	os.Exit(m.Run())
+	os.Exit(testutil.Run(m, TestHelper, true))
 }
 
 func TestServiceProfiles(t *testing.T) {
diff --git a/test/stat/stat_test.go b/test/stat/stat_test.go
index 1e80c9dd2..9d1036ab1 100644
--- a/test/stat/stat_test.go
+++ b/test/stat/stat_test.go
@@ -19,7 +19,7 @@ var TestHelper *testutil.TestHelper
 
 func TestMain(m *testing.M) {
 	TestHelper = testutil.NewTestHelper()
-	os.Exit(m.Run())
+	os.Exit(testutil.Run(m, TestHelper, true))
 }
 
 //////////////////////
diff --git a/test/tap/tap_test.go b/test/tap/tap_test.go
index 2bea554e4..1935a4462 100644
--- a/test/tap/tap_test.go
+++ b/test/tap/tap_test.go
@@ -18,7 +18,7 @@ var TestHelper *testutil.TestHelper
 
 func TestMain(m *testing.M) {
 	TestHelper = testutil.NewTestHelper()
-	os.Exit(m.Run())
+	os.Exit(testutil.Run(m, TestHelper, true))
 }
 
 type tapEvent struct {
diff --git a/test/tracing/tracing_test.go b/test/tracing/tracing_test.go
index bc7f7ec2e..9c6426edc 100644
--- a/test/tracing/tracing_test.go
+++ b/test/tracing/tracing_test.go
@@ -33,7 +33,7 @@ var TestHelper *testutil.TestHelper
 
 func TestMain(m *testing.M) {
 	TestHelper = testutil.NewTestHelper()
-	os.Exit(m.Run())
+	os.Exit(testutil.Run(m, TestHelper, true))
 }
 
 //////////////////////
diff --git a/test/trafficsplit/trafficsplit_test.go b/test/trafficsplit/trafficsplit_test.go
index b8512daa3..ea2e64590 100644
--- a/test/trafficsplit/trafficsplit_test.go
+++ b/test/trafficsplit/trafficsplit_test.go
@@ -16,7 +16,7 @@ const zeroRPS = "0.0rps"
 
 func TestMain(m *testing.M) {
 	TestHelper = testutil.NewTestHelper()
-	os.Exit(m.Run())
+	os.Exit(testutil.Run(m, TestHelper, true))
 }
 
 type statTsRow struct {
diff --git a/test/uninstall/uninstall_test.go b/test/uninstall/uninstall_test.go
index afe8f0153..6d2dc1faa 100644
--- a/test/uninstall/uninstall_test.go
+++ b/test/uninstall/uninstall_test.go
@@ -16,7 +16,7 @@ func TestMain(m *testing.M) {
 		fmt.Fprintln(os.Stderr, "Uninstall test disabled")
 		os.Exit(0)
 	}
-	os.Exit(m.Run())
+	os.Exit(testutil.Run(m, TestHelper, true))
 }
 
 func TestInstall(t *testing.T) {
diff --git a/testutil/logs_events.go b/testutil/logs_events.go
new file mode 100644
index 000000000..bb12ba273
--- /dev/null
+++ b/testutil/logs_events.go
@@ -0,0 +1,161 @@
+package testutil
+
+import (
+	"errors"
+	"fmt"
+	"regexp"
+	"strings"
+	"time"
+
+	"github.com/linkerd/linkerd2/pkg/k8s"
+	corev1 "k8s.io/api/core/v1"
+)
+
+var (
+	// Linkerd commonly logs these errors during testing, remove these once
+	// they're addressed: https://github.com/linkerd/linkerd2/issues/2453
+	knownControllerErrorsRegex = regexp.MustCompile(strings.Join([]string{
+		`.*linkerd-controller-.*-.* tap time=".*" level=error msg="\[.*\] encountered an error: rpc error: code = Canceled desc = context canceled"`,
+		`.*linkerd-web-.*-.* web time=".*" level=error msg="Post http://linkerd-controller-api\..*\.svc\.cluster\.local:8085/api/v1/Version: context canceled"`,
+		`.*linkerd-proxy-injector-.*-.* proxy-injector time=".*" level=warning msg="failed to retrieve replicaset from indexer .*-smoke-test.*/smoke-test-.*-.*: replicaset\.apps \\"smoke-test-.*-.*\\" not found"`,
+		`.*linkerd-destination-.* destination time=".*" level=warning msg="failed to retrieve replicaset from indexer .* not found"`,
+	}, "|"))
+
+	knownProxyErrorsRegex = regexp.MustCompile(strings.Join([]string{
+		// k8s hitting readiness endpoints before components are ready
+		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|sp-validator|web|tap)-.*-.* linkerd-proxy ERR! \[ +\d+.\d+s\] proxy={server=in listen=0\.0\.0\.0:4143 remote=.*} linkerd2_proxy::app::errors unexpected error: an IO error occurred: Connection reset by peer \(os error 104\)`,
+		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|sp-validator|web|tap)-.*-.* linkerd-proxy ERR! \[ *\d+.\d+s\] proxy={server=in listen=0\.0\.0\.0:4143 remote=.*} linkerd2_proxy::(proxy::http::router service|app::errors unexpected) error: an error occurred trying to connect: Connection refused \(os error 111\) \(address: 127\.0\.0\.1:.*\)`,
+		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|sp-validator|web|tap)-.*-.* linkerd-proxy ERR! \[ *\d+.\d+s\] proxy={server=out listen=127\.0\.0\.1:4140 remote=.*} linkerd2_proxy::(proxy::http::router service|app::errors unexpected) error: an error occurred trying to connect: Connection refused \(os error 111\) \(address: .*\)`,
+		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|sp-validator|web|tap)-.*-.* linkerd-proxy ERR! \[ *\d+.\d+s\] proxy={server=out listen=127\.0\.0\.1:4140 remote=.*} linkerd2_proxy::(proxy::http::router service|app::errors unexpected) error: an error occurred trying to connect: operation timed out after 1s`,
+		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|sp-validator|web|tap)-.*-.* linkerd-proxy WARN \[ *\d+.\d+s\] .* linkerd2_proxy::proxy::reconnect connect error to ControlAddr .*`,
+
+		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|sp-validator|web|tap)-.*-.* linkerd-proxy ERR! \[ *\d+.\d+s\] admin={server=metrics listen=0\.0\.0\.0:4191 remote=.*} linkerd2_proxy::control::serve_http error serving metrics: Error { kind: Shutdown, .* }`,
+		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|sp-validator|web|tap)-.*-.* linkerd-proxy ERR! \[ +\d+.\d+s\] admin={server=admin listen=127\.0\.0\.1:4191 remote=.*} linkerd2_proxy::control::serve_http error serving admin: Error { kind: Shutdown, cause: Os { code: 107, kind: NotConnected, message: "Transport endpoint is not connected" } }`,
+
+		`.* linkerd-web-.*-.* linkerd-proxy WARN trust_dns_proto::xfer::dns_exchange failed to associate send_message response to the sender`,
+		`.* linkerd-(controller|identity|grafana|prometheus|proxy-injector|web|tap)-.*-.* linkerd-proxy WARN \[.*\] linkerd2_proxy::proxy::canonicalize failed to refine linkerd-.*\..*\.svc\.cluster\.local: deadline has elapsed; using original name`,
+
+		// prometheus scrape failures of control-plane
+		`.* linkerd-prometheus-.*-.* linkerd-proxy ERR! \[ +\d+.\d+s\] proxy={server=out listen=127\.0\.0\.1:4140 remote=.*} linkerd2_proxy::proxy::http::router service error: an error occurred trying to connect: .*`,
+	}, "|"))
+
+	knownEventWarningsRegex = regexp.MustCompile(strings.Join([]string{
+		`MountVolume.SetUp failed for volume .* : couldn't propagate object cache: timed out waiting for the condition`, // pre k8s 1.16
+		`MountVolume.SetUp failed for volume .* : failed to sync .* cache: timed out waiting for the condition`,         // post k8s 1.16
+		`(Liveness|Readiness) probe failed: HTTP probe failed with statuscode: 50(2|3)`,
+		`(Liveness|Readiness) probe failed: Get http://.*: dial tcp .*: connect: connection refused`,
+		`(Liveness|Readiness) probe failed: Get http://.*: read tcp .*: read: connection reset by peer`,
+		`(Liveness|Readiness) probe failed: Get http://.*: net/http: request canceled .*\(Client\.Timeout exceeded while awaiting headers\)`,
+		`Failed to update endpoint .*/linkerd-.*: Operation cannot be fulfilled on endpoints "linkerd-.*": the object has been modified; please apply your changes to the latest version and try again`,
+		`error killing pod: failed to "KillPodSandbox" for ".*" with KillPodSandboxError: "rpc error: code = Unknown desc`,
+	}, "|"))
+)
+
+// FetchAndCheckLogs retrieves the logs from the control plane containers and matches them
+// to the list in knownControllerErrorsRegex and knownProxyErrorsRegex.
+// It returns the list of matched entries and the list of unmatched entries (as errors).
+func FetchAndCheckLogs(helper *TestHelper) ([]string, []error) {
+	okMessages := []string{}
+	errs := []error{}
+
+	controllerRegex := regexp.MustCompile("level=(panic|fatal|error|warn)")
+	proxyRegex := regexp.MustCompile(fmt.Sprintf("%s (ERR|WARN)", k8s.ProxyContainerName))
+	clientGoRegex := regexp.MustCompile("client-go@")
+	hasClientGoLogs := false
+
+	for deploy, spec := range LinkerdDeployReplicas {
+		deploy := strings.TrimPrefix(deploy, "linkerd-")
+		containers := append(spec.Containers, k8s.ProxyContainerName)
+
+		for _, container := range containers {
+			container := container // pin
+			name := fmt.Sprintf("%s/%s", deploy, container)
+
+			proxy := false
+			errRegex := controllerRegex
+			knownErrorsRegex := knownControllerErrorsRegex
+			if container == k8s.ProxyContainerName {
+				proxy = true
+				errRegex = proxyRegex
+				knownErrorsRegex = knownProxyErrorsRegex
+			}
+
+			outputStream, err := helper.LinkerdRunStream(
+				"logs", "--no-color",
+				"--control-plane-component", deploy,
+				"--container", container,
+			)
+			if err != nil {
+				errs = append(errs, fmt.Errorf("error running command:\n%s", err))
+				continue
+			}
+			defer outputStream.Stop()
+			// Ignore the error returned, since ReadUntil will return an error if it
+			// does not return 10,000 after 2 seconds. We don't need 10,000 log lines.
+			outputLines, _ := outputStream.ReadUntil(10000, 2*time.Second)
+			if len(outputLines) == 0 {
+				// Retry one time for 30 more seconds, in case the cluster is slow to
+				// produce log lines.
+				outputLines, _ = outputStream.ReadUntil(10000, 30*time.Second)
+				if len(outputLines) == 0 {
+					errs = append(errs, fmt.Errorf("no logs found for %s", name))
+				}
+			}
+
+			for _, line := range outputLines {
+				if errRegex.MatchString(line) {
+					if knownErrorsRegex.MatchString(line) {
+						// report all known logging errors in the output
+						okMessages = append(okMessages, fmt.Sprintf("found known error in %s log: %s", name, line))
+					} else {
+						if proxy {
+							okMessages = append(okMessages, fmt.Sprintf("found unexpected proxy error in %s log: %s", name, line))
+						} else {
+							errs = append(errs, fmt.Errorf("found unexpected controller error in %s log: %s", name, line))
+						}
+					}
+				}
+				if clientGoRegex.MatchString((line)) {
+					hasClientGoLogs = true
+				}
+			}
+		}
+	}
+	if !hasClientGoLogs {
+		errs = append(errs, errors.New("didn't find any client-go entries"))
+	}
+
+	return okMessages, errs
+}
+
+// FetchAndCheckEvents retrieves the events from k8s for the current namespace, matches
+// them against knownEventWarningsRegex and returns the list of unmatched entries
+// (as errors).
+func FetchAndCheckEvents(helper *TestHelper) []error {
+	out, err := helper.Kubectl("",
+		"--namespace", helper.GetLinkerdNamespace(),
+		"get", "events", "-ojson",
+	)
+	if err != nil {
+		return []error{fmt.Errorf("'kubectl get events' command failed with %s\n%s", err, out)}
+	}
+
+	events, err := ParseEvents(out)
+	if err != nil {
+		return []error{err}
+	}
+
+	errs := []error{}
+	for _, e := range events {
+		if e.Type == corev1.EventTypeNormal {
+			continue
+		}
+
+		evtStr := fmt.Errorf("found unexpected warning event: reason: [%s] Object: [%s] Message: [%s]", e.Reason, e.InvolvedObject.Name, e.Message)
+		if !knownEventWarningsRegex.MatchString(e.Message) {
+			errs = append(errs, evtStr)
+		}
+	}
+
+	return errs
+}
diff --git a/testutil/test_helper.go b/testutil/test_helper.go
index 7ad0e005c..0cd0ec5d7 100644
--- a/testutil/test_helper.go
+++ b/testutil/test_helper.go
@@ -13,6 +13,7 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strings"
+	"testing"
 	"time"
 
 	log "github.com/sirupsen/logrus"
@@ -482,3 +483,23 @@ func ParseEvents(out string) ([]*corev1.Event, error) {
 
 	return events, nil
 }
+
+// Run calls `m.Run()`, shows unexpected logs/events if showLogs is true,
+// and returns the exit code for the tests
+func Run(m *testing.M, helper *TestHelper, showLogs bool) int {
+	code := m.Run()
+	if code != 0 && showLogs {
+		_, errs1 := FetchAndCheckLogs(helper)
+		for _, err := range errs1 {
+			fmt.Println(err)
+		}
+		errs2 := FetchAndCheckEvents(helper)
+		for _, err := range errs2 {
+			fmt.Println(err)
+		}
+		if len(errs1) == 0 && len(errs2) == 0 {
+			fmt.Println("No unexpected log entries or events found")
+		}
+	}
+	return code
+}