Add volumesnapshot content deletion policy

This PR adds DeletionPolicy which describes a policy for end-of-life maintenance of volume snapshot contents.
There are two types of policies, delete and retain. By default, the
policy is delete. In this case, when volume snaphsot is deleted, the
volume snapshot content for it should be deleted, as well as the
physical snapshot. If the policy is set the retain, the volume snapshot
content will remain even if the volume snapshot it binds to is deleted.

pkg/apis/volumesnapshot/v1alpha1/types.go

@@ -132,6 +132,11 @@ type VolumeSnapshotClass struct {
 	// to the snapshotter.
 	// +optional
 	Parameters map[string]string `json:"parameters,omitempty" protobuf:"bytes,3,rep,name=parameters"`
+
+	// Optional: what happens to a snapshot content when released from its snapshot.
+	// The default policy is Delete if not specified.
+	// +optional
+	DeletionPolicy *DeletionPolicy `json:"deletionPolicy,omitempty" protobuf:"bytes,4,opt,name=deletionPolicy"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
@@ -195,6 +200,11 @@ type VolumeSnapshotContentSpec struct {
 	// be used if it is available.
 	// +optional
 	VolumeSnapshotClassName *string `json:"snapshotClassName" protobuf:"bytes,4,opt,name=snapshotClassName"`
+
+	// Optional: what happens to a snapshot content when released from its snapshot. It will be set to Delete by default
+	// if not specified
+	// +optional
+	DeletionPolicy *DeletionPolicy `json:"deletionPolicy" protobuf:"bytes,5,opt,name=deletionPolicy"`
 }
 
 // VolumeSnapshotSource represents the actual location and type of the snapshot. Only one of its members may be specified.
@@ -232,3 +242,15 @@ type CSIVolumeSnapshotSource struct {
 	// +optional
 	RestoreSize *int64 `json:"restoreSize,omitempty" protobuf:"bytes,4,opt,name=restoreSize"`
 }
+
+// DeletionPolicy describes a policy for end-of-life maintenance of volume snapshot contents
+type DeletionPolicy string
+
+const (
+	// VolumeSnapshotContentDelete means the snapshot content will be deleted from Kubernetes on release from its volume snapshot.
+	VolumeSnapshotContentDelete DeletionPolicy = "Delete"
+
+	// VolumeSnapshotContentRetain means the snapshot will be left in its current state on release from its volume snapshot.
+	// The default policy is Retain if not specified.
+	VolumeSnapshotContentRetain DeletionPolicy = "Retain"
+)

pkg/controller/csi_handler.go

@@ -77,7 +77,7 @@ func (handler *csiHandler) DeleteSnapshot(content *crdv1.VolumeSnapshotContent,
 
 	err := handler.csiConnection.DeleteSnapshot(ctx, content.Spec.CSI.SnapshotHandle, snapshotterCredentials)
 	if err != nil {
-		return fmt.Errorf("failed to delete snapshot data %s: %q", content.Name, err)
+		return fmt.Errorf("failed to delete snapshot content %s: %q", content.Name, err)
 	}
 
 	return nil
@@ -92,7 +92,7 @@ func (handler *csiHandler) GetSnapshotStatus(content *crdv1.VolumeSnapshotConten
 
 	csiSnapshotStatus, timestamp, size, err := handler.csiConnection.GetSnapshotStatus(ctx, content.Spec.CSI.SnapshotHandle)
 	if err != nil {
-		return false, 0, 0, fmt.Errorf("failed to list snapshot data %s: %q", content.Name, err)
+		return false, 0, 0, fmt.Errorf("failed to list snapshot content %s: %q", content.Name, err)
 	}
 	return csiSnapshotStatus, timestamp, size, nil
 

pkg/controller/framework_test.go

@@ -744,7 +744,7 @@ func newTestController(kubeClient kubernetes.Interface, clientset clientset.Inte
 }
 
 // newContent returns a new content with given attributes
-func newContent(name, className, snapshotHandle, volumeUID, volumeName, boundToSnapshotUID, boundToSnapshotName string, size *int64, creationTime *int64) *crdv1.VolumeSnapshotContent {
+func newContent(name, className, snapshotHandle, volumeUID, volumeName, boundToSnapshotUID, boundToSnapshotName string, deletionPolicy *crdv1.DeletionPolicy, size *int64, creationTime *int64) *crdv1.VolumeSnapshotContent {
 	content := crdv1.VolumeSnapshotContent{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:            name,
@@ -766,6 +766,7 @@ func newContent(name, className, snapshotHandle, volumeUID, volumeName, boundToS
 				UID:        types.UID(volumeUID),
 				Name:       volumeName,
 			},
+			DeletionPolicy: deletionPolicy,
 		},
 	}
 	if boundToSnapshotName != "" {
@@ -781,14 +782,14 @@ func newContent(name, className, snapshotHandle, volumeUID, volumeName, boundToS
 	return &content
 }
 
-func newContentArray(name, className, snapshotHandle, volumeUID, volumeName, boundToSnapshotUID, boundToSnapshotName string, size *int64, creationTime *int64) []*crdv1.VolumeSnapshotContent {
+func newContentArray(name, className, snapshotHandle, volumeUID, volumeName, boundToSnapshotUID, boundToSnapshotName string, deletionPolicy *crdv1.DeletionPolicy, size *int64, creationTime *int64) []*crdv1.VolumeSnapshotContent {
 	return []*crdv1.VolumeSnapshotContent{
-		newContent(name, className, snapshotHandle, volumeUID, volumeName, boundToSnapshotUID, boundToSnapshotName, size, creationTime),
+		newContent(name, className, snapshotHandle, volumeUID, volumeName, boundToSnapshotUID, boundToSnapshotName, deletionPolicy, size, creationTime),
 	}
 }
 
-func newContentWithUnmatchDriverArray(name, className, snapshotHandle, volumeUID, volumeName, boundToSnapshotUID, boundToSnapshotName string, size *int64, creationTime *int64) []*crdv1.VolumeSnapshotContent {
-	content := newContent(name, className, snapshotHandle, volumeUID, volumeName, boundToSnapshotUID, boundToSnapshotName, size, creationTime)
+func newContentWithUnmatchDriverArray(name, className, snapshotHandle, volumeUID, volumeName, boundToSnapshotUID, boundToSnapshotName string, deletionPolicy *crdv1.DeletionPolicy, size *int64, creationTime *int64) []*crdv1.VolumeSnapshotContent {
+	content := newContent(name, className, snapshotHandle, volumeUID, volumeName, boundToSnapshotUID, boundToSnapshotName, deletionPolicy, size, creationTime)
 	content.Spec.VolumeSnapshotSource.CSI.Driver = "fake"
 	return []*crdv1.VolumeSnapshotContent{
 		content,

pkg/controller/snapshot_controller.go

@@ -125,9 +125,24 @@ func (ctrl *csiSnapshotController) syncContent(content *crdv1.VolumeSnapshotCont
 		snapshot = nil
 	}
 	if snapshot == nil {
-		ctrl.deleteSnapshotContent(content)
+		if content.Spec.DeletionPolicy != nil {
+			switch *content.Spec.DeletionPolicy {
+			case crdv1.VolumeSnapshotContentRetain:
+				glog.V(4).Infof("VolumeSnapshotContent[%s]: policy is Retain, nothing to do", content.Name)
+
+			case crdv1.VolumeSnapshotContentDelete:
+				glog.V(4).Infof("VolumeSnapshotContent[%s]: policy is Delete", content.Name)
+				ctrl.deleteSnapshotContent(content)
+			default:
+				// Unknown VolumeSnapshotDeletionolicy
+				ctrl.eventRecorder.Event(content, v1.EventTypeWarning, "SnapshotUnknownDeletionPolicy", "Volume Snapshot Content has unrecognized deletion policy")
+			}
+			return nil
+		}
+		// By default, we use Retain policy if it is not set by users
+		glog.V(4).Infof("VolumeSnapshotContent[%s]: by default the policy is Retain", content.Name)
+
 	}
-
 	return nil
 }
 
@@ -453,19 +468,18 @@ func (ctrl *csiSnapshotController) createSnapshotOperation(snapshot *crdv1.Volum
 
 	className := snapshot.Spec.VolumeSnapshotClassName
 	glog.V(5).Infof("createSnapshotOperation [%s]: VolumeSnapshotClassName [%s]", snapshot.Name, *className)
-	var class *crdv1.VolumeSnapshotClass
-	var err error
-	if className != nil {
-		class, err = ctrl.GetSnapshotClass(*className)
-		if err != nil {
-			glog.Errorf("createSnapshotOperation failed to getClassFromVolumeSnapshot %s", err)
-			return nil, err
-		}
-	} else {
+
+	if className == nil {
 		glog.Errorf("failed to take snapshot %s without a snapshot class", snapshot.Name)
 		return nil, fmt.Errorf("failed to take snapshot %s without a snapshot class", snapshot.Name)
 	}
 
+	class, err := ctrl.GetSnapshotClass(*className)
+	if err != nil {
+		glog.Errorf("createSnapshotOperation failed to getClassFromVolumeSnapshot %s", err)
+		return nil, err
+	}
+
 	volume, err := ctrl.getVolumeFromVolumeSnapshot(snapshot)
 	if err != nil {
 		glog.Errorf("createSnapshotOperation failed to get PersistentVolume object [%s]: Error: [%#v]", snapshot.Name, err)
@@ -515,6 +529,10 @@ func (ctrl *csiSnapshotController) createSnapshotOperation(snapshot *crdv1.Volum
 		return nil, err
 	}
 
+	if class.DeletionPolicy == nil {
+		class.DeletionPolicy = new(crdv1.DeletionPolicy)
+		*class.DeletionPolicy = crdv1.VolumeSnapshotContentDelete
+	}
 	snapshotContent := &crdv1.VolumeSnapshotContent{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: contentName,
@@ -531,8 +549,10 @@ func (ctrl *csiSnapshotController) createSnapshotOperation(snapshot *crdv1.Volum
 				},
 			},
 			VolumeSnapshotClassName: &(class.Name),
+			DeletionPolicy: class.DeletionPolicy,
 		},
 	}
+	glog.V(3).Infof("volume snapshot content %%v", snapshotContent)
 	// Try to create the VolumeSnapshotContent object several times
 	for i := 0; i < ctrl.createSnapshotContentRetryCount; i++ {
 		glog.V(5).Infof("createSnapshot [%s]: trying to save volume snapshot content %s", snapshotKey(snapshot), snapshotContent.Name)
@@ -542,7 +562,7 @@ func (ctrl *csiSnapshotController) createSnapshotOperation(snapshot *crdv1.Volum
 				glog.V(3).Infof("volume snapshot content %q for snapshot %q already exists, reusing", snapshotContent.Name, snapshotKey(snapshot))
 				err = nil
 			} else {
-				glog.V(3).Infof("volume snapshot content %q for snapshot %q saved", snapshotContent.Name, snapshotKey(snapshot))
+				glog.V(3).Infof("volume snapshot content %q for snapshot %q saved, %v", snapshotContent.Name, snapshotKey(snapshot), snapshotContent)
 			}
 			break
 		}

pkg/controller/snapshot_controller_test.go

@@ -23,7 +23,7 @@ import (
 )
 
 func storeVersion(t *testing.T, prefix string, c cache.Store, version string, expectedReturn bool) {
-	content := newContent("contentName", classEmpty, "sid1-1", "vuid1-1", "volume1-1", "snapuid1-1", "snap1-1", nil, nil)
+	content := newContent("contentName", classEmpty, "sid1-1", "vuid1-1", "volume1-1", "snapuid1-1", "snap1-1", nil, nil, nil)
 	content.ResourceVersion = version
 	ret, err := storeObjectUpdate(c, content, "content")
 	if err != nil {
@@ -82,7 +82,7 @@ func TestControllerCacheParsingError(t *testing.T) {
 	// There must be something in the cache to compare with
 	storeVersion(t, "Step1", c, "1", true)
 
-	content := newContent("contentName", classEmpty, "sid1-1", "vuid1-1", "volume1-1", "snapuid1-1", "snap1-1", nil, nil)
+	content := newContent("contentName", classEmpty, "sid1-1", "vuid1-1", "volume1-1", "snapuid1-1", "snap1-1", nil, nil, nil)
 	content.ResourceVersion = "xxx"
 	_, err := storeObjectUpdate(c, content, "content")
 	if err == nil {