Repository: linkerd/linkerd2
Pull request #8054 (2022-03-11 23:53:03)

build(deps): bump tower-http from 0.2.4 to 0.2.5
Bumps [tower-http](https://github.com/tower-rs/tower-http) from 0.2.4 to 0.2.5.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/tower-rs/tower-http/releases">tower-http's releases</a>.</em></p>
<blockquote>
<h2>v0.2.5</h2>
<h2>Fixed</h2>
<ul>
<li>Fix content-length calculation on range requests (<a href="https://github-redirect.dependabot.com/tower-rs/tower-http/issues/228">#228</a>)</li>
</ul>
<p><a href="https://github-redirect.dependabot.com/tower-rs/tower-http/issues/228">#228</a>: <a href="https://github-redirect.dependabot.com/tower-rs/tower-http/pull/228">tower-rs/tower-http#228</a></p>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/tower-rs/tower-http/commit/fa8d49ecba34dcd35294356c4aaf39359128b555"><code>fa8d49e</code></a> Release 0.2.5</li>
<li><a href="https://github.com/tower-rs/tower-http/commit/2abd741ba99c9163e72d5ea3299ceef1d3dea62f"><code>2abd741</code></a> fix content-length calculation on range requests (<a href="https://github-redirect.dependabot.com/tower-rs/tower-http/issues/228">#228</a>) (<a href="https://github-redirect.dependabot.com/tower-rs/tower-http/issues/230">#230</a>)</li>
<li>See full diff in <a href="https://github.com/tower-rs/tower-http/compare/tower-http-0.2.4...tower-http-0.2.5">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=tower-http&package-manager=cargo&previous-version=0.2.4&new-version=0.2.5)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


Patch:
diff --git a/Cargo.lock b/Cargo.lock
index ee7c7eae8..2c3f0b322 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -39,9 +39,9 @@ checksum = "4361135be9122e0870de935d7c439aef945b9f9ddd4199a553b5270b49c82a27"
 
 [[package]]
 name = "async-stream"
-version = "0.3.2"
+version = "0.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "171374e7e3b2504e0e5236e3b59260560f9fe94bfe9ac39ba5e4e929c5590625"
+checksum = "dad5c83079eae9969be7fadefe640a1c566901f05ff91ab221de4b6f68d9507e"
 dependencies = [
  "async-stream-impl",
  "futures-core",
@@ -49,9 +49,9 @@ dependencies = [
 
 [[package]]
 name = "async-stream-impl"
-version = "0.3.2"
+version = "0.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "648ed8c8d2ce5409ccd57453d9d1b214b342a0d69376a6feda1fd6cae3299308"
+checksum = "10f203db73a71dfa2fb6dd22763990fa26f3d2625a6da2da900d23b87d26be27"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -806,9 +806,9 @@ dependencies = [
 
 [[package]]
 name = "kubert"
-version = "0.3.3"
+version = "0.3.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b587b7718e2c44d3c045fdd5c3a3515317200b14319439ad41089818078d16e2"
+checksum = "db6d5039ebf63d1978760db7e835579aa790511979cb78eb9d4dcdd45f1112c5"
 dependencies = [
  "clap",
  "drain",
@@ -1954,9 +1954,9 @@ dependencies = [
 
 [[package]]
 name = "tower-http"
-version = "0.2.4"
+version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "90c125fdea84614a4368fd35786b51d0682ab8d42705e061e92f0b955dea40fb"
+checksum = "aba3f3efabf7fb41fae8534fc20a817013dd1c12cb45441efb6c82e6556b4cd8"
 dependencies = [
  "base64",
  "bitflags",
diff --git a/multicluster/cmd/service-mirror/main.go b/multicluster/cmd/service-mirror/main.go
index bae0dca1c..75ef86947 100644
--- a/multicluster/cmd/service-mirror/main.go
+++ b/multicluster/cmd/service-mirror/main.go
@@ -180,11 +180,19 @@ func restartClusterWatcher(
 		probeWorker.Stop()
 	}
 
+	// Start probe worker
+	workerMetrics, err := metrics.NewWorkerMetrics(link.TargetClusterName)
+	if err != nil {
+		return fmt.Errorf("failed to create metrics for cluster watcher: %w", err)
+	}
+	probeWorker = servicemirror.NewProbeWorker(fmt.Sprintf("probe-gateway-%s", link.TargetClusterName), &link.ProbeSpec, workerMetrics, link.TargetClusterName)
+	probeWorker.Start()
+
+	// Start cluster watcher
 	cfg, err := clientcmd.RESTConfigFromKubeConfig(creds)
 	if err != nil {
 		return fmt.Errorf("unable to parse kube config: %w", err)
 	}
-
 	clusterWatcher, err = servicemirror.NewRemoteClusterServiceWatcher(
 		ctx,
 		namespace,
@@ -193,22 +201,16 @@ func restartClusterWatcher(
 		&link,
 		requeueLimit,
 		repairPeriod,
+		probeWorker.Liveness,
 		enableHeadlessSvc,
 	)
 	if err != nil {
 		return fmt.Errorf("unable to create cluster watcher: %w", err)
 	}
-
 	err = clusterWatcher.Start(ctx)
 	if err != nil {
 		return fmt.Errorf("failed to start cluster watcher: %w", err)
 	}
 
-	workerMetrics, err := metrics.NewWorkerMetrics(link.TargetClusterName)
-	if err != nil {
-		return fmt.Errorf("failed to create metrics for cluster watcher: %w", err)
-	}
-	probeWorker = servicemirror.NewProbeWorker(fmt.Sprintf("probe-gateway-%s", link.TargetClusterName), &link.ProbeSpec, workerMetrics, link.TargetClusterName)
-	probeWorker.Start()
 	return nil
 }
diff --git a/multicluster/service-mirror/cluster_watcher.go b/multicluster/service-mirror/cluster_watcher.go
index 426caca4e..bc13e3f54 100644
--- a/multicluster/service-mirror/cluster_watcher.go
+++ b/multicluster/service-mirror/cluster_watcher.go
@@ -50,6 +50,8 @@ type (
 		eventsQueue             workqueue.RateLimitingInterface
 		requeueLimit            int
 		repairPeriod            time.Duration
+		gatewayAlive            bool
+		liveness                chan bool
 		headlessServicesEnabled bool
 	}
 
@@ -156,6 +158,7 @@ func NewRemoteClusterServiceWatcher(
 	link *multicluster.Link,
 	requeueLimit int,
 	repairPeriod time.Duration,
+	liveness chan bool,
 	enableHeadlessSvc bool,
 ) (*RemoteClusterServiceWatcher, error) {
 	remoteAPI, err := k8s.InitializeAPIForConfig(ctx, cfg, false, k8s.Svc, k8s.Endpoint)
@@ -191,6 +194,7 @@ func NewRemoteClusterServiceWatcher(
 		eventsQueue:             workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
 		requeueLimit:            requeueLimit,
 		repairPeriod:            repairPeriod,
+		liveness:                liveness,
 		headlessServicesEnabled: enableHeadlessSvc,
 	}, nil
 }
@@ -434,7 +438,8 @@ func (rcsw *RemoteClusterServiceWatcher) handleRemoteServiceUpdated(ctx context.
 	}
 	copiedEndpoints.Annotations[consts.RemoteGatewayIdentity] = rcsw.link.GatewayIdentity
 
-	if _, err := rcsw.localAPIClient.Client.CoreV1().Endpoints(copiedEndpoints.Namespace).Update(ctx, copiedEndpoints, metav1.UpdateOptions{}); err != nil {
+	err = rcsw.createOrUpdateEndpoints(ctx, copiedEndpoints)
+	if err != nil {
 		return RetryableError{[]error{err}}
 	}
 
@@ -594,8 +599,16 @@ func (rcsw *RemoteClusterServiceWatcher) createGatewayEndpoints(ctx context.Cont
 				Ports:     rcsw.getEndpointsPorts(exportedService),
 			},
 		}
+	} else if !empty {
+		endpointsToCreate.Subsets = []corev1.EndpointSubset{
+			{
+				NotReadyAddresses: gatewayAddresses,
+				Ports:             rcsw.getEndpointsPorts(exportedService),
+			},
+		}
+		rcsw.log.Warnf("could not resolve gateway addresses for %s; setting endpoint subsets to not ready", serviceInfo)
 	} else {
-		rcsw.log.Warnf("exported service is empty or gateway for %s does not have ready addresses, skipping subsets", serviceInfo)
+		rcsw.log.Warnf("exported service %s is empty", serviceInfo)
 	}
 
 	if rcsw.link.GatewayIdentity != "" {
@@ -603,7 +616,8 @@ func (rcsw *RemoteClusterServiceWatcher) createGatewayEndpoints(ctx context.Cont
 	}
 
 	rcsw.log.Infof("Creating a new endpoints for %s", serviceInfo)
-	if _, err := rcsw.localAPIClient.Client.CoreV1().Endpoints(exportedService.Namespace).Create(ctx, endpointsToCreate, metav1.CreateOptions{}); err != nil {
+	err = rcsw.createOrUpdateEndpoints(ctx, endpointsToCreate)
+	if err != nil {
 		if svcErr := rcsw.localAPIClient.Client.CoreV1().Services(exportedService.Namespace).Delete(ctx, localServiceName, metav1.DeleteOptions{}); svcErr != nil {
 			rcsw.log.Errorf("Failed to delete service %s after endpoints creation failed: %s", localServiceName, svcErr)
 		}
@@ -671,13 +685,12 @@ func (rcsw *RemoteClusterServiceWatcher) createOrUpdateService(service *corev1.S
 	return nil
 }
 
-func (rcsw *RemoteClusterServiceWatcher) getMirrorServices() ([]*corev1.Service, error) {
+func (rcsw *RemoteClusterServiceWatcher) getMirrorServices() (*corev1.ServiceList, error) {
 	matchLabels := map[string]string{
 		consts.MirroredResourceLabel:  "true",
 		consts.RemoteClusterNameLabel: rcsw.link.TargetClusterName,
 	}
-
-	services, err := rcsw.localAPIClient.Svc().Lister().List(labels.Set(matchLabels).AsSelector())
+	services, err := rcsw.localAPIClient.Client.CoreV1().Services("").List(context.Background(), metav1.ListOptions{LabelSelector: labels.SelectorFromSet(matchLabels).String()})
 	if err != nil {
 		return nil, err
 	}
@@ -859,6 +872,11 @@ func (rcsw *RemoteClusterServiceWatcher) Start(ctx context.Context) error {
 			case <-ticker.C:
 				ev := RepairEndpoints{}
 				rcsw.eventsQueue.Add(&ev)
+			case alive := <-rcsw.liveness:
+				rcsw.log.Debugf("gateway liveness change from %t to %t", rcsw.gatewayAlive, alive)
+				rcsw.gatewayAlive = alive
+				ev := RepairEndpoints{}
+				rcsw.eventsQueue.Add(&ev)
 			case <-rcsw.stopper:
 				return
 			}
@@ -900,44 +918,17 @@ func (rcsw *RemoteClusterServiceWatcher) resolveGatewayAddress() ([]corev1.Endpo
 }
 
 func (rcsw *RemoteClusterServiceWatcher) repairEndpoints(ctx context.Context) error {
-	gatewayAddresses, err := rcsw.resolveGatewayAddress()
-	if err != nil {
-		return err
-	}
-
 	endpointRepairCounter.With(prometheus.Labels{
 		gatewayClusterName: rcsw.link.TargetClusterName,
 	}).Inc()
 
-	// Create or update gateway mirror endpoints.
-	gatewayMirrorName := fmt.Sprintf("probe-gateway-%s", rcsw.link.TargetClusterName)
-
-	gatewayMirrorEndpoints := &corev1.Endpoints{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      gatewayMirrorName,
-			Namespace: rcsw.serviceMirrorNamespace,
-			Labels: map[string]string{
-				consts.RemoteClusterNameLabel: rcsw.link.TargetClusterName,
-			},
-			Annotations: map[string]string{
-				consts.RemoteGatewayIdentity: rcsw.link.GatewayIdentity,
-			},
-		},
-		Subsets: []corev1.EndpointSubset{
-			{
-				Addresses: gatewayAddresses,
-				Ports: []corev1.EndpointPort{
-					{
-						Name:     "mc-probe",
-						Port:     int32(rcsw.link.ProbeSpec.Port),
-						Protocol: "TCP",
-					},
-				},
-			},
-		},
+	// Create or update the gateway mirror endpoints responsible for driving
+	// the cluster watcher's gateway liveness status.
+	gatewayAddresses, err := rcsw.resolveGatewayAddress()
+	if err != nil {
+		return err
 	}
-
-	err = rcsw.createOrUpdateEndpoints(ctx, gatewayMirrorEndpoints)
+	err = rcsw.createOrUpdateGatewayEndpoints(ctx, gatewayAddresses)
 	if err != nil {
 		rcsw.log.Errorf("Failed to create/update gateway mirror endpoints: %s", err)
 	}
@@ -947,8 +938,8 @@ func (rcsw *RemoteClusterServiceWatcher) repairEndpoints(ctx context.Context) er
 	if err != nil {
 		rcsw.log.Errorf("Failed to list mirror services: %s", err)
 	}
-	for _, svc := range mirrorServices {
-		updatedService := svc.DeepCopy()
+	for _, svc := range mirrorServices.Items {
+		updatedService := svc
 
 		// Mirrors for headless services are also headless, and their
 		// Endpoints point to auxiliary services instead of pointing to
@@ -967,15 +958,21 @@ func (rcsw *RemoteClusterServiceWatcher) repairEndpoints(ctx context.Context) er
 
 		endpoints, err := rcsw.localAPIClient.Endpoint().Lister().Endpoints(svc.Namespace).Get(svc.Name)
 		if err != nil {
-			rcsw.log.Errorf("Could not get local endpoints: %s", err)
-			continue
+			if !kerrors.IsNotFound(err) {
+				rcsw.log.Errorf("failed to list local endpoints: %s", err)
+				continue
+			}
+			endpoints, err = rcsw.localAPIClient.Client.CoreV1().Endpoints(svc.Namespace).Get(ctx, svc.Name, metav1.GetOptions{})
+			if err != nil {
+				rcsw.log.Errorf("failed to get local endpoints %s/%s: %s", svc.Namespace, svc.Name, err)
+			}
 		}
 
 		updatedEndpoints := endpoints.DeepCopy()
 		updatedEndpoints.Subsets = []corev1.EndpointSubset{
 			{
 				Addresses: gatewayAddresses,
-				Ports:     rcsw.getEndpointsPorts(updatedService),
+				Ports:     rcsw.getEndpointsPorts(&updatedService),
 			},
 		}
 
@@ -988,6 +985,7 @@ func (rcsw *RemoteClusterServiceWatcher) repairEndpoints(ctx context.Context) er
 			continue
 		}
 		if empty {
+			rcsw.log.Warnf("exported service %s/%s is empty", targetService.Namespace, targetService.Name)
 			updatedEndpoints.Subsets = []corev1.EndpointSubset{}
 		}
 
@@ -996,13 +994,12 @@ func (rcsw *RemoteClusterServiceWatcher) repairEndpoints(ctx context.Context) er
 		}
 		updatedEndpoints.Annotations[consts.RemoteGatewayIdentity] = rcsw.link.GatewayIdentity
 
-		_, err = rcsw.localAPIClient.Client.CoreV1().Services(updatedService.Namespace).Update(ctx, updatedService, metav1.UpdateOptions{})
+		_, err = rcsw.localAPIClient.Client.CoreV1().Services(updatedService.Namespace).Update(ctx, &updatedService, metav1.UpdateOptions{})
 		if err != nil {
 			rcsw.log.Error(err)
 			continue
 		}
-
-		_, err = rcsw.localAPIClient.Client.CoreV1().Endpoints(updatedService.Namespace).Update(ctx, updatedEndpoints, metav1.UpdateOptions{})
+		err = rcsw.createOrUpdateEndpoints(ctx, updatedEndpoints)
 		if err != nil {
 			rcsw.log.Error(err)
 		}
@@ -1011,22 +1008,57 @@ func (rcsw *RemoteClusterServiceWatcher) repairEndpoints(ctx context.Context) er
 	return nil
 }
 
-func (rcsw *RemoteClusterServiceWatcher) createOrUpdateEndpoints(ctx context.Context, ep *corev1.Endpoints) error {
-	_, err := rcsw.localAPIClient.Client.CoreV1().Endpoints(ep.Namespace).Get(ctx, ep.Name, metav1.GetOptions{})
+// createOrUpdateGatewayEndpoints will create or update the gateway mirror
+// endpoints for a remote cluster. These endpoints are required for the probe
+// worker responsible for probing gateway liveness, so these endpoints are
+// never in a not ready state.
+func (rcsw *RemoteClusterServiceWatcher) createOrUpdateGatewayEndpoints(ctx context.Context, addressses []v1.EndpointAddress) error {
+	gatewayMirrorName := fmt.Sprintf("probe-gateway-%s", rcsw.link.TargetClusterName)
+	endpoints := &corev1.Endpoints{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      gatewayMirrorName,
+			Namespace: rcsw.serviceMirrorNamespace,
+			Labels: map[string]string{
+				consts.RemoteClusterNameLabel: rcsw.link.TargetClusterName,
+			},
+			Annotations: map[string]string{
+				consts.RemoteGatewayIdentity: rcsw.link.GatewayIdentity,
+			},
+		},
+		Subsets: []corev1.EndpointSubset{
+			{
+				Addresses: addressses,
+				Ports: []corev1.EndpointPort{
+					{
+						Name:     "mc-probe",
+						Port:     int32(rcsw.link.ProbeSpec.Port),
+						Protocol: "TCP",
+					},
+				},
+			},
+		},
+	}
+	_, err := rcsw.localAPIClient.Client.CoreV1().Endpoints(endpoints.Namespace).Get(ctx, endpoints.Name, metav1.GetOptions{})
 	if err != nil {
 		if !kerrors.IsNotFound(err) {
 			return err
 		}
 
-		// Does not exist so we should create it.
-		_, err = rcsw.localAPIClient.Client.CoreV1().Endpoints(ep.Namespace).Create(ctx, ep, metav1.CreateOptions{})
+		// Mirror endpoints for the gateway do not exist so they need to be
+		// created. As mentioned above, these endpoints are required for the
+		// probe worker and therefore should never be put in a not ready
+		// state.
+		_, err = rcsw.localAPIClient.Client.CoreV1().Endpoints(endpoints.Namespace).Create(ctx, endpoints, metav1.CreateOptions{})
 		if err != nil {
 			return err
 		}
+		return nil
 	}
 
-	// Exists so we should update it.
-	_, err = rcsw.localAPIClient.Client.CoreV1().Endpoints(ep.Namespace).Update(ctx, ep, metav1.UpdateOptions{})
+	// Mirror endpoints for the gateway already exist so they need to be
+	// updated. As mentioned above, these endpoints are required for the probe
+	// worker and therefore should never be put in a not ready state.
+	_, err = rcsw.localAPIClient.Client.CoreV1().Endpoints(endpoints.Namespace).Update(ctx, endpoints, metav1.UpdateOptions{})
 	return err
 }
 
@@ -1078,8 +1110,33 @@ func (rcsw *RemoteClusterServiceWatcher) handleCreateOrUpdateEndpoints(
 			},
 		}
 	}
+	return rcsw.createOrUpdateEndpoints(ctx, ep)
+}
 
-	_, err = rcsw.localAPIClient.Client.CoreV1().Endpoints(exportedEndpoints.Namespace).Update(ctx, ep, metav1.UpdateOptions{})
+// createOrUpdateEndpoints will create or update endpoints based off gateway
+// liveness. If the gateway is not alive, then the addresses in each subset
+// will be set to not ready; these will moved back to ready once the gateway
+// is alive and an update is received.
+func (rcsw *RemoteClusterServiceWatcher) createOrUpdateEndpoints(ctx context.Context, endpoints *corev1.Endpoints) error {
+	if !rcsw.gatewayAlive {
+		rcsw.log.Warnf("gateway for %s/%s does not have ready addresses; setting addresses to not ready", endpoints.Namespace, endpoints.Name)
+		for i := range endpoints.Subsets {
+			endpoints.Subsets[i].NotReadyAddresses = append(endpoints.Subsets[i].NotReadyAddresses, endpoints.Subsets[i].Addresses...)
+			endpoints.Subsets[i].Addresses = nil
+		}
+	}
+	_, err := rcsw.localAPIClient.Client.CoreV1().Endpoints(endpoints.Namespace).Get(ctx, endpoints.Name, metav1.GetOptions{})
+	if err != nil {
+		if !kerrors.IsNotFound(err) {
+			return fmt.Errorf("Failed to get endpoints for %s/%s: %w", endpoints.Namespace, endpoints.Name, err)
+		}
+		_, err := rcsw.localAPIClient.Client.CoreV1().Endpoints(endpoints.Namespace).Create(ctx, endpoints, metav1.CreateOptions{})
+		if err != nil {
+			return fmt.Errorf("Failed to create endpoints for %s/%s: %w", endpoints.Namespace, endpoints.Name, err)
+		}
+		return nil
+	}
+	_, err = rcsw.localAPIClient.Client.CoreV1().Endpoints(endpoints.Namespace).Update(ctx, endpoints, metav1.UpdateOptions{})
 	return err
 }
 
diff --git a/multicluster/service-mirror/cluster_watcher_headless.go b/multicluster/service-mirror/cluster_watcher_headless.go
index 6bad31777..5b44e1045 100644
--- a/multicluster/service-mirror/cluster_watcher_headless.go
+++ b/multicluster/service-mirror/cluster_watcher_headless.go
@@ -167,9 +167,9 @@ func (rcsw *RemoteClusterServiceWatcher) createOrUpdateHeadlessEndpoints(ctx con
 		return RetryableError{errors}
 	}
 
-	// Update
+	// Update endpoints
 	mirrorEndpoints.Subsets = newSubsets
-	_, err = rcsw.localAPIClient.Client.CoreV1().Endpoints(mirrorEndpoints.Namespace).Update(ctx, mirrorEndpoints, metav1.UpdateOptions{})
+	err = rcsw.createOrUpdateEndpoints(ctx, mirrorEndpoints)
 	if err != nil {
 		return RetryableError{[]error{err}}
 	}
@@ -377,6 +377,13 @@ func (rcsw *RemoteClusterServiceWatcher) createEndpointMirrorService(ctx context
 		}
 	}
 
+	// todo: The endpoints are created without checking for gateway liveness.
+	// We do this because if we do check — and the gateway is down — these endpoints are
+	// never repaired. This is because in repairEndpoints we skip local
+	// endpoints that are mirroring headless services since they may not have
+	// a matching endpoint on the remote cluster.
+	//
+	// Explained by cluster_watcher.go L943-L945.
 	rcsw.log.Infof("Creating a new endpoints object for endpoint mirror service %s", endpointMirrorInfo)
 	if _, err := rcsw.localAPIClient.Client.CoreV1().Endpoints(endpointMirrorService.Namespace).Create(ctx, endpointMirrorEndpoints, metav1.CreateOptions{}); err != nil {
 		if svcErr := rcsw.localAPIClient.Client.CoreV1().Services(endpointMirrorService.Namespace).Delete(ctx, endpointMirrorName, metav1.DeleteOptions{}); svcErr != nil {
diff --git a/multicluster/service-mirror/probe_worker.go b/multicluster/service-mirror/probe_worker.go
index cf983029c..a1d0104e2 100644
--- a/multicluster/service-mirror/probe_worker.go
+++ b/multicluster/service-mirror/probe_worker.go
@@ -16,6 +16,8 @@ const httpGatewayTimeoutMillis = 50000
 // ProbeWorker is responsible for monitoring gateways using a probe specification
 type ProbeWorker struct {
 	localGatewayName string
+	alive            bool
+	Liveness         chan bool
 	*sync.RWMutex
 	probeSpec *multicluster.ProbeSpec
 	stopCh    chan struct{}
@@ -27,6 +29,7 @@ type ProbeWorker struct {
 func NewProbeWorker(localGatewayName string, spec *multicluster.ProbeSpec, metrics *ProbeMetrics, probekey string) *ProbeWorker {
 	return &ProbeWorker{
 		localGatewayName: localGatewayName,
+		Liveness:         make(chan bool, 10),
 		RWMutex:          &sync.RWMutex{},
 		probeSpec:        spec,
 		stopCh:           make(chan struct{}),
@@ -97,17 +100,29 @@ func (pw *ProbeWorker) doProbe() {
 		pw.log.Warnf("Problem connecting with gateway. Marking as unhealthy %s", err)
 		pw.metrics.alive.Set(0)
 		pw.metrics.probes.With(notSuccessLabel).Inc()
+		if pw.alive {
+			pw.alive = false
+			pw.Liveness <- false
+		}
 		return
 	}
 	if resp.StatusCode != 200 {
 		pw.log.Warnf("Gateway returned unexpected status %d. Marking as unhealthy", resp.StatusCode)
 		pw.metrics.alive.Set(0)
 		pw.metrics.probes.With(notSuccessLabel).Inc()
+		if pw.alive {
+			pw.alive = false
+			pw.Liveness <- false
+		}
 	} else {
 		pw.log.Debug("Gateway is healthy")
 		pw.metrics.alive.Set(1)
 		pw.metrics.latencies.Observe(float64(end.Milliseconds()))
 		pw.metrics.probes.With(successLabel).Inc()
+		if !pw.alive {
+			pw.alive = true
+			pw.Liveness <- true
+		}
 	}
 
 	if err := resp.Body.Close(); err != nil {
diff --git a/policy-controller/Cargo.toml b/policy-controller/Cargo.toml
index a5cf2eb68..a5d380ded 100644
--- a/policy-controller/Cargo.toml
+++ b/policy-controller/Cargo.toml
@@ -35,7 +35,7 @@ features = [
 ]
 
 [dependencies.kubert]
-version = "0.3.3"
+version = "0.3.5"
 default-features = false
 features = [
     "clap",
diff --git a/policy-controller/k8s/index/Cargo.toml b/policy-controller/k8s/index/Cargo.toml
index 2152e1a70..dc901e8de 100644
--- a/policy-controller/k8s/index/Cargo.toml
+++ b/policy-controller/k8s/index/Cargo.toml
@@ -11,7 +11,7 @@ anyhow = "1"
 async-stream = "0.3"
 async-trait = "0.1"
 futures = { version = "0.3", default-features = false }
-kubert = { version = "0.3.3", default-features = false, features = ["initialized"] }
+kubert = { version = "0.3.5", default-features = false, features = ["initialized"] }
 linkerd-policy-controller-core = { path = "../../core" }
 linkerd-policy-controller-k8s-api = { path = "../api" }
 parking_lot = "0.12"
diff --git a/viz/tap/api/handlers.go b/viz/tap/api/handlers.go
index 562951ed5..d740dca27 100644
--- a/viz/tap/api/handlers.go
+++ b/viz/tap/api/handlers.go
@@ -157,7 +157,7 @@ func (h *handler) handleTap(w http.ResponseWriter, req *http.Request, p httprout
 
 	url := pkg.TapReqToURL(&tapReq)
 	if url != req.URL.Path {
-		err = fmt.Errorf("tap request body did not match APIServer URL: %+v != %q", url, req.URL.Path)
+		err = fmt.Errorf("tap request body did not match APIServer URL: %q != %q", url, req.URL.Path)
 		h.log.Error(err)
 		protohttp.WriteErrorToHTTPResponse(w, err)
 		return
@@ -175,7 +175,7 @@ func (h *handler) handleTap(w http.ResponseWriter, req *http.Request, p httprout
 	//nolint:staticcheck
 	err = h.grpcTapServer.TapByResource(&tapReq, &serverStream)
 	if err != nil {
-		h.log.Error(err)
+		h.log.Errorf("TapByResource failed: %q", err)
 		protohttp.WriteErrorToHTTPResponse(flushableWriter, err)
 		return
 	}
diff --git a/web/app/package.json b/web/app/package.json
index 63af95c02..0ca22f7ac 100644
--- a/web/app/package.json
+++ b/web/app/package.json
@@ -56,7 +56,7 @@
     "babel-plugin-transform-react-remove-prop-types": "0.4.24",
     "chai": "4.3.6",
     "clean-webpack-plugin": "4.0.0",
-    "css-loader": "^6.7.0",
+    "css-loader": "^6.7.1",
     "enzyme": "3.11.0",
     "enzyme-adapter-react-16": "^1.15.6",
     "eslint": "^8.10.0",
diff --git a/web/app/yarn.lock b/web/app/yarn.lock
index 78f20e591..c9daaea83 100644
--- a/web/app/yarn.lock
+++ b/web/app/yarn.lock
@@ -2990,10 +2990,10 @@ cross-spawn@^7.0.2, cross-spawn@^7.0.3:
     shebang-command "^2.0.0"
     which "^2.0.1"
 
-css-loader@^6.7.0:
-  version "6.7.0"
-  resolved "https://registry.yarnpkg.com/css-loader/-/css-loader-6.7.0.tgz#c1200da1dfffe6643b18bda20fdd84cad3e36d39"
-  integrity sha512-S7HCfCiDHLA+VXKqdZwyRZgoO0R9BnKDnVIoHMq5grl3N86zAu7MB+FBWHr5xOJC8SmvpTLha/2NpfFkFEN/ig==
+css-loader@^6.7.1:
+  version "6.7.1"
+  resolved "https://registry.yarnpkg.com/css-loader/-/css-loader-6.7.1.tgz#e98106f154f6e1baf3fc3bc455cb9981c1d5fd2e"
+  integrity sha512-yB5CNFa14MbPJcomwNh3wLThtkZgcNyI2bNMRt8iE5Z8Vwl7f8vQXFAzn2HDOJvtDq2NTZBUGMSUNNyrv3/+cw==
   dependencies:
     icss-utils "^5.1.0"
     postcss "^8.4.7"


Test patch:
diff --git a/multicluster/service-mirror/cluster_watcher_mirroring_test.go b/multicluster/service-mirror/cluster_watcher_mirroring_test.go
index fa6e9c0ee..c062e470b 100644
--- a/multicluster/service-mirror/cluster_watcher_mirroring_test.go
+++ b/multicluster/service-mirror/cluster_watcher_mirroring_test.go
@@ -242,6 +242,7 @@ func TestLocalNamespaceCreatedAfterServiceExport(t *testing.T) {
 		log:                     logging.WithFields(logging.Fields{"cluster": clusterName}),
 		eventsQueue:             q,
 		requeueLimit:            0,
+		gatewayAlive:            true,
 		headlessServicesEnabled: true,
 	}
 
@@ -293,6 +294,251 @@ func TestLocalNamespaceCreatedAfterServiceExport(t *testing.T) {
 	}
 }
 
+func TestServiceCreatedGatewayAlive(t *testing.T) {
+	remoteAPI, err := k8s.NewFakeAPI(
+		gatewayAsYaml("gateway", "gateway-ns", "1", "192.0.0.1", "gateway", 888, "gateway-identity", defaultProbePort, defaultProbePath, defaultProbePeriod),
+		remoteServiceAsYaml("svc", "ns", "1", []corev1.ServicePort{}),
+		endpointsAsYaml("svc", "ns", "192.0.0.1", "gateway-identity", []corev1.EndpointPort{}),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	localAPI, err := k8s.NewFakeAPI(
+		namespaceAsYaml("ns"),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	remoteAPI.Sync(nil)
+	localAPI.Sync(nil)
+
+	events := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
+	watcher := RemoteClusterServiceWatcher{
+		link: &multicluster.Link{
+			TargetClusterName:   clusterName,
+			TargetClusterDomain: clusterDomain,
+			GatewayIdentity:     "gateway-identity",
+			GatewayAddress:      "192.0.0.1",
+			GatewayPort:         888,
+			ProbeSpec:           defaultProbeSpec,
+			Selector:            *defaultSelector,
+		},
+		remoteAPIClient: remoteAPI,
+		localAPIClient:  localAPI,
+		log:             logging.WithFields(logging.Fields{"cluster": clusterName}),
+		eventsQueue:     events,
+		requeueLimit:    0,
+		gatewayAlive:    true,
+	}
+
+	events.Add(&RemoteServiceCreated{
+		service: remoteService("svc", "ns", "1", map[string]string{
+			consts.DefaultExportedServiceSelector: "true",
+		}, []corev1.ServicePort{
+			{
+				Name:     "port",
+				Protocol: "TCP",
+				Port:     111,
+			},
+		}),
+	})
+	for events.Len() > 0 {
+		watcher.processNextEvent(context.Background())
+	}
+
+	// Expect Service svc-remote to be created with ready endpoints because
+	// the Namespace ns exists and the gateway is alive.
+	_, err = localAPI.Client.CoreV1().Services("ns").Get(context.Background(), "svc-remote", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("error getting svc-remote Service: %v", err)
+	}
+	endpoints, err := localAPI.Client.CoreV1().Endpoints("ns").Get(context.Background(), "svc-remote", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("error getting svc-remote Endpoints: %v", err)
+	}
+	if len(endpoints.Subsets) == 0 {
+		t.Fatal("expected svc-remote Endpoints subsets")
+	}
+	for _, ss := range endpoints.Subsets {
+		if len(ss.Addresses) == 0 {
+			t.Fatal("svc-remote Endpoints should have addresses")
+		}
+		if len(ss.NotReadyAddresses) != 0 {
+			t.Fatalf("svc-remote Endpoints should not have not ready addresses: %v", ss.NotReadyAddresses)
+		}
+	}
+
+	// The gateway is now down which triggers repairing Endpoints on the local
+	// cluster.
+	watcher.gatewayAlive = false
+	events.Add(&RepairEndpoints{})
+	for events.Len() > 0 {
+		watcher.processNextEvent(context.Background())
+	}
+
+	// When repairing Endpoints on the local cluster, the gateway address
+	// should have been moved to NotReadyAddresses meaning that Endpoints
+	// for the mirrored Service svc-remote should have no ready addresses.
+	endpoints, err = localAPI.Client.CoreV1().Endpoints("ns").Get(context.Background(), "svc-remote", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("error getting svc-remote Endpoints locally: %v", err)
+	}
+	if len(endpoints.Subsets) == 0 {
+		t.Fatal("expected svc-remote Endpoints subsets")
+	}
+	for _, ss := range endpoints.Subsets {
+		if len(ss.NotReadyAddresses) == 0 {
+			t.Fatal("svc-remote Endpoints should have not ready addresses")
+		}
+		if len(ss.Addresses) != 0 {
+			t.Fatalf("svc-remote Endpoints should not have addresses: %v", ss.Addresses)
+		}
+	}
+
+	// Issue an update for the remote Service which adds a new label
+	// 'new-label'. This should exercise RemoteServiceUpdated which should
+	// update svc-remote; the gateway is still not alive though so we expect
+	// the Endpoints of svc-remote to still have no ready addresses.
+	events.Add(&RemoteServiceUpdated{
+		localService:   remoteService("svc-remote", "ns", "2", nil, nil),
+		localEndpoints: endpoints,
+		remoteUpdate: remoteService("svc", "ns", "2", map[string]string{
+			consts.DefaultExportedServiceSelector: "true",
+			"new-label":                           "hi",
+		}, []corev1.ServicePort{
+			{
+				Name:     "port",
+				Protocol: "TCP",
+				Port:     111,
+			},
+		}),
+	})
+	for events.Len() > 0 {
+		watcher.processNextEvent(context.Background())
+	}
+	service, err := localAPI.Client.CoreV1().Services("ns").Get(context.Background(), "svc-remote", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("error getting svc-remote Service: %v", err)
+	}
+	_, ok := service.Labels["new-label"]
+	if !ok {
+		t.Fatalf("error updating svc-remote Service: %v", err)
+	}
+	endpoints, err = localAPI.Client.CoreV1().Endpoints("ns").Get(context.Background(), "svc-remote", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("error getting svc-remote Endpoints: %v", err)
+	}
+	if len(endpoints.Subsets) == 0 {
+		t.Fatal("expected svc-remote Endpoints subsets")
+	}
+	for _, ss := range endpoints.Subsets {
+		if len(ss.NotReadyAddresses) == 0 {
+			t.Fatal("svc-remote Endpoints should have not ready addresses")
+		}
+		if len(ss.Addresses) != 0 {
+			t.Fatalf("svc-remote Endpoints should not have addresses: %v", ss.Addresses)
+		}
+	}
+}
+
+func TestServiceCreatedGatewayDown(t *testing.T) {
+	remoteAPI, err := k8s.NewFakeAPI(
+		gatewayAsYaml("gateway", "gateway-ns", "1", "192.0.0.1", "gateway", 888, "gateway-identity", defaultProbePort, defaultProbePath, defaultProbePeriod),
+		remoteServiceAsYaml("svc", "ns", "1", []corev1.ServicePort{}),
+		endpointsAsYaml("svc", "ns", "192.0.0.1", "gateway-identity", []corev1.EndpointPort{}),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	localAPI, err := k8s.NewFakeAPI(
+		namespaceAsYaml("ns"),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	remoteAPI.Sync(nil)
+	localAPI.Sync(nil)
+
+	events := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
+	watcher := RemoteClusterServiceWatcher{
+		link: &multicluster.Link{
+			TargetClusterName:   clusterName,
+			TargetClusterDomain: clusterDomain,
+			GatewayIdentity:     "gateway-identity",
+			GatewayAddress:      "192.0.0.1",
+			GatewayPort:         888,
+			ProbeSpec:           defaultProbeSpec,
+			Selector:            *defaultSelector,
+		},
+		remoteAPIClient: remoteAPI,
+		localAPIClient:  localAPI,
+		log:             logging.WithFields(logging.Fields{"cluster": clusterName}),
+		eventsQueue:     events,
+		requeueLimit:    0,
+		gatewayAlive:    false,
+	}
+
+	events.Add(&RemoteServiceCreated{
+		service: remoteService("svc", "ns", "1", map[string]string{
+			consts.DefaultExportedServiceSelector: "true",
+		}, []corev1.ServicePort{
+			{
+				Name:     "port",
+				Protocol: "TCP",
+				Port:     111,
+			},
+		}),
+	})
+	for events.Len() > 0 {
+		watcher.processNextEvent(context.Background())
+	}
+
+	// Expect Service svc-remote to be created with Endpoints subsets
+	// that are not ready because the gateway is down.
+	_, err = localAPI.Client.CoreV1().Services("ns").Get(context.Background(), "svc-remote", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("error getting svc-remote Service: %v", err)
+	}
+	endpoints, err := localAPI.Client.CoreV1().Endpoints("ns").Get(context.Background(), "svc-remote", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("error getting svc-remote Endpoints: %v", err)
+	}
+	if len(endpoints.Subsets) == 0 {
+		t.Fatal("expected svc-remote Endpoints subsets")
+	}
+	for _, ss := range endpoints.Subsets {
+		if len(ss.NotReadyAddresses) == 0 {
+			t.Fatal("svc-remote Endpoints should have not ready addresses")
+		}
+		if len(ss.Addresses) != 0 {
+			t.Fatalf("svc-remote Endpoints should not have addresses: %v", ss.Addresses)
+		}
+	}
+
+	// The gateway is now alive which triggers repairing Endpoints on the
+	// local cluster.
+	watcher.gatewayAlive = true
+	events.Add(&RepairEndpoints{})
+	for events.Len() > 0 {
+		watcher.processNextEvent(context.Background())
+	}
+	endpoints, err = localAPI.Client.CoreV1().Endpoints("ns").Get(context.Background(), "svc-remote", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("error getting svc-remote Endpoints locally: %v", err)
+	}
+	if len(endpoints.Subsets) == 0 {
+		t.Fatal("expected svc-remote Endpoints subsets")
+	}
+	for _, ss := range endpoints.Subsets {
+		if len(ss.Addresses) == 0 {
+			t.Fatal("svc-remote Endpoints should have addresses")
+		}
+		if len(ss.NotReadyAddresses) != 0 {
+			t.Fatalf("svc-remote Service endpoints should not have not ready addresses: %v", ss.NotReadyAddresses)
+		}
+	}
+}
+
 func TestRemoteServiceDeletedMirroring(t *testing.T) {
 	for _, tt := range []mirroringTestCase{
 		{
diff --git a/multicluster/service-mirror/cluster_watcher_test_util.go b/multicluster/service-mirror/cluster_watcher_test_util.go
index a276a12eb..f811eaf8e 100644
--- a/multicluster/service-mirror/cluster_watcher_test_util.go
+++ b/multicluster/service-mirror/cluster_watcher_test_util.go
@@ -62,6 +62,7 @@ func (te *testEnvironment) runEnvironment(watcherQueue workqueue.RateLimitingInt
 		log:                     logging.WithFields(logging.Fields{"cluster": clusterName}),
 		eventsQueue:             watcherQueue,
 		requeueLimit:            0,
+		gatewayAlive:            true,
 		headlessServicesEnabled: true,
 	}