Repository: linkerd/linkerd2
Pull request #7388 (2021-12-01 23:28:22)

Update `GetProfile` clients with policy server updates
### What

`GetProfile` clients do not receive destinatin profiles that consider Server protocol fields the way that `Get` clients do. If a Server exists for a `GetProfile` destination that specifies the protocol for that destination is `opaque`, this information is not passed back to the client.

#7184 added this for `Get` by subscribing clients to Endpoint/EndpointSlice updates. When there is an update, or there is a Server update, the endpoints watcher passes this information back to the endpoint translator which handles sending the update back to the client.

For `GetProfile` the situation is different. As with `Get`, we only consider Servers when dealing with Pod IPs, but this only occurs in two situations for `GetProfile`.

1. The destination is a Pod IP and port
2. The destionation is an Instance ID and port

In both of these cases, we need to check if a already Server selects the endpoint and we need to subscribe for Server updates incase one is added or deleted which selects the endpoint.

### How

First we check if there is already a Server which selects the endpoint. This is so that when the first destionation profile is returned, the client knows if the destination is `opaque` or not.

After sending that first update, we then subscribe the client for any future updates which will come from a Server being added or deleted.

This is handled by the new `ServerWatcher` which watches for Server updates on the cluster; when an update occurs it sends that to the `endpointProfileTranslator` which translates the protcol update into a DestinationProfile.

By introducing the `endpointProfileTranslator` which only handles protocol updates, we're able to decouple the endpoint logic from `profileTranslator`—it's `endpoint` field has been removed now that it only handles updates for ServiceProfiles for Services.

### Testing

A unit test has been added and below are some manual testing instructions to see how it interacts with Server updates:

<details>
	<summary>app.yaml</summary>

	```yaml
	apiVersion: v1
	kind: Pod
	metadata:
	  name: pod
	  labels:
		app: pod
	spec:
	  containers:
	  - name: app
		image: nginx
		ports:
		  - name: http
			containerPort: 80
	---
	apiVersion: policy.linkerd.io/v1beta1
	kind: Server
	metadata:
	  name: srv
	  labels:
		policy: srv
	spec:
	  podSelector:
		matchLabels:
		  app: pod
	  port: 80
	  proxyProtocol: opaque
	```
</details>

```shell
$ go run ./controller/cmd/main.go destination
```

```shell
$ linkerd inject app.yaml |kubectl apply -f -
...
$ kubectl get pods -o wide
NAME   READY   STATUS    RESTARTS   AGE   IP           NODE                       NOMINATED NODE   READINESS GATES
pod    2/2     Running   0          53m   10.42.0.34   k3d-k3s-default-server-0   <none>           <none>
$ go run ./controller/script/destination-client/main.go -method getProfile -path 10.42.0.34:80
...
```

You can add/delete `srv` as well as edit its `proxyProtocol` field to observe the correct DestinationProfile updates.

Signed-off-by: Kevin Leimkuhler <kleimkuhler@icloud.com>

kleimkuhler: @adleong Thanks for the review! By introducing the `endpointProfileTranslator`, we're able to handle protocol updates for endpoints and decouple the endpoint logic from `profileTranslator`.

This also allows us to remove the `endpoint` field from `profileTranslator` now that it is no longer used.

Patch:
diff --git a/.github/workflows/policy_controller.yml b/.github/workflows/policy_controller.yml
index cbad46765..a6fb0ca09 100644
--- a/.github/workflows/policy_controller.yml
+++ b/.github/workflows/policy_controller.yml
@@ -42,7 +42,7 @@ jobs:
     continue-on-error: ${{ matrix.checks == 'advisories' }}
     steps:
     - uses: actions/checkout@ec3a7ce113134d7a93b817d10a8272cb61118579
-    - uses: EmbarkStudios/cargo-deny-action@0ca727bbae7b7b578b9a5f98186caac35aa2a00d
+    - uses: EmbarkStudios/cargo-deny-action@f2d2f98857d524436b31aa639bac5edc10863b08
       with:
         command: check bans licenses sources
 
diff --git a/controller/api/destination/endpoint_profile_translator.go b/controller/api/destination/endpoint_profile_translator.go
new file mode 100644
index 000000000..cbebcf2cb
--- /dev/null
+++ b/controller/api/destination/endpoint_profile_translator.go
@@ -0,0 +1,63 @@
+package destination
+
+import (
+	pb "github.com/linkerd/linkerd2-proxy-api/go/destination"
+	"github.com/sirupsen/logrus"
+	v1 "k8s.io/api/core/v1"
+)
+
+type endpointProfileTranslator struct {
+	pod      *v1.Pod
+	port     uint32
+	endpoint *pb.WeightedAddr
+	stream   pb.Destination_GetProfileServer
+	log      *logrus.Entry
+}
+
+// newEndpointProfileTranslator translates protocol updates to
+// DestinationProfiles for endpoints. When a Server on the cluster is updated
+// it is possible that it selects an endpoint that is being watched, if that
+// is the case then an update will be sent to the client if the Server has
+// changed the endpoint's supported protocol—mainly being opaque or not.
+func newEndpointProfileTranslator(pod *v1.Pod, port uint32, endpoint *pb.WeightedAddr, stream pb.Destination_GetProfileServer, log *logrus.Entry) *endpointProfileTranslator {
+	return &endpointProfileTranslator{
+		pod:      pod,
+		port:     port,
+		endpoint: endpoint,
+		stream:   stream,
+		log:      log,
+	}
+}
+
+func (ept *endpointProfileTranslator) UpdateProtocol(opaqueProtocol bool) {
+	// The protocol for an endpoint should only be updated if there is a pod,
+	// endpoint, and the endpoint has a protocol hint. If there is an endpoint
+	// but it does not have a protocol hint, that means we could not determine
+	// if it has a peer proxy so a opaque traffic would not be supported.
+	if ept.pod != nil && ept.endpoint != nil && ept.endpoint.ProtocolHint != nil {
+		if !opaqueProtocol {
+			ept.endpoint.ProtocolHint.OpaqueTransport = nil
+		} else if ept.endpoint.ProtocolHint.OpaqueTransport == nil {
+			port, err := getInboundPort(&ept.pod.Spec)
+			if err != nil {
+				ept.log.Error(err)
+			} else {
+				ept.endpoint.ProtocolHint.OpaqueTransport = &pb.ProtocolHint_OpaqueTransport{
+					InboundPort: port,
+				}
+			}
+		}
+
+	}
+	profile := ept.createDefaultProfile(opaqueProtocol)
+	ept.log.Debugf("sending protocol update: %+v", profile)
+	ept.stream.Send(profile)
+}
+
+func (ept *endpointProfileTranslator) createDefaultProfile(opaqueProtocol bool) *pb.DestinationProfile {
+	return &pb.DestinationProfile{
+		RetryBudget:    defaultRetryBudget(),
+		Endpoint:       ept.endpoint,
+		OpaqueProtocol: opaqueProtocol,
+	}
+}
diff --git a/controller/api/destination/endpoint_translator.go b/controller/api/destination/endpoint_translator.go
index 40e1f69c7..b62a76662 100644
--- a/controller/api/destination/endpoint_translator.go
+++ b/controller/api/destination/endpoint_translator.go
@@ -220,11 +220,11 @@ func (et *endpointTranslator) sendClientAdd(set watcher.AddressSet) {
 			err         error
 		)
 		if address.Pod != nil {
-			opaquePorts, err = getPodOpaquePorts(address.Pod, et.defaultOpaquePorts)
+			opaquePorts, err = getAnnotatedOpaquePorts(address.Pod, et.defaultOpaquePorts)
 			if err != nil {
 				et.log.Errorf("failed to get opaque ports for pod %s/%s: %s", address.Pod.Namespace, address.Pod.Name, err)
 			}
-			wa, err = toWeightedAddr(address, opaquePorts, et.enableH2Upgrade, et.identityTrustDomain, et.controllerNS, et.log)
+			wa, err = createWeightedAddr(address, opaquePorts, et.enableH2Upgrade, et.identityTrustDomain, et.controllerNS, et.log)
 		} else {
 			var authOverride *pb.AuthorityOverride
 			if address.AuthorityOverride != "" {
@@ -314,7 +314,7 @@ func toAddr(address watcher.Address) (*net.TcpAddress, error) {
 	}, nil
 }
 
-func toWeightedAddr(address watcher.Address, opaquePorts map[uint32]struct{}, enableH2Upgrade bool, identityTrustDomain string, controllerNS string, log *logging.Entry) (*pb.WeightedAddr, error) {
+func createWeightedAddr(address watcher.Address, opaquePorts map[uint32]struct{}, enableH2Upgrade bool, identityTrustDomain string, controllerNS string, log *logging.Entry) (*pb.WeightedAddr, error) {
 	// When converting an address to a weighted addr, it should be backed by a Pod.
 	if address.Pod == nil {
 		return nil, fmt.Errorf("endpoint not backed by Pod: %s:%d", address.IP, address.Port)
diff --git a/controller/api/destination/profile_translator.go b/controller/api/destination/profile_translator.go
index 42422d6a9..80e110509 100644
--- a/controller/api/destination/profile_translator.go
+++ b/controller/api/destination/profile_translator.go
@@ -22,16 +22,14 @@ type profileTranslator struct {
 	log                *logging.Entry
 	fullyQualifiedName string
 	port               uint32
-	endpoint           *pb.WeightedAddr
 }
 
-func newProfileTranslator(stream pb.Destination_GetProfileServer, log *logging.Entry, fqn string, port uint32, endpoint *pb.WeightedAddr) *profileTranslator {
+func newProfileTranslator(stream pb.Destination_GetProfileServer, log *logging.Entry, fqn string, port uint32) *profileTranslator {
 	return &profileTranslator{
 		stream:             stream,
 		log:                log.WithField("component", "profile-translator"),
 		fullyQualifiedName: fqn,
 		port:               port,
-		endpoint:           endpoint,
 	}
 }
 
@@ -40,7 +38,7 @@ func (pt *profileTranslator) Update(profile *sp.ServiceProfile) {
 		pt.stream.Send(pt.defaultServiceProfile())
 		return
 	}
-	destinationProfile, err := pt.toServiceProfile(profile)
+	destinationProfile, err := pt.createDestinationProfile(profile)
 	if err != nil {
 		pt.log.Error(err)
 		return
@@ -54,7 +52,6 @@ func (pt *profileTranslator) defaultServiceProfile() *pb.DestinationProfile {
 		Routes:             []*pb.Route{},
 		RetryBudget:        defaultRetryBudget(),
 		FullyQualifiedName: pt.fullyQualifiedName,
-		Endpoint:           pt.endpoint,
 	}
 }
 
@@ -78,9 +75,9 @@ func toDuration(d time.Duration) *duration.Duration {
 	}
 }
 
-// toServiceProfile returns a Proxy API DestinationProfile, given a
+// createDestinationProfile returns a Proxy API DestinationProfile, given a
 // ServiceProfile.
-func (pt *profileTranslator) toServiceProfile(profile *sp.ServiceProfile) (*pb.DestinationProfile, error) {
+func (pt *profileTranslator) createDestinationProfile(profile *sp.ServiceProfile) (*pb.DestinationProfile, error) {
 	routes := make([]*pb.Route, 0)
 	for _, route := range profile.Spec.Routes {
 		pbRoute, err := toRoute(profile, route)
@@ -108,7 +105,6 @@ func (pt *profileTranslator) toServiceProfile(profile *sp.ServiceProfile) (*pb.D
 		RetryBudget:        budget,
 		DstOverrides:       toDstOverrides(profile.Spec.DstOverrides, pt.port),
 		FullyQualifiedName: pt.fullyQualifiedName,
-		Endpoint:           pt.endpoint,
 		OpaqueProtocol:     opaqueProtocol,
 	}, nil
 }
@@ -256,7 +252,7 @@ func toResponseMatch(rspMatch *sp.ResponseMatch) (*pb.ResponseMatch, error) {
 	}
 
 	if len(matches) == 0 {
-		return nil, errors.New("A response match must have a field set")
+		return nil, errors.New("a response match must have a field set")
 	}
 	if len(matches) == 1 {
 		return matches[0], nil
@@ -350,7 +346,7 @@ func toRequestMatch(reqMatch *sp.RequestMatch) (*pb.RequestMatch, error) {
 	}
 
 	if len(matches) == 0 {
-		return nil, errors.New("A request match must have a field set")
+		return nil, errors.New("a request match must have a field set")
 	}
 	if len(matches) == 1 {
 		return matches[0], nil
diff --git a/controller/api/destination/server.go b/controller/api/destination/server.go
index 35d4a5e6c..0cc61b62d 100644
--- a/controller/api/destination/server.go
+++ b/controller/api/destination/server.go
@@ -10,7 +10,6 @@ import (
 
 	pb "github.com/linkerd/linkerd2-proxy-api/go/destination"
 	"github.com/linkerd/linkerd2/controller/api/destination/watcher"
-	sp "github.com/linkerd/linkerd2/controller/gen/apis/serviceprofile/v1alpha2"
 	"github.com/linkerd/linkerd2/controller/k8s"
 	labels "github.com/linkerd/linkerd2/pkg/k8s"
 	"github.com/linkerd/linkerd2/pkg/prometheus"
@@ -31,6 +30,7 @@ type (
 		endpoints   *watcher.EndpointsWatcher
 		opaquePorts *watcher.OpaquePortsWatcher
 		profiles    *watcher.ProfileWatcher
+		servers     *watcher.ServerWatcher
 		nodes       coreinformers.NodeInformer
 
 		enableH2Upgrade     bool
@@ -82,12 +82,14 @@ func NewServer(
 	endpoints := watcher.NewEndpointsWatcher(k8sAPI, log, enableEndpointSlices)
 	opaquePorts := watcher.NewOpaquePortsWatcher(k8sAPI, log, defaultOpaquePorts)
 	profiles := watcher.NewProfileWatcher(k8sAPI, log)
+	servers := watcher.NewServerWatcher(k8sAPI, log)
 
 	srv := server{
 		pb.UnimplementedDestinationServer{},
 		endpoints,
 		opaquePorts,
 		profiles,
+		servers,
 		k8sAPI.Node(),
 		enableH2Upgrade,
 		controllerNS,
@@ -207,14 +209,35 @@ func (s *server) GetProfile(dest *pb.GetDestination, stream pb.Destination_GetPr
 				return err
 			}
 
-			// The IP may or may not map to a pod (pod argument can be nil). If
-			// pod is not nil we will return a single endpoint in the
-			// DestinationProfile response, otherwise we return a default
-			// profile response.
-			err = s.sendEndpointProfile(stream, pod, port)
+			opaquePorts, err := getAnnotatedOpaquePorts(pod, s.defaultOpaquePorts)
 			if err != nil {
-				log.Debugf("Failed to send profile response for endpoint %s:%d: %v", ip.String(), port, err)
-				return err
+				return fmt.Errorf("failed to get opaque ports for pod: %s", err)
+			}
+			var address watcher.Address
+			var endpoint *pb.WeightedAddr
+			if pod != nil {
+				address, err = s.createAddress(pod, port)
+				if err != nil {
+					return fmt.Errorf("failed to create address: %s", err)
+				}
+				endpoint, err = s.createEndpoint(address, opaquePorts)
+				if err != nil {
+					return fmt.Errorf("failed to create endpoint: %s", err)
+				}
+			}
+			translator := newEndpointProfileTranslator(pod, port, endpoint, stream, s.log)
+
+			// If the endpoint's port is annotated as opaque, we don't need to
+			// subscribe for updates because it will always be opaque
+			// regardless of any Servers that may select it.
+			if _, ok := opaquePorts[port]; ok {
+				translator.UpdateProtocol(true)
+			} else if pod == nil {
+				translator.UpdateProtocol(false)
+			} else {
+				translator.UpdateProtocol(address.OpaqueProtocol)
+				s.servers.Subscribe(pod, port, translator)
+				defer s.servers.Unsubscribe(pod, port, translator)
 			}
 
 			select {
@@ -222,7 +245,6 @@ func (s *server) GetProfile(dest *pb.GetDestination, stream pb.Destination_GetPr
 			case <-stream.Context().Done():
 				log.Debugf("GetProfile(%+v) cancelled", dest)
 			}
-
 			return nil
 		}
 	} else {
@@ -239,21 +261,43 @@ func (s *server) GetProfile(dest *pb.GetDestination, stream pb.Destination_GetPr
 		if hostname != "" {
 			pod, err := getPodByHostname(s.k8sAPI, hostname, service)
 			if err != nil {
-				log.Errorf("Failed to get pod for hostname %s: %v", hostname, err)
+				log.Errorf("failed to get pod for hostname %s: %v", hostname, err)
 			}
-
-			err = s.sendEndpointProfile(stream, pod, port)
+			opaquePorts, err := getAnnotatedOpaquePorts(pod, s.defaultOpaquePorts)
 			if err != nil {
-				log.Debugf("Failed to send profile response for host %s: %v", hostname, err)
-				return err
+				return fmt.Errorf("failed to get opaque ports for pod: %s", err)
+			}
+			var address watcher.Address
+			var endpoint *pb.WeightedAddr
+			if pod != nil {
+				address, err = s.createAddress(pod, port)
+				if err != nil {
+					return fmt.Errorf("failed to create address: %s", err)
+				}
+				endpoint, err = s.createEndpoint(address, opaquePorts)
+				if err != nil {
+					return fmt.Errorf("failed to create endpoint: %s", err)
+				}
+			}
+			translator := newEndpointProfileTranslator(pod, port, endpoint, stream, s.log)
+
+			// If the endpoint's port is annotated as opaque, we don't need to
+			// subscribe for updates because it will always be opaque
+			// regardless of any Servers that may select it.
+			if _, ok := opaquePorts[port]; ok {
+				translator.UpdateProtocol(true)
+			} else if pod == nil {
+				translator.UpdateProtocol(false)
+			} else {
+				translator.UpdateProtocol(address.OpaqueProtocol)
+				s.servers.Subscribe(pod, port, translator)
+				defer s.servers.Unsubscribe(pod, port, translator)
 			}
-
 			select {
 			case <-s.shutdown:
 			case <-stream.Context().Done():
 				log.Debugf("GetProfile(%+v) cancelled", dest)
 			}
-
 			return nil
 		}
 
@@ -263,7 +307,7 @@ func (s *server) GetProfile(dest *pb.GetDestination, stream pb.Destination_GetPr
 	// We build up the pipeline of profile updaters backwards, starting from
 	// the translator which takes profile updates, translates them to protobuf
 	// and pushes them onto the gRPC stream.
-	translator := newProfileTranslator(stream, log, fqn, port, nil)
+	translator := newProfileTranslator(stream, log, fqn, port)
 
 	// The opaque ports adaptor merges profile updates with service opaque
 	// port annotation updates; it then publishes the result to the traffic
@@ -325,53 +369,33 @@ func (s *server) GetProfile(dest *pb.GetDestination, stream pb.Destination_GetPr
 	return nil
 }
 
-// sendEndpointProfile sends a DestinationProfile response back to the client.
-// If the pod argument is provided, the profile sent to the client will
-// include an endpoint. Otherwise, the default profile is sent.
-func (s *server) sendEndpointProfile(stream pb.Destination_GetProfileServer, pod *corev1.Pod, port uint32) error {
-	log := s.log
-	var weightedAddr *pb.WeightedAddr
-	opaquePorts := make(map[uint32]struct{})
-	var err error
-	if pod != nil {
-		ownerKind, ownerName := s.k8sAPI.GetOwnerKindAndName(context.Background(), pod, true)
-		address := watcher.Address{
-			IP:        pod.Status.PodIP,
-			Port:      port,
-			Pod:       pod,
-			OwnerName: ownerName,
-			OwnerKind: ownerKind,
-		}
-		opaquePorts, err = getPodOpaquePorts(pod, s.defaultOpaquePorts)
-		if err != nil {
-			log.Errorf("failed to get opaque ports for pod %s/%s: %s", pod.Namespace, pod.Name, err)
-		}
-		weightedAddr, err = toWeightedAddr(address, opaquePorts, s.enableH2Upgrade, s.identityTrustDomain, s.controllerNS, s.log)
-		if err != nil {
-			return err
-		}
-
-		// `Get` doesn't include the namespace in the per-endpoint
-		// metadata, so it needs to be special-cased.
-		weightedAddr.MetricLabels["namespace"] = pod.Namespace
+func (s *server) createAddress(pod *corev1.Pod, port uint32) (watcher.Address, error) {
+	ownerKind, ownerName := s.k8sAPI.GetOwnerKindAndName(context.Background(), pod, true)
+	address := watcher.Address{
+		IP:        pod.Status.PodIP,
+		Port:      port,
+		Pod:       pod,
+		OwnerName: ownerName,
+		OwnerKind: ownerKind,
 	}
+	err := watcher.SetToServerProtocol(s.k8sAPI, &address, port)
+	if err != nil {
+		return watcher.Address{}, fmt.Errorf("failed to set address OpaqueProtocol: %s", err)
+	}
+	return address, nil
+}
 
-	// Send the default profile without subscribing for future updates. The
-	// profile response will also include an endpoint if the IP (or hostname)
-	// sent in the profile request maps to a pod.
-	translator := newProfileTranslator(stream, log, "", port, weightedAddr)
-
-	// If there are opaque ports then update the profile translator
-	// with a service profile that has those values
-	if len(opaquePorts) != 0 {
-		sp := sp.ServiceProfile{}
-		sp.Spec.OpaquePorts = opaquePorts
-		translator.Update(&sp)
-	} else {
-		translator.Update(nil)
+func (s *server) createEndpoint(address watcher.Address, opaquePorts map[uint32]struct{}) (*pb.WeightedAddr, error) {
+	weightedAddr, err := createWeightedAddr(address, opaquePorts, s.enableH2Upgrade, s.identityTrustDomain, s.controllerNS, s.log)
+	if err != nil {
+		return nil, err
 	}
 
-	return nil
+	// `Get` doesn't include the namespace in the per-endpoint
+	// metadata, so it needs to be special-cased.
+	weightedAddr.MetricLabels["namespace"] = address.Pod.Namespace
+
+	return weightedAddr, err
 }
 
 // getSvcID returns the service that corresponds to a Cluster IP address if one
@@ -551,7 +575,7 @@ func profileID(authority string, ctxToken contextToken, clusterDomain string) (w
 func getHostAndPort(authority string) (string, watcher.Port, error) {
 	hostPort := strings.Split(authority, ":")
 	if len(hostPort) > 2 {
-		return "", 0, fmt.Errorf("Invalid destination %s", authority)
+		return "", 0, fmt.Errorf("invalid destination %s", authority)
 	}
 	host := hostPort[0]
 	port := 80
@@ -559,7 +583,7 @@ func getHostAndPort(authority string) (string, watcher.Port, error) {
 		var err error
 		port, err = strconv.Atoi(hostPort[1])
 		if err != nil || port <= 0 || port > 65535 {
-			return "", 0, fmt.Errorf("Invalid port %s", hostPort[1])
+			return "", 0, fmt.Errorf("invalid port %s", hostPort[1])
 		}
 	}
 	return host, watcher.Port(port), nil
@@ -616,7 +640,10 @@ func hasSuffix(slice []string, suffix []string) bool {
 	return true
 }
 
-func getPodOpaquePorts(pod *corev1.Pod, defaultPorts map[uint32]struct{}) (map[uint32]struct{}, error) {
+func getAnnotatedOpaquePorts(pod *corev1.Pod, defaultPorts map[uint32]struct{}) (map[uint32]struct{}, error) {
+	if pod == nil {
+		return defaultPorts, nil
+	}
 	annotation, ok := pod.Annotations[labels.ProxyOpaquePortsAnnotation]
 	if !ok {
 		return defaultPorts, nil
diff --git a/controller/api/destination/watcher/endpoints_watcher.go b/controller/api/destination/watcher/endpoints_watcher.go
index 49457432a..efdad8233 100644
--- a/controller/api/destination/watcher/endpoints_watcher.go
+++ b/controller/api/destination/watcher/endpoints_watcher.go
@@ -416,6 +416,8 @@ func (ew *EndpointsWatcher) getServicePublisher(id ServiceID) (sp *servicePublis
 }
 
 func (ew *EndpointsWatcher) addServer(obj interface{}) {
+	ew.Lock()
+	defer ew.Unlock()
 	server := obj.(*v1beta1.Server)
 	for _, sp := range ew.publishers {
 		sp.updateServer(server, true)
@@ -423,6 +425,8 @@ func (ew *EndpointsWatcher) addServer(obj interface{}) {
 }
 
 func (ew *EndpointsWatcher) deleteServer(obj interface{}) {
+	ew.Lock()
+	defer ew.Unlock()
 	server := obj.(*v1beta1.Server)
 	for _, sp := range ew.publishers {
 		sp.updateServer(server, false)
@@ -770,7 +774,7 @@ func (pp *portPublisher) endpointSliceToAddresses(es *discovery.EndpointSlice) A
 					pp.log.Errorf("Unable to create new address:%v", err)
 					continue
 				}
-				err = setToServerProtocol(pp.k8sAPI, &address, resolvedPort)
+				err = SetToServerProtocol(pp.k8sAPI, &address, resolvedPort)
 				if err != nil {
 					pp.log.Errorf("failed to set address OpaqueProtocol: %s", err)
 					continue
@@ -823,7 +827,7 @@ func (pp *portPublisher) endpointsToAddresses(endpoints *corev1.Endpoints) Addre
 					pp.log.Errorf("Unable to create new address:%v", err)
 					continue
 				}
-				err = setToServerProtocol(pp.k8sAPI, &address, resolvedPort)
+				err = SetToServerProtocol(pp.k8sAPI, &address, resolvedPort)
 				if err != nil {
 					pp.log.Errorf("failed to set address OpaqueProtocol: %s", err)
 					continue
@@ -1142,7 +1146,9 @@ func isValidSlice(es *discovery.EndpointSlice) bool {
 	return true
 }
 
-func setToServerProtocol(k8sAPI *k8s.API, address *Address, port Port) error {
+// SetToServerProtocol sets the address's OpaqueProtocol field based off any
+// Servers that select it and override the expected protocol.
+func SetToServerProtocol(k8sAPI *k8s.API, address *Address, port Port) error {
 	servers, err := k8sAPI.Srv().Lister().Servers("").List(labels.Everything())
 	if err != nil {
 		return fmt.Errorf("failed to list Servers: %s", err)
diff --git a/controller/api/destination/watcher/server_watcher.go b/controller/api/destination/watcher/server_watcher.go
new file mode 100644
index 000000000..a8e91ccf8
--- /dev/null
+++ b/controller/api/destination/watcher/server_watcher.go
@@ -0,0 +1,151 @@
+package watcher
+
+import (
+	"sync"
+
+	"github.com/linkerd/linkerd2/controller/gen/apis/server/v1beta1"
+	"github.com/linkerd/linkerd2/controller/k8s"
+	logging "github.com/sirupsen/logrus"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/client-go/tools/cache"
+)
+
+// ServerWatcher watches all the servers in the cluster. When there is an
+// update, it only sends updates to listeners if their endpoint's protocol
+// is changed by the Server.
+type ServerWatcher struct {
+	subscriptions map[podPort][]ServerUpdateListener
+	k8sAPI        *k8s.API
+	log           *logging.Entry
+	sync.RWMutex
+}
+
+type podPort struct {
+	pod  *corev1.Pod
+	port Port
+}
+
+// ServerUpdateListener is the interface that subscribers must implement.
+type ServerUpdateListener interface {
+	// UpdateProtocol takes a bool which is set to true if the endpoint is
+	// opaque and false otherwise. This value is used to send a
+	// DestinationProfile update to listeners for that endpoint.
+	UpdateProtocol(bool)
+}
+
+// NewServerWatcher creates a new ServerWatcher.
+func NewServerWatcher(k8sAPI *k8s.API, log *logging.Entry) *ServerWatcher {
+	sw := &ServerWatcher{
+		subscriptions: make(map[podPort][]ServerUpdateListener),
+		k8sAPI:        k8sAPI,
+		log:           log,
+	}
+	k8sAPI.Srv().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    sw.addServer,
+		DeleteFunc: sw.deleteServer,
+		UpdateFunc: func(_, obj interface{}) { sw.addServer(obj) },
+	})
+	return sw
+}
+
+// Subscribe subscribes a listener for any Server updates that may select the
+// endpoint and change its expected protocol.
+func (sw *ServerWatcher) Subscribe(pod *corev1.Pod, port Port, listener ServerUpdateListener) {
+	sw.Lock()
+	defer sw.Unlock()
+	pp := podPort{
+		pod:  pod,
+		port: port,
+	}
+	listeners, ok := sw.subscriptions[pp]
+	if !ok {
+		sw.subscriptions[pp] = []ServerUpdateListener{listener}
+		return
+	}
+	listeners = append(listeners, listener)
+	sw.subscriptions[pp] = listeners
+}
+
+// Unsubscribe unsubcribes a listener from any Server updates.
+func (sw *ServerWatcher) Unsubscribe(pod *corev1.Pod, port Port, listener ServerUpdateListener) {
+	sw.Lock()
+	defer sw.Unlock()
+	pp := podPort{
+		pod:  pod,
+		port: port,
+	}
+	listeners, ok := sw.subscriptions[pp]
+	if !ok {
+		sw.log.Errorf("cannot unsubscribe from unknown Pod: %s/%s:%d", pod.Namespace, pod.Name, port)
+		return
+	}
+	for i, l := range listeners {
+		if l == listener {
+			n := len(listeners)
+			listeners[i] = listeners[n-1]
+			listeners[n-1] = nil
+			listeners = listeners[:n-1]
+		}
+	}
+	sw.subscriptions[pp] = listeners
+}
+
+func (sw *ServerWatcher) addServer(obj interface{}) {
+	server := obj.(*v1beta1.Server)
+	selector, err := metav1.LabelSelectorAsSelector(server.Spec.PodSelector)
+	if err != nil {
+		sw.log.Errorf("failed to create Selector: %s", err)
+		return
+	}
+	sw.updateServer(server, selector, true)
+}
+
+func (sw *ServerWatcher) deleteServer(obj interface{}) {
+	server := obj.(*v1beta1.Server)
+	selector, err := metav1.LabelSelectorAsSelector(server.Spec.PodSelector)
+	if err != nil {
+		sw.log.Errorf("failed to create Selector: %s", err)
+		return
+	}
+	sw.updateServer(server, selector, false)
+}
+
+func (sw *ServerWatcher) updateServer(server *v1beta1.Server, selector labels.Selector, isAdd bool) {
+	sw.Lock()
+	defer sw.Unlock()
+	for pp, listeners := range sw.subscriptions {
+		if selector.Matches(labels.Set(pp.pod.Labels)) {
+			var portMatch bool
+			switch server.Spec.Port.Type {
+			case intstr.Int:
+				if server.Spec.Port.IntVal == int32(pp.port) {
+					portMatch = true
+				}
+			case intstr.String:
+				for _, c := range pp.pod.Spec.Containers {
+					for _, p := range c.Ports {
+						if p.ContainerPort == int32(pp.port) && p.Name == server.Spec.Port.StrVal {
+							portMatch = true
+						}
+					}
+				}
+			default:
+				continue
+			}
+			if portMatch {
+				var isOpaque bool
+				if isAdd && server.Spec.ProxyProtocol == opaqueProtocol {
+					isOpaque = true
+				} else {
+					isOpaque = false
+				}
+				for _, listener := range listeners {
+					listener.UpdateProtocol(isOpaque)
+				}
+			}
+		}
+	}
+}
diff --git a/go.mod b/go.mod
index 19589bf49..40aaac89c 100644
--- a/go.mod
+++ b/go.mod
@@ -35,8 +35,8 @@ require (
 	github.com/spf13/cobra v1.2.1
 	github.com/spf13/pflag v1.0.5
 	go.opencensus.io v0.23.0
-	golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d
-	golang.org/x/tools v0.1.7
+	golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f
+	golang.org/x/tools v0.1.8
 	google.golang.org/grpc v1.42.0
 	google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0
 	google.golang.org/protobuf v1.27.1
@@ -140,12 +140,12 @@ require (
 	github.com/xlab/treeprint v0.0.0-20181112141820-a009c3971eca // indirect
 	go.starlark.net v0.0.0-20200306205701-8dd3e2ee1dd5 // indirect
 	golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a // indirect
-	golang.org/x/mod v0.4.2 // indirect
+	golang.org/x/mod v0.5.1 // indirect
 	golang.org/x/oauth2 v0.0.0-20210514164344-f6687ab2804c // indirect
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
-	golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e // indirect
+	golang.org/x/sys v0.0.0-20211019181941-9d821ace8654 // indirect
 	golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d // indirect
-	golang.org/x/text v0.3.6 // indirect
+	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	google.golang.org/api v0.44.0 // indirect
diff --git a/go.sum b/go.sum
index 6ec213fc5..c0360d3a3 100644
--- a/go.sum
+++ b/go.sum
@@ -943,7 +943,7 @@ github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9de
 github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
-github.com/yuin/goldmark v1.4.0/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
+github.com/yuin/goldmark v1.4.1/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 github.com/yvasiyarov/go-metrics v0.0.0-20140926110328-57bccd1ccd43 h1:+lm10QQTNSBd8DVTNGHx7o/IKu9HYDvLMffDhbyLccI=
 github.com/yvasiyarov/go-metrics v0.0.0-20140926110328-57bccd1ccd43/go.mod h1:aX5oPXxHm3bOH+xeAttToC8pqch2ScQN/JoXYupl6xs=
 github.com/yvasiyarov/gorelic v0.0.0-20141212073537-a9bba5b9ab50 h1:hlE8//ciYMztlGpl/VA+Zm1AcTPHYkHJPbHqE6WJUXE=
@@ -1051,8 +1051,9 @@ golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.1/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.4.2 h1:Gz96sIWK3OalVv/I/qNygP42zyoKp3xptRVCWRFEBvo=
 golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.5.1 h1:OJxoQ/rynoF0dcCdI7cLPktw/hR2cueqYfjm43oqK38=
+golang.org/x/mod v0.5.1/go.mod h1:5OXOZSfqPIIbmVBIIKWRFfZjPR0E5r58TLhUjH0a2Ro=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -1105,8 +1106,8 @@ golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLd
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/net v0.0.0-20210520170846-37e1c6afe023/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d h1:20cMwl2fHAzkJMEA+8J4JgqBQcQGzbisXo31MIeenXI=
-golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f h1:OfiFi4JbukWwe3lzw+xunroH1mnC1e2Gy5cxNJApiSY=
+golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -1221,8 +1222,8 @@ golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e h1:WUoyKPm6nCo1BnNUvPGnFG3T5DUVem42yDJZZ4CNxMA=
-golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20211019181941-9d821ace8654 h1:id054HUawV2/6IGm2IV8KZQjqtwAOo2CYlOToYqa0d0=
+golang.org/x/sys v0.0.0-20211019181941-9d821ace8654/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d h1:SZxvLBoTP5yHO3Frd4z4vrF+DBX9vMVanchswa69toE=
@@ -1234,8 +1235,9 @@ golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
@@ -1302,8 +1304,8 @@ golang.org/x/tools v0.0.0-20210105154028-b0ab187a4818/go.mod h1:emZCQorbCU4vsT4f
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0=
 golang.org/x/tools v0.1.2/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
-golang.org/x/tools v0.1.7 h1:6j8CgantCy3yc8JGBqkDLMKWqZ0RDU2g1HVgacojGWQ=
-golang.org/x/tools v0.1.7/go.mod h1:LGqMHiF4EqQNHR1JncWGqT5BVaXmza+X+BDGol+dOxo=
+golang.org/x/tools v0.1.8 h1:P1HhGGuLW4aAclzjtmJdf0mJOjVUZUzOTqkAkWL+l6w=
+golang.org/x/tools v0.1.8/go.mod h1:nABZi5QlRsZVlzPpHl034qft6wpY4eDcsTt5AaioBiU=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
diff --git a/pkg/k8s/fake.go b/pkg/k8s/fake.go
index f504d2c66..a8cebcc6c 100644
--- a/pkg/k8s/fake.go
+++ b/pkg/k8s/fake.go
@@ -87,6 +87,8 @@ func NewFakeClientSets(configs ...string) (
 			discoveryObjs = append(discoveryObjs, obj)
 		case ServiceProfile:
 			spObjs = append(spObjs, obj)
+		case Server:
+			spObjs = append(spObjs, obj)
 		default:
 			objs = append(objs, obj)
 		}
diff --git a/web/app/package.json b/web/app/package.json
index f96fc408d..a951aac0e 100644
--- a/web/app/package.json
+++ b/web/app/package.json
@@ -16,7 +16,7 @@
     "@material-ui/icons": "4.11.2",
     "@material-ui/lab": "^4.0.0-alpha.60",
     "classnames": "2.3.1",
-    "core-js": "^3.19.2",
+    "core-js": "^3.19.3",
     "css-mediaquery": "^0.1.2",
     "d3-drag": "3.0.0",
     "d3-force": "3.0.0",
@@ -59,7 +59,7 @@
     "css-loader": "^6.5.1",
     "enzyme": "3.11.0",
     "enzyme-adapter-react-16": "^1.15.6",
-    "eslint": "^8.3.0",
+    "eslint": "^8.4.1",
     "eslint-config-airbnb": "^19.0.2",
     "eslint-plugin-import": "^2.25.3",
     "eslint-plugin-jsx-a11y": "^6.5.1",
diff --git a/web/app/yarn.lock b/web/app/yarn.lock
index 0c2e5dacd..170417aa2 100644
--- a/web/app/yarn.lock
+++ b/web/app/yarn.lock
@@ -1014,14 +1014,14 @@
     ts-node "^9"
     tslib "^2"
 
-"@eslint/eslintrc@^1.0.4":
-  version "1.0.4"
-  resolved "https://registry.yarnpkg.com/@eslint/eslintrc/-/eslintrc-1.0.4.tgz#dfe0ff7ba270848d10c5add0715e04964c034b31"
-  integrity sha512-h8Vx6MdxwWI2WM8/zREHMoqdgLNXEL4QX3MWSVMdyNJGvXVOs+6lp+m2hc3FnuMHDc4poxFNI20vCk0OmI4G0Q==
+"@eslint/eslintrc@^1.0.5":
+  version "1.0.5"
+  resolved "https://registry.yarnpkg.com/@eslint/eslintrc/-/eslintrc-1.0.5.tgz#33f1b838dbf1f923bfa517e008362b78ddbbf318"
+  integrity sha512-BLxsnmK3KyPunz5wmCCpqy0YelEoxxGmH73Is+Z74oOTMtExcjkr3dDR6quwrjh1YspA8DH9gnX1o069KiS9AQ==
   dependencies:
     ajv "^6.12.4"
     debug "^4.3.2"
-    espree "^9.0.0"
+    espree "^9.2.0"
     globals "^13.9.0"
     ignore "^4.0.6"
     import-fresh "^3.2.1"
@@ -1062,19 +1062,19 @@
   dependencies:
     prop-types "^15.7.2"
 
-"@humanwhocodes/config-array@^0.6.0":
-  version "0.6.0"
-  resolved "https://registry.yarnpkg.com/@humanwhocodes/config-array/-/config-array-0.6.0.tgz#b5621fdb3b32309d2d16575456cbc277fa8f021a"
-  integrity sha512-JQlEKbcgEUjBFhLIF4iqM7u/9lwgHRBcpHrmUNCALK0Q3amXN6lxdoXLnF0sm11E9VqTmBALR87IlUg1bZ8A9A==
+"@humanwhocodes/config-array@^0.9.2":
+  version "0.9.2"
+  resolved "https://registry.yarnpkg.com/@humanwhocodes/config-array/-/config-array-0.9.2.tgz#68be55c737023009dfc5fe245d51181bb6476914"
+  integrity sha512-UXOuFCGcwciWckOpmfKDq/GyhlTf9pN/BzG//x8p8zTOFEcGuA68ANXheFS0AGvy3qgZqLBUkMs7hqzqCKOVwA==
   dependencies:
-    "@humanwhocodes/object-schema" "^1.2.0"
+    "@humanwhocodes/object-schema" "^1.2.1"
     debug "^4.1.1"
     minimatch "^3.0.4"
 
-"@humanwhocodes/object-schema@^1.2.0":
-  version "1.2.0"
-  resolved "https://registry.yarnpkg.com/@humanwhocodes/object-schema/-/object-schema-1.2.0.tgz#87de7af9c231826fdd68ac7258f77c429e0e5fcf"
-  integrity sha512-wdppn25U8z/2yiaT6YGquE6X8sSv7hNMWSXYSSU1jGv/yd6XqjXgTDJ8KP4NgjTXfJ3GbRjeeb8RTV7a/VpM+w==
+"@humanwhocodes/object-schema@^1.2.1":
+  version "1.2.1"
+  resolved "https://registry.yarnpkg.com/@humanwhocodes/object-schema/-/object-schema-1.2.1.tgz#b520529ec21d8e5945a1851dfd1c32e94e39ff45"
+  integrity sha512-ZnQMnLV4e7hDlUvw8H+U8ASL02SS2Gn6+9Ac3wGGLIe7+je2AeAOxPY+izIPJDfFDb7eDjev0Us8MO1iFRN8hA==
 
 "@istanbuljs/load-nyc-config@^1.0.0":
   version "1.1.0"
@@ -2811,10 +2811,10 @@ core-js-pure@^3.16.0:
   resolved "https://registry.yarnpkg.com/core-js-pure/-/core-js-pure-3.18.0.tgz#e5187347bae66448c9e2d67c01c34c4df3261dc5"
   integrity sha512-ZnK+9vyuMhKulIGqT/7RHGRok8RtkHMEX/BGPHkHx+ouDkq+MUvf9mfIgdqhpmPDu8+V5UtRn/CbCRc9I4lX4w==
 
-core-js@^3.19.2:
-  version "3.19.2"
-  resolved "https://registry.yarnpkg.com/core-js/-/core-js-3.19.2.tgz#ae216d7f4f7e924d9a2e3ff1e4b1940220f9157b"
-  integrity sha512-ciYCResnLIATSsXuXnIOH4CbdfgV+H1Ltg16hJFN7/v6OxqnFr/IFGeLacaZ+fHLAm0TBbXwNK9/DNBzBUrO/g==
+core-js@^3.19.3:
+  version "3.19.3"
+  resolved "https://registry.yarnpkg.com/core-js/-/core-js-3.19.3.tgz#6df8142a996337503019ff3235a7022d7cdf4559"
+  integrity sha512-LeLBMgEGSsG7giquSzvgBrTS7V5UL6ks3eQlUSbN8dJStlLFiRzUm5iqsRyzUB8carhfKjkJ2vzKqE6z1Vga9g==
 
 core-util-is@~1.0.0:
   version "1.0.3"
@@ -3639,13 +3639,13 @@ eslint-webpack-plugin@^3.1.1:
     normalize-path "^3.0.0"
     schema-utils "^3.1.1"
 
-eslint@^8.3.0:
-  version "8.3.0"
-  resolved "https://registry.yarnpkg.com/eslint/-/eslint-8.3.0.tgz#a3c2409507403c1c7f6c42926111d6cbefbc3e85"
-  integrity sha512-aIay56Ph6RxOTC7xyr59Kt3ewX185SaGnAr8eWukoPLeriCrvGjvAubxuvaXOfsxhtwV5g0uBOsyhAom4qJdww==
+eslint@^8.4.1:
+  version "8.4.1"
+  resolved "https://registry.yarnpkg.com/eslint/-/eslint-8.4.1.tgz#d6531bbf3e598dffd7c0c7d35ec52a0b30fdfa2d"
+  integrity sha512-TxU/p7LB1KxQ6+7aztTnO7K0i+h0tDi81YRY9VzB6Id71kNz+fFYnf5HD5UOQmxkzcoa0TlVZf9dpMtUv0GpWg==
   dependencies:
-    "@eslint/eslintrc" "^1.0.4"
-    "@humanwhocodes/config-array" "^0.6.0"
+    "@eslint/eslintrc" "^1.0.5"
+    "@humanwhocodes/config-array" "^0.9.2"
     ajv "^6.10.0"
     chalk "^4.0.0"
     cross-spawn "^7.0.2"
@@ -3656,7 +3656,7 @@ eslint@^8.3.0:
     eslint-scope "^7.1.0"
     eslint-utils "^3.0.0"
     eslint-visitor-keys "^3.1.0"
-    espree "^9.1.0"
+    espree "^9.2.0"
     esquery "^1.4.0"
     esutils "^2.0.2"
     fast-deep-equal "^3.1.3"
@@ -3683,10 +3683,10 @@ eslint@^8.3.0:
     text-table "^0.2.0"
     v8-compile-cache "^2.0.3"
 
-espree@^9.0.0, espree@^9.1.0:
-  version "9.1.0"
-  resolved "https://registry.yarnpkg.com/espree/-/espree-9.1.0.tgz#ba9d3c9b34eeae205724124e31de4543d59fbf74"
-  integrity sha512-ZgYLvCS1wxOczBYGcQT9DDWgicXwJ4dbocr9uYN+/eresBAUuBu+O4WzB21ufQ/JqQT8gyp7hJ3z8SHii32mTQ==
+espree@^9.2.0:
+  version "9.2.0"
+  resolved "https://registry.yarnpkg.com/espree/-/espree-9.2.0.tgz#c50814e01611c2d0f8bd4daa83c369eabba80dbc"
+  integrity sha512-oP3utRkynpZWF/F2x/HZJ+AGtnIclaR7z1pYPxy7NYM2fSO6LgK/Rkny8anRSPK/VwEA1eqm2squui0T7ZMOBg==
   dependencies:
     acorn "^8.6.0"
     acorn-jsx "^5.3.1"


Test patch:
diff --git a/controller/api/destination/server_test.go b/controller/api/destination/server_test.go
index 9825e1bf4..d800dc734 100644
--- a/controller/api/destination/server_test.go
+++ b/controller/api/destination/server_test.go
@@ -25,6 +25,7 @@ const podIP1 = "172.17.0.12"
 const podIP2 = "172.17.0.13"
 const podIPOpaque = "172.17.0.14"
 const podIPSkipped = "172.17.0.15"
+const podIPPolicy = "172.17.0.16"
 const podIPStatefulSet = "172.17.13.15"
 const port uint32 = 8989
 const opaquePort uint32 = 4242
@@ -313,12 +314,52 @@ status:
   podIP: 172.17.13.15`,
 	}
 
+	policyResources := []string{
+		`
+apiVersion: v1
+kind: Pod
+metadata:
+  labels:
+    linkerd.io/control-plane-ns: linkerd
+    app: policy-test
+  name: pod-policyResources
+  namespace: ns
+status:
+  phase: Running
+  podIP: 172.17.0.16
+spec:
+  containers:
+    - name: linkerd-proxy
+      env:
+      - name: LINKERD2_PROXY_INBOUND_LISTEN_ADDR
+        value: 0.0.0.0:4143
+    - name: app
+      image: nginx
+      ports:
+      - containerPort: 80
+        name: http
+        protocol: TCP`,
+		`
+apiVersion: policy.linkerd.io/v1beta1
+kind: Server
+metadata:
+  name: srv
+  namespace: ns
+spec:
+  podSelector:
+    matchLabels:
+      app: policy-test
+  port: 80
+  proxyProtocol: opaque`,
+	}
+
 	res := append(meshedPodResources, clientSP...)
 	res = append(res, unmeshedPod)
 	res = append(res, meshedOpaquePodResources...)
 	res = append(res, meshedOpaqueServiceResources...)
 	res = append(res, meshedSkippedPodResource...)
 	res = append(res, meshedStatefulSetPodResource...)
+	res = append(res, policyResources...)
 	k8sAPI, err := k8s.NewFakeAPI(res...)
 	if err != nil {
 		t.Fatalf("NewFakeAPI returned an error: %s", err)
@@ -341,6 +382,7 @@ status:
 	endpoints := watcher.NewEndpointsWatcher(k8sAPI, log, false)
 	opaquePorts := watcher.NewOpaquePortsWatcher(k8sAPI, log, defaultOpaquePorts)
 	profiles := watcher.NewProfileWatcher(k8sAPI, log)
+	servers := watcher.NewServerWatcher(k8sAPI, log)
 
 	// Sync after creating watchers so that the the indexers added get updated
 	// properly
@@ -351,6 +393,7 @@ status:
 		endpoints,
 		opaquePorts,
 		profiles,
+		servers,
 		k8sAPI.Node(),
 		true,
 		"linkerd",
@@ -915,7 +958,7 @@ func TestGetProfiles(t *testing.T) {
 		if first.Endpoint.ProtocolHint == nil {
 			t.Fatalf("Expected protocol hint but found none")
 		}
-		if first.Endpoint.ProtocolHint.GetOpaqueTransport().InboundPort != 4143 {
+		if first.Endpoint.ProtocolHint.GetOpaqueTransport().GetInboundPort() != 4143 {
 			t.Fatalf("Expected pod to support opaque traffic on port 4143")
 		}
 		if first.Endpoint.Addr.String() != epAddr.String() {
@@ -986,6 +1029,46 @@ func TestGetProfiles(t *testing.T) {
 			t.Fatalf("Expected TLS identity for %s to be nil but got %+v", path, addr.TlsIdentity)
 		}
 	})
+
+	t.Run("Return profile with opaque protocol when using Pod IP selected by a Server", func(t *testing.T) {
+		server := makeServer(t)
+		stream := &bufferingGetProfileStream{
+			updates:          []*pb.DestinationProfile{},
+			MockServerStream: util.NewMockServerStream(),
+		}
+		stream.Cancel()
+
+		_, err := toAddress(podIPPolicy, 80)
+		if err != nil {
+			t.Fatalf("Got error: %s", err)
+		}
+		err = server.GetProfile(&pb.GetDestination{
+			Scheme: "k8s",
+			Path:   fmt.Sprintf("%s:%d", podIPPolicy, 80),
+		}, stream)
+		if err != nil {
+			t.Fatalf("Got error: %s", err)
+		}
+
+		// Test that the first update has a destination profile with an
+		// opaque protocol and opaque transport.
+		if len(stream.updates) == 0 {
+			t.Fatalf("Expected at least 1 update but got 0")
+		}
+		update := stream.updates[0]
+		if update.Endpoint == nil {
+			t.Fatalf("Expected response to have endpoint field")
+		}
+		if !update.OpaqueProtocol {
+			t.Fatalf("Expected port %d to be an opaque protocol, but it was not", 80)
+		}
+		if update.Endpoint.ProtocolHint == nil {
+			t.Fatalf("Expected protocol hint but found none")
+		}
+		if update.Endpoint.ProtocolHint.GetOpaqueTransport().GetInboundPort() != 4143 {
+			t.Fatalf("Expected pod to support opaque traffic on port 4143")
+		}
+	})
 }
 
 func TestTokenStructure(t *testing.T) {
diff --git a/controller/api/destination/test_helper.go b/controller/api/destination/test_helper.go
index 5dbd91e50..3d86713be 100644
--- a/controller/api/destination/test_helper.go
+++ b/controller/api/destination/test_helper.go
@@ -25,7 +25,7 @@ func (c *MockAPIClient) Get(ctx context.Context, in *destinationPb.GetDestinatio
 // GetProfile provides a mock of a destination API method
 func (c *MockAPIClient) GetProfile(ctx context.Context, _ *destinationPb.GetDestination, _ ...grpc.CallOption) (destinationPb.Destination_GetProfileClient, error) {
 	// Not implemented through this client. The proxies use the gRPC server directly instead.
-	return nil, errors.New("Not implemented")
+	return nil, errors.New("not implemented")
 }
 
 // MockDestinationGetClient satisfies the Destination_GetClient gRPC interface.