Enable PVC as DataSource for PVC creation

This enables the ability to specify and existing PVC as a DataSource in
a new PVC Spec (eg "clone" and existing volume).

pkg/api/persistentvolumeclaim/util.go

@@ -28,7 +28,7 @@ func DropDisabledFields(pvcSpec, oldPVCSpec *core.PersistentVolumeClaimSpec) {
 	if !utilfeature.DefaultFeatureGate.Enabled(features.BlockVolume) && !volumeModeInUse(oldPVCSpec) {
 		pvcSpec.VolumeMode = nil
 	}
-	if !utilfeature.DefaultFeatureGate.Enabled(features.VolumeSnapshotDataSource) && !volumeSnapshotDataSourceInUse(oldPVCSpec) {
+	if !dataSourceIsEnabled(pvcSpec) && !dataSourceInUse(oldPVCSpec) {
 		pvcSpec.DataSource = nil
 	}
 }
@@ -43,7 +43,7 @@ func volumeModeInUse(oldPVCSpec *core.PersistentVolumeClaimSpec) bool {
 	return false
 }
 
-func volumeSnapshotDataSourceInUse(oldPVCSpec *core.PersistentVolumeClaimSpec) bool {
+func dataSourceInUse(oldPVCSpec *core.PersistentVolumeClaimSpec) bool {
 	if oldPVCSpec == nil {
 		return false
 	}
@@ -52,3 +52,17 @@ func volumeSnapshotDataSourceInUse(oldPVCSpec *core.PersistentVolumeClaimSpec) b
 	}
 	return false
 }
+
+func dataSourceIsEnabled(pvcSpec *core.PersistentVolumeClaimSpec) bool {
+	if pvcSpec.DataSource != nil {
+		if pvcSpec.DataSource.Kind == "PersistentVolumeClaim" && utilfeature.DefaultFeatureGate.Enabled(features.VolumeDataSource) {
+			return true
+
+		}
+		if pvcSpec.DataSource.Kind == "VolumeSnapshot" && utilfeature.DefaultFeatureGate.Enabled(features.VolumeSnapshotDataSource) {
+			return true
+
+		}
+	}
+	return false
+}

pkg/apis/core/types.go

@@ -413,11 +413,14 @@ type PersistentVolumeClaimSpec struct {
 	// This is a beta feature.
 	// +optional
 	VolumeMode *PersistentVolumeMode
-	// This field requires the VolumeSnapshotDataSource alpha feature gate to be
-	// enabled and currently VolumeSnapshot is the only supported data source.
-	// If the provisioner can support VolumeSnapshot data source, it will create
-	// a new volume and data will be restored to the volume at the same time.
-	// If the provisioner does not support VolumeSnapshot data source, volume will
+	// This field can be used to specify either:
+	// * An existing VolumeSnapshot
+	// * An existing Volume (PVC, Clone operation)
+	// In order to use either of these DataSource types, the appropriate feature gate
+	// must be enabled (VolumeSnapshotDataSource, VolumeDataSource)
+	// If the provisioner can support the specified data source, it will create
+	// a new volume based on the contents of the specified PVC or Snapshot.
+	// If the provisioner does not support the specified data source, the volume will
 	// not be created and the failure will be reported as an event.
 	// In the future, we plan to support more data source types and the behavior
 	// of the provisioner may change.

pkg/apis/core/validation/validation.go

@@ -1537,6 +1537,7 @@ var supportedVolumeModes = sets.NewString(string(core.PersistentVolumeBlock), st
 
 var supportedDataSourceAPIGroupKinds = map[schema.GroupKind]bool{
 	{Group: "snapshot.storage.k8s.io", Kind: "VolumeSnapshot"}: true,
+	{Group: "", Kind: "PersistentVolumeClaim"}:                 true,
 }
 
 func ValidatePersistentVolume(pv *core.PersistentVolume) field.ErrorList {

pkg/apis/core/validation/validation_test.go

@@ -13288,3 +13288,48 @@ func TestValidateWindowsSecurityContextOptions(t *testing.T) {
 		})
 	}
 }
+
+func testDataSourceInSpec(name string, kind string, apiGroup string) *core.PersistentVolumeClaimSpec {
+	scName := "csi-plugin"
+	dataSourceInSpec := core.PersistentVolumeClaimSpec{
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadOnlyMany,
+		},
+		Resources: core.ResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+			},
+		},
+		StorageClassName: &scName,
+		DataSource: &core.TypedLocalObjectReference{
+			APIGroup: &apiGroup,
+			Kind:     kind,
+			Name:     name,
+		},
+	}
+
+	return &dataSourceInSpec
+}
+
+func TestAlphaVolumeDataSource(t *testing.T) {
+	successTestCases := []core.PersistentVolumeClaimSpec{
+		*testDataSourceInSpec("test_snapshot", "VolumeSnapshot", "snapshot.storage.k8s.io"),
+		*testDataSourceInSpec("test_pvc", "PersistentVolumeClaim", ""),
+	}
+	failedTestCases := []core.PersistentVolumeClaimSpec{
+		*testDataSourceInSpec("", "VolumeSnapshot", "snapshot.storage.k8s.io"),
+		*testDataSourceInSpec("test_snapshot", "PersistentVolumeClaim", "snapshot.storage.k8s.io"),
+		*testDataSourceInSpec("test_snapshot", "VolumeSnapshot", "storage.k8s.io"),
+	}
+
+	for _, tc := range successTestCases {
+		if errs := ValidatePersistentVolumeClaimSpec(&tc, field.NewPath("spec")); len(errs) != 0 {
+			t.Errorf("expected success: %v", errs)
+		}
+	}
+	for _, tc := range failedTestCases {
+		if errs := ValidatePersistentVolumeClaimSpec(&tc, field.NewPath("spec")); len(errs) == 0 {
+			t.Errorf("expected failure: %v", errs)
+		}
+	}
+}