rbd: Fix connection leak when a non-nil volume is returned (backport #6432)

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
@@ -526,7 +529,7 @@ func (ri *rbdImage) getImageID() error {
 	if ri.ImageID != "" {
 		return nil
 	}
-	image, err := ri.open()
+	image, err := ri.openReadOnly()
 	if err != nil {
 		return err
 	}
@@ -632,7 +635,7 @@ func (ri *rbdImage) openReadOnly() (*librbd.Image, error) {
 // isInUse is called with exponential backoff to check the image is used by
 // anyone else the returned bool value is discarded if its a RWX access.
 func (ri *rbdImage) isInUse() (bool, error) {
-	image, err := ri.open()
+	image, err := ri.openReadOnly()
 	if err != nil {
 		if errors.Is(err, rbderrors.ErrImageNotFound) || errors.Is(err, util.ErrPoolNotFound) {
 			return false, err
@@ -658,8 +661,8 @@ func (ri *rbdImage) isInUse() (bool, error) {
 		return false, fmt.Errorf("cannot map image %s it is not primary", ri)
 	}
 
-	// because we opened the image, there is at least one watcher
-	defaultWatchers := 1
+	// openReadOnly does not register a watcher on the image
+	defaultWatchers := 0
 	if mirrorInfo.Primary {
 		count, err := util.GetRBDMirrorDaemonCount(util.CsiConfigFile, ri.ClusterID)
 		if err != nil {
@@ -1785,7 +1788,7 @@ func (ri *rbdImage) GetCreationTime(ctx context.Context) (*time.Time, error) {
 // getImageInfo queries rbd about the given image and returns its metadata, and returns
 // ErrImageNotFound if provided image is not found.
 func (ri *rbdImage) getImageInfo() error {
-	image, err := ri.open()
+	image, err := ri.openReadOnly()
 	if err != nil {
 		return err
 	}
@@ -2058,7 +2061,7 @@ func (ri *rbdImage) resize(newSize int64) error {
 }
 
 func (ri *rbdImage) GetMetadata(key string) (string, error) {
-	image, err := ri.open()
+	image, err := ri.openReadOnly()
 	if err != nil {
 		return "", err
 	}