Add OpaqueTransport field to destination protocol hints (#5421)

## What

When the destination service returns a destination profile for an endpoint,
indicate if the endpoint can receive opaque traffic.

## Why

Closes #5400

## How

When translating a pod address to a destination profile, the destination service
checks if the pod is controlled by any linkerd control plane. If it is, it can
set a protocol hint where we indicate that it supports H2 and opaque traffic.

If the pod supports opaque traffic, we need to get the port that it expects
inbound traffic on. We do this by getting the proxy container and reading it's
`LINKERD2_PROXY_INBOUND_LISTEN_ADDR` environment variable. If we successfully
parse that into a port, we can set the opaque transport field in the destination
profile.

## Testing

A test has been added to the destination server where a pod has a
`linkerd-proxy` container. We can expect the `OpaqueTransport` field to be set
in the returned destination profile's protocol hint.

Signed-off-by: Kevin Leimkuhler <kevin@kleimkuhler.com>

controller/api/destination/endpoint_translator.go

@@ -3,6 +3,8 @@ package destination
 import (
 	"context"
 	"fmt"
+	"strconv"
+	"strings"
 
 	pb "github.com/linkerd/linkerd2-proxy-api/go/destination"
 	"github.com/linkerd/linkerd2-proxy-api/go/net"
@@ -15,7 +17,12 @@ import (
 	"k8s.io/client-go/kubernetes"
 )
 
-const defaultWeight uint32 = 10000
+const (
+	defaultWeight uint32 = 10000
+	// inboundListenAddr is the environment variable holding the inbound
+	// listening address for the proxy container.
+	envInboundListenAddr = "LINKERD2_PROXY_INBOUND_LISTEN_ADDR"
+)
 
 // endpointTranslator satisfies EndpointUpdateListener and translates updates
 // into Destination.Get messages.
@@ -209,7 +216,7 @@ func (et *endpointTranslator) sendClientAdd(set watcher.AddressSet) {
 			err error
 		)
 		if address.Pod != nil {
-			wa, err = toWeightedAddr(address, et.enableH2Upgrade, et.identityTrustDomain, et.controllerNS)
+			wa, err = toWeightedAddr(address, et.enableH2Upgrade, et.identityTrustDomain, et.controllerNS, et.log)
 		} else {
 			var authOverride *pb.AuthorityOverride
 			if address.AuthorityOverride != "" {
@@ -299,19 +306,44 @@ func toAddr(address watcher.Address) (*net.TcpAddress, error) {
 	}, nil
 }
 
-func toWeightedAddr(address watcher.Address, enableH2Upgrade bool, identityTrustDomain string, controllerNS string) (*pb.WeightedAddr, error) {
+func toWeightedAddr(address watcher.Address, enableH2Upgrade bool, identityTrustDomain string, controllerNS string, log *logging.Entry) (*pb.WeightedAddr, error) {
 	controllerNSLabel := address.Pod.Labels[k8s.ControllerNSLabel]
 	sa, ns := k8s.GetServiceAccountAndNS(address.Pod)
 	labels := k8s.GetPodLabels(address.OwnerKind, address.OwnerName, address.Pod)
 
-	// If the pod is controlled by any Linkerd control plane, then it can be hinted
-	// that this destination knows H2 (and handles our orig-proto translation).
+	// If the pod is controlled by any Linkerd control plane, then it can be
+	// hinted that this destination knows H2 (and handles our orig-proto
+	// translation) and supports receiving opaque traffic.
 	var hint *pb.ProtocolHint
 	if enableH2Upgrade && controllerNSLabel != "" {
+		// Get the inbound port from the proxy container's environment
+		// variable so that it can be set in the protocol hint.
+		var inboundPort uint32
+	loop:
+		for _, containerSpec := range address.Pod.Spec.Containers {
+			if containerSpec.Name != k8s.ProxyContainerName {
+				continue
+			}
+			for _, envVar := range containerSpec.Env {
+				if envVar.Name != envInboundListenAddr {
+					continue
+				}
+				addr := strings.Split(envVar.Value, ":")
+				port, err := strconv.ParseUint(addr[1], 10, 32)
+				if err != nil {
+					log.Errorf("failed to parse inbound port for proxy container: %s", err)
+				}
+				inboundPort = uint32(port)
+				break loop
+			}
+		}
 		hint = &pb.ProtocolHint{
 			Protocol: &pb.ProtocolHint_H2_{
 				H2: &pb.ProtocolHint_H2{},
 			},
+			OpaqueTransport: &pb.ProtocolHint_OpaqueTransport{
+				InboundPort: inboundPort,
+			},
 		}
 	}
 

controller/api/destination/server.go

@@ -208,7 +208,7 @@ func (s *server) GetProfile(dest *pb.GetDestination, stream pb.Destination_GetPr
 					Namespace: pod.Namespace,
 					Name:      pod.Name,
 				}
-				endpoint, err = toWeightedAddr(podSet.Addresses[podID], s.enableH2Upgrade, s.identityTrustDomain, s.controllerNS)
+				endpoint, err = toWeightedAddr(podSet.Addresses[podID], s.enableH2Upgrade, s.identityTrustDomain, s.controllerNS, log)
 				if err != nil {
 					return err
 				}

controller/api/destination/server_test.go

@@ -84,7 +84,13 @@ metadata:
   namespace: ns
 status:
   phase: Running
-  podIP: 172.17.0.12`,
+  podIP: 172.17.0.12
+spec:
+  containers:
+    - env:
+      - name: LINKERD2_PROXY_INBOUND_LISTEN_ADDR
+        value: 0.0.0.0:4143
+      name: linkerd-proxy`,
 		`
 apiVersion: v1
 kind: Pod
@@ -476,6 +482,9 @@ 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 {
+			t.Fatalf("Expected pod to support opaque traffic on port 4143")
+		}
 		if first.Endpoint.Addr.String() != epAddr.String() {
 			t.Fatalf("Expected endpoint IP to be %s, but it was %s", epAddr.Ip, first.Endpoint.Addr.Ip)
 		}
@@ -613,8 +622,8 @@ func TestGetProfiles(t *testing.T) {
 		if first.Endpoint.ProtocolHint == nil {
 			t.Fatalf("Expected protocol hint but found none")
 		}
-		if !first.OpaqueProtocol {
-			t.Fatalf("Expected protocol to be opaque but it was not")
+		if first.Endpoint.ProtocolHint.GetOpaqueTransport().InboundPort != 4143 {
+			t.Fatalf("Expected pod to support opaque traffic on port 4143")
 		}
 		if first.Endpoint.Addr.String() != epAddr.String() {
 			t.Fatalf("Expected endpoint IP port to be %d, but it was %d", epAddr.Port, first.Endpoint.Addr.Port)

go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/imdario/mergo v0.3.8
 	github.com/julienschmidt/httprouter v1.2.0
-	github.com/linkerd/linkerd2-proxy-api v0.1.15
+	github.com/linkerd/linkerd2-proxy-api v0.1.16
 	github.com/linkerd/linkerd2-proxy-init v1.3.8
 	github.com/mattn/go-isatty v0.0.12
 	github.com/mattn/go-runewidth v0.0.4

go.sum

@@ -461,6 +461,10 @@ github.com/lightstep/lightstep-tracer-common/golang/gogo v0.0.0-20190605223551-b
 github.com/lightstep/lightstep-tracer-go v0.18.1/go.mod h1:jlF1pusYV4pidLvZ+XD0UBX0ZE6WURAspgAczcDHrL4=
 github.com/linkerd/linkerd2-proxy-api v0.1.15 h1:hy/36GwG+BnKxxh3BAnC5cfUsH/ZAxuzuOSsGM3KfkM=
 github.com/linkerd/linkerd2-proxy-api v0.1.15/go.mod h1:yFz+DCCEomC3vpsChFzfCuOuSJtzx7jMNNHBIlbFil0=
+github.com/linkerd/linkerd2-proxy-api v0.1.16-0.20201221204305-2913e990c702 h1:CweHoGEw9PC1uz7GaHH+N276QDcW+Svk9taapkvGqX4=
+github.com/linkerd/linkerd2-proxy-api v0.1.16-0.20201221204305-2913e990c702/go.mod h1:yFz+DCCEomC3vpsChFzfCuOuSJtzx7jMNNHBIlbFil0=
+github.com/linkerd/linkerd2-proxy-api v0.1.16 h1:Qjqbw5Bw3QYUJpUSpYHr4nkJqBRnTlsFUwncCtCdOEs=
+github.com/linkerd/linkerd2-proxy-api v0.1.16/go.mod h1:yFz+DCCEomC3vpsChFzfCuOuSJtzx7jMNNHBIlbFil0=
 github.com/linkerd/linkerd2-proxy-init v1.3.7 h1:S/xBSHArQyd+hPrnkcAr+eOf+SGOoCmxr27n40fI+fo=
 github.com/linkerd/linkerd2-proxy-init v1.3.7/go.mod h1:M6iaaLLi06ofuIV6x74SDknSFi7VS/MFqa5m+CwHgLY=
 github.com/linkerd/linkerd2-proxy-init v1.3.8 h1:fo/LbrIS3FHssAPLkVXi5h8K/3mWP7ncVwOU2oI6Dm8=