fix(destination): properly deal with native sidecar port opacity in getProfile (#14791)

(Extracted from #14566)

When hitting pods directly at their their IPs, ports in native sidecars
that were marked as opaque via the `config.linkerd.io/opaque-ports`
annotation, weren't really being marked as opaque.

More concretely, the issue layed in the getProfile API, that was
forgoing init containers when iterating over containers in this
particular case.

Endpoint profile translator tests got expanded, testing for opaque ports
in both init and regular containers.

Author: alpeb

controller/api/destination/endpoint_profile_translator_test.go

@@ -18,20 +18,49 @@ func TestEndpointProfileTranslator(t *testing.T) {
 	// logging.SetLevel(logging.TraceLevel)
 	// defer logging.SetLevel(logging.PanicLevel)
 
+	pod := corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Annotations: map[string]string{
+				consts.ProxyOpaquePortsAnnotation: "sidecar,http",
+			},
+		},
+		Spec: corev1.PodSpec{
+			InitContainers: []corev1.Container{
+				corev1.Container{
+					Ports: []corev1.ContainerPort{
+						corev1.ContainerPort{
+							Name:          "sidecar",
+							ContainerPort: 8081,
+						},
+					},
+				},
+			},
+			Containers: []corev1.Container{
+				corev1.Container{
+					Ports: []corev1.ContainerPort{
+						corev1.ContainerPort{
+							Name:          "http",
+							ContainerPort: 8080,
+						},
+					},
+				},
+			},
+		},
+	}
+
 	addr := &watcher.Address{
 		IP:   "10.10.11.11",
 		Port: 8080,
 	}
-	podAddr := &watcher.Address{
+	podAddr1 := &watcher.Address{
 		IP:   "10.10.11.11",
 		Port: 8080,
-		Pod: &corev1.Pod{
-			ObjectMeta: metav1.ObjectMeta{
-				Annotations: map[string]string{
-					consts.ProxyOpaquePortsAnnotation: "8080",
-				},
-			},
-		},
+		Pod:  &pod,
+	}
+	podAddr2 := &watcher.Address{
+		IP:   "10.10.11.11",
+		Port: 8081,
+		Pod:  &pod,
 	}
 
 	t.Run("Sends update", func(t *testing.T) {
@@ -68,13 +97,22 @@ func TestEndpointProfileTranslator(t *testing.T) {
 		case <-time.After(1 * time.Second):
 		}
 
-		if err := translator.Update(podAddr); err != nil {
+		if err := translator.Update(podAddr1); err != nil {
 			t.Fatal("Expected update")
 		}
-		select {
-		case p := <-mockGetProfileServer.profilesReceived:
-			log.Debugf("Received update: %v", p)
-		case <-time.After(1 * time.Second):
+
+		p1 := <-mockGetProfileServer.profilesReceived
+		if !p1.GetOpaqueProtocol() {
+			t.Errorf("Expected port 8080 to be opaque")
+		}
+
+		if err := translator.Update(podAddr2); err != nil {
+			t.Fatal("Expected update")
+		}
+
+		p2 := <-mockGetProfileServer.profilesReceived
+		if !p2.GetOpaqueProtocol() {
+			t.Errorf("Expected port 8081 to be opaque")
 		}
 	})
 
@@ -97,7 +135,7 @@ func TestEndpointProfileTranslator(t *testing.T) {
 		// queue and we can test the overflow behavior.
 
 		for i := 0; i < queueCapacity/2; i++ {
-			if err := translator.Update(podAddr); err != nil {
+			if err := translator.Update(podAddr1); err != nil {
 				t.Fatal("Expected update")
 			}
 			select {
@@ -118,7 +156,7 @@ func TestEndpointProfileTranslator(t *testing.T) {
 
 		// The queue should be full and the next update should fail.
 		t.Logf("Queue length=%d capacity=%d", translator.queueLen(), queueCapacity)
-		if err := translator.Update(podAddr); err == nil {
+		if err := translator.Update(podAddr1); err == nil {
 			if !errors.Is(err, http.ErrServerClosed) {
 				t.Fatalf("Expected update to fail; queue=%d; capacity=%d", translator.queueLen(), queueCapacity)
 			}

controller/api/destination/watcher/workload_watcher.go

@@ -794,7 +794,7 @@ func GetAnnotatedOpaquePorts(pod *corev1.Pod, defaultPorts map[uint32]struct{})
 		return defaultPorts
 	}
 	opaquePorts := make(map[uint32]struct{})
-	namedPorts := util.GetNamedPorts(pod.Spec.Containers)
+	namedPorts := util.GetNamedPorts(append(pod.Spec.InitContainers, pod.Spec.Containers...))
 	if annotation != "" {
 		for _, pr := range util.ParseContainerOpaquePorts(annotation, namedPorts) {
 			for _, port := range pr.Ports() {