feat: customize the nfs version

README.md

@@ -37,3 +37,20 @@ enable_cascading_deletion: true
 ```
 
 **Important**: This feature should be used with caution as it will automatically delete datasets that reference the source dataset. Consider the impact on dependent workloads before enabling this feature.
+
+### NFS Protocol Version
+
+For `NFS` datasets, the controller sets the NFS mount option `nfsvers` when it creates a PersistentVolume. The supported versions are `4.0` and `4.1`; the default is `4.1`.
+
+Set the version in the controller configuration file:
+
+```yaml
+dataset_nfs_version: "4.0"
+```
+
+The `DATASET_NFS_VERSION` environment variable takes precedence over the configuration file. When using the Helm chart, set the corresponding value:
+
+```yaml
+config:
+  dataset_nfs_version: "4.0"
+```

cmd/main.go

@@ -74,6 +74,7 @@ func main() {
 		setupLog.Error(err, "unable to load config")
 		os.Exit(1)
 	}
+	setupLog.Info("configured NFS protocol version", "version", config2.GetDatasetNFSVersion())
 
 	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
 		Scheme:                 scheme,

config/config.go

@@ -1,6 +1,7 @@
 package config
 
 import (
+	"fmt"
 	"os"
 	"strings"
 
@@ -12,9 +13,34 @@ var (
 	config *configuration
 )
 
+const (
+	defaultDatasetNFSVersion = "4.1"
+	datasetNFSVersionEnv     = "DATASET_NFS_VERSION"
+)
+
 type configuration struct {
 	DatasetJobSpecYaml      string `json:"dataset_job_spec_yaml"`
 	EnableCascadingDeletion bool   `json:"enable_cascading_deletion"`
+	DatasetNFSVersion       string `json:"dataset_nfs_version"`
+}
+
+func validateDatasetNFSVersion(version string) error {
+	switch version {
+	case "4.0", "4.1":
+		return nil
+	default:
+		return fmt.Errorf("unsupported dataset NFS version %q, must be one of: 4.0, 4.1", version)
+	}
+}
+
+// GetDatasetNFSVersion returns the NFS protocol version used for newly-created NFS PVs.
+// DATASET_NFS_VERSION is bound during configuration loading and takes precedence over
+// the config file value. The default keeps the existing behavior at NFS 4.1.
+func GetDatasetNFSVersion() string {
+	if config == nil || config.DatasetNFSVersion == "" {
+		return defaultDatasetNFSVersion
+	}
+	return config.DatasetNFSVersion
 }
 
 func GetDatasetJobSpecYaml() string {
@@ -72,15 +98,26 @@ func ParseConfigFromFile(configPath string) error {
 	viper.SetConfigFile(configPath)
 	viper.AutomaticEnv()
 	viper.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
+	viper.SetDefault("dataset_nfs_version", defaultDatasetNFSVersion)
+	if err := viper.BindEnv("dataset_nfs_version", datasetNFSVersionEnv); err != nil {
+		return err
+	}
 	if err := viper.ReadInConfig(); err != nil {
 		return err
 	}
 	err := viper.Unmarshal(cfg, func(c *mapstructure.DecoderConfig) {
 		c.TagName = "json"
 	})
-	config = cfg
 	if err != nil {
 		return err
 	}
+	cfg.DatasetNFSVersion = strings.TrimSpace(viper.GetString("dataset_nfs_version"))
+	if cfg.DatasetNFSVersion == "" {
+		cfg.DatasetNFSVersion = defaultDatasetNFSVersion
+	}
+	if err := validateDatasetNFSVersion(cfg.DatasetNFSVersion); err != nil {
+		return err
+	}
+	config = cfg
 	return nil
 }

config/config_test.go

@@ -0,0 +1,49 @@
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDatasetNFSVersion(t *testing.T) {
+	tests := []struct {
+		name       string
+		configData string
+		envValue   string
+		want       string
+	}{
+		{
+			name:       "default",
+			configData: "enable_cascading_deletion: false",
+			want:       "4.1",
+		},
+		{
+			name:       "config value",
+			configData: "dataset_nfs_version: \"4.0\"",
+			want:       "4.0",
+		},
+		{
+			name:       "environment overrides config",
+			configData: "dataset_nfs_version: \"4.1\"",
+			envValue:   "4.0",
+			want:       "4.0",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Setenv(datasetNFSVersionEnv, tt.envValue)
+			require.NoError(t, ParseConfigFromFileContent(tt.configData))
+			assert.Equal(t, tt.want, GetDatasetNFSVersion())
+		})
+	}
+}
+
+func TestParseConfigRejectsUnsupportedDatasetNFSVersion(t *testing.T) {
+	t.Setenv(datasetNFSVersionEnv, "4.2")
+	err := ParseConfigFromFileContent("dataset_nfs_version: \"4.1\"")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "unsupported dataset NFS version")
+}

config/sample-config.yaml

@@ -1,13 +1,17 @@
 # Configuration for the Dataset Controller
-# 
+#
 # To enable cascading deletion of reference datasets when the source dataset is deleted,
 # set enable_cascading_deletion to true. When enabled, if a dataset is deleted and other
 # datasets reference it (via DatasetTypeReference), those referencing datasets will also
 # be automatically deleted along with their associated retained PVs.
-# 
+#
 # Default: false (disabled for safety)
 enable_cascading_deletion: false
 
+# NFS protocol version used when creating new NFS PVs (default: 4.1).
+# DATASET_NFS_VERSION takes precedence over this value.
+# dataset_nfs_version: "4.1"
+
 # Custom job specification for dataset loading jobs (optional)
 # If not specified, a default job specification will be used
 # dataset_job_spec_yaml: |

internal/controller/dataset/dataset_controller.go

@@ -58,6 +58,23 @@ const (
 	condTypeJobStatus = "JobStatus"
 	condTypeJob       = "Job"
 	condTypeConfigMap = "ConfigMap"
+
+	nfsPersistentVolumeTemplate = `
+apiVersion: v1
+kind: PersistentVolume
+metadata:
+  annotations:
+    pv.kubernetes.io/provisioned-by: nfs.csi.k8s.io
+spec:
+  capacity:
+    storage: 100Ti
+  accessModes:
+    - ReadWriteMany
+  persistentVolumeReclaimPolicy: Retain
+  storageClassName: nfs-csi
+  csi:
+    driver: nfs.csi.k8s.io
+`
 )
 
 // DatasetReconciler reconciles a Dataset object
@@ -334,27 +351,11 @@ func (r *DatasetReconciler) reconcilePVC(ctx context.Context, ds *datasetv1alpha
 
 		// NFS 需要先创建一个 PV
 		var pvTemp corev1.PersistentVolume
-		err := yaml.Unmarshal([]byte(`
-apiVersion: v1
-kind: PersistentVolume
-metadata:
-  annotations:
-    pv.kubernetes.io/provisioned-by: nfs.csi.k8s.io
-spec:
-  capacity:
-    storage: 100Ti
-  accessModes:
-    - ReadWriteMany
-  persistentVolumeReclaimPolicy: Retain
-  storageClassName: nfs-csi
-  mountOptions:
-    - nfsvers=4.1
-  csi:
-    driver: nfs.csi.k8s.io
-`), &pvTemp)
+		err := yaml.Unmarshal([]byte(nfsPersistentVolumeTemplate), &pvTemp)
 		if err != nil {
 			return err
 		}
+		pvTemp.Spec.MountOptions = []string{fmt.Sprintf("nfsvers=%s", config.GetDatasetNFSVersion())}
 		u, err := url.Parse(ds.Spec.Source.URI)
 		if err != nil {
 			return err

internal/controller/dataset/dataset_controller_test.go

@@ -439,3 +439,98 @@ func TestDatasetReconciler_reconcileClaimPVC(t *testing.T) {
 		})
 	}
 }
+
+func TestDatasetReconciler_reconcilePVCNFSVersion(t *testing.T) {
+	tests := []struct {
+		name     string
+		envValue string
+		want     string
+	}{
+		{name: "nfs 4.0", envValue: "4.0", want: "nfsvers=4.0"},
+		{name: "nfs 4.1", envValue: "4.1", want: "nfsvers=4.1"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Setenv("DATASET_NFS_VERSION", tt.envValue)
+			require.NoError(t, config.ParseConfigFromFileContent(""))
+
+			scheme := runtime.NewScheme()
+			require.NoError(t, datasetv1alpha1.AddToScheme(scheme))
+			require.NoError(t, corev1.AddToScheme(scheme))
+
+			ds := &datasetv1alpha1.Dataset{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "netapp-dataset",
+					Namespace: "default",
+				},
+				Spec: datasetv1alpha1.DatasetSpec{
+					Source: datasetv1alpha1.DatasetSource{
+						Type: datasetv1alpha1.DatasetTypeNFS,
+						URI:  "nfs://10.0.0.1/export/path",
+					},
+				},
+			}
+			fakeClient := fake.NewClientBuilder().WithScheme(scheme).Build()
+			reconciler := &DatasetReconciler{Client: fakeClient, Scheme: scheme}
+
+			require.NoError(t, reconciler.reconcilePVC(context.Background(), ds))
+
+			pv := &corev1.PersistentVolume{}
+			require.NoError(t, fakeClient.Get(context.Background(), client.ObjectKey{
+				Name: "dataset-default-pvc-netapp-dataset",
+			}, pv))
+			assert.Equal(t, []string{tt.want}, pv.Spec.MountOptions)
+		})
+	}
+}
+
+func TestDatasetReconciler_reconcilePVCNFSVersionDoesNotUpdateExistingPV(t *testing.T) {
+	t.Setenv("DATASET_NFS_VERSION", "4.0")
+	require.NoError(t, config.ParseConfigFromFileContent(""))
+
+	scheme := runtime.NewScheme()
+	require.NoError(t, datasetv1alpha1.AddToScheme(scheme))
+	require.NoError(t, corev1.AddToScheme(scheme))
+
+	ds := &datasetv1alpha1.Dataset{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "netapp-dataset",
+			Namespace: "default",
+		},
+		Spec: datasetv1alpha1.DatasetSpec{
+			Source: datasetv1alpha1.DatasetSource{
+				Type: datasetv1alpha1.DatasetTypeNFS,
+				URI:  "nfs://10.0.0.1/export/path",
+			},
+		},
+	}
+	pv := &corev1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "dataset-default-pvc-netapp-dataset",
+			Labels: map[string]string{
+				constants.DatasetNameLabel: ds.Name,
+			},
+		},
+		Spec: corev1.PersistentVolumeSpec{
+			MountOptions: []string{"nfsvers=4.1"},
+		},
+	}
+	pvc := &corev1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      ds.Name,
+			Namespace: ds.Namespace,
+			Labels: map[string]string{
+				constants.DatasetNameLabel: ds.Name,
+			},
+		},
+	}
+	fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(pv, pvc).Build()
+	reconciler := &DatasetReconciler{Client: fakeClient, Scheme: scheme}
+
+	require.NoError(t, reconciler.reconcilePVC(context.Background(), ds))
+
+	storedPV := &corev1.PersistentVolume{}
+	require.NoError(t, fakeClient.Get(context.Background(), client.ObjectKey{Name: pv.Name}, storedPV))
+	assert.Equal(t, []string{"nfsvers=4.1"}, storedPV.Spec.MountOptions)
+}

manifests/dataset/templates/deployment.yaml

@@ -35,6 +35,9 @@ spec:
             {{- toYaml .Values.securityContext | nindent 12 }}
           image: {{ template "dataset.controller.image" . }}
           imagePullPolicy: {{ .Values.global.imagePullPolicy }}
+          env:
+            - name: DATASET_NFS_VERSION
+              value: {{ .Values.config.dataset_nfs_version | default "4.1" | quote }}
           readinessProbe:
             httpGet:
               path: /readyz

manifests/dataset/values.yaml

@@ -9,6 +9,9 @@ global:
 
 config:
   dataset_job_spec: {}
+  # NFS protocol version used when creating new NFS PVs. This
+  # value will be passed to the controller through DATASET_NFS_VERSION.
+  dataset_nfs_version: "4.1"
   # Enable cascading deletion of reference datasets when source dataset is deleted
   # Default: false (disabled for safety)
   enable_cascading_deletion: false