endpoints resolver and manager (#11)

* endpoints resolver and manager
* fix version info
* increase log verbosity
* refactor function names
* setup field indexer for policy reference
* update manager UTs
* do not ignore not found error for updates
* fix periodic reconcile
* comment service cluster IP resolver function
* add endpoint resolver UTs

Author: kishorj

cmd/main.go

@@ -42,7 +42,7 @@ import (
 	"github.com/aws/amazon-network-policy-controller-k8s/pkg/k8s"
 	"github.com/aws/amazon-network-policy-controller-k8s/pkg/policyendpoints"
 	"github.com/aws/amazon-network-policy-controller-k8s/pkg/utils/configmap"
-	"github.com/aws/amazon-network-policy-controller-k8s/version"
+	"github.com/aws/amazon-network-policy-controller-k8s/pkg/version"
 	//+kubebuilder:scaffold:imports
 )
 

internal/controllers/policy_controller.go

@@ -18,6 +18,7 @@ package controllers
 
 import (
 	"context"
+	policyinfo "github.com/aws/amazon-network-policy-controller-k8s/api/v1alpha1"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -85,7 +86,10 @@ func (r *policyReconciler) Reconcile(ctx context.Context, request reconcile.Requ
 	return ctrl.Result{}, r.reconcile(ctx, request)
 }
 
-func (r *policyReconciler) SetupWithManager(_ context.Context, mgr ctrl.Manager) error {
+func (r *policyReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error {
+	if err := r.setupIndexes(ctx, mgr.GetFieldIndexer()); err != nil {
+		return err
+	}
 	policyEventChan := make(chan event.GenericEvent)
 	policyEventHandler := eventhandlers.NewEnqueueRequestForPolicyEvent(r.policyTracker, r.podUpdateBatchPeriodDuration,
 		r.logger.WithName("eventHandler").WithName("policy"))
@@ -139,3 +143,11 @@ func (r *policyReconciler) cleanupPolicy(ctx context.Context, policy *networking
 	}
 	return nil
 }
+
+func (r *policyReconciler) setupIndexes(ctx context.Context, fieldIndexer client.FieldIndexer) error {
+	if err := fieldIndexer.IndexField(ctx, &policyinfo.PolicyEndpoint{}, policyendpoints.IndexKeyPolicyReferenceName,
+		policyendpoints.IndexFunctionPolicyReferenceName); err != nil {
+		return err
+	}
+	return nil
+}

internal/eventhandlers/namespace.go

@@ -70,7 +70,7 @@ func (h *enqueueRequestForNamespaceEvent) enqueueReferredPolicies(ctx context.Co
 	}
 	for i := range referredPolicies {
 		policy := &referredPolicies[i]
-		h.logger.Info("Enqueue from namespace reference", "policy", k8s.NamespacedName(policy), "namespace", k8s.NamespacedName(ns))
+		h.logger.V(1).Info("Enqueue from namespace reference", "policy", k8s.NamespacedName(policy), "namespace", k8s.NamespacedName(ns))
 		h.policyEventChan <- event.GenericEvent{
 			Object: policy,
 		}

internal/eventhandlers/pod.go

@@ -92,7 +92,7 @@ func (h *enqueueRequestForPodEvent) enqueueReferredPolicies(ctx context.Context,
 	}
 	for i := range referredPolicies {
 		policy := &referredPolicies[i]
-		h.logger.Info("Enqueue from pod reference", "policy", k8s.NamespacedName(policy), "pod", k8s.NamespacedName(pod))
+		h.logger.V(1).Info("Enqueue from pod reference", "policy", k8s.NamespacedName(policy), "pod", k8s.NamespacedName(pod))
 		h.policyEventChan <- event.GenericEvent{
 			Object: policy,
 		}

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 equality.Semantic.DeepEqual(oldPolicy.Spec, newPolicy.Spec) &&
+	if oldPolicy.Generation != newPolicy.Generation && equality.Semantic.DeepEqual(oldPolicy.Spec, newPolicy.Spec) &&
 		equality.Semantic.DeepEqual(oldPolicy.DeletionTimestamp.IsZero(), newPolicy.DeletionTimestamp.IsZero()) {
 		return
 	}

internal/eventhandlers/service.go

@@ -84,7 +84,7 @@ func (h *enqueueRequestForServiceEvent) enqueueReferredPolicies(ctx context.Cont
 	}
 	for i := range referredPolicies {
 		policy := &referredPolicies[i]
-		h.logger.Info("Enqueue policies from service reference", "policy", k8s.NamespacedName(policy), "svc", k8s.NamespacedName(svc))
+		h.logger.V(1).Info("Enqueue policies from service reference", "policy", k8s.NamespacedName(policy), "svc", k8s.NamespacedName(svc))
 		h.policyEventChan <- event.GenericEvent{
 			Object: policy,
 		}

pkg/policyendpoints/indexer.go

@@ -0,0 +1,16 @@
+package policyendpoints
+
+import (
+	policyinfo "github.com/aws/amazon-network-policy-controller-k8s/api/v1alpha1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+const (
+	IndexKeyPolicyReferenceName = "spec.policyRef.name"
+)
+
+// IndexFunctionPolicyReferenceName is IndexFunc for "PolicyReference" index.
+func IndexFunctionPolicyReferenceName(obj client.Object) []string {
+	policyEndpoint := obj.(*policyinfo.PolicyEndpoint)
+	return []string{policyEndpoint.Spec.PolicyRef.Name}
+}