feat(backend): add subPath support for PVC mounts

Signed-off-by: vmridul <vermamridul1302@gmail.com>

Author: vmridul

.github/workflows/trivy-cache.yml

@@ -13,22 +13,16 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Run Trivy vulnerability scanner in repo mode
-        uses: aquasecurity/trivy-action@0.33.1
+        uses: aquasecurity/trivy-action@0.28.0
         with:
           scan-type: 'fs'
           ignore-unfixed: true
           format: 'sarif'
           output: 'trivy-results.sarif'
           severity: 'CRITICAL,HIGH'
           skip-dirs: 'components'
-        env:
-          # Use ECR instead of GitHub Container Registry to avoid rate limiting.
-          TRIVY_DB_REPOSITORY: public.ecr.aws/aquasecurity/trivy-db:2
-          TRIVY_JAVA_DB_REPOSITORY: public.ecr.aws/aquasecurity/trivy-java-db:1
-          TRIVY_SKIP_DB_UPDATE: true
-          TRIVY_SKIP_JAVA_DB_UPDATE: true
 
       - name: Upload Trivy scan results to GitHub Security tab
-        uses: github/codeql-action/upload-sarif@v4
+        uses: github/codeql-action/upload-sarif@v3
         with:
           sarif_file: 'trivy-results.sarif'

.github/workflows/trivy.yml

@@ -0,0 +1,30 @@
+name: Workflow Linter
+on:
+  push:
+    branches:
+      - 'master'
+      - 'release-*'
+
+  pull_request:
+    paths:
+      - '.github/workflows/**'
+
+jobs:
+  actionlint:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      # actionlint requires Go to run.
+      - name: Setup Go 🛠️
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.24.7' # Use a recent, supported Go version
+
+      - name: Install actionlint
+        run: go install github.com/rhysd/actionlint/cmd/actionlint@latest
+
+      # actionlint automatically finds all workflow files in .github/workflows/
+      - name: Run actionlint
+        run: actionlint -color

.github/workflows/workflow-linter.yml

@@ -73,8 +73,3 @@ repos:
         language: golang
         require_serial: true
         pass_filenames: false
-  - repo: https://github.com/rhysd/actionlint
-    rev: v1.7.9
-    hooks:
-      - id: actionlint
-    types: ["yml", "yaml"]

.pre-commit-config.yaml

@@ -19,6 +19,7 @@
 import collections
 import kubernetes
 import yaml
+import urllib3
 from time import sleep
 import lru
 
@@ -399,9 +400,7 @@ def is_kfp_v2_pod(pod) -> bool:
         # If the for loop ended, a server-side timeout occurred. Continue watching.
         pass
 
-    except Exception as e:
-        # Handle any errors, print stack trace, and continue watching.
-        import traceback
-        print(traceback.format_exc())
+    except urllib3.exceptions.ReadTimeoutError as e:
+        # Client side timeout, continue watching.
         continue
 

backend/metadata_writer/src/metadata_writer.py

@@ -460,6 +460,7 @@ func (s *RunServerV1) ReportRunMetricsV1(ctx context.Context, request *apiv1beta
 	return &apiv1beta1.ReportRunMetricsResponse{Results: apiResults}, nil
 }
 
+
 // Terminates a run.
 // Applies common logic on v1beta1 and v2beta1 API.
 func (s *BaseRunServer) terminateRun(ctx context.Context, runId string) error {
@@ -604,6 +605,7 @@ func (s *RunServer) DeleteRun(ctx context.Context, request *apiv2beta1.DeleteRun
 	return &emptypb.Empty{}, nil
 }
 
+
 // Terminates a run.
 // Supports v2beta1 behavior.
 func (s *RunServer) TerminateRun(ctx context.Context, request *apiv2beta1.TerminateRunRequest) (*emptypb.Empty, error) {

backend/src/apiserver/server/run_server.go

@@ -1479,6 +1479,7 @@ func TestCanAccessRun_Unauthenticated(t *testing.T) {
 	)
 }
 
+
 func TestRetryRun(t *testing.T) {
 	clients, manager, experiment := initWithExperiment(t)
 	defer clients.Close()

backend/src/apiserver/server/run_server_test.go

@@ -105,9 +105,6 @@ func (t *V2Spec) ScheduledWorkflow(modelJob *model.Job) (*scheduledworkflow.Sche
 		return nil, util.Wrap(err, "Failed to convert runtime config")
 	}
 	job.RuntimeConfig = jobRuntimeConfig
-
-	// Parameter macros like [[CurrentTime]], [[RunUUID]], [[ScheduledTime]], [[Index]] are not formatted here.
-	// They remain unformatted in the ScheduledWorkflow spec and will be formatted by the scheduled workflow controller
 	if err = t.validatePipelineJobInputs(job); err != nil {
 		return nil, util.Wrap(err, "invalid pipeline job inputs")
 	}
@@ -319,37 +316,6 @@ func (t *V2Spec) RunWorkflow(modelRun *model.Run, options RunWorkflowOptions) (u
 		return nil, util.NewInternalServerError(err, "Failed to convert to PipelineJob RuntimeConfig")
 	}
 	job.RuntimeConfig = jobRuntimeConfig
-
-	// Format parameters to expand macros like [[CurrentTime]], [[RunUUID]], [[ScheduledTime]], [[Index]] (V1 forward compatibility).
-	// Uses NewSWFParameterFormatter to support all macros. For standalone runs, [[ScheduledTime]] and [[Index]] remain unformatted
-
-	if job.RuntimeConfig != nil && len(job.RuntimeConfig.GetParameterValues()) > 0 {
-		scheduledEpoch := int64(-1) // disabled by default
-
-		if modelRun.ScheduledAtInSec > 0 {
-			scheduledEpoch = modelRun.ScheduledAtInSec
-		}
-		formatter := util.NewSWFParameterFormatter(
-			options.RunID,
-			scheduledEpoch,
-			options.RunAt,
-			-1,
-		)
-		// Convert structpb.Value to strings, format, convert back
-		paramValues := job.RuntimeConfig.GetParameterValues()
-		stringParams := make(map[string]string)
-		for key, val := range paramValues {
-			if strVal := val.GetStringValue(); strVal != "" {
-				stringParams[key] = strVal
-			}
-		}
-		// Format the string parameters
-		formattedParams := formatter.FormatWorkflowParameters(stringParams)
-		// Convert formatted strings back to structpb.Value
-		for key, formattedVal := range formattedParams {
-			paramValues[key] = structpb.NewStringValue(formattedVal)
-		}
-	}
 	if err = t.validatePipelineJobInputs(job); err != nil {
 		return nil, util.Wrap(err, "invalid pipeline job inputs")
 	}

backend/src/apiserver/template/v2_template.go

@@ -1011,12 +1011,14 @@ func makeVolumeMountPatch(
 		pvcName := resolvedPvcName.GetStringValue()
 
 		pvcMountPath := pvcMount.GetMountPath()
+		pvcSubPath := pvcMount.GetSubPath()
 		if pvcName == "" || pvcMountPath == "" {
 			return nil, nil, fmt.Errorf("failed to mount volume, missing mountpath or pvc name")
 		}
 		volumeMount := k8score.VolumeMount{
 			Name:      pvcName,
 			MountPath: pvcMountPath,
+			SubPath:   pvcSubPath,
 		}
 		volume := k8score.Volume{
 			Name: pvcName,

backend/src/v2/driver/k8s.go

@@ -27,6 +27,7 @@ func Test_makeVolumeMountPatch(t *testing.T) {
 		args        args
 		wantPath    string
 		wantName    string
+		wantSubPath string
 		inputParams map[string]*structpb.Value
 	}{
 		{
@@ -43,6 +44,7 @@ func Test_makeVolumeMountPatch(t *testing.T) {
 			},
 			"/mnt/path",
 			"pvc-name",
+			"",
 			nil,
 		},
 		{
@@ -60,6 +62,7 @@ func Test_makeVolumeMountPatch(t *testing.T) {
 			},
 			"/mnt/path",
 			"pvc-name",
+			"",
 			nil,
 		},
 		{
@@ -76,10 +79,29 @@ func Test_makeVolumeMountPatch(t *testing.T) {
 			},
 			"/mnt/path",
 			"pvc-name",
+			"",
 			map[string]*structpb.Value{
 				"param_1": structpb.NewStringValue("pvc-name"),
 			},
 		},
+		{
+			"pvc with subPath",
+			args{
+				[]*kubernetesplatform.PvcMount{
+					{
+						MountPath:        "/mnt/path",
+						SubPath:          "logs/experiment1",
+						PvcNameParameter: inputParamConstant("pvc-name"),
+					},
+				},
+				nil,
+				nil,
+			},
+			"/mnt/path",
+			"pvc-name",
+			"logs/experiment1",
+			nil,
+		},
 	}
 
 	for _, tt := range tests {
@@ -100,6 +122,7 @@ func Test_makeVolumeMountPatch(t *testing.T) {
 			assert.Equal(t, volumeMounts[0].Name, tt.wantName)
 			assert.Equal(t, volumes[0].Name, tt.wantName)
 			assert.Equal(t, volumes[0].PersistentVolumeClaim.ClaimName, tt.wantName)
+			assert.Equal(t, volumeMounts[0].SubPath, tt.wantSubPath)
 		})
 	}
 }