feat: add S3 TLS options and per-dataset NFS versions

README.md

@@ -42,6 +42,12 @@ enable_cascading_deletion: true
 
 For `NFS` datasets, the controller sets the NFS mount option `nfsvers` when it creates a PersistentVolume. The supported versions are `3`, `4.0`, `4.1`, and `4.2`; the default is `4.1`.
 
+An explicit port in the NFS URI (for example, `nfs://server:20490/export`)
+sets only the NFS service port. NFSv3 still requires rpcbind, mountd, and
+other auxiliary services to be correctly configured and reachable. The URI
+does not support specifying their ports separately. When only a single
+custom port is exposed, NFSv4 (4.0, 4.1, or 4.2) is recommended.
+
 Set the version in the controller configuration file:
 
 ```yaml
@@ -54,3 +60,16 @@ The `DATASET_NFS_VERSION` environment variable takes precedence over the configu
 config:
   dataset_nfs_version: "4.0"
 ```
+
+Set a version for an individual NFS dataset to override the controller default:
+
+```yaml
+spec:
+  source:
+    type: NFS
+    uri: nfs://server/export
+    options:
+      nfsVersion: "3"
+```
+
+Omit `nfsVersion` to use the controller default. An explicit version cannot be added, changed, or removed after creation. Changing the controller default does not update existing PVs. External PVCs (`volumeClaimRef`) use their existing PV options and cannot specify `nfsVersion`. Reference datasets inherit their source PV mount options.

api/dataset/v1alpha1/dataset_types.go

@@ -61,6 +61,8 @@ const (
 	AccessModeReadWrite AccessMode = "ReadWrite"
 )
 
+// +kubebuilder:validation:XValidation:rule="!has(self.options) || !('nfsVersion' in self.options) || (self.type == 'NFS' && self.options['nfsVersion'] in ['3', '4.0', '4.1', '4.2'])",message="nfsVersion must be 3, 4.0, 4.1 or 4.2 and is only supported for NFS datasets"
+// +kubebuilder:validation:XValidation:rule="(has(self.options) && ('nfsVersion' in self.options)) ? (has(oldSelf.options) && ('nfsVersion' in oldSelf.options) && self.options['nfsVersion'] == oldSelf.options['nfsVersion']) : !(has(oldSelf.options) && ('nfsVersion' in oldSelf.options))",message="nfsVersion cannot be added, changed or removed after creation"
 type DatasetSource struct {
 	// +kubebuilder:validation:Enum=GIT;S3;HTTP;PVC;NFS;CONDA;REFERENCE;HUGGING_FACE;MODEL_SCOPE;DATABASE;HADOOP;MANUAL
 	// +kubebuilder:validation:XValidation:rule="self == oldSelf",message="Value is immutable"
@@ -87,10 +89,13 @@ type DatasetSource struct {
 	// options is a map of key-value pairs that can be used to specify additional options for the dataset source, e.g. {"branch": "master"}
 	// supported keys for each type of dataset source are:
 	// - GIT: branch, commit, depth, submodules
-	// - S3: region, endpoint, provider, syncMode
+	// - S3: region, endpoint, provider, syncMode, tlsMode, tlsCASecretName
+	//   tlsMode: system (default), customCA, insecureSkipVerify.
+	//   customCA requires a same-namespace Secret reference (tlsCASecretName), key ca.crt.
+	//   CA PEM and the internal loader path tlsCAFile must not be stored in options.
 	// - HTTP: any key-value pair will be passed to the underlying http client as http headers
 	// - PVC:
-	// - NFS:
+	// - NFS: nfsVersion: 3, 4.0, 4.1 or 4.2. Omit to use the controller default. Immutable after creation.
 	// - CONDA: requirements.txt, environment.yaml
 	// - REFERENCE:
 	// - HUGGING_FACE: repo, repoType, endpoint, include, exclude, revision
@@ -128,6 +133,7 @@ type MountOptions struct {
 // DatasetSpec defines the desired state of Dataset
 // +kubebuilder:validation:XValidation:rule="!has(self.shareAccess) || size(self.shareAccess.rules) > 0",message="shareAccess.rules must contain at least one rule when shareAccess is configured"
 // +kubebuilder:validation:XValidation:rule="has(self.volumeClaimRef) || self.source.type != 'PVC' || (self.source.uri.matches('^pvc://[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?([.][a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?)*(/[^?#]*)?$') && size(self.source.uri.split('/')[2]) <= 253)",message="PVC dataset source URI must be pvc://<valid-pvc-name>[/<path>] when volumeClaimRef is not set"
+// +kubebuilder:validation:XValidation:rule="!has(self.volumeClaimRef) || !has(self.source.options) || !('nfsVersion' in self.source.options)",message="nfsVersion cannot be set when volumeClaimRef is used"
 type DatasetSpec struct {
 	// Share indicates whether the model is shareable with others.
 	// When set to true, the model can be shared according to the specified selector.

api/dataset/v1alpha1/dataset_types_envtest_test.go

@@ -2,6 +2,7 @@ package v1alpha1_test
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"path/filepath"
 	goruntime "runtime"
@@ -117,4 +118,51 @@ func TestShareAccessCRDValidation(t *testing.T) {
 		immutable.Spec.ShareAccess.Rules[0].AccessMode = datasetv1alpha1.AccessModeReadWrite
 		require.NoError(t, apiClient.Update(ctx, immutable))
 	})
+
+	t.Run("NFS version schema and transitions", func(t *testing.T) {
+		for index, options := range []map[string]string{nil, {}, {"other": "value"}, {"nfsVersion": "3"}, {"nfsVersion": "4.0"}, {"nfsVersion": "4.1"}, {"nfsVersion": "4.2"}} {
+			ds := newDataset(fmt.Sprintf("nfs-version-%d", index))
+			ds.Spec.Source = datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeNFS, URI: "nfs://server/export", Options: options}
+			require.NoError(t, apiClient.Create(ctx, ds))
+			ds.Annotations = map[string]string{"description": "updated"}
+			require.NoError(t, apiClient.Update(ctx, ds))
+			unchanged := ds.DeepCopy()
+			if unchanged.Spec.Source.Options == nil {
+				unchanged.Spec.Source.Options = map[string]string{}
+			}
+			unchanged.Spec.Source.Options["other"] = "updated"
+			require.NoError(t, apiClient.Update(ctx, unchanged))
+			delete(unchanged.Spec.Source.Options, "other")
+			require.NoError(t, apiClient.Update(ctx, unchanged))
+			changed := unchanged.DeepCopy()
+			if changed.Spec.Source.Options == nil {
+				changed.Spec.Source.Options = map[string]string{}
+			}
+			changed.Spec.Source.Options["nfsVersion"] = "3"
+			if options["nfsVersion"] == "3" {
+				changed.Spec.Source.Options["nfsVersion"] = "4.1"
+			}
+			require.Error(t, apiClient.Update(ctx, changed))
+			if _, present := options["nfsVersion"]; present {
+				removed := unchanged.DeepCopy()
+				delete(removed.Spec.Source.Options, "nfsVersion")
+				require.Error(t, apiClient.Update(ctx, removed))
+				removed.Spec.Source.Options = nil
+				require.Error(t, apiClient.Update(ctx, removed))
+			}
+		}
+		for index, version := range []string{"", "4", "5", "auto"} {
+			ds := newDataset(fmt.Sprintf("nfs-invalid-%d", index))
+			ds.Spec.Source = datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeNFS, URI: "nfs://server/export", Options: map[string]string{"nfsVersion": version}}
+			require.Error(t, apiClient.Create(ctx, ds))
+		}
+		ds := newDataset("nfs-version-wrong-type")
+		ds.Spec.Source.Options = map[string]string{"nfsVersion": "3"}
+		require.Error(t, apiClient.Create(ctx, ds))
+		ds = newDataset("nfs-version-external")
+		ds.Spec.Source = datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeNFS, URI: "nfs://server/export", Options: map[string]string{"nfsVersion": "3"}}
+		ds.Spec.VolumeClaimRef = &datasetv1alpha1.VolumeClaimRef{Name: "external"}
+		require.Error(t, apiClient.Create(ctx, ds))
+	})
+
 }

config/config.go

@@ -1,7 +1,7 @@
 package config
 
 import (
-	"fmt"
+	"github.com/BaizeAI/dataset/pkg/nfsconfig"
 	"os"
 	"strings"
 
@@ -14,8 +14,8 @@ var (
 )
 
 const (
-	defaultDatasetNFSVersion = "4.1"
-	datasetNFSVersionEnv     = "DATASET_NFS_VERSION"
+	defaultDatasetNFSVersion = nfsconfig.DefaultVersion
+	datasetNFSVersionEnv     = nfsconfig.EnvironmentKey
 )
 
 type configuration struct {
@@ -25,12 +25,7 @@ type configuration struct {
 }
 
 func validateDatasetNFSVersion(version string) error {
-	switch version {
-	case "3", "4.0", "4.1", "4.2":
-		return nil
-	default:
-		return fmt.Errorf("unsupported dataset NFS version %q, must be one of: 3, 4.0, 4.1, 4.2", version)
-	}
+	return nfsconfig.Validate(version)
 }
 
 // GetDatasetNFSVersion returns the NFS protocol version used for newly-created NFS PVs.

config/crd/bases/dataset.baizeai.io_datasets.yaml

@@ -297,10 +297,13 @@ spec:
                       options is a map of key-value pairs that can be used to specify additional options for the dataset source, e.g. {"branch": "master"}
                       supported keys for each type of dataset source are:
                       - GIT: branch, commit, depth, submodules
-                      - S3: region, endpoint, provider, syncMode
+                      - S3: region, endpoint, provider, syncMode, tlsMode, tlsCASecretName
+                        tlsMode: system (default), customCA, insecureSkipVerify.
+                        customCA requires a same-namespace Secret reference (tlsCASecretName), key ca.crt.
+                        CA PEM and the internal loader path tlsCAFile must not be stored in options.
                       - HTTP: any key-value pair will be passed to the underlying http client as http headers
                       - PVC:
-                      - NFS:
+                      - NFS: nfsVersion: 3, 4.0, 4.1 or 4.2. Omit to use the controller default. Immutable after creation.
                       - CONDA: requirements.txt, environment.yaml
                       - REFERENCE:
                       - HUGGING_FACE: repo, repoType, endpoint, include, exclude, revision
@@ -353,6 +356,17 @@ spec:
                 - type
                 - uri
                 type: object
+                x-kubernetes-validations:
+                - message: nfsVersion must be 3, 4.0, 4.1 or 4.2 and is only supported
+                    for NFS datasets
+                  rule: '!has(self.options) || !(''nfsVersion'' in self.options) ||
+                    (self.type == ''NFS'' && self.options[''nfsVersion''] in [''3'',
+                    ''4.0'', ''4.1'', ''4.2''])'
+                - message: nfsVersion cannot be added, changed or removed after creation
+                  rule: '(has(self.options) && (''nfsVersion'' in self.options)) ?
+                    (has(oldSelf.options) && (''nfsVersion'' in oldSelf.options) &&
+                    self.options[''nfsVersion''] == oldSelf.options[''nfsVersion''])
+                    : !(has(oldSelf.options) && (''nfsVersion'' in oldSelf.options))'
               volumeClaimRef:
                 description: volumeClaimRef is the reference to an existing PVC.
                 properties:
@@ -796,6 +810,9 @@ spec:
                 when volumeClaimRef is not set
               rule: has(self.volumeClaimRef) || self.source.type != 'PVC' || (self.source.uri.matches('^pvc://[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?([.][a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?)*(/[^?#]*)?$')
                 && size(self.source.uri.split('/')[2]) <= 253)
+            - message: nfsVersion cannot be set when volumeClaimRef is used
+              rule: '!has(self.volumeClaimRef) || !has(self.source.options) || !(''nfsVersion''
+                in self.source.options)'
           status:
             description: DatasetStatus defines the observed state of Dataset
             properties:

config/sample-config.yaml

@@ -8,7 +8,7 @@
 # Default: false (disabled for safety)
 enable_cascading_deletion: false
 
-# NFS protocol version used when creating new NFS PVs (supported: 3, 4.0, 4.1, 4.2; default: 4.1).
+# Default NFS version for new PVs when source.options.nfsVersion is omitted (supported: 3, 4.0, 4.1, 4.2; default: 4.1).
 # DATASET_NFS_VERSION takes precedence over this value.
 # dataset_nfs_version: "4.1"
 

internal/controller/dataset/dataset_controller.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/BaizeAI/dataset/pkg/kubeutils"
 	"github.com/BaizeAI/dataset/pkg/mountpolicy"
+	"github.com/BaizeAI/dataset/pkg/nfsconfig"
 
 	"github.com/samber/lo"
 	batchv1 "k8s.io/api/batch/v1"
@@ -458,11 +459,22 @@ func (r *DatasetReconciler) reconcilePVC(ctx context.Context, ds *datasetv1alpha
 		if err != nil {
 			return err
 		}
-		pvTemp.Spec.MountOptions = []string{fmt.Sprintf("nfsvers=%s", config.GetDatasetNFSVersion())}
+		version, err := nfsconfig.Resolve(ds.Spec.Source.Options, config.GetDatasetNFSVersion())
+		if err != nil {
+			return err
+		}
+		pvTemp.Spec.MountOptions = []string{fmt.Sprintf("nfsvers=%s", version)}
 		u, err := url.Parse(ds.Spec.Source.URI)
 		if err != nil {
 			return err
 		}
+		server, port, err := nfsServerAndPort(u)
+		if err != nil {
+			return err
+		}
+		if port != "" {
+			pvTemp.Spec.MountOptions = append(pvTemp.Spec.MountOptions, "port="+port)
+		}
 		forceStorageClass = pvTemp.Spec.StorageClassName
 
 		pv := &corev1.PersistentVolume{}
@@ -488,7 +500,7 @@ func (r *DatasetReconciler) reconcilePVC(ctx context.Context, ds *datasetv1alpha
 			if pvTemp.Spec.CSI.VolumeAttributes == nil {
 				pvTemp.Spec.CSI.VolumeAttributes = make(map[string]string)
 			}
-			pvTemp.Spec.CSI.VolumeAttributes["server"] = u.Host
+			pvTemp.Spec.CSI.VolumeAttributes["server"] = server
 			pvTemp.Spec.CSI.VolumeAttributes["share"] = "/"
 			pvTemp.Spec.CSI.VolumeAttributes["subdir"] = u.Path
 			pvTemp.Spec.CSI.VolumeAttributes["onDelete"] = "retain"
@@ -500,6 +512,8 @@ func (r *DatasetReconciler) reconcilePVC(ctx context.Context, ds *datasetv1alpha
 			if pvTemp.Spec.CSI.VolumeAttributes["mountPermissions"] == "" {
 				pvTemp.Spec.CSI.VolumeAttributes["mountPermissions"] = ds.Spec.MountOptions.Mode
 			}
+			// Preserve the existing volume identity, including the URI port.
+			// The mount server and NFS port are supplied separately above.
 			pvTemp.Spec.CSI.VolumeHandle = fmt.Sprintf("%s#%s#%s#", u.Host, u.Path, pvName)
 
 			if err := r.Create(ctx, &pvTemp); err != nil {
@@ -747,6 +761,11 @@ func (r *DatasetReconciler) reconcileJob(ctx context.Context, ds *datasetv1alpha
 		}
 
 		podSpec := &jobSpec.Template.Spec
+		if ds.Spec.Source.Type == datasetv1alpha1.DatasetTypeS3 {
+			if err := configureS3TLS(options, podSpec, container); err != nil {
+				return err
+			}
+		}
 
 		// conda 类型需要将 ConfigMap mount 到容器
 		condaKeyItems := make([]corev1.KeyToPath, 0, 2)
@@ -1166,6 +1185,12 @@ func (r *DatasetReconciler) getSourceDataset(ctx context.Context, ds *datasetv1a
 }
 
 func (r *DatasetReconciler) validate(ctx context.Context, ds *datasetv1alpha1.Dataset) error {
+	if err := nfsconfig.ValidateOptions(string(ds.Spec.Source.Type), ds.Spec.Source.Options); err != nil {
+		return err
+	}
+	if _, present := ds.Spec.Source.Options[nfsconfig.VersionKey]; present && ds.Spec.VolumeClaimRef != nil {
+		return fmt.Errorf("nfsVersion cannot be set when volumeClaimRef is used")
+	}
 	if ds.Spec.Source.Type == datasetv1alpha1.DatasetTypeManual && ds.Spec.Source.URI != "manual://" {
 		return fmt.Errorf("MANUAL dataset source URI must be manual://")
 	}

internal/controller/dataset/nfs_uri.go

@@ -0,0 +1,38 @@
+package dataset
+
+import (
+	"fmt"
+	"net/netip"
+	"net/url"
+	"strconv"
+	"strings"
+)
+
+// nfsServerAndPort separates the mount server from the NFS service port.
+// Keep the original host spelling (including IPv6 brackets) for compatibility.
+func nfsServerAndPort(u *url.URL) (string, string, error) {
+	server := u.Host
+	if u.Hostname() == "" {
+		return "", "", fmt.Errorf("NFS URI must include a server")
+	}
+	// url.URL.Port treats the final numeric segment of an unbracketed IPv6
+	// address as a port. Preserve legacy address-only URIs before splitting.
+	if addr, err := netip.ParseAddr(server); err == nil && addr.Is6() {
+		return server, "", nil
+	}
+	if strings.Count(server, ":") > 1 && !strings.HasPrefix(server, "[") {
+		return "", "", fmt.Errorf("invalid NFS server %q: IPv6 addresses with a port must be bracketed", server)
+	}
+	port := u.Port()
+	if port == "" {
+		if strings.HasSuffix(server, ":") {
+			return "", "", fmt.Errorf("invalid NFS port: port must not be empty")
+		}
+		return server, "", nil
+	}
+	number, err := strconv.ParseUint(port, 10, 16)
+	if err != nil || number == 0 {
+		return "", "", fmt.Errorf("invalid NFS port %q: must be between 1 and 65535", port)
+	}
+	return strings.TrimSuffix(server, ":"+port), strconv.FormatUint(number, 10), nil
+}