WIP: controller: nrop: per-tree processing and progress

WIP: implement "vertical split" vs current "horizontal split"

Signed-off-by: Francesco Romani <fromani@redhat.com>

Author: ffromani

internal/controller/numaresourcesoperator_controller.go

@@ -64,6 +64,7 @@ import (
 	"github.com/openshift-kni/numaresources-operator/pkg/images"
 	"github.com/openshift-kni/numaresources-operator/pkg/loglevel"
 	"github.com/openshift-kni/numaresources-operator/pkg/objectnames"
+	"github.com/openshift-kni/numaresources-operator/pkg/objectstate"
 	apistate "github.com/openshift-kni/numaresources-operator/pkg/objectstate/api"
 	rtestate "github.com/openshift-kni/numaresources-operator/pkg/objectstate/rte"
 	rteupdate "github.com/openshift-kni/numaresources-operator/pkg/objectupdate/rte"
@@ -243,82 +244,172 @@ func (r *NUMAResourcesOperatorReconciler) reconcileResourceAPI(ctx context.Conte
 	return intreconcile.StepSuccess()
 }
 
-func (r *NUMAResourcesOperatorReconciler) reconcileResourceMachineConfig(ctx context.Context, instance *nropv1.NUMAResourcesOperator, existing *rtestate.ExistingManifests, trees []nodegroupv1.Tree) intreconcile.Step {
-	// we need to sync machine configs first and wait for the MachineConfigPool updates
-	// before checking additional components for updates
-	waitByPool, pausedMCPNames, err := r.syncMachineConfigs(ctx, instance, existing, trees)
+type treeReconcileResult struct {
+	tree           nodegroupv1.Tree
+	mcpStatuses    []nropv1.MachineConfigPool
+	dsInfo         []poolDaemonSet
+	pausedMCPNames []string
+	step           intreconcile.Step
+}
+
+func (r *NUMAResourcesOperatorReconciler) reconcileTreeMachineConfig(ctx context.Context, instance *nropv1.NUMAResourcesOperator, existing *rtestate.ExistingManifests, tree nodegroupv1.Tree) treeReconcileResult {
+	result := treeReconcileResult{tree: tree}
+	singleTreeSlice := []nodegroupv1.Tree{tree}
+
+	waitByPool, pausedMCPNames, err := r.syncMachineConfigs(ctx, instance, existing, singleTreeSlice)
+	result.pausedMCPNames = pausedMCPNames
 	if err != nil {
-		r.Recorder.Eventf(instance, corev1.EventTypeWarning, "FailedMCSync", "Failed to set up machine configuration for worker nodes: %v", err)
-		err = fmt.Errorf("failed to sync machine configs: %w", err)
-		return intreconcile.StepFailed(err)
+		result.step = intreconcile.StepFailed(fmt.Errorf("failed to sync machine configs: %w", err))
+		return result
 	}
-	r.Recorder.Eventf(instance, corev1.EventTypeNormal, "SuccessfulMCSync", "Enabled machine configuration for worker nodes")
 
-	// MCO needs to update the SELinux context removal and other stuff, and need to trigger a reboot.
-	// It can take a while.
-	mcpStatuses, mcpNamePending := syncMachineConfigPoolsStatuses(instance.Name, trees, r.ForwardMCPConds, waitByPool)
-	instance.Status.MachineConfigPools = mcpStatuses
-	instance.Status.Conditions = updateMachineConfigPoolPausedCondition(instance.Status.Conditions, instance.Generation, pausedMCPNames)
+	mcpStatuses, mcpNamePending := syncMachineConfigPoolsStatuses(instance.Name, singleTreeSlice, r.ForwardMCPConds, waitByPool)
+	result.mcpStatuses = mcpStatuses
 
 	if mcpNamePending != "" {
-		// the Machine Config Pool still did not apply the machine config, wait for one minute
-		return intreconcile.StepOngoing(numaResourcesRetryPeriod).WithReason("MachineConfigPoolIsUpdating").WithMessage(mcpNamePending + " is updating")
+		result.step = intreconcile.StepOngoing(numaResourcesRetryPeriod).WithReason("MachineConfigPoolIsUpdating").WithMessage(mcpNamePending + " is updating")
+		return result
 	}
-	instance.Status.MachineConfigPools = syncMachineConfigPoolNodeGroupConfigStatuses(instance.Status.MachineConfigPools, trees)
+	result.mcpStatuses = syncMachineConfigPoolNodeGroupConfigStatuses(result.mcpStatuses, singleTreeSlice)
 
-	return intreconcile.StepSuccess()
+	result.step = intreconcile.StepSuccess()
+	return result
 }
 
-func (r *NUMAResourcesOperatorReconciler) reconcileResourceDaemonSet(ctx context.Context, instance *nropv1.NUMAResourcesOperator, existing *rtestate.ExistingManifests, trees []nodegroupv1.Tree) ([]poolDaemonSet, intreconcile.Step) {
-	daemonSetsInfoPerPool, err := r.syncNUMAResourcesOperatorResources(ctx, instance, existing, trees)
+func (r *NUMAResourcesOperatorReconciler) reconcileTreeDaemonSet(ctx context.Context, instance *nropv1.NUMAResourcesOperator, existing *rtestate.ExistingManifests, tree nodegroupv1.Tree) treeReconcileResult {
+	result := treeReconcileResult{tree: tree}
+
+	dsPoolPairs, err := r.syncTreeDaemonSet(ctx, instance, existing, r.RTEManifests, tree)
 	if err != nil {
-		r.Recorder.Eventf(instance, corev1.EventTypeWarning, "FailedRTECreate", "Failed to create Resource-Topology-Exporter DaemonSets: %v", err)
-		err = fmt.Errorf("FailedRTESync: %w", err)
-		return nil, intreconcile.StepFailed(err)
+		result.step = intreconcile.StepFailed(fmt.Errorf("FailedRTESync: %w", err))
+		return result
 	}
 
-	if len(daemonSetsInfoPerPool) == 0 {
-		return nil, intreconcile.StepSuccess()
+	dssWithReadyStatus, dsNamePending, err := r.syncDaemonSetsStatuses(ctx, r.Client, dsPoolPairs)
+	result.dsInfo = dsPoolPairs
+	if err != nil {
+		result.step = intreconcile.StepFailed(err)
+		return result
+	}
+	if dsNamePending != "" {
+		result.step = intreconcile.StepOngoing(5 * time.Second).WithReason("DaemonSetIsUpdating").WithMessage(dsNamePending + " is updating")
+		return result
 	}
 
-	r.Recorder.Eventf(instance, corev1.EventTypeNormal, "SuccessfulRTECreate", "Created Resource-Topology-Exporter DaemonSets")
+	result.dsInfo = dsPoolPairs
+	_ = dssWithReadyStatus // consumed via dsInfo for status aggregation
+	result.step = intreconcile.StepSuccess()
+	return result
+}
 
-	dssWithReadyStatus, dsNamePending, err := r.syncDaemonSetsStatuses(ctx, r.Client, daemonSetsInfoPerPool)
-	instance.Status.DaemonSets = dssWithReadyStatus
-	instance.Status.RelatedObjects = relatedobjects.ResourceTopologyExporter(r.Namespace, dssWithReadyStatus)
-	if err != nil {
-		return nil, intreconcile.StepFailed(err)
+func aggregateTreeResults(results []treeReconcileResult) (intreconcile.Step, []nropv1.MachineConfigPool, []poolDaemonSet, []string) {
+	var allMCPStatuses []nropv1.MachineConfigPool
+	var allDSInfo []poolDaemonSet
+	var allPausedMCPNames []string
+
+	worstStep := intreconcile.StepSuccess()
+	var messages []string
+
+	for _, result := range results {
+		allMCPStatuses = append(allMCPStatuses, result.mcpStatuses...)
+		allPausedMCPNames = append(allPausedMCPNames, result.pausedMCPNames...)
+
+		if result.step.Done() {
+			allDSInfo = append(allDSInfo, result.dsInfo...)
+			continue
+		}
+
+		if result.step.Error != nil {
+			messages = append(messages, result.step.Error.Error())
+		} else if result.step.Result.RequeueAfter > 0 {
+			messages = append(messages, result.step.ConditionInfo.Message)
+		}
+
+		if shouldReplaceStep(worstStep, result.step) {
+			worstStep = result.step
+		}
 	}
-	if dsNamePending != "" {
-		return nil, intreconcile.StepOngoing(5 * time.Second).WithReason("DaemonSetIsUpdating").WithMessage(dsNamePending + " is updating")
+
+	if !worstStep.Done() && len(messages) > 0 {
+		worstStep = worstStep.WithMessage(strings.Join(messages, "; "))
 	}
 
-	return daemonSetsInfoPerPool, intreconcile.StepSuccess()
+	return worstStep, allMCPStatuses, allDSInfo, allPausedMCPNames
+}
+
+func shouldReplaceStep(current, candidate intreconcile.Step) bool {
+	if current.Done() {
+		return true
+	}
+	if candidate.Error != nil && current.Error == nil {
+		return true
+	}
+	if candidate.Result.RequeueAfter > 0 && current.Result.RequeueAfter > 0 {
+		return candidate.Result.RequeueAfter < current.Result.RequeueAfter
+	}
+	return false
 }
 
 func (r *NUMAResourcesOperatorReconciler) reconcileResource(ctx context.Context, instance *nropv1.NUMAResourcesOperator, trees []nodegroupv1.Tree) intreconcile.Step {
 	if step := r.reconcileResourceAPI(ctx, instance, trees); step.EarlyStop() {
 		return step
 	}
 
-	existing := rtestate.FromClient(ctx, r.Client, r.Platform, r.RTEManifests, instance, trees, r.Namespace)
+	existing := rtestate.FromClientShared(ctx, r.Client, r.Platform, r.RTEManifests, instance, r.Namespace)
 
+	if err := r.prepareAndApplySharedResources(ctx, instance, existing); err != nil {
+		r.Recorder.Eventf(instance, corev1.EventTypeWarning, "FailedRTECreate", "Failed to create Resource-Topology-Exporter shared resources: %v", err)
+		return intreconcile.StepFailed(fmt.Errorf("FailedSharedResourceSync: %w", err))
+	}
+
+	err := dangling.DeleteUnusedDaemonSets(r.Client, ctx, instance, trees)
+	if err != nil {
+		klog.ErrorS(err, "failed to deleted unused daemonsets")
+	}
 	if r.Platform == platform.OpenShift {
-		if step := r.reconcileResourceMachineConfig(ctx, instance, existing, trees); step.EarlyStop() {
-			return step
+		err = dangling.DeleteUnusedMachineConfigs(r.Client, ctx, instance, trees)
+		if err != nil {
+			klog.ErrorS(err, "failed to deleted unused machineconfigs")
 		}
 	}
 
-	dsPerPool, step := r.reconcileResourceDaemonSet(ctx, instance, existing, trees)
-	if step.EarlyStop() {
-		return step
+	var results []treeReconcileResult
+	for _, tree := range trees {
+		treeExisting := existing.ForTree(ctx, r.Client, tree)
+
+		if r.Platform == platform.OpenShift {
+			mcResult := r.reconcileTreeMachineConfig(ctx, instance, treeExisting, tree)
+			if mcResult.step.EarlyStop() {
+				results = append(results, mcResult)
+				continue
+			}
+			dsResult := r.reconcileTreeDaemonSet(ctx, instance, treeExisting, tree)
+			dsResult.mcpStatuses = mcResult.mcpStatuses
+			dsResult.pausedMCPNames = mcResult.pausedMCPNames
+			results = append(results, dsResult)
+		} else {
+			dsResult := r.reconcileTreeDaemonSet(ctx, instance, treeExisting, tree)
+			results = append(results, dsResult)
+		}
 	}
 
-	// all fields of NodeGroupStatus are required so publish the status only when all daemonset and MCPs are updated which
-	// is a certain thing if we got to this point otherwise the function would have returned already
-	instance.Status.NodeGroups = syncNodeGroupsStatus(instance, dsPerPool)
+	step, allMCPStatuses, allDSInfo, allPausedMCPNames := aggregateTreeResults(results)
 
-	return intreconcile.StepSuccess()
+	instance.Status.MachineConfigPools = allMCPStatuses
+	instance.Status.Conditions = updateMachineConfigPoolPausedCondition(instance.Status.Conditions, instance.Generation, allPausedMCPNames)
+
+	dssReady := []nropv1.NamespacedName{}
+	for _, dsInfo := range allDSInfo {
+		dssReady = append(dssReady, dsInfo.DaemonSet)
+	}
+	instance.Status.DaemonSets = dssReady
+	instance.Status.RelatedObjects = relatedobjects.ResourceTopologyExporter(r.Namespace, dssReady)
+
+	if step.Done() {
+		instance.Status.NodeGroups = syncNodeGroupsStatus(instance, allDSInfo)
+	}
+
+	return step
 }
 
 func (r *NUMAResourcesOperatorReconciler) syncDaemonSetsStatuses(ctx context.Context, rd client.Reader, daemonSetsInfo []poolDaemonSet) ([]nropv1.NamespacedName, string, error) {
@@ -511,49 +602,55 @@ func getMachineConfigPoolStatusByName(mcpStatuses []nropv1.MachineConfigPool, na
 	return nropv1.MachineConfigPool{Name: name}
 }
 
-func (r *NUMAResourcesOperatorReconciler) syncNUMAResourcesOperatorResources(ctx context.Context, instance *nropv1.NUMAResourcesOperator, existing *rtestate.ExistingManifests, trees []nodegroupv1.Tree) ([]poolDaemonSet, error) {
-	klog.V(4).InfoS("RTESync start", "trees", len(trees))
-	defer klog.V(4).Info("RTESync stop")
-
-	err := dangling.DeleteUnusedDaemonSets(r.Client, ctx, instance, trees)
-	if err != nil {
-		klog.ErrorS(err, "failed to deleted unused daemonsets")
-	}
-
-	if r.Platform == platform.OpenShift {
-		err = dangling.DeleteUnusedMachineConfigs(r.Client, ctx, instance, trees)
+func (r *NUMAResourcesOperatorReconciler) applyObjectStates(ctx context.Context, instance *nropv1.NUMAResourcesOperator, objStates []objectstate.ObjectState) error {
+	for _, objState := range objStates {
+		if objState.Error != nil {
+			klog.Warningf("error loading object: %v", objState.Error)
+		}
+		if objState.UpdateError != nil {
+			return fmt.Errorf("failed to update (%s) %s/%s: %w", objState.Desired.GetObjectKind().GroupVersionKind(), objState.Desired.GetNamespace(), objState.Desired.GetName(), objState.UpdateError)
+		}
+		err := controllerutil.SetControllerReference(instance, objState.Desired, r.Scheme)
 		if err != nil {
-			klog.ErrorS(err, "failed to deleted unused machineconfigs")
+			return fmt.Errorf("failed to set controller reference to %s %s: %w", objState.Desired.GetNamespace(), objState.Desired.GetName(), err)
+		}
+		_, _, err = apply.ApplyObject(ctx, r.Client, objState)
+		if err != nil {
+			return fmt.Errorf("failed to apply (%s) %s/%s: %w", objState.Desired.GetObjectKind().GroupVersionKind(), objState.Desired.GetNamespace(), objState.Desired.GetName(), err)
 		}
 	}
+	return nil
+}
 
+func (r *NUMAResourcesOperatorReconciler) prepareAndApplySharedResources(ctx context.Context, instance *nropv1.NUMAResourcesOperator, existing *rtestate.ExistingManifests) error {
 	rteupdate.DaemonSetRolloutSettings(r.RTEManifests.Core.DaemonSet)
-	err = rteupdate.DaemonSetAffinitySettings(r.RTEManifests.Core.DaemonSet, r.RTEManifests.Core.DaemonSet.Spec.Template.Labels)
+	err := rteupdate.DaemonSetAffinitySettings(r.RTEManifests.Core.DaemonSet, r.RTEManifests.Core.DaemonSet.Spec.Template.Labels)
 	if err != nil {
 		klog.ErrorS(err, "failed to update RTE affinity settings")
 	}
 
-	dsPoolPairs := []poolDaemonSet{}
-
-	// using a slice of poolDaemonSet instead of a map because Go maps assignment order is not consistent and non-deterministic
 	err = rteupdate.DaemonSetUserImageSettings(r.RTEManifests.Core.DaemonSet, instance.Spec.ExporterImage, r.Images.Preferred(), r.ImagePullPolicy)
 	if err != nil {
-		return dsPoolPairs, err
+		return err
 	}
 
 	err = rteupdate.DaemonSetPauseContainerSettings(r.RTEManifests.Core.DaemonSet)
 	if err != nil {
-		return dsPoolPairs, err
+		return err
 	}
 
 	err = loglevel.UpdatePodSpec(&r.RTEManifests.Core.DaemonSet.Spec.Template.Spec, manifests.ContainerNameRTE, instance.Spec.LogLevel)
 	if err != nil {
-		return dsPoolPairs, err
+		return err
 	}
 
 	rteupdate.SecurityContextConstraint(r.RTEManifests.Core.SecurityContextConstraint, true) // force to legacy context
-	// SCC v2 needs no updates
 
+	return r.applyObjectStates(ctx, instance, existing.SharedState(r.RTEManifests))
+}
+
+func (r *NUMAResourcesOperatorReconciler) syncTreeDaemonSet(ctx context.Context, instance *nropv1.NUMAResourcesOperator, existing *rtestate.ExistingManifests, mf rtestate.Manifests, tree nodegroupv1.Tree) ([]poolDaemonSet, error) {
+	var dsPoolPairs []poolDaemonSet
 	existing = existing.WithManifestsUpdater(func(poolName string, gdm *rtestate.GeneratedDesiredManifest) error {
 		err := daemonsetUpdater(poolName, gdm, r.RTEMetricsTLS)
 		if err != nil {
@@ -563,28 +660,8 @@ func (r *NUMAResourcesOperatorReconciler) syncNUMAResourcesOperatorResources(ctx
 		return nil
 	})
 
-	for _, objState := range existing.State(r.RTEManifests) {
-		if objState.Error != nil {
-			// We are likely in the bootstrap scenario. In this case, which is expected once, everything is fine.
-			// If it happens past bootstrap, still carry on. We know what to do, and we do want to enforce the desired state.
-			klog.Warningf("error loading object: %v", objState.Error)
-		}
-		if objState.UpdateError != nil {
-			// this is an internal error. Should not happen. But if it happen, we don't want to send garbage to the cluster, so we abort
-			return nil, fmt.Errorf("failed to update (%s) %s/%s: %w", objState.Desired.GetObjectKind().GroupVersionKind(), objState.Desired.GetNamespace(), objState.Desired.GetName(), err)
-		}
-		err := controllerutil.SetControllerReference(instance, objState.Desired, r.Scheme)
-		if err != nil {
-			return nil, fmt.Errorf("failed to set controller reference to %s %s: %w", objState.Desired.GetNamespace(), objState.Desired.GetName(), err)
-		}
-		_, _, err = apply.ApplyObject(ctx, r.Client, objState)
-		if err != nil {
-			return nil, fmt.Errorf("failed to apply (%s) %s/%s: %w", objState.Desired.GetObjectKind().GroupVersionKind(), objState.Desired.GetNamespace(), objState.Desired.GetName(), err)
-		}
-	}
-
-	if len(dsPoolPairs) < len(trees) {
-		klog.Warningf("daemonset and tree size mismatch: expected %d got in daemonsets %d", len(trees), len(dsPoolPairs))
+	if err := r.applyObjectStates(ctx, instance, existing.TreeState(mf, tree)); err != nil {
+		return nil, err
 	}
 	return dsPoolPairs, nil
 }

internal/controller/numaresourcesoperator_controller_test.go

@@ -1751,13 +1751,10 @@ var _ = Describe("Test NUMAResourcesOperator Reconcile", func() {
 								Expect(err).ToNot(HaveOccurred())
 							})
 							It("should wait", func() {
-								// check reconcile first loop result
-								// wait one minute to update MCP, thus RTE daemonsets and complete status update is not going to be achieved at this point
 								Expect(result).To(Equal(reconcile.Result{RequeueAfter: time.Minute}))
 
 								Expect(reconciler.Client.Get(context.TODO(), key, nro)).ToNot(HaveOccurred())
-								Expect(nro.Status.MachineConfigPools).To(HaveLen(1))
-								Expect(nro.Status.MachineConfigPools[0].Name).To(Equal("test1"))
+								Expect(nro.Status.MachineConfigPools).To(HaveLen(2))
 							})
 						})
 
@@ -2286,6 +2283,53 @@ var _ = Describe("Test NUMAResourcesOperator Reconcile", func() {
 				Expect(pausedCond.Status).To(Equal(metav1.ConditionTrue))
 			})
 
+			It("should create DaemonSet for ready tree while another tree is still pending", func(ctx context.Context) {
+				// pool1 has custom policy -> MC created, needs MCO rollout
+				// pool2 has no custom annotation -> default policy (no MC to wait for)
+				nro.Spec.NodeGroups[0].Annotations[annotations.SELinuxPolicyConfigAnnotation] = annotations.SELinuxPolicyCustom
+
+				var err error
+				reconciler, err = NewFakeNUMAResourcesOperatorReconciler(platform.OpenShift, defaultOCPVersion, nro, mcp1, mcp2)
+				Expect(err).ToNot(HaveOccurred())
+
+				key := client.ObjectKeyFromObject(nro)
+
+				By("First reconcile: pool1 MC pending, pool2 ready immediately")
+				Expect(reconciler.Reconcile(ctx, reconcile.Request{NamespacedName: key})).To(CauseRequeue())
+
+				By("Verify pool2 DaemonSet exists even though pool1 is still pending")
+				ds := &appsv1.DaemonSet{}
+				mcp2DSKey := client.ObjectKey{
+					Name:      objectnames.GetComponentName(nro.Name, mcp2.Name),
+					Namespace: testNamespace,
+				}
+				Expect(reconciler.Client.Get(ctx, mcp2DSKey, ds)).To(Succeed())
+
+				By("Verify global status is Progressing, not Available")
+				Expect(reconciler.Client.Get(ctx, client.ObjectKeyFromObject(nro), nro)).To(Succeed())
+				progressingCond := getConditionByType(nro.Status.Conditions, status.ConditionProgressing)
+				Expect(progressingCond).ToNot(BeNil())
+				Expect(progressingCond.Status).To(Equal(metav1.ConditionTrue))
+
+				By("Make pool1 ready and reconcile again")
+				Expect(reconciler.Client.Get(ctx, client.ObjectKeyFromObject(mcp1), mcp1)).To(Succeed())
+				ensureMCPIsReady(mcp1, nro.Name)
+				Expect(reconciler.Client.Update(ctx, mcp1)).To(Succeed())
+
+				Expect(reconciler.Reconcile(ctx, reconcile.Request{NamespacedName: key})).ToNot(CauseRequeue())
+
+				By("Now both DaemonSets exist and status is Available")
+				mcp1DSKey := client.ObjectKey{
+					Name:      objectnames.GetComponentName(nro.Name, mcp1.Name),
+					Namespace: testNamespace,
+				}
+				Expect(reconciler.Client.Get(ctx, mcp1DSKey, ds)).To(Succeed())
+				Expect(reconciler.Client.Get(ctx, mcp2DSKey, ds)).To(Succeed())
+
+				Expect(reconciler.Client.Get(ctx, client.ObjectKeyFromObject(nro), nro)).To(Succeed())
+				Expect(nro).To(BeInCondition(status.ConditionAvailable))
+			})
+
 			It("should report paused condition while Progressing", func(ctx context.Context) {
 				// pool1 has custom policy -> MC created, needs MCO rollout
 				// pool2 is paused -> skipped