WIP - do not merge - support parameter inputs in kfp-kubernetes

proof of concept for supporting parameter inputs in kfp-kubernetes.
We've added support first to secret.py.

Ref: kubeflow#10534
Ref: kubeflow#10914

backend/src/v2/driver/driver.go

@@ -544,14 +544,34 @@ func extendPodSpecPatch(
 	// Get secret mount information
 	for _, secretAsVolume := range kubernetesExecutorConfig.GetSecretAsVolume() {
 		optional := secretAsVolume.Optional != nil && *secretAsVolume.Optional
+
+		// Read the secret name passed in. If it has braces around it, it is dynamic, and we
+		// need to check the parameter inputs map for the actual runtime value to use.
+		// If there are no braces, it IS the name of the secret to mount -- just use it.
+		secretName := secretAsVolume.GetSecretName()
+		if strings.HasPrefix(secretName, "{{") {
+			// it looks like this in the protobuf:
+			// secretName: '{{my_secret}}'
+			// strip the braces
+			key := secretName[2 : len(secretName)-2]
+
+			// get the value from the parameter inputs map
+			inputParams, _, err := dag.Execution.GetParameters()
+			if err != nil {
+				return err
+			}
+			val := inputParams[key]
+			secretName = val.GetStringValue()
+		}
+
 		secretVolume := k8score.Volume{
-			Name: secretAsVolume.GetSecretName(),
+			Name: secretName,
 			VolumeSource: k8score.VolumeSource{
-				Secret: &k8score.SecretVolumeSource{SecretName: secretAsVolume.GetSecretName(), Optional: &optional},
+				Secret: &k8score.SecretVolumeSource{SecretName: secretName, Optional: &optional},
 			},
 		}
 		secretVolumeMount := k8score.VolumeMount{
-			Name:      secretAsVolume.GetSecretName(),
+			Name:      secretName,
 			MountPath: secretAsVolume.GetMountPath(),
 		}
 		podSpec.Volumes = append(podSpec.Volumes, secretVolume)

kubernetes_platform/python/kfp/kubernetes/secret.py

@@ -18,7 +18,8 @@
 from kfp.dsl import PipelineTask
 from kfp.kubernetes import common
 from kfp.kubernetes import kubernetes_executor_config_pb2 as pb
-
+from typing import Union
+from kfp.dsl.pipeline_channel import PipelineParameterChannel
 
 def use_secret_as_env(
     task: PipelineTask,
@@ -59,7 +60,7 @@ def use_secret_as_env(
 
 def use_secret_as_volume(
     task: PipelineTask,
-    secret_name: str,
+    secret_name: Union[str, PipelineParameterChannel],
     mount_path: str,
     optional: bool = False,
 ) -> PipelineTask:
@@ -75,11 +76,16 @@ def use_secret_as_volume(
     Returns:
         Task object with updated secret configuration.
     """
-
     msg = common.get_existing_kubernetes_config_as_message(task)
 
+    val = secret_name
+    # if secret_name is a PipelineParameterChannel, then we don't know what secret to mount until RUNTIME
+    # so, treat is as a map KEY instead of a secret name
+    if isinstance(secret_name, PipelineParameterChannel):
+        val = "{{" + secret_name.name + "}}"
+
     secret_as_vol = pb.SecretAsVolume(
-        secret_name=secret_name,
+        secret_name=val,
         mount_path=mount_path,
         optional=optional,
     )

test-pipeline-1.py

@@ -0,0 +1,18 @@
+from kfp import dsl
+from kfp import kubernetes
+
+
+@dsl.component
+def comp():
+    pass
+
+@dsl.pipeline
+def my_pipeline(my_secret: str):
+    task = comp()
+    task.set_caching_options(False)
+
+    kubernetes.use_secret_as_volume(
+        task,
+        secret_name=my_secret,
+        mount_path='/mnt/my_secret',
+    )

test-pipeline-1.yaml

@@ -0,0 +1,65 @@
+# PIPELINE DEFINITION
+# Name: my-pipeline
+# Inputs:
+#    my_secret: str
+components:
+  comp-comp:
+    executorLabel: exec-comp
+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.8.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():\n    pass\n\n"
+        image: python:3.8
+pipelineInfo:
+  name: my-pipeline
+root:
+  dag:
+    tasks:
+      comp:
+        cachingOptions:
+          enableCache: false
+        componentRef:
+          name: comp-comp
+        taskInfo:
+          name: comp
+  inputDefinitions:
+    parameters:
+      my_secret:
+        parameterType: STRING
+schemaVersion: 2.1.0
+sdkVersion: kfp-2.8.0
+---
+platforms:
+  kubernetes:
+    deploymentSpec:
+      executors:
+        exec-comp:
+          secretAsVolume:
+          - mountPath: /mnt/my_secret
+            optional: false
+            secretName: '{{my_secret}}'

test-pipeline-2.py

@@ -0,0 +1,18 @@
+from kfp import dsl
+from kfp import kubernetes
+
+
+@dsl.component
+def comp():
+    pass
+
+@dsl.pipeline
+def my_pipeline(my_secret: str = 'some-default-secret'):
+    task = comp()
+    task.set_caching_options(False)
+
+    kubernetes.use_secret_as_volume(
+        task,
+        secret_name=my_secret,
+        mount_path='/mnt/my_secret',
+    )

test-pipeline-2.yaml

@@ -0,0 +1,67 @@
+# PIPELINE DEFINITION
+# Name: my-pipeline
+# Inputs:
+#    my_secret: str [Default: 'some-default-secret']
+components:
+  comp-comp:
+    executorLabel: exec-comp
+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.8.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():\n    pass\n\n"
+        image: python:3.8
+pipelineInfo:
+  name: my-pipeline
+root:
+  dag:
+    tasks:
+      comp:
+        cachingOptions:
+          enableCache: false
+        componentRef:
+          name: comp-comp
+        taskInfo:
+          name: comp
+  inputDefinitions:
+    parameters:
+      my_secret:
+        defaultValue: some-default-secret
+        isOptional: true
+        parameterType: STRING
+schemaVersion: 2.1.0
+sdkVersion: kfp-2.8.0
+---
+platforms:
+  kubernetes:
+    deploymentSpec:
+      executors:
+        exec-comp:
+          secretAsVolume:
+          - mountPath: /mnt/my_secret
+            optional: false
+            secretName: '{{my_secret}}'

test-pipeline-3.py

@@ -0,0 +1,17 @@
+from kfp import dsl
+from kfp import kubernetes
+
+
+@dsl.component
+def comp():
+    pass
+
+@dsl.pipeline
+def my_pipeline():
+    task = comp()
+
+    kubernetes.use_secret_as_volume(
+        task,
+        secret_name='my-hardcoded-secret',
+        mount_path='/mnt/my_secret',
+    )

test-pipeline-3.yaml

@@ -0,0 +1,59 @@
+# PIPELINE DEFINITION
+# Name: my-pipeline
+components:
+  comp-comp:
+    executorLabel: exec-comp
+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.8.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():\n    pass\n\n"
+        image: python:3.8
+pipelineInfo:
+  name: my-pipeline
+root:
+  dag:
+    tasks:
+      comp:
+        cachingOptions:
+          enableCache: true
+        componentRef:
+          name: comp-comp
+        taskInfo:
+          name: comp
+schemaVersion: 2.1.0
+sdkVersion: kfp-2.8.0
+---
+platforms:
+  kubernetes:
+    deploymentSpec:
+      executors:
+        exec-comp:
+          secretAsVolume:
+          - mountPath: /mnt/my_secret
+            optional: false
+            secretName: my-hardcoded-secret

test-pipeline-4.py

@@ -0,0 +1,19 @@
+from kfp import dsl
+from kfp import kubernetes
+
+
+@dsl.component
+def comp():
+    pass
+
+@dsl.pipeline
+def my_pipeline():
+    task = comp()
+
+    my_secret = 'my-hardcoded-secret2'
+
+    kubernetes.use_secret_as_volume(
+        task,
+        secret_name=my_secret,
+        mount_path='/mnt/my_secret',
+    )