Repository: linkerd/linkerd2
Pull request #4510 (2020-05-29 14:56:30)

Probe manager simplification
There are a few notable things happening in this PR: 

- the probe manager has been decoupled from the cluster_watcher. Now its only responsibility is to watch for mirrored gateways beeing created and to probe them. This means that probes are initiated for all gateways no matter whether there are mirrored services being paired
- the number of paired services is derived from the existing services in the cluster rather than being published as a metric by the prober
- there are no events being exchanged between the cluster watcher and the probe manager

zaharidichev: I will most likely now take a pass at annotation/labels namings 

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)