Add dynamic secret name support in use_secret_as_volume function

- Implemented support for dynamically specifying secret names in the `use_secret_as_volume` function.
- Updated driver code to handle secret name substitution at runtime based on input parameters.
- Introduced a `{{my_secret}}` template string representation for secret names in the compiled DSL.
- Added a test to validate secret name template creation in IR.

Co-authored-by: Greg Sheremeta <gshereme@redhat.com>
Signed-off-by: ddalvi <ddalvi@redhat.com>

backend/src/v2/driver/driver.go

@@ -17,10 +17,12 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/kubeflow/pipelines/backend/src/v2/objectstore"
 	"strconv"
+	"strings"
 	"time"
 
+	"github.com/kubeflow/pipelines/backend/src/v2/objectstore"
+
 	"github.com/golang/glog"
 	"github.com/golang/protobuf/ptypes/timestamp"
 	"github.com/google/uuid"
@@ -536,6 +538,32 @@ func extendPodSpecPatch(
 	// Get secret mount information
 	for _, secretAsVolume := range kubernetesExecutorConfig.GetSecretAsVolume() {
 		optional := secretAsVolume.Optional != nil && *secretAsVolume.Optional
+
+		secretName := secretAsVolume.GetSecretName()
+
+		if strings.HasPrefix(secretName, "{{") && strings.HasSuffix(secretName, "}}") {
+			// Strip the braces
+			key := secretName[2 : len(secretName)-2]
+
+			// Check if the key exists in the parameter inputs map
+			inputParams, _, err := dag.Execution.GetParameters()
+			if err != nil {
+				return fmt.Errorf("failed to get input parameters: %v", err)
+			}
+
+			val, ok := inputParams[key]
+			if !ok {
+				return fmt.Errorf("dynamic secret name key '%s' not found in input parameters", key)
+			}
+
+			secretName = val.GetStringValue()
+			if secretName == "" {
+				return fmt.Errorf("secret name for key '%s' is empty", key)
+			}
+		} else if strings.TrimSpace(secretName) == "" {
+			return fmt.Errorf("secret name is empty or invalid")
+		}
+
 		secretVolume := k8score.Volume{
 			Name: secretAsVolume.GetSecretName(),
 			VolumeSource: k8score.VolumeSource{

kubernetes_platform/python/kfp/kubernetes/secret.py

@@ -76,13 +76,16 @@ def use_secret_as_volume(
     Returns:
         Task object with updated secret configuration.
     """
-    # Extract the actual string value if secret_name is a PipelineParameterChannel
-    if isinstance(secret_name, PipelineParameterChannel):
-        secret_name = secret_name.name
     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,
     )

kubernetes_platform/python/test/unit/test_secret.py

@@ -157,14 +157,15 @@ def my_pipeline(secret_name: str = 'my-secret'):
                 mount_path='secretpath',
             )
 
+        expected_secret_name = '{{my_secret}}'
         assert json_format.MessageToDict(my_pipeline.platform_spec) == {
             'platforms': {
                 'kubernetes': {
                     'deploymentSpec': {
                         'executors': {
                             'exec-comp': {
                                 'secretAsVolume': [{
-                                    'secretName': 'secret_name',
+                                    'secretName': 'expected_secret_name',
                                     'mountPath': 'secretpath',
                                     'optional': False
                                 }]