journal: replace getStoredImageId with go-ceph

moved implementation of getStoredImageID to go-ceph
to get better performance.

Signed-off-by: Madhu Rajanna <madhupr007@gmail.com>

internal/journal/voljournal.go

@@ -587,6 +587,7 @@ type ImageAttributes struct {
 	SourceName    string // Contains the parent image name for the passed in UUID, if it is a snapshot
 	ImageName     string // Contains the image or subvolume name for the passed in UUID
 	KmsID         string // Contains encryption KMS, if it is an encrypted image
+	ImageID       string // Contains the image id
 	JournalPoolID int64  // Pool ID of the CSI journal pool, stored in big endian format (on-disk data)
 }
 
@@ -609,6 +610,7 @@ func (conn *Connection) GetImageAttributes(ctx context.Context, pool, objectUUID
 		cj.encryptKMSKey,
 		cj.csiJournalPool,
 		cj.cephSnapSourceKey,
+		cj.csiImageIDKey,
 	}
 	values, err := getOMapValues(
 		ctx, conn, pool, cj.namespace, cj.cephUUIDDirectoryPrefix+objectUUID,
@@ -625,6 +627,7 @@ func (conn *Connection) GetImageAttributes(ctx context.Context, pool, objectUUID
 	var found bool
 	imageAttributes.RequestName = values[cj.csiNameKey]
 	imageAttributes.KmsID = values[cj.encryptKMSKey]
+	imageAttributes.ImageID = values[cj.csiImageIDKey]
 
 	// image key was added at a later point, so not all volumes will have this
 	// key set when ceph-csi was upgraded

internal/rbd/controllerserver.go

@@ -945,23 +945,9 @@ func (cs *ControllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteS
 			return nil, status.Error(codes.Internal, err.Error())
 		}
 	} else {
-		// save image ID
-		var j = &journal.Connection{}
-		j, err = snapJournal.Connect(rbdSnap.Monitors, cr)
-		if err != nil {
-			klog.Errorf(util.Log(ctx, "failed to connect to cluster: %v"), err)
-			return nil, status.Error(codes.Internal, err.Error())
-		}
-		defer j.Destroy()
-		// TODO replace GetStoredImageID with GetImageAttributes in all places
-		rbdVol.ImageID, err = j.GetStoredImageID(ctx, rbdSnap.JournalPool, rbdSnap.ReservedID, cr)
-		if err != nil {
-			klog.Errorf(util.Log(ctx, "failed to get reserved image id: %v"), err)
-			return nil, status.Error(codes.Internal, err.Error())
-		}
+		rbdVol.ImageID = rbdSnap.ImageID
 		// update parent name to delete the snapshot
 		rbdSnap.RbdImageName = rbdVol.RbdImageName
-
 		err = cleanUpSnapshot(ctx, rbdVol, rbdSnap, rbdVol, cr)
 		if err != nil {
 			klog.Errorf(util.Log(ctx, "failed to delete image: %v"), err)

internal/rbd/rbd_journal.go

@@ -133,6 +133,7 @@ func checkSnapCloneExists(ctx context.Context, parentVol *rbdVolume, rbdSnap *rb
 	}
 	snapUUID := snapData.ImageUUID
 	rbdSnap.RbdSnapName = snapData.ImageAttributes.ImageName
+	rbdSnap.ImageID = snapData.ImageAttributes.ImageID
 
 	// it should never happen that this disagrees, but check
 	if rbdSnap.Pool != snapData.ImagePool {
@@ -193,8 +194,7 @@ func checkSnapCloneExists(ctx context.Context, parentVol *rbdVolume, rbdSnap *rb
 		return false, err
 	}
 
-	vol.ImageID, err = j.GetStoredImageID(ctx, vol.JournalPool, vol.ReservedID, cr)
-	if _, ok := err.(util.ErrKeyNotFound); ok {
+	if vol.ImageID == "" {
 		sErr := vol.getImageID()
 		if sErr != nil {
 			klog.Errorf(util.Log(ctx, "failed to get image id %s: %v"), vol, sErr)
@@ -254,7 +254,7 @@ func (rv *rbdVolume) Exists(ctx context.Context) (bool, error) {
 
 	rv.ReservedID = imageData.ImageUUID
 	rv.RbdImageName = imageData.ImageAttributes.ImageName
-
+	rv.ImageID = imageData.ImageAttributes.ImageID
 	// check if topology constraints match what is found
 	rv.Topology, err = util.MatchTopologyForPool(rv.TopologyPools, rv.TopologyRequirement,
 		imageData.ImagePool)
@@ -282,8 +282,7 @@ func (rv *rbdVolume) Exists(ctx context.Context) (bool, error) {
 		return false, err
 	}
 
-	rv.ImageID, err = j.GetStoredImageID(ctx, rv.JournalPool, rv.ReservedID, rv.conn.Creds)
-	if _, ok := err.(util.ErrKeyNotFound); ok {
+	if rv.ImageID == "" {
 		err = rv.getImageID()
 		if err != nil {
 			klog.Errorf(util.Log(ctx, "failed to get image id %s: %v"), rv, err)

internal/rbd/rbd_util.go

@@ -124,12 +124,14 @@ type rbdSnapshot struct {
 	// RequestName is the CSI generated snapshot name for the rbdSnapshot
 	// JournalPool is the ceph pool in which the CSI snapshot Journal is stored
 	// Pool is where the image snapshot journal and snapshot is stored, and could be the same as `JournalPool`
+	// ImageID contains the image id of cloned image
 	SourceVolumeID string
 	RbdImageName   string
 	ReservedID     string
 	NamePrefix     string
 	RbdSnapName    string
 	SnapID         string
+	ImageID        string
 	Monitors       string
 	JournalPool    string
 	Pool           string
@@ -563,6 +565,7 @@ func genSnapFromSnapID(ctx context.Context, rbdSnap *rbdSnapshot, snapshotID str
 	if err != nil {
 		return err
 	}
+	rbdSnap.ImageID = imageAttributes.ImageID
 	rbdSnap.RequestName = imageAttributes.RequestName
 	rbdSnap.RbdImageName = imageAttributes.SourceName
 	rbdSnap.RbdSnapName = imageAttributes.ImageName
@@ -633,6 +636,7 @@ func genVolFromVolID(ctx context.Context, volumeID string, cr *util.Credentials,
 	rbdVol.RequestName = imageAttributes.RequestName
 	rbdVol.RbdImageName = imageAttributes.ImageName
 	rbdVol.ReservedID = vi.ObjectUUID
+	rbdVol.ImageID = imageAttributes.ImageID
 
 	if imageAttributes.KmsID != "" {
 		rbdVol.Encrypted = true
@@ -650,8 +654,7 @@ func genVolFromVolID(ctx context.Context, volumeID string, cr *util.Credentials,
 		}
 	}
 
-	rbdVol.ImageID, err = j.GetStoredImageID(ctx, rbdVol.JournalPool, rbdVol.ReservedID, cr)
-	if _, ok := err.(util.ErrKeyNotFound); ok {
+	if rbdVol.ImageID == "" {
 		err = rbdVol.getImageID()
 		if err != nil {
 			klog.Errorf(util.Log(ctx, "failed to get image id %s: %v"), rbdVol, err)

internal/rbd/snapshot.go

@@ -87,6 +87,7 @@ func generateVolFromSnap(rbdSnap *rbdSnapshot) *rbdVolume {
 	vol.Pool = rbdSnap.Pool
 	vol.JournalPool = rbdSnap.JournalPool
 	vol.RbdImageName = rbdSnap.RbdSnapName
+	vol.ImageID = rbdSnap.ImageID
 	return vol
 }