feat: implement partial apply

api/v1/kustomization_types.go

@@ -167,6 +167,12 @@ type KustomizationSpec struct {
 	// Components specifies relative paths to specifications of other Components.
 	// +optional
 	Components []string `json:"components,omitempty"`
+
+	// PartialApply instructs the controller to apply the kustomization partially
+	// if there are errors during the apply phase.
+	// +kubebuilder:default:=false
+	// +optional
+	PartialApply bool `json:"partialApply,omitempty"`
 }
 
 // CommonMetadata defines the common labels and annotations.

config/crd/bases/kustomize.toolkit.fluxcd.io_kustomizations.yaml

@@ -231,6 +231,12 @@ spec:
                 maxLength: 200
                 minLength: 1
                 type: string
+              partialApply:
+                default: false
+                description: |-
+                  PartialApply instructs the controller to apply the kustomization partially
+                  if there are errors during the apply phase.
+                type: boolean
               patches:
                 description: |-
                   Strategic merge and JSON patches, defined as inline YAML objects,

docs/api/v1/kustomize.md

@@ -380,6 +380,19 @@ resources. When enabled, the HealthChecks are ignored. Defaults to false.</p>
 <p>Components specifies relative paths to specifications of other Components.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>partialApply</code><br>
+<em>
+bool
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>PartialApply instructs the controller to apply the kustomization partially
+if there are errors during the apply phase.</p>
+</td>
+</tr>
 </table>
 </td>
 </tr>
@@ -888,6 +901,19 @@ resources. When enabled, the HealthChecks are ignored. Defaults to false.</p>
 <p>Components specifies relative paths to specifications of other Components.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>partialApply</code><br>
+<em>
+bool
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>PartialApply instructs the controller to apply the kustomization partially
+if there are errors during the apply phase.</p>
+</td>
+</tr>
 </tbody>
 </table>
 </div>

internal/controller/kustomization_controller.go

@@ -29,6 +29,7 @@ import (
 	securejoin "github.com/cyphar/filepath-securejoin"
 	"github.com/fluxcd/pkg/ssa/normalize"
 	ssautil "github.com/fluxcd/pkg/ssa/utils"
+	"golang.org/x/sync/errgroup"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
@@ -420,7 +421,7 @@ func (r *KustomizationReconciler) reconcile(
 	}
 
 	// Validate and apply resources in stages.
-	drifted, changeSet, err := r.apply(ctx, resourceManager, obj, revision, objects)
+	drifted, changeSet, err, errDuringPartialApply := r.apply(ctx, resourceManager, obj, revision, objects)
 	if err != nil {
 		conditions.MarkFalse(obj, meta.ReadyCondition, meta.ReconciliationFailedReason, "%s", err)
 		return err
@@ -467,11 +468,15 @@ func (r *KustomizationReconciler) reconcile(
 	// Set last applied revision.
 	obj.Status.LastAppliedRevision = revision
 
-	// Mark the object as ready.
-	conditions.MarkTrue(obj,
-		meta.ReadyCondition,
-		meta.ReconciliationSucceededReason,
-		fmt.Sprintf("Applied revision: %s", revision))
+	if errDuringPartialApply != nil {
+		conditions.MarkFalse(obj, meta.ReadyCondition, meta.ReconciliationFailedReason, fmt.Sprintf("Applied revision: %s with, err: %s", revision, errDuringPartialApply))
+	} else {
+		// Mark the object as ready.
+		conditions.MarkTrue(obj,
+			meta.ReadyCondition,
+			meta.ReconciliationSucceededReason,
+			fmt.Sprintf("Applied revision: %s", revision))
+	}
 
 	return nil
 }
@@ -657,11 +662,11 @@ func (r *KustomizationReconciler) apply(ctx context.Context,
 	manager *ssa.ResourceManager,
 	obj *kustomizev1.Kustomization,
 	revision string,
-	objects []*unstructured.Unstructured) (bool, *ssa.ChangeSet, error) {
+	objects []*unstructured.Unstructured) (bool, *ssa.ChangeSet, error, error) {
 	log := ctrl.LoggerFrom(ctx)
 
 	if err := normalize.UnstructuredList(objects); err != nil {
-		return false, nil, err
+		return false, nil, err, nil
 	}
 
 	if cmeta := obj.Spec.CommonMetadata; cmeta != nil {
@@ -751,7 +756,7 @@ func (r *KustomizationReconciler) apply(ctx context.Context,
 		if decryptor.IsEncryptedSecret(u) {
 			return false, nil,
 				fmt.Errorf("%s is SOPS encrypted, configuring decryption is required for this secret to be reconciled",
-					ssautil.FmtUnstructured(u))
+					ssautil.FmtUnstructured(u)), nil
 		}
 
 		switch {
@@ -771,7 +776,7 @@ func (r *KustomizationReconciler) apply(ctx context.Context,
 	if len(defStage) > 0 {
 		changeSet, err := manager.ApplyAll(ctx, defStage, applyOpts)
 		if err != nil {
-			return false, nil, err
+			return false, nil, err, nil
 		}
 
 		if changeSet != nil && len(changeSet.Entries) > 0 {
@@ -788,7 +793,7 @@ func (r *KustomizationReconciler) apply(ctx context.Context,
 				Interval: 2 * time.Second,
 				Timeout:  obj.GetTimeout(),
 			}); err != nil {
-				return false, nil, err
+				return false, nil, err, nil
 			}
 		}
 	}
@@ -797,7 +802,7 @@ func (r *KustomizationReconciler) apply(ctx context.Context,
 	if len(classStage) > 0 {
 		changeSet, err := manager.ApplyAll(ctx, classStage, applyOpts)
 		if err != nil {
-			return false, nil, err
+			return false, nil, err, nil
 		}
 
 		if changeSet != nil && len(changeSet.Entries) > 0 {
@@ -814,25 +819,54 @@ func (r *KustomizationReconciler) apply(ctx context.Context,
 				Interval: 2 * time.Second,
 				Timeout:  obj.GetTimeout(),
 			}); err != nil {
-				return false, nil, err
+				return false, nil, err, nil
 			}
 		}
 	}
 
+	var errorDuringPartialApply error = nil
 	// sort by kind, validate and apply all the others objects
 	sort.Sort(ssa.SortableUnstructureds(resStage))
 	if len(resStage) > 0 {
-		changeSet, err := manager.ApplyAll(ctx, resStage, applyOpts)
-		if err != nil {
-			return false, nil, fmt.Errorf("%w\n%s", err, changeSetLog.String())
+		changeSet := ssa.NewChangeSet()
+		if obj.Spec.PartialApply {
+			collectedChanges := make([]ssa.ChangeSetEntry, len(resStage))
+			g := &errgroup.Group{}
+			g.SetLimit(r.ConcurrentSSA)
+
+			for i, resource := range resStage {
+				g.Go(func() error {
+					changeSetEntry, err := manager.Apply(ctx, resource, applyOpts)
+					if err != nil {
+						r.event(obj, revision, eventv1.EventSeverityError, fmt.Sprintf("error during apply for %v: %v", client.ObjectKeyFromObject(resource), err), nil)
+						return err
+					}
+					collectedChanges[i] = *changeSetEntry
+					return nil
+				})
+			}
+			if err := g.Wait(); err != nil {
+				errorDuringPartialApply = err
+			}
+			changeSet.Append(collectedChanges)
+		} else {
+			var err error
+			changeSet, err = manager.ApplyAll(ctx, resStage, applyOpts)
+			if err != nil {
+				return false, nil, fmt.Errorf("%w\n%s", err, changeSetLog.String()), nil
+			}
 		}
 
 		if changeSet != nil && len(changeSet.Entries) > 0 {
 			resultSet.Append(changeSet.Entries)
 
-			log.Info("server-side apply completed", "output", changeSet.ToMap(), "revision", revision)
+			if errorDuringPartialApply != nil {
+				log.Info("server-side partial apply completed with error", "output", changeSet.ToMap(), "revision", revision, "err", errorDuringPartialApply)
+			} else {
+				log.Info("server-side apply completed", "output", changeSet.ToMap(), "revision", revision)
+			}
 			for _, change := range changeSet.Entries {
-				if HasChanged(change.Action) {
+				if change.Action != "" && HasChanged(change.Action) {
 					changeSetLog.WriteString(change.String() + "\n")
 				}
 			}
@@ -845,7 +879,7 @@ func (r *KustomizationReconciler) apply(ctx context.Context,
 		r.event(obj, revision, eventv1.EventSeverityInfo, applyLog, nil)
 	}
 
-	return applyLog != "", resultSet, nil
+	return applyLog != "", resultSet, nil, errorDuringPartialApply
 }
 
 func (r *KustomizationReconciler) checkHealth(ctx context.Context,