ingress doesn't need to check src ports (#72)

<!--  Thanks for sending a pull request!  Here are some tips for you:
1. Ensure you have added the unit tests for your changes.
2. Ensure you have included output of manual testing done in the Testing
section.
3. Ensure number of lines of code for new or existing methods are within
the reasonable limit.
4. Ensure your change works on existing clusters after upgrade.
-->
**What type of PR is this?**
fix a bug
<!--
Add one of the following:
bug
cleanup
documentation
feature
-->
bug
**Which issue does this PR fix**:
#71 

**What does this PR do / Why do we need it**:
The controller shouldn't loop through src pods to check if their ports
matching policies'.

**If an issue # is not available please add steps to reproduce and the
controller logs**:


**Testing done on this change**:
<!--
output of manual testing/integration tests results and also attach logs
showing the fix being resolved
-->
Unit tests are updated and functional tests were done

policy applied pod (dst)
```
ports:
        - containerPort: 80
          name: http
          protocol: TCP
        - containerPort: 443
          name: https
          protocol: TCP
```
src pod
```
ports:
    - name: src
      containerPort: 9999
      protocol: TCP
```
svc
```
spec:
  ports:
  - port: 80
    name: http
    targetPort: http
  - port: 443
    name: https
    targetPort: https
```
policyendpoint
```
spec:
  ingress:
  - cidr: 192.168.xx.xxx
    ports:
    - port: 80
      protocol: TCP
    - port: 443
      protocol: TCP
  podIsolation:
  - Ingress
```


**Automation added to e2e**:
<!--
List the e2e tests you added as part of this PR.
If no, create an issue with enhancement/testing label
-->
no
**Will this PR introduce any new dependencies?**:
<!--
e.g. new K8s API
-->
no
**Will this break upgrades or downgrades. Has updating a running cluster
been tested?**:

no
**Does this PR introduce any user-facing change?**:
<!--
If yes, a release note update is required:
Enter your extended release note in the block below. If the PR requires
additional actions
from users switching to the new release, include the string "action
required".
-->
no
```release-note

```

By submitting this pull request, I confirm that my contribution is made
under the terms of the Apache 2.0 license.

Author: haouc

pkg/resolvers/endpoints.go

@@ -191,15 +191,8 @@ func (r *defaultEndpointsResolver) resolveNetworkPeers(ctx context.Context, poli
 			namespaces = []string{policy.Namespace}
 		}
 
-		var portsToApply []policyinfo.Port
-		// populate the policy applied targets' ports
-		// only populate ports for Ingress and from network policy namespaces as destination ports
-		if policyType == networking.PolicyTypeIngress {
-			portsToApply = r.getIngressRulesPorts(ctx, policy.Namespace, &policy.Spec.PodSelector, ports)
-		}
-
 		for _, ns := range namespaces {
-			networkPeers = append(networkPeers, r.getMatchingPodAddresses(ctx, peer.PodSelector, ns, portsToApply, ports, policyType)...)
+			networkPeers = append(networkPeers, r.getMatchingPodAddresses(ctx, peer.PodSelector, ns, policy, ports, policyType)...)
 		}
 
 	}
@@ -317,9 +310,22 @@ func (r *defaultEndpointsResolver) resolveNamespaces(ctx context.Context, ls *me
 }
 
 func (r *defaultEndpointsResolver) getMatchingPodAddresses(ctx context.Context, ls *metav1.LabelSelector, namespace string,
-	policyPorts []policyinfo.Port, ports []networking.NetworkPolicyPort, rule networking.PolicyType) []policyinfo.EndpointInfo {
+	policy *networking.NetworkPolicy, rulePorts []networking.NetworkPolicyPort, policyType networking.PolicyType) []policyinfo.EndpointInfo {
 	var addresses []policyinfo.EndpointInfo
 
+	var portList []policyinfo.Port
+	// populate the policy applied targets' ports
+	// only populate ports for Ingress and from network policy namespaces as destination ports
+	if policyType == networking.PolicyTypeIngress {
+		portList = r.getIngressRulesPorts(ctx, policy.Namespace, &policy.Spec.PodSelector, rulePorts)
+		if len(rulePorts) != len(portList) && len(portList) == 0 {
+			r.logger.Info("Couldn't get matched port list from ingress of policy", "policy", types.NamespacedName{Name: policy.Name, Namespace: policy.Namespace}.String(),
+				"ingressPorts", rulePorts, "derivedPorts", portList)
+			return nil
+		}
+	}
+
+	// populate src pods for ingress and dst pods for egress
 	podList := &corev1.PodList{}
 	if err := r.k8sClient.List(ctx, podList, &client.ListOptions{
 		LabelSelector: r.createPodLabelSelector(ls),
@@ -329,25 +335,25 @@ func (r *defaultEndpointsResolver) getMatchingPodAddresses(ctx context.Context,
 		return nil
 	}
 	r.logger.V(1).Info("Got pods for label selector", "count", len(podList.Items), "selector", ls.String())
+
 	for _, pod := range podList.Items {
 		podIP := k8s.GetPodIP(&pod)
 		if len(podIP) == 0 {
 			r.logger.Info("pod IP not assigned yet", "pod", k8s.NamespacedName(&pod))
 			continue
 		}
-		portList := r.getPortList(pod, ports)
-		if len(ports) != len(portList) && len(portList) == 0 {
-			r.logger.Info("Couldn't get matched port list from the pod", "pod", k8s.NamespacedName(&pod), "expectedPorts", ports)
-			continue
+
+		if policyType == networking.PolicyTypeEgress {
+			portList = r.getPortList(pod, rulePorts)
+			if len(rulePorts) != len(portList) && len(portList) == 0 {
+				r.logger.Info("Couldn't get matched port list from the pod", "pod", k8s.NamespacedName(&pod), "expectedPorts", rulePorts)
+				continue
+			}
 		}
+
 		addresses = append(addresses, policyinfo.EndpointInfo{
-			CIDR: policyinfo.NetworkAddress(podIP),
-			Ports: func(policyType networking.PolicyType) []policyinfo.Port {
-				if policyType == networking.PolicyTypeIngress {
-					return policyPorts
-				}
-				return portList
-			}(rule),
+			CIDR:  policyinfo.NetworkAddress(podIP),
+			Ports: portList,
 		})
 	}
 

pkg/resolvers/endpoints_test.go

@@ -658,7 +658,7 @@ func TestEndpointsResolver_ResolveNetworkPeers(t *testing.T) {
 						{
 							ContainerPort: port80,
 							Protocol:      corev1.ProtocolTCP,
-							Name:          "test-port",
+							Name:          "src-port",
 						},
 					},
 				},
@@ -668,6 +668,7 @@ func TestEndpointsResolver_ResolveNetworkPeers(t *testing.T) {
 			PodIP: "1.0.0.1",
 		},
 	}
+
 	dstPodOne := corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "pod2",
@@ -681,7 +682,7 @@ func TestEndpointsResolver_ResolveNetworkPeers(t *testing.T) {
 						{
 							ContainerPort: port8080,
 							Protocol:      corev1.ProtocolTCP,
-							Name:          "test-port",
+							Name:          "dst-port",
 						},
 					},
 				},
@@ -715,6 +716,12 @@ func TestEndpointsResolver_ResolveNetworkPeers(t *testing.T) {
 		},
 	}
 
+	portsMap := map[string]int32{
+		"src-port": port80,
+		"dst-port": port8080,
+	}
+
+	// the policy is applied to dst namespace on dst pod
 	policy := &networking.NetworkPolicy{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "netpol",
@@ -737,7 +744,7 @@ func TestEndpointsResolver_ResolveNetworkPeers(t *testing.T) {
 					Ports: []networking.NetworkPolicyPort{
 						{
 							Protocol: &protocolTCP,
-							Port:     &intstr.IntOrString{Type: intstr.String, StrVal: "test-port"},
+							Port:     &intstr.IntOrString{Type: intstr.String, StrVal: "dst-port"},
 						},
 					},
 				},
@@ -756,7 +763,7 @@ func TestEndpointsResolver_ResolveNetworkPeers(t *testing.T) {
 					Ports: []networking.NetworkPolicyPort{
 						{
 							Protocol: &protocolTCP,
-							Port:     &intstr.IntOrString{Type: intstr.Int, IntVal: port8080},
+							Port:     &intstr.IntOrString{Type: intstr.String, StrVal: "src-port"},
 							EndPort:  &port9090,
 						},
 					},
@@ -798,6 +805,7 @@ func TestEndpointsResolver_ResolveNetworkPeers(t *testing.T) {
 			// getting ingress endpoint calls listing pods with dst NS first
 			mockClient.EXPECT().List(gomock.Any(), podList, gomock.Any()).DoAndReturn(
 				func(ctx context.Context, podList *corev1.PodList, opts ...client.ListOption) error {
+					podList.Items = []corev1.Pod{dstPodOne, dstPodTwo}
 					podList.Items = []corev1.Pod{dstPodOne, dstPodTwo}
 					return nil
 				},
@@ -820,7 +828,7 @@ func TestEndpointsResolver_ResolveNetworkPeers(t *testing.T) {
 
 		dstNS := corev1.Namespace{
 			ObjectMeta: metav1.ObjectMeta{
-				Name: "dst",
+				Name: "src",
 			},
 		}
 
@@ -834,6 +842,7 @@ func TestEndpointsResolver_ResolveNetworkPeers(t *testing.T) {
 			),
 			mockClient.EXPECT().List(gomock.Any(), podList, gomock.Any()).DoAndReturn(
 				func(ctx context.Context, podList *corev1.PodList, opts ...client.ListOption) error {
+					podList.Items = []corev1.Pod{dstPodOne, dstPodTwo}
 					podList.Items = []corev1.Pod{dstPodOne, dstPodTwo}
 					return nil
 				},
@@ -866,16 +875,23 @@ func TestEndpointsResolver_ResolveNetworkPeers(t *testing.T) {
 		}
 	}
 
+	// the policy is applied to dst namespace
+	// the ingress should have cidr from src pod and ports from dst pod
+	// the egress should have cidr from src pod and ports from src pod
 	for _, ingPE := range ingressEndpoints {
 		assert.Equal(t, srcPod.Status.PodIP, string(ingPE.CIDR))
 		assert.Equal(t, dstPodOne.Spec.Containers[0].Ports[0].ContainerPort, *ingPE.Ports[0].Port)
 		assert.Equal(t, 1, len(ingPE.Ports))
+		assert.Equal(t, dstPodOne.Spec.Containers[0].Ports[0].ContainerPort, *ingPE.Ports[0].Port)
+		assert.Equal(t, 1, len(ingPE.Ports))
 	}
 
 	for _, egPE := range egressEndpoints {
 		assert.True(t, string(egPE.CIDR) == dstPodOne.Status.PodIP || string(egPE.CIDR) == dstPodTwo.Status.PodIP)
 		assert.Equal(t, dstPodOne.Spec.Containers[0].Ports[0].ContainerPort, *egPE.Ports[0].Port)
-		assert.Equal(t, policy.Spec.Egress[0].Ports[0].Port.IntVal, *egPE.Ports[0].Port)
+		assert.Equal(t, srcPod.Status.PodIP, string(egPE.CIDR))
+		assert.Equal(t, srcPod.Spec.Containers[0].Ports[0].ContainerPort, *egPE.Ports[0].Port)
+		assert.Equal(t, portsMap[policy.Spec.Egress[0].Ports[0].Port.StrVal], *egPE.Ports[0].Port)
 		assert.Equal(t, *policy.Spec.Egress[0].Ports[0].EndPort, *egPE.Ports[0].EndPort)
 	}
 }