feat(backend): Pipeline specpatch from APIServer envvar

- Allows a KFP admin to apply a custom specpatch to all incoming
  pipelines
- Useful for installment-wide configuration, such as workflow ttl
- Take in a JSON string provided via the environment variable
  `COMPILED_PIPELINE_SPEC_PATCH`

Signed-off-by: Giulio Frasca <[email protected]>

Author: gmfrasca

backend/src/apiserver/common/config.go

@@ -32,6 +32,7 @@ const (
 	KubeflowUserIDPrefix                    string = "KUBEFLOW_USERID_PREFIX"
 	UpdatePipelineVersionByDefault          string = "AUTO_UPDATE_PIPELINE_DEFAULT_VERSION"
 	TokenReviewAudience                     string = "TOKEN_REVIEW_AUDIENCE"
+	CompiledPipelineSpecPatch               string = "COMPILED_PIPELINE_SPEC_PATCH"
 )
 
 func IsPipelineVersionUpdatedByDefault() bool {
@@ -127,3 +128,7 @@ func GetKubeflowUserIDPrefix() string {
 func GetTokenReviewAudience() string {
 	return GetStringConfigWithDefault(TokenReviewAudience, DefaultTokenReviewAudience)
 }
+
+func GetCompiledPipelineSpecPatch() string {
+	return GetStringConfigWithDefault(CompiledPipelineSpecPatch, "{}")
+}

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

@@ -183,8 +183,17 @@ func Compile(jobArg *pipelinespec.PipelineJob, kubernetesSpecArg *pipelinespec.S
 
 	// compile
 	err = compiler.Accept(job, kubernetesSpec, c)
+	if err != nil {
+		return nil, err
+	}
+
+	// Apply any workflow spec patches from environment variable
+	patchJSON := common.GetCompiledPipelineSpecPatch()
+	if err := c.ApplyWorkflowSpecPatch(patchJSON); err != nil {
+		return nil, fmt.Errorf("failed to apply workflow spec patch: %w", err)
+	}
 
-	return c.wf, err
+	return c.wf, nil
 }
 
 func retrieveLastValidString(s string) string {

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

@@ -15,7 +15,13 @@
 package argocompiler
 
 import (
+	"encoding/json"
+	"fmt"
+
+	wfapi "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
+	log "github.com/sirupsen/logrus"
 	"google.golang.org/protobuf/types/known/structpb"
+	"k8s.io/apimachinery/pkg/util/strategicpatch"
 )
 
 func (c *workflowCompiler) AddKubernetesSpec(name string, kubernetesSpec *structpb.Struct) error {
@@ -25,3 +31,52 @@ func (c *workflowCompiler) AddKubernetesSpec(name string, kubernetesSpec *struct
 	}
 	return nil
 }
+
+// ApplyWorkflowSpecPatch applies a JSON patch to the compiled workflow specification.
+// It validates the JSON and applies it using Kubernetes strategic merge patch.
+// Only the workflow's "spec" field can be patched for security reasons.
+func (c *workflowCompiler) ApplyWorkflowSpecPatch(patchJSON string) error {
+	if c.wf == nil {
+		return fmt.Errorf("workflow is nil")
+	}
+
+	// Check for empty patch string
+	if patchJSON == "" {
+		log.Debug("Empty workflow spec patch string provided, skipping patching")
+		return nil
+	}
+
+	log.Debug("Applying workflow spec patch")
+
+	// Validate that the patch is valid JSON by attempting to unmarshal it
+	var specPatchValidation map[string]interface{}
+	if err := json.Unmarshal([]byte(patchJSON), &specPatchValidation); err != nil {
+		return fmt.Errorf("invalid JSON in COMPILED_PIPELINE_SPEC_PATCH: %w", err)
+	}
+
+	// Check if the patch is empty (no fields to patch)
+	if len(specPatchValidation) == 0 {
+		log.Debug("Empty workflow spec patch provided, skipping patching")
+		return nil
+	}
+
+	// Convert the current workflow spec to JSON
+	originalSpecJSON, err := json.Marshal(c.wf.Spec)
+	if err != nil {
+		return fmt.Errorf("failed to marshal workflow spec to JSON: %w", err)
+	}
+
+	// Apply the strategic merge patch to the spec directly
+	patchedSpecJSON, err := strategicpatch.StrategicMergePatch(originalSpecJSON, []byte(patchJSON), wfapi.WorkflowSpec{})
+	if err != nil {
+		return fmt.Errorf("failed to apply strategic merge patch to workflow spec: %w", err)
+	}
+
+	// Unmarshal the patched spec back into the workflow
+	if err := json.Unmarshal(patchedSpecJSON, &c.wf.Spec); err != nil {
+		return fmt.Errorf("failed to unmarshal patched workflow spec: %w", err)
+	}
+
+	log.Debug("Successfully applied workflow spec patch")
+	return nil
+}

manifests/kustomize/base/pipeline/ml-pipeline-apiserver-deployment.yaml

@@ -126,6 +126,9 @@ spec:
               value: ghcr.io/kubeflow/kfp-driver:2.14.3
             - name: V2_LAUNCHER_IMAGE
               value: ghcr.io/kubeflow/kfp-launcher:2.14.3
+            # JSON patch to apply to compiled workflow specifications
+            - name: COMPILED_PIPELINE_SPEC_PATCH
+              value: "{}"
           image: ghcr.io/kubeflow/kfp-api-server:dummy
           imagePullPolicy: IfNotPresent
           name: ml-pipeline-api-server