Merge pull request GoogleCloudPlatform#2595 from barney-s/refactor-re…

…conciler-2

Compositions: Implicit Getter - first attempt

experiments/compositions/composition/go.mod

@@ -26,7 +26,7 @@ require (
 	sigs.k8s.io/cli-utils v0.35.0
 	sigs.k8s.io/controller-runtime v0.17.2
 	sigs.k8s.io/kubebuilder-declarative-pattern v0.15.0-beta.1.0.20240614185435-a248ed1e894c
-	sigs.k8s.io/kubebuilder-declarative-pattern/applylib v0.0.0-20240813172011-a8a1382c63c4
+	sigs.k8s.io/kubebuilder-declarative-pattern/applylib v0.0.0-20240830014331-1a63d5a3bb9d
 	sigs.k8s.io/kustomize/kstatus v0.0.2-0.20200509233124-065f70705d4d
 	tailscale.com v1.62.0
 )

experiments/compositions/composition/internal/controller/expander_reconciler.go

@@ -246,53 +246,34 @@ func (r *ExpanderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	loggerCR := logger
 	stagesApplied := []string{}
 	values := map[string]interface{}{}
-	waitRequested := false
+	requeueAgain := false
+
+	// ---------- Evaluate and Apply expanders in order ---------------------
 	for index, expander := range compositionCR.Spec.Expanders {
 		planUpdated := false
+		reason := ""
 
 		// ------------------- EVALUATION SECTION -----------------------
-		logger = loggerCR.WithName(expander.Name).WithName("Expand")
-
-		uri, ev, reason, err := r.getExpanderConfig(ctx, expander.Version, expander.Type)
-		if err != nil {
-			logger.Error(err, "Error getting expander version", "expander", expander.Type,
-				"version", expander.Version, "reason", reason)
-			newStatus.AppendErrorCondition(expander.Name, err.Error(), reason)
-			return ctrl.Result{}, err
-		}
-
-		logger.Info("Got valid expander uri", "uri", uri)
-
-		if expanderDebugLogsEnabled {
-			r.Recorder.Event(&inputcr, "Normal", fmt.Sprintf("Running expander stage %d: %s", index, expander.Name), expanderDebugLog(&inputcr))
-		}
 
-		if ev.Spec.Type == compositionv1alpha1.ExpanderTypeJob {
-			reason, err = r.runJob(ctx, logger, &inputcr, expander.Name, planNN.Name, uri, ev.Spec.ImageRegistry)
-		} else {
-			values, planUpdated, reason, err = r.evaluateAndSavePlan(ctx, logger, &inputcr, values, expander, planNN, ev, uri, expanderDebugLogsEnabled)
-			_, iswaitErr := err.(*EvaluateWaitError)
-			if iswaitErr {
-				newStatus.AppendWaitingCondition(expander.Name, err.Error(), reason)
-				// Subsume the error
-				waitRequested = true
-				// continue to apply phase
-				break
-			}
+		values, planUpdated, reason, err = r.evaluate(ctx, logger, &inputcr, planNN, expander, values, expanderDebugLogsEnabled)
+		_, iswaitErr := err.(*EvaluateWaitError)
+		if iswaitErr {
+			newStatus.AppendWaitingCondition(expander.Name, err.Error(), reason)
+			// Subsume the error
+			requeueAgain = true
+			break
 		}
 
 		if err != nil {
-			newStatus.AppendErrorCondition(expander.Name, err.Error(), reason)
 			// Skip apply phase and return
+			newStatus.AppendErrorCondition(expander.Name, err.Error(), reason)
 			return ctrl.Result{}, err
 		}
+
 		// Inject plan.Ready Condition with list of expanders
 		newStatus.ClearCondition(compositionv1alpha1.Ready)
 		message := fmt.Sprintf("Evaluated stage: %s", expander.Name)
 		newStatus.AppendCondition(compositionv1alpha1.Ready, metav1.ConditionFalse, message, "EvaluationPending")
-		if expanderDebugLogsEnabled {
-			r.Recorder.Event(&inputcr, "Normal", fmt.Sprintf("Evaluated expander stage %d: %s", index, expander.Name), expanderDebugLog(&inputcr))
-		}
 
 		// ------------------- APPLIER SECTION -----------------------
 
@@ -390,10 +371,16 @@ func (r *ExpanderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 			logger.Info("Applied successfully but some resources did not become healthy")
 			// Inject plan.Waiting Condition
 			newStatus.AppendWaitingCondition(expander.Name, "Not all resources are healthy", "WaitingForAppliedResources")
-			return ctrl.Result{}, fmt.Errorf("Some applied resources are not healthy")
+
+			// Request a re-reconcile
+			requeueAgain = true
+			break
 		}
 		logger.Info("Applied resources successfully.")
 
+		// Implicit getter: Make the applied objects available in the values passed to subsequent stages
+		values = applier.AddAppliedObjectsIntoValues(values)
+
 		stagesApplied = append(stagesApplied, expander.Name)
 		r.Recorder.Event(&inputcr, "Normal", "ResourcesReconciled", fmt.Sprintf("All applied resources were reconciled. name: %s", expander.Name))
 		// Inject plan.Ready Condition with list of expanders
@@ -418,12 +405,48 @@ func (r *ExpanderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	newStatus.Generation = plancr.GetGeneration()
 	newStatus.CompositionGeneration = compositionCR.GetGeneration()
 	newStatus.CompositionUID = compositionCR.GetUID()
-	if waitRequested {
+	if requeueAgain {
 		return ctrl.Result{RequeueAfter: time.Second * 5}, nil
 	}
 	return ctrl.Result{}, nil
 }
 
+func (r *ExpanderReconciler) evaluate(ctx context.Context, logger logr.Logger,
+	cr *unstructured.Unstructured, planNN types.NamespacedName,
+	expander compositionv1alpha1.Expander, values map[string]interface{},
+	expanderDebugLogsEnabled bool) (map[string]interface{}, bool, string, error) {
+
+	planUpdated := false
+
+	logger = logger.WithName(expander.Name).WithName("Expand")
+
+	uri, ev, reason, err := r.getExpanderConfig(ctx, expander.Version, expander.Type)
+	if err != nil {
+		logger.Error(err, "Error getting expander version", "expander", expander.Type,
+			"version", expander.Version, "reason", reason)
+		return values, planUpdated, reason, err
+	}
+
+	logger.Info("Got valid expander uri", "uri", uri)
+
+	if expanderDebugLogsEnabled {
+		r.Recorder.Event(cr, "Normal", fmt.Sprintf("Running expander stage: %s", expander.Name), expanderDebugLog(cr))
+	}
+
+	if ev.Spec.Type == compositionv1alpha1.ExpanderTypeJob {
+		reason, err = r.runJob(ctx, logger, cr, expander.Name, planNN.Name, uri, ev.Spec.ImageRegistry)
+	} else {
+		values, planUpdated, reason, err = r.evaluateAndSavePlan(ctx, logger, cr, values, expander, planNN, ev, uri, expanderDebugLogsEnabled)
+	}
+
+	if err == nil && expanderDebugLogsEnabled {
+		r.Recorder.Event(cr, "Normal", fmt.Sprintf("Evaluated expander stage: %s", expander.Name), expanderDebugLog(cr))
+	}
+
+	return values, planUpdated, reason, err
+
+}
+
 func (r *ExpanderReconciler) getExpanderConfig(
 	ctx context.Context, inputExpanderVersion string, expanderType string,
 ) (string, *compositionv1alpha1.ExpanderVersion, string, error) {

experiments/compositions/composition/pkg/applier/applier.go

@@ -17,6 +17,7 @@ package applier
 import (
 	"context"
 	"fmt"
+	"strings"
 
 	compositionv1alpha1 "github.com/GoogleCloudPlatform/k8s-config-connector/experiments/compositions/composition/api/v1alpha1"
 	"github.com/GoogleCloudPlatform/k8s-config-connector/experiments/compositions/composition/pkg/cel"
@@ -394,3 +395,70 @@ func (a *Applier) AreResourcesReady() (bool, error) {
 
 	return allReady, nil
 }
+
+func (a *Applier) AddAppliedObjectsIntoValues(values map[string]interface{}) map[string]interface{} {
+	for _, resultObj := range a.results.Objects {
+		if resultObj.Apply.IsPruned {
+			continue
+		}
+		obj := resultObj.LastApplied
+		name := obj.GetName()
+		gvk := obj.GroupVersionKind()
+		kind := strings.ToLower(gvk.Kind)
+
+		// short path: values.<kind>.<name>. May clash
+		_, ok := values[kind]
+		if !ok {
+			values[kind] = map[string]interface{}{}
+		}
+		ref := values[kind].(map[string]interface{})
+
+		_, ok = ref[name]
+		if ok {
+			// Clash !! We will ignore
+			a.logger.Info("Clash when adding applied objects to values.", "kind", kind, "name", name)
+		} else {
+			ref[name] = obj.Object
+		}
+
+		// long path: values.<group>.<kind>.<namespace>.<name> will not clash
+		// Long path may not be practical since the namespace is not part of the composition and
+		// most templating languages dont support nested templatable variable.
+		// ex: {{ values.deployment.teampage.status.something  }} will work in jinja2
+		// but this {{ values.apps.deployment.{{teampage.metadata.namespace}}.status.something }} wont work
+		// So leaving this code commented for now.
+		/*
+			group := strings.ReplaceAll(strings.ToLower(gvk.Group), ".", "_")
+			namespace := obj.GetNamespace()
+
+			if group == "" {
+				group = "core"
+			}
+			_, ok = values[group]
+			if !ok {
+				values[group] = map[string]interface{}{}
+			}
+			ref = values[group].(map[string]interface{})
+
+			_, ok = ref[kind]
+			if !ok {
+				ref[kind] = map[string]interface{}{}
+			}
+			ref = ref[kind].(map[string]interface{})
+
+			_, ok = ref[namespace]
+			if !ok {
+				ref[namespace] = map[string]interface{}{}
+			}
+			ref = ref[namespace].(map[string]interface{})
+
+			_, ok = ref[name]
+			if !ok {
+				ref[name] = map[string]interface{}{}
+			}
+			ref[name] = obj.Object
+		*/
+
+	}
+	return values
+}

experiments/compositions/composition/tests/data/TestCustomStatusCELRule/input.yaml

@@ -124,7 +124,7 @@ spec:
   - name: v1alpha1
     schema:
       openAPIV3Schema:
-        description: Create GCS Bucket with CORS and retention
+        description: Team Page Creation
         properties:
           apiVersion:
             type: string