Follows the HostPort mapping when a request for a pod comes in on node network (#9819)

Maps the request port to the container's port if the request comes in from the node network and has a hostPort mapping.

Problem:

When a request for a container comes in from the node network, the node port is used ignoring the hostPort mapping.

Solution:

When a request is seen coming from the node network, get the container Port from the Spec.

Validation:

Fixed an existing unit test and wrote a new one driving GetProfile specifically.

Fixes #9677 

Signed-off-by: Steve Jenson <stevej@buoyant.io>

controller/api/destination/server.go

@@ -218,7 +218,7 @@ func (s *server) GetProfile(dest *pb.GetDestination, stream pb.Destination_GetPr
 			var address watcher.Address
 			var endpoint *pb.WeightedAddr
 			if pod != nil {
-				address, err = s.createAddress(pod, port)
+				address, err = s.createAddress(pod, host, port)
 				if err != nil {
 					return fmt.Errorf("failed to create address: %w", err)
 				}
@@ -364,8 +364,47 @@ func (s *server) GetProfile(dest *pb.GetDestination, stream pb.Destination_GetPr
 	return nil
 }
 
-func (s *server) createAddress(pod *corev1.Pod, port uint32) (watcher.Address, error) {
+// getPortForPod returns the port that a `pod` is listening on.
+//
+// Proxies usually receive traffic targeting `podIp:containerPort`.
+// However, they may be receiving traffic on `nodeIp:nodePort`. In this
+// case, we convert the port to the containerPort for discovery. In k8s parlance,
+// this is the 'HostPort' mapping.
+func (s *server) getPortForPod(pod *corev1.Pod, targetIP string, port uint32) (uint32, error) {
+	if pod == nil {
+		return port, fmt.Errorf("getPortForPod passed a nil pod")
+	}
+
+	if net.ParseIP(targetIP) == nil {
+		return port, fmt.Errorf("failed to parse hostIP into net.IP: %s", targetIP)
+	}
+
+	if containsIP(pod.Status.PodIPs, targetIP) {
+		return port, nil
+	}
+
+	if targetIP == pod.Status.HostIP {
+		for _, container := range pod.Spec.Containers {
+			for _, containerPort := range container.Ports {
+				if uint32(containerPort.HostPort) == port {
+					return uint32(containerPort.ContainerPort), nil
+				}
+			}
+		}
+	}
+
+	s.log.Warnf("unable to find container port as host (%s) matches neither PodIP nor HostIP (%s)", targetIP, pod)
+	return port, nil
+}
+
+func (s *server) createAddress(pod *corev1.Pod, targetIP string, port uint32) (watcher.Address, error) {
 	ownerKind, ownerName := s.k8sAPI.GetOwnerKindAndName(context.Background(), pod, true)
+
+	port, err := s.getPortForPod(pod, targetIP, port)
+	if err != nil {
+		return watcher.Address{}, fmt.Errorf("failed to find Port for Pod: %w", err)
+	}
+
 	address := watcher.Address{
 		IP:        pod.Status.PodIP,
 		Port:      port,
@@ -373,8 +412,8 @@ func (s *server) createAddress(pod *corev1.Pod, port uint32) (watcher.Address, e
 		OwnerName: ownerName,
 		OwnerKind: ownerKind,
 	}
-	err := watcher.SetToServerProtocol(s.k8sAPI, &address, port)
-	if err != nil {
+
+	if err := watcher.SetToServerProtocol(s.k8sAPI, &address, port); err != nil {
 		return watcher.Address{}, fmt.Errorf("failed to set address OpaqueProtocol: %w", err)
 	}
 	return address, nil
@@ -446,7 +485,7 @@ func (s *server) getEndpointByHostname(k8sAPI *k8s.API, hostname string, svcID w
 					if err != nil {
 						return nil, err
 					}
-					address, err := s.createAddress(pod, port)
+					address, err := s.createAddress(pod, addr.IP, port)
 					if err != nil {
 						return nil, err
 					}
@@ -677,3 +716,14 @@ func getPodSkippedInboundPortsAnnotations(pod *corev1.Pod) (map[uint32]struct{},
 
 	return util.ParsePorts(annotation)
 }
+
+// Given a list of PodIP, determine is `targetIP` is a member
+func containsIP(podIPs []corev1.PodIP, targetIP string) bool {
+	for _, ip := range podIPs {
+		if ip.String() == targetIP {
+			return true
+		}
+	}
+
+	return false
+}

controller/api/destination/server_test.go

@@ -23,6 +23,7 @@ const clusterIP = "172.17.12.0"
 const clusterIPOpaque = "172.17.12.1"
 const podIP1 = "172.17.0.12"
 const podIP2 = "172.17.0.13"
+const podIP3 = "172.17.0.17"
 const podIPOpaque = "172.17.0.14"
 const podIPSkipped = "172.17.0.15"
 const podIPPolicy = "172.17.0.16"
@@ -796,6 +797,30 @@ func toAddress(path string, port uint32) (*net.TcpAddress, error) {
 	}, nil
 }
 
+func TestHostPortMapping(t *testing.T) {
+	hostPort := uint32(7777)
+	containerPort := uint32(80)
+	server := makeServer(t)
+
+	pod, err := getPodByIP(server.k8sAPI, externalIP, hostPort, server.log)
+	if err != nil {
+		t.Fatalf("error retrieving pod by external IP %s", err)
+	}
+
+	address, err := server.createAddress(pod, externalIP, hostPort)
+	if err != nil {
+		t.Fatalf("error calling createAddress() %s", err)
+	}
+
+	if address.IP != podIP3 {
+		t.Fatalf("expected podIP (%s), received other IP (%s)", podIP3, address.IP)
+	}
+
+	if address.Port != containerPort {
+		t.Fatalf("expected containerPort (%d) but received port (%d) instead", containerPort, address.Port)
+	}
+}
+
 func TestIpWatcherGetSvcID(t *testing.T) {
 	name := "service"
 	namespace := "test"
@@ -854,8 +879,8 @@ spec:
 func TestIpWatcherGetPod(t *testing.T) {
 	podIP := "10.255.0.1"
 	hostIP := "172.0.0.1"
-	var hostPort1 uint32 = 12345
-	var hostPort2 uint32 = 12346
+	var hostPort1 uint32 = 22345
+	var hostPort2 uint32 = 22346
 	expectedPodName := "hostPortPod1"
 	k8sConfigs := []string{`
 apiVersion: v1
@@ -870,13 +895,13 @@ spec:
     ports:
     - containerPort: 12345
       hostIP: 172.0.0.1
-      hostPort: 12345
+      hostPort: 22345
   - image: test
     name: hostPortContainer2
     ports:
     - containerPort: 12346
       hostIP: 172.0.0.1
-      hostPort: 12346
+      hostPort: 22346
 status:
   phase: Running
   podIP: 10.255.0.1

controller/api/destination/test_util.go

@@ -314,6 +314,26 @@ spec:
   proxyProtocol: opaque`,
 	}
 
+	hostPortMapping := []string{
+		`
+kind: Pod
+apiVersion: v1
+metadata:
+  name: hostport-mapping
+status:
+  phase: Running
+  hostIP: 192.168.1.20
+  podIP: 172.17.0.17
+spec:
+  containers:
+  - name: nginx
+    image: nginx
+    ports:
+    - containerPort: 80
+      hostPort: 7777
+      name: nginx-7777`,
+	}
+
 	res := append(meshedPodResources, clientSP...)
 	res = append(res, unmeshedPod)
 	res = append(res, meshedOpaquePodResources...)
@@ -321,6 +341,7 @@ spec:
 	res = append(res, meshedSkippedPodResource...)
 	res = append(res, meshedStatefulSetPodResource...)
 	res = append(res, policyResources...)
+	res = append(res, hostPortMapping...)
 	k8sAPI, err := k8s.NewFakeAPI(res...)
 	if err != nil {
 		t.Fatalf("NewFakeAPI returned an error: %s", err)