Enhancement - Optimize Reconciler Triggers Using Predicates (#3633)

Updated reconcilers to ignore updates that do not modify the spec
Refactored the logic to get the worker threads
Added unit tests wherever applicable

Author: kolluria

pkg/syncer/cnsoperator/controller/cnsfileaccessconfig/cnsfileaccessconfig_controller.go

@@ -27,6 +27,7 @@ import (
 
 	vmoperatortypes "github.com/vmware-tanzu/vm-operator/api/v1alpha5"
 	cnstypes "github.com/vmware/govmomi/cns/types"
+	vsanfstypes "github.com/vmware/govmomi/vsan/vsanfs/types"
 	v1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -39,13 +40,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
-
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
-
-	vsanfstypes "github.com/vmware/govmomi/vsan/vsanfs/types"
 	cnsoperatorapis "sigs.k8s.io/vsphere-csi-driver/v3/pkg/apis/cnsoperator"
 	cnsfileaccessconfigv1alpha1 "sigs.k8s.io/vsphere-csi-driver/v3/pkg/apis/cnsoperator/cnsfileaccessconfig/v1alpha1"
 	volumes "sigs.k8s.io/vsphere-csi-driver/v3/pkg/common/cns-lib/volume"
@@ -57,13 +55,14 @@ import (
 	k8s "sigs.k8s.io/vsphere-csi-driver/v3/pkg/kubernetes"
 	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/syncer"
 	cnsoperatortypes "sigs.k8s.io/vsphere-csi-driver/v3/pkg/syncer/cnsoperator/types"
-	cnsoperatorutil "sigs.k8s.io/vsphere-csi-driver/v3/pkg/syncer/cnsoperator/util"
+	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/syncer/cnsoperator/util"
 )
 
 const (
-	defaultMaxWorkerThreadsForFileAccessConfig = 10
-	capvVmLabelKey                             = "capv.vmware.com"
-	devopsUserLabelKey                         = "cns.vmware.com/user-created"
+	workerThreadsEnvVar     = "WORKER_THREADS_CNS_FILE_ACCESS_CONFIG"
+	defaultMaxWorkerThreads = 10
+	capvVmLabelKey          = "capv.vmware.com"
+	devopsUserLabelKey      = "cns.vmware.com/user-created"
 )
 
 // backOffDuration is a map of cnsfileaccessconfig name's to the time after
@@ -87,18 +86,16 @@ func Add(mgr manager.Manager, clusterFlavor cnstypes.CnsClusterFlavor,
 		return nil
 	}
 
-	if clusterFlavor == cnstypes.CnsClusterFlavorWorkload {
-		if commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx, common.TKGsHA) {
-			clusterComputeResourceMoIds, _, err := common.GetClusterComputeResourceMoIds(ctx)
-			if err != nil {
-				log.Errorf("failed to get clusterComputeResourceMoIds. err: %v", err)
-				return err
-			}
-			if len(clusterComputeResourceMoIds) > 1 &&
-				!commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx, common.WorkloadDomainIsolation) {
-				log.Infof("Not initializing the CnsFileAccessConfig Controller as stretched supervisor is detected.")
-				return nil
-			}
+	if commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx, common.TKGsHA) {
+		clusterComputeResourceMoIds, _, err := common.GetClusterComputeResourceMoIds(ctx)
+		if err != nil {
+			log.Errorf("failed to get clusterComputeResourceMoIds. err: %v", err)
+			return err
+		}
+		if len(clusterComputeResourceMoIds) > 1 &&
+			!commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx, common.WorkloadDomainIsolation) {
+			log.Infof("Not initializing the CnsFileAccessConfig Controller as stretched supervisor is detected.")
+			return nil
 		}
 	}
 	volumePermissionLockMap = &sync.Map{}
@@ -164,17 +161,23 @@ func Add(mgr manager.Manager, clusterFlavor cnstypes.CnsClusterFlavor,
 func newReconciler(mgr manager.Manager, configInfo *commonconfig.ConfigurationInfo,
 	volumeManager volumes.Manager, vmOperatorClient client.Client, dynamicClient dynamic.Interface,
 	recorder record.EventRecorder) reconcile.Reconciler {
-	return &ReconcileCnsFileAccessConfig{client: mgr.GetClient(), scheme: mgr.GetScheme(),
-		configInfo: configInfo, volumeManager: volumeManager, vmOperatorClient: vmOperatorClient,
-		dynamicClient: dynamicClient, recorder: recorder}
+	return &ReconcileCnsFileAccessConfig{
+		client:           mgr.GetClient(),
+		scheme:           mgr.GetScheme(),
+		configInfo:       configInfo,
+		volumeManager:    volumeManager,
+		vmOperatorClient: vmOperatorClient,
+		dynamicClient:    dynamicClient,
+		recorder:         recorder,
+	}
 }
 
 // add adds a new Controller to mgr with r as the reconcile.Reconciler.
 func add(mgr manager.Manager, r reconcile.Reconciler) error {
 	ctx, log := logger.GetNewContextWithLogger()
 
-	maxWorkerThreads := getMaxWorkerThreadsToReconcileCnsFileAccessConfig(ctx)
-
+	maxWorkerThreads := util.GetMaxWorkerThreads(ctx,
+		workerThreadsEnvVar, defaultMaxWorkerThreads)
 	// Create a new controller.
 	c, err := controller.New("cnsfileaccessconfig-controller", mgr,
 		controller.Options{Reconciler: r, MaxConcurrentReconciles: maxWorkerThreads})
@@ -261,7 +264,7 @@ func (r *ReconcileCnsFileAccessConfig) Reconcile(ctx context.Context,
 				instance.Name, instance.Spec.VMName)
 			// Fetch the PVC and PV instance and get volume ID
 			skipConfigureVolumeACL := false
-			volumeID, err := cnsoperatorutil.GetVolumeID(ctx, r.client, instance.Spec.PvcName, instance.Namespace)
+			volumeID, err := util.GetVolumeID(ctx, r.client, instance.Spec.PvcName, instance.Namespace)
 			if err != nil {
 				if apierrors.IsNotFound(err) {
 					// If PVC instance is NotFound (deleted), then there is no need to configure ACL on file volume.
@@ -330,7 +333,7 @@ func (r *ReconcileCnsFileAccessConfig) Reconcile(ctx context.Context,
 	if instance.DeletionTimestamp != nil {
 		log.Infof("CnsFileAccessConfig instance %q has deletion timestamp set", instance.Name)
 		volumeExists := true
-		volumeID, err := cnsoperatorutil.GetVolumeID(ctx, r.client, instance.Spec.PvcName, instance.Namespace)
+		volumeID, err := util.GetVolumeID(ctx, r.client, instance.Spec.PvcName, instance.Namespace)
 		if err != nil {
 			if ifFileVolumesWithVmserviceVmsSupported &&
 				apierrors.IsNotFound(err) {
@@ -447,7 +450,7 @@ func (r *ReconcileCnsFileAccessConfig) Reconcile(ctx context.Context,
 	log.Infof("Reconciling CnsFileAccessConfig with instance: %q from namespace: %q. timeout %q seconds",
 		instance.Name, instance.Namespace, timeout)
 	if !instance.Status.Done {
-		volumeID, err := cnsoperatorutil.GetVolumeID(ctx, r.client, instance.Spec.PvcName, instance.Namespace)
+		volumeID, err := util.GetVolumeID(ctx, r.client, instance.Spec.PvcName, instance.Namespace)
 		if err != nil {
 			msg := fmt.Sprintf("Failed to get volumeID from pvcName: %q. Error: %+v", instance.Spec.PvcName, err)
 			log.Error(msg)
@@ -780,7 +783,7 @@ func (r *ReconcileCnsFileAccessConfig) configureVolumeACLs(ctx context.Context,
 func (r *ReconcileCnsFileAccessConfig) getVMExternalIP(ctx context.Context,
 	vm *vmoperatortypes.VirtualMachine) (string, error) {
 	log := logger.GetLogger(ctx)
-	networkProvider, err := cnsoperatorutil.GetNetworkProvider(ctx)
+	networkProvider, err := util.GetNetworkProvider(ctx)
 	if err != nil {
 		return "", logger.LogNewErrorf(log, "Failed to identify the network provider. Error: %+v", err)
 	}
@@ -789,11 +792,11 @@ func (r *ReconcileCnsFileAccessConfig) getVMExternalIP(ctx context.Context,
 		return "", logger.LogNewError(log, "unable to find network provider information")
 	}
 
-	networkTypes := []string{cnsoperatorutil.NSXTNetworkProvider, cnsoperatorutil.
+	networkTypes := []string{util.NSXTNetworkProvider, util.
 		VDSNetworkProvider}
 
 	if commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx, common.VPCCapabilitySupervisor) {
-		networkTypes = append(networkTypes, cnsoperatorutil.VPCNetworkProvider)
+		networkTypes = append(networkTypes, util.VPCNetworkProvider)
 	}
 
 	supported_found := false
@@ -808,7 +811,7 @@ func (r *ReconcileCnsFileAccessConfig) getVMExternalIP(ctx context.Context,
 		return "", logger.LogNewErrorf(log, "Unknown network provider. Error: %+v", err)
 	}
 
-	tkgVMIP, err := cnsoperatorutil.GetTKGVMIP(ctx, r.vmOperatorClient,
+	tkgVMIP, err := util.GetTKGVMIP(ctx, r.vmOperatorClient,
 		r.dynamicClient, vm.Namespace, vm.Name, networkProvider)
 	if err != nil {
 		return "", logger.LogNewErrorf(log, "Failed to get external facing IP address for VM %q/%q. Err: %+v",

pkg/syncer/cnsoperator/controller/cnsfileaccessconfig/util.go

@@ -19,9 +19,7 @@ package cnsfileaccessconfig
 import (
 	"context"
 	"fmt"
-	"os"
 	"reflect"
-	"strconv"
 
 	vmoperatortypes "github.com/vmware-tanzu/vm-operator/api/v1alpha5"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -70,38 +68,3 @@ func setInstanceOwnerRef(instance *cnsfileaccessconfigv1alpha1.CnsFileAccessConf
 		},
 	}
 }
-
-// getMaxWorkerThreadsToReconcileCnsFileAccessConfig returns the maximum number
-// of worker threads which can be run to reconcile CnsFileAccessConfig instances.
-// If environment variable WORKER_THREADS_FILE_ACCESS_CONFIG is set and valid,
-// return the value read from environment variable otherwise, use the default
-// value.
-func getMaxWorkerThreadsToReconcileCnsFileAccessConfig(ctx context.Context) int {
-	log := logger.GetLogger(ctx)
-	workerThreads := defaultMaxWorkerThreadsForFileAccessConfig
-	if v := os.Getenv("WORKER_THREADS_FILE_ACCESS_CONFIG"); v != "" {
-		if value, err := strconv.Atoi(v); err == nil {
-			if value <= 0 {
-				log.Warnf("Maximum number of worker threads to run set in env variable "+
-					"WORKER_THREADS_FILE_ACCESS_CONFIG %s is less than 1, will use the default value %d",
-					v, defaultMaxWorkerThreadsForFileAccessConfig)
-			} else if value > defaultMaxWorkerThreadsForFileAccessConfig {
-				log.Warnf("Maximum number of worker threads to run set in env variable "+
-					"WORKER_THREADS_FILE_ACCESS_CONFIG %s is greater than %d, will use the default value %d",
-					v, defaultMaxWorkerThreadsForFileAccessConfig, defaultMaxWorkerThreadsForFileAccessConfig)
-			} else {
-				workerThreads = value
-				log.Debugf("Maximum number of worker threads to run to reconcile "+
-					"CnsFileAccessConfig instances is set to %d", workerThreads)
-			}
-		} else {
-			log.Warnf("Maximum number of worker threads to run set in env variable "+
-				"WORKER_THREADS_FILE_ACCESS_CONFIG %s is invalid, will use the default value %d",
-				v, defaultMaxWorkerThreadsForFileAccessConfig)
-		}
-	} else {
-		log.Debugf("WORKER_THREADS_FILE_ACCESS_CONFIG is not set. Picking the default value %d",
-			defaultMaxWorkerThreadsForFileAccessConfig)
-	}
-	return workerThreads
-}

pkg/syncer/cnsoperator/controller/cnsnodevmattachment/cnsnodevmattachment_controller.go

@@ -20,8 +20,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"os"
-	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -55,10 +53,12 @@ import (
 	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/csi/service/logger"
 	k8s "sigs.k8s.io/vsphere-csi-driver/v3/pkg/kubernetes"
 	cnsoptypes "sigs.k8s.io/vsphere-csi-driver/v3/pkg/syncer/cnsoperator/types"
+	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/syncer/cnsoperator/util"
 )
 
 const (
-	defaultMaxWorkerThreadsForNodeVMAttach = 10
+	workerThreadsEnvVar     = "WORKER_THREADS_NODEVM_ATTACH"
+	defaultMaxWorkerThreads = 10
 )
 
 // backOffDuration is a map of cnsnodevmattachment name's to the time after
@@ -135,15 +135,17 @@ func newReconciler(mgr manager.Manager, configInfo *config.ConfigurationInfo,
 // add adds a new Controller to mgr with r as the reconcile.Reconciler.
 func add(mgr manager.Manager, r reconcile.Reconciler) error {
 	ctx, log := logger.GetNewContextWithLogger()
-	maxWorkerThreads := getMaxWorkerThreadsToReconcileCnsNodeVmAttachment(ctx)
+
+	maxWorkerThreads := util.GetMaxWorkerThreads(ctx,
+		workerThreadsEnvVar, defaultMaxWorkerThreads)
 	// Create a new controller.
 	err := ctrl.NewControllerManagedBy(mgr).Named("cnsnodevmattachment-controller").
 		For(&v1a1.CnsNodeVmAttachment{}).
 		WithEventFilter(predicate.GenerationChangedPredicate{}).
 		WithOptions(controller.Options{MaxConcurrentReconciles: maxWorkerThreads}).
 		Complete(r)
 	if err != nil {
-		log.Errorf("failed to create new CnsNodeVmAttachment controller with error: %+v", err)
+		log.Errorf("Failed to build application controller. Err: %v", err)
 		return err
 	}
 
@@ -869,41 +871,6 @@ func updateCnsNodeVMAttachment(ctx context.Context, client client.Client,
 	return err
 }
 
-// getMaxWorkerThreadsToReconcileCnsNodeVmAttachment returns the maximum
-// number of worker threads which can be run to reconcile CnsNodeVmAttachment
-// instances. If environment variable WORKER_THREADS_NODEVM_ATTACH is set and
-// valid, return the value read from environment variable otherwise, use the
-// default value.
-func getMaxWorkerThreadsToReconcileCnsNodeVmAttachment(ctx context.Context) int {
-	log := logger.GetLogger(ctx)
-	workerThreads := defaultMaxWorkerThreadsForNodeVMAttach
-	if v := os.Getenv("WORKER_THREADS_NODEVM_ATTACH"); v != "" {
-		if value, err := strconv.Atoi(v); err == nil {
-			if value <= 0 {
-				log.Warnf("Maximum number of worker threads to run set in env variable "+
-					"WORKER_THREADS_NODEVM_ATTACH %s is less than 1, will use the default value %d",
-					v, defaultMaxWorkerThreadsForNodeVMAttach)
-			} else if value > defaultMaxWorkerThreadsForNodeVMAttach {
-				log.Warnf("Maximum number of worker threads to run set in env variable "+
-					"WORKER_THREADS_NODEVM_ATTACH %s is greater than %d, will use the default value %d",
-					v, defaultMaxWorkerThreadsForNodeVMAttach, defaultMaxWorkerThreadsForNodeVMAttach)
-			} else {
-				workerThreads = value
-				log.Debugf("Maximum number of worker threads to run to reconcile CnsNodeVmAttachment "+
-					"instances is set to %d", workerThreads)
-			}
-		} else {
-			log.Warnf("Maximum number of worker threads to run set in env variable "+
-				"WORKER_THREADS_NODEVM_ATTACH %s is invalid, will use the default value %d",
-				v, defaultMaxWorkerThreadsForNodeVMAttach)
-		}
-	} else {
-		log.Debugf("WORKER_THREADS_NODEVM_ATTACH is not set. Picking the default value %d",
-			defaultMaxWorkerThreadsForNodeVMAttach)
-	}
-	return workerThreads
-}
-
 // recordEvent records the event, sets the backOffDuration for the instance
 // appropriately and logs the message.
 // backOffDuration is reset to 1 second on success and doubled on failure.

pkg/syncer/cnsoperator/controller/cnsnodevmbatchattachment/cnsnodevmbatchattachment_controller.go

@@ -20,13 +20,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"os"
-	"strconv"
 	"strings"
 	"sync"
 	"time"
 
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/syncer/cnsoperator/util"
 
 	cnsvsphere "sigs.k8s.io/vsphere-csi-driver/v3/pkg/common/cns-lib/vsphere"
 	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/csi/service/common/commonco"
@@ -50,7 +49,7 @@ import (
 	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 
 	cnsoperatorapis "sigs.k8s.io/vsphere-csi-driver/v3/pkg/apis/cnsoperator"
-	v1alpha1 "sigs.k8s.io/vsphere-csi-driver/v3/pkg/apis/cnsoperator/cnsnodevmbatchattachment/v1alpha1"
+	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/apis/cnsoperator/cnsnodevmbatchattachment/v1alpha1"
 	volumes "sigs.k8s.io/vsphere-csi-driver/v3/pkg/common/cns-lib/volume"
 	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/common/config"
 	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/csi/service/common"
@@ -69,6 +68,7 @@ var (
 )
 
 const (
+	workerThreadsEnvVar     = "WORKER_THREADS_NODEVM_BATCH_ATTACH"
 	defaultMaxWorkerThreads = 10
 )
 
@@ -132,10 +132,9 @@ func newReconciler(mgr manager.Manager, configInfo *config.ConfigurationInfo,
 // add adds this package's controller to the provided manager.
 func add(mgr manager.Manager, r reconcile.Reconciler) error {
 	ctx, log := logger.GetNewContextWithLogger()
-	maxWorkerThreads := getMaxWorkerThreads(ctx)
-
-	backOffDuration = make(map[types.NamespacedName]time.Duration)
 
+	maxWorkerThreads := util.GetMaxWorkerThreads(ctx,
+		workerThreadsEnvVar, defaultMaxWorkerThreads)
 	// Create a new controller.
 	err := ctrl.NewControllerManagedBy(mgr).
 		Named("cnsnodevmbatchattachment-controller").
@@ -145,11 +144,12 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
 			MaxConcurrentReconciles: maxWorkerThreads,
 		}).
 		Complete(r)
-
 	if err != nil {
-		log.Errorf("failed to watch for changes to CnsNodeVmBatchAttachment resource with error: %+v", err)
+		log.Errorf("Failed to build application controller. Err: %v", err)
 		return err
 	}
+
+	backOffDuration = make(map[types.NamespacedName]time.Duration)
 	return nil
 }
 
@@ -168,42 +168,6 @@ type Reconciler struct {
 	instanceLock sync.Map
 }
 
-// getMaxWorkerThreads returns the maximum
-// number of worker threads which can be run to reconcile CnsNodeVmBatchAttachment
-// instances. If environment variable WORKER_THREADS_NODEVM_BATCH_ATTACH is set and
-// valid, return the value read from environment variable otherwise, use the
-// default value.
-func getMaxWorkerThreads(ctx context.Context) int {
-	log := logger.GetLogger(ctx)
-
-	workerThreads := defaultMaxWorkerThreads
-	envVal := os.Getenv("WORKER_THREADS_NODEVM_BATCH_ATTACH")
-	if envVal == "" {
-		log.Debugf("WORKER_THREADS_NODEVM_BATCH_ATTACH is not set. Picking the default value %d",
-			defaultMaxWorkerThreads)
-		return workerThreads
-	}
-
-	value, err := strconv.Atoi(envVal)
-	if err != nil {
-		log.Warnf("Invalid value for WORKER_THREADS_NODEVM_BATCH_ATTACH: %s. Using default value %d",
-			envVal, defaultMaxWorkerThreads)
-		return workerThreads
-	}
-
-	switch {
-	case value <= 0 || value > defaultMaxWorkerThreads:
-		log.Warnf("Value %s for WORKER_THREADS_NODEVM_BATCH_ATTACH is invalid. Using default value %d",
-			envVal, defaultMaxWorkerThreads)
-	default:
-		workerThreads = value
-		log.Debugf("Maximum number of worker threads to reconcile CnsNodeVmBatchAttachment is set to %d",
-			workerThreads)
-
-	}
-	return workerThreads
-}
-
 // Reconcile over CnsNodeVmBatchAttachment CR.
 // Reconcile stops when all volumes have been attached or detached successfully.
 func (r *Reconciler) Reconcile(ctx context.Context,