Repository: GoogleContainerTools/kpt
Pull request #3416 (2022-07-26 15:13:14)

Introduce alpha apply command
First task is a more compact table view, so start by duplicating the table printer logic.


Patch:
diff --git a/commands/alphalivecmd.go b/commands/alphalivecmd.go
index b5794f2f..3c7ea50a 100644
--- a/commands/alphalivecmd.go
+++ b/commands/alphalivecmd.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"os"
 
+	"github.com/GoogleContainerTools/kpt/internal/cmdapply"
 	"github.com/GoogleContainerTools/kpt/internal/cmdplan"
 	"github.com/GoogleContainerTools/kpt/internal/docs/generated/livedocs"
 	"github.com/spf13/cobra"
@@ -39,6 +40,9 @@ func GetAlphaLiveCommand(ctx context.Context, _, version string) *cobra.Command
 
 	f := newFactory(liveCmd, version)
 
+	applyCmd := cmdapply.NewCommand(ctx, f, ioStreams, true)
+	liveCmd.AddCommand(applyCmd)
+
 	planCmd := cmdplan.NewCommand(ctx, f, ioStreams)
 	liveCmd.AddCommand(planCmd)
 
diff --git a/commands/livecmd.go b/commands/livecmd.go
index 880effa7..69e12f71 100644
--- a/commands/livecmd.go
+++ b/commands/livecmd.go
@@ -49,7 +49,7 @@ func GetLiveCommand(ctx context.Context, _, version string) *cobra.Command {
 	// Init command which updates a Kptfile for the ResourceGroup inventory object.
 	klog.V(2).Infoln("init command updates Kptfile for ResourceGroup inventory")
 	initCmd := cmdliveinit.NewCommand(ctx, f, ioStreams)
-	applyCmd := cmdapply.NewCommand(ctx, f, ioStreams)
+	applyCmd := cmdapply.NewCommand(ctx, f, ioStreams, false)
 	destroyCmd := cmddestroy.NewCommand(ctx, f, ioStreams)
 	statusCmd := status.NewCommand(ctx, f)
 	installRGCmd := cmdinstallrg.NewCommand(ctx, f, ioStreams)
diff --git a/internal/alpha/printers/table/collector.go b/internal/alpha/printers/table/collector.go
new file mode 100644
index 00000000..a0352b9e
--- /dev/null
+++ b/internal/alpha/printers/table/collector.go
@@ -0,0 +1,381 @@
+// Copyright 2020 The Kubernetes Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package table
+
+import (
+	"fmt"
+	"sort"
+	"sync"
+
+	"k8s.io/klog/v2"
+	"sigs.k8s.io/cli-utils/pkg/apply/event"
+	pe "sigs.k8s.io/cli-utils/pkg/kstatus/polling/event"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/status"
+	"sigs.k8s.io/cli-utils/pkg/object"
+	"sigs.k8s.io/cli-utils/pkg/object/validation"
+	"sigs.k8s.io/cli-utils/pkg/print/stats"
+	"sigs.k8s.io/cli-utils/pkg/print/table"
+)
+
+const InvalidStatus status.Status = "Invalid"
+
+func newResourceStateCollector(resourceGroups []event.ActionGroup) *resourceStateCollector {
+	resourceInfos := make(map[object.ObjMetadata]*resourceInfo)
+	for _, group := range resourceGroups {
+		action := group.Action
+		// Keep the action that describes the operation for the resource
+		// rather than that we will wait for it.
+		if action == event.WaitAction {
+			continue
+		}
+		for _, identifier := range group.Identifiers {
+			resourceInfos[identifier] = &resourceInfo{
+				identifier: identifier,
+				resourceStatus: &pe.ResourceStatus{
+					Identifier: identifier,
+					Status:     status.UnknownStatus,
+				},
+				ResourceAction: action,
+			}
+		}
+	}
+	return &resourceStateCollector{
+		resourceInfos: resourceInfos,
+	}
+}
+
+// resourceStateCollector consumes the events from the applier
+// eventChannel and keeps track of the latest state for all resources.
+// It also provides functionality for fetching the latest seen
+// state and return it in format that can be used by the
+// BaseTablePrinter.
+type resourceStateCollector struct {
+	mux sync.RWMutex
+
+	// resourceInfos contains a mapping from the unique
+	// resource identifier to a ResourceInfo object that captures
+	// the latest state for the given resource.
+	resourceInfos map[object.ObjMetadata]*resourceInfo
+
+	// stats collect statistics from handled events
+	stats stats.Stats
+
+	err error
+}
+
+// resourceInfo captures the latest seen state of a single resource.
+// This is used for top-level resources that have a ResourceAction
+// associated with them.
+type resourceInfo struct {
+	// identifier contains the information that identifies a
+	// single resource.
+	identifier object.ObjMetadata
+
+	// resourceStatus contains the latest status information
+	// about the resource.
+	resourceStatus *pe.ResourceStatus
+
+	// ResourceAction defines the action we are performing
+	// on this particular resource. This can be either Apply
+	// or Prune.
+	ResourceAction event.ResourceAction
+
+	// Error is set if an error occurred trying to perform
+	// the desired action on the resource.
+	Error error
+
+	// ApplyStatus contains the result after
+	// a resource has been applied to the cluster.
+	ApplyStatus event.ApplyEventStatus
+
+	// PruneStatus contains the result after
+	// a prune operation on a resource
+	PruneStatus event.PruneEventStatus
+
+	// DeleteStatus contains the result after
+	// a delete operation on a resource
+	DeleteStatus event.DeleteEventStatus
+
+	// WaitStatus contains the result after
+	// a wait operation on a resource
+	WaitStatus event.WaitEventStatus
+}
+
+// Identifier returns the identifier for the given resource.
+func (r *resourceInfo) Identifier() object.ObjMetadata {
+	return r.identifier
+}
+
+// ResourceStatus returns the latest seen status for the
+// resource.
+func (r *resourceInfo) ResourceStatus() *pe.ResourceStatus {
+	return r.resourceStatus
+}
+
+// SubResources returns a slice of Resource which contains
+// any resources created and managed by this resource.
+func (r *resourceInfo) SubResources() []table.Resource {
+	var resources []table.Resource
+	for _, res := range r.resourceStatus.GeneratedResources {
+		resources = append(resources, &subResourceInfo{
+			resourceStatus: res,
+		})
+	}
+	return resources
+}
+
+// subResourceInfo captures the latest seen state of a
+// single subResource, i.e. resources that are created and
+// managed by one of the top-level resources we either apply
+// or prune.
+type subResourceInfo struct {
+	// resourceStatus contains the latest status information
+	// about the subResource.
+	resourceStatus *pe.ResourceStatus
+}
+
+// Identifier returns the identifier for the given subResource.
+func (r *subResourceInfo) Identifier() object.ObjMetadata {
+	return r.resourceStatus.Identifier
+}
+
+// ResourceStatus returns the latest seen status for the
+// subResource.
+func (r *subResourceInfo) ResourceStatus() *pe.ResourceStatus {
+	return r.resourceStatus
+}
+
+// SubResources returns a slice of Resource which contains
+// any resources created and managed by this resource.
+func (r *subResourceInfo) SubResources() []table.Resource {
+	var resources []table.Resource
+	for _, res := range r.resourceStatus.GeneratedResources {
+		resources = append(resources, &subResourceInfo{
+			resourceStatus: res,
+		})
+	}
+	return resources
+}
+
+// Listen starts a new goroutine that will listen for events on the
+// provided eventChannel and keep track of the latest state for
+// the resources. The goroutine will exit when the provided
+// eventChannel is closed.
+// The function returns a channel. When this channel is closed, the
+// goroutine has processed all events in the eventChannel and
+// exited.
+func (r *resourceStateCollector) Listen(eventChannel <-chan event.Event) <-chan listenerResult {
+	completed := make(chan listenerResult)
+	go func() {
+		defer close(completed)
+		for ev := range eventChannel {
+			if err := r.processEvent(ev); err != nil {
+				completed <- listenerResult{err: err}
+				return
+			}
+		}
+	}()
+	return completed
+}
+
+type listenerResult struct {
+	err error
+}
+
+// processEvent processes an event and updates the state.
+func (r *resourceStateCollector) processEvent(ev event.Event) error {
+	r.mux.Lock()
+	defer r.mux.Unlock()
+	switch ev.Type {
+	case event.ValidationType:
+		return r.processValidationEvent(ev.ValidationEvent)
+	case event.StatusType:
+		r.processStatusEvent(ev.StatusEvent)
+	case event.ApplyType:
+		r.processApplyEvent(ev.ApplyEvent)
+	case event.PruneType:
+		r.processPruneEvent(ev.PruneEvent)
+	case event.DeleteType:
+		r.processDeleteEvent(ev.DeleteEvent)
+	case event.WaitType:
+		r.processWaitEvent(ev.WaitEvent)
+	case event.ErrorType:
+		return ev.ErrorEvent.Err
+	}
+	return nil
+}
+
+// processValidationEvent handles events pertaining to a validation error
+// for a resource.
+func (r *resourceStateCollector) processValidationEvent(e event.ValidationEvent) error {
+	klog.V(7).Infoln("processing validation event")
+	// unwrap validation errors
+	err := e.Error
+	if vErr, ok := err.(*validation.Error); ok {
+		err = vErr.Unwrap()
+	}
+	if len(e.Identifiers) == 0 {
+		// no objects, invalid event
+		return fmt.Errorf("invalid validation event: no identifiers: %w", err)
+	}
+	for _, id := range e.Identifiers {
+		previous, found := r.resourceInfos[id]
+		if !found {
+			klog.V(4).Infof("%s status event not found in ResourceInfos; no processing", id)
+			continue
+		}
+		previous.resourceStatus = &pe.ResourceStatus{
+			Identifier: id,
+			Status:     InvalidStatus,
+			Message:    e.Error.Error(),
+		}
+	}
+	return nil
+}
+
+// processStatusEvent handles events pertaining to a status
+// update for a resource.
+func (r *resourceStateCollector) processStatusEvent(e event.StatusEvent) {
+	klog.V(7).Infoln("processing status event")
+	previous, found := r.resourceInfos[e.Identifier]
+	if !found {
+		klog.V(4).Infof("%s status event not found in ResourceInfos; no processing", e.Identifier)
+		return
+	}
+	previous.resourceStatus = e.PollResourceInfo
+}
+
+// processApplyEvent handles events relating to apply operations
+func (r *resourceStateCollector) processApplyEvent(e event.ApplyEvent) {
+	identifier := e.Identifier
+	klog.V(7).Infof("processing apply event for %s", identifier)
+	previous, found := r.resourceInfos[identifier]
+	if !found {
+		klog.V(4).Infof("%s apply event not found in ResourceInfos; no processing", identifier)
+		return
+	}
+	if e.Error != nil {
+		previous.Error = e.Error
+	}
+	previous.ApplyStatus = e.Status
+	r.stats.ApplyStats.Inc(e.Status)
+}
+
+// processPruneEvent handles event related to prune operations.
+func (r *resourceStateCollector) processPruneEvent(e event.PruneEvent) {
+	identifier := e.Identifier
+	klog.V(7).Infof("processing prune event for %s", identifier)
+	previous, found := r.resourceInfos[identifier]
+	if !found {
+		klog.V(4).Infof("%s prune event not found in ResourceInfos; no processing", identifier)
+		return
+	}
+	if e.Error != nil {
+		previous.Error = e.Error
+	}
+	previous.PruneStatus = e.Status
+	r.stats.PruneStats.Inc(e.Status)
+}
+
+// processDeleteEvent handles event related to delete operations.
+func (r *resourceStateCollector) processDeleteEvent(e event.DeleteEvent) {
+	identifier := e.Identifier
+	klog.V(7).Infof("processing delete event for %s", identifier)
+	previous, found := r.resourceInfos[identifier]
+	if !found {
+		klog.V(4).Infof("%s delete event not found in ResourceInfos; no processing", identifier)
+		return
+	}
+	if e.Error != nil {
+		previous.Error = e.Error
+	}
+	previous.DeleteStatus = e.Status
+	r.stats.DeleteStats.Inc(e.Status)
+}
+
+// processPruneEvent handles event related to prune operations.
+func (r *resourceStateCollector) processWaitEvent(e event.WaitEvent) {
+	identifier := e.Identifier
+	klog.V(7).Infof("processing wait event for %s", identifier)
+	previous, found := r.resourceInfos[identifier]
+	if !found {
+		klog.V(4).Infof("%s wait event not found in ResourceInfos; no processing", identifier)
+		return
+	}
+	previous.WaitStatus = e.Status
+	r.stats.WaitStats.Inc(e.Status)
+}
+
+// ResourceState contains the latest state for all the resources.
+type ResourceState struct {
+	resourceInfos ResourceInfos
+
+	err error
+}
+
+// Resources returns a slice containing the latest state
+// for each individual resource.
+func (r *ResourceState) Resources() []table.Resource {
+	var resources []table.Resource
+	for _, res := range r.resourceInfos {
+		resources = append(resources, res)
+	}
+	return resources
+}
+
+func (r *ResourceState) Error() error {
+	return r.err
+}
+
+// LatestState returns a ResourceState object that contains
+// a copy of the latest state for all resources.
+func (r *resourceStateCollector) LatestState() *ResourceState {
+	r.mux.RLock()
+	defer r.mux.RUnlock()
+
+	var resourceInfos ResourceInfos
+	for _, ri := range r.resourceInfos {
+		resourceInfos = append(resourceInfos, &resourceInfo{
+			identifier:     ri.identifier,
+			resourceStatus: ri.resourceStatus,
+			ResourceAction: ri.ResourceAction,
+			ApplyStatus:    ri.ApplyStatus,
+			PruneStatus:    ri.PruneStatus,
+			DeleteStatus:   ri.DeleteStatus,
+			WaitStatus:     ri.WaitStatus,
+		})
+	}
+	sort.Sort(resourceInfos)
+
+	return &ResourceState{
+		resourceInfos: resourceInfos,
+		err:           r.err,
+	}
+}
+
+type ResourceInfos []*resourceInfo
+
+func (g ResourceInfos) Len() int {
+	return len(g)
+}
+
+func (g ResourceInfos) Less(i, j int) bool {
+	idI := g[i].identifier
+	idJ := g[j].identifier
+
+	if idI.Namespace != idJ.Namespace {
+		return idI.Namespace < idJ.Namespace
+	}
+	if idI.GroupKind.Group != idJ.GroupKind.Group {
+		return idI.GroupKind.Group < idJ.GroupKind.Group
+	}
+	if idI.GroupKind.Kind != idJ.GroupKind.Kind {
+		return idI.GroupKind.Kind < idJ.GroupKind.Kind
+	}
+	return idI.Name < idJ.Name
+}
+
+func (g ResourceInfos) Swap(i, j int) {
+	g[i], g[j] = g[j], g[i]
+}
diff --git a/internal/alpha/printers/table/printer.go b/internal/alpha/printers/table/printer.go
new file mode 100644
index 00000000..29ee4c52
--- /dev/null
+++ b/internal/alpha/printers/table/printer.go
@@ -0,0 +1,189 @@
+// Copyright 2020 The Kubernetes Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package table
+
+import (
+	"fmt"
+	"io"
+	"time"
+
+	"k8s.io/cli-runtime/pkg/genericclioptions"
+	"sigs.k8s.io/cli-utils/pkg/apply/event"
+	"sigs.k8s.io/cli-utils/pkg/common"
+	printcommon "sigs.k8s.io/cli-utils/pkg/print/common"
+	"sigs.k8s.io/cli-utils/pkg/print/table"
+)
+
+type Printer struct {
+	IOStreams genericclioptions.IOStreams
+}
+
+func (t *Printer) Print(ch <-chan event.Event, _ common.DryRunStrategy, _ bool) error {
+	// Wait for the init event that will give us the set of
+	// resources.
+	var initEvent event.InitEvent
+	for e := range ch {
+		if e.Type == event.InitType {
+			initEvent = e.InitEvent
+			break
+		}
+		// If we get an error event, we just print it and
+		// exit. The error event signals a fatal error.
+		if e.Type == event.ErrorType {
+			return e.ErrorEvent.Err
+		}
+	}
+	// Create a new collector and initialize it with the resources
+	// we are interested in.
+	coll := newResourceStateCollector(initEvent.ActionGroups)
+
+	stop := make(chan struct{})
+
+	// Start the goroutine that is responsible for
+	// printing the latest state on a regular cadence.
+	printCompleted := t.runPrintLoop(coll, stop)
+
+	// Make the collector start listening on the eventChannel.
+	done := coll.Listen(ch)
+
+	// Block until all the collector has shut down. This means the
+	// eventChannel has been closed and all events have been processed.
+	var err error
+	for msg := range done {
+		err = msg.err
+	}
+
+	// Close the stop channel to notify the print goroutine that it should
+	// shut down.
+	close(stop)
+
+	// Wait until the printCompleted channel is closed. This means
+	// the printer has updated the UI with the latest state and
+	// exited from the goroutine.
+	<-printCompleted
+
+	if err != nil {
+		return err
+	}
+	// If no fatal errors happened, we will return a ResultError if
+	// one or more resources failed to apply/prune or reconcile.
+	return printcommon.ResultErrorFromStats(coll.stats)
+}
+
+// columns defines the columns we want to print
+//TODO: We should have the number of columns and their widths be
+// dependent on the space available.
+var (
+	actionColumnDef = table.ColumnDef{
+		// Column containing the resource type and name. Currently it does not
+		// print group or version since those are rarely needed to uniquely
+		// distinguish two resources from each other. Just name and kind should
+		// be enough in almost all cases and saves space in the output.
+		ColumnName:   "action",
+		ColumnHeader: "ACTION",
+		ColumnWidth:  12,
+		PrintResourceFunc: func(w io.Writer, width int, r table.Resource) (int,
+			error) {
+			var resInfo *resourceInfo
+			switch res := r.(type) {
+			case *resourceInfo:
+				resInfo = res
+			default:
+				return 0, nil
+			}
+
+			var text string
+			switch resInfo.ResourceAction {
+			case event.ApplyAction:
+				if resInfo.ApplyStatus != event.ApplyFailed {
+					text = resInfo.ApplyStatus.String()
+				}
+			case event.PruneAction:
+				if resInfo.PruneStatus != event.PruneFailed {
+					text = resInfo.PruneStatus.String()
+				}
+			}
+
+			if len(text) > width {
+				text = text[:width]
+			}
+			_, err := fmt.Fprint(w, text)
+			return len(text), err
+		},
+	}
+
+	reconciledColumnDef = table.ColumnDef{
+		// Column containing the reconciliation status.
+		ColumnName:   "reconciled",
+		ColumnHeader: "RECONCILED",
+		ColumnWidth:  10,
+		PrintResourceFunc: func(w io.Writer, width int, r table.Resource) (
+			int,
+			error,
+		) {
+			var resInfo *resourceInfo
+			switch res := r.(type) {
+			case *resourceInfo:
+				resInfo = res
+			default:
+				return 0, nil
+			}
+
+			var text string
+			switch resInfo.ResourceAction {
+			case event.WaitAction:
+				text = resInfo.WaitStatus.String()
+			}
+
+			if len(text) > width {
+				text = text[:width]
+			}
+			_, err := fmt.Fprint(w, text)
+			return len(text), err
+		},
+	}
+
+	columns = []table.ColumnDefinition{
+		table.MustColumn("namespace"),
+		table.MustColumn("resource"),
+		actionColumnDef,
+		table.MustColumn("status"),
+		reconciledColumnDef,
+		table.MustColumn("conditions"),
+		table.MustColumn("age"),
+		table.MustColumn("message"),
+	}
+)
+
+// runPrintLoop starts a new goroutine that will regularly fetch the
+// latest state from the collector and update the table.
+func (t *Printer) runPrintLoop(coll *resourceStateCollector, stop chan struct{}) chan struct{} {
+	finished := make(chan struct{})
+
+	baseTablePrinter := table.BaseTablePrinter{
+		IOStreams: t.IOStreams,
+		Columns:   columns,
+	}
+
+	linesPrinted := baseTablePrinter.PrintTable(coll.LatestState(), 0)
+
+	go func() {
+		defer close(finished)
+		ticker := time.NewTicker(500 * time.Millisecond)
+		for {
+			select {
+			case <-stop:
+				ticker.Stop()
+				latestState := coll.LatestState()
+				linesPrinted = baseTablePrinter.PrintTable(latestState, linesPrinted)
+				_, _ = fmt.Fprint(t.IOStreams.Out, "\n")
+				return
+			case <-ticker.C:
+				latestState := coll.LatestState()
+				linesPrinted = baseTablePrinter.PrintTable(latestState, linesPrinted)
+			}
+		}
+	}()
+	return finished
+}
diff --git a/internal/cmdapply/cmdapply.go b/internal/cmdapply/cmdapply.go
index c57981ab..a3786abe 100644
--- a/internal/cmdapply/cmdapply.go
+++ b/internal/cmdapply/cmdapply.go
@@ -20,6 +20,7 @@ import (
 	"os"
 	"time"
 
+	alphaprinterstable "github.com/GoogleContainerTools/kpt/internal/alpha/printers/table"
 	"github.com/GoogleContainerTools/kpt/internal/cmdutil"
 	"github.com/GoogleContainerTools/kpt/internal/docs/generated/livedocs"
 	"github.com/GoogleContainerTools/kpt/internal/util/argutil"
@@ -35,6 +36,7 @@ import (
 	"sigs.k8s.io/cli-utils/pkg/common"
 	"sigs.k8s.io/cli-utils/pkg/inventory"
 	"sigs.k8s.io/cli-utils/pkg/printers"
+	cliutilsprinter "sigs.k8s.io/cli-utils/pkg/printers/printer"
 )
 
 // NewRunner returns a command runner
@@ -42,12 +44,14 @@ func NewRunner(
 	ctx context.Context,
 	factory util.Factory,
 	ioStreams genericclioptions.IOStreams,
+	alpha bool,
 ) *Runner {
 	r := &Runner{
 		ctx:         ctx,
 		ioStreams:   ioStreams,
 		factory:     factory,
 		applyRunner: runApply,
+		alpha:       alpha,
 	}
 	c := &cobra.Command{
 		Use:     "apply [PKG_PATH | -]",
@@ -86,13 +90,14 @@ func NewRunner(
 }
 
 func NewCommand(ctx context.Context, factory util.Factory,
-	ioStreams genericclioptions.IOStreams) *cobra.Command {
-	return NewRunner(ctx, factory, ioStreams).Command
+	ioStreams genericclioptions.IOStreams, alpha bool) *cobra.Command {
+	return NewRunner(ctx, factory, ioStreams, alpha).Command
 }
 
 // Runner contains the run function
 type Runner struct {
 	ctx        context.Context
+	alpha      bool
 	Command    *cobra.Command
 	PreProcess func(info inventory.Info, strategy common.DryRunStrategy) (inventory.Policy, error)
 	ioStreams  genericclioptions.IOStreams
@@ -251,6 +256,13 @@ func runApply(r *Runner, invInfo inventory.Info, objs []*unstructured.Unstructur
 
 	// The printer will print updates from the channel. It will block
 	// until the channel is closed.
-	printer := printers.GetPrinter(r.output, r.ioStreams)
+	var printer cliutilsprinter.Printer
+	if r.alpha && r.output == printers.TablePrinter {
+		printer = &alphaprinterstable.Printer{
+			IOStreams: r.ioStreams,
+		}
+	} else {
+		printer = printers.GetPrinter(r.output, r.ioStreams)
+	}
 	return printer.Print(ch, dryRunStrategy, r.printStatusEvents)
 }
diff --git a/internal/pkg/pkg.go b/internal/pkg/pkg.go
index 68b0bcf6..e4c55fa7 100644
--- a/internal/pkg/pkg.go
+++ b/internal/pkg/pkg.go
@@ -195,7 +195,7 @@ type Pkg struct {
 // Use ReadKptfile or ReadPipeline on the return value to read meta resources from filesystem.
 func New(fs filesys.FileSystem, path string) (*Pkg, error) {
 	if !filepath.IsAbs(path) {
-		return nil, fmt.Errorf("need absolute path")
+		return nil, fmt.Errorf("provided path %s must be absolute", path)
 	}
 	absPath := filepath.Clean(path)
 	pkg := &Pkg{
diff --git a/internal/util/addmergecomment/addmergecomment.go b/internal/util/addmergecomment/addmergecomment.go
index 2b4a3903..94630055 100644
--- a/internal/util/addmergecomment/addmergecomment.go
+++ b/internal/util/addmergecomment/addmergecomment.go
@@ -71,7 +71,6 @@ func Process(paths ...string) error {
 // One known caveat is that upstream meta change can cause downstream origin mismatch. This potentially causes the 3-way merge
 // to fail in pkg update step.
 func addUpstreamAnnotation(object *kyaml.RNode, mergeComment string) error {
-	annotations := object.GetAnnotations()
 	group, _ := resid.ParseGroupVersion(object.GetApiVersion())
 	var name, namespace string
 	if strings.Contains(mergeComment, merge.MergeCommentPrefix) {
@@ -90,8 +89,8 @@ func addUpstreamAnnotation(object *kyaml.RNode, mergeComment string) error {
 	} else if object.GetNamespace() == resid.TotallyNotANamespace {
 		namespace = unknownNamespace
 	}
-	annotations[upstreamIdentifier] = fmt.Sprintf(upstreamIdentifierFmt, group, object.GetKind(), namespace, name)
-	return object.SetAnnotations(annotations)
+	upstreamIdentifierValue := fmt.Sprintf(upstreamIdentifierFmt, group, object.GetKind(), namespace, name)
+	return object.PipeE(kyaml.SetAnnotation(upstreamIdentifier, upstreamIdentifierValue))
 }
 
 // Filter implements kyaml.Filter
diff --git a/porch/pkg/engine/render.go b/porch/pkg/engine/render.go
index e5ad2fd5..922a8d55 100644
--- a/porch/pkg/engine/render.go
+++ b/porch/pkg/engine/render.go
@@ -92,7 +92,7 @@ func writeResources(fs filesys.FileSystem, resources repository.PackageResources
 		if base == "Kptfile" {
 			// Found Kptfile. Check if the current directory is ancestor of the current
 			// topmost package directory. If so, use it instead.
-			if packageDir == "" || strings.HasPrefix(packageDir, dir+"/") {
+			if packageDir == "" || dir == "/" || strings.HasPrefix(packageDir, dir+"/") {
 				packageDir = dir
 			}
 		}
diff --git a/porch/pkg/git/draft.go b/porch/pkg/git/draft.go
index b2adf5ca..79923f14 100644
--- a/porch/pkg/git/draft.go
+++ b/porch/pkg/git/draft.go
@@ -16,6 +16,7 @@ package git
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"path"
@@ -26,7 +27,9 @@ import (
 	"github.com/go-git/go-git/v5"
 	"github.com/go-git/go-git/v5/config"
 	"github.com/go-git/go-git/v5/plumbing"
+	"github.com/go-git/go-git/v5/plumbing/transport"
 	"go.opentelemetry.io/otel/trace"
+	"k8s.io/klog/v2"
 )
 
 type gitPackageDraft struct {
@@ -186,24 +189,44 @@ func (r *gitRepository) closeDraft(ctx context.Context, d *gitPackageDraft) (*gi
 	}, nil
 }
 
+// doGitWithAuth fetches auth information for git and provides it
+// to the provided function which performs the operation against a git repo.
+func (r *gitRepository) doGitWithAuth(ctx context.Context, op func(transport.AuthMethod) error) error {
+	auth, err := r.getAuthMethod(ctx, false)
+	if err != nil {
+		return fmt.Errorf("failed to obtain git credentials: %w", err)
+	}
+	err = op(auth)
+	if err != nil {
+		if !errors.Is(err, transport.ErrAuthenticationRequired) {
+			return err
+		}
+		klog.Infof("Authentication failed. Trying to refresh credentials")
+		// TODO: Consider having some kind of backoff here.
+		auth, err := r.getAuthMethod(ctx, true)
+		if err != nil {
+			return fmt.Errorf("failed to obtain git credentials: %w", err)
+		}
+		return op(auth)
+	}
+	return nil
+}
+
 func (r *gitRepository) commitPackageToMain(ctx context.Context, d *gitPackageDraft) (commitHash, newPackageTreeHash plumbing.Hash, base *plumbing.Reference, err error) {
 	branch := r.branch
 	localRef := branch.RefInLocal()
 
 	var zero plumbing.Hash
-	auth, err := r.getAuthMethod(ctx)
-
-	if err != nil {
-		return zero, zero, nil, fmt.Errorf("failed to obtain git credentials: %w", err)
-	}
 
 	repo := r.repo
 
 	// Fetch main
-	switch err := repo.Fetch(&git.FetchOptions{
-		RemoteName: OriginName,
-		RefSpecs:   []config.RefSpec{branch.ForceFetchSpec()},
-		Auth:       auth,
+	switch err := r.doGitWithAuth(ctx, func(auth transport.AuthMethod) error {
+		return repo.Fetch(&git.FetchOptions{
+			RemoteName: OriginName,
+			RefSpecs:   []config.RefSpec{branch.ForceFetchSpec()},
+			Auth:       auth,
+		})
 	}); err {
 	case nil, git.NoErrAlreadyUpToDate:
 		// ok
diff --git a/porch/pkg/git/git.go b/porch/pkg/git/git.go
index 9212ed12..e5d24de8 100644
--- a/porch/pkg/git/git.go
+++ b/porch/pkg/git/git.go
@@ -22,6 +22,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"
 
 	kptfilev1 "github.com/GoogleContainerTools/kpt/pkg/api/kptfile/v1"
@@ -33,7 +34,6 @@ import (
 	"github.com/go-git/go-git/v5/plumbing"
 	"github.com/go-git/go-git/v5/plumbing/object"
 	"github.com/go-git/go-git/v5/plumbing/transport"
-	"github.com/go-git/go-git/v5/plumbing/transport/http"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/trace"
 	"k8s.io/klog/v2"
@@ -140,9 +140,13 @@ type gitRepository struct {
 	branch             BranchName // The main branch from repository registration (defaults to 'main' if unspecified)
 	directory          string     // Directory within the repository where to look for packages.
 	repo               *git.Repository
-	cachedCredentials  transport.AuthMethod
 	credentialResolver repository.CredentialResolver
 	userInfoProvider   repository.UserInfoProvider
+
+	// credential contains the information needed to authenticate against
+	// a git repository.
+	credential repository.Credential
+	mutex      sync.Mutex
 }
 
 func (r *gitRepository) ListPackageRevisions(ctx context.Context, filter repository.ListPackageRevisionFilter) ([]repository.PackageRevision, error) {
@@ -614,33 +618,26 @@ func (r *gitRepository) dumpAllRefs() {
 	}
 }
 
-func resolveCredential(ctx context.Context, namespace, name string, resolver repository.CredentialResolver) (transport.AuthMethod, error) {
-	cred, err := resolver.ResolveCredential(ctx, namespace, name)
-	if err != nil {
-		return nil, fmt.Errorf("failed to obtain credential from secret %s/%s: %w", namespace, name, err)
-	}
-
-	username := cred.Data["username"]
-	password := cred.Data["password"]
+// getAuthMethod fetches the credentials for authenticating to git. It caches the
+// credentials between calls and refresh credentials when the tokens have expired.
+func (r *gitRepository) getAuthMethod(ctx context.Context, forceRefresh bool) (transport.AuthMethod, error) {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
 
-	return &http.BasicAuth{
-		Username: string(username),
-		Password: string(password),
-	}, nil
-}
+	// If no secret is provided, we try without any auth.
+	if r.secret == "" {
+		return nil, nil
+	}
 
-func (r *gitRepository) getAuthMethod(ctx context.Context) (transport.AuthMethod, error) {
-	if r.cachedCredentials == nil {
-		if r.secret != "" {
-			if auth, err := resolveCredential(ctx, r.namespace, r.secret, r.credentialResolver); err != nil {
-				return nil, err
-			} else {
-				r.cachedCredentials = auth
-			}
+	if r.credential == nil || !r.credential.Valid() || forceRefresh {
+		if cred, err := r.credentialResolver.ResolveCredential(ctx, r.namespace, r.secret); err != nil {
+			return nil, fmt.Errorf("failed to obtain credential from secret %s/%s: %w", r.namespace, r.secret, err)
+		} else {
+			r.credential = cred
 		}
 	}
 
-	return r.cachedCredentials, nil
+	return r.credential.ToAuthMethod(), nil
 }
 
 func (r *gitRepository) getRepo() (string, error) {
@@ -656,16 +653,13 @@ func (r *gitRepository) fetchRemoteRepository(ctx context.Context) error {
 	ctx, span := tracer.Start(ctx, "gitRepository::fetchRemoteRepository", trace.WithAttributes())
 	defer span.End()
 
-	auth, err := r.getAuthMethod(ctx)
-	if err != nil {
-		return err
-	}
-
 	// Fetch
-	switch err := r.repo.Fetch(&git.FetchOptions{
-		RemoteName: OriginName,
-		Auth:       auth,
-		Prune:      git.Prune,
+	switch err := r.doGitWithAuth(ctx, func(auth transport.AuthMethod) error {
+		return r.repo.Fetch(&git.FetchOptions{
+			RemoteName: OriginName,
+			Auth:       auth,
+			Prune:      git.Prune,
+		})
 	}); err {
 	case nil: // OK
 	case git.NoErrAlreadyUpToDate:
@@ -698,11 +692,6 @@ func (r *gitRepository) verifyRepository(opts *GitRepositoryOptions) error {
 // If the branch doesn't exist, will return zero hash and no error.
 func (r *gitRepository) createPackageDeleteCommit(ctx context.Context, branch plumbing.ReferenceName, pkg *gitPackageRevision) (plumbing.Hash, error) {
 	var zero plumbing.Hash
-	auth, err := r.getAuthMethod(ctx)
-	if err != nil {
-		return zero, fmt.Errorf("failed to obtain git credentials: %w", err)
-	}
-
 	repo := r.repo
 
 	local, err := refInRemoteFromRefInLocal(branch)
@@ -711,11 +700,13 @@ func (r *gitRepository) createPackageDeleteCommit(ctx context.Context, branch pl
 	}
 	// Fetch the branch
 	// TODO: Fetch only as part of conflict resolution & Retry
-	switch err := repo.Fetch(&git.FetchOptions{
-		RemoteName: OriginName,
-		RefSpecs:   []config.RefSpec{config.RefSpec(fmt.Sprintf("+%s:%s", local, branch))},
-		Auth:       auth,
-		Tags:       git.NoTags,
+	switch err := r.doGitWithAuth(ctx, func(auth transport.AuthMethod) error {
+		return repo.Fetch(&git.FetchOptions{
+			RemoteName: OriginName,
+			RefSpecs:   []config.RefSpec{config.RefSpec(fmt.Sprintf("+%s:%s", local, branch))},
+			Auth:       auth,
+			Tags:       git.NoTags,
+		})
 	}); err {
 	case nil, git.NoErrAlreadyUpToDate:
 		// ok
@@ -768,24 +759,22 @@ func (r *gitRepository) createPackageDeleteCommit(ctx context.Context, branch pl
 }
 
 func (r *gitRepository) pushAndCleanup(ctx context.Context, ph *pushRefSpecBuilder) error {
-	auth, err := r.getAuthMethod(ctx)
-	if err != nil {
-		return fmt.Errorf("failed to obtain git credentials: %w", err)
-	}
-
 	specs, require, err := ph.BuildRefSpecs()
 	if err != nil {
 		return err
 	}
 
 	klog.Infof("pushing refs: %v", specs)
-	if err := r.repo.Push(&git.PushOptions{
-		RemoteName:        OriginName,
-		RefSpecs:          specs,
-		Auth:              auth,
-		RequireRemoteRefs: require,
-		// TODO(justinsb): Need to ensure this is a compare-and-swap
-		Force: true,
+
+	if err := r.doGitWithAuth(ctx, func(auth transport.AuthMethod) error {
+		return r.repo.Push(&git.PushOptions{
+			RemoteName:        OriginName,
+			RefSpecs:          specs,
+			Auth:              auth,
+			RequireRemoteRefs: require,
+			// TODO(justinsb): Need to ensure this is a compare-and-swap
+			Force: true,
+		})
 	}); err != nil {
 		return err
 	}
diff --git a/porch/pkg/registry/porch/secret.go b/porch/pkg/registry/porch/secret.go
index da0b75d6..1c34373d 100644
--- a/porch/pkg/registry/porch/secret.go
+++ b/porch/pkg/registry/porch/secret.go
@@ -17,11 +17,12 @@ package porch
 import (
 	"context"
 	"fmt"
-	"sync"
 	"time"
 
 	"github.com/GoogleContainerTools/kpt/porch/pkg/registry/porch/wi"
 	"github.com/GoogleContainerTools/kpt/porch/pkg/repository"
+	"github.com/go-git/go-git/v5/plumbing/transport"
+	"github.com/go-git/go-git/v5/plumbing/transport/http"
 	"golang.org/x/oauth2"
 	stsv1 "google.golang.org/api/sts/v1"
 	core "k8s.io/api/core/v1"
@@ -65,19 +66,19 @@ func (r *secretResolver) ResolveCredential(ctx context.Context, namespace, name
 		Namespace: namespace,
 		Name:      name,
 	}, &secret); err != nil {
-		return repository.Credential{}, fmt.Errorf("cannot resolve credentials in a secret %s/%s: %w", namespace, name, err)
+		return nil, fmt.Errorf("cannot resolve credentials in a secret %s/%s: %w", namespace, name, err)
 	}
 
 	for _, resolver := range r.resolverChain {
 		cred, found, err := resolver.Resolve(ctx, secret)
 		if err != nil {
-			return repository.Credential{}, fmt.Errorf("error resolving credential: %w", err)
+			return nil, fmt.Errorf("error resolving credential: %w", err)
 		}
 		if found {
 			return cred, nil
 		}
 	}
-	return repository.Credential{}, &NoMatchingResolverError{
+	return nil, &NoMatchingResolverError{
 		Type: string(secret.Type),
 	}
 }
@@ -108,35 +109,51 @@ type BasicAuthResolver struct{}
 
 func (b *BasicAuthResolver) Resolve(_ context.Context, secret core.Secret) (repository.Credential, bool, error) {
 	if secret.Type != BasicAuthType {
-		return repository.Credential{}, false, nil
+		return nil, false, nil
 	}
 
-	return repository.Credential{
-		Data: secret.Data,
+	return &BasicAuthCredential{
+		Username: string(secret.Data["username"]),
+		Password: string(secret.Data["password"]),
 	}, true, nil
 }
 
+type BasicAuthCredential struct {
+	Username string
+	Password string
+}
+
+var _ repository.Credential = &BasicAuthCredential{}
+
+func (b *BasicAuthCredential) Valid() bool {
+	return true
+}
+
+func (b *BasicAuthCredential) ToAuthMethod() transport.AuthMethod {
+	return &http.BasicAuth{
+		Username: string(b.Username),
+		Password: string(b.Password),
+	}
+}
+
 func NewGcloudWIResolver(corev1Client *corev1client.CoreV1Client, stsClient *stsv1.Service) Resolver {
 	return &GcloudWIResolver{
 		coreV1Client: corev1Client,
 		exchanger:    wi.NewWITokenExchanger(corev1Client, stsClient),
-		tokenCache:   make(map[tokenCacheKey]*oauth2.Token),
+		circuitBreaker: &circuitBreaker{
+			duration:    5 * time.Second, // We always wait at least 5 seconds before trying again.
+			factor:      2,               // We double the wait time for every consecutive failure.
+			maxDuration: 5 * time.Minute, // Max wait time is 5 minutes.
+		},
 	}
 }
 
 var _ Resolver = &GcloudWIResolver{}
 
 type GcloudWIResolver struct {
-	coreV1Client *corev1client.CoreV1Client
-	exchanger    *wi.WITokenExchanger
-
-	mutex      sync.Mutex
-	tokenCache map[tokenCacheKey]*oauth2.Token
-}
-
-type tokenCacheKey struct {
-	ksa types.NamespacedName
-	gsa string
+	coreV1Client   *corev1client.CoreV1Client
+	exchanger      *wi.WITokenExchanger
+	circuitBreaker *circuitBreaker
 }
 
 var porchKSA = types.NamespacedName{
@@ -146,18 +163,20 @@ var porchKSA = types.NamespacedName{
 
 func (g *GcloudWIResolver) Resolve(ctx context.Context, secret core.Secret) (repository.Credential, bool, error) {
 	if secret.Type != WorkloadIdentityAuthType {
-		return repository.Credential{}, false, nil
+		return nil, false, nil
 	}
 
-	token, err := g.getToken(ctx)
+	var token *oauth2.Token
+	err := g.circuitBreaker.do(func() error {
+		var tokenErr error
+		token, tokenErr = g.getToken(ctx)
+		return tokenErr
+	})
 	if err != nil {
-		return repository.Credential{}, true, err
+		return nil, true, err
 	}
-	return repository.Credential{
-		Data: map[string][]byte{
-			"username": []byte("token"), // username doesn't matter here.
-			"password": []byte(token.AccessToken),
-		},
+	return &GcloudWICredential{
+		token: token,
 	}, true, nil
 }
 
@@ -167,31 +186,11 @@ func (g *GcloudWIResolver) getToken(ctx context.Context) (*oauth2.Token, error)
 		return nil, err
 	}
 
-	tokenKey := tokenCacheKey{
-		ksa: porchKSA,
-		gsa: gsa,
-	}
-
-	g.mutex.Lock()
-	token, found := g.tokenCache[tokenKey]
-	g.mutex.Unlock()
-
-	if found {
-		timeLeft := time.Until(token.Expiry)
-		if timeLeft > 5*time.Minute {
-			return token, nil
-		}
-	}
-
-	token, err = g.exchanger.Exchange(ctx, porchKSA, gsa)
+	token, err := g.exchanger.Exchange(ctx, porchKSA, gsa)
 	if err != nil {
 		return nil, err
 	}
 
-	g.mutex.Lock()
-	g.tokenCache[tokenKey] = token
-	g.mutex.Unlock()
-
 	return token, nil
 }
 
@@ -209,3 +208,86 @@ func (g *GcloudWIResolver) lookupGSAEmail(ctx context.Context, name, namespace s
 	}
 	return gsa, nil
 }
+
+type GcloudWICredential struct {
+	token *oauth2.Token
+}
+
+var _ repository.Credential = &GcloudWICredential{}
+
+func (b *GcloudWICredential) Valid() bool {
+	timeLeft := time.Until(b.token.Expiry)
+	return timeLeft > 5*time.Minute
+}
+
+func (b *GcloudWICredential) ToAuthMethod() transport.AuthMethod {
+	return &http.BasicAuth{
+		Username: "token", // username doesn't matter here.
+		Password: string(b.token.AccessToken),
+	}
+}
+
+// circuitBreaker makes sure that failing operations are retried
+// using exponential backoff.
+type circuitBreaker struct {
+	// duration defines how long to wait after the first failure.
+	duration time.Duration
+	// factor is multiplied with the previous wait time after a failure.
+	factor float64
+	// maxDuration is the maximum wait time we require.
+	maxDuration time.Duration
+
+	open       bool
+	delay      time.Duration
+	expiration time.Time
+	lastErr    error
+}
+
+func (cb *circuitBreaker) do(action func() error) error {
+	if cb.open {
+		// If the circuit breaker is open and the required amount of
+		// time hasn't passed yet, we don't retry but instead return
+		// and error wrapping the error from the last attempt.
+		if time.Now().Before(cb.expiration) {
+			return &CircuitBreakerError{
+				Err: cb.lastErr,
+			}
+		}
+	}
+
+	err := action()
+	if err != nil {
+		// After an error, we determine delay before we allow
+		// another attempt.
+		if !cb.open {
+			cb.delay = cb.duration
+		} else {
+			cb.delay = cb.delay * time.Duration(cb.factor)
+		}
+		if cb.delay > cb.maxDuration {
+			cb.delay = cb.maxDuration
+		}
+		// Set the new expiration in the future.
+		cb.expiration = time.Now().Add(cb.delay)
+		cb.open = true
+		cb.lastErr = err
+		return err
+	}
+	// If the opration succeeded, reset the circuit breaker.
+	cb.open = false
+	cb.delay = 0
+	cb.lastErr = nil
+	return nil
+}
+
+type CircuitBreakerError struct {
+	Err error
+}
+
+func (cbe *CircuitBreakerError) Error() string {
+	return fmt.Sprintf("circuit breaker is open. Last error: %s", cbe.Err.Error())
+}
+
+func (cbe *CircuitBreakerError) Unwrap() error {
+	return cbe.Err
+}
diff --git a/porch/pkg/repository/repository.go b/porch/pkg/repository/repository.go
index 3b6d45da..96e30ac3 100644
--- a/porch/pkg/repository/repository.go
+++ b/porch/pkg/repository/repository.go
@@ -20,6 +20,7 @@ import (
 
 	kptfile "github.com/GoogleContainerTools/kpt/pkg/api/kptfile/v1"
 	"github.com/GoogleContainerTools/kpt/porch/api/porch/v1alpha1"
+	"github.com/go-git/go-git/v5/plumbing/transport"
 )
 
 // TODO: 	"sigs.k8s.io/kustomize/kyaml/filesys" FileSystem?
@@ -130,9 +131,9 @@ type FunctionRepository interface {
 // They are located in repository package because repository is one such package though thematically
 // they rather belong to a package of their own.
 
-type Credential struct {
-	// TODO: support different credential types
-	Data map[string][]byte
+type Credential interface {
+	Valid() bool
+	ToAuthMethod() transport.AuthMethod
 }
 
 type CredentialResolver interface {


Test patch:
diff --git a/internal/cmdapply/cmdapply_test.go b/internal/cmdapply/cmdapply_test.go
index 8f8956bf..a19da430 100644
--- a/internal/cmdapply/cmdapply_test.go
+++ b/internal/cmdapply/cmdapply_test.go
@@ -136,7 +136,7 @@ func TestCmd(t *testing.T) {
 			revert := testutil.Chdir(t, w.WorkspaceDirectory)
 			defer revert()
 
-			runner := NewRunner(fake.CtxWithDefaultPrinter(), tf, ioStreams)
+			runner := NewRunner(fake.CtxWithDefaultPrinter(), tf, ioStreams, false)
 			runner.Command.SetArgs(tc.args)
 			runner.applyRunner = func(_ *Runner, inv inventory.Info,
 				_ []*unstructured.Unstructured, _ common.DryRunStrategy) error {
diff --git a/internal/util/addmergecomment/addmergecomment_test.go b/internal/util/addmergecomment/addmergecomment_test.go
index d40f414e..d1c13b52 100644
--- a/internal/util/addmergecomment/addmergecomment_test.go
+++ b/internal/util/addmergecomment/addmergecomment_test.go
@@ -47,7 +47,7 @@ metadata: # kpt-merge: my-space/nginx-deployment
   name: nginx-deployment
   namespace: my-space
   annotations:
-    internal.kpt.dev/upstream-identifier: apps|Deployment|my-space|nginx-deployment
+    internal.kpt.dev/upstream-identifier: 'apps|Deployment|my-space|nginx-deployment'
 spec:
   replicas: 3
  `,
@@ -70,7 +70,7 @@ metadata: # kpt-merge: default/nginx-deployment
   name: nginx-deployment
   namespace: default
   annotations:
-    internal.kpt.dev/upstream-identifier: apps|Deployment|default|nginx-deployment
+    internal.kpt.dev/upstream-identifier: 'apps|Deployment|default|nginx-deployment'
 spec:
   replicas: 3
  `,
@@ -91,7 +91,7 @@ kind: Deployment
 metadata: # kpt-merge: /nginx-deployment
   name: nginx-deployment
   annotations:
-    internal.kpt.dev/upstream-identifier: apps|Deployment|default|nginx-deployment
+    internal.kpt.dev/upstream-identifier: 'apps|Deployment|default|nginx-deployment'
 spec:
   replicas: 3
  `,
@@ -112,7 +112,7 @@ kind: Deployment
 metadata: # kpt-merge: my-space/nginx-deployment
   name: nginx-deployment-new
   annotations:
-    internal.kpt.dev/upstream-identifier: apps|Deployment|my-space|nginx-deployment
+    internal.kpt.dev/upstream-identifier: 'apps|Deployment|my-space|nginx-deployment'
 spec:
   replicas: 3
  `,
@@ -143,6 +143,35 @@ spec:
   path: /spec
   value:
     group: kubeflow.org
+ `,
+		},
+		{
+			name: "Preserve label comments and orders during upstream-id label editing",
+			input: `
+apiVersion: v1
+kind: Namespace
+metadata: # kpt-merge: /some-app
+  name: some-app # kpt-set: ${some-app}
+  annotations:
+    a-custom: value1
+    b-custom: value2
+    ncp/static_snat_ip: 122.122.122.122 # kpt-set: ${gateway-snat-ip}
+
+spec:
+  replicas: 3 # kpt-set: ${other}
+ `,
+			expected: `
+apiVersion: v1
+kind: Namespace
+metadata: # kpt-merge: /some-app
+  name: some-app # kpt-set: ${some-app}
+  annotations:
+    a-custom: value1
+    b-custom: value2
+    ncp/static_snat_ip: 122.122.122.122 # kpt-set: ${gateway-snat-ip}
+    internal.kpt.dev/upstream-identifier: '|Namespace|default|some-app'
+spec:
+  replicas: 3 # kpt-set: ${other}
  `,
 		},
 	}
diff --git a/porch/pkg/engine/clone_test.go b/porch/pkg/engine/clone_test.go
index efb648b0..db5678c2 100644
--- a/porch/pkg/engine/clone_test.go
+++ b/porch/pkg/engine/clone_test.go
@@ -35,6 +35,8 @@ import (
 	gogit "github.com/go-git/go-git/v5"
 	"github.com/go-git/go-git/v5/plumbing"
 	"github.com/go-git/go-git/v5/plumbing/object"
+	"github.com/go-git/go-git/v5/plumbing/transport"
+	githttp "github.com/go-git/go-git/v5/plumbing/transport/http"
 	"github.com/go-git/go-git/v5/storage/memory"
 )
 
@@ -151,10 +153,28 @@ func startGitServer(t *testing.T, repo *gogit.Repository, opts ...git.GitServerO
 	return fmt.Sprintf("http://%s", address)
 }
 
+// TODO(mortent): See if we can restruture the packages to
+// avoid having to create separate implementations of the auth
+// interfaces here.
 type credentialResolver struct {
 	username, password string
 }
 
+type credential struct {
+	username, password string
+}
+
+func (c *credential) Valid() bool {
+	return true
+}
+
+func (c *credential) ToAuthMethod() transport.AuthMethod {
+	return &githttp.BasicAuth{
+		Username: c.username,
+		Password: c.password,
+	}
+}
+
 func randomString(n int) string {
 	const letters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
 	result := make([]byte, n)
@@ -172,11 +192,9 @@ func randomCredentials() *credentialResolver {
 }
 
 func (r *credentialResolver) ResolveCredential(ctx context.Context, namespace, name string) (repository.Credential, error) {
-	return repository.Credential{
-		Data: map[string][]byte{
-			"username": []byte(r.username),
-			"password": []byte(r.password),
-		},
+	return &credential{
+		username: r.username,
+		password: r.password,
 	}, nil
 }
 
diff --git a/porch/pkg/registry/porch/secret_test.go b/porch/pkg/registry/porch/secret_test.go
index 28579442..dbd2ebb4 100644
--- a/porch/pkg/registry/porch/secret_test.go
+++ b/porch/pkg/registry/porch/secret_test.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/GoogleContainerTools/kpt/porch/pkg/repository"
 	"github.com/stretchr/testify/assert"
@@ -63,11 +64,9 @@ func TestCredentialResolver(t *testing.T) {
 					"password": []byte("password"),
 				},
 			},
-			expectedCredential: repository.Credential{
-				Data: map[string][]byte{
-					"username": []byte("username"),
-					"password": []byte("password"),
-				},
+			expectedCredential: &BasicAuthCredential{
+				Username: "username",
+				Password: "password",
 			},
 		},
 		"no resolver for secret type": {
@@ -140,3 +139,33 @@ type fakeResolver struct {
 func (fr *fakeResolver) Resolve(ctx context.Context, secret core.Secret) (repository.Credential, bool, error) {
 	return fr.credential, fr.resolved, fr.err
 }
+
+func TestCircuitBreaker(t *testing.T) {
+	cb := &circuitBreaker{
+		duration:    2 * time.Second,
+		factor:      4,
+		maxDuration: 10 * time.Second,
+	}
+
+	ticker := time.NewTicker(500 * time.Millisecond)
+	timer := time.NewTimer(5 * time.Second)
+
+	actionCounter := 0
+loop:
+	for {
+		select {
+		case <-ticker.C:
+			_ = cb.do(func() error {
+				actionCounter++
+				return fmt.Errorf("error")
+			})
+		case <-timer.C:
+			ticker.Stop()
+			timer.Stop()
+			break loop
+		}
+	}
+	if got, want := actionCounter, 2; got != want {
+		t.Errorf("expected function to be called %d times, but got %d", want, got)
+	}
+}