journal: added csi.volume.backingsnapshotid image attribute

Signed-off-by: Robert Vasek <robert.vasek@cern.ch>

internal/cephfs/store/fsjournal.go

@@ -100,7 +100,7 @@ func CheckVolExists(ctx context.Context,
 	volOptions.VolID = vid.FsSubvolName
 
 	vol := core.NewSubVolume(volOptions.conn, &volOptions.SubVolume, volOptions.ClusterID)
-	if sID != nil || pvID != nil {
+	if (sID != nil || pvID != nil) && imageData.ImageAttributes.BackingSnapshotID == "" {
 		cloneState, cloneStateErr := vol.GetCloneState(ctx)
 		if cloneStateErr != nil {
 			if errors.Is(cloneStateErr, cerrors.ErrVolumeNotFound) {
@@ -153,6 +153,8 @@ func CheckVolExists(ctx context.Context,
 			return nil, fmt.Errorf("clone is not in complete state for %s", vid.FsSubvolName)
 		}
 	}
+
+	if imageData.ImageAttributes.BackingSnapshotID == "" {
 		volOptions.RootPath, err = vol.GetVolumeRootPathCeph(ctx)
 		if err != nil {
 			if errors.Is(err, cerrors.ErrVolumeNotFound) {
@@ -173,6 +175,7 @@ func CheckVolExists(ctx context.Context,
 
 			return nil, err
 		}
+	}
 
 	// check if topology constraints match what is found
 	// TODO: we need an API to fetch subvolume attributes (size/datapool and others), based
@@ -271,7 +274,7 @@ func ReserveVol(ctx context.Context, volOptions *VolumeOptions, secret map[strin
 	imageUUID, vid.FsSubvolName, err = j.ReserveName(
 		ctx, volOptions.MetadataPool, util.InvalidPoolID,
 		volOptions.MetadataPool, util.InvalidPoolID, volOptions.RequestName,
-		volOptions.NamePrefix, "", "", volOptions.ReservedID, "")
+		volOptions.NamePrefix, "", "", volOptions.ReservedID, "", volOptions.BackingSnapshotID)
 	if err != nil {
 		return nil, err
 	}
@@ -314,7 +317,7 @@ func ReserveSnap(
 	imageUUID, vid.FsSnapshotName, err = j.ReserveName(
 		ctx, volOptions.MetadataPool, util.InvalidPoolID,
 		volOptions.MetadataPool, util.InvalidPoolID, snap.RequestName,
-		snap.NamePrefix, parentSubVolName, "", snap.ReservedID, "")
+		snap.NamePrefix, parentSubVolName, "", snap.ReservedID, "", "")
 	if err != nil {
 		return nil, err
 	}

internal/journal/voljournal.go

@@ -152,6 +152,9 @@ type Config struct {
 	// ownerKey is used to identify the owner of the volume, can be used with some KMS configurations
 	ownerKey string
 
+	// backingSnapshotIDKey is snapshot ID on which a shallow CephFS volume is based.
+	backingSnapshotIDKey string
+
 	// commonPrefix is the prefix common to all omap keys for this Config
 	commonPrefix string
 }
@@ -170,6 +173,7 @@ func NewCSIVolumeJournal(suffix string) *Config {
 		csiImageIDKey:           "csi.imageid",
 		encryptKMSKey:           "csi.volume.encryptKMS",
 		ownerKey:                "csi.volume.owner",
+		backingSnapshotIDKey:    "csi.volume.backingsnapshotid",
 		commonPrefix:            "csi.",
 	}
 }
@@ -528,6 +532,7 @@ Input arguments:
 	- kmsConf: Name of the key management service used to encrypt the image (optional)
 	- volUUID: UUID need to be reserved instead of auto-generating one (this is useful for mirroring and metro-DR)
 	- owner: the owner of the volume (optional)
+	- backingSnapshotID:  (optional)
 
 Return values:
 	- string: Contains the UUID that was reserved for the passed in reqName
@@ -537,7 +542,8 @@ Return values:
 func (conn *Connection) ReserveName(ctx context.Context,
 	journalPool string, journalPoolID int64,
 	imagePool string, imagePoolID int64,
-	reqName, namePrefix, parentName, kmsConf, volUUID, owner string,
+	reqName, namePrefix, parentName, kmsConf, volUUID, owner,
+	backingSnapshotID string,
 ) (string, string, error) {
 	// TODO: Take in-arg as ImageAttributes?
 	var (
@@ -639,6 +645,11 @@ func (conn *Connection) ReserveName(ctx context.Context,
 		omapValues[cj.cephSnapSourceKey] = parentName
 	}
 
+	// Update backing snapshot ID for shallow CephFS volume
+	if backingSnapshotID != "" {
+		omapValues[cj.backingSnapshotIDKey] = backingSnapshotID
+	}
+
 	err = setOMapKeys(ctx, conn, journalPool, cj.namespace, oid, omapValues)
 	if err != nil {
 		return "", "", err
@@ -656,6 +667,7 @@ type ImageAttributes struct {
 	Owner             string // Contains the owner to be used in combination with KmsID (for some KMS)
 	ImageID           string // Contains the image id
 	JournalPoolID     int64  // Pool ID of the CSI journal pool, stored in big endian format (on-disk data)
+	BackingSnapshotID string // ID of the backing snapshot of a shallow CephFS volume
 }
 
 // GetImageAttributes fetches all keys and their values, from a UUID directory, returning ImageAttributes structure.
@@ -684,6 +696,7 @@ func (conn *Connection) GetImageAttributes(
 		cj.cephSnapSourceKey,
 		cj.csiImageIDKey,
 		cj.ownerKey,
+		cj.backingSnapshotIDKey,
 	}
 	values, err := getOMapValues(
 		ctx, conn, pool, cj.namespace, cj.cephUUIDDirectoryPrefix+objectUUID,
@@ -700,6 +713,7 @@ func (conn *Connection) GetImageAttributes(
 	imageAttributes.KmsID = values[cj.encryptKMSKey]
 	imageAttributes.Owner = values[cj.ownerKey]
 	imageAttributes.ImageID = values[cj.csiImageIDKey]
+	imageAttributes.BackingSnapshotID = values[cj.backingSnapshotIDKey]
 
 	// image key was added at a later point, so not all volumes will have this
 	// key set when ceph-csi was upgraded

internal/rbd/rbd_journal.go

@@ -393,7 +393,7 @@ func reserveSnap(ctx context.Context, rbdSnap *rbdSnapshot, rbdVol *rbdVolume, c
 
 	rbdSnap.ReservedID, rbdSnap.RbdSnapName, err = j.ReserveName(
 		ctx, rbdSnap.JournalPool, journalPoolID, rbdSnap.Pool, imagePoolID,
-		rbdSnap.RequestName, rbdSnap.NamePrefix, rbdVol.RbdImageName, kmsID, rbdSnap.ReservedID, rbdVol.Owner)
+		rbdSnap.RequestName, rbdSnap.NamePrefix, rbdVol.RbdImageName, kmsID, rbdSnap.ReservedID, rbdVol.Owner, "")
 	if err != nil {
 		return err
 	}
@@ -473,7 +473,7 @@ func reserveVol(ctx context.Context, rbdVol *rbdVolume, rbdSnap *rbdSnapshot, cr
 
 	rbdVol.ReservedID, rbdVol.RbdImageName, err = j.ReserveName(
 		ctx, rbdVol.JournalPool, journalPoolID, rbdVol.Pool, imagePoolID,
-		rbdVol.RequestName, rbdVol.NamePrefix, "", kmsID, rbdVol.ReservedID, rbdVol.Owner)
+		rbdVol.RequestName, rbdVol.NamePrefix, "", kmsID, rbdVol.ReservedID, rbdVol.Owner, "")
 	if err != nil {
 		return err
 	}
@@ -637,7 +637,7 @@ func RegenerateJournal(
 
 	rbdVol.ReservedID, rbdVol.RbdImageName, err = j.ReserveName(
 		ctx, rbdVol.JournalPool, journalPoolID, rbdVol.Pool, imagePoolID,
-		rbdVol.RequestName, rbdVol.NamePrefix, "", kmsID, vi.ObjectUUID, rbdVol.Owner)
+		rbdVol.RequestName, rbdVol.NamePrefix, "", kmsID, vi.ObjectUUID, rbdVol.Owner, "")
 	if err != nil {
 		return "", err
 	}