Add possibility for pipe-tasks to pipe env-files

config/crds/kudo.dev_operatorversions.yaml

@@ -234,6 +234,8 @@ spec:
                           description: PipeSpec describes how a file generated by
                             a PipeTask is stored and referenced
                           properties:
+                            envFile:
+                              type: string
                             file:
                               type: string
                             key:

pkg/apis/kudo/v1beta1/operatorversion_types.go

@@ -192,6 +192,8 @@ type PipeSpec struct {
 	// +optional
 	File string `json:"file"`
 	// +optional
+	EnvFile string `json:"envFile"`
+	// +optional
 	Kind string `json:"kind"`
 	// +optional
 	Key string `json:"key"`

pkg/engine/task/env_file.go

@@ -0,0 +1,77 @@
+package task
+
+// This file has been adopted from kubectl/pkg/generate/versioned/env_file.go and used to read files containing
+// env var pairs in pipe-tasks.
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"strings"
+	"unicode"
+	"unicode/utf8"
+
+	"k8s.io/apimachinery/pkg/util/validation"
+)
+
+var utf8bom = []byte{0xEF, 0xBB, 0xBF}
+
+// proccessEnvFileLine returns a blank key if the line is empty or a comment.
+// The value will be retrieved from the environment if necessary.
+func proccessEnvFileLine(line []byte, currentLine int) (key, value string, err error) {
+
+	if !utf8.Valid(line) {
+		return ``, ``, fmt.Errorf("invalid utf8 bytes at line %d: %v", currentLine+1, line)
+	}
+
+	// We trim UTF8 BOM from the first line of the file but no others
+	if currentLine == 0 {
+		line = bytes.TrimPrefix(line, utf8bom)
+	}
+
+	// trim the line from all leading whitespace first
+	line = bytes.TrimLeftFunc(line, unicode.IsSpace)
+
+	// If the line is empty or a comment, we return a blank key/value pair.
+	if len(line) == 0 || line[0] == '#' {
+		return ``, ``, nil
+	}
+
+	data := strings.SplitN(string(line), "=", 2)
+	if len(data) != 2 {
+		return ``, ``, fmt.Errorf("%q is not a valid env var definition (KEY=VAL)", line)
+	}
+
+	key = data[0]
+	if errs := validation.IsEnvVarName(key); len(errs) != 0 {
+		return ``, ``, fmt.Errorf("%q is not a valid key name: %s", key, strings.Join(errs, ";"))
+	}
+	value = data[1]
+
+	return key, value, nil
+}
+
+// addFromEnvFile processes an env file allows a generic addTo to handle the
+// collection of key value pairs or returns an error.
+func addFromEnvFile(data []byte, addTo func(key, value string)) error {
+	r := bytes.NewReader(data)
+	scanner := bufio.NewScanner(r)
+	currentLine := 0
+	for scanner.Scan() {
+		// Process the current line, retrieving a key/value pair if possible.
+		scannedBytes := scanner.Bytes()
+		key, value, err := proccessEnvFileLine(scannedBytes, currentLine)
+		if err != nil {
+			return err
+		}
+		currentLine++
+
+		if len(key) == 0 {
+			// no key means line was empty or a comment
+			continue
+		}
+
+		addTo(key, value)
+	}
+	return nil
+}

pkg/engine/task/task.go

@@ -115,7 +115,7 @@ func newPipe(task *v1beta1.Task) (Tasker, error) {
 
 	var pipeFiles []PipeFile
 	for _, pp := range task.Spec.PipeTaskSpec.Pipe {
-		pf := PipeFile{File: pp.File, Kind: PipeFileKind(pp.Kind), Key: pp.Key}
+		pf := PipeFile{File: pp.File, EnvFile: pp.EnvFile, Kind: PipeFileKind(pp.Kind), Key: pp.Key}
 		// validate pipe file
 		if err := validPipeFile(pf); err != nil {
 			return nil, err
@@ -151,9 +151,12 @@ var (
 )
 
 func validPipeFile(pf PipeFile) error {
-	if pf.File == "" {
-		return fmt.Errorf("task validation error: pipe file is empty: %v", pf)
+	fl := pf.File != ""
+	efl := pf.EnvFile != ""
+	if fl == efl {
+		return fmt.Errorf("task validation error: pipe file %v must have either 'file' or 'envFile' field set but not both", pf)
 	}
+
 	if pf.Kind != PipeFileKindSecret && pf.Kind != PipeFileKindConfigMap {
 		return fmt.Errorf("task validation error: invalid pipe kind (must be Secret or ConfigMap): %v", pf)
 	}

pkg/engine/task/task_pipe.go

@@ -49,9 +49,19 @@ type PipeTask struct {
 }
 
 type PipeFile struct {
-	File string
-	Kind PipeFileKind
-	Key  string
+	File    string
+	EnvFile string
+	Kind    PipeFileKind
+	Key     string
+}
+
+// fileSource return either File or EnvFile depending on which one of them is set. Note that only one can be set at a
+// time which is enforced by the validation.
+func (pf PipeFile) fileSource() string {
+	if pf.File != "" {
+		return pf.File
+	}
+	return pf.EnvFile
 }
 
 func (pt PipeTask) Run(ctx Context) (bool, error) {
@@ -218,11 +228,11 @@ func validate(pod *corev1.Pod, ff []PipeFile) error {
 
 	// check if all referenced pipe files are children of the container mountPath
 	for _, f := range ff {
-		if !isRelative(mountPath, f.File) {
-			return fmt.Errorf("pipe file %s should be a child of %s mount path", f.File, mountPath)
+		if !isRelative(mountPath, f.fileSource()) {
+			return fmt.Errorf("pipe file %s should be a child of %s mount path", f.fileSource(), mountPath)
 		}
 
-		fileName := path.Base(f.File)
+		fileName := path.Base(f.fileSource())
 		// Same as k8s we use file names as ConfigMap data keys. A valid key name for a ConfigMap must consist
 		// of alphanumeric characters, '-', '_' or '.' (e.g. 'key.name',  or 'KEY_NAME',  or 'key-name', regex
 		// used for validation is '[-._a-zA-Z0-9]+')
@@ -271,12 +281,12 @@ func copyFiles(fs afero.Fs, ff []PipeFile, pod *corev1.Pod, ctx Context) error {
 
 	for _, f := range ff {
 		f := f
-		log.Printf("PipeTask: %s/%s copying pipe file %s", ctx.Meta.InstanceNamespace, ctx.Meta.InstanceName, f.File)
+		log.Printf("PipeTask: %s/%s copying pipe file %s", ctx.Meta.InstanceNamespace, ctx.Meta.InstanceName, f.fileSource())
 		g.Go(func() error {
 			// Check the size of the pipe file first. K87 has a inherent limit on the size of
 			// Secret/ConfigMap, so we avoid unnecessary copying of files that are too big by
 			// checking its size first.
-			size, err := podexec.FileSize(f.File, pod, pipePodContainerName, ctx.Config)
+			size, err := podexec.FileSize(f.fileSource(), pod, pipePodContainerName, ctx.Config)
 			if err != nil {
 				// Any remote command exit code > 0 is treated as a fatal error since retrying it doesn't make sense
 				if podexec.HasCommandFailed(err) {
@@ -286,10 +296,10 @@ func copyFiles(fs afero.Fs, ff []PipeFile, pod *corev1.Pod, ctx Context) error {
 			}
 
 			if size > maxPipeFileSize {
-				return fatalExecutionError(fmt.Errorf("pipe file %s size %d exceeds maximum file size of %d bytes", f.File, size, maxPipeFileSize), pipeTaskError, ctx.Meta)
+				return fatalExecutionError(fmt.Errorf("pipe file %s size %d exceeds maximum file size of %d bytes", f.fileSource(), size, maxPipeFileSize), pipeTaskError, ctx.Meta)
 			}
 
-			if err = podexec.DownloadFile(fs, f.File, pod, pipePodContainerName, ctx.Config); err != nil {
+			if err = podexec.DownloadFile(fs, f.fileSource(), pod, pipePodContainerName, ctx.Config); err != nil {
 				// Any remote command exit code > 0 is treated as a fatal error since retrying it doesn't make sense
 				if podexec.HasCommandFailed(err) {
 					return fatalExecutionError(err, pipeTaskError, ctx.Meta)
@@ -309,9 +319,9 @@ func createArtifacts(fs afero.Fs, files []PipeFile, meta renderer.Metadata) (map
 	artifacts := map[string]string{}
 
 	for _, pf := range files {
-		data, err := afero.ReadFile(fs, pf.File)
+		data, err := afero.ReadFile(fs, pf.fileSource())
 		if err != nil {
-			return nil, fmt.Errorf("error opening pipe file %s", pf.File)
+			return nil, fmt.Errorf("error opening pipe file %s", pf.fileSource())
 		}
 
 		var art string
@@ -337,7 +347,7 @@ func createArtifacts(fs afero.Fs, files []PipeFile, meta renderer.Metadata) (map
 // as Secret data key. Secret name will be of the form <instance>.<plan>.<phase>.<step>.<task>-<PipeFile.Key>
 func createSecret(pf PipeFile, data []byte, meta renderer.Metadata) (string, error) {
 	name := PipeArtifactName(meta, pf.Key)
-	key := path.Base(pf.File)
+	key := path.Base(pf.fileSource())
 
 	secret := corev1.Secret{
 		TypeMeta: metav1.TypeMeta{
@@ -347,10 +357,23 @@ func createSecret(pf PipeFile, data []byte, meta renderer.Metadata) (string, err
 		ObjectMeta: metav1.ObjectMeta{
 			Name: name,
 		},
-		Data: map[string][]byte{key: data},
+		Data: map[string][]byte{},
 		Type: corev1.SecretTypeOpaque,
 	}
 
+	if pf.File != "" {
+		secret.Data = map[string][]byte{key: data}
+	}
+	if pf.EnvFile != "" {
+		err := addFromEnvFile(data, func(key, value string) {
+			secret.Data[key] = []byte(value)
+		})
+
+		if err != nil {
+			return "", fmt.Errorf("failed to read env var file %q: %v", pf.fileSource(), err)
+		}
+	}
+
 	b, err := yaml.Marshal(secret)
 	if err != nil {
 		return "", fmt.Errorf("failed to marshal pipe secret for pipe file %s: %v", pf.File, err)
@@ -363,7 +386,7 @@ func createSecret(pf PipeFile, data []byte, meta renderer.Metadata) (string, err
 // as ConfigMap data key. ConfigMap name will be of the form <instance>.<plan>.<phase>.<step>.<task>-<PipeFile.Key>
 func createConfigMap(pf PipeFile, data []byte, meta renderer.Metadata) (string, error) {
 	name := PipeArtifactName(meta, pf.Key)
-	key := path.Base(pf.File)
+	key := path.Base(pf.fileSource())
 
 	configMap := corev1.ConfigMap{
 		TypeMeta: metav1.TypeMeta{
@@ -373,7 +396,21 @@ func createConfigMap(pf PipeFile, data []byte, meta renderer.Metadata) (string,
 		ObjectMeta: metav1.ObjectMeta{
 			Name: name,
 		},
-		BinaryData: map[string][]byte{key: data},
+		Data:       map[string]string{},
+		BinaryData: map[string][]byte{},
+	}
+
+	if pf.File != "" {
+		configMap.BinaryData = map[string][]byte{key: data}
+	}
+	if pf.EnvFile != "" {
+		err := addFromEnvFile(data, func(key, value string) {
+			configMap.Data[key] = value
+		})
+
+		if err != nil {
+			return "", fmt.Errorf("failed to read env var file %q: %v", pf.fileSource(), err)
+		}
 	}
 
 	b, err := yaml.Marshal(configMap)

pkg/engine/task/task_pipe_test.go

@@ -383,7 +383,7 @@ func Test_pipeFiles(t *testing.T) {
 		wantErr      bool
 	}{
 		{
-			name: "pipe a secret",
+			name: "pipe a file to a secret",
 			data: map[string]string{"/tmp/foo.txt": "foo"},
 			file: PipeFile{
 				File: "/tmp/foo.txt",
@@ -400,7 +400,32 @@ func Test_pipeFiles(t *testing.T) {
 			wantErr: false,
 		},
 		{
-			name: "pipe a configMap",
+			name: "pipe an env file to a secret",
+			data: map[string]string{"/tmp/foo.env": `
+				enemies=aliens
+				lives=3
+				allowed="true"
+			`},
+			file: PipeFile{
+				EnvFile: "/tmp/foo.env",
+				Kind:    PipeFileKindSecret,
+				Key:     "Foo",
+			},
+			meta: meta,
+			wantArtifact: v1.Secret{
+				TypeMeta:   metav1.TypeMeta{Kind: "Secret", APIVersion: "v1"},
+				ObjectMeta: metav1.ObjectMeta{Name: "fooinstance.deploy.first.step.genfiles.foo"},
+				Data: map[string][]byte{
+					"enemies": []byte("aliens"),
+					"lives":   []byte("3"),
+					"allowed": []byte("\"true\""),
+				},
+				Type: v1.SecretTypeOpaque,
+			},
+			wantErr: false,
+		},
+		{
+			name: "pipe a file to a configMap",
 			data: map[string]string{"/tmp/bar.txt": "bar"},
 			file: PipeFile{
 				File: "/tmp/bar.txt",
@@ -415,6 +440,30 @@ func Test_pipeFiles(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "pipe an env file to a configMap",
+			data: map[string]string{"/tmp/bar.env": `
+				enemies=aliens
+				lives=3
+				allowed="true"
+			`},
+			file: PipeFile{
+				EnvFile: "/tmp/bar.env",
+				Kind:    PipeFileKindConfigMap,
+				Key:     "Bar",
+			},
+			meta: meta,
+			wantArtifact: v1.ConfigMap{
+				TypeMeta:   metav1.TypeMeta{Kind: "ConfigMap", APIVersion: "v1"},
+				ObjectMeta: metav1.ObjectMeta{Name: "fooinstance.deploy.first.step.genfiles.bar"},
+				Data: map[string]string{
+					"enemies": "aliens",
+					"lives":   "3",
+					"allowed": "\"true\"",
+				},
+			},
+			wantErr: false,
+		},
 		{
 			name: "return an error for an invalid pipe",
 			data: map[string]string{"nope.txt": ""},
@@ -428,6 +477,7 @@ func Test_pipeFiles(t *testing.T) {
 			wantErr:      true,
 		},
 	}
+
 	for _, tt := range tests {
 		tt := tt