- fix(launcher): missing executorInput parameter values caused by {{$}} evaluation order (kubeflow#11925)

Signed-off-by: arpechenin <[email protected]>
Co-authored-by: arpechenin <[email protected]>

Author: ntny

backend/src/v2/component/launcher_v2.go

@@ -364,26 +364,16 @@ func executeV2(
 	}
 
 	// Fill in placeholders with runtime values.
-	placeholders, err := getPlaceholders(executorInputWithDefault)
+	compiledCmd, compiledArgs, err := compileCmdAndArgs(executorInputWithDefault, cmd, args)
 	if err != nil {
 		return nil, nil, err
 	}
-	for placeholder, replacement := range placeholders {
-		cmd = strings.ReplaceAll(cmd, placeholder, replacement)
-	}
-	for i := range args {
-		arg := args[i]
-		for placeholder, replacement := range placeholders {
-			arg = strings.ReplaceAll(arg, placeholder, replacement)
-		}
-		args[i] = arg
-	}
 
 	executorOutput, err := execute(
 		ctx,
 		executorInput,
-		cmd,
-		args,
+		compiledCmd,
+		compiledArgs,
 		bucket,
 		bucketConfig,
 		namespace,
@@ -737,6 +727,31 @@ func fetchNonDefaultBuckets(
 
 }
 
+func compileCmdAndArgs(executorInput *pipelinespec.ExecutorInput, cmd string, args []string) (string, []string, error) {
+	placeholders, err := getPlaceholders(executorInput)
+
+	executorInputJSON, err := protojson.Marshal(executorInput)
+	if err != nil {
+		return "", nil, fmt.Errorf("failed to convert ExecutorInput into JSON: %w", err)
+	}
+	executorInputJSONKey := "{{$}}"
+	executorInputJSONString := string(executorInputJSON)
+
+	compiledCmd := strings.ReplaceAll(cmd, executorInputJSONKey, executorInputJSONString)
+	compiledArgs := make([]string, 0, len(args))
+	for placeholder, replacement := range placeholders {
+		cmd = strings.ReplaceAll(cmd, placeholder, replacement)
+	}
+	for _, arg := range args {
+		compiledArgTemplate := strings.ReplaceAll(arg, executorInputJSONKey, executorInputJSONString)
+		for placeholder, replacement := range placeholders {
+			compiledArgTemplate = strings.ReplaceAll(compiledArgTemplate, placeholder, replacement)
+		}
+		compiledArgs = append(compiledArgs, compiledArgTemplate)
+	}
+	return compiledCmd, compiledArgs, nil
+}
+
 // Add executor input placeholders to provided map.
 func getPlaceholders(executorInput *pipelinespec.ExecutorInput) (placeholders map[string]string, err error) {
 	defer func() {
@@ -745,11 +760,9 @@ func getPlaceholders(executorInput *pipelinespec.ExecutorInput) (placeholders ma
 		}
 	}()
 	placeholders = make(map[string]string)
-	executorInputJSON, err := protojson.Marshal(executorInput)
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert ExecutorInput into JSON: %w", err)
 	}
-	placeholders["{{$}}"] = string(executorInputJSON)
 
 	// Read input artifact metadata.
 	for name, artifactList := range executorInput.GetInputs().GetArtifacts() {

backend/src/v2/component/launcher_v2_test.go

@@ -15,10 +15,13 @@ package component
 
 import (
 	"context"
+	"encoding/json"
 	"io"
 	"os"
 	"testing"
 
+	"google.golang.org/protobuf/encoding/protojson"
+
 	"github.com/kubeflow/pipelines/api/v2alpha1/go/pipelinespec"
 	"github.com/kubeflow/pipelines/backend/src/v2/metadata"
 	"github.com/kubeflow/pipelines/backend/src/v2/objectstore"
@@ -106,7 +109,7 @@ func Test_executeV2_Parameters(t *testing.T) {
 	}
 }
 
-func Test_executeV2_publishLogss(t *testing.T) {
+func Test_executeV2_publishLogs(t *testing.T) {
 	tests := []struct {
 		name          string
 		executorInput *pipelinespec.ExecutorInput
@@ -167,6 +170,75 @@ func Test_executeV2_publishLogss(t *testing.T) {
 	}
 }
 
+func Test_executorInput_compileCmdAndArgs(t *testing.T) {
+	executorInputJSON := `{
+		"inputs": {
+			"parameterValues": {
+				"config": {
+					"category_ids": "{{$.inputs.parameters['pipelinechannel--category_ids']}}",
+					"dump_filename": "{{$.inputs.parameters['pipelinechannel--dump_filename']}}",
+					"sphinx_host": "{{$.inputs.parameters['pipelinechannel--sphinx_host']}}",
+					"sphinx_port": "{{$.inputs.parameters['pipelinechannel--sphinx_port']}}"
+				},
+				"pipelinechannel--category_ids": "116",
+				"pipelinechannel--dump_filename": "dump_filename_test.txt",
+				"pipelinechannel--sphinx_host": "sphinx-default-host.ru",
+				"pipelinechannel--sphinx_port": 9312
+			}
+		},
+		"outputs": {
+			"artifacts": {
+				"dataset": {
+					"artifacts": [{
+						"type": {
+							"schemaTitle": "system.Dataset",
+							"schemaVersion": "0.0.1"
+						},
+						"uri": "s3://aviflow-stage-kfp-artifacts/debug-component-pipeline/ae02034e-bd96-4b8a-a06b-55c99fe9eccb/sayhello/c98ac032-2448-4637-bf37-3ad1e13a112c/dataset"
+					}]
+				}
+			},
+			"outputFile": "/tmp/kfp_outputs/output_metadata.json"
+		}
+	}`
+
+	executorInput := &pipelinespec.ExecutorInput{}
+	err := protojson.Unmarshal([]byte(executorInputJSON), executorInput)
+
+	assert.NoError(t, err)
+
+	cmd := "sh"
+	args := []string{
+		"--executor_input", "{{$}}",
+		"--function_to_execute", "sayHello",
+	}
+	cmd, args, err = compileCmdAndArgs(executorInput, cmd, args)
+
+	assert.NoError(t, err)
+
+	var actualExecutorInput string
+	for i := 0; i < len(args)-1; i++ {
+		if args[i] == "--executor_input" {
+			actualExecutorInput = args[i+1]
+			break
+		}
+	}
+	assert.NotEmpty(t, actualExecutorInput, "--executor_input not found")
+
+	var parsed map[string]any
+	err = json.Unmarshal([]byte(actualExecutorInput), &parsed)
+	assert.NoError(t, err)
+
+	inputs := parsed["inputs"].(map[string]any)
+	paramValues := inputs["parameterValues"].(map[string]any)
+	config := paramValues["config"].(map[string]any)
+
+	assert.Equal(t, "116", config["category_ids"])
+	assert.Equal(t, "dump_filename_test.txt", config["dump_filename"])
+	assert.Equal(t, "sphinx-default-host.ru", config["sphinx_host"])
+	assert.Equal(t, "9312", config["sphinx_port"])
+}
+
 func Test_get_log_Writer(t *testing.T) {
 	old := osCreateFunc
 	defer func() { osCreateFunc = old }()