Issue #615 - have PackageVariant set readiness gate on PackageRevisions

- PackageVariant controller now uses a readiness gate to allow
  a PackageRevision to complete its mutation pipeline before it
  is allowed to be proposed/approved
- refactored conversion of Kptfiles to YAML since readiness condition
  information is stored in the package Kptfile
  - unified all cases to the same kyaml/yaml-based method
    (KptFile.ToYamlString() and ToYamlString(*fn.KubeObject))
  - this ensures consistency in the YAML (indentation, field order etc.)
  - and reduces the chances of Git conflicts when setting and updating
    readiness conditions
- added more info to error message in case of Git conflict when applying
  a patch

nephio-project/nephio#615

controllers/main.go

@@ -137,7 +137,8 @@ func run(ctx context.Context) error {
 		Client: client.Options{
 			Cache: &client.CacheOptions{
 				DisableFor: []client.Object{
-					&porchapi.PackageRevisionResources{}},
+					&porchapi.PackageRevisionResources{},
+					&porchapi.PackageRevision{}},
 			},
 		},
 	}

controllers/packagevariants/pkg/controllers/packagevariant/injection.go

@@ -18,10 +18,11 @@ import (
 	"context"
 	"fmt"
 	"path/filepath"
-	"sigs.k8s.io/kustomize/kyaml/kio"
 	"sort"
 	"strings"
 
+	"sigs.k8s.io/kustomize/kyaml/kio"
+
 	"github.com/GoogleContainerTools/kpt-functions-sdk/go/fn"
 	porchapi "github.com/nephio-project/porch/api/porch/v1alpha1"
 	api "github.com/nephio-project/porch/controllers/packagevariants/api/v1alpha1"
@@ -121,7 +122,11 @@ func ensureConfigInjection(ctx context.Context,
 
 	setInjectionPointConditionsAndGates(kptfile, injectionPoints)
 
-	prr.Spec.Resources["Kptfile"] = kptfile.String()
+	kptfileYaml, err := kptfilev1.ToYamlString(kptfile)
+	if err != nil {
+		return err
+	}
+	prr.Spec.Resources[kptfilev1.KptFileName] = kptfileYaml
 
 	return nil
 }

controllers/packagevariants/pkg/controllers/packagevariant/packagevariant_controller.go

@@ -56,8 +56,24 @@ type PackageVariantReconciler struct {
 const (
 	workspaceNamePrefix = "packagevariant-"
 
-	ConditionTypeStalled = "Stalled" // whether or not the packagevariant object is making progress or not
-	ConditionTypeReady   = "Ready"   // whether or not the reconciliation succeeded
+	ConditionTypeStalled        = "Stalled"                // whether or not the packagevariant object is making progress or not
+	ConditionTypeReady          = "Ready"                  // whether or not the reconciliation succeeded
+	ConditionTypePipelinePassed = "MutationPipelinePassed" // whether or not the mutation pipeline has completed successufully
+)
+
+var (
+	conditionPipelineNotPassed = porchapi.Condition{
+		Type:    ConditionTypePipelinePassed,
+		Status:  porchapi.ConditionFalse,
+		Reason:  "WaitingOnPipeline",
+		Message: "waiting for mutation pipeline to pass",
+	}
+	conditionPipelinePassed = porchapi.Condition{
+		Type:    ConditionTypePipelinePassed,
+		Status:  porchapi.ConditionTrue,
+		Reason:  "PipelinePassed",
+		Message: "mutation pipeline completed successfully",
+	}
 )
 
 //go:generate go run sigs.k8s.io/controller-tools/cmd/controller-gen@v0.16.1 rbac:headerFile=../../../../../scripts/boilerplate.yaml.txt,roleName=porch-controllers-packagevariants webhook paths="." output:rbac:artifacts:config=../../../config/rbac
@@ -334,6 +350,13 @@ func (r *PackageVariantReconciler) ensurePackageVariant(ctx context.Context,
 	if err = r.Client.Create(ctx, newPR); err != nil {
 		return nil, err
 	}
+
+	setPrReadinessGate(newPR, ConditionTypePipelinePassed)
+	setPrStatusCondition(newPR, conditionPipelineNotPassed)
+	if err := r.Client.Update(ctx, newPR); err != nil {
+		return nil, err
+	}
+
 	klog.Infoln(fmt.Sprintf("package variant %q created package revision %q", pv.Name, newPR.Name))
 
 	prr, changed, err := r.calculateDraftResources(ctx, pv, newPR)
@@ -347,6 +370,29 @@ func (r *PackageVariantReconciler) ensurePackageVariant(ctx context.Context,
 		}
 	}
 
+	// TODO: remove if OK to exclude PackageRevision from cache of r.Client
+	//
+	// var tmpClient client.Client
+	// if tmpClient, err = client.New(config.GetConfigOrDie(), client.Options{
+	// 	Cache: &client.CacheOptions{
+	// 		DisableFor: []client.Object{
+	// 			&porchapi.PackageRevisionResources{}},
+	// 	},
+	// }); err != nil {
+	// 	return nil, err
+	// }
+	var refreshedPR porchapi.PackageRevision
+	if err := r.Client.Get(ctx, types.NamespacedName{Name: newPR.GetName(), Namespace: newPR.GetNamespace()}, &refreshedPR); err != nil {
+		return nil, err
+	}
+	newPR.ResourceVersion = refreshedPR.ResourceVersion
+	newPR.Spec.Tasks = refreshedPR.Spec.Tasks
+
+	setPrStatusCondition(newPR, conditionPipelinePassed)
+	if err := r.Client.Update(ctx, newPR); err != nil {
+		return nil, err
+	}
+
 	return []*porchapi.PackageRevision{newPR}, nil
 }
 
@@ -716,10 +762,33 @@ func setTargetStatusConditions(pv *api.PackageVariant, targets []*porchapi.Packa
 		Type:    ConditionTypeReady,
 		Status:  "True",
 		Reason:  "NoErrors",
-		Message: "successfully ensured downstream package variant",
+		Message: "successfully ensured downstream target package revision",
 	})
 }
 
+func setPrReadinessGate(pr *porchapi.PackageRevision, conditionType string) {
+	for _, aGate := range pr.Spec.ReadinessGates {
+		if aGate.ConditionType == conditionType {
+			return
+		}
+	}
+
+	pr.Spec.ReadinessGates = append(pr.Spec.ReadinessGates, porchapi.ReadinessGate{
+		ConditionType: conditionType,
+	})
+}
+
+func setPrStatusCondition(pr *porchapi.PackageRevision, condition porchapi.Condition) {
+	for index, aCondition := range pr.Status.Conditions {
+		if aCondition.Type == condition.Type {
+			pr.Status.Conditions[index] = condition
+			return
+		}
+	}
+
+	pr.Status.Conditions = append(pr.Status.Conditions, condition)
+}
+
 // SetupWithManager sets up the controller with the Manager.
 func (r *PackageVariantReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	if err := api.AddToScheme(mgr.GetScheme()); err != nil {
@@ -1011,7 +1080,11 @@ func ensureKRMFunctions(pv *api.PackageVariant,
 	}
 
 	// update kptfile
-	prr.Spec.Resources[kptfilev1.KptFileName] = kptfile.String()
+	kptfileYaml, err := kptfilev1.ToYamlString(kptfile)
+	if err != nil {
+		return err
+	}
+	prr.Spec.Resources[kptfilev1.KptFileName] = kptfileYaml
 
 	return nil
 }

pkg/engine/engine.go

@@ -200,6 +200,7 @@ func (cad *cadEngine) CreatePackageRevision(ctx context.Context, repositoryObj *
 		return nil, err
 	}
 	repoPkgRev.SetMeta(pkgRevMeta)
+
 	sent := cad.watcherManager.NotifyPackageRevisionChange(watch.Added, repoPkgRev)
 	klog.Infof("engine: sent %d for new PackageRevision %s/%s", sent, repoPkgRev.KubeObjectNamespace(), repoPkgRev.KubeObjectName())
 	return repoPkgRev, nil
@@ -301,7 +302,9 @@ func (cad *cadEngine) UpdatePackageRevision(ctx context.Context, version string,
 		return nil, err
 	}
 
-	cad.taskHandler.DoPRMutations(ctx, repositoryObj.Namespace, repoPr, oldObj, newObj, draft)
+	if err := cad.taskHandler.DoPRMutations(ctx, repositoryObj.Namespace, repoPr, oldObj, newObj, draft); err != nil {
+		return nil, err
+	}
 
 	if err := draft.UpdateLifecycle(ctx, newObj.Spec.Lifecycle); err != nil {
 		return nil, err

pkg/kpt/api/kptfile/v1/types.go

@@ -20,6 +20,7 @@ package v1
 import (
 	"fmt"
 
+	"github.com/GoogleContainerTools/kpt-functions-sdk/go/fn"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"sigs.k8s.io/kustomize/kyaml/yaml"
 )
@@ -85,6 +86,31 @@ type KptFile struct {
 	Status *Status `yaml:"status,omitempty" json:"status,omitempty"`
 }
 
+func FromKubeObject(kptfileKubeObject *fn.KubeObject) (KptFile, error) {
+	var apiKptfile KptFile
+	if err := kptfileKubeObject.As(&apiKptfile); err != nil {
+		return KptFile{}, err
+	}
+	return apiKptfile, nil
+}
+
+func (file *KptFile) ToYamlString() (string, error) {
+	b, err := yaml.MarshalWithOptions(file, &yaml.EncoderOptions{SeqIndent: yaml.WideSequenceStyle})
+	if err != nil {
+		return "", err
+	}
+	return string(b), nil
+}
+
+func ToYamlString(kptfileKubeObject *fn.KubeObject) (string, error) {
+	kptfile, err := FromKubeObject(kptfileKubeObject)
+	if err != nil {
+		return "", err
+	}
+
+	return kptfile.ToYamlString()
+}
+
 // OriginType defines the type of origin for a package.
 type OriginType string
 

pkg/kpt/clone.go

@@ -20,7 +20,6 @@ import (
 
 	internalpkg "github.com/nephio-project/porch/internal/kpt/pkg"
 	kptfilev1 "github.com/nephio-project/porch/pkg/kpt/api/kptfile/v1"
-	"sigs.k8s.io/kustomize/kyaml/yaml"
 )
 
 // TODO: Accept a virtual filesystem or other package abstraction
@@ -38,12 +37,12 @@ func UpdateUpstream(kptfileContents string, name string, upstream kptfilev1.Upst
 		kptfile.Name = name
 	}
 
-	b, err := yaml.MarshalWithOptions(kptfile, &yaml.EncoderOptions{SeqIndent: yaml.WideSequenceStyle})
+	kptfileYaml, err := kptfile.ToYamlString()
 	if err != nil {
 		return "", fmt.Errorf("cannot save Kptfile: %w", err)
 	}
 
-	return string(b), nil
+	return kptfileYaml, nil
 }
 
 func UpdateName(kptfileContents string, name string) (string, error) {
@@ -55,12 +54,12 @@ func UpdateName(kptfileContents string, name string) (string, error) {
 	// update the name of the package
 	kptfile.Name = name
 
-	b, err := yaml.MarshalWithOptions(kptfile, &yaml.EncoderOptions{SeqIndent: yaml.WideSequenceStyle})
+	kptfileYaml, err := kptfile.ToYamlString()
 	if err != nil {
 		return "", fmt.Errorf("cannot save Kptfile: %w", err)
 	}
 
-	return string(b), nil
+	return kptfileYaml, nil
 }
 
 // TODO: accept a virtual filesystem

pkg/registry/porch/packagerevisionresources.go

@@ -154,13 +154,13 @@ func (r *packageRevisionResources) Update(ctx context.Context, name string, objI
 		klog.Infof("update failed to construct UpdatedObject: %v", err)
 		return nil, false, err
 	}
-	newObj, ok := newRuntimeObj.(*api.PackageRevisionResources)
+	newPkgRevResources, ok := newRuntimeObj.(*api.PackageRevisionResources)
 	if !ok {
 		return nil, false, apierrors.NewBadRequest(fmt.Sprintf("expected PackageRevisionResources object, got %T", newRuntimeObj))
 	}
 
 	if updateValidation != nil {
-		err := updateValidation(ctx, newObj, oldApiPkgRevResources)
+		err := updateValidation(ctx, newPkgRevResources, oldApiPkgRevResources)
 		if err != nil {
 			klog.Infof("update failed validation: %v", err)
 			return nil, false, err
@@ -181,7 +181,7 @@ func (r *packageRevisionResources) Update(ctx context.Context, name string, objI
 		return nil, false, apierrors.NewInternalError(fmt.Errorf("error getting repository %v: %w", repositoryID, err))
 	}
 
-	rev, renderStatus, err := r.cad.UpdatePackageResources(ctx, &repositoryObj, oldRepoPkgRev, oldApiPkgRevResources, newObj)
+	rev, renderStatus, err := r.cad.UpdatePackageResources(ctx, &repositoryObj, oldRepoPkgRev, oldApiPkgRevResources, newPkgRevResources)
 	if err != nil {
 		return nil, false, apierrors.NewInternalError(err)
 	}

pkg/repository/repository.go

@@ -85,7 +85,7 @@ type PackageRevision interface {
 	// GetUpstreamLock returns the kpt lock information.
 	GetUpstreamLock(context.Context) (kptfile.Upstream, kptfile.UpstreamLock, error)
 
-	// GetKptfile returns the Kptfile for hte package
+	// GetKptfile returns the Kptfile for the package
 	GetKptfile(context.Context) (kptfile.KptFile, error)
 
 	// GetLock returns the current revision's lock information.

pkg/task/generictaskhandler.go

@@ -15,10 +15,10 @@
 package task
 
 import (
-	"bytes"
 	"context"
 	"fmt"
 
+	sdkfn "github.com/GoogleContainerTools/kpt-functions-sdk/go/fn"
 	api "github.com/nephio-project/porch/api/porch/v1alpha1"
 	configapi "github.com/nephio-project/porch/api/porchconfig/v1alpha1"
 	"github.com/nephio-project/porch/internal/kpt/builtins"
@@ -339,14 +339,9 @@ func createKptfilePatchTask(ctx context.Context, oldPackage repository.PackageRe
 		return nil, false, err
 	}
 
-	var orgKfString string
-	{
-		var buf bytes.Buffer
-		d := yaml.NewEncoder(&buf)
-		if err := d.Encode(kf); err != nil {
-			return nil, false, err
-		}
-		orgKfString = buf.String()
+	var origKfString string
+	if origKfString, err = kf.ToYamlString(); err != nil {
+		return nil, false, fmt.Errorf("cannot read original Kptfile: %w", err)
 	}
 
 	var readinessGates []kptfile.ReadinessGate
@@ -381,15 +376,11 @@ func createKptfilePatchTask(ctx context.Context, oldPackage repository.PackageRe
 	}
 
 	var newKfString string
-	{
-		var buf bytes.Buffer
-		d := yaml.NewEncoder(&buf)
-		if err := d.Encode(kf); err != nil {
-			return nil, false, err
-		}
-		newKfString = buf.String()
+	if newKfString, err = kf.ToYamlString(); err != nil {
+		return nil, false, fmt.Errorf("cannot read Kptfile after updating: %w", err)
 	}
-	patchSpec, err := GeneratePatch(kptfile.KptFileName, orgKfString, newKfString)
+
+	patchSpec, err := GeneratePatch(kptfile.KptFileName, origKfString, newKfString)
 	if err != nil {
 		return nil, false, err
 	}
@@ -534,6 +525,17 @@ func healConfig(old, new map[string]string) (map[string]string, error) {
 
 	healed := out.output.Contents
 
+	var kptfileKubeObject *sdkfn.KubeObject
+	if kptfileKubeObject, err = sdkfn.ParseKubeObject([]byte(healed[kptfile.KptFileName])); err != nil {
+		return nil, err
+	}
+
+	kptfileYaml, err := kptfile.ToYamlString(kptfileKubeObject)
+	if err != nil {
+		return nil, err
+	}
+	healed[kptfile.KptFileName] = kptfileYaml
+
 	for k, v := range extra {
 		healed[k] = v
 	}

pkg/task/patchgen.go

@@ -100,22 +100,23 @@ func (m *applyPatchMutation) apply(ctx context.Context, resources repository.Pac
 				return result, nil, fmt.Errorf("patch had unexpected preamble %q", preamble)
 			}
 
-			if files[0].OldName != patchSpec.File {
+			file := files[0]
+			if file.OldName != patchSpec.File {
 				return result, nil, fmt.Errorf("patch contained unexpected name; got %q, want %q", files[0].OldName, patchSpec.File)
 			}
 
-			if files[0].IsBinary {
+			if file.IsBinary {
 				return result, nil, fmt.Errorf("patch was a binary diff; expected text diff")
 			}
-			if files[0].IsCopy || files[0].IsDelete || files[0].IsNew || files[0].IsRename {
+			if file.IsCopy || file.IsDelete || file.IsNew || file.IsRename {
 				return result, nil, fmt.Errorf("patch was of an unexpected type (copy/delete/new/rename)")
 			}
-			if files[0].OldMode != files[0].NewMode {
+			if file.OldMode != file.NewMode {
 				return result, nil, fmt.Errorf("patch contained file mode change")
 			}
 			var output bytes.Buffer
-			if err := gitdiff.Apply(&output, strings.NewReader(oldContents), files[0]); err != nil {
-				return result, nil, fmt.Errorf("error applying patch: %w", err)
+			if err := gitdiff.Apply(&output, strings.NewReader(oldContents), file); err != nil {
+				return result, nil, fmt.Errorf("error applying patch %q to file %q: %w", patchSpec.Contents, oldContents, err)
 			}
 
 			patched := output.String()

test/e2e/suite.go

@@ -795,11 +795,11 @@ func (t *TestSuite) ParseKptfileF(resources *porchapi.PackageRevisionResources)
 
 func (t *TestSuite) SaveKptfileF(resources *porchapi.PackageRevisionResources, kptfile *kptfilev1.KptFile) {
 	t.Helper()
-	b, err := yaml.MarshalWithOptions(kptfile, &yaml.EncoderOptions{SeqIndent: yaml.WideSequenceStyle})
+	kptfileYaml, err := kptfile.ToYamlString()
 	if err != nil {
 		t.Fatalf("Failed saving Kptfile: %v", err)
 	}
-	resources.Spec.Resources[kptfilev1.KptFileName] = string(b)
+	resources.Spec.Resources[kptfilev1.KptFileName] = kptfileYaml
 }
 
 func (t *TestSuite) FindAndDecodeF(resources *porchapi.PackageRevisionResources, name string, value interface{}) {