Merge pull request #182 from hbelmiro/fix-cache

UPSTREAM<drop> Fix boolean flag syntax for --cache_disabled and update test to cover pipelines with outputs

Author: openshift-merge-bot[bot]

backend/src/v2/compiler/argocompiler/container.go

@@ -208,7 +208,7 @@ func (c *workflowCompiler) addContainerDriverTemplate() string {
 		"--ca_cert_path", common.GetCaCertPath(),
 	}
 	if c.cacheDisabled {
-		args = append(args, "--cache_disabled", "true")
+		args = append(args, "--cache_disabled")
 	}
 	if value, ok := os.LookupEnv(PipelineLogLevelEnvVar); ok {
 		args = append(args, "--log_level", value)
@@ -352,7 +352,7 @@ func (c *workflowCompiler) addContainerExecutorTemplate(task *pipelinespec.Pipel
 		"--copy", component.KFPLauncherPath,
 	}
 	if c.cacheDisabled {
-		args = append(args, "--cache_disabled", "true")
+		args = append(args, "--cache_disabled")
 	}
 	if value, ok := os.LookupEnv(PipelineLogLevelEnvVar); ok {
 		args = append(args, "--log_level", value)

backend/src/v2/compiler/argocompiler/dag.go

@@ -565,7 +565,7 @@ func (c *workflowCompiler) addDAGDriverTemplate() string {
 		"--ca_cert_path", common.GetCaCertPath(),
 	}
 	if c.cacheDisabled {
-		args = append(args, "--cache_disabled", "true")
+		args = append(args, "--cache_disabled")
 	}
 	if value, ok := os.LookupEnv(PipelineLogLevelEnvVar); ok {
 		args = append(args, "--log_level", value)

backend/src/v2/compiler/argocompiler/importer.go

@@ -85,7 +85,7 @@ func (c *workflowCompiler) addImporterTemplate() string {
 		"--ca_cert_path", common.GetCaCertPath(),
 	}
 	if c.cacheDisabled {
-		args = append(args, "--cache_disabled", "true")
+		args = append(args, "--cache_disabled")
 	}
 	if value, ok := os.LookupEnv(PipelineLogLevelEnvVar); ok {
 		args = append(args, "--log_level", value)

backend/src/v2/compiler/argocompiler/testdata/hello_world_cache_disabled.yaml

@@ -72,7 +72,6 @@ spec:
       - --ca_cert_path
       - ""
       - --cache_disabled
-      - 'true'
       command:
       - driver
       env:
@@ -176,7 +175,6 @@ spec:
       - --copy
       - /kfp-launcher/launch
       - --cache_disabled
-      - "true"
       command:
       - launcher-v2
       image: ghcr.io/kubeflow/kfp-launcher
@@ -287,7 +285,6 @@ spec:
       - --ca_cert_path
       - ""
       - --cache_disabled
-      - 'true'
       command:
       - driver
       env:

backend/src/v2/driver/driver.go

@@ -216,7 +216,7 @@ func initPodSpecPatch(
 		"--ca_cert_path", caCertPath,
 	}
 	if cacheDisabled == "true" {
-		launcherCmd = append(launcherCmd, "--cache_disabled", cacheDisabled)
+		launcherCmd = append(launcherCmd, "--cache_disabled")
 	}
 	if pipelineLogLevel != "1" {
 		// Add log level to user code launcher if not default (set to 1)

backend/test/v2/integration/cache_test.go

@@ -136,6 +136,11 @@ func (s *CacheTestSuite) TestCacheRecurringRun() {
 				IntervalSecond: 60,
 			},
 		},
+		RuntimeConfig: &recurring_run_model.V2beta1RuntimeConfig{
+			Parameters: map[string]interface{}{
+				"message": "Hello world",
+			},
+		},
 	}}
 	helloWorldRecurringRun, err := s.recurringRunClient.Create(createRecurringRunRequest)
 	require.NoError(t, err)
@@ -251,6 +256,11 @@ func (s *CacheTestSuite) createRun(pipelineVersion *pipeline_upload_model.V2beta
 			PipelineID:        pipelineVersion.PipelineID,
 			PipelineVersionID: pipelineVersion.PipelineVersionID,
 		},
+		RuntimeConfig: &run_model.V2beta1RuntimeConfig{
+			Parameters: map[string]interface{}{
+				"message": "Hello world",
+			},
+		},
 	}}
 	pipelineRunDetail, err := s.runClient.Create(createRunRequest)
 	require.NoError(s.T(), err)
@@ -268,12 +278,12 @@ func (s *CacheTestSuite) createRun(pipelineVersion *pipeline_upload_model.V2beta
 }
 
 func (s *CacheTestSuite) preparePipeline() *pipeline_upload_model.V2beta1PipelineVersion {
-	pipeline, err := s.pipelineUploadClient.UploadFile("../resources/hello-world.yaml", uploadParams.NewUploadPipelineParams())
+	pipeline, err := s.pipelineUploadClient.UploadFile("../resources/hello-world-with-returning-component.yaml", uploadParams.NewUploadPipelineParams())
 	require.NoError(s.T(), err)
 
 	time.Sleep(1 * time.Second)
 	pipelineVersion, err := s.pipelineUploadClient.UploadPipelineVersion(
-		"../resources/hello-world.yaml", &uploadParams.UploadPipelineVersionParams{
+		"../resources/hello-world-with-returning-component.yaml", &uploadParams.UploadPipelineVersionParams{
 			Name:       util.StringPointer("hello-world-version"),
 			Pipelineid: util.StringPointer(pipeline.PipelineID),
 		})

backend/test/v2/resources/hello-world-with-returning-component.py

@@ -0,0 +1,12 @@
+from kfp import dsl
+
+
+@dsl.component(base_image="public.ecr.aws/docker/library/python:3.12")
+def comp(message: str) -> str:
+    print(message)
+    return message
+
+
+@dsl.pipeline
+def my_pipeline(message: str) -> str:
+    return comp(message=message).output

backend/test/v2/resources/hello-world-with-returning-component.yaml

@@ -0,0 +1,80 @@
+# PIPELINE DEFINITION
+# Name: my-pipeline
+# Inputs:
+#    message: str
+# Outputs:
+#    Output: str
+components:
+  comp-comp:
+    executorLabel: exec-comp
+    inputDefinitions:
+      parameters:
+        message:
+          parameterType: STRING
+    outputDefinitions:
+      parameters:
+        Output:
+          parameterType: STRING
+deploymentSpec:
+  executors:
+    exec-comp:
+      container:
+        args:
+        - --executor_input
+        - '{{$}}'
+        - --function_to_execute
+        - comp
+        command:
+        - sh
+        - -c
+        - "\nif ! [ -x \"$(command -v pip)\" ]; then\n    python3 -m ensurepip ||\
+          \ python3 -m ensurepip --user || apt-get install python3-pip\nfi\n\nPIP_DISABLE_PIP_VERSION_CHECK=1\
+          \ python3 -m pip install --quiet --no-warn-script-location 'kfp==2.13.0'\
+          \ '--no-deps' 'typing-extensions>=3.7.4,<5; python_version<\"3.9\"' && \"\
+          $0\" \"$@\"\n"
+        - sh
+        - -ec
+        - 'program_path=$(mktemp -d)
+
+
+          printf "%s" "$0" > "$program_path/ephemeral_component.py"
+
+          _KFP_RUNTIME=true python3 -m kfp.dsl.executor_main                         --component_module_path                         "$program_path/ephemeral_component.py"                         "$@"
+
+          '
+        - "\nimport kfp\nfrom kfp import dsl\nfrom kfp.dsl import *\nfrom typing import\
+          \ *\n\ndef comp(message: str) -> str:\n    print(message)\n    return message\n\
+          \n"
+        image: public.ecr.aws/docker/library/python:3.12
+pipelineInfo:
+  name: my-pipeline
+root:
+  dag:
+    outputs:
+      parameters:
+        Output:
+          valueFromParameter:
+            outputParameterKey: Output
+            producerSubtask: comp
+    tasks:
+      comp:
+        cachingOptions:
+          enableCache: true
+        componentRef:
+          name: comp-comp
+        inputs:
+          parameters:
+            message:
+              componentInputParameter: message
+        taskInfo:
+          name: comp
+  inputDefinitions:
+    parameters:
+      message:
+        parameterType: STRING
+  outputDefinitions:
+    parameters:
+      Output:
+        parameterType: STRING
+schemaVersion: 2.1.0
+sdkVersion: kfp-2.13.0