Repository: GoogleContainerTools/kpt
Pull request #2745 (2022-02-05 02:08:33)

Connect to gRPC function runner
Add grpc function-runner address to Porch command line arguments.




Patch:
diff --git a/internal/fnruntime/runner.go b/internal/fnruntime/runner.go
index 557b6a37..1f814be4 100644
--- a/internal/fnruntime/runner.go
+++ b/internal/fnruntime/runner.go
@@ -19,7 +19,6 @@ import (
 	goerrors "errors"
 	"fmt"
 	"io/ioutil"
-	"os"
 	"path"
 	"path/filepath"
 	"strings"
@@ -33,6 +32,7 @@ import (
 	kptfilev1 "github.com/GoogleContainerTools/kpt/pkg/api/kptfile/v1"
 	"github.com/GoogleContainerTools/kpt/pkg/fn"
 	"github.com/google/shlex"
+	"sigs.k8s.io/kustomize/kyaml/filesys"
 	"sigs.k8s.io/kustomize/kyaml/fn/framework"
 	"sigs.k8s.io/kustomize/kyaml/fn/runtime/runtimeutil"
 	"sigs.k8s.io/kustomize/kyaml/kio"
@@ -43,12 +43,12 @@ import (
 // NewRunner returns a kio.Filter given a specification of a function
 // and it's config.
 func NewRunner(
-	ctx context.Context, f *kptfilev1.Function,
+	ctx context.Context, fsys filesys.FileSystem, f *kptfilev1.Function,
 	pkgPath types.UniquePath, fnResults *fnresult.ResultList,
 	imagePullPolicy ImagePullPolicy, displayResourceCount bool,
 	runtime fn.FunctionRuntime) (kio.Filter, error) {
 
-	config, err := newFnConfig(f, pkgPath)
+	config, err := newFnConfig(fsys, f, pkgPath)
 	if err != nil {
 		return nil, err
 	}
@@ -469,7 +469,7 @@ func (ri *multiLineFormatter) String() string {
 	return b.String()
 }
 
-func newFnConfig(f *kptfilev1.Function, pkgPath types.UniquePath) (*yaml.RNode, error) {
+func newFnConfig(fsys filesys.FileSystem, f *kptfilev1.Function, pkgPath types.UniquePath) (*yaml.RNode, error) {
 	const op errors.Op = "fn.readConfig"
 	var fn errors.Fn = errors.Fn(f.Image)
 
@@ -477,11 +477,12 @@ func newFnConfig(f *kptfilev1.Function, pkgPath types.UniquePath) (*yaml.RNode,
 	switch {
 	case f.ConfigPath != "":
 		path := filepath.Join(string(pkgPath), f.ConfigPath)
-		file, err := os.Open(path)
+		file, err := fsys.Open(path)
 		if err != nil {
 			return nil, errors.E(op, fn,
 				fmt.Errorf("missing function config %q", f.ConfigPath))
 		}
+		defer file.Close()
 		b, err := ioutil.ReadAll(file)
 		if err != nil {
 			return nil, errors.E(op, fn, err)
diff --git a/internal/util/render/executor.go b/internal/util/render/executor.go
index dbc7d065..d16c8d9f 100644
--- a/internal/util/render/executor.go
+++ b/internal/util/render/executor.go
@@ -488,7 +488,7 @@ func (pn *pkgNode) runValidators(ctx context.Context, hctx *hydrationContext, in
 			}
 			hctx.dockerCheckDone = true
 		}
-		validator, err = fnruntime.NewRunner(ctx, &function, pn.pkg.UniquePath, hctx.fnResults, hctx.imagePullPolicy, displayResourceCount, hctx.runtime)
+		validator, err = fnruntime.NewRunner(ctx, hctx.fileSystem, &function, pn.pkg.UniquePath, hctx.fnResults, hctx.imagePullPolicy, displayResourceCount, hctx.runtime)
 		if err != nil {
 			return err
 		}
@@ -605,7 +605,7 @@ func fnChain(ctx context.Context, hctx *hydrationContext, pkgPath types.UniquePa
 			}
 			hctx.dockerCheckDone = true
 		}
-		runner, err = fnruntime.NewRunner(ctx, &function, pkgPath, hctx.fnResults, hctx.imagePullPolicy, displayResourceCount, hctx.runtime)
+		runner, err = fnruntime.NewRunner(ctx, hctx.fileSystem, &function, pkgPath, hctx.fnResults, hctx.imagePullPolicy, displayResourceCount, hctx.runtime)
 		if err != nil {
 			return nil, err
 		}
diff --git a/pkg/api/kptfile/v1/validation.go b/pkg/api/kptfile/v1/validation.go
index 25914937..1f88a20b 100644
--- a/pkg/api/kptfile/v1/validation.go
+++ b/pkg/api/kptfile/v1/validation.go
@@ -177,6 +177,7 @@ func GetValidatedFnConfigFromPath(fsys filesys.FileSystem, pkgPath types.UniqueP
 	if err != nil {
 		return nil, fmt.Errorf("functionConfig must exist in the current package")
 	}
+	defer file.Close()
 	reader := kio.ByteReader{Reader: file, PreserveSeqIndent: true, WrapBareSeqNode: true, DisableUnwrapping: true}
 	nodes, err := reader.Read()
 	if err != nil {
diff --git a/porch/.vscode/launch.json b/porch/.vscode/launch.json
index 48998e43..d53fb75c 100644
--- a/porch/.vscode/launch.json
+++ b/porch/.vscode/launch.json
@@ -14,8 +14,9 @@
                 "--secure-port=9443",
                 "--v=7",
                 "--standalone-debug-mode",
-                "--kubeconfig=hack/local/kubeconfig",
-                "--cache-directory=${workspaceFolder}/.cache"
+                "--kubeconfig=${workspaceFolder}/hack/local/kubeconfig",
+                "--cache-directory=${workspaceFolder}/.cache",
+                "--function-runner=192.168.8.202:9445"
             ],
             "cwd": "${workspaceFolder}"
         },
diff --git a/porch/Makefile b/porch/Makefile
index 743e7daf..2ed85529 100644
--- a/porch/Makefile
+++ b/porch/Makefile
@@ -111,7 +111,8 @@ run-local:
 	--secure-port 9443 \
 	--standalone-debug-mode \
 	--kubeconfig="$(KUBECONFIG)" \
-	--cache-directory="$(CACHEDIR)"
+	--cache-directory="$(CACHEDIR)" \
+	--function-runner 192.168.8.202:9445
 
 .PHONY: run-jaeger
 run-jaeger:
diff --git a/porch/engine/pkg/engine/engine.go b/porch/engine/pkg/engine/engine.go
index 11c43f7f..d08872ab 100644
--- a/porch/engine/pkg/engine/engine.go
+++ b/porch/engine/pkg/engine/engine.go
@@ -15,7 +15,6 @@
 package engine
 
 import (
-	"bytes"
 	"context"
 	"fmt"
 	"io/fs"
@@ -25,7 +24,6 @@ import (
 	"reflect"
 	"strings"
 
-	v1 "github.com/GoogleContainerTools/kpt/pkg/api/kptfile/v1"
 	"github.com/GoogleContainerTools/kpt/pkg/fn"
 	api "github.com/GoogleContainerTools/kpt/porch/api/porch/v1alpha1"
 	configapi "github.com/GoogleContainerTools/kpt/porch/controllers/pkg/apis/porch/v1alpha1"
@@ -36,8 +34,6 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 	"k8s.io/klog/v2"
-	"sigs.k8s.io/kustomize/kyaml/kio"
-	"sigs.k8s.io/kustomize/kyaml/yaml"
 )
 
 type CaDEngine interface {
@@ -378,85 +374,6 @@ func loadResourcesFromDirectory(dir string) (repository.PackageResources, error)
 	return result, nil
 }
 
-type evalFunctionMutation struct {
-	runtime fn.FunctionRuntime
-	task    *api.Task
-}
-
-func (m *evalFunctionMutation) Apply(ctx context.Context, resources repository.PackageResources) (repository.PackageResources, *api.Task, error) {
-	e := m.task.Eval
-
-	// TODO: Apply should accept filesystem instead of PackageResources
-
-	runner, err := m.runtime.GetRunner(ctx, &v1.Function{
-		Image:     e.Image,
-		ConfigMap: e.ConfigMap,
-	})
-	if err != nil {
-		return repository.PackageResources{}, nil, fmt.Errorf("failed to create function runner: %w", err)
-	}
-
-	pr := &packageReader{
-		input: resources,
-		extra: map[string]string{},
-	}
-
-	// r := &kio.LocalPackageReader{
-	// 	PackagePath:        "/",
-	// 	IncludeSubpackages: true,
-	// 	FileSystem:         filesys.FileSystemOrOnDisk{FileSystem: fs},
-	// 	WrapBareSeqNode:    true,
-	// }
-
-	var rl bytes.Buffer
-	w := &kio.ByteWriter{
-		Writer:                &rl,
-		KeepReaderAnnotations: true,
-		FunctionConfig:        &yaml.RNode{},
-		WrappingKind:          kio.ResourceListKind,
-		WrappingAPIVersion:    kio.ResourceListAPIVersion,
-	}
-
-	pipeline := kio.Pipeline{
-		Inputs:  []kio.Reader{pr},
-		Outputs: []kio.Writer{w},
-	}
-
-	if err := pipeline.Execute(); err != nil {
-		return repository.PackageResources{}, nil, fmt.Errorf("failed to serialize package: %w", err)
-	}
-
-	// Evaluate the function
-	var output bytes.Buffer
-	if err := runner.Run(&rl, &output); err != nil {
-		return repository.PackageResources{}, nil, fmt.Errorf("failed to evaluate function: %w", err)
-	}
-
-	result := repository.PackageResources{
-		Contents: map[string]string{},
-	}
-
-	if err := (kio.Pipeline{
-		Inputs: []kio.Reader{&kio.ByteReader{
-			Reader:            &output,
-			PreserveSeqIndent: true,
-			WrapBareSeqNode:   true,
-		}},
-		Outputs: []kio.Writer{&packageWriter{
-			output: result,
-		}},
-	}.Execute()); err != nil {
-		return repository.PackageResources{}, nil, fmt.Errorf("failed to de-serialize function result: %w", err)
-	}
-
-	// Return extras. TODO: Apply should accept FS.
-	for k, v := range pr.extra {
-		result.Contents[k] = v
-	}
-
-	return result, m.task, nil
-}
-
 type mutationReplaceResources struct {
 	newResources *api.PackageRevisionResources
 	oldResources *api.PackageRevisionResources
diff --git a/porch/engine/pkg/engine/eval.go b/porch/engine/pkg/engine/eval.go
new file mode 100644
index 00000000..2bb86f02
--- /dev/null
+++ b/porch/engine/pkg/engine/eval.go
@@ -0,0 +1,97 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package engine
+
+import (
+	"context"
+	"fmt"
+
+	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/engine/pkg/kpt"
+	"github.com/GoogleContainerTools/kpt/porch/repository/pkg/repository"
+	"sigs.k8s.io/kustomize/kyaml/fn/runtime/runtimeutil"
+	"sigs.k8s.io/kustomize/kyaml/kio"
+	"sigs.k8s.io/kustomize/kyaml/yaml"
+)
+
+type evalFunctionMutation struct {
+	runtime fn.FunctionRuntime
+	task    *api.Task
+}
+
+func (m *evalFunctionMutation) Apply(ctx context.Context, resources repository.PackageResources) (repository.PackageResources, *api.Task, error) {
+	e := m.task.Eval
+
+	// TODO: Apply should accept filesystem instead of PackageResources
+
+	runner, err := m.runtime.GetRunner(ctx, &v1.Function{
+		Image: e.Image,
+	})
+	if err != nil {
+		return repository.PackageResources{}, nil, fmt.Errorf("failed to create function runner: %w", err)
+	}
+
+	var functionConfig *yaml.RNode
+	if m.task.Eval.ConfigMap != nil {
+		if cm, err := kpt.NewConfigMap(m.task.Eval.ConfigMap); err != nil {
+			return repository.PackageResources{}, nil, fmt.Errorf("failed to create function config: %w", err)
+		} else {
+			functionConfig = cm
+		}
+	}
+
+	ff := &runtimeutil.FunctionFilter{
+		Run:            runner.Run,
+		FunctionConfig: functionConfig,
+		Results:        &yaml.RNode{},
+	}
+
+	pr := &packageReader{
+		input: resources,
+		extra: map[string]string{},
+	}
+
+	// r := &kio.LocalPackageReader{
+	// 	PackagePath:        "/",
+	// 	IncludeSubpackages: true,
+	// 	FileSystem:         filesys.FileSystemOrOnDisk{FileSystem: fs},
+	// 	WrapBareSeqNode:    true,
+	// }
+
+	result := repository.PackageResources{
+		Contents: map[string]string{},
+	}
+
+	pipeline := kio.Pipeline{
+		Inputs:  []kio.Reader{pr},
+		Filters: []kio.Filter{ff},
+		Outputs: []kio.Writer{&packageWriter{
+			output: result,
+		}},
+	}
+
+	if err := pipeline.Execute(); err != nil {
+		return repository.PackageResources{}, nil, fmt.Errorf("failed to evaluate function: %w", err)
+	}
+
+	// Return extras. TODO: Apply should accept FS.
+	for k, v := range pr.extra {
+		result.Contents[k] = v
+	}
+
+	return result, m.task, nil
+}
diff --git a/porch/engine/pkg/kpt/eval.go b/porch/engine/pkg/kpt/eval.go
index 52fa4704..0313983b 100644
--- a/porch/engine/pkg/kpt/eval.go
+++ b/porch/engine/pkg/kpt/eval.go
@@ -24,6 +24,7 @@ import (
 	"github.com/GoogleContainerTools/kpt/porch/engine/pkg/internal"
 	"sigs.k8s.io/kustomize/kyaml/fn/framework"
 	"sigs.k8s.io/kustomize/kyaml/kio"
+	"sigs.k8s.io/kustomize/kyaml/yaml"
 )
 
 func NewSimpleFunctionRuntime() FunctionRuntime {
@@ -43,7 +44,6 @@ func (e *runtime) GetRunner(ctx context.Context, fn *kptfilev1.Function) (fn.Fun
 
 	return &runner{
 		ctx:       ctx,
-		fn:        *fn,
 		processor: processor,
 	}, nil
 }
@@ -54,7 +54,6 @@ func (e *runtime) Close() error {
 
 type runner struct {
 	ctx       context.Context
-	fn        kptfilev1.Function
 	processor framework.ResourceListProcessorFunc
 }
 
@@ -69,3 +68,22 @@ 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
+}


Test patch:
diff --git a/internal/fnruntime/runner_test.go b/internal/fnruntime/runner_test.go
index 97fd2b1f..37bbe218 100644
--- a/internal/fnruntime/runner_test.go
+++ b/internal/fnruntime/runner_test.go
@@ -29,6 +29,7 @@ import (
 	"github.com/GoogleContainerTools/kpt/internal/types"
 	kptfilev1 "github.com/GoogleContainerTools/kpt/pkg/api/kptfile/v1"
 	"github.com/stretchr/testify/assert"
+	"sigs.k8s.io/kustomize/kyaml/filesys"
 	"sigs.k8s.io/kustomize/kyaml/fn/framework"
 	"sigs.k8s.io/kustomize/kyaml/kio"
 	"sigs.k8s.io/kustomize/kyaml/yaml"
@@ -89,7 +90,8 @@ data: {foo: bar}
 				assert.NoError(t, err, "unexpected error")
 				c.fn.ConfigPath = path.Base(tmp.Name())
 			}
-			cn, err := newFnConfig(&c.fn, types.UniquePath(os.TempDir()))
+			fsys := filesys.MakeFsOnDisk()
+			cn, err := newFnConfig(fsys, &c.fn, types.UniquePath(os.TempDir()))
 			assert.NoError(t, err, "unexpected error")
 			actual, err := cn.String()
 			assert.NoError(t, err, "unexpected error")
diff --git a/porch/engine/pkg/kpt/eval_test.go b/porch/engine/pkg/kpt/eval_test.go
index 30b722bd..13ef5863 100644
--- a/porch/engine/pkg/kpt/eval_test.go
+++ b/porch/engine/pkg/kpt/eval_test.go
@@ -89,22 +89,3 @@ func TestSetLabels(t *testing.T) {
 		}
 	}
 }
-
-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
-}