Repository: linkerd/linkerd2
Pull request #4134 (2020-03-04 15:03:48)

Enable dst service to work with remote gateway endpoints
The purpose of this change is to enable the destinations service to recognize remote gateway endpoints and be able to set the corresponding labels, identitiy and the rest of the data that will be needed for the proxy to handle the traffic properly. 

@adleong since you mostly wrote that code, it would be good to get your opinion on the direction of these changes. Effectivelly I want to express the idea that endpoints and hende `WeightedAddress` instaces can be the product of things different than pods. Still have to add more docs and write some more tests for this.

Description: 

The most prominant change here is that we break up the concrete `Address` type into three types: 

- `PodAddress` -  is an address/port combo that comes from Endpoints that have been created because of pods
- `HeadlessServiceAddress` - what it sounds like. These have been created because of a service with an associated Endpoints object that does not have ties to a Pod
- `RemoteGatewayAddress` - this is a subcase of `HeadlessServiceAddress`  but contains additional infromation for remote gateway identitiy and service fq name. 

Note that this PR is the first part of this change.  We still need to set these annotations when we create endpoitns in the mirroring controller. We also need to add metrics labels, etc and we need to eventually change the API to allow for carying data about the remote gateway (such as fully qualified name of the service we are targeting on the other end). These will come as a seprate smaller PRs. 


Signed-off-by: Zahari Dichev <zaharidichev@gmail.com>


olix0r: @zaharidichev I don't want to create extra work, but would it be possible to get a "clean" change to the existing endpoints code without introducing the new features, and then following on with a PR that adds remote gateway support?
olix0r: > would it be possible to get a "clean" change

Actually, on second look, the actual surface area of this change is smaller than it first appeared (thanks gofmt). If I understand correctly, this is basically introduce a polymorpish address type instead of the single concrete `Address` type? If so, i don't know a separate PR is worth it.
zaharidichev: > If I understand correctly, this is basically introduce a polymorpish address type instead of the single concrete Address type?


@olix0r that is the just of it (for some definition of polymorpish that Go adopts .... )

Patch:
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 08de82bf8..010165911 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -289,7 +289,7 @@ jobs:
         until RES=$(curl -sL https://run.linkerd.io/$INSTALL | grep "LINKERD2_VERSION=\${LINKERD2_VERSION:-$TAG}") \
           || (( count++ >= 10 ))
         do
-          sleep 5
+          sleep 30
         done
         if [[ -z $RES ]]; then
           echo "::error::The version '$TAG' was NOT found published in the website"
diff --git a/charts/linkerd2/README.md b/charts/linkerd2/README.md
index 4d576c57d..d5e1a3400 100644
--- a/charts/linkerd2/README.md
+++ b/charts/linkerd2/README.md
@@ -155,6 +155,28 @@ The following table lists the configurable parameters of the Linkerd2 chart and
 | `webhookFailurePolicy`                | Failure policy for the proxy injector                                                                                                                                                 | `Ignore`                             |
 | `webImage`                            | Docker image for the web container                                                                                                                                                    | `gcr.io/linkerd-io/web`              |
 
+## Add-Ons Configuration
+
+### Tracing Add-On
+
+The following table lists the configurable parameters for the Tracing Add-On.
+
+| Parameter                             | Description                                                                                                                                                                           | Default                              |
+|:--------------------------------------|:--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|:-------------------------------------|
+| `tracing.enabled`                     | Flag to enable tracing components to be installed                                                                                                                                                | `false`
+| `tracing.collector.name`                | Name of the trace collector Service                                                                                                                                                 | `linkerd-collector`                             |
+| `tracing.collector.image`                | Docker image for the trace collector                                                                                                                                                 | `omnition/opencensus-collector:0.1.10`                             |
+| `tracing.collector.resources.cpu.limit`       | Maximum amount of CPU units that the trace collector container can use                                                                                                                     | `1`                               |
+| `tracing.collector.resources.cpu.request`     | Amount of CPU units that the trace collector container requests                                                                                                                            | `200m`                                |
+| `tracing.collector.resources.memory.limit`    | Maximum amount of memory that the trace collector container can use                                                                                                                        | `2Gi`                               |
+| `tracing.collector.resources.memory.request`  | Amount of memory that the trace collector container requests                                                                                                                               | `400Mi`                               |
+| `tracing.jaeger.name`                | Name of the jaeger instance                                                                                                                                                 | `linkerd-jaeger`                             |
+| `tracing.jaeger.image`                | Docker image for the jaeger instance                                                                                                                                                 | `jaegertracing/all-in-one:1.8`                             |
+| `tracing.jaeger.resources.cpu.limit`       | Maximum amount of CPU units that the jaeger container can use                                                                                                                     ||
+| `tracing.jaeger.resources.cpu.request`     | Amount of CPU units that the jaeger container requests                                                                                                                            ||
+| `tracing.jaeger.resources.memory.limit`    | Maximum amount of memory that the jaeger container can use                                                                                                                        ||
+| `tracing.jaeger.resources.memory.request`  | Amount of memory that the jaeger container requests                                                                                                                               ||
+
 ## Get involved
 
 * Check out Linkerd's source code at [Github][linkerd2].
diff --git a/controller/api/destination/endpoint_translator.go b/controller/api/destination/endpoint_translator.go
index d8e30d737..c4acdbc7c 100644
--- a/controller/api/destination/endpoint_translator.go
+++ b/controller/api/destination/endpoint_translator.go
@@ -39,9 +39,9 @@ func newEndpointTranslator(
 	return &endpointTranslator{controllerNS, identityTrustDomain, enableH2Upgrade, stream, log}
 }
 
-func (et *endpointTranslator) Add(set watcher.PodSet) {
+func (et *endpointTranslator) Add(set watcher.AddressSet) {
 	addrs := []*pb.WeightedAddr{}
-	for _, address := range set.Pods {
+	for _, address := range set.Addresses {
 		var (
 			wa  *pb.WeightedAddr
 			err error
@@ -49,14 +49,31 @@ func (et *endpointTranslator) Add(set watcher.PodSet) {
 		if address.Pod != nil {
 			wa, err = et.toWeightedAddr(address)
 		} else {
-			// handling address with no associated pod, such as those
-			// belonging to headless services
+			// handling address with no associated pod
 			var addr *net.TcpAddress
 			addr, err = et.toAddr(address)
 			wa = &pb.WeightedAddr{
 				Addr:   addr,
 				Weight: defaultWeight,
 			}
+
+			if address.Identity != "" {
+				wa.TlsIdentity = &pb.TlsIdentity{
+					Strategy: &pb.TlsIdentity_DnsLikeIdentity_{
+						DnsLikeIdentity: &pb.TlsIdentity_DnsLikeIdentity{
+							Name: address.Identity,
+						},
+					},
+				}
+				// in this case we most likely have a proxy on the side, so set protocol hint as well.
+				if et.enableH2Upgrade {
+					wa.ProtocolHint = &pb.ProtocolHint{
+						Protocol: &pb.ProtocolHint_H2_{
+							H2: &pb.ProtocolHint_H2{},
+						},
+					}
+				}
+			}
 		}
 		if err != nil {
 			et.log.Errorf("Failed to translate endpoints to weighted addr: %s", err)
@@ -78,9 +95,9 @@ func (et *endpointTranslator) Add(set watcher.PodSet) {
 	}
 }
 
-func (et *endpointTranslator) Remove(set watcher.PodSet) {
+func (et *endpointTranslator) Remove(set watcher.AddressSet) {
 	addrs := []*net.TcpAddress{}
-	for _, address := range set.Pods {
+	for _, address := range set.Addresses {
 		tcpAddr, err := et.toAddr(address)
 		if err != nil {
 			et.log.Errorf("Failed to translate endpoints to addr: %s", err)
diff --git a/controller/api/destination/watcher/endpoints_watcher.go b/controller/api/destination/watcher/endpoints_watcher.go
index 875c99370..a1ecdfe99 100644
--- a/controller/api/destination/watcher/endpoints_watcher.go
+++ b/controller/api/destination/watcher/endpoints_watcher.go
@@ -7,6 +7,7 @@ import (
 	"sync"
 
 	"github.com/linkerd/linkerd2/controller/k8s"
+	consts "github.com/linkerd/linkerd2/pkg/k8s"
 	"github.com/prometheus/client_golang/prometheus"
 	logging "github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
@@ -24,19 +25,24 @@ const (
 // https://github.com/linkerd/linkerd2/issues/2204
 
 type (
-	// Address represents an individual port on an specific pod.
+	// Address represents an individual port on a specific endpoint.
+	// This endpoint might be the result of a the existence of a pod
+	// that is targeted by this service; alternatively it can be the
+	// case that this endpoint is not associated with a pod and maps
+	// to some other IP (i.e. a remote service gateway)
 	Address struct {
 		IP        string
 		Port      Port
 		Pod       *corev1.Pod
 		OwnerName string
 		OwnerKind string
+		Identity  string
 	}
 
-	// PodSet is a set of pods, indexed by IP.
-	PodSet struct {
-		Pods   map[PodID]Address
-		Labels map[string]string
+	// AddressSet is a set of Address, indexed by ID.
+	AddressSet struct {
+		Addresses map[ID]Address
+		Labels    map[string]string
 	}
 
 	portAndHostname struct {
@@ -89,15 +95,15 @@ type (
 		k8sAPI     *k8s.API
 
 		exists    bool
-		pods      PodSet
+		addresses AddressSet
 		listeners []EndpointUpdateListener
 		metrics   endpointsMetrics
 	}
 
 	// EndpointUpdateListener is the interface that subscribers must implement.
 	EndpointUpdateListener interface {
-		Add(set PodSet)
-		Remove(set PodSet)
+		Add(set AddressSet)
+		Remove(set AddressSet)
 		NoEndpoints(exists bool)
 	}
 )
@@ -417,32 +423,32 @@ func (sp *servicePublisher) metricsLabels(port Port, hostname string) prometheus
 // portPublisher.
 
 func (pp *portPublisher) updateEndpoints(endpoints *corev1.Endpoints) {
-	newPods := pp.endpointsToAddresses(endpoints)
-	if len(newPods.Pods) == 0 {
+	newAddressSet := pp.endpointsToAddresses(endpoints)
+	if len(newAddressSet.Addresses) == 0 {
 		for _, listener := range pp.listeners {
 			listener.NoEndpoints(true)
 		}
 	} else {
-		add, remove := diffPods(pp.pods, newPods)
+		add, remove := diffAddresses(pp.addresses, newAddressSet)
 		for _, listener := range pp.listeners {
-			if len(remove.Pods) > 0 {
+			if len(remove.Addresses) > 0 {
 				listener.Remove(remove)
 			}
-			if len(add.Pods) > 0 {
+			if len(add.Addresses) > 0 {
 				listener.Add(add)
 			}
 		}
 	}
 	pp.exists = true
-	pp.pods = newPods
+	pp.addresses = newAddressSet
 
 	pp.metrics.incUpdates()
-	pp.metrics.setPods(len(pp.pods.Pods))
+	pp.metrics.setPods(len(pp.addresses.Addresses))
 	pp.metrics.setExists(true)
 }
 
-func (pp *portPublisher) endpointsToAddresses(endpoints *corev1.Endpoints) PodSet {
-	pods := make(map[PodID]Address)
+func (pp *portPublisher) endpointsToAddresses(endpoints *corev1.Endpoints) AddressSet {
+	addresses := make(map[ID]Address)
 	for _, subset := range endpoints.Subsets {
 		resolvedPort := pp.resolveTargetPort(subset)
 		for _, endpoint := range subset.Addresses {
@@ -458,9 +464,11 @@ func (pp *portPublisher) endpointsToAddresses(endpoints *corev1.Endpoints) PodSe
 					}, "-"),
 					Namespace: endpoints.ObjectMeta.Namespace,
 				}
-				pods[id] = Address{
-					IP:   endpoint.IP,
-					Port: resolvedPort,
+
+				addresses[id] = Address{
+					IP:       endpoint.IP,
+					Port:     resolvedPort,
+					Identity: endpoints.Annotations[consts.RemoteGatewayIdentity],
 				}
 				continue
 			}
@@ -475,7 +483,7 @@ func (pp *portPublisher) endpointsToAddresses(endpoints *corev1.Endpoints) PodSe
 					continue
 				}
 				ownerKind, ownerName := pp.k8sAPI.GetOwnerKindAndName(pod, false)
-				pods[id] = Address{
+				addresses[id] = Address{
 					IP:        endpoint.IP,
 					Port:      resolvedPort,
 					Pod:       pod,
@@ -485,9 +493,9 @@ func (pp *portPublisher) endpointsToAddresses(endpoints *corev1.Endpoints) PodSe
 			}
 		}
 	}
-	return PodSet{
-		Pods:   pods,
-		Labels: map[string]string{"service": endpoints.Name, "namespace": endpoints.Namespace},
+	return AddressSet{
+		Addresses: addresses,
+		Labels:    map[string]string{"service": endpoints.Name, "namespace": endpoints.Namespace},
 	}
 }
 
@@ -517,7 +525,7 @@ func (pp *portPublisher) updatePort(targetPort namedPort) {
 
 func (pp *portPublisher) noEndpoints(exists bool) {
 	pp.exists = exists
-	pp.pods = PodSet{}
+	pp.addresses = AddressSet{}
 	for _, listener := range pp.listeners {
 		listener.NoEndpoints(exists)
 	}
@@ -529,8 +537,8 @@ func (pp *portPublisher) noEndpoints(exists bool) {
 
 func (pp *portPublisher) subscribe(listener EndpointUpdateListener) {
 	if pp.exists {
-		if len(pp.pods.Pods) > 0 {
-			listener.Add(pp.pods)
+		if len(pp.addresses.Addresses) > 0 {
+			listener.Add(pp.addresses)
 		} else {
 			listener.NoEndpoints(true)
 		}
@@ -584,28 +592,28 @@ func getTargetPort(service *corev1.Service, port Port) namedPort {
 	return targetPort
 }
 
-func diffPods(oldPods, newPods PodSet) (add, remove PodSet) {
+func diffAddresses(oldAddresses, newAddresses AddressSet) (add, remove AddressSet) {
 	// TODO: this detects pods which have been added or removed, but does not
-	// detect pods which have been modified.  A modified pod should trigger
+	// detect addresses which have been modified.  A modified address should trigger
 	// an add of the new version.
-	addPods := make(map[PodID]Address)
-	removePods := make(map[PodID]Address)
-	for id, pod := range newPods.Pods {
-		if _, ok := oldPods.Pods[id]; !ok {
-			addPods[id] = pod
+	addAddesses := make(map[ID]Address)
+	removeAddresses := make(map[ID]Address)
+	for id, address := range newAddresses.Addresses {
+		if _, ok := oldAddresses.Addresses[id]; !ok {
+			addAddesses[id] = address
 		}
 	}
-	for id, pod := range oldPods.Pods {
-		if _, ok := newPods.Pods[id]; !ok {
-			removePods[id] = pod
+	for id, address := range oldAddresses.Addresses {
+		if _, ok := newAddresses.Addresses[id]; !ok {
+			removeAddresses[id] = address
 		}
 	}
-	add = PodSet{
-		Pods:   addPods,
-		Labels: newPods.Labels,
+	add = AddressSet{
+		Addresses: addAddesses,
+		Labels:    newAddresses.Labels,
 	}
-	remove = PodSet{
-		Pods: removePods,
+	remove = AddressSet{
+		Addresses: removeAddresses,
 	}
 	return
 }
diff --git a/controller/api/destination/watcher/ip_watcher.go b/controller/api/destination/watcher/ip_watcher.go
index a38aa1330..b0a45d017 100644
--- a/controller/api/destination/watcher/ip_watcher.go
+++ b/controller/api/destination/watcher/ip_watcher.go
@@ -29,7 +29,7 @@ type (
 
 		// At most one of service or pod may be non-zero.
 		service ServiceID
-		pod     PodSet
+		pod     AddressSet
 
 		listeners map[EndpointUpdateListener]Port
 		endpoints *EndpointsWatcher
@@ -150,7 +150,7 @@ func (iw *IPWatcher) addPod(obj interface{}) {
 		return
 	}
 	ss := iw.getOrNewServiceSubscriptions(pod.Status.PodIP)
-	ss.updatePod(iw.podToPodSet(pod))
+	ss.updatePod(iw.podToAddressSet(pod))
 }
 
 func (iw *IPWatcher) deletePod(obj interface{}) {
@@ -228,7 +228,7 @@ func (iw *IPWatcher) getOrNewServiceSubscriptions(clusterIP string) *serviceSubs
 				iw.log.Errorf("Pod IP conflict: %v, %v", objs[0], objs[1])
 			}
 			if len(pods) == 1 {
-				ss.pod = iw.podToPodSet(pods[0])
+				ss.pod = iw.podToAddressSet(pods[0])
 			}
 		}
 
@@ -244,14 +244,14 @@ func (iw *IPWatcher) getServiceSubscriptions(clusterIP string) (ss *serviceSubsc
 	return
 }
 
-func (iw *IPWatcher) podToPodSet(pod *corev1.Pod) PodSet {
+func (iw *IPWatcher) podToAddressSet(pod *corev1.Pod) AddressSet {
 	if pod.Spec.HostNetwork {
-		return PodSet{}
+		return AddressSet{}
 	}
 
 	ownerKind, ownerName := iw.k8sAPI.GetOwnerKindAndName(pod, true)
-	return PodSet{
-		Pods: map[PodID]Address{
+	return AddressSet{
+		Addresses: map[PodID]Address{
 			PodID{
 				Name:      pod.Name,
 				Namespace: pod.Namespace,
@@ -292,7 +292,7 @@ func (ss *serviceSubscriptions) updateService(service *corev1.Service) {
 			}
 		}
 		ss.service = id
-		ss.pod = PodSet{}
+		ss.pod = AddressSet{}
 	}
 }
 
@@ -309,13 +309,13 @@ func (ss *serviceSubscriptions) deleteService() {
 	ss.service = ServiceID{}
 }
 
-func (ss *serviceSubscriptions) updatePod(podSet PodSet) {
+func (ss *serviceSubscriptions) updatePod(podSet AddressSet) {
 	ss.Lock()
 	defer ss.Unlock()
 
 	for listener, port := range ss.listeners {
 		listener.NoEndpoints(true) // Clear out previous endpoints.
-		if len(podSet.Pods) != 0 {
+		if len(podSet.Addresses) != 0 {
 			podSetWithPort := withPort(podSet, port)
 			listener.Add(podSetWithPort)
 		}
@@ -330,7 +330,7 @@ func (ss *serviceSubscriptions) deletePod() {
 		listener.NoEndpoints(true) // Clear out previous endpoints.
 		listener.Add(singletonAddress(ss.clusterIP, port))
 	}
-	ss.pod = PodSet{}
+	ss.pod = AddressSet{}
 }
 
 func (ss *serviceSubscriptions) subscribe(port Port, listener EndpointUpdateListener) error {
@@ -342,7 +342,7 @@ func (ss *serviceSubscriptions) subscribe(port Port, listener EndpointUpdateList
 		if err != nil {
 			return err
 		}
-	} else if len(ss.pod.Pods) != 0 {
+	} else if len(ss.pod.Addresses) != 0 {
 		podSetWithPort := withPort(ss.pod, port)
 		listener.Add(podSetWithPort)
 	} else {
@@ -362,21 +362,21 @@ func (ss *serviceSubscriptions) unsubscribe(port Port, listener EndpointUpdateLi
 	delete(ss.listeners, listener)
 }
 
-func withPort(pods PodSet, port Port) PodSet {
-	wp := PodSet{
-		Pods:   map[PodID]Address{},
-		Labels: pods.Labels,
+func withPort(pods AddressSet, port Port) AddressSet {
+	wp := AddressSet{
+		Addresses: map[PodID]Address{},
+		Labels:    pods.Labels,
 	}
-	for id, pod := range pods.Pods {
-		pod.Port = port
-		wp.Pods[id] = pod
+	for id, addr := range pods.Addresses {
+		addr.Port = port
+		wp.Addresses[id] = addr
 	}
 	return wp
 }
 
-func singletonAddress(ip string, port Port) PodSet {
-	return PodSet{
-		Pods: map[PodID]Address{
+func singletonAddress(ip string, port Port) AddressSet {
+	return AddressSet{
+		Addresses: map[PodID]Address{
 			PodID{}: Address{
 				IP:   ip,
 				Port: port,
diff --git a/pkg/k8s/labels.go b/pkg/k8s/labels.go
index fbff494de..4a59499bb 100644
--- a/pkg/k8s/labels.go
+++ b/pkg/k8s/labels.go
@@ -374,6 +374,9 @@ const (
 	// in cases we detect a change in a remote gateway
 	RemoteGatewayResourceVersionAnnotation = SvcMirrorPrefix + "/remote-gateway-resource-version"
 
+	// RemoteGatewayIdentity follows the same kind of logic as RemoteGatewayNameLabel
+	RemoteGatewayIdentity = SvcMirrorPrefix + "/remote-gateway-identity"
+
 	// ConfigKeyName is the key in the secret that stores the kubeconfig needed to connect
 	// to a remote cluster
 	ConfigKeyName = "kubeconfig"
diff --git a/pkg/smimetrics/client.go b/pkg/smimetrics/client.go
index 019129fa0..841744288 100644
--- a/pkg/smimetrics/client.go
+++ b/pkg/smimetrics/client.go
@@ -2,6 +2,7 @@ package smimetrics
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"net/http"
@@ -47,6 +48,14 @@ type (
 		Unit  string `json:"unit"`
 		Value string `json:"value"`
 	}
+
+	// Status is an API response that is returned when the operation is not
+	// successful.
+	Status struct {
+		Kind    string `json:"kind"`
+		Status  string `json:"status"`
+		Message string `json:"message"`
+	}
 )
 
 const apiBase = "/apis/metrics.smi-spec.io/v1alpha1"
@@ -142,24 +151,47 @@ func getMetricsResponse(k8sAPI *k8s.KubernetesAPI, path string, params map[strin
 	return ioutil.ReadAll(httpRsp.Body)
 }
 
+func handleStatusResponse(bytes []byte) error {
+	var status Status
+	json.Unmarshal(bytes, &status)
+	if status.Kind == "Status" {
+		return errors.New(status.Message)
+	}
+	return nil
+}
+
 func parseTrafficMetrics(bytes []byte) (*TrafficMetrics, error) {
+	err := handleStatusResponse(bytes)
+	if err != nil {
+		return nil, err
+	}
+
 	var metrics TrafficMetrics
-	err := json.Unmarshal(bytes, &metrics)
+	err = json.Unmarshal(bytes, &metrics)
 	if err != nil {
 		log.Errorf("Failed to decode response as TrafficMetrics [%s]: %v", string(bytes), err)
-		return nil, err
+		return nil, errors.New(string(bytes))
 	}
 
+	log.Debugf("Parsed TrafficMetrics: %+v", metrics)
+
 	return &metrics, nil
 }
 
 func parseTrafficMetricsList(bytes []byte) (*TrafficMetricsList, error) {
+	err := handleStatusResponse(bytes)
+	if err != nil {
+		return nil, err
+	}
+
 	var metrics TrafficMetricsList
-	err := json.Unmarshal(bytes, &metrics)
+	err = json.Unmarshal(bytes, &metrics)
 	if err != nil {
 		log.Errorf("Failed to decode response as TrafficMetricsList [%s]: %v", string(bytes), err)
-		return nil, err
+		return nil, errors.New(string(bytes))
 	}
 
+	log.Debugf("Parsed TrafficMetricsList: %+v", metrics)
+
 	return &metrics, nil
 }


Test patch:
diff --git a/controller/api/destination/endpoint_translator_test.go b/controller/api/destination/endpoint_translator_test.go
index 7c57b1e45..3f5e22b38 100644
--- a/controller/api/destination/endpoint_translator_test.go
+++ b/controller/api/destination/endpoint_translator_test.go
@@ -1,8 +1,10 @@
 package destination
 
 import (
+	"fmt"
 	"reflect"
 	"sort"
+	"strings"
 	"testing"
 
 	pb "github.com/linkerd/linkerd2-proxy-api/go/destination"
@@ -92,6 +94,17 @@ var (
 			},
 		},
 	}
+
+	remoteGatewayWithNoTLS = watcher.Address{
+		IP:   "1.1.1.1",
+		Port: 1,
+	}
+
+	remoteGatewayWithTLS = watcher.Address{
+		IP:       "1.1.1.2",
+		Port:     2,
+		Identity: "some-identity",
+	}
 )
 
 func makeEndpointTranslator(t *testing.T) (*mockDestinationGetServer, *endpointTranslator) {
@@ -99,7 +112,7 @@ func makeEndpointTranslator(t *testing.T) (*mockDestinationGetServer, *endpointT
 	translator := newEndpointTranslator(
 		"linkerd",
 		"trust.domain",
-		false,
+		true,
 		"service-name.service-ns",
 		mockGetServer,
 		logging.WithField("test", t.Name()),
@@ -107,12 +120,77 @@ func makeEndpointTranslator(t *testing.T) (*mockDestinationGetServer, *endpointT
 	return mockGetServer, translator
 }
 
-func TestEndpointTranslator(t *testing.T) {
+func TestEndpointTranslatorForRemoteGateways(t *testing.T) {
+	t.Run("Sends one update for add and another for remove", func(t *testing.T) {
+		mockGetServer, translator := makeEndpointTranslator(t)
+
+		translator.Add(mkAddressSetForServices(remoteGatewayWithNoTLS))
+		translator.Remove(mkAddressSetForServices(remoteGatewayWithTLS))
+
+		expectedNumUpdates := 2
+		actualNumUpdates := len(mockGetServer.updatesReceived)
+		if actualNumUpdates != expectedNumUpdates {
+			t.Fatalf("Expecting [%d] updates, got [%d]. Updates: %v", expectedNumUpdates, actualNumUpdates, mockGetServer.updatesReceived)
+		}
+	})
+
+	t.Run("Sends TlsIdentity when enabled", func(t *testing.T) {
+		expectedTLSIdentity := &pb.TlsIdentity_DnsLikeIdentity{
+			Name: "some-identity",
+		}
+
+		expectedProtocolHint := &pb.ProtocolHint{
+			Protocol: &pb.ProtocolHint_H2_{
+				H2: &pb.ProtocolHint_H2{},
+			},
+		}
+
+		mockGetServer, translator := makeEndpointTranslator(t)
+
+		translator.Add(mkAddressSetForServices(remoteGatewayWithTLS))
+
+		addrs := mockGetServer.updatesReceived[0].GetAdd().GetAddrs()
+		if len(addrs) != 1 {
+			t.Fatalf("Expected [1] address returned, got %v", addrs)
+		}
+
+		actualTLSIdentity := addrs[0].GetTlsIdentity().GetDnsLikeIdentity()
+		if !reflect.DeepEqual(actualTLSIdentity, expectedTLSIdentity) {
+			t.Fatalf("Expected TlsIdentity to be [%v] but was [%v]", expectedTLSIdentity, actualTLSIdentity)
+		}
+
+		actualProtocolHint := addrs[0].GetProtocolHint()
+		if !reflect.DeepEqual(actualProtocolHint, expectedProtocolHint) {
+			t.Fatalf("Expected ProtoclHint to be [%v] but was [%v]", expectedProtocolHint, actualProtocolHint)
+		}
+	})
+
+	t.Run("Does not send TlsIdentity when not present", func(t *testing.T) {
+		mockGetServer, translator := makeEndpointTranslator(t)
+
+		translator.Add(mkAddressSetForServices(remoteGatewayWithNoTLS))
+
+		addrs := mockGetServer.updatesReceived[0].GetAdd().GetAddrs()
+		if len(addrs) != 1 {
+			t.Fatalf("Expected [1] address returned, got %v", addrs)
+		}
+
+		if addrs[0].TlsIdentity != nil {
+			t.Fatalf("Expected no TlsIdentity to be sent, but got [%v]", addrs[0].TlsIdentity)
+		}
+		if addrs[0].ProtocolHint != nil {
+			t.Fatalf("Expected no ProtocolHint to be sent, but got [%v]", addrs[0].TlsIdentity)
+		}
+	})
+
+}
+
+func TestEndpointTranslatorForPods(t *testing.T) {
 	t.Run("Sends one update for add and another for remove", func(t *testing.T) {
 		mockGetServer, translator := makeEndpointTranslator(t)
 
-		translator.Add(mkPodSet(normalPod))
-		translator.Remove(mkPodSet(tlsOptionalPod))
+		translator.Add(mkAddressSetForPods(normalPod))
+		translator.Remove(mkAddressSetForPods(tlsOptionalPod))
 
 		expectedNumUpdates := 2
 		actualNumUpdates := len(mockGetServer.updatesReceived)
@@ -124,8 +202,8 @@ func TestEndpointTranslator(t *testing.T) {
 	t.Run("Sends addresses as removed or added", func(t *testing.T) {
 		mockGetServer, translator := makeEndpointTranslator(t)
 
-		translator.Add(mkPodSet(normalPod, tlsOptionalPod))
-		translator.Remove(mkPodSet(tlsDisabledPod))
+		translator.Add(mkAddressSetForPods(normalPod, tlsOptionalPod))
+		translator.Remove(mkAddressSetForPods(tlsDisabledPod))
 
 		addressesAdded := mockGetServer.updatesReceived[0].GetAdd().Addrs
 		actualNumberOfAdded := len(addressesAdded)
@@ -152,7 +230,7 @@ func TestEndpointTranslator(t *testing.T) {
 	t.Run("Sends metric labels with added addresses", func(t *testing.T) {
 		mockGetServer, translator := makeEndpointTranslator(t)
 
-		translator.Add(mkPodSet(normalPod))
+		translator.Add(mkAddressSetForPods(normalPod))
 
 		actualGlobalMetricLabels := mockGetServer.updatesReceived[0].GetAdd().MetricLabels
 		expectedGlobalMetricLabels := map[string]string{"namespace": "service-ns", "service": "service-name"}
@@ -179,7 +257,7 @@ func TestEndpointTranslator(t *testing.T) {
 
 		mockGetServer, translator := makeEndpointTranslator(t)
 
-		translator.Add(mkPodSet(normalPod))
+		translator.Add(mkAddressSetForPods(normalPod))
 
 		addrs := mockGetServer.updatesReceived[0].GetAdd().GetAddrs()
 		if len(addrs) != 1 {
@@ -195,7 +273,7 @@ func TestEndpointTranslator(t *testing.T) {
 	t.Run("Does not send TlsIdentity for non-default identity-modes", func(t *testing.T) {
 		mockGetServer, translator := makeEndpointTranslator(t)
 
-		translator.Add(mkPodSet(tlsOptionalPod))
+		translator.Add(mkAddressSetForPods(tlsOptionalPod))
 
 		addrs := mockGetServer.updatesReceived[0].GetAdd().GetAddrs()
 		if len(addrs) != 1 {
@@ -210,7 +288,7 @@ func TestEndpointTranslator(t *testing.T) {
 	t.Run("Does not send TlsIdentity for other meshes", func(t *testing.T) {
 		mockGetServer, translator := makeEndpointTranslator(t)
 
-		translator.Add(mkPodSet(otherMeshPod))
+		translator.Add(mkAddressSetForPods(otherMeshPod))
 
 		addrs := mockGetServer.updatesReceived[0].GetAdd().GetAddrs()
 		if len(addrs) != 1 {
@@ -225,7 +303,7 @@ func TestEndpointTranslator(t *testing.T) {
 	t.Run("Does not send TlsIdentity when not enabled", func(t *testing.T) {
 		mockGetServer, translator := makeEndpointTranslator(t)
 
-		translator.Add(mkPodSet(tlsDisabledPod))
+		translator.Add(mkAddressSetForPods(tlsDisabledPod))
 
 		addrs := mockGetServer.updatesReceived[0].GetAdd().GetAddrs()
 		if len(addrs) != 1 {
@@ -238,14 +316,33 @@ func TestEndpointTranslator(t *testing.T) {
 	})
 }
 
-func mkPodSet(pods ...watcher.Address) watcher.PodSet {
-	set := watcher.PodSet{
-		Pods:   make(map[watcher.PodID]watcher.Address),
-		Labels: map[string]string{"service": "service-name", "namespace": "service-ns"},
+func mkAddressSetForServices(gatewayAddresses ...watcher.Address) watcher.AddressSet {
+	set := watcher.AddressSet{
+		Addresses: make(map[watcher.ServiceID]watcher.Address),
+		Labels:    map[string]string{"service": "service-name", "namespace": "service-ns"},
+	}
+	for _, a := range gatewayAddresses {
+		a := a // pin
+
+		id := watcher.ServiceID{
+			Name: strings.Join([]string{
+				a.IP,
+				fmt.Sprint(a.Port),
+			}, "-"),
+		}
+		set.Addresses[id] = a
+	}
+	return set
+}
+
+func mkAddressSetForPods(podAddresses ...watcher.Address) watcher.AddressSet {
+	set := watcher.AddressSet{
+		Addresses: make(map[watcher.PodID]watcher.Address),
+		Labels:    map[string]string{"service": "service-name", "namespace": "service-ns"},
 	}
-	for _, p := range pods {
+	for _, p := range podAddresses {
 		id := watcher.PodID{Name: p.Pod.Name, Namespace: p.Pod.Namespace}
-		set.Pods[id] = p
+		set.Addresses[id] = p
 	}
 	return set
 }
diff --git a/controller/api/destination/watcher/endpoints_watcher_test.go b/controller/api/destination/watcher/endpoints_watcher_test.go
index 6798ca2ca..07d566848 100644
--- a/controller/api/destination/watcher/endpoints_watcher_test.go
+++ b/controller/api/destination/watcher/endpoints_watcher_test.go
@@ -27,17 +27,21 @@ func newBufferingEndpointListener() *bufferingEndpointListener {
 }
 
 func addressString(address Address) string {
-	return fmt.Sprintf("%s:%d", address.IP, address.Port)
+	addressString := fmt.Sprintf("%s:%d", address.IP, address.Port)
+	if address.Identity != "" {
+		addressString = fmt.Sprintf("%s/%s", addressString, address.Identity)
+	}
+	return addressString
 }
 
-func (bel *bufferingEndpointListener) Add(set PodSet) {
-	for _, address := range set.Pods {
+func (bel *bufferingEndpointListener) Add(set AddressSet) {
+	for _, address := range set.Addresses {
 		bel.added = append(bel.added, addressString(address))
 	}
 }
 
-func (bel *bufferingEndpointListener) Remove(set PodSet) {
-	for _, address := range set.Pods {
+func (bel *bufferingEndpointListener) Remove(set AddressSet) {
+	for _, address := range set.Addresses {
 		bel.removed = append(bel.removed, addressString(address))
 	}
 }
@@ -271,7 +275,7 @@ status:
 			expectedError:                    false,
 		},
 		{
-			serviceType: "local services with missing pods",
+			serviceType: "local services with missing addresses",
 			k8sConfigs: []string{`
 apiVersion: v1
 kind: Service
@@ -615,5 +619,165 @@ status:
 		})
 
 	}
+}
+
+func TestEndpointsWatcherServiceMirrors(t *testing.T) {
+	for _, tt := range []struct {
+		serviceType                      string
+		k8sConfigs                       []string
+		id                               ServiceID
+		hostname                         string
+		port                             Port
+		expectedAddresses                []string
+		expectedNoEndpoints              bool
+		expectedNoEndpointsServiceExists bool
+	}{
+		{
+			k8sConfigs: []string{`
+apiVersion: v1
+kind: Service
+metadata:
+  name: name1-remote
+  namespace: ns
+spec:
+  type: LoadBalancer
+  ports:
+  - port: 8989`,
+				`
+apiVersion: v1
+kind: Endpoints
+metadata:
+  name: name1-remote
+  namespace: ns
+  annotations:
+    mirror.linkerd.io/remote-gateway-identity: "gateway-identity-1"
+    mirror.linkerd.io/remote-svc-fq-name: "name1-remote-fq"
+  labels:
+    mirror.linkerd.io/mirrored-service: "true"
+subsets:
+- addresses:
+  - ip: 172.17.0.12
+  ports:
+  - port: 8989`,
+			},
+			serviceType: "mirrored service with identity",
+			id:          ServiceID{Name: "name1-remote", Namespace: "ns"},
+			port:        8989,
+			expectedAddresses: []string{
+				"172.17.0.12:8989/gateway-identity-1",
+			},
+			expectedNoEndpoints:              false,
+			expectedNoEndpointsServiceExists: false,
+		},
+		{
+			k8sConfigs: []string{`
+apiVersion: v1
+kind: Service
+metadata:
+  name: name1-remote
+  namespace: ns
+spec:
+  type: LoadBalancer
+  ports:
+  - port: 8989`,
+				`
+apiVersion: v1
+kind: Endpoints
+metadata:
+  name: name1-remote
+  namespace: ns
+  annotations:
+    mirror.linkerd.io/remote-svc-fq-name: "name1-remote-fq"
+  labels:
+    mirror.linkerd.io/mirrored-service: "true"
+subsets:
+- addresses:
+  - ip: 172.17.0.12
+  ports:
+  - port: 8989`,
+			},
+			serviceType: "mirrored service without identity",
+			id:          ServiceID{Name: "name1-remote", Namespace: "ns"},
+			port:        8989,
+			expectedAddresses: []string{
+				"172.17.0.12:8989",
+			},
+			expectedNoEndpoints:              false,
+			expectedNoEndpointsServiceExists: false,
+		},
+
+		{
+			k8sConfigs: []string{`
+apiVersion: v1
+kind: Service
+metadata:
+  name: name1-remote
+  namespace: ns
+spec:
+  type: LoadBalancer
+  ports:
+  - port: 8989`,
+				`
+apiVersion: v1
+kind: Endpoints
+metadata:
+  name: name1-remote
+  namespace: ns
+  annotations:
+    mirror.linkerd.io/remote-gateway-identity: "gateway-identity-1"
+    mirror.linkerd.io/remote-svc-fq-name: "name1-remote-fq"
+  labels:
+    mirror.linkerd.io/mirrored-service: "true"
+subsets:
+- addresses:
+  - ip: 172.17.0.12
+  ports:
+  - port: 9999`,
+			},
+			serviceType: "mirrored service with remapped port in endpoints",
+			id:          ServiceID{Name: "name1-remote", Namespace: "ns"},
+			port:        9999,
+			expectedAddresses: []string{
+				"172.17.0.12:9999/gateway-identity-1",
+			},
+			expectedNoEndpoints:              false,
+			expectedNoEndpointsServiceExists: false,
+		},
+	} {
+		tt := tt // pin
+		t.Run("subscribes listener to "+tt.serviceType, func(t *testing.T) {
+			k8sAPI, err := k8s.NewFakeAPI(tt.k8sConfigs...)
+			if err != nil {
+				t.Fatalf("NewFakeAPI returned an error: %s", err)
+			}
+
+			watcher := NewEndpointsWatcher(k8sAPI, logging.WithField("test", t.Name()))
+
+			k8sAPI.Sync(nil)
+
+			listener := newBufferingEndpointListener()
+
+			err = watcher.Subscribe(tt.id, tt.port, tt.hostname, listener)
+
+			if err != nil {
+				t.Fatalf("NewFakeAPI returned an error: %s", err)
+			}
+
+			actualAddresses := make([]string, 0)
+			actualAddresses = append(actualAddresses, listener.added...)
+			sort.Strings(actualAddresses)
+
+			testCompare(t, tt.expectedAddresses, actualAddresses)
+
+			if listener.noEndpointsCalled != tt.expectedNoEndpoints {
+				t.Fatalf("Expected noEndpointsCalled to be [%t], got [%t]",
+					tt.expectedNoEndpoints, listener.noEndpointsCalled)
+			}
 
+			if listener.noEndpointsExists != tt.expectedNoEndpointsServiceExists {
+				t.Fatalf("Expected noEndpointsExists to be [%t], got [%t]",
+					tt.expectedNoEndpointsServiceExists, listener.noEndpointsExists)
+			}
+		})
+	}
 }
diff --git a/controller/api/destination/watcher/ip_watcher_test.go b/controller/api/destination/watcher/ip_watcher_test.go
index 2e4218088..17207a2b9 100644
--- a/controller/api/destination/watcher/ip_watcher_test.go
+++ b/controller/api/destination/watcher/ip_watcher_test.go
@@ -531,7 +531,7 @@ status:
 		deletingPod    bool
 	}{
 		{
-			description:    "can delete pods",
+			description:    "can delete addresses",
 			k8sConfigs:     podK8sConfig,
 			host:           "172.17.0.12",
 			port:           8989,
@@ -539,7 +539,7 @@ status:
 			deletingPod:    true,
 		},
 		{
-			description:    "can delete pods wrapped in a DeletedFinalStateUnknown",
+			description:    "can delete addresses wrapped in a DeletedFinalStateUnknown",
 			k8sConfigs:     podK8sConfig,
 			host:           "172.17.0.12",
 			port:           8989,