rbd: Fix connection leak when a non-nil volume is returned

This patch calls `Destroy` on the non-nil Volume/Snap object
so that conn pool GC can clean it up.

Not doing this always kept ref count > 0, preventing cleanup and
leading to lock contention, which in turn kicked in librbd lock
breaking leading to blacklisting of actual clients.

Signed-off-by: Niraj Yadav <niryadav@redhat.com>
(cherry picked from commit 7ef770a)

Author: black-dragon74

internal/rbd/controllerserver.go

@@ -853,6 +853,7 @@ func checkContentSource(
 		}
 		rbdSnap, err := genSnapFromSnapID(ctx, snapshotID, cr, req.GetSecrets())
 		if err != nil {
+			rbdSnap.Destroy(ctx)
 			log.ErrorLog(ctx, "failed to get backend snapshot for %s: %v", snapshotID, err)
 			if !errors.Is(err, rbderrors.ErrSnapNotFound) {
 				return nil, nil, status.Error(codes.Internal, err.Error())
@@ -873,6 +874,7 @@ func checkContentSource(
 		}
 		rbdvol, err := GenVolFromVolID(ctx, volID, cr, req.GetSecrets())
 		if err != nil {
+			rbdvol.Destroy(ctx)
 			log.ErrorLog(ctx, "failed to get backend image for %s: %v", volID, err)
 			if !errors.Is(err, rbderrors.ErrImageNotFound) {
 				return nil, nil, status.Error(codes.Internal, err.Error())
@@ -1498,6 +1500,8 @@ func (cs *ControllerServer) DeleteSnapshot(
 
 	rbdSnap, err := genSnapFromSnapID(ctx, snapshotID, cr, req.GetSecrets())
 	if err != nil {
+		rbdSnap.Destroy(ctx)
+
 		// if error is ErrPoolNotFound, the pool is already deleted we don't
 		// need to worry about deleting snapshot or omap data, return success
 		if errors.Is(err, util.ErrPoolNotFound) {
@@ -1812,6 +1816,8 @@ func (cs *ControllerServer) ControllerUnpublishVolume(
 
 	rv, err := GenVolFromVolID(ctx, volumeId, credentials, secrets)
 	if err != nil {
+		rv.Destroy(ctx)
+
 		return nil, status.Errorf(codes.Internal, "failed to generate volume from volume ID %s: %v",
 			volumeId, err)
 	}

internal/rbd/manager.go

@@ -75,15 +75,13 @@ func (mgr *rbdManager) GetVolumeByID(ctx context.Context, id string) (types.Volu
 
 	volume, err := GenVolFromVolID(ctx, id, creds, mgr.secrets)
 	if err != nil {
+		volume.Destroy(ctx)
+
 		switch {
 		case errors.Is(err, rbderrors.ErrImageNotFound):
-			err = fmt.Errorf("volume %s not found: %w", id, err)
-
-			return nil, err
+			return nil, fmt.Errorf("volume %s not found: %w", id, err)
 		case errors.Is(err, util.ErrPoolNotFound):
-			err = fmt.Errorf("pool %s not found for %s: %w", volume.Pool, id, err)
-
-			return nil, err
+			return nil, fmt.Errorf("pool not found for %s: %w", id, err)
 		default:
 			return nil, fmt.Errorf("failed to get volume from id %q: %w", id, err)
 		}
@@ -100,17 +98,15 @@ func (mgr *rbdManager) GetSnapshotByID(ctx context.Context, id string) (types.Sn
 
 	snapshot, err := genSnapFromSnapID(ctx, id, creds, mgr.secrets)
 	if err != nil {
+		snapshot.Destroy(ctx)
+
 		switch {
 		case errors.Is(err, rbderrors.ErrImageNotFound):
-			err = fmt.Errorf("volume %s not found: %w", id, err)
-
-			return nil, err
+			return nil, fmt.Errorf("snapshot %s not found: %w", id, err)
 		case errors.Is(err, util.ErrPoolNotFound):
-			err = fmt.Errorf("pool %s not found for %s: %w", snapshot.Pool, id, err)
-
-			return nil, err
+			return nil, fmt.Errorf("pool not found for %s: %w", id, err)
 		default:
-			return nil, fmt.Errorf("failed to get volume from id %q: %w", id, err)
+			return nil, fmt.Errorf("failed to get snapshot from id %q: %w", id, err)
 		}
 	}
 

internal/rbd/nodeserver.go

@@ -1597,6 +1597,8 @@ func (ns *NodeServer) blockNodeGetVolumeStats(
 
 	rv, err := GenVolFromVolID(ctx, volumeId, credentials, secrets)
 	if err != nil {
+		rv.Destroy(ctx)
+
 		return nil, status.Errorf(codes.Internal, "failed to generate volume from volume ID %s: %v",
 			volumeId, err)
 	}

internal/rbd/rbd_util.go

@@ -420,6 +420,9 @@ func (ri *rbdImage) Connect(cr *util.Credentials) error {
 // Destroy cleans up the rbdVolume and closes the connection to the Ceph
 // cluster in case one was setup.
 func (ri *rbdImage) Destroy(ctx context.Context) {
+	if ri == nil {
+		return
+	}
 	if ri.ioctx != nil {
 		ri.ioctx.Destroy()
 		ri.ioctx = nil