fix

Signed-off-by: agoins <[email protected]>

# Conflicts:
#	samples/v2/sample_test.py

Author: alyssacgoins

api/v2alpha1/go/pipelinespec/pipeline_spec.pb.go

@@ -368,6 +368,7 @@ func executeV2(
 		return nil, nil, err
 	}
 
+	// ensure executorOutput contains custompath
 	executorOutput, err := execute(
 		ctx,
 		executorInput,
@@ -545,27 +546,32 @@ func uploadOutputArtifacts(ctx context.Context, executorInput *pipelinespec.Exec
 				mergeRuntimeArtifacts(list.Artifacts[0], outputArtifact)
 			}
 
-			// If artifact customPath is set, upload to remote storages from this path. Otherwise, upload from local path.
-			var artifactDir string
+			var dir string
+			var copyDir string
 			var err error
+			// If artifact customPath is set, upload to remote storages from this path. Otherwise, upload from local path (outputArtifact.URI)
 			if *outputArtifact.CustomPath != "" {
-				artifactDir = *outputArtifact.CustomPath
+				dir = *outputArtifact.CustomPath
+				copyDir = *outputArtifact.CustomPath
 			} else {
-				artifactDir, err = LocalPathForURI(outputArtifact.Uri)
+				dir = outputArtifact.Uri
+				copyDir, err = LocalPathForURI(dir)
+				if err != nil {
+					glog.Warningf("Output Artifact %q does not have a recognized storage URI %q. Skipping uploading to remote storage.", name, dir)
+				}
 			}
-			if err != nil {
-				glog.Warningf("Output Artifact %q does not have a recognized storage URI %q. Skipping uploading to remote storage.", name, outputArtifact.Uri)
-			} else if !strings.HasPrefix(outputArtifact.Uri, "oci://") {
-				blobKey, err := opts.bucketConfig.KeyFromURI(outputArtifact.Uri)
+
+			if !strings.HasPrefix(dir, "oci://") {
+				blobKey, err := opts.bucketConfig.KeyFromURI(dir)
 				if err != nil {
 					return nil, fmt.Errorf("failed to upload output artifact %q: %w", name, err)
 				}
-				if err := objectstore.UploadBlob(ctx, opts.bucket, artifactDir, blobKey); err != nil {
+				if err := objectstore.UploadBlob(ctx, opts.bucket, copyDir, blobKey); err != nil {
 					//  We allow components to not produce output files
 					if errors.Is(err, os.ErrNotExist) {
-						glog.Warningf("Local filepath %q does not exist", artifactDir)
+						glog.Warningf("Local filepath %q does not exist", copyDir)
 					} else {
-						return nil, fmt.Errorf("failed to upload output artifact %q to remote storage URI %q: %w", name, outputArtifact.Uri, err)
+						return nil, fmt.Errorf("failed to upload output artifact %q to remote storage URI %q: %w", name, dir, err)
 					}
 				}
 			}

backend/src/v2/component/launcher_v2.go

@@ -52,6 +52,7 @@ var addNumbersComponent = &pipelinespec.ComponentSpec{
 
 // Tests that launcher correctly executes the user component and successfully writes output parameters to file.
 func Test_executeV2_Parameters(t *testing.T) {
+	customLocalPath := "/tmp/custom/artifact.txt"
 	tests := []struct {
 		name          string
 		executorInput *pipelinespec.ExecutorInput
@@ -64,6 +65,17 @@ func Test_executeV2_Parameters(t *testing.T) {
 				Inputs: &pipelinespec.ExecutorInput_Inputs{
 					ParameterValues: map[string]*structpb.Value{"a": structpb.NewNumberValue(1), "b": structpb.NewNumberValue(2)},
 				},
+				Outputs: &pipelinespec.ExecutorInput_Outputs{
+					Artifacts: map[string]*pipelinespec.ArtifactList{
+						"dataset": {
+							Artifacts: []*pipelinespec.RuntimeArtifact{
+								{
+									Uri: customLocalPath,
+								},
+							},
+						},
+					},
+				},
 			},
 			[]string{"-c", "test {{$.inputs.parameters['a']}} -eq 1 || exit 1\ntest {{$.inputs.parameters['b']}} -eq 2 || exit 1"},
 			false,

backend/src/v2/component/launcher_v2_test.go

@@ -0,0 +1,24 @@
+from sdk.python.kfp.dsl import Output
+from sdk.python.kfp.dsl.types.artifact_types import Artifact
+from sdk.python.kfp import dsl
+
+@dsl.component
+def create_list() -> list:
+    return [1, 2, 3, 4]
+
+@dsl.component
+def append_to_list(digit: int, input_list: Output[Artifact]) -> list:
+    input_list.append(digit)
+    return input_list
+
+@dsl.component
+def validate_custom_path(exp_path: str, input_list: Output[Artifact]) -> bool:
+    #todo: is this the correct comparison? (or should use != instead?)
+    if input_list.path is not exp_path:
+        raise ValueError(f"File uri is {input_list.path} but should be {exp_path}.")
+
+@dsl.pipeline
+def pipeline_with_custom_path_artifact():
+    task1 = create_list()
+    task1.output.set_custom_path('/etc/test/file/path')
+    # task2 = validate_custom_path(path='/etc/test/file/path', input_list=task1.output)

samples/v2/component_with_runtime_artifct_custom_path.py

@@ -0,0 +1,25 @@
+from kfp.dsl import Output
+from kfp.dsl.types.artifact_types import Artifact
+from kfp.v2 import dsl
+
+
+@dsl.component
+def create_list() -> list:
+    return [1, 2, 3, 4]
+
+@dsl.component
+def append_to_list(digit: int, input_list: Output[Artifact]) -> list:
+    input_list.append(digit)
+    return input_list
+
+@dsl.component
+def validate_custom_path(exp_path: str, input_list: Output[Artifact]) -> bool:
+    #todo: is this the correct comparison? (or should use != instead?)
+    if input_list.path is not exp_path:
+        raise ValueError(f"File uri is {input_list.path} but should be {exp_path}.")
+
+@dsl.pipeline
+def pipeline_with_custom_path_artifact():
+    task1 = create_list()
+    task1.output.set_custom_path('/etc/test/file/path')
+    task2 = validate_custom_path(path='/etc/test/file/path', input_list=task1.output)