rbd: set metadata on the snapshot

Set snapshot-name/snapshot-namespace/snapshotcontent-name details
on RBD backend snapshot image as metadata on snapshot

Signed-off-by: Prasanna Kumar Kalever <prasanna.kalever@redhat.com>

internal/rbd/controllerserver.go

@@ -963,6 +963,7 @@ func (cs *ControllerServer) ValidateVolumeCapabilities(
 }
 
 // CreateSnapshot creates the snapshot in backend and stores metadata in store.
+// nolint:cyclop // TODO: reduce complexity
 func (cs *ControllerServer) CreateSnapshot(
 	ctx context.Context,
 	req *csi.CreateSnapshotRequest) (*csi.CreateSnapshotResponse, error) {
@@ -1036,7 +1037,7 @@ func (cs *ControllerServer) CreateSnapshot(
 		return nil, status.Errorf(codes.Internal, err.Error())
 	}
 	if found {
-		return cloneFromSnapshot(ctx, rbdVol, rbdSnap, cr)
+		return cloneFromSnapshot(ctx, rbdVol, rbdSnap, cr, req.GetParameters())
 	}
 
 	err = flattenTemporaryClonedImages(ctx, rbdVol, cr)
@@ -1062,6 +1063,16 @@ func (cs *ControllerServer) CreateSnapshot(
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
+	// Update the metadata on snapshot not on the original image
+	rbdVol.RbdImageName = rbdSnap.RbdSnapName
+
+	// Set snapshot-name/snapshot-namespace/snapshotcontent-name details
+	// on RBD backend image as metadata on create
+	err = rbdVol.setSnapshotMetadata(req.GetParameters())
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
 	return &csi.CreateSnapshotResponse{
 		Snapshot: &csi.Snapshot{
 			SizeBytes:      vol.VolSize,
@@ -1079,7 +1090,8 @@ func cloneFromSnapshot(
 	ctx context.Context,
 	rbdVol *rbdVolume,
 	rbdSnap *rbdSnapshot,
-	cr *util.Credentials) (*csi.CreateSnapshotResponse, error) {
+	cr *util.Credentials,
+	parameters map[string]string) (*csi.CreateSnapshotResponse, error) {
 	vol := generateVolFromSnap(rbdSnap)
 	err := vol.Connect(cr)
 	if err != nil {
@@ -1112,6 +1124,15 @@ func cloneFromSnapshot(
 		return nil, status.Errorf(codes.Internal, err.Error())
 	}
 
+	// Update snapshot-name/snapshot-namespace/snapshotcontent-name details on
+	// RBD backend image as metadata on restart of provisioner pod when image exist
+	if len(parameters) != 0 {
+		err = rbdVol.setSnapshotMetadata(parameters)
+		if err != nil {
+			return nil, status.Error(codes.Internal, err.Error())
+		}
+	}
+
 	return &csi.CreateSnapshotResponse{
 		Snapshot: &csi.Snapshot{
 			SizeBytes:      rbdSnap.VolSize,

internal/rbd/rbd_util.go

@@ -1933,3 +1933,16 @@ func (rv *rbdVolume) setVolumeMetadata(parameters map[string]string) error {
 
 	return nil
 }
+
+// setSnapshotMetadata Set snapshot-name/snapshot-namespace/snapshotcontent-name metadata
+// on RBD image.
+func (rv *rbdVolume) setSnapshotMetadata(parameters map[string]string) error {
+	for k, v := range k8s.GetSnapshotMetadata(parameters) {
+		err := rv.SetMetadata(k, v)
+		if err != nil {
+			return fmt.Errorf("failed to set metadata key %q, value %q on image: %w", k, v, err)
+		}
+	}
+
+	return nil
+}

internal/util/k8s/parameters.go

@@ -29,6 +29,11 @@ const (
 	pvcNameKey      = "csi.storage.k8s.io/pvc/name"
 	pvcNamespaceKey = "csi.storage.k8s.io/pvc/namespace"
 	pvNameKey       = "csi.storage.k8s.io/pv/name"
+
+	// snapshot metadata keys.
+	volSnapNameKey        = "csi.storage.k8s.io/volumesnapshot/name"
+	volSnapNamespaceKey   = "csi.storage.k8s.io/volumesnapshot/namespace"
+	volSnapContentNameKey = "csi.storage.k8s.io/volumesnapshotcontent/name"
 )
 
 // RemoveCSIPrefixedParameters removes parameters prefixed with csiParameterPrefix.
@@ -79,3 +84,19 @@ func PrepareVolumeMetadata(pvcName, pvcNamespace, pvName string) map[string]stri
 
 	return newParam
 }
+
+// GetSnapshotMetadata filter parameters, only return
+// snapshot-name/snapshot-namespace/snapshotcontent-name metadata.
+func GetSnapshotMetadata(parameters map[string]string) map[string]string {
+	keys := []string{volSnapNameKey, volSnapNamespaceKey, volSnapContentNameKey}
+	newParam := map[string]string{}
+	for k, v := range parameters {
+		for _, key := range keys {
+			if strings.Contains(k, key) {
+				newParam[k] = v
+			}
+		}
+	}
+
+	return newParam
+}