feat: enable the new work applier (#1042)

Author: michaelawyu

cmd/memberagent/main.go

@@ -43,13 +43,14 @@ import (
 	fleetv1alpha1 "go.goms.io/fleet/apis/v1alpha1"
 	imcv1alpha1 "go.goms.io/fleet/pkg/controllers/internalmembercluster/v1alpha1"
 	imcv1beta1 "go.goms.io/fleet/pkg/controllers/internalmembercluster/v1beta1"
-	"go.goms.io/fleet/pkg/controllers/work"
+	"go.goms.io/fleet/pkg/controllers/workapplier"
 	workv1alpha1controller "go.goms.io/fleet/pkg/controllers/workv1alpha1"
 	fleetmetrics "go.goms.io/fleet/pkg/metrics"
 	"go.goms.io/fleet/pkg/propertyprovider"
 	"go.goms.io/fleet/pkg/propertyprovider/azure"
 	"go.goms.io/fleet/pkg/utils"
 	"go.goms.io/fleet/pkg/utils/httpclient"
+	"go.goms.io/fleet/pkg/utils/parallelizer"
 	//+kubebuilder:scaffold:imports
 )
 
@@ -68,12 +69,14 @@ var (
 	metricsAddr          = flag.String("metrics-bind-address", ":8090", "The address the metric endpoint binds to.")
 	enableLeaderElection = flag.Bool("leader-elect", false,
 		"Enable leader election for controller manager. Enabling this will ensure there is only one active controller manager.")
-	leaderElectionNamespace = flag.String("leader-election-namespace", "kube-system", "The namespace in which the leader election resource will be created.")
-	enableV1Alpha1APIs      = flag.Bool("enable-v1alpha1-apis", true, "If set, the agents will watch for the v1alpha1 APIs.")
-	enableV1Beta1APIs       = flag.Bool("enable-v1beta1-apis", false, "If set, the agents will watch for the v1beta1 APIs.")
-	propertyProvider        = flag.String("property-provider", "none", "The property provider to use for the agent.")
-	region                  = flag.String("region", "", "The region where the member cluster resides.")
-	cloudConfigFile         = flag.String("cloud-config", "/etc/kubernetes/provider/config.json", "The path to the cloud cloudconfig file.")
+	leaderElectionNamespace   = flag.String("leader-election-namespace", "kube-system", "The namespace in which the leader election resource will be created.")
+	enableV1Alpha1APIs        = flag.Bool("enable-v1alpha1-apis", true, "If set, the agents will watch for the v1alpha1 APIs.")
+	enableV1Beta1APIs         = flag.Bool("enable-v1beta1-apis", false, "If set, the agents will watch for the v1beta1 APIs.")
+	propertyProvider          = flag.String("property-provider", "none", "The property provider to use for the agent.")
+	region                    = flag.String("region", "", "The region where the member cluster resides.")
+	cloudConfigFile           = flag.String("cloud-config", "/etc/kubernetes/provider/config.json", "The path to the cloud cloudconfig file.")
+	availabilityCheckInterval = flag.Int("availability-check-interval", 5, "The interval in seconds between attempts to check for resource availability when resources are not yet available.")
+	driftDetectionInterval    = flag.Int("drift-detection-interval", 15, "The interval in seconds between attempts to detect configuration drifts in the cluster.")
 )
 
 func init() {
@@ -349,11 +352,20 @@ func Start(ctx context.Context, hubCfg, memberConfig *rest.Config, hubOpts, memb
 			return err
 		}
 		// create the work controller, so we can pass it to the internal member cluster reconciler
-		workController := work.NewApplyWorkReconciler(
+		workController := workapplier.NewReconciler(
 			hubMgr.GetClient(),
+			targetNS,
 			spokeDynamicClient,
 			memberMgr.GetClient(),
-			restMapper, hubMgr.GetEventRecorderFor("work_controller"), 5, targetNS)
+			restMapper,
+			hubMgr.GetEventRecorderFor("work_applier"),
+			// The number of concurrent reconcilations. This is set to 5 to boost performance in
+			// resource processing.
+			5,
+			// Use the default worker count (4) for parallelized manifest processing.
+			parallelizer.DefaultNumOfWorkers,
+			time.Second*time.Duration(*availabilityCheckInterval),
+			time.Second*time.Duration(*driftDetectionInterval))
 
 		if err = workController.SetupWithManager(hubMgr); err != nil {
 			klog.ErrorS(err, "Failed to create v1beta1 controller", "controller", "work")

pkg/controllers/internalmembercluster/v1beta1/member_suite_test.go

@@ -10,6 +10,7 @@ import (
 	"path/filepath"
 	"sync"
 	"testing"
+	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -30,7 +31,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/metrics/server"
 
 	clusterv1beta1 "go.goms.io/fleet/apis/cluster/v1beta1"
-	"go.goms.io/fleet/pkg/controllers/work"
+	"go.goms.io/fleet/pkg/controllers/workapplier"
 	"go.goms.io/fleet/pkg/propertyprovider"
 )
 
@@ -60,8 +61,8 @@ var (
 	hubClient         client.Client
 	member1Client     client.Client
 	member2Client     client.Client
-	workApplier1      *work.ApplyWorkReconciler
-	workApplier2      *work.ApplyWorkReconciler
+	workApplier1      *workapplier.Reconciler
+	workApplier2      *workapplier.Reconciler
 	propertyProvider1 *manuallyUpdatedProvider
 
 	ctx    context.Context
@@ -367,7 +368,7 @@ var _ = BeforeSuite(func() {
 
 	// This controller is created for testing purposes only; no reconciliation loop is actually
 	// run.
-	workApplier1 = work.NewApplyWorkReconciler(hubClient, nil, nil, nil, nil, 0, "")
+	workApplier1 = workapplier.NewReconciler(hubClient, member1ReservedNSName, nil, nil, nil, nil, 0, 1, time.Second*5, time.Second*5)
 
 	propertyProvider1 = &manuallyUpdatedProvider{}
 	member1Reconciler, err := NewReconciler(ctx, hubClient, member1Cfg, member1Client, workApplier1, propertyProvider1)
@@ -390,7 +391,7 @@ var _ = BeforeSuite(func() {
 
 	// This controller is created for testing purposes only; no reconciliation loop is actually
 	// run.
-	workApplier2 = work.NewApplyWorkReconciler(hubClient, nil, nil, nil, nil, 0, "")
+	workApplier2 = workapplier.NewReconciler(hubClient, member2ReservedNSName, nil, nil, nil, nil, 0, 1, time.Second*5, time.Second*5)
 
 	member2Reconciler, err := NewReconciler(ctx, hubClient, member2Cfg, member2Client, workApplier2, nil)
 	Expect(err).NotTo(HaveOccurred())

pkg/controllers/rollout/controller.go

@@ -30,7 +30,7 @@ import (
 
 	fleetv1alpha1 "go.goms.io/fleet/apis/placement/v1alpha1"
 	fleetv1beta1 "go.goms.io/fleet/apis/placement/v1beta1"
-	"go.goms.io/fleet/pkg/controllers/work"
+	"go.goms.io/fleet/pkg/controllers/workapplier"
 	bindingutils "go.goms.io/fleet/pkg/utils/binding"
 	"go.goms.io/fleet/pkg/utils/condition"
 	"go.goms.io/fleet/pkg/utils/controller"
@@ -626,7 +626,10 @@ func isBindingReady(binding *fleetv1beta1.ClusterResourceBinding, readyTimeCutOf
 	// find the latest applied condition that has the same generation as the binding
 	availableCondition := binding.GetCondition(string(fleetv1beta1.ResourceBindingAvailable))
 	if condition.IsConditionStatusTrue(availableCondition, binding.GetGeneration()) {
-		if availableCondition.Reason != work.WorkNotTrackableReason {
+		// TO-DO (chenyu1): currently it checks for both the new and the old reason
+		// (as set previously by the work generator) to avoid compatibility issues.
+		// the check for the old reason can be removed once the rollout completes successfully.
+		if availableCondition.Reason != condition.WorkNotAvailabilityTrackableReason && availableCondition.Reason != workapplier.WorkNotAllManifestsTrackableReason {
 			return 0, true
 		}
 
@@ -1035,7 +1038,7 @@ func (r *Reconciler) processApplyStrategyUpdates(
 		// Verify if the binding has the latest apply strategy set.
 		if equality.Semantic.DeepEqual(binding.Spec.ApplyStrategy, applyStrategy) {
 			// The binding already has the latest apply strategy set; no need to push the update.
-			klog.V(3).InfoS("The binding already has the latest apply strategy; skip the apply strategy update", "clusterResourceBinding", klog.KObj(binding))
+			klog.V(2).InfoS("The binding already has the latest apply strategy; skip the apply strategy update", "clusterResourceBinding", klog.KObj(binding), "bindingGeneration", binding.Generation)
 			continue
 		}
 
@@ -1052,7 +1055,7 @@ func (r *Reconciler) processApplyStrategyUpdates(
 				klog.ErrorS(err, "Failed to update binding with new apply strategy", "clusterResourceBinding", klog.KObj(binding))
 				return controller.NewAPIServerError(false, err)
 			}
-			klog.V(2).InfoS("Updated binding with new apply strategy", "clusterResourceBinding", klog.KObj(binding))
+			klog.V(2).InfoS("Updated binding with new apply strategy", "clusterResourceBinding", klog.KObj(binding), "beforeUpdateBindingGeneration", binding.Generation, "afterUpdateBindingGeneration", updatedBinding.Generation)
 			return nil
 		})
 	}

pkg/controllers/rollout/controller_integration_test.go

@@ -22,7 +22,7 @@ import (
 	"k8s.io/utils/ptr"
 
 	fleetv1beta1 "go.goms.io/fleet/apis/placement/v1beta1"
-	"go.goms.io/fleet/pkg/controllers/work"
+	"go.goms.io/fleet/pkg/controllers/workapplier"
 	"go.goms.io/fleet/pkg/utils"
 	"go.goms.io/fleet/pkg/utils/condition"
 )
@@ -1045,7 +1045,7 @@ func markBindingAvailable(binding *fleetv1beta1.ClusterResourceBinding, trackabl
 	Eventually(func() error {
 		reason := "trackable"
 		if !trackable {
-			reason = work.WorkNotTrackableReason
+			reason = workapplier.WorkNotAllManifestsTrackableReason
 		}
 		binding.SetConditions(metav1.Condition{
 			Type:               string(fleetv1beta1.ResourceBindingAvailable),

pkg/controllers/rollout/controller_test.go

@@ -26,7 +26,7 @@ import (
 	clusterv1beta1 "go.goms.io/fleet/apis/cluster/v1beta1"
 	fleetv1alpha1 "go.goms.io/fleet/apis/placement/v1alpha1"
 	fleetv1beta1 "go.goms.io/fleet/apis/placement/v1beta1"
-	"go.goms.io/fleet/pkg/controllers/work"
+	"go.goms.io/fleet/pkg/controllers/workapplier"
 	"go.goms.io/fleet/pkg/utils/condition"
 	"go.goms.io/fleet/pkg/utils/controller"
 )
@@ -668,7 +668,7 @@ func TestIsBindingReady(t *testing.T) {
 							LastTransitionTime: metav1.Time{
 								Time: now.Add(-time.Millisecond),
 							},
-							Reason: work.WorkNotTrackableReason,
+							Reason: condition.WorkNotAvailabilityTrackableReason,
 						},
 					},
 				},
@@ -691,7 +691,30 @@ func TestIsBindingReady(t *testing.T) {
 							LastTransitionTime: metav1.Time{
 								Time: now.Add(time.Millisecond),
 							},
-							Reason: work.WorkNotTrackableReason,
+							Reason: condition.WorkNotAvailabilityTrackableReason,
+						},
+					},
+				},
+			},
+			readyTimeCutOff: now,
+			wantReady:       false,
+			wantWaitTime:    time.Millisecond,
+		},
+		"binding available (not trackable with old reason) after the ready time cut off should return not ready with a wait time": {
+			binding: &fleetv1beta1.ClusterResourceBinding{
+				ObjectMeta: metav1.ObjectMeta{
+					Generation: 10,
+				},
+				Status: fleetv1beta1.ResourceBindingStatus{
+					Conditions: []metav1.Condition{
+						{
+							Type:               string(fleetv1beta1.ResourceBindingAvailable),
+							Status:             metav1.ConditionTrue,
+							ObservedGeneration: 10,
+							LastTransitionTime: metav1.Time{
+								Time: now.Add(time.Millisecond),
+							},
+							Reason: workapplier.WorkNotAllManifestsTrackableReason,
 						},
 					},
 				},
@@ -804,7 +827,7 @@ func TestIsBindingReady(t *testing.T) {
 							Type:               string(fleetv1beta1.ResourceBindingAvailable),
 							Status:             metav1.ConditionTrue,
 							LastTransitionTime: metav1.NewTime(now.Add(-5 * time.Second)),
-							Reason:             work.WorkAvailableReason,
+							Reason:             workapplier.WorkAllManifestsAvailableReason,
 							ObservedGeneration: 10,
 						},
 						{
@@ -2275,7 +2298,7 @@ func generateReadyClusterResourceBinding(state fleetv1beta1.BindingState, resour
 		{
 			Type:   string(fleetv1beta1.ResourceBindingAvailable),
 			Status: metav1.ConditionTrue,
-			Reason: work.WorkAvailableReason, // Make it ready
+			Reason: workapplier.WorkAllManifestsAvailableReason, // Make it ready
 		},
 	}
 	return binding
@@ -2292,7 +2315,7 @@ func generateNotTrackableClusterResourceBinding(state fleetv1beta1.BindingState,
 			Type:               string(fleetv1beta1.ResourceBindingAvailable),
 			Status:             metav1.ConditionTrue,
 			LastTransitionTime: lastTransitionTime,
-			Reason:             work.WorkNotTrackableReason, // Make it not ready
+			Reason:             condition.WorkNotAvailabilityTrackableReason, // Make it not ready
 		},
 	}
 	return binding

pkg/controllers/workapplier/controller.go

@@ -44,6 +44,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 
 	fleetv1beta1 "go.goms.io/fleet/apis/placement/v1beta1"
+	"go.goms.io/fleet/pkg/controllers/work"
 	"go.goms.io/fleet/pkg/utils/controller"
 	"go.goms.io/fleet/pkg/utils/defaulter"
 	"go.goms.io/fleet/pkg/utils/parallelizer"
@@ -113,7 +114,18 @@ func NewReconciler(
 
 var (
 	// Some exported reasons for Work object conditions. Currently only the untrackable reason is being actively used.
-	WorkNotAllManifestsTrackableReason    = "SomeManifestsAreNotAvailabilityTrackable"
+
+	// This is a new reason for the Availability condition when the manifests are not
+	// trackable for availability. This value is currently unused.
+	//
+	// TO-DO (chenyu1): switch to the new reason after proper rollout.
+	WorkNotAllManifestsTrackableReasonNew = "SomeManifestsAreNotAvailabilityTrackable"
+	// This reason uses the exact same value as the one kept in the old work applier for
+	// compatibility reasons. It helps guard the case where the member agent is upgraded
+	// before the hub agent.
+	//
+	// TO-DO (chenyu1): switch off the old reason after proper rollout.
+	WorkNotAllManifestsTrackableReason    = work.WorkNotTrackableReason
 	WorkAllManifestsAppliedReason         = "AllManifestsApplied"
 	WorkAllManifestsAvailableReason       = "AllManifestsAvailable"
 	WorkAllManifestsDiffReportedReason    = "AllManifestsDiffReported"
@@ -143,7 +155,8 @@ const (
 	ManifestProcessingApplyResultTypeNotTakenOver                   manifestProcessingAppliedResultType = "NotTakenOver"
 	ManifestProcessingApplyResultTypeFailedToRunDriftDetection      manifestProcessingAppliedResultType = "FailedToRunDriftDetection"
 	ManifestProcessingApplyResultTypeFoundDrifts                    manifestProcessingAppliedResultType = "FoundDrifts"
-	ManifestProcessingApplyResultTypeFailedToApply                  manifestProcessingAppliedResultType = "FailedToApply"
+	// Note that the reason string below uses the same value as kept in the old work applier.
+	ManifestProcessingApplyResultTypeFailedToApply manifestProcessingAppliedResultType = "ManifestApplyFailed"
 
 	// The result type and description for partially successfully processing attempts.
 	ManifestProcessingApplyResultTypeAppliedWithFailedDriftDetection manifestProcessingAppliedResultType = "AppliedWithFailedDriftDetection"
@@ -168,13 +181,17 @@ const (
 	// The result type for availability check failures.
 	ManifestProcessingAvailabilityResultTypeFailed ManifestProcessingAvailabilityResultType = "Failed"
 
+	// The description for availability check failures.
 	ManifestProcessingAvailabilityResultTypeFailedDescription = "Failed to track the availability of the applied manifest (error = %s)"
 
 	// The result types for completed availability checks.
-	ManifestProcessingAvailabilityResultTypeAvailable       ManifestProcessingAvailabilityResultType = "Available"
-	ManifestProcessingAvailabilityResultTypeNotYetAvailable ManifestProcessingAvailabilityResultType = "NotYetAvailable"
-	ManifestProcessingAvailabilityResultTypeNotTrackable    ManifestProcessingAvailabilityResultType = "NotTrackable"
+	ManifestProcessingAvailabilityResultTypeAvailable ManifestProcessingAvailabilityResultType = "Available"
+	// Note that the reason string below uses the same value as kept in the old work applier.
+	ManifestProcessingAvailabilityResultTypeNotYetAvailable ManifestProcessingAvailabilityResultType = "ManifestNotAvailableYet"
+
+	ManifestProcessingAvailabilityResultTypeNotTrackable ManifestProcessingAvailabilityResultType = "NotTrackable"
 
+	// The descriptions for completed availability checks.
 	ManifestProcessingAvailabilityResultTypeAvailableDescription       = "Manifest is available"
 	ManifestProcessingAvailabilityResultTypeNotYetAvailableDescription = "Manifest is not yet available; Fleet will check again later"
 	ManifestProcessingAvailabilityResultTypeNotTrackableDescription    = "Manifest's availability is not trackable; Fleet assumes that the applied manifest is available"

pkg/controllers/workgenerator/controller.go

@@ -44,7 +44,7 @@ import (
 
 	clusterv1beta1 "go.goms.io/fleet/apis/cluster/v1beta1"
 	fleetv1beta1 "go.goms.io/fleet/apis/placement/v1beta1"
-	"go.goms.io/fleet/pkg/controllers/work"
+	"go.goms.io/fleet/pkg/controllers/workapplier"
 	"go.goms.io/fleet/pkg/utils"
 	"go.goms.io/fleet/pkg/utils/condition"
 	"go.goms.io/fleet/pkg/utils/controller"
@@ -1159,8 +1159,21 @@ func setAllWorkAvailableCondition(works map[string]*fleetv1beta1.Work, binding *
 	for _, w := range works {
 		availableCond := meta.FindStatusCondition(w.Status.Conditions, fleetv1beta1.WorkConditionTypeAvailable)
 		switch {
-		case condition.IsConditionStatusTrue(availableCond, w.GetGeneration()) && availableCond.Reason == work.WorkNotTrackableReason:
-			// The Work object has completed the availability check successfully, due to the resources being untrackable.
+		case condition.IsConditionStatusTrue(availableCond, w.GetGeneration()) && availableCond.Reason == workapplier.WorkNotAllManifestsTrackableReasonNew:
+			// The Work object has completed the availability check successfully, due to the
+			// resources being untrackable.
+			//
+			// This branch is currently never visited as the work applier would still populate
+			// the Available condition using the old reason string for compatibility reasons.
+			if firstWorkWithSuccessfulAvailabilityCheckDueToUntrackableRes == nil {
+				firstWorkWithSuccessfulAvailabilityCheckDueToUntrackableRes = w
+			}
+		case condition.IsConditionStatusTrue(availableCond, w.GetGeneration()) && availableCond.Reason == workapplier.WorkNotAllManifestsTrackableReason:
+			// The Work object has completed the availability check successfully, due to the
+			// resources being untrackable. This is the same branch as the one above but checks
+			// for the old reason string; it is kept for compatibility reasons.
+			//
+			// TO-DO (chenyu1): drop this branch after the rollout completes.
 			if firstWorkWithSuccessfulAvailabilityCheckDueToUntrackableRes == nil {
 				firstWorkWithSuccessfulAvailabilityCheckDueToUntrackableRes = w
 			}
@@ -1218,7 +1231,7 @@ func setAllWorkAvailableCondition(works map[string]*fleetv1beta1.Work, binding *
 		meta.SetStatusCondition(&binding.Status.Conditions, metav1.Condition{
 			Status:             metav1.ConditionTrue,
 			Type:               string(fleetv1beta1.ResourceBindingAvailable),
-			Reason:             work.WorkNotTrackableReason,
+			Reason:             condition.WorkNotAvailabilityTrackableReason,
 			Message:            fmt.Sprintf("The availability of work object %s is not trackable", firstWorkWithSuccessfulAvailabilityCheckDueToUntrackableRes.Name),
 			ObservedGeneration: binding.GetGeneration(),
 		})

pkg/controllers/workgenerator/controller_integration_test.go

@@ -29,7 +29,6 @@ import (
 	clusterv1beta1 "go.goms.io/fleet/apis/cluster/v1beta1"
 	placementv1alpha1 "go.goms.io/fleet/apis/placement/v1alpha1"
 	placementv1beta1 "go.goms.io/fleet/apis/placement/v1beta1"
-	fleetwork "go.goms.io/fleet/pkg/controllers/work"
 	"go.goms.io/fleet/pkg/controllers/workapplier"
 	"go.goms.io/fleet/pkg/utils"
 	"go.goms.io/fleet/pkg/utils/condition"
@@ -2839,7 +2838,7 @@ var _ = Describe("Test Work Generator Controller", func() {
 				// actually handles resources; ConfigMap objects are considered to be
 				// trackable (available immediately after placement). Here it is set
 				// to NotTrackable for testing purposes.
-				Reason:             fleetwork.WorkNotTrackableReason,
+				Reason:             workapplier.WorkNotAllManifestsTrackableReason,
 				ObservedGeneration: work.GetGeneration(),
 			}
 
@@ -3209,7 +3208,7 @@ var _ = Describe("Test Work Generator Controller", func() {
 			availableCond := metav1.Condition{
 				Type:               string(placementv1beta1.ResourceBindingAvailable),
 				Status:             metav1.ConditionTrue,
-				Reason:             fleetwork.WorkNotTrackableReason,
+				Reason:             condition.WorkNotAvailabilityTrackableReason,
 				ObservedGeneration: clusterResourceBinding.GetGeneration(),
 			}