combine endpoints based on cidr (#80)

pkg/policyendpoints/manager.go

@@ -167,7 +167,37 @@ func (m *policyEndpointsManager) computePolicyEndpoints(policy *networking.Netwo
 		}
 	}
 
-	return createPolicyEndpoints, updatePolicyEndpoints, deletePolicyEndpoints, nil
+	return m.processPolicyEndpoints(createPolicyEndpoints), m.processPolicyEndpoints(updatePolicyEndpoints), deletePolicyEndpoints, nil
+}
+
+func (m *policyEndpointsManager) processPolicyEndpoints(pes []policyinfo.PolicyEndpoint) []policyinfo.PolicyEndpoint {
+	var newPEs []policyinfo.PolicyEndpoint
+	for _, pe := range pes {
+		pe.Spec.Ingress = combineRulesEndpoints(pe.Spec.Ingress)
+		pe.Spec.Egress = combineRulesEndpoints(pe.Spec.Egress)
+		newPEs = append(newPEs, pe)
+	}
+	m.logger.Info("manager processed policy endpoints to consolidate rules", "preLen", len(pes), "postLen", len(newPEs), "newPEs", newPEs)
+	return newPEs
+}
+
+// the controller should consolidate the ingress and egress endpoints and put entries to one CIDR if they belong to a same CIDR
+func combineRulesEndpoints(ingressEndpoints []policyinfo.EndpointInfo) []policyinfo.EndpointInfo {
+	combinedMap := make(map[string]policyinfo.EndpointInfo)
+	for _, iep := range ingressEndpoints {
+		if _, ok := combinedMap[string(iep.CIDR)]; ok {
+			tempIEP := combinedMap[string(iep.CIDR)]
+			tempIEP.Ports = append(combinedMap[string(iep.CIDR)].Ports, iep.Ports...)
+			tempIEP.Except = append(combinedMap[string(iep.CIDR)].Except, iep.Except...)
+			combinedMap[string(iep.CIDR)] = tempIEP
+		} else {
+			combinedMap[string(iep.CIDR)] = iep
+		}
+	}
+	if len(combinedMap) > 0 {
+		return maps.Values(combinedMap)
+	}
+	return nil
 }
 
 func (m *policyEndpointsManager) newPolicyEndpoint(policy *networking.NetworkPolicy,

pkg/policyendpoints/manager_test.go

@@ -9,6 +9,7 @@ import (
 	networking "k8s.io/api/networking/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 
 	policyinfo "github.com/aws/amazon-network-policy-controller-k8s/api/v1alpha1"
 )
@@ -494,3 +495,61 @@ func Test_policyEndpointsManager_computePolicyEndpoints(t *testing.T) {
 		})
 	}
 }
+
+func Test_processPolicyEndpoints(t *testing.T) {
+	m := &policyEndpointsManager{
+		logger: zap.New(),
+	}
+
+	p80 := int32(80)
+	p8080 := int32(8080)
+	pTCP := corev1.ProtocolTCP
+	pUDP := corev1.ProtocolUDP
+
+	pes := m.processPolicyEndpoints([]policyinfo.PolicyEndpoint{
+		{
+			Spec: policyinfo.PolicyEndpointSpec{
+				Ingress: []policyinfo.EndpointInfo{
+					{
+						CIDR: "1.2.3.4",
+						Ports: []policyinfo.Port{
+							{Port: &p80, Protocol: &pTCP},
+						},
+					},
+					{
+						CIDR: "1.2.3.4",
+						Ports: []policyinfo.Port{
+							{Port: &p8080, Protocol: &pTCP},
+						},
+					},
+					{
+						CIDR: "1.2.3.4",
+						Ports: []policyinfo.Port{
+							{Protocol: &pUDP},
+						},
+					},
+				},
+				Egress: []policyinfo.EndpointInfo{
+					{
+						CIDR: "1.2.3.5",
+						Ports: []policyinfo.Port{
+							{Port: &p80, Protocol: &pTCP},
+						},
+					},
+					{
+						CIDR: "1.2.3.5",
+						Ports: []policyinfo.Port{
+							{Port: &p8080, Protocol: &pTCP},
+						},
+					},
+				},
+			},
+		},
+	})
+	assert.Equal(t, 1, len(pes[0].Spec.Ingress))
+	assert.Equal(t, 1, len(pes[0].Spec.Egress))
+	assert.Equal(t, "1.2.3.4", string(pes[0].Spec.Ingress[0].CIDR))
+	assert.Equal(t, "1.2.3.5", string(pes[0].Spec.Egress[0].CIDR))
+	assert.Equal(t, 3, len(pes[0].Spec.Ingress[0].Ports))
+	assert.Equal(t, 2, len(pes[0].Spec.Egress[0].Ports))
+}