Sync from open-cluster-management-io/governance-policy-propagator: #130

README.md

@@ -76,18 +76,6 @@ make kind-delete-cluster
 ### Updating Deployment resources
 Some of the deployment resources are generated by kubebuilder - the crds are generated into `./deploy/crds` and the rbac details from kubebuilder comments are compiled into `./deploy/rbac/role.yaml`.  Other details are managed independently - in particular, the details in `./deploy/manager/manager.yaml`. When any of those details need to be changed, the main deployment yaml `./deploy/operator.yaml` must be regenerated through the `make generate-operator-yaml` target. The `./deploy/operator.yaml` SHOULD NOT be manually updated.
 
-## Configuration
-
-The following environment variables can be set to configure the controller:
-
-* `CONTROLLER_CONFIG_CONCURRENCY_PER_POLICY` - The maximum number of placement decisions that can be
-    processed concurrently per policy. This defaults to `5`.
-* `CONTROLLER_CONFIG_REQUEUE_ERROR_DELAY` - The number of minutes to delay before retrying to
-    process a reconcile event after one or more placement decisions failed to be processed. This is
-    not a blocking delay. This defaults to `5`.
-* `CONTROLLER_CONFIG_RETRY_ATTEMPTS` - The number of times to retry a failed Kubernetes API call
-    when processing a placement decision. This defaults to `3`.
-
 ## Running the Compliance Events API
 
 Create the KinD cluster and install Postgres with the following commands:

controllers/common/common.go

@@ -221,18 +221,6 @@ func GetDecisions(c client.Client, pb *policiesv1.PlacementBinding) ([]appsv1.Pl
 	return nil, fmt.Errorf("placement binding %s/%s reference is not valid", pb.Name, pb.Namespace)
 }
 
-// GetNumWorkers is a helper function to return the number of workers to handle concurrent tasks
-func GetNumWorkers(listLength int, concurrencyPerPolicy int) int {
-	var numWorkers int
-	if listLength > concurrencyPerPolicy {
-		numWorkers = concurrencyPerPolicy
-	} else {
-		numWorkers = listLength
-	}
-
-	return numWorkers
-}
-
 func ParseRootPolicyLabel(rootPlc string) (name, namespace string, err error) {
 	// namespaces can't have a `.` (but names can) so this always correctly pulls the namespace out
 	namespace, name, found := strings.Cut(rootPlc, ".")

controllers/common/policy_mapper.go

@@ -15,7 +15,7 @@ import (
 
 // PolicyMapper looks at object and returns a slice of reconcile.Request to reconcile
 // owners of object from label: policy.open-cluster-management.io/root-policy
-func PolicyMapper(c client.Client) handler.MapFunc {
+func MapToRootPolicy(c client.Client) handler.MapFunc {
 	return func(ctx context.Context, object client.Object) []reconcile.Request {
 		log := ctrl.Log.WithValues("name", object.GetName(), "namespace", object.GetNamespace())
 

controllers/encryptionkeys/encryptionkeys_controller.go

@@ -35,17 +35,14 @@ const (
 var (
 	log                       = ctrl.Log.WithName(ControllerName)
 	errLastRotationParseError = fmt.Errorf(`failed to parse the "%s" annotation`, propagator.LastRotatedAnnotation)
-	// The number of retries when performing an operation that can fail temporarily and can't be
-	// requeued for a retry later. This is not a const so it can be overwritten during the tests.
-	retries uint = 4
 )
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *EncryptionKeysReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *EncryptionKeysReconciler) SetupWithManager(mgr ctrl.Manager, maxConcurrentReconciles uint) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		// The work queue prevents the same item being reconciled concurrently:
 		// https://github.com/kubernetes-sigs/controller-runtime/issues/1416#issuecomment-899833144
-		WithOptions(controller.Options{MaxConcurrentReconciles: int(r.MaxConcurrentReconciles)}).
+		WithOptions(controller.Options{MaxConcurrentReconciles: int(maxConcurrentReconciles)}).
 		Named(ControllerName).
 		For(&corev1.Secret{}).
 		Complete(r)
@@ -58,9 +55,8 @@ var _ reconcile.Reconciler = &EncryptionKeysReconciler{}
 // for all managed clusters.
 type EncryptionKeysReconciler struct { //nolint:golint,revive
 	client.Client
-	KeyRotationDays         uint
-	MaxConcurrentReconciles uint
-	Scheme                  *runtime.Scheme
+	KeyRotationDays uint
+	Scheme          *runtime.Scheme
 }
 
 //+kubebuilder:rbac:groups=policy.open-cluster-management.io,resources=policies,verbs=get;list;patch

controllers/encryptionkeys/encryptionkeys_controller_test.go

@@ -205,10 +205,9 @@ func getReconciler(encryptionSecret *corev1.Secret) *EncryptionKeysReconciler {
 	client := builder.Build()
 
 	return &EncryptionKeysReconciler{
-		Client:                  client,
-		KeyRotationDays:         30,
-		MaxConcurrentReconciles: 1,
-		Scheme:                  scheme,
+		Client:          client,
+		KeyRotationDays: 30,
+		Scheme:          scheme,
 	}
 }
 
@@ -464,11 +463,6 @@ func TestReconcileAPIFails(t *testing.T) {
 	t.Parallel()
 	RegisterFailHandler(Fail)
 
-	originalRetries := retries
-	retries = 0
-
-	t.Cleanup(func() { retries = originalRetries })
-
 	tests := []struct {
 		ExpectedRotation bool
 		GetError         bool

controllers/policymetrics/policymetrics_controller.go

@@ -24,11 +24,11 @@ const ControllerName string = "policy-metrics"
 var log = ctrl.Log.WithName(ControllerName)
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *MetricReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *MetricReconciler) SetupWithManager(mgr ctrl.Manager, maxConcurrentReconciles uint) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		// The work queue prevents the same item being reconciled concurrently:
 		// https://github.com/kubernetes-sigs/controller-runtime/issues/1416#issuecomment-899833144
-		WithOptions(controller.Options{MaxConcurrentReconciles: int(r.MaxConcurrentReconciles)}).
+		WithOptions(controller.Options{MaxConcurrentReconciles: int(maxConcurrentReconciles)}).
 		Named(ControllerName).
 		For(&policiesv1.Policy{}).
 		Complete(r)
@@ -40,8 +40,7 @@ var _ reconcile.Reconciler = &MetricReconciler{}
 // MetricReconciler reconciles the metrics for the Policy
 type MetricReconciler struct {
 	client.Client
-	MaxConcurrentReconciles uint
-	Scheme                  *runtime.Scheme
+	Scheme *runtime.Scheme
 }
 
 //+kubebuilder:rbac:groups=policy.open-cluster-management.io,resources=policies,verbs=get;list;watch;create;update;patch;delete

controllers/propagator/aggregation.go

@@ -4,69 +4,58 @@ import (
 	"context"
 	"sort"
 
+	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
 
 	policiesv1 "open-cluster-management.io/governance-policy-propagator/api/v1"
 )
 
 // calculatePerClusterStatus lists up all policies replicated from the input policy, and stores
-// their compliance states in the result list. Additionally, clusters in the failedClusters input
-// will be marked as NonCompliant in the result. The result is sorted by cluster name. An error
-// will be returned if lookup of the replicated policies fails, and the retries also fail.
-func (r *PolicyReconciler) calculatePerClusterStatus(
-	instance *policiesv1.Policy, allDecisions, failedClusters decisionSet,
+// their compliance states in the result list. The result is sorted by cluster name. An error
+// will be returned if lookup of a replicated policy fails, but all lookups will still be attempted.
+func (r *Propagator) calculatePerClusterStatus(
+	instance *policiesv1.Policy, decisions decisionSet,
 ) ([]*policiesv1.CompliancePerClusterStatus, error) {
 	if instance.Spec.Disabled {
 		return nil, nil
 	}
 
-	status := make([]*policiesv1.CompliancePerClusterStatus, 0, len(allDecisions))
+	status := make([]*policiesv1.CompliancePerClusterStatus, 0, len(decisions))
+	var lookupErr error // save until end, to attempt all lookups
 
 	// Update the status based on the processed decisions
-	for decision := range allDecisions {
-		if failedClusters[decision] {
-			// Skip the replicated policies that failed to be properly replicated
-			// for now. This will be handled later.
-			continue
-		}
-
-		rPlc := &policiesv1.Policy{}
+	for dec := range decisions {
+		replicatedPolicy := &policiesv1.Policy{}
 		key := types.NamespacedName{
-			Namespace: decision.ClusterNamespace, Name: instance.Namespace + "." + instance.Name,
+			Namespace: dec.ClusterNamespace, Name: instance.Namespace + "." + instance.Name,
 		}
 
-		err := r.Get(context.TODO(), key, rPlc)
+		err := r.Get(context.TODO(), key, replicatedPolicy)
 		if err != nil {
-			return nil, err
-		}
+			if errors.IsNotFound(err) {
+				status = append(status, &policiesv1.CompliancePerClusterStatus{
+					ClusterName:      dec.ClusterName,
+					ClusterNamespace: dec.ClusterNamespace,
+				})
 
-		status = append(status, &policiesv1.CompliancePerClusterStatus{
-			ComplianceState:  rPlc.Status.ComplianceState,
-			ClusterName:      decision.ClusterName,
-			ClusterNamespace: decision.ClusterNamespace,
-		})
-	}
+				continue
+			}
 
-	// Add cluster statuses for the clusters that did not get their policies properly
-	// replicated. This is not done in the previous loop since some replicated polices may not
-	// have been created at all.
-	for clusterDecision := range failedClusters {
-		log.Info(
-			"Setting the policy to noncompliant since the replication failed", "cluster", clusterDecision,
-		)
+			lookupErr = err
+		}
 
 		status = append(status, &policiesv1.CompliancePerClusterStatus{
-			ComplianceState:  policiesv1.NonCompliant,
-			ClusterName:      clusterDecision.ClusterName,
-			ClusterNamespace: clusterDecision.ClusterNamespace,
+			ComplianceState:  replicatedPolicy.Status.ComplianceState,
+			ClusterName:      dec.ClusterName,
+			ClusterNamespace: dec.ClusterNamespace,
 		})
 	}
 
 	sort.Slice(status, func(i, j int) bool {
 		return status[i].ClusterName < status[j].ClusterName
 	})
 
-	return status, nil
+	return status, lookupErr
 }
 
 // CalculateRootCompliance uses the input per-cluster statuses to determine what a root policy's

controllers/propagator/encryption.go

@@ -26,7 +26,7 @@ const (
 
 // getEncryptionKey will get the encryption key for a managed cluster used for policy template encryption. If it doesn't
 // already exist as a secret on the Hub cluster, it will be generated.
-func (r *PolicyReconciler) getEncryptionKey(clusterName string) ([]byte, error) {
+func (r *Propagator) getEncryptionKey(clusterName string) ([]byte, error) {
 	ctx := context.TODO()
 	objectKey := types.NamespacedName{
 		Name:      EncryptionKeySecret,
@@ -90,7 +90,7 @@ func GenerateEncryptionKey() ([]byte, error) {
 // getInitializationVector retrieves the initialization vector from the annotation
 // "policy.open-cluster-management.io/encryption-iv" if the annotation exists or generates a new
 // initialization vector and adds it to the annotations object if it's missing.
-func (r *PolicyReconciler) getInitializationVector(
+func (r *Propagator) getInitializationVector(
 	policyName string, clusterName string, annotations map[string]string,
 ) ([]byte, error) {
 	log := log.WithValues("policy", policyName, "cluster", clusterName)

controllers/propagator/encryption_test.go

@@ -28,7 +28,7 @@ func TestGetEncryptionKeyNoSecret(_ *testing.T) {
 	RegisterFailHandler(Fail)
 
 	client := fake.NewClientBuilder().Build()
-	r := PolicyReconciler{Client: client}
+	r := Propagator{Client: client}
 	key, err := r.getEncryptionKey(clusterName)
 
 	Expect(err).ToNot(HaveOccurred())
@@ -68,7 +68,7 @@ func TestGetEncryptionKeySecretExists(_ *testing.T) {
 
 	client := fake.NewClientBuilder().WithObjects(encryptionSecret).Build()
 
-	r := PolicyReconciler{Client: client}
+	r := Propagator{Client: client}
 	key, err = r.getEncryptionKey(clusterName)
 
 	Expect(err).ToNot(HaveOccurred())
@@ -103,7 +103,7 @@ func TestGetInitializationVector(t *testing.T) {
 		},
 	}
 
-	r := PolicyReconciler{}
+	r := Propagator{}
 
 	for _, test := range tests {
 		subTest := test