fix a duplicated ingress port issue (#69)

<!--  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?**

<!--
Add one of the following:
bug
cleanup
documentation
feature
-->
bug
**Which issue does this PR fix**:
This PR is fixing the duplicated ingress ports issue when scale dst
pods.

**What does this PR do / Why do we need it**:
The bug can cause duplicated ports breaching the limit and fails ports
beyond the limit.

**If an issue # is not available please add steps to reproduce and the
controller logs**:
1, create two deployments (simple nginx) with replicas = 1
2, create NPs for one deployment pods with ingress from the other
deployment pods
3, scale the dst pods to a larger number
4, print out the policyendpoint and check if the ingress has duplicated
ports with a same cidr in one entry

**Testing done on this change**:
<!--
output of manual testing/integration tests results and also attach logs
showing the fix being resolved
-->
Before this fix
```
  ingress:
  - cidr: 192.168.163.178
    ports:
    - port: 80
      protocol: TCP
    - port: 80
      protocol: TCP
    - port: 80
      protocol: TCP
  podIsolation:
  - Ingress
  - Egress
```
after this fix
```
  ingress:
  - cidr: 192.168.51.76
    ports:
    - port: 80
      protocol: TCP
  podIsolation:
  - Ingress
  - Egress
```
with more than one port
```
  ingress:
  - cidr: 192.168.7.80
    ports:
    - port: 80
      protocol: TCP
    - port: 8080
      protocol: TCP
  podIsolation:
  - Ingress
  - Egress
```

Updated Unit Tests
Before the fix
```
Error Trace:	/local/home/zhuhz/go/src/amazon-network-policy-controller-k8s/pkg/resolvers/endpoints_test.go:872
        	Error:      	Not equal: 
        	            	expected: 1
        	            	actual  : 2
        	Test:       	TestEndpointsResolver_ResolveNetworkPeers
```
After the fix
```
=== RUN   TestEndpointsResolver_ResolveNetworkPeers
--- PASS: TestEndpointsResolver_ResolveNetworkPeers (0.00s)
PASS
ok      github.com/aws/amazon-network-policy-controller-k8s/pkg/resolvers       (cached)
```

**Automation added to e2e**:
<!--
List the e2e tests you added as part of this PR.
If no, create an issue with enhancement/testing label
-->

**Will this PR introduce any new dependencies?**:
<!--
e.g. new K8s API
-->

**Will this break upgrades or downgrades. Has updating a running cluster
been tested?**:


**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".
-->

```release-note

```

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

Author: achevuru

pkg/resolvers/endpoints.go

@@ -2,15 +2,19 @@ package resolvers
 
 import (
 	"context"
+	"fmt"
+	"strconv"
 
 	policyinfo "github.com/aws/amazon-network-policy-controller-k8s/api/v1alpha1"
 	"github.com/aws/amazon-network-policy-controller-k8s/pkg/k8s"
 	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
+	"golang.org/x/exp/maps"
 	corev1 "k8s.io/api/core/v1"
 	networking "k8s.io/api/networking/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -216,10 +220,14 @@ func (r *defaultEndpointsResolver) getIngressRulesPorts(ctx context.Context, pol
 	var portList []policyinfo.Port
 	for _, pod := range podList.Items {
 		portList = append(portList, r.getPortList(pod, ports)...)
-		r.logger.Info("Got ingress port", "port", portList, "pod", pod)
+		r.logger.Info("Got ingress port from pod", "pod", types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}.String())
 	}
 
-	return portList
+	// since we pull ports from dst pods, we should deduplicate them
+	dedupedPorts := dedupPorts(portList)
+	r.logger.Info("Got ingress ports from dst pods", "port", dedupedPorts)
+
+	return dedupedPorts
 }
 
 func (r *defaultEndpointsResolver) getPortList(pod corev1.Pod, ports []networking.NetworkPolicyPort) []policyinfo.Port {
@@ -455,3 +463,25 @@ func (r *defaultEndpointsResolver) getMatchingServicePort(ctx context.Context, s
 	}
 	return 0, errors.Errorf("unable to find matching service listen port %s for service %s", port.String(), k8s.NamespacedName(svc))
 }
+
+func dedupPorts(policyPorts []policyinfo.Port) []policyinfo.Port {
+	ports := make(map[string]policyinfo.Port)
+	for _, port := range policyPorts {
+		prot, p, ep := "", "", ""
+		if port.Protocol != nil {
+			prot = string(*port.Protocol)
+		}
+		if port.Port != nil {
+			p = strconv.FormatInt(int64(*port.Port), 10)
+		}
+		if port.EndPort != nil {
+			ep = strconv.FormatInt(int64(*port.EndPort), 10)
+		}
+
+		ports[fmt.Sprintf("%s@%s@%s", prot, p, ep)] = port
+	}
+	if len(ports) > 0 {
+		return maps.Values(ports)
+	}
+	return nil
+}

pkg/resolvers/endpoints_test.go

@@ -668,7 +668,7 @@ func TestEndpointsResolver_ResolveNetworkPeers(t *testing.T) {
 			PodIP: "1.0.0.1",
 		},
 	}
-	dstPod := corev1.Pod{
+	dstPodOne := corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "pod2",
 			Namespace: "dst",
@@ -691,6 +691,29 @@ func TestEndpointsResolver_ResolveNetworkPeers(t *testing.T) {
 			PodIP: "1.0.0.2",
 		},
 	}
+	dstPodTwo := corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "pod3",
+			Namespace: "dst",
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				{
+					Name: "pod3",
+					Ports: []corev1.ContainerPort{
+						{
+							ContainerPort: port8080,
+							Protocol:      corev1.ProtocolTCP,
+							Name:          "test-port",
+						},
+					},
+				},
+			},
+		},
+		Status: corev1.PodStatus{
+			PodIP: "1.0.0.3",
+		},
+	}
 
 	policy := &networking.NetworkPolicy{
 		ObjectMeta: metav1.ObjectMeta{
@@ -775,7 +798,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{dstPod}
+					podList.Items = []corev1.Pod{dstPodOne, dstPodTwo}
 					return nil
 				},
 			),
@@ -811,7 +834,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{dstPod}
+					podList.Items = []corev1.Pod{dstPodOne, dstPodTwo}
 					return nil
 				},
 			),
@@ -845,12 +868,13 @@ func TestEndpointsResolver_ResolveNetworkPeers(t *testing.T) {
 
 	for _, ingPE := range ingressEndpoints {
 		assert.Equal(t, srcPod.Status.PodIP, string(ingPE.CIDR))
-		assert.Equal(t, dstPod.Spec.Containers[0].Ports[0].ContainerPort, *ingPE.Ports[0].Port)
+		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.Equal(t, dstPod.Status.PodIP, string(egPE.CIDR))
-		assert.Equal(t, dstPod.Spec.Containers[0].Ports[0].ContainerPort, *egPE.Ports[0].Port)
+		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, *policy.Spec.Egress[0].Ports[0].EndPort, *egPE.Ports[0].EndPort)
 	}