fix sync period handling, headless svc (#15)

Fix policy event handler to refer resourceRevision
do not include headless service in the endpoints
policyendpoints manager cleanup

Author: kishorj

internal/eventhandlers/policy.go

@@ -61,7 +61,7 @@ func (h *enqueueRequestForPolicyEvent) Update(_ context.Context, e event.UpdateE
 	newPolicy := e.ObjectNew.(*networking.NetworkPolicy)
 
 	h.logger.V(1).Info("Handling update event", "policy", k8s.NamespacedName(newPolicy))
-	if oldPolicy.Generation != newPolicy.Generation && equality.Semantic.DeepEqual(oldPolicy.Spec, newPolicy.Spec) &&
+	if !equality.Semantic.DeepEqual(newPolicy.ResourceVersion, oldPolicy.ResourceVersion) && equality.Semantic.DeepEqual(oldPolicy.Spec, newPolicy.Spec) &&
 		equality.Semantic.DeepEqual(oldPolicy.DeletionTimestamp.IsZero(), newPolicy.DeletionTimestamp.IsZero()) {
 		return
 	}

pkg/k8s/service_utils.go

@@ -42,3 +42,11 @@ func LookupListenPortFromPodSpec(svc *corev1.Service, pod *corev1.Pod, port ints
 	}
 	return 0, errors.Errorf("unable to find listener port for port %s on service %s", port.String(), NamespacedName(svc))
 }
+
+// IsServiceHeadless returns true if the service is headless
+func IsServiceHeadless(svc *corev1.Service) bool {
+	if svc.Spec.ClusterIP == "" || svc.Spec.ClusterIP == "None" {
+		return true
+	}
+	return false
+}

pkg/k8s/service_utils_test.go

@@ -305,3 +305,42 @@ func Test_LookupListenPortFromPodSpec(t *testing.T) {
 		})
 	}
 }
+
+func Test_IsServiceHeadless(t *testing.T) {
+	tests := []struct {
+		name string
+		svc  *corev1.Service
+		want bool
+	}{
+		{
+			name: "headless service",
+			svc: &corev1.Service{
+				Spec: corev1.ServiceSpec{
+					ClusterIP: corev1.ClusterIPNone,
+				},
+			},
+			want: true,
+		},
+		{
+			name: "empty IP",
+			svc: &corev1.Service{
+				Spec: corev1.ServiceSpec{},
+			},
+			want: true,
+		},
+		{
+			name: "some cluster IP",
+			svc: &corev1.Service{
+				Spec: corev1.ServiceSpec{
+					ClusterIP: "10.100.0.209",
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, IsServiceHeadless(tt.svc))
+		})
+	}
+
+}

pkg/policyendpoints/manager.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"golang.org/x/exp/maps"
 	"strconv"
 
 	"github.com/go-logr/logr"
@@ -67,7 +68,7 @@ func (m *policyEndpointsManager) Reconcile(ctx context.Context, policy *networki
 	if err != nil {
 		return err
 	}
-	m.logger.V(1).Info("Got policy endpoints lists", "create", len(createList), "update", len(updateList), "delete", len(deleteList))
+	m.logger.Info("Got policy endpoints lists", "create", len(createList), "update", len(updateList), "delete", len(deleteList))
 	for _, policyEndpoint := range createList {
 		if err := m.k8sClient.Create(ctx, &policyEndpoint); err != nil {
 			return err
@@ -139,31 +140,28 @@ func (m *policyEndpointsManager) computePolicyEndpoints(policy *networking.Netwo
 	// Go over the existing endpoints, and remove entries that are no longer needed
 	var modifiedEndpoints []policyinfo.PolicyEndpoint
 	var potentialDeletes []policyinfo.PolicyEndpoint
-	usedIngressRuleKeys := sets.Set[string]{}
-	usedEgressRulesKeys := sets.Set[string]{}
-	usedPodEndpoints := sets.Set[policyinfo.PodEndpoint]{}
 	for i := range existingPolicyEndpoints {
 		ingEndpointList := make([]policyinfo.EndpointInfo, 0, len(existingPolicyEndpoints[i].Spec.Ingress))
 		for _, ingRule := range existingPolicyEndpoints[i].Spec.Ingress {
 			ruleKey := m.getEndpointInfoKey(ingRule)
 			if _, exists := ingressEndpointsMap[ruleKey]; exists {
 				ingEndpointList = append(ingEndpointList, ingRule)
-				usedIngressRuleKeys.Insert(ruleKey)
+				delete(ingressEndpointsMap, ruleKey)
 			}
 		}
 		egEndpointList := make([]policyinfo.EndpointInfo, 0, len(existingPolicyEndpoints[i].Spec.Egress))
 		for _, egRule := range existingPolicyEndpoints[i].Spec.Egress {
 			ruleKey := m.getEndpointInfoKey(egRule)
 			if _, exists := egressEndpointsMap[ruleKey]; exists {
 				egEndpointList = append(egEndpointList, egRule)
-				usedEgressRulesKeys.Insert(ruleKey)
+				delete(egressEndpointsMap, ruleKey)
 			}
 		}
 		podSelectorEndpointList := make([]policyinfo.PodEndpoint, 0, len(existingPolicyEndpoints[i].Spec.PodSelectorEndpoints))
 		for _, ps := range existingPolicyEndpoints[i].Spec.PodSelectorEndpoints {
 			if podSelectorEndpointSet.Has(ps) {
 				podSelectorEndpointList = append(podSelectorEndpointList, ps)
-				usedPodEndpoints.Insert(ps)
+				podSelectorEndpointSet.Delete(ps)
 			}
 		}
 		policyEndpointChanged := false
@@ -188,22 +186,7 @@ func (m *policyEndpointsManager) computePolicyEndpoints(policy *networking.Netwo
 		}
 	}
 
-	remainingIngressRuleKeys := sets.Set[string]{}
-	remainingEgressRulesKeys := sets.Set[string]{}
-	remainingPodEndpoints := podSelectorEndpointSet.Difference(usedPodEndpoints)
-
-	for key := range ingressEndpointsMap {
-		if !usedIngressRuleKeys.Has(key) {
-			remainingIngressRuleKeys.Insert(key)
-		}
-	}
-	for key := range egressEndpointsMap {
-		if !usedEgressRulesKeys.Has(key) {
-			remainingEgressRulesKeys.Insert(key)
-		}
-	}
-
-	ingressRuleChunks := lo.Chunk(remainingIngressRuleKeys.UnsortedList(), m.endpointChunkSize)
+	ingressRuleChunks := lo.Chunk(maps.Keys(ingressEndpointsMap), m.endpointChunkSize)
 	doNotDelete := sets.Set[types.NamespacedName]{}
 	for _, chunk := range ingressRuleChunks {
 		// check in the existing lists if chunk fits, otherwise allocate a new ep
@@ -228,7 +211,7 @@ func (m *policyEndpointsManager) computePolicyEndpoints(policy *networking.Netwo
 		createPolicyEndpoints = append(createPolicyEndpoints, newEP)
 	}
 
-	egressRuleChunks := lo.Chunk(remainingEgressRulesKeys.UnsortedList(), m.endpointChunkSize)
+	egressRuleChunks := lo.Chunk(maps.Keys(egressEndpointsMap), m.endpointChunkSize)
 	for _, chunk := range egressRuleChunks {
 		// check in the existing to-update/to-delete list if chunk fits, otherwise allocate a new ep
 		var assigned bool
@@ -251,7 +234,7 @@ func (m *policyEndpointsManager) computePolicyEndpoints(policy *networking.Netwo
 		newEP := m.newPolicyEndpoint(policy, nil, m.getListOfEndpointInfoFromHash(chunk, egressEndpointsMap), nil)
 		createPolicyEndpoints = append(createPolicyEndpoints, newEP)
 	}
-	podEndpointChunks := lo.Chunk(remainingPodEndpoints.UnsortedList(), m.endpointChunkSize)
+	podEndpointChunks := lo.Chunk(podSelectorEndpointSet.UnsortedList(), m.endpointChunkSize)
 	for _, chunk := range podEndpointChunks {
 		var assigned bool
 		for _, sliceToCheck := range [][]policyinfo.PolicyEndpoint{createPolicyEndpoints, modifiedEndpoints, potentialDeletes} {

pkg/policyendpoints/manager_test.go

@@ -372,6 +372,76 @@ func Test_policyEndpointsManager_computePolicyEndpoints(t *testing.T) {
 				updateCount: 2,
 			},
 		},
+		{
+			name: "cleanup endpoints with same entries",
+			fields: fields{
+				endpointChunkSize: 3,
+			},
+			args: args{
+				policy: &networking.NetworkPolicy{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "policy-namespace",
+						Name:      "policy-name",
+					},
+					Spec: networking.NetworkPolicySpec{},
+				},
+				policyEndpoints: []policyinfo.PolicyEndpoint{
+					{
+						ObjectMeta: metav1.ObjectMeta{
+							Namespace: "policy-namespace",
+							Name:      "policy-name-1",
+						},
+						Spec: policyinfo.PolicyEndpointSpec{
+							Ingress: getEPInfoHelper([]policyinfo.NetworkAddress{"1.2.3.4", "1.2.3.5"}, nil, 3),
+							Egress:  getEPInfoHelper([]policyinfo.NetworkAddress{"2.2.0.0/16", "2.3.0.0/16"}, []policyinfo.NetworkAddress{"2.2.3.4"}, 1),
+							PodSelectorEndpoints: []policyinfo.PodEndpoint{
+								{
+									Name: "pod1",
+								},
+								{
+									Name: "pod2",
+								},
+							},
+						},
+					},
+					{
+						ObjectMeta: metav1.ObjectMeta{
+							Namespace: "policy-namespace",
+							Name:      "policy-name-dup",
+						},
+						Spec: policyinfo.PolicyEndpointSpec{
+							Ingress: getEPInfoHelper([]policyinfo.NetworkAddress{"1.2.3.5", "1.2.3.4"}, nil, 3),
+							Egress:  getEPInfoHelper([]policyinfo.NetworkAddress{"2.3.0.0/16", "2.2.0.0/16"}, []policyinfo.NetworkAddress{"2.2.3.4"}, 1),
+							PodSelectorEndpoints: []policyinfo.PodEndpoint{
+								{
+									Name: "pod1",
+								},
+								{
+									Name: "pod2",
+								},
+							},
+						},
+					},
+				},
+				ingressRules: getEPInfoHelper([]policyinfo.NetworkAddress{"1.2.3.4", "1.2.3.5"}, nil, 3),
+				egressRules:  getEPInfoHelper([]policyinfo.NetworkAddress{"2.2.0.0/16", "2.3.0.0/16"}, []policyinfo.NetworkAddress{"2.2.3.4"}, 1),
+				podselectorEndpoints: []policyinfo.PodEndpoint{
+					{
+						Name: "pod1",
+					},
+					{
+						Name: "pod2",
+					},
+				},
+				epValidator: func(_ *networking.NetworkPolicy, _ *policyinfo.PolicyEndpoint) bool {
+					return true
+				},
+			},
+			want: want{
+				deleteCount: 1,
+				updateCount: 1,
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

pkg/resolvers/endpoints.go

@@ -339,7 +339,7 @@ func (r *defaultEndpointsResolver) getMatchingServiceClusterIPs(ctx context.Cont
 		return nil
 	}
 	for _, svc := range svcList.Items {
-		if !svcSelector.Matches(labels.Set(svc.Spec.Selector)) {
+		if k8s.IsServiceHeadless(&svc) || !svcSelector.Matches(labels.Set(svc.Spec.Selector)) {
 			continue
 		}
 		var portList []policyinfo.Port

pkg/resolvers/endpoints_test.go

@@ -372,6 +372,36 @@ func TestEndpointsResolver_Resolve(t *testing.T) {
 				{PodIP: "1.0.0.3", Name: "pod3", Namespace: "ns"},
 			},
 		},
+		{
+			name: "exclude headless service",
+			args: args{
+				netpol: egressPolicy,
+				podListCalls: []podListCall{
+					{
+						pods: []corev1.Pod{pod2, podNoIP, pod3},
+					},
+				},
+				serviceListCalls: []serviceListCall{
+					{
+						services: []corev1.Service{
+							{
+								Spec: corev1.ServiceSpec{
+									ClusterIP: "None",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantEgressEndpoints: []policyinfo.EndpointInfo{
+				{CIDR: "1.0.0.2"},
+				{CIDR: "1.0.0.3"},
+			},
+			wantPodEndpoints: []policyinfo.PodEndpoint{
+				{PodIP: "1.0.0.2", Name: "pod2", Namespace: "ns"},
+				{PodIP: "1.0.0.3", Name: "pod3", Namespace: "ns"},
+			},
+		},
 		{
 			name: "resolve network peers, ingress/egress",
 			args: args{

pkg/resolvers/policies_for_service.go

@@ -16,7 +16,7 @@ import (
 
 // getReferredPoliciesForService returns the list of policies that refer to the service.
 func (r *defaultPolicyReferenceResolver) getReferredPoliciesForService(ctx context.Context, svc, svcOld *corev1.Service) ([]networking.NetworkPolicy, error) {
-	if svc.Spec.ClusterIP == "" || svc.Spec.ClusterIP == "None" {
+	if k8s.IsServiceHeadless(svc) {
 		r.logger.V(1).Info("Ignoring headless service", "svc", k8s.NamespacedName(svc))
 		return nil, nil
 	}