improve node stage volume mounting

- add volume locks to avoid concurrent operations on the same volume
- add detection if symlinks don't add up to the right volume
- increase logging once more

Author: mweibel

driver/driver.go

@@ -65,6 +65,10 @@ type Driver struct {
 	mounter          Mounter
 	log              *logrus.Entry
 
+	// A map storing all volumes with ongoing operations so that additional operations
+	// for that same volume (as defined by VolumeID) return an Aborted error
+	volumeLocks *VolumeLocks
+
 	// ready defines whether the driver is ready to function. This value will
 	// be used by the `Identity` service via the `Probe()` method.
 	readyMu sync.Mutex // protects ready
@@ -113,6 +117,7 @@ func NewDriver(ep, token, urlstr string, logLevel logrus.Level) (*Driver, error)
 		cloudscaleClient: cloudscaleClient,
 		mounter:          newMounter(log),
 		log:              log,
+		volumeLocks:      NewVolumeLocks(),
 	}, nil
 }
 

driver/driver_test.go

@@ -29,15 +29,11 @@ import (
 	"time"
 
 	"github.com/cenkalti/backoff/v5"
-	"github.com/google/uuid"
-	"k8s.io/mount-utils"
-
 	"github.com/cloudscale-ch/cloudscale-go-sdk/v6"
+	"github.com/google/uuid"
 	"github.com/kubernetes-csi/csi-test/v5/pkg/sanity"
 	"github.com/sirupsen/logrus"
 	"k8s.io/mount-utils"
-
-	"github.com/cloudscale-ch/cloudscale-go-sdk/v4"
 )
 
 func init() {
@@ -74,6 +70,7 @@ func TestDriverSuite(t *testing.T) {
 		cloudscaleClient: cloudscaleClient,
 		mounter:          fm,
 		log:              logrus.New().WithField("test_enabed", true),
+		volumeLocks:      NewVolumeLocks(),
 	}
 	defer driver.Stop()
 

driver/driver_volume_type_test.go

@@ -90,7 +90,7 @@ func TestCreateVolumeInvalidType(t *testing.T) {
 	)
 
 	assert.Error(t, err)
-	//assert.Error(t, err, "invalid volume capabilities requested for LUKS xx.")
+	// assert.Error(t, err, "invalid volume capabilities requested for LUKS xx.")
 }
 
 func TestCreateVolumeInvalidLUKSAndRaw(t *testing.T) {
@@ -183,5 +183,6 @@ func createDriverForTest(t *testing.T) *Driver {
 		mounter:          &fakeMounter{},
 		log:              logrus.New().WithField("test_enabled", true),
 		cloudscaleClient: cloudscaleClient,
+		volumeLocks:      NewVolumeLocks(),
 	}
 }

driver/luks_util.go

@@ -20,7 +20,6 @@ package driver
 import (
 	"errors"
 	"fmt"
-	"io/ioutil"
 	"os"
 	"os/exec"
 	"strings"
@@ -204,9 +203,9 @@ func luksPrepareMount(source string, ctx LuksContext, log *logrus.Entry) (string
 		return "", err
 	}
 	defer func() {
-		e := os.Remove(filename)
-		if e != nil {
-			log.Errorf("cannot delete temporary file %s: %s", filename, e.Error())
+		err := os.Remove(filename)
+		if err != nil {
+			log.Errorf("cannot delete temporary file %s: %s", filename, err.Error())
 		}
 	}()
 
@@ -378,10 +377,14 @@ func writeLuksKey(key string, log *logrus.Entry) (string, error) {
 	if !checkTmpFs("/tmp") {
 		return "", errors.New("temporary directory /tmp is not a tmpfs volume; refusing to write luks key to a volume backed by a disk")
 	}
-	tmpFile, err := ioutil.TempFile("/tmp", "luks-")
+	tmpFile, err := os.CreateTemp("/tmp", "luks-")
 	if err != nil {
 		return "", err
 	}
+	defer func() {
+		_ = tmpFile.Close()
+	}()
+
 	_, err = tmpFile.WriteString(key)
 	if err != nil {
 		log.WithField("tmp_file", tmpFile.Name()).Warnf("Unable to write luks key file: %s", err.Error())

driver/mounter.go

@@ -25,15 +25,13 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"regexp"
 	"strconv"
 	"strings"
 	"sync"
 	"syscall"
 	"time"
 
-	"k8s.io/mount-utils"
-	kexec "k8s.io/utils/exec"
-
 	"github.com/sirupsen/logrus"
 	"golang.org/x/sys/unix"
 	"k8s.io/mount-utils"
@@ -204,7 +202,9 @@ func (m *mounter) Mount(source, target, fsType string, luksContext LuksContext,
 		if err != nil {
 			return fmt.Errorf("failed to create target file for raw block bind mount: %v", err)
 		}
-		file.Close()
+		if err := file.Close(); err != nil {
+			m.log.WithFields(logrus.Fields{"target": target}).Error("failed to close file handle")
+		}
 	} else {
 		// create target, os.Mkdirall is noop if directory exists
 		err := os.MkdirAll(target, 0750)
@@ -222,6 +222,7 @@ func (m *mounter) Mount(source, target, fsType string, luksContext LuksContext,
 			}).Error("failed to prepare luks volume for mounting")
 			return err
 		}
+		// source is /dev/mapper/<volumeName> now
 		source = luksSource
 	}
 
@@ -267,6 +268,9 @@ func (m *mounter) Unmount(target string, luksContext LuksContext) error {
 	// a luks volume needs to be closed after unmounting; get the source
 	// of the mount to check if that is a luks volume
 	mountSources, err := getMountSources(target)
+	if err != nil {
+		return fmt.Errorf("failed to get mount sources for target %q: %v", target, err)
+	}
 
 	err = mount.CleanupMountPoint(target, m.kMounter, true)
 	if err != nil {
@@ -476,15 +480,35 @@ func (m *mounter) FinalizeVolumeAttachmentAndFindPath(logger *logrus.Entry, volu
 				logger.WithFields(logrus.Fields{
 					"disk_id_path": diskIDPath,
 					"error":        err,
-				}).Debug("FinalizeVolumeAttachmentAndFindPath: found path but failed to resolve symlink")
-			} else {
+				}).Error("FinalizeVolumeAttachmentAndFindPath: found path but failed to resolve symlink")
+				return "", fmt.Errorf("FinalizeVolumeAttachmentAndFindPath: found path %s but failed to resolve symlink: %w", diskIDPath, err)
+			}
+			logger.WithFields(logrus.Fields{
+				"disk_id_path":    diskIDPath,
+				"resolved_device": resolved,
+				"num_tries":       numTries,
+			}).Debug("FinalizeVolumeAttachmentAndFindPath: found device path")
+
+			devFsSerial, innerErr := getScsiSerial(resolved)
+			if innerErr != nil {
+				logger.WithFields(logrus.Fields{
+					"disk_id_path":    diskIDPath,
+					"resolved_device": resolved,
+					"num_tries":       numTries,
+				}).Error("FinalizeVolumeAttachmentAndFindPath: unable to get device serial")
+				return "", fmt.Errorf("FinalizeVolumeAttachmentAndFindPath: unable to get serial number for disk %s at path %s: %w", diskIDPath, resolved, innerErr)
+			}
+			// success: found a path in /dev/disk/by-id/* which resolved to a symlink in /dev/* and that returned the right serial.
+			if devFsSerial != "" && devFsSerial == volumeID {
 				logger.WithFields(logrus.Fields{
 					"disk_id_path":    diskIDPath,
 					"resolved_device": resolved,
+					"serial":          devFsSerial,
 					"num_tries":       numTries,
-				}).Debug("FinalizeVolumeAttachmentAndFindPath: found device path")
+				}).Debug("FinalizeVolumeAttachmentAndFindPath: found device and resolved serial")
+				return diskIDPath, nil
 			}
-			return diskIDPath, nil
+			// A /dev/* path exists, but it's not matching the right serial. Attempt to repair by triggering udevadm.
 		}
 
 		logger.WithFields(logrus.Fields{
@@ -502,6 +526,41 @@ func (m *mounter) FinalizeVolumeAttachmentAndFindPath(logger *logrus.Entry, volu
 	return "", errors.New("FinalizeVolumeAttachmentAndFindPath: Timeout after 30s")
 }
 
+// getScsiSerial assumes that scsiIdPath exists and will error if it
+// doesnt. It is the callers responsibility to verify the existence of this
+// tool. Calls scsi_id on the given devicePath to get the serial number reported
+// by that device.
+func getScsiSerial(devicePath string) (string, error) {
+	out, err := exec.Command(
+		"/usr/lib/udev/scsi_id",
+		"--page=0x83",
+		"--whitelisted",
+		fmt.Sprintf("--device=%v", devicePath)).CombinedOutput()
+	if err != nil {
+		return "", fmt.Errorf("scsi_id failed for device %q with output %s: %w", devicePath, string(out), err)
+	}
+
+	return parseScsiSerial(string(out))
+}
+
+var (
+	// scsi_id output should be in the form of:
+	// 0QEMU QEMU HARDDISK <disk id>
+	scsiPattern = `^0QEMU\s+QEMU\sHARDDISK\s+([\S]+)\s*$`
+	// regex to parse scsi_id output and extract the serial
+	scsiRegex = regexp.MustCompile(scsiPattern)
+)
+
+// Parse the output returned by scsi_id and extract the serial number
+func parseScsiSerial(output string) (string, error) {
+	substrings := scsiRegex.FindStringSubmatch(output)
+	if substrings == nil {
+		return "", fmt.Errorf("scsi_id output cannot be parsed: %q", output)
+	}
+
+	return substrings[1], nil
+}
+
 func runCmdWithTimeout(name string, args []string, logger *logrus.Entry, timeout time.Duration) {
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()

driver/node.go

@@ -15,14 +15,6 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// Code generated by protoc-gen-go. DO NOT EDIT.
-
-// NOTE: THIS IS NOT GENERATED. We have to add the line above to prevent golint
-// checking this file. This is needed because some methods end with xxxId, but
-// golint wants them to be xxxID. But we're not able to change it as the
-// official CSI spec is that way and we have to implement the interface
-// exactly.
-
 package driver
 
 import (
@@ -40,8 +32,6 @@ import (
 )
 
 const (
-	diskDOPrefix = "scsi-0DO_Volume_"
-
 	// Current technical limit is 128
 	//   - 1 for root
 	//   - 1 for /var/lib/docker
@@ -72,6 +62,11 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 		return nil, status.Error(codes.InvalidArgument, "NodeStageVolume Volume Capability must be provided")
 	}
 
+	if acquired := d.volumeLocks.TryAcquire(req.VolumeId); !acquired {
+		return nil, status.Errorf(codes.Aborted, "an operation with the given Volume ID %s already exists", req.VolumeId)
+	}
+	defer d.volumeLocks.Release(req.VolumeId)
+
 	// Apparently sometimes we need to call udevadm trigger to get the volume
 	// properly registered in /dev/disk. More information can be found here:
 	// https://github.com/cloudscale-ch/csi-cloudscale/issues/9
@@ -80,12 +75,17 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 		return nil, err
 	}
 
+	d.log.WithFields(logrus.Fields{
+		"volume_id":   req.VolumeId,
+		"device_path": source,
+	}).Info("successfully found attached volume_id at device_path")
+
 	// Debug logging to help diagnose potential race conditions with concurrent volume mounts
 	resolvedSource, resolveErr := filepath.EvalSymlinks(source)
 	if resolveErr != nil {
 		d.log.WithFields(logrus.Fields{
-			"volume_id":    req.VolumeId,
-			"source":       source,
+			"volume_id":     req.VolumeId,
+			"source":        source,
 			"resolve_error": resolveErr,
 		}).Debug("failed to resolve source symlink")
 	} else {
@@ -115,7 +115,7 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 		return &csi.NodeStageVolumeResponse{}, nil
 	}
 
-	target := req.StagingTargetPath
+	stagingTargetPath := req.StagingTargetPath
 
 	mnt := req.VolumeCapability.GetMount()
 	options := mnt.MountFlags
@@ -153,19 +153,21 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 		ll.Info("source device is already formatted")
 	}
 
-	ll.Info("mounting the volume for staging")
+	ll.Info("checking if stagingTargetPath is already mounted")
 
-	mounted, err := d.mounter.IsMounted(target)
+	mounted, err := d.mounter.IsMounted(stagingTargetPath)
 	if err != nil {
+		ll.WithError(err).Error("unable to check if already mounted")
 		return nil, err
 	}
 
 	if !mounted {
-		if err := d.mounter.Mount(source, target, fsType, luksContext, options...); err != nil {
+		ll.Info("not mounted yet, mounting the volume for staging")
+		if err := d.mounter.Mount(source, stagingTargetPath, fsType, luksContext, options...); err != nil {
 			return nil, status.Error(codes.Internal, err.Error())
 		}
 	} else {
-		ll.Info("source device is already mounted to the target path")
+		ll.Info("source device is already mounted to the stagingTargetPath path")
 	}
 
 	ll.Info("formatting and mounting stage volume is finished")
@@ -182,6 +184,11 @@ func (d *Driver) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstageVolu
 		return nil, status.Error(codes.InvalidArgument, "NodeUnstageVolume Staging Target Path must be provided")
 	}
 
+	if acquired := d.volumeLocks.TryAcquire(req.VolumeId); !acquired {
+		return nil, status.Errorf(codes.Aborted, "an operation with the given Volume ID %s already exists", req.VolumeId)
+	}
+	defer d.volumeLocks.Release(req.VolumeId)
+
 	luksContext := LuksContext{VolumeLifecycle: VolumeLifecycleNodeUnstageVolume}
 
 	ll := d.log.WithFields(logrus.Fields{
@@ -229,6 +236,11 @@ func (d *Driver) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolu
 		return nil, status.Error(codes.InvalidArgument, "NodePublishVolume Volume Capability must be provided")
 	}
 
+	if acquired := d.volumeLocks.TryAcquire(req.VolumeId); !acquired {
+		return nil, status.Errorf(codes.Aborted, "an operation with the given Volume ID %s already exists", req.VolumeId)
+	}
+	defer d.volumeLocks.Release(req.VolumeId)
+
 	publishContext := req.GetPublishContext()
 	if publishContext == nil {
 		return nil, status.Error(codes.InvalidArgument, "PublishContext must be provided")
@@ -276,6 +288,11 @@ func (d *Driver) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublish
 		return nil, status.Error(codes.InvalidArgument, "NodeUnpublishVolume Target Path must be provided")
 	}
 
+	if acquired := d.volumeLocks.TryAcquire(req.VolumeId); !acquired {
+		return nil, status.Errorf(codes.Aborted, "an operation with the given Volume ID %s already exists", req.VolumeId)
+	}
+	defer d.volumeLocks.Release(req.VolumeId)
+
 	luksContext := LuksContext{VolumeLifecycle: VolumeLifecycleNodeUnpublishVolume}
 
 	ll := d.log.WithFields(logrus.Fields{