feat(job): Enhance mount flag to pass additional mount options for gcsfuse mounts

cmd/job/submit.go

@@ -137,7 +137,7 @@ func init() {
 	SubmitCmd.Flags().StringVarP(&dryRunManifest, "dry-run-out", "o", "", "Path to output the generated Kubernetes manifest instead of applying it.")
 	SubmitCmd.Flags().StringVarP(&platform, "platform", "f", "linux/amd64", "Target platform for the image build (e.g., 'linux/amd64', 'linux/arm64'). Used with --base-image.")
 
-	SubmitCmd.Flags().StringSliceVar(&volumeStr, "mount", nil, "Volumes to mount (format: <src>:<dest>[:<mode>], mode can be 'ro' or 'rw', default 'ro').")
+	SubmitCmd.Flags().StringArrayVar(&volumeStr, "mount", nil, "Volumes to mount (format: <src>;<dest>[;<mode>][;options=<options>], mode can be 'ro' or 'rw', default 'ro').")
 	SubmitCmd.Flags().StringArrayVar(&envVars, "env", []string{}, "Custom environment variables to pass to the workload container in KEY=VALUE format. Can be specified multiple times.")
 
 	SubmitCmd.Flags().StringVarP(&workloadName, "name", "n", "", "Name of the workload to create. Required.")

docs/gcluster_job_guide.md

@@ -161,8 +161,9 @@ If you want to run a job across multiple groups of GPU nodes (e.g., 2 groups of
 
 You can mount Cloud Storage buckets, Filestore instances, existing PVCs (e.g., for Lustre), or host paths using the `--mount` flag.
 
-Mounts must use the format: `--mount "<src>:<dest>[:<mode>]"`
-* `mode` is optional and defaults to `ro` (read-only). To allow writes, append `:rw`.
+Mounts must use the format: `--mount "<src>;<dest>[;<mode>][;options=<options>]"`
+* `mode` is optional and defaults to `ro` (read-only). To allow writes, append `;rw`.
+* `options` is optional and allows passing custom mount options (currently only supported for GCS fuse volumes).
 
 **Supported volume sources (`<src>`):**
 * **Cloud Storage**: `gs://<bucket-name>` (mounts via GCS Fused Driver)
@@ -187,7 +188,7 @@ Mounting a GCS bucket (read-write):
   --compute-type n2-standard-32 \
   --base-image python:3.9-slim \
   --build-context job_details \
-  --mount "gs://<YOUR_BUCKET_NAME>:/data:rw"
+  --mount "gs://<YOUR_BUCKET_NAME>;/data;rw;options=logging:severity:info,enable-atomic-rename-object:true"
 ```
 
 Mounting an existing PVC named `lustre-pvc` (read-only):
@@ -199,7 +200,7 @@ Mounting an existing PVC named `lustre-pvc` (read-only):
   --compute-type n2-standard-32 \
   --base-image python:3.9-slim \
   --build-context job_details \
-  --mount "lustre-pvc:/data"
+  --mount "lustre-pvc;/data"
 ```
 
 ### 4.5 Example: Submit Job with Custom Environment Variables
@@ -1094,7 +1095,7 @@ The `gcluster job submit` command deploys a container image as a job (Kubernetes
 | `--num-slices` | `int` | Number of independent groups/slices to use (Default: `1`). |
 | `--num-nodes` | `int` | Number of nodes to use per group/slice (Default: `1`). Auto-calculated for TPUs based on topology. |
 | `--restarts` | `int` | Maximum number of restarts allowed for the JobSet before marked as failed (Default: `1`). |
-| `--mount` | `stringArray` | Mount storage volumes, buckets, filestore instances, or PVCs using the `<src>:<dest>[:<mode>]` format. Examples of `<src>`: `gs://my-bucket`, `filestore://my-instance/share`, `my-pvc` (for Lustre/etc), or `/host/path`. |
+| `--mount` | `stringArray` | Mount storage volumes, buckets, filestore instances, or PVCs using the `<src>;<dest>[;<mode>][;options=<options>]` format. Examples of `<src>`: `gs://my-bucket`, `filestore://my-instance/share`, `my-pvc` (for Lustre/etc), or `/host/path`. |
 | `--env` | `stringArray` | Custom environment variables to pass exclusively to the user's workload container in KEY=VALUE format (e.g. `--env KEY=VALUE`). Applies to both standard and Pathways workloads. Can be specified multiple times. |
 | `--await-job-completion` | `bool` | If true, the CLI waits for the job to complete before exiting. |
 | `--timeout` | `string` | Time to wait for job completion (e.g., `1h`, `10m`). Used with `--await-job-completion`. |

pkg/orchestrator/gke/gke_job_orchestrator_test.go

@@ -310,9 +310,9 @@ func TestGenerateGKEManifest_Volumes(t *testing.T) {
 		ClusterLocation: "us-central1-a",
 		ComputeType:     "n2-standard-4",
 		RawMounts: []string{
-			"gs://my-bucket:/data",
-			"/host/path:/host",
-			"my-pvc:/pvc",
+			"gs://my-bucket;/data",
+			"/host/path;/host",
+			"my-pvc;/pvc",
 		},
 	}
 

pkg/orchestrator/gke/storage.go

@@ -41,7 +41,7 @@ func (sm *StorageManager) ProcessMounts(mounts []string, job orchestrator.JobDef
 	var mountInfos []MountInfo
 	var additionalManifests []string
 	for i, vStr := range mounts {
-		src, dest, readOnly, err := sm.parseSingleVolume(vStr)
+		src, dest, options, readOnly, err := sm.parseSingleVolume(vStr)
 		if err != nil {
 			return nil, nil, err
 		}
@@ -65,13 +65,15 @@ func (sm *StorageManager) ProcessMounts(mounts []string, job orchestrator.JobDef
 			volType = "hostPath"
 		}
 
-		mountInfos = append(mountInfos, MountInfo{
+		mountInfo := MountInfo{
 			Name:      fmt.Sprintf("vol-%d", i),
 			Source:    src,
 			MountPath: dest,
 			Type:      volType,
 			ReadOnly:  readOnly,
-		})
+			Options:   options,
+		}
+		mountInfos = append(mountInfos, mountInfo)
 	}
 
 	return mountInfos, additionalManifests, nil
@@ -83,7 +85,7 @@ func (sm *StorageManager) ValidateMounts(mounts []string) error {
 	seenDestinations := make(map[string]bool)
 
 	for _, vStr := range mounts {
-		src, dest, _, err := sm.parseSingleVolume(vStr)
+		src, dest, _, _, err := sm.parseSingleVolume(vStr)
 		if err != nil {
 			return err
 		}
@@ -97,68 +99,51 @@ func (sm *StorageManager) ValidateMounts(mounts []string) error {
 		seenSources[src] = true
 		seenDestinations[dest] = true
 	}
-
 	return nil
 }
 
-func (sm *StorageManager) parseSingleVolume(vStr string) (src, dest string, readOnly bool, err error) {
-	src, dest, readOnly, err = parseSrcDest(vStr)
-	if err != nil {
-		return "", "", false, err
-	}
-	if err := validateSrcScheme(src, vStr); err != nil {
-		return "", "", false, err
-	}
-	return src, dest, readOnly, nil
-}
-
 func missingDestOrFormatErr(vStr string) error {
 	if strings.HasPrefix(vStr, "gs://") || strings.HasPrefix(vStr, "filestore://") {
 		return fmt.Errorf("invalid volume format: %s. Missing destination.", vStr)
 	}
-	return fmt.Errorf("invalid volume format: %s. Expected format: <src>:<dest>[:<mode>]", vStr)
-}
-
-func parseVolumeMode(vStr string) (stripped string, readOnly bool, err error) {
-	idx := strings.LastIndex(vStr, ":")
-	if idx <= 0 || idx == len(vStr)-1 {
-		return "", false, missingDestOrFormatErr(vStr)
-	}
-
-	lastPart := vStr[idx+1:]
-	if lastPart == "ro" || lastPart == "rw" {
-		return vStr[:idx], (lastPart == "ro"), nil
-	}
-	return vStr, true, nil
+	return fmt.Errorf("invalid volume format: %s. Expected format: <src>;<dest>[;<mode>][;options=<options>]", vStr)
 }
 
-func parseSrcDest(vStr string) (src, dest string, readOnly bool, err error) {
-	origVStr := vStr
-
-	vStrWithoutMode, readOnly, err := parseVolumeMode(vStr)
-	if err != nil {
-		return "", "", false, err
+func (sm *StorageManager) parseSingleVolume(vStr string) (src, dest, options string, readOnly bool, err error) {
+	parts := strings.Split(vStr, ";")
+	if len(parts) < 2 {
+		return "", "", "", false, missingDestOrFormatErr(vStr)
+	}
+
+	src = parts[0]
+	dest = parts[1]
+	readOnly = true // default
+
+	for _, part := range parts[2:] {
+		if part == "ro" {
+			readOnly = true
+		} else if part == "rw" {
+			readOnly = false
+		} else if strings.HasPrefix(part, "options=") {
+			options = strings.TrimPrefix(part, "options=")
+		} else {
+			return "", "", "", false, fmt.Errorf("invalid volume option or mode: %s", part)
+		}
 	}
 
-	idx := strings.LastIndex(vStrWithoutMode, ":")
-	if idx <= 0 || idx == len(vStrWithoutMode)-1 {
-		return "", "", false, missingDestOrFormatErr(origVStr)
+	if src == "" || dest == "" || !strings.HasPrefix(dest, "/") {
+		return "", "", "", false, missingDestOrFormatErr(vStr)
 	}
 
-	src = vStrWithoutMode[:idx]
-	dest = vStrWithoutMode[idx+1:]
-
-	if src == "" || dest == "" || !strings.HasPrefix(dest, "/") {
-		return "", "", false, missingDestOrFormatErr(origVStr)
+	if options != "" && !strings.HasPrefix(src, "gs://") {
+		return "", "", "", false, fmt.Errorf("options= is currently only supported for GCS fuse volumes (gs://...)")
 	}
 
-	// Ensure scheme colons are not mistaken for delimiters when destination is missing
-	if (strings.HasPrefix(vStrWithoutMode, "gs://") && !strings.HasPrefix(src, "gs://")) ||
-		(strings.HasPrefix(vStrWithoutMode, "filestore://") && !strings.HasPrefix(src, "filestore://")) {
-		return "", "", false, missingDestOrFormatErr(origVStr)
+	if err := validateSrcScheme(src, vStr); err != nil {
+		return "", "", "", false, err
 	}
 
-	return src, dest, readOnly, nil
+	return src, dest, options, readOnly, nil
 }
 
 func extractHost(hostStr string) string {
@@ -332,12 +317,16 @@ func buildVolumeSpec(v MountInfo) map[string]interface{} {
 	}
 	switch v.Type {
 	case "gcsfuse":
+		volumeAttributes := map[string]interface{}{
+			"bucketName": strings.TrimPrefix(v.Source, "gs://"),
+		}
+		if v.Options != "" {
+			volumeAttributes["mountOptions"] = v.Options
+		}
 		spec["csi"] = map[string]interface{}{
-			"driver":   "gcsfuse.csi.storage.gke.io",
-			"readOnly": v.ReadOnly,
-			"volumeAttributes": map[string]interface{}{
-				"bucketName": strings.TrimPrefix(v.Source, "gs://"),
-			},
+			"driver":           "gcsfuse.csi.storage.gke.io",
+			"readOnly":         v.ReadOnly,
+			"volumeAttributes": volumeAttributes,
 		}
 	case "hostPath":
 		spec["hostPath"] = map[string]interface{}{