Add flag to enable deviceInUseTimeout. Pass args through NodeServer constructor

Author: davis-haba

cmd/gce-pd-csi-driver/main.go

@@ -63,7 +63,9 @@ var (
 	waitForOpBackoffJitter    = flag.Float64("wait-op-backoff-jitter", 0.0, "Jitter for wait for operation backoff")
 	waitForOpBackoffSteps     = flag.Int("wait-op-backoff-steps", 100, "Steps for wait for operation backoff")
 	waitForOpBackoffCap       = flag.Duration("wait-op-backoff-cap", 0, "Cap for wait for operation backoff")
-	deviceInUseTimeout        = flag.Duration("device-in-use-timeout", 30*time.Second, "Max time to wait for a device to be unsed when attempting to unstage")
+
+	enableDeviceInUseTimeout = flag.Bool("enable-device-in-use-timeout", true, "If set to true, ignores device in use errors when attempting to unstage a device if it has been stuck for longer than 'device-in-use-timeout'")
+	deviceInUseTimeout       = flag.Duration("device-in-use-timeout", 30*time.Second, "Max time to wait for a device to be unused when attempting to unstage")
 
 	maxProcs                = flag.Int("maxprocs", 1, "GOMAXPROCS override")
 	maxConcurrentFormat     = flag.Int("max-concurrent-format", 1, "The maximum number of concurrent format exec calls")
@@ -176,7 +178,6 @@ func handle() {
 	defer cancel()
 
 	// Initialize driver
-	driver.DeviceInUseTimeout = *deviceInUseTimeout
 	gceDriver := driver.GetGCEDriver()
 
 	// Initialize identity server
@@ -242,7 +243,11 @@ func handle() {
 		if err != nil {
 			klog.Fatalf("Failed to set up metadata service: %v", err.Error())
 		}
-		nodeServer = driver.NewNodeServer(gceDriver, mounter, deviceUtils, meta, statter)
+		nsArgs := driver.NodeServerArgs{
+			EnableDeviceInUseTimeout: *enableDeviceInUseTimeout,
+			DeviceInUseTimeout:       *deviceInUseTimeout,
+		}
+		nodeServer = driver.NewNodeServer(gceDriver, mounter, deviceUtils, meta, statter, nsArgs)
 		if *maxConcurrentFormatAndMount > 0 {
 			nodeServer = nodeServer.WithSerializedFormatAndMount(*formatAndMountTimeout, *maxConcurrentFormatAndMount)
 		}

pkg/gce-pd-csi-driver/device_error_map.go

@@ -24,37 +24,45 @@ import (
 // deviceErrMap is an atomic data datastructure for recording deviceInUseError times
 // for specified devices
 type deviceErrMap struct {
+	enabled           bool
+	timeout           time.Duration
 	mux               sync.Mutex
 	deviceInUseErrors map[string]time.Time
 }
 
-var DeviceInUseTimeout = time.Second * 30
+func newDeviceErrMap(shouldEnable bool, timeout time.Duration) *deviceErrMap {
+	return &deviceErrMap{
+		deviceInUseErrors: make(map[string]time.Time),
+		enabled:           shouldEnable,
+		timeout:           timeout,
+	}
+}
 
 // checkDeviceErrorTimeout returns true an error was encountered for the specified deviceName,
 // where the error happened at least `deviceInUseTimeout` seconds ago.
 func (devErrMap *deviceErrMap) checkDeviceErrorTimeout(deviceName string) bool {
+	if !devErrMap.enabled {
+		return false
+	}
+
 	devErrMap.mux.Lock()
 	defer devErrMap.mux.Unlock()
 
-	if devErrMap.deviceInUseErrors == nil {
-		devErrMap.deviceInUseErrors = make(map[string]time.Time)
-	}
-
 	lastErrTime, exists := devErrMap.deviceInUseErrors[deviceName]
-	return exists && time.Now().Sub(lastErrTime).Seconds() >= DeviceInUseTimeout.Seconds()
+	return exists && time.Now().Sub(lastErrTime).Seconds() >= devErrMap.timeout.Seconds()
 }
 
 // markDeviceError updates the internal `deviceInUseErrors` map to denote an error was encounted
 // for the specified deviceName at the current time. If an error had previously been recorded, the
 // time will not be updated.
 func (devErrMap *deviceErrMap) markDeviceError(deviceName string) {
+	if !devErrMap.enabled {
+		return
+	}
+
 	devErrMap.mux.Lock()
 	defer devErrMap.mux.Unlock()
 
-	if devErrMap.deviceInUseErrors == nil {
-		devErrMap.deviceInUseErrors = make(map[string]time.Time)
-	}
-
 	// If an earlier error has already been recorded, do not overwrite it
 	if _, exists := devErrMap.deviceInUseErrors[deviceName]; !exists {
 		now := time.Now()
@@ -65,6 +73,10 @@ func (devErrMap *deviceErrMap) markDeviceError(deviceName string) {
 
 // deleteDevice removes a specified device name from the map
 func (devErrMap *deviceErrMap) deleteDevice(deviceName string) {
+	if !devErrMap.enabled {
+		return
+	}
+
 	devErrMap.mux.Lock()
 	defer devErrMap.mux.Unlock()
 	delete(devErrMap.deviceInUseErrors, deviceName)

pkg/gce-pd-csi-driver/gce-pd-driver.go

@@ -145,14 +145,15 @@ func NewIdentityServer(gceDriver *GCEDriver) *GCEIdentityServer {
 	}
 }
 
-func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, deviceUtils deviceutils.DeviceUtils, meta metadataservice.MetadataService, statter mountmanager.Statter) *GCENodeServer {
+func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, deviceUtils deviceutils.DeviceUtils, meta metadataservice.MetadataService, statter mountmanager.Statter, args NodeServerArgs) *GCENodeServer {
 	return &GCENodeServer{
-		Driver:          gceDriver,
-		Mounter:         mounter,
-		DeviceUtils:     deviceUtils,
-		MetadataService: meta,
-		volumeLocks:     common.NewVolumeLocks(),
-		VolumeStatter:   statter,
+		Driver:            gceDriver,
+		Mounter:           mounter,
+		DeviceUtils:       deviceUtils,
+		MetadataService:   meta,
+		volumeLocks:       common.NewVolumeLocks(),
+		VolumeStatter:     statter,
+		deviceInUseErrors: newDeviceErrMap(args.EnableDeviceInUseTimeout, args.DeviceInUseTimeout),
 	}
 }
 

pkg/gce-pd-csi-driver/node.go

@@ -53,7 +53,7 @@ type GCENodeServer struct {
 
 	// deviceInUseErrors keeps tracks of device names and a timestamp for when an error is
 	// encounted for that device
-	deviceInUseErrors deviceErrMap
+	deviceInUseErrors *deviceErrMap
 
 	// If set, this semaphore will be used to serialize formatAndMount. It will be raised
 	// when the operation starts, and lowered either when finished, or when
@@ -69,6 +69,15 @@ type GCENodeServer struct {
 	csi.UnimplementedNodeServer
 }
 
+type NodeServerArgs struct {
+	// EnableDeviceInUseTimeout enables functionality which will cause unstage requests
+	// that are stucking waiting for a device to be unused to return succesfully
+	// after `DeviceInUseTimeout` has elapsed
+	EnableDeviceInUseTimeout bool
+
+	DeviceInUseTimeout time.Duration
+}
+
 var _ csi.NodeServer = &GCENodeServer{}
 
 // The constants are used to map from the machine type to the limit of
@@ -465,7 +474,7 @@ func (ns *GCENodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUns
 		if errors.As(err, &ignoreableErr) {
 			klog.Warningf("Unabled to check if device for %s is unused. Device has been unmounted successfully. Ignoring and continuing with unstaging. (%v)", volumeID, err)
 		} else if ns.deviceInUseErrors.checkDeviceErrorTimeout(volumeID) {
-			klog.Warningf("Device %s could not be released after timeout of %d seconds. NodeUnstageVolume will return success.", volumeID, DeviceInUseTimeout)
+			klog.Warningf("Device %s could not be released after timeout of %f seconds. NodeUnstageVolume will return success.", volumeID, ns.deviceInUseErrors.timeout.Seconds())
 		} else {
 			ns.deviceInUseErrors.markDeviceError(volumeID)
 			return nil, status.Errorf(codes.Internal, "NodeUnstageVolume for volume %s failed: %v", volumeID, err)

pkg/gce-pd-csi-driver/node_test.go

@@ -51,7 +51,7 @@ func getTestGCEDriverWithCustomMounter(t *testing.T, mounter *mount.SafeFormatAn
 
 func getCustomTestGCEDriver(t *testing.T, mounter *mount.SafeFormatAndMount, deviceUtils deviceutils.DeviceUtils, metaService metadataservice.MetadataService) *GCEDriver {
 	gceDriver := GetGCEDriver()
-	nodeServer := NewNodeServer(gceDriver, mounter, deviceUtils, metaService, mountmanager.NewFakeStatter(mounter))
+	nodeServer := NewNodeServer(gceDriver, mounter, deviceUtils, metaService, mountmanager.NewFakeStatter(mounter), NodeServerArgs{false, 0})
 	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, nil, nil, nodeServer)
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)
@@ -62,7 +62,7 @@ func getCustomTestGCEDriver(t *testing.T, mounter *mount.SafeFormatAndMount, dev
 func getTestBlockingMountGCEDriver(t *testing.T, readyToExecute chan chan struct{}) *GCEDriver {
 	gceDriver := GetGCEDriver()
 	mounter := mountmanager.NewFakeSafeBlockingMounter(readyToExecute)
-	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(false), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter))
+	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(false), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter), NodeServerArgs{false, 0})
 	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, nil, nil, nodeServer)
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)
@@ -73,7 +73,7 @@ func getTestBlockingMountGCEDriver(t *testing.T, readyToExecute chan chan struct
 func getTestBlockingFormatAndMountGCEDriver(t *testing.T, readyToExecute chan chan struct{}) *GCEDriver {
 	gceDriver := GetGCEDriver()
 	mounter := mountmanager.NewFakeSafeBlockingMounter(readyToExecute)
-	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(false), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter)).WithSerializedFormatAndMount(5*time.Second, 1)
+	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(false), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter), NodeServerArgs{false, 0}).WithSerializedFormatAndMount(5*time.Second, 1)
 
 	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, nil, nil, nodeServer)
 	if err != nil {

test/e2e/utils/utils.go

@@ -68,7 +68,7 @@ func GCEClientAndDriverSetup(instance *remote.InstanceInfo, driverConfig DriverC
 		"--use-instance-api-to-list-volumes-published-nodes",
 		"--supports-dynamic-iops-provisioning=hyperdisk-balanced,hyperdisk-extreme",
 		"--supports-dynamic-throughput-provisioning=hyperdisk-balanced,hyperdisk-throughput,hyperdisk-ml",
-		"--device-in-use-timeout=10s", // Set lower than the usual value to expidite tests
+		"--device-in-use-timeout=10s", // Set lower than the usual value to expedite tests
 		fmt.Sprintf("--fallback-requisite-zones=%s", strings.Join(driverConfig.Zones, ",")),
 	}
 	extra_flags = append(extra_flags, fmt.Sprintf("--compute-endpoint=%s", driverConfig.ComputeEndpoint))

test/sanity/sanity_test.go

@@ -77,7 +77,7 @@ func TestSanity(t *testing.T) {
 	identityServer := driver.NewIdentityServer(gceDriver)
 	controllerServer := driver.NewControllerServer(gceDriver, cloudProvider, 0, 5*time.Minute, fallbackRequisiteZones, enableStoragePools, multiZoneVolumeHandleConfig, listVolumesConfig, provisionableDisksConfig)
 	fakeStatter := mountmanager.NewFakeStatterWithOptions(mounter, mountmanager.FakeStatterOptions{IsBlock: false})
-	nodeServer := driver.NewNodeServer(gceDriver, mounter, deviceUtils, metadataservice.NewFakeService(), fakeStatter)
+	nodeServer := driver.NewNodeServer(gceDriver, mounter, deviceUtils, metadataservice.NewFakeService(), fakeStatter, driver.NodeServerArgs{EnableDeviceInUseTimeout: false, DeviceInUseTimeout: 0})
 	err = gceDriver.SetupGCEDriver(driverName, vendorVersion, extraLabels, nil, identityServer, controllerServer, nodeServer)
 	if err != nil {
 		t.Fatalf("Failed to initialize GCE CSI Driver: %v", err.Error())