destination: send Opaque protocol hint for opaque ports

The outbound proxy handles endpoints with the `opaque_transport` flag by
opening a direct connection to the inbound proxy's inbound listener
port, and sending a ProtoBuf `TransportHeader` including the target port
of the originating outbound connection and an (optional)
`SessionProtocol` describing the protocol used on that connection.

Currently, outbound proxies initiating direct connections will *always*
send `SessionProtocol` values communicating the protocol as understood
by the outbound proxy. However, this is not always the desired behavior.
Direct connections with `TransportHeader`s are used in two cases: for
gateway connections, and for ports which are marked as opaque. When the
inbound port is marked as opaque, the presence of a `SessionProtocol`
tells the inbound proxy to handle that connection as the indicated
protocol, which results in incorrect behavior when the inbound proxy's
ServerPolicy configures the target port as opaque (see #9888).

Therefore, the `Destination` proxy API has been updated to add a new
`ProtocolHint`, `Opaque`, which indicates that an outbound proxy should
_not_ send a `SessionProtocol` when initiating a direct connection, even
if the outbound proxy handled the connection as HTTP. This hint was
added to the proxy API in linkerd/linkerd2-proxy-api#197, and released
in `linkerd2-proxy-api` v0.8.0.

This branch updates the Destination controller's dependency on
`linkerd2-proxy-api` to v0.8.0, and changes the controller to send an
`Opaque` protocol hint when the target port is marked as opaque on the
destination pod. This should override the `H2` protocol hint that is
added when the destination is meshed. I've also added a new test for
this behavior.

Fixes #9888 (along with linkerd/linkerd2-proxy#2209, which changes the
proxy to actually handle the `Opaque` protocol hint).

controller/api/destination/endpoint_translator.go

@@ -392,7 +392,8 @@ func createWeightedAddr(address watcher.Address, opaquePorts map[uint32]struct{}
 		}
 		// If address is set as opaque by a Server, or its port is set as
 		// opaque by annotation or default value, then hint its proxy's
-		// inbound port.
+		// inbound port, and set the hinted protocol to Opaque, so that the
+		// client proxy does not send a SessionProtocol.
 		_, opaquePort := opaquePorts[address.Port]
 		if address.OpaqueProtocol || opaquePort {
 			port, err := getInboundPort(&address.Pod.Spec)
@@ -402,6 +403,9 @@ func createWeightedAddr(address watcher.Address, opaquePorts map[uint32]struct{}
 				weightedAddr.ProtocolHint.OpaqueTransport = &pb.ProtocolHint_OpaqueTransport{
 					InboundPort: port,
 				}
+				weightedAddr.ProtocolHint.Protocol = &pb.ProtocolHint_Opaque_{
+					Opaque: &pb.ProtocolHint_Opaque{},
+				}
 			}
 		}
 	}

controller/api/destination/endpoint_translator_test.go

@@ -69,6 +69,38 @@ var (
 		},
 	}
 
+	podOpaque = watcher.Address{
+		IP:   "1.1.1.4",
+		Port: 4,
+		Pod: &corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "pod4",
+				Namespace: "ns",
+				Labels: map[string]string{
+					k8s.ControllerNSLabel:    "linkerd",
+					k8s.ProxyDeploymentLabel: "deployment-name",
+				},
+				Annotations: map[string]string{
+					k8s.ProxyOpaquePortsAnnotation: "4",
+				},
+			},
+			Spec: corev1.PodSpec{
+				Containers: []corev1.Container{
+					{
+						Name: k8s.ProxyContainerName,
+						Env: []corev1.EnvVar{
+							{
+								Name:  envInboundListenAddr,
+								Value: "0.0.0.0:4143",
+							},
+						},
+					},
+				},
+			},
+		},
+		OpaqueProtocol: true,
+	}
+
 	remoteGateway1 = watcher.Address{
 		IP:   "1.1.1.1",
 		Port: 1,
@@ -340,6 +372,31 @@ func TestEndpointTranslatorForPods(t *testing.T) {
 			t.Fatalf("Expected TlsIdentity to be [%v] but was [%v]", expectedTLSIdentity, actualTLSIdentity)
 		}
 	})
+
+	t.Run("Sends Opaque ProtocolHint for opaque ports", func(t *testing.T) {
+		expectedProtocolHint := &pb.ProtocolHint{
+			Protocol: &pb.ProtocolHint_Opaque_{
+				Opaque: &pb.ProtocolHint_Opaque{},
+			},
+			OpaqueTransport: &pb.ProtocolHint_OpaqueTransport{
+				InboundPort: 4143,
+			},
+		}
+
+		mockGetServer, translator := makeEndpointTranslator(t)
+
+		translator.Add(mkAddressSetForServices(podOpaque))
+
+		addrs := mockGetServer.updatesReceived[0].GetAdd().GetAddrs()
+		if len(addrs) != 1 {
+			t.Fatalf("Expected [1] address returned, got %v", addrs)
+		}
+
+		actualProtocolHint := addrs[0].GetProtocolHint()
+		if diff := deep.Equal(actualProtocolHint, expectedProtocolHint); diff != nil {
+			t.Fatalf("ProtocolHint: %v", diff)
+		}
+	})
 }
 
 func TestEndpointTranslatorForZonedAddresses(t *testing.T) {