Repository: GoogleContainerTools/kpt
Pull request #3154 (2022-05-13 03:37:38)

feat: cache upstream during update
fixes https://github.com/GoogleContainerTools/kpt/issues/2988

This introduces caching for upstream repos so we can skip some git exec commands like `ls-remote` and `fetch` for repos that are already upto date. Looking for some early feedback on the approach and happy to add additional tests/benchmarks. Currently packages are not updated in parallel so I haven't investigated a thread safe approach.

bharathkkb: /cc @droot 
bharathkkb: @droot can you PTAL?
droot: sorry @bharathkkb for the delay. Will take a look at it today.

Patch:
diff --git a/internal/docs/generated/fndocs/docs.go b/internal/docs/generated/fndocs/docs.go
index 54376b73..5f419978 100644
--- a/internal/docs/generated/fndocs/docs.go
+++ b/internal/docs/generated/fndocs/docs.go
@@ -58,18 +58,6 @@ Flags:
     By default, container function is executed as ` + "`" + `nobody` + "`" + ` user. You may want to use
     this flag to run higher privilege operations such as mounting the local filesystem.
   
-  --match-api-version:
-    Select resources matching the given apiVersion.
-  
-  --match-kind
-    Select resources matching the given kind.
-  
-  --match-name:
-    Select resources matching the given name.
-    
-  --match-namespace:
-    Select resources matching the given namespace.
-  
   --env, e:
     List of local environment variables to be exported to the container function.
     By default, none of local environment variables are made available to the
@@ -106,6 +94,18 @@ Flags:
     (DEPRECATED) include-meta-resources is no longer necessary because meta
     resources are included by default with kpt version v1.0.0-beta.15+.
   
+  --match-api-version:
+    Select resources matching the given apiVersion.
+  
+  --match-kind
+    Select resources matching the given kind.
+  
+  --match-name:
+    Select resources matching the given name.
+    
+  --match-namespace:
+    Select resources matching the given namespace.
+  
   --mount:
     List of storage options to enable reading from the local filesytem. By default,
     container functions can not access the local filesystem. It accepts the same options
@@ -125,11 +125,21 @@ Flags:
     3. OUT_DIR_PATH: output resources are written to provided directory.
        The provided directory must not already exist.
   
+  --type, t;
+    Specify the function type. Accept value ` + "`" + `mutator` + "`" + ` (default), ` + "`" + `validator` + "`" + `. 
+    If used with ` + "`" + `--save` + "`" + `, this flag will save the evaluated function to the corresponding
+     Kptfile section: ` + "`" + `.pipeline.mutators` + "`" + ` if type is ` + "`" + `mutator` + "`" + `; ` + "`" + `.pipeline.validators` + "`" + ` if type
+      is ` + "`" + `validator` + "`" + `.
+  
   --results-dir:
     Path to a directory to write structured results. Directory will be created if
     it doesn't exist. Structured results emitted by the functions are aggregated and saved
     to ` + "`" + `results.yaml` + "`" + ` file in the specified directory.
     If not specified, no result files are written to the local filesystem.
+    
+  --save, s:
+    Save the function image and fn-config to Kptfile. Require ` + "`" + ` + "` + "`" + `" + ` + "`" + `--image` + "`" + ` + "` + "`" + `" + ` + "`" + `.
+    
 
 Environment Variables:
 
@@ -148,6 +158,12 @@ var EvalExamples = `
   # execute container my-fn with an input ConfigMap containing ` + "`" + `data: {foo: bar}` + "`" + `
   $ kpt fn eval DIR -i gcr.io/example.com/my-fn:v1.0.0 -- foo=bar
 
+  # execute container my-fn and save it to Kptfile ` + "`" + `pipeline.mutators` + "`" + ` (Default) list.
+  $ kpt fn eval DIR -s -i gcr.io/example.com/my-fn:v1.0.0 -- foo=bar
+
+  # execute container my-fn and save it to Kptfile ` + "`" + `pipeline.validators` + "`" + ` list.
+  $ kpt fn eval DIR -s -t validator -i gcr.io/example.com/my-fn:v1.0.0 -- foo=bar
+
   # execute executable my-fn on the resources in DIR directory and
   # write output back to DIR
   $ kpt fn eval DIR --exec ./my-fn
diff --git a/internal/fnruntime/runner.go b/internal/fnruntime/runner.go
index 1e62b9a6..98c404b2 100644
--- a/internal/fnruntime/runner.go
+++ b/internal/fnruntime/runner.go
@@ -508,19 +508,7 @@ func newFnConfig(fsys filesys.FileSystem, f *kptfilev1.Function, pkgPath types.U
 		// directly use the config from file
 		return node, nil
 	case len(f.ConfigMap) != 0:
-		node = yaml.NewMapRNode(&f.ConfigMap)
-		if node == nil {
-			return nil, nil
-		}
-		// create a ConfigMap only for configMap config
-		configNode := yaml.MustParse(`
-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: function-input
-data: {}
-`)
-		err := configNode.PipeE(yaml.SetField("data", node))
+		configNode, err := NewConfigMap(f.ConfigMap)
 		if err != nil {
 			return nil, errors.E(op, fn, err)
 		}
diff --git a/internal/fnruntime/utils.go b/internal/fnruntime/utils.go
index acfe6d65..b41a3624 100644
--- a/internal/fnruntime/utils.go
+++ b/internal/fnruntime/utils.go
@@ -221,3 +221,30 @@ func annoMatch(node *yaml.RNode, selector kptfilev1.Selector) bool {
 	}
 	return true
 }
+
+func NewConfigMap(data map[string]string) (*yaml.RNode, error) {
+	node := yaml.NewMapRNode(&data)
+	if node == nil {
+		return nil, nil
+	}
+	// create a ConfigMap only for configMap config
+	configMap := yaml.MustParse(`
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: function-input
+data: {}
+`)
+	if err := node.VisitFields(func(node *yaml.MapNode) error {
+		v := node.Value.YNode()
+		v.Tag = yaml.NodeTagString
+		node.Value.SetYNode(v)
+		return nil
+	}); err != nil {
+		return nil, err
+	}
+	if err := configMap.PipeE(yaml.SetField("data", node)); err != nil {
+		return nil, err
+	}
+	return configMap, nil
+}
diff --git a/internal/gitutil/gitutil.go b/internal/gitutil/gitutil.go
index 0a37e4cb..b3c7e34c 100644
--- a/internal/gitutil/gitutil.go
+++ b/internal/gitutil/gitutil.go
@@ -137,13 +137,26 @@ func (g *GitLocalRunner) run(ctx context.Context, verbose bool, command string,
 	}, nil
 }
 
+type NewGitUpstreamRepoOption func(*GitUpstreamRepo)
+
+func WithFetchedRefs(a map[string]bool) NewGitUpstreamRepoOption {
+	return func(g *GitUpstreamRepo) {
+		g.fetchedRefs = a
+	}
+}
+
 // NewGitUpstreamRepo returns a new GitUpstreamRepo for an upstream package.
-func NewGitUpstreamRepo(ctx context.Context, uri string) (*GitUpstreamRepo, error) {
+func NewGitUpstreamRepo(ctx context.Context, uri string, opts ...NewGitUpstreamRepoOption) (*GitUpstreamRepo, error) {
 	const op errors.Op = "gitutil.NewGitUpstreamRepo"
-
 	g := &GitUpstreamRepo{
 		URI: uri,
 	}
+	for _, opt := range opts {
+		opt(g)
+	}
+	if g.fetchedRefs == nil {
+		g.fetchedRefs = map[string]bool{}
+	}
 	if err := g.updateRefs(ctx); err != nil {
 		return nil, errors.E(op, errors.Repo(uri), err)
 	}
@@ -161,6 +174,17 @@ type GitUpstreamRepo struct {
 	// Tags contains all tag refs in the upstream repo as well as the
 	// each of the are referencing.
 	Tags map[string]string
+
+	// fetchedRefs keeps track of refs already fetched from remote
+	fetchedRefs map[string]bool
+}
+
+func (gur *GitUpstreamRepo) GetFetchedRefs() []string {
+	fetchedRefs := make([]string, 0, len(gur.fetchedRefs))
+	for ref := range gur.fetchedRefs {
+		fetchedRefs = append(fetchedRefs, ref)
+	}
+	return fetchedRefs
 }
 
 // updateRefs fetches all refs from the upstream git repo, parses the results
@@ -374,20 +398,26 @@ loop:
 		// commit sha.
 		validFullSha := s == strings.TrimSpace(rr.Stdout)
 		_, resolved := gur.ResolveRef(s)
-
+		// check if ref was previously fetched
+		// we use the ref s as the cache key
+		_, fetched := gur.fetchedRefs[s]
 		switch {
+		case fetched:
+			// skip refetching if previously fetched
+			break
 		case resolved || validFullSha:
 			// If the ref references a branch or a tag, or is a valid commit
-			// sha, we can fetch just a single commit.
+			// sha and has not already been fetched, we can fetch just a single commit.
 			if _, err := gitRunner.RunVerbose(ctx, "fetch", "origin", "--depth=1", s); err != nil {
 				AmendGitExecError(err, func(e *GitExecError) {
 					e.Repo = uri
-					e.Command = "origin"
+					e.Command = "fetch"
 					e.Ref = s
 				})
 				return "", errors.E(op, errors.Git, fmt.Errorf(
 					"error running `git fetch` for ref %q: %w", s, err))
 			}
+			gur.fetchedRefs[s] = true
 		default:
 			// In other situations (like a short commit sha), we have to do
 			// a full fetch from the remote.
@@ -407,6 +437,7 @@ loop:
 				return "", errors.E(op, errors.Git, fmt.Errorf(
 					"error verifying results from fetch: %w", err))
 			}
+			gur.fetchedRefs[s] = true
 			// If we did a full fetch, we already have all refs, so we can just
 			// exit the loop.
 			break loop
diff --git a/internal/util/fetch/fetch.go b/internal/util/fetch/fetch.go
index 6b2040e0..156e58d2 100644
--- a/internal/util/fetch/fetch.go
+++ b/internal/util/fetch/fetch.go
@@ -62,7 +62,7 @@ func (c Command) Run(ctx context.Context) error {
 		Path:    g.Directory,
 		Ref:     g.Ref,
 	}
-	err = cloneAndCopy(ctx, repoSpec, c.Pkg.UniquePath.String())
+	err = NewCloner(repoSpec).cloneAndCopy(ctx, c.Pkg.UniquePath.String())
 	if err != nil {
 		return errors.E(op, c.Pkg.UniquePath, err)
 	}
@@ -94,21 +94,55 @@ func (c Command) validate(kf *kptfilev1.KptFile) error {
 	return nil
 }
 
+// Cloner clones an upstream repo defined by a repoSpec.
+// Optionally, previously cloned repos can be cached
+// rather than recloning them each time.
+type Cloner struct {
+	// repoSpec spec to clone
+	repoSpec *git.RepoSpec
+
+	// cachedRepos
+	cachedRepo map[string]*gitutil.GitUpstreamRepo
+}
+
+type NewClonerOption func(*Cloner)
+
+func WithCachedRepo(r map[string]*gitutil.GitUpstreamRepo) NewClonerOption {
+	return func(c *Cloner) {
+		c.cachedRepo = r
+	}
+}
+
+func NewCloner(r *git.RepoSpec, opts ...NewClonerOption) *Cloner {
+	c := &Cloner{
+		repoSpec: r,
+	}
+	for _, opt := range opts {
+		opt(c)
+	}
+	if c.cachedRepo == nil {
+		c.cachedRepo = make(map[string]*gitutil.GitUpstreamRepo)
+	}
+	return c
+}
+
 // cloneAndCopy fetches the provided repo and copies the content into the
 // directory specified by dest. The provided name is set as `metadata.name`
 // of the Kptfile of the package.
-func cloneAndCopy(ctx context.Context, r *git.RepoSpec, dest string) error {
+func (c *Cloner) cloneAndCopy(ctx context.Context, dest string) error {
 	const op errors.Op = "fetch.cloneAndCopy"
 	pr := printer.FromContextOrDie(ctx)
 
-	err := ClonerUsingGitExec(ctx, r)
+	err := c.ClonerUsingGitExec(ctx)
 	if err != nil {
 		return errors.E(op, errors.Git, types.UniquePath(dest), err)
 	}
-	defer os.RemoveAll(r.Dir)
+	defer os.RemoveAll(c.repoSpec.Dir)
+	// update cache before removing clone dir
+	defer delete(c.cachedRepo, c.repoSpec.CloneSpec())
 
-	sourcePath := filepath.Join(r.Dir, r.Path)
-	pr.Printf("Adding package %q.\n", strings.TrimPrefix(r.Path, "/"))
+	sourcePath := filepath.Join(c.repoSpec.Dir, c.repoSpec.Path)
+	pr.Printf("Adding package %q.\n", strings.TrimPrefix(c.repoSpec.Path, "/"))
 	if err := pkgutil.CopyPackage(sourcePath, dest, true, pkg.All); err != nil {
 		return errors.E(op, types.UniquePath(dest), err)
 	}
@@ -117,7 +151,7 @@ func cloneAndCopy(ctx context.Context, r *git.RepoSpec, dest string) error {
 		return errors.E(op, types.UniquePath(dest), err)
 	}
 
-	if err := kptfileutil.UpdateUpstreamLockFromGit(dest, r); err != nil {
+	if err := kptfileutil.UpdateUpstreamLockFromGit(dest, c.repoSpec); err != nil {
 		return errors.E(op, errors.Git, types.UniquePath(dest), err)
 	}
 	return nil
@@ -129,46 +163,51 @@ func cloneAndCopy(ctx context.Context, r *git.RepoSpec, dest string) error {
 // for versioning multiple kpt packages in a single repo independently. It
 // relies on the private clonerUsingGitExec function to try fetching different
 // refs.
-func ClonerUsingGitExec(ctx context.Context, repoSpec *git.RepoSpec) error {
+func (c *Cloner) ClonerUsingGitExec(ctx context.Context) error {
 	const op errors.Op = "fetch.ClonerUsingGitExec"
 
 	// Create a local representation of the upstream repo. This will initialize
 	// the cache for the specified repo uri if it isn't already there. It also
 	// fetches and caches all tag and branch refs from the upstream repo.
-	upstreamRepo, err := gitutil.NewGitUpstreamRepo(ctx, repoSpec.CloneSpec())
-	if err != nil {
-		return errors.E(op, errors.Git, errors.Repo(repoSpec.CloneSpec()), err)
+	upstreamRepo, exists := c.cachedRepo[c.repoSpec.CloneSpec()]
+	if !exists {
+		newUpstreamRemp, err := gitutil.NewGitUpstreamRepo(ctx, c.repoSpec.CloneSpec())
+		if err != nil {
+			return errors.E(op, errors.Git, errors.Repo(c.repoSpec.CloneSpec()), err)
+		}
+		upstreamRepo = newUpstreamRemp
+		c.cachedRepo[c.repoSpec.CloneSpec()] = upstreamRepo
 	}
 
 	// Check if we have a ref in the upstream that matches the package-specific
 	// reference. If we do, we use that reference.
-	ps := strings.Split(repoSpec.Path, "/")
+	ps := strings.Split(c.repoSpec.Path, "/")
 	for len(ps) != 0 {
 		p := path.Join(ps...)
-		packageRef := path.Join(strings.TrimLeft(p, "/"), repoSpec.Ref)
+		packageRef := path.Join(strings.TrimLeft(p, "/"), c.repoSpec.Ref)
 		if _, found := upstreamRepo.ResolveTag(packageRef); found {
-			repoSpec.Ref = packageRef
+			c.repoSpec.Ref = packageRef
 			break
 		}
 		ps = ps[:len(ps)-1]
 	}
 
 	// Pull the required ref into the repo git cache.
-	dir, err := upstreamRepo.GetRepo(ctx, []string{repoSpec.Ref})
+	dir, err := upstreamRepo.GetRepo(ctx, []string{c.repoSpec.Ref})
 	if err != nil {
-		return errors.E(op, errors.Git, errors.Repo(repoSpec.CloneSpec()), err)
+		return errors.E(op, errors.Git, errors.Repo(c.repoSpec.CloneSpec()), err)
 	}
 
 	gitRunner, err := gitutil.NewLocalGitRunner(dir)
 	if err != nil {
-		return errors.E(op, errors.Git, errors.Repo(repoSpec.CloneSpec()), err)
+		return errors.E(op, errors.Git, errors.Repo(c.repoSpec.CloneSpec()), err)
 	}
 
 	// Find the commit SHA for the ref that was just fetched. We need the SHA
 	// rather than the ref to be able to do a hard reset of the cache repo.
-	commit, found := upstreamRepo.ResolveRef(repoSpec.Ref)
+	commit, found := upstreamRepo.ResolveRef(c.repoSpec.Ref)
 	if !found {
-		commit = repoSpec.Ref
+		commit = c.repoSpec.Ref
 	}
 
 	// Reset the local repo to the commit we need. Doing a hard reset instead of
@@ -178,34 +217,34 @@ func ClonerUsingGitExec(ctx context.Context, repoSpec *git.RepoSpec) error {
 	_, err = gitRunner.Run(ctx, "reset", "--hard", commit)
 	if err != nil {
 		gitutil.AmendGitExecError(err, func(e *gitutil.GitExecError) {
-			e.Repo = repoSpec.CloneSpec()
+			e.Repo = c.repoSpec.CloneSpec()
 			e.Ref = commit
 		})
-		return errors.E(op, errors.Git, errors.Repo(repoSpec.CloneSpec()), err)
+		return errors.E(op, errors.Git, errors.Repo(c.repoSpec.CloneSpec()), err)
 	}
 
 	// We need to create a temp directory where we can copy the content of the repo.
 	// During update, we need to checkout multiple versions of the same repo, so
 	// we can't do merges directly from the cache.
-	repoSpec.Dir, err = ioutil.TempDir("", "kpt-get-")
+	c.repoSpec.Dir, err = ioutil.TempDir("", "kpt-get-")
 	if err != nil {
 		return errors.E(op, errors.Internal, fmt.Errorf("error creating temp directory: %w", err))
 	}
-	repoSpec.Commit = commit
+	c.repoSpec.Commit = commit
 
-	pkgPath := filepath.Join(dir, repoSpec.Path)
+	pkgPath := filepath.Join(dir, c.repoSpec.Path)
 	// Verify that the requested path exists in the repo.
 	_, err = os.Stat(pkgPath)
 	if os.IsNotExist(err) {
 		return errors.E(op,
 			errors.Internal,
 			err,
-			fmt.Errorf("path %q does not exist in repo %q", repoSpec.Path, repoSpec.OrgRepo))
+			fmt.Errorf("path %q does not exist in repo %q", c.repoSpec.Path, c.repoSpec.OrgRepo))
 	}
 
 	// Copy the content of the pkg into the temp directory.
 	// Note that we skip the content outside the package directory.
-	err = copyDir(ctx, pkgPath, repoSpec.AbsPath())
+	err = copyDir(ctx, pkgPath, c.repoSpec.AbsPath())
 	if err != nil {
 		return errors.E(op, errors.Internal, fmt.Errorf("error copying package: %w", err))
 	}
@@ -226,7 +265,7 @@ func ClonerUsingGitExec(ctx context.Context, repoSpec *git.RepoSpec) error {
 		var kfError *pkg.KptfileError
 		if errors.As(err, &kfError) {
 			return &pkg.RemoteKptfileError{
-				RepoSpec: repoSpec,
+				RepoSpec: c.repoSpec,
 				Err:      kfError.Err,
 			}
 		}
diff --git a/internal/util/update/update.go b/internal/util/update/update.go
index df0658c3..ef50a91b 100644
--- a/internal/util/update/update.go
+++ b/internal/util/update/update.go
@@ -25,6 +25,7 @@ import (
 	"strings"
 
 	"github.com/GoogleContainerTools/kpt/internal/errors"
+	"github.com/GoogleContainerTools/kpt/internal/gitutil"
 	"github.com/GoogleContainerTools/kpt/internal/pkg"
 	"github.com/GoogleContainerTools/kpt/internal/printer"
 	"github.com/GoogleContainerTools/kpt/internal/types"
@@ -102,10 +103,13 @@ type Command struct {
 
 	// Strategy is the update strategy to use
 	Strategy kptfilev1.UpdateStrategyType
+
+	// cachedUpstreamRepos is an upstream repo already fetched for a given repoSpec CloneRef
+	cachedUpstreamRepos map[string]*gitutil.GitUpstreamRepo
 }
 
 // Run runs the Command.
-func (u Command) Run(ctx context.Context) error {
+func (u *Command) Run(ctx context.Context) error {
 	const op errors.Op = "update.Run"
 	pr := printer.FromContextOrDie(ctx)
 
@@ -133,7 +137,9 @@ func (u Command) Run(ctx context.Context) error {
 	if err != nil {
 		return errors.E(op, u.Pkg.UniquePath, err)
 	}
-
+	if u.cachedUpstreamRepos == nil {
+		u.cachedUpstreamRepos = make(map[string]*gitutil.GitUpstreamRepo)
+	}
 	packageCount := 0
 
 	// Use stack to keep track of paths with a Kptfile that might contain
@@ -182,6 +188,11 @@ func (u Command) Run(ctx context.Context) error {
 	return nil
 }
 
+// GetCachedUpstreamRepos returns repos cached during update
+func (u Command) GetCachedUpstreamRepos() map[string]*gitutil.GitUpstreamRepo {
+	return u.cachedUpstreamRepos
+}
+
 // updateSubKf updates subpackage with given ref and update strategy
 func updateSubKf(subKf *kptfilev1.KptFile, ref string, strategy kptfilev1.UpdateStrategyType) {
 	// check if explicit ref provided
@@ -250,7 +261,8 @@ func (u Command) updateRootPackage(ctx context.Context, p *pkg.Pkg) error {
 	g := kf.Upstream.Git
 	updated := &git.RepoSpec{OrgRepo: g.Repo, Path: g.Directory, Ref: g.Ref}
 	pr.Printf("Fetching upstream from %s@%s\n", kf.Upstream.Git.Repo, kf.Upstream.Git.Ref)
-	if err := fetch.ClonerUsingGitExec(ctx, updated); err != nil {
+	cloner := fetch.NewCloner(updated, fetch.WithCachedRepo(u.cachedUpstreamRepos))
+	if err := cloner.ClonerUsingGitExec(ctx); err != nil {
 		return errors.E(op, p.UniquePath, err)
 	}
 	defer os.RemoveAll(updated.AbsPath())
@@ -260,7 +272,7 @@ func (u Command) updateRootPackage(ctx context.Context, p *pkg.Pkg) error {
 		gLock := kf.UpstreamLock.Git
 		originRepoSpec := &git.RepoSpec{OrgRepo: gLock.Repo, Path: gLock.Directory, Ref: gLock.Commit}
 		pr.Printf("Fetching origin from %s@%s\n", kf.Upstream.Git.Repo, kf.Upstream.Git.Ref)
-		if err := fetch.ClonerUsingGitExec(ctx, originRepoSpec); err != nil {
+		if err := fetch.NewCloner(originRepoSpec, fetch.WithCachedRepo(u.cachedUpstreamRepos)).ClonerUsingGitExec(ctx); err != nil {
 			return errors.E(op, p.UniquePath, err)
 		}
 		origin = originRepoSpec
diff --git a/porch/pkg/engine/eval.go b/porch/pkg/engine/eval.go
index 5ef7fe84..74274794 100644
--- a/porch/pkg/engine/eval.go
+++ b/porch/pkg/engine/eval.go
@@ -18,10 +18,10 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/GoogleContainerTools/kpt/internal/fnruntime"
 	v1 "github.com/GoogleContainerTools/kpt/pkg/api/kptfile/v1"
 	"github.com/GoogleContainerTools/kpt/pkg/fn"
 	api "github.com/GoogleContainerTools/kpt/porch/api/porch/v1alpha1"
-	"github.com/GoogleContainerTools/kpt/porch/pkg/kpt"
 	"github.com/GoogleContainerTools/kpt/porch/pkg/repository"
 	"go.opentelemetry.io/otel/trace"
 	"sigs.k8s.io/kustomize/kyaml/fn/runtime/runtimeutil"
@@ -51,7 +51,7 @@ func (m *evalFunctionMutation) Apply(ctx context.Context, resources repository.P
 
 	var functionConfig *yaml.RNode
 	if m.task.Eval.ConfigMap != nil {
-		if cm, err := kpt.NewConfigMap(m.task.Eval.ConfigMap); err != nil {
+		if cm, err := fnruntime.NewConfigMap(m.task.Eval.ConfigMap); err != nil {
 			return repository.PackageResources{}, nil, fmt.Errorf("failed to create function config: %w", err)
 		} else {
 			functionConfig = cm
diff --git a/porch/pkg/kpt/eval.go b/porch/pkg/kpt/eval.go
index 26d2cfc8..549007ae 100644
--- a/porch/pkg/kpt/eval.go
+++ b/porch/pkg/kpt/eval.go
@@ -23,7 +23,6 @@ import (
 	"github.com/GoogleContainerTools/kpt/porch/pkg/kpt/internal"
 	"sigs.k8s.io/kustomize/kyaml/fn/framework"
 	"sigs.k8s.io/kustomize/kyaml/kio"
-	"sigs.k8s.io/kustomize/kyaml/yaml"
 )
 
 func NewSimpleFunctionRuntime() FunctionRuntime {
@@ -67,22 +66,3 @@ func (fr *runner) Run(r io.Reader, w io.Writer) error {
 
 	return framework.Execute(fr.processor, rw)
 }
-
-func NewConfigMap(data map[string]string) (*yaml.RNode, error) {
-	node := yaml.NewMapRNode(&data)
-	if node == nil {
-		return nil, nil
-	}
-	// create a ConfigMap only for configMap config
-	configMap := yaml.MustParse(`
-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: function-input
-data: {}
-`)
-	if err := configMap.PipeE(yaml.SetField("data", node)); err != nil {
-		return nil, err
-	}
-	return configMap, nil
-}
diff --git a/porch/pkg/kpt/pkgupdate.go b/porch/pkg/kpt/pkgupdate.go
index e6263845..e0b56a67 100644
--- a/porch/pkg/kpt/pkgupdate.go
+++ b/porch/pkg/kpt/pkgupdate.go
@@ -86,7 +86,7 @@ func PkgUpdate(ctx context.Context, ref string, packageDir string, opts PkgUpdat
 		// 	return errors.E(op, p.UniquePath, err)
 		// }
 		updated := *upstream
-		if err := fetch.ClonerUsingGitExec(ctx, &updated); err != nil {
+		if err := fetch.NewCloner(&updated).ClonerUsingGitExec(ctx); err != nil {
 			return err
 		}
 		defer os.RemoveAll(updated.AbsPath())
@@ -102,7 +102,7 @@ func PkgUpdate(ctx context.Context, ref string, packageDir string, opts PkgUpdat
 			// if err := fetch.ClonerUsingGitExec(ctx, originRepoSpec); err != nil {
 			// 	return errors.E(op, p.UniquePath, err)
 			// }
-			if err := fetch.ClonerUsingGitExec(ctx, originRepoSpec); err != nil {
+			if err := fetch.NewCloner(originRepoSpec).ClonerUsingGitExec(ctx); err != nil {
 				return err //errors.E(op, p.UniquePath, err)
 			}
 			originDir = originRepoSpec.AbsPath()
diff --git a/site/reference/cli/fn/eval/README.md b/site/reference/cli/fn/eval/README.md
index 2f189311..233a06d6 100644
--- a/site/reference/cli/fn/eval/README.md
+++ b/site/reference/cli/fn/eval/README.md
@@ -58,18 +58,6 @@ fn-args:
   By default, container function is executed as `nobody` user. You may want to use
   this flag to run higher privilege operations such as mounting the local filesystem.
 
---match-api-version:
-  Select resources matching the given apiVersion.
-
---match-kind
-  Select resources matching the given kind.
-
---match-name:
-  Select resources matching the given name.
-  
---match-namespace:
-  Select resources matching the given namespace.
-
 --env, e:
   List of local environment variables to be exported to the container function.
   By default, none of local environment variables are made available to the
@@ -106,6 +94,18 @@ fn-args:
   (DEPRECATED) include-meta-resources is no longer necessary because meta
   resources are included by default with kpt version v1.0.0-beta.15+.
 
+--match-api-version:
+  Select resources matching the given apiVersion.
+
+--match-kind
+  Select resources matching the given kind.
+
+--match-name:
+  Select resources matching the given name.
+  
+--match-namespace:
+  Select resources matching the given namespace.
+
 --mount:
   List of storage options to enable reading from the local filesytem. By default,
   container functions can not access the local filesystem. It accepts the same options
@@ -125,11 +125,21 @@ fn-args:
   3. OUT_DIR_PATH: output resources are written to provided directory.
      The provided directory must not already exist.
 
+--type, t;
+  Specify the function type. Accept value `mutator` (default), `validator`. 
+  If used with `--save`, this flag will save the evaluated function to the corresponding
+   Kptfile section: `.pipeline.mutators` if type is `mutator`; `.pipeline.validators` if type
+    is `validator`.
+
 --results-dir:
   Path to a directory to write structured results. Directory will be created if
   it doesn't exist. Structured results emitted by the functions are aggregated and saved
   to `results.yaml` file in the specified directory.
   If not specified, no result files are written to the local filesystem.
+  
+--save, s:
+  Save the function image and fn-config to Kptfile. Require ` + "`" + `--image` + "`" + `.
+  
 ```
 
 #### Environment Variables
@@ -162,6 +172,16 @@ $ kpt fn eval DIR -i gcr.io/example.com/my-fn --fn-config my-fn-config
 $ kpt fn eval DIR -i gcr.io/example.com/my-fn:v1.0.0 -- foo=bar
 ```
 
+```shell
+# execute container my-fn and save it to Kptfile `pipeline.mutators` (Default) list.
+$ kpt fn eval DIR -s -i gcr.io/example.com/my-fn:v1.0.0 -- foo=bar
+```
+
+```shell
+# execute container my-fn and save it to Kptfile `pipeline.validators` list.
+$ kpt fn eval DIR -s -t validator -i gcr.io/example.com/my-fn:v1.0.0 -- foo=bar
+```
+
 ```shell
 # execute executable my-fn on the resources in DIR directory and
 # write output back to DIR


Test patch:
diff --git a/internal/fnruntime/utils_test.go b/internal/fnruntime/utils_test.go
index dafb230d..b2f57776 100644
--- a/internal/fnruntime/utils_test.go
+++ b/internal/fnruntime/utils_test.go
@@ -140,3 +140,19 @@ spec:
 		})
 	}
 }
+
+func TestNewConfigMap(t *testing.T) {
+	data := map[string]string{
+		"normal string": "abc",
+		"integer":       "8081",
+		"float":         "1.23",
+		"bool":          "true",
+	}
+	m, err := NewConfigMap(data)
+	assert.NoError(t, err)
+	mapAsString := m.MustString()
+	assert.Contains(t, mapAsString, `bool: "true"`)
+	assert.Contains(t, mapAsString, `normal string: abc`)
+	assert.Contains(t, mapAsString, `integer: "8081"`)
+	assert.Contains(t, mapAsString, `float: "1.23"`)
+}
diff --git a/internal/util/update/update_test.go b/internal/util/update/update_test.go
index e1ea14e5..162a51a6 100644
--- a/internal/util/update/update_test.go
+++ b/internal/util/update/update_test.go
@@ -76,15 +76,25 @@ func TestCommand_Run_noRefChanges(t *testing.T) {
 				return
 			}
 			upstreamRepo := g.Repos[testutil.Upstream]
-
-			// Update the local package
-			if !assert.NoError(t, Command{
+			cmd := &Command{
 				Pkg:      pkgtest.CreatePkgOrFail(t, g.LocalWorkspace.FullPackagePath()),
 				Strategy: strategy,
-			}.Run(fake.CtxWithDefaultPrinter())) {
+			}
+			// Update the local package
+			if !assert.NoError(t, cmd.Run(fake.CtxWithDefaultPrinter())) {
 				return
 			}
 
+			// Expect upstream repo to be cached
+			cachedGURs := cmd.GetCachedUpstreamRepos()
+			assert.Equal(t, 1, len(cachedGURs))
+			for _, gur := range cachedGURs {
+				assert.Equal(t, upstreamRepo.RepoDirectory, gur.URI)
+				// Expect two fetched refs
+				// 1 for upstream ref (master) and 1 for previous upstream lock
+				assert.Equal(t, []string{masterBranch, upstreamRepo.Commits[0]}, gur.GetFetchedRefs())
+			}
+
 			// Expect the local package to have Dataset2
 			if !g.AssertLocalDataEquals(testutil.Dataset2, true) {
 				return
@@ -129,11 +139,11 @@ func TestCommand_Run_subDir(t *testing.T) {
 			upstreamRepo := g.Repos[testutil.Upstream]
 
 			// Update the local package
-			if !assert.NoError(t, Command{
+			if !assert.NoError(t, (&Command{
 				Pkg:      pkgtest.CreatePkgOrFail(t, g.LocalWorkspace.FullPackagePath()),
 				Ref:      "v1.2",
 				Strategy: strategy,
-			}.Run(fake.CtxWithDefaultPrinter())) {
+			}).Run(fake.CtxWithDefaultPrinter())) {
 				return
 			}
 
@@ -185,10 +195,10 @@ func TestCommand_Run_noChanges(t *testing.T) {
 			upstreamRepo := g.Repos[testutil.Upstream]
 
 			// Update the local package
-			err := Command{
+			err := (&Command{
 				Pkg:      pkgtest.CreatePkgOrFail(t, g.LocalWorkspace.FullPackagePath()),
 				Strategy: u.updater,
-			}.Run(fake.CtxWithDefaultPrinter())
+			}).Run(fake.CtxWithDefaultPrinter())
 			if u.err == "" {
 				if !assert.NoError(t, err) {
 					return
@@ -367,11 +377,11 @@ func TestCommand_Run_localPackageChanges(t *testing.T) {
 			}
 
 			// run the command
-			err = Command{
+			err = (&Command{
 				Pkg:      pkgtest.CreatePkgOrFail(t, g.LocalWorkspace.FullPackagePath()),
 				Ref:      masterBranch,
 				Strategy: tc.strategy,
-			}.Run(fake.CtxWithDefaultPrinter())
+			}).Run(fake.CtxWithDefaultPrinter())
 
 			// check the error response
 			if tc.expectedErr == "" {
@@ -437,11 +447,11 @@ func TestCommand_Run_toBranchRef(t *testing.T) {
 			upstreamRepo := g.Repos[testutil.Upstream]
 
 			// Update the local package
-			if !assert.NoError(t, Command{
+			if !assert.NoError(t, (&Command{
 				Pkg:      pkgtest.CreatePkgOrFail(t, g.LocalWorkspace.FullPackagePath()),
 				Strategy: strategy,
 				Ref:      "exp",
-			}.Run(fake.CtxWithDefaultPrinter())) {
+			}).Run(fake.CtxWithDefaultPrinter())) {
 				return
 			}
 
@@ -801,11 +811,11 @@ func TestCommand_Run_toBranchRefWithSubpkgs(t *testing.T) {
 			}
 
 			// Update the local package
-			if !assert.NoError(t, Command{
+			if !assert.NoError(t, (&Command{
 				Pkg:      pkgtest.CreatePkgOrFail(t, path.Join(g.LocalWorkspace.FullPackagePath(), tc.updateSubPkg)),
 				Strategy: tc.strategy,
 				Ref:      tc.updateRef,
-			}.Run(fake.CtxWithDefaultPrinter())) {
+			}).Run(fake.CtxWithDefaultPrinter())) {
 				return
 			}
 			expectedPath := tc.expectedLocal.ExpandPkgWithName(t,
@@ -847,11 +857,11 @@ func TestCommand_Run_toTagRef(t *testing.T) {
 			upstreamRepo := g.Repos[testutil.Upstream]
 
 			// Update the local package
-			if !assert.NoError(t, Command{
+			if !assert.NoError(t, (&Command{
 				Pkg:      pkgtest.CreatePkgOrFail(t, g.LocalWorkspace.FullPackagePath()),
 				Strategy: strategy,
 				Ref:      "v1.0",
-			}.Run(fake.CtxWithDefaultPrinter())) {
+			}).Run(fake.CtxWithDefaultPrinter())) {
 				return
 			}
 
@@ -904,11 +914,11 @@ func TestCommand_ResourceMerge_NonKRMUpdates(t *testing.T) {
 			upstreamRepo := g.Repos[testutil.Upstream]
 
 			// Update the local package
-			if !assert.NoError(t, Command{
+			if !assert.NoError(t, (&Command{
 				Pkg:      pkgtest.CreatePkgOrFail(t, g.LocalWorkspace.FullPackagePath()),
 				Strategy: strategy,
 				Ref:      "v1.0",
-			}.Run(fake.CtxWithDefaultPrinter())) {
+			}).Run(fake.CtxWithDefaultPrinter())) {
 				t.FailNow()
 			}
 
@@ -947,9 +957,9 @@ func TestCommand_Run_noUpstreamReference(t *testing.T) {
 	testutil.AddKptfileToWorkspace(t, w, kf)
 
 	// Update the local package
-	err := Command{
+	err := (&Command{
 		Pkg: pkgtest.CreatePkgOrFail(t, w.FullPackagePath()),
-	}.Run(fake.CtxWithDefaultPrinter())
+	}).Run(fake.CtxWithDefaultPrinter())
 
 	assert.Contains(t, err.Error(), "must have an upstream reference")
 }
@@ -960,10 +970,10 @@ func TestCommand_Run_failInvalidPath(t *testing.T) {
 		strategy := kptfilev1.UpdateStrategies[i]
 		t.Run(string(strategy), func(t *testing.T) {
 			path := filepath.Join("fake", "path")
-			err := Command{
+			err := (&Command{
 				Pkg:      pkgtest.CreatePkgOrFail(t, path),
 				Strategy: strategy,
-			}.Run(fake.CtxWithDefaultPrinter())
+			}).Run(fake.CtxWithDefaultPrinter())
 			if assert.Error(t, err) {
 				assert.Contains(t, err.Error(), "no such file or directory")
 			}
@@ -1025,9 +1035,9 @@ func TestCommand_Run_badUpstreamLock(t *testing.T) {
 			}
 
 			// Update the local package.
-			err = Command{
+			err = (&Command{
 				Pkg: pkgtest.CreatePkgOrFail(t, g.LocalWorkspace.FullPackagePath()),
-			}.Run(fake.CtxWithDefaultPrinter())
+			}).Run(fake.CtxWithDefaultPrinter())
 
 			if assert.Error(t, err) {
 				assert.Contains(t, err.Error(), tc.expectedErrorMsg)
@@ -1060,11 +1070,11 @@ func TestCommand_Run_failInvalidRef(t *testing.T) {
 				return
 			}
 
-			err := Command{
+			err := (&Command{
 				Pkg:      pkgtest.CreatePkgOrFail(t, g.LocalWorkspace.FullPackagePath()),
 				Ref:      "exp",
 				Strategy: strategy,
-			}.Run(fake.CtxWithDefaultPrinter())
+			}).Run(fake.CtxWithDefaultPrinter())
 			if !assert.Error(t, err) {
 				return
 			}
@@ -1118,9 +1128,9 @@ func TestCommand_Run_manualChange(t *testing.T) {
 		return
 	}
 
-	err := Command{
+	err := (&Command{
 		Pkg: pkgtest.CreatePkgOrFail(t, g.LocalWorkspace.FullPackagePath()),
-	}.Run(fake.CtxWithDefaultPrinter())
+	}).Run(fake.CtxWithDefaultPrinter())
 	if !assert.NoError(t, err) {
 		t.FailNow()
 	}
@@ -1172,9 +1182,9 @@ func TestCommand_Run_symlinks(t *testing.T) {
 	}
 	upstreamRepo := g.Repos[testutil.Upstream]
 
-	err := Command{
+	err := (&Command{
 		Pkg: pkgtest.CreatePkgOrFail(t, g.LocalWorkspace.FullPackagePath()),
-	}.Run(fake.CtxWithDefaultPrinter())
+	}).Run(fake.CtxWithDefaultPrinter())
 	if !assert.NoError(t, err) {
 		t.FailNow()
 	}
@@ -1220,10 +1230,10 @@ func TestCommand_Run_badStrategy(t *testing.T) {
 	}
 
 	// Update the local package
-	err := Command{
+	err := (&Command{
 		Pkg:      pkgtest.CreatePkgOrFail(t, g.LocalWorkspace.FullPackagePath()),
 		Strategy: strategy,
-	}.Run(fake.CtxWithDefaultPrinter())
+	}).Run(fake.CtxWithDefaultPrinter())
 	if !assert.Error(t, err, strategy) {
 		return
 	}
@@ -1999,10 +2009,10 @@ func TestCommand_Run_local_subpackages(t *testing.T) {
 					return
 				}
 
-				err := Command{
+				err := (&Command{
 					Pkg:      pkgtest.CreatePkgOrFail(t, g.LocalWorkspace.FullPackagePath()),
 					Strategy: strategy,
-				}.Run(fake.CtxWithDefaultPrinter())
+				}).Run(fake.CtxWithDefaultPrinter())
 
 				result := findExpectedResultForStrategy(test.expectedResults, strategy)
 
@@ -3539,9 +3549,9 @@ func TestRun_remote_subpackages(t *testing.T) {
 				return
 			}
 
-			err := Command{
+			err := (&Command{
 				Pkg: pkgtest.CreatePkgOrFail(t, g.LocalWorkspace.FullPackagePath()),
-			}.Run(fake.CtxWithDefaultPrinter())
+			}).Run(fake.CtxWithDefaultPrinter())
 
 			if tc.expectedErrMsg != "" {
 				if !assert.Error(t, err) {
@@ -3684,9 +3694,9 @@ func TestRootPackageIsUnfetched(t *testing.T) {
 			}
 			testutil.AddKptfileToWorkspace(t, w, kf)
 
-			err := Command{
+			err := (&Command{
 				Pkg: pkgtest.CreatePkgOrFail(t, w.FullPackagePath()),
-			}.Run(fake.CtxWithDefaultPrinter())
+			}).Run(fake.CtxWithDefaultPrinter())
 			if !assert.NoError(t, err) {
 				t.FailNow()
 			}
@@ -3697,6 +3707,97 @@ func TestRootPackageIsUnfetched(t *testing.T) {
 	}
 }
 
+// TestMultiUpdateCache verifies that multiple sub packages
+// with same upstream leverage the cache.
+func TestMultiUpdateCache(t *testing.T) {
+	numSubPkgs := 10
+	blueprintsUpstream := "blueprints"
+	// Generate multiple subpackages referencing blueprintsUpstream
+	subPkgs := make([]*pkgbuilder.SubPkg, 0, numSubPkgs)
+	for i := 1; i <= numSubPkgs; i++ {
+		sub := pkgbuilder.NewSubPkg(fmt.Sprintf("subpkg-%d", i)).
+			WithKptfile(
+				pkgbuilder.NewKptfile().
+					WithUpstreamRef(blueprintsUpstream, "/", masterBranch, "fast-forward").
+					WithUpstreamLockRef(blueprintsUpstream, "/", masterBranch, 0),
+			).
+			WithResource(pkgbuilder.DeploymentResource)
+		subPkgs = append(subPkgs, sub)
+	}
+
+	reposChanges := map[string][]testutil.Content{
+		testutil.Upstream: {
+			{
+				Pkg: pkgbuilder.NewRootPkg().
+					WithSubPackages(subPkgs...),
+				Branch: masterBranch,
+			},
+		},
+		blueprintsUpstream: {
+			{
+				Pkg: pkgbuilder.NewRootPkg().
+					WithResource(pkgbuilder.DeploymentResource),
+				Branch: masterBranch,
+			},
+		},
+	}
+
+	expectedLocal := pkgbuilder.NewRootPkg().
+		WithKptfile(
+			pkgbuilder.NewKptfile().
+				WithUpstreamRef(testutil.Upstream, "/", masterBranch, "resource-merge").
+				WithUpstreamLockRef(testutil.Upstream, "/", masterBranch, 0),
+		).
+		WithSubPackages(subPkgs...)
+
+	repos, w, clean := testutil.SetupReposAndWorkspace(t, reposChanges)
+	defer clean()
+
+	w.PackageDir = testPackageName
+	kf := kptfileutil.DefaultKptfile(testPackageName)
+	kf.Upstream = &kptfilev1.Upstream{
+		Type: kptfilev1.GitOrigin,
+		Git: &kptfilev1.Git{
+			Repo:      repos[testutil.Upstream].RepoDirectory,
+			Directory: "/",
+			Ref:       masterBranch,
+		},
+		UpdateStrategy: kptfilev1.ResourceMerge,
+	}
+	testutil.AddKptfileToWorkspace(t, w, kf)
+	cmd := Command{
+		Pkg: pkgtest.CreatePkgOrFail(t, w.FullPackagePath()),
+	}
+	err := cmd.Run(fake.CtxWithDefaultPrinter())
+	if !assert.NoError(t, err) {
+		t.FailNow()
+	}
+
+	// Expect 2 cached repo refs - testutil.Upstream and blueprintsUpstream
+	cachedGURs := cmd.GetCachedUpstreamRepos()
+	assert.Equal(t, 2, len(cachedGURs))
+
+	// Assert cached refs for each upstream repo
+	upstreamRepo := repos[testutil.Upstream].RepoDirectory
+	blueprintsRepo := repos[blueprintsUpstream].RepoDirectory
+	blueprintsRepoCommit := repos[blueprintsUpstream].Commits
+	for cr, gur := range cachedGURs {
+		switch cr {
+		case upstreamRepo:
+			// Expect only upstream ref master to be cached as there was no lock
+			assert.Equal(t, []string{masterBranch}, gur.GetFetchedRefs())
+		case blueprintsRepo:
+			// Expect both upstream ref master and upstream lock commit to be cached
+			assert.Equal(t, []string{masterBranch, blueprintsRepoCommit[0]}, gur.GetFetchedRefs())
+		default:
+			t.Fatalf("Unexpected upstream repo %s", cr)
+		}
+	}
+
+	expectedPath := expectedLocal.ExpandPkgWithName(t, testPackageName, testutil.ToReposInfo(repos))
+	testutil.KptfileAwarePkgEqual(t, expectedPath, w.FullPackagePath(), true)
+}
+
 type nonKRMTestCase struct {
 	name            string
 	updated         string
diff --git a/porch/pkg/kpt/eval_test.go b/porch/pkg/kpt/eval_test.go
index 13ef5863..bafe1789 100644
--- a/porch/pkg/kpt/eval_test.go
+++ b/porch/pkg/kpt/eval_test.go
@@ -20,6 +20,7 @@ import (
 	"path"
 	"testing"
 
+	"github.com/GoogleContainerTools/kpt/internal/fnruntime"
 	v1 "github.com/GoogleContainerTools/kpt/pkg/api/kptfile/v1"
 	"sigs.k8s.io/kustomize/kyaml/kio"
 	"sigs.k8s.io/kustomize/kyaml/yaml"
@@ -56,7 +57,7 @@ func TestSetLabels(t *testing.T) {
 		t.Errorf("GetRunner failed: %v", err)
 	}
 
-	config, err := NewConfigMap(map[string]string{
+	config, err := fnruntime.NewConfigMap(map[string]string{
 		"label-key": "label-value",
 	})
 	if err != nil {