Merge pull request #97013 from jsafrane/fc-fixcorrupted-fs

Fix FibreChannel volume plugin corrupting filesystem on detach

pkg/volume/fc/attacher.go

@@ -132,6 +132,7 @@ func (attacher *fcAttacher) MountDevice(spec *volume.Spec, devicePath string, de
 type fcDetacher struct {
 	mounter mount.Interface
 	manager diskManager
+	host    volume.VolumeHost
 }
 
 var _ volume.Detacher = &fcDetacher{}
@@ -142,6 +143,7 @@ func (plugin *fcPlugin) NewDetacher() (volume.Detacher, error) {
 	return &fcDetacher{
 		mounter: plugin.host.GetMounter(plugin.GetPluginName()),
 		manager: &fcUtil{},
+		host:    plugin.host,
 	}, nil
 }
 
@@ -170,7 +172,7 @@ func (detacher *fcDetacher) UnmountDevice(deviceMountPath string) error {
 	if devName == "" {
 		return nil
 	}
-	unMounter := volumeSpecToUnmounter(detacher.mounter)
+	unMounter := volumeSpecToUnmounter(detacher.mounter, detacher.host)
 	err = detacher.manager.DetachDisk(*unMounter, devName)
 	if err != nil {
 		return fmt.Errorf("fc: failed to detach disk: %s\nError: %v", devName, err)
@@ -230,12 +232,13 @@ func volumeSpecToMounter(spec *volume.Spec, host volume.VolumeHost) (*fcDiskMoun
 	}, nil
 }
 
-func volumeSpecToUnmounter(mounter mount.Interface) *fcDiskUnmounter {
+func volumeSpecToUnmounter(mounter mount.Interface, host volume.VolumeHost) *fcDiskUnmounter {
 	return &fcDiskUnmounter{
 		fcDisk: &fcDisk{
 			io: &osIOHandler{},
 		},
 		mounter:    mounter,
 		deviceUtil: volumeutil.NewDeviceHandler(volumeutil.NewIOHandler()),
+		exec:       host.GetExec(fcPluginName),
 	}
 }

pkg/volume/fc/fc.go

@@ -189,10 +189,10 @@ func (plugin *fcPlugin) newBlockVolumeMapperInternal(spec *volume.Spec, podUID t
 
 func (plugin *fcPlugin) NewUnmounter(volName string, podUID types.UID) (volume.Unmounter, error) {
 	// Inject real implementations here, test through the internal function.
-	return plugin.newUnmounterInternal(volName, podUID, &fcUtil{}, plugin.host.GetMounter(plugin.GetPluginName()))
+	return plugin.newUnmounterInternal(volName, podUID, &fcUtil{}, plugin.host.GetMounter(plugin.GetPluginName()), plugin.host.GetExec(plugin.GetPluginName()))
 }
 
-func (plugin *fcPlugin) newUnmounterInternal(volName string, podUID types.UID, manager diskManager, mounter mount.Interface) (volume.Unmounter, error) {
+func (plugin *fcPlugin) newUnmounterInternal(volName string, podUID types.UID, manager diskManager, mounter mount.Interface, exec utilexec.Interface) (volume.Unmounter, error) {
 	return &fcDiskUnmounter{
 		fcDisk: &fcDisk{
 			podUID:  podUID,
@@ -203,14 +203,15 @@ func (plugin *fcPlugin) newUnmounterInternal(volName string, podUID types.UID, m
 		},
 		mounter:    mounter,
 		deviceUtil: util.NewDeviceHandler(util.NewIOHandler()),
+		exec:       exec,
 	}, nil
 }
 
 func (plugin *fcPlugin) NewBlockVolumeUnmapper(volName string, podUID types.UID) (volume.BlockVolumeUnmapper, error) {
-	return plugin.newUnmapperInternal(volName, podUID, &fcUtil{})
+	return plugin.newUnmapperInternal(volName, podUID, &fcUtil{}, plugin.host.GetExec(plugin.GetPluginName()))
 }
 
-func (plugin *fcPlugin) newUnmapperInternal(volName string, podUID types.UID, manager diskManager) (volume.BlockVolumeUnmapper, error) {
+func (plugin *fcPlugin) newUnmapperInternal(volName string, podUID types.UID, manager diskManager, exec utilexec.Interface) (volume.BlockVolumeUnmapper, error) {
 	return &fcDiskUnmapper{
 		fcDisk: &fcDisk{
 			podUID:  podUID,
@@ -219,6 +220,7 @@ func (plugin *fcPlugin) newUnmapperInternal(volName string, podUID types.UID, ma
 			plugin:  plugin,
 			io:      &osIOHandler{},
 		},
+		exec:       exec,
 		deviceUtil: util.NewDeviceHandler(util.NewIOHandler()),
 	}, nil
 }
@@ -373,6 +375,7 @@ type fcDiskUnmounter struct {
 	*fcDisk
 	mounter    mount.Interface
 	deviceUtil util.DeviceUtil
+	exec       utilexec.Interface
 }
 
 var _ volume.Unmounter = &fcDiskUnmounter{}
@@ -400,6 +403,7 @@ var _ volume.BlockVolumeMapper = &fcDiskMapper{}
 type fcDiskUnmapper struct {
 	*fcDisk
 	deviceUtil util.DeviceUtil
+	exec       utilexec.Interface
 }
 
 var _ volume.BlockVolumeUnmapper = &fcDiskUnmapper{}

pkg/volume/fc/fc_test.go

@@ -190,7 +190,7 @@ func doTestPlugin(t *testing.T, spec *volume.Spec) {
 
 	fakeManager2 := newFakeDiskManager()
 	defer fakeManager2.Cleanup()
-	unmounter, err := plug.(*fcPlugin).newUnmounterInternal("vol1", types.UID("poduid"), fakeManager2, fakeMounter)
+	unmounter, err := plug.(*fcPlugin).newUnmounterInternal("vol1", types.UID("poduid"), fakeManager2, fakeMounter, fakeExec)
 	if err != nil {
 		t.Errorf("Failed to make a new Unmounter: %v", err)
 	}

pkg/volume/fc/fc_util.go

@@ -27,6 +27,7 @@ import (
 
 	"k8s.io/klog/v2"
 	"k8s.io/mount-utils"
+	utilexec "k8s.io/utils/exec"
 
 	"k8s.io/kubernetes/pkg/volume"
 	volumeutil "k8s.io/kubernetes/pkg/volume/util"
@@ -111,6 +112,16 @@ func findDiskWWIDs(wwid string, io ioHandler, deviceUtil volumeutil.DeviceUtil)
 	return "", ""
 }
 
+// Flushes any outstanding I/O to the device
+func flushDevice(deviceName string, exec utilexec.Interface) {
+	out, err := exec.Command("blockdev", "--flushbufs", deviceName).CombinedOutput()
+	if err != nil {
+		// Ignore the error and continue deleting the device. There is will be no retry on error.
+		klog.Warningf("Failed to flush device %s: %s\n%s", deviceName, err, string(out))
+	}
+	klog.V(4).Infof("Flushed device %s", deviceName)
+}
+
 // Removes a scsi device based upon /dev/sdX name
 func removeFromScsiSubsystem(deviceName string, io ioHandler) {
 	fileName := "/sys/block/" + deviceName + "/device/delete"
@@ -257,6 +268,9 @@ func (util *fcUtil) DetachDisk(c fcDiskUnmounter, devicePath string) error {
 	// Find slave
 	if strings.HasPrefix(dstPath, "/dev/dm-") {
 		devices = c.deviceUtil.FindSlaveDevicesOnMultipath(dstPath)
+		if err := util.deleteMultipathDevice(c.exec, dstPath); err != nil {
+			return err
+		}
 	} else {
 		// Add single devicepath to devices
 		devices = append(devices, dstPath)
@@ -264,7 +278,7 @@ func (util *fcUtil) DetachDisk(c fcDiskUnmounter, devicePath string) error {
 	klog.V(4).Infof("fc: DetachDisk devicePath: %v, dstPath: %v, devices: %v", devicePath, dstPath, devices)
 	var lastErr error
 	for _, device := range devices {
-		err := util.detachFCDisk(c.io, device)
+		err := util.detachFCDisk(c.io, c.exec, device)
 		if err != nil {
 			klog.Errorf("fc: detachFCDisk failed. device: %v err: %v", device, err)
 			lastErr = fmt.Errorf("fc: detachFCDisk failed. device: %v err: %v", device, err)
@@ -278,11 +292,12 @@ func (util *fcUtil) DetachDisk(c fcDiskUnmounter, devicePath string) error {
 }
 
 // detachFCDisk removes scsi device file such as /dev/sdX from the node.
-func (util *fcUtil) detachFCDisk(io ioHandler, devicePath string) error {
+func (util *fcUtil) detachFCDisk(io ioHandler, exec utilexec.Interface, devicePath string) error {
 	// Remove scsi device from the node.
 	if !strings.HasPrefix(devicePath, "/dev/") {
 		return fmt.Errorf("fc detach disk: invalid device name: %s", devicePath)
 	}
+	flushDevice(devicePath, exec)
 	arr := strings.Split(devicePath, "/")
 	dev := arr[len(arr)-1]
 	removeFromScsiSubsystem(dev, io)
@@ -354,13 +369,16 @@ func (util *fcUtil) DetachBlockFCDisk(c fcDiskUnmapper, mapPath, devicePath stri
 	if len(dm) != 0 {
 		// Find all devices which are managed by multipath
 		devices = c.deviceUtil.FindSlaveDevicesOnMultipath(dm)
+		if err := util.deleteMultipathDevice(c.exec, dm); err != nil {
+			return err
+		}
 	} else {
 		// Add single device path to devices
 		devices = append(devices, dstPath)
 	}
 	var lastErr error
 	for _, device := range devices {
-		err = util.detachFCDisk(c.io, device)
+		err = util.detachFCDisk(c.io, c.exec, device)
 		if err != nil {
 			klog.Errorf("fc: detachFCDisk failed. device: %v err: %v", device, err)
 			lastErr = fmt.Errorf("fc: detachFCDisk failed. device: %v err: %v", device, err)
@@ -373,6 +391,15 @@ func (util *fcUtil) DetachBlockFCDisk(c fcDiskUnmapper, mapPath, devicePath stri
 	return nil
 }
 
+func (util *fcUtil) deleteMultipathDevice(exec utilexec.Interface, dmDevice string) error {
+	out, err := exec.Command("multipath", "-f", dmDevice).CombinedOutput()
+	if err != nil {
+		return fmt.Errorf("failed to flush multipath device %s: %s\n%s", dmDevice, err, string(out))
+	}
+	klog.V(4).Infof("Flushed multipath device: %s", dmDevice)
+	return nil
+}
+
 func checkPathExists(path string) (bool, error) {
 	if pathExists, pathErr := mount.PathExists(path); pathErr != nil {
 		return pathExists, fmt.Errorf("Error checking if path exists: %v", pathErr)