Make deviceInUseErrors atomic. Make deviceInUseTimeout a CLI flag

Author: davis-haba

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

@@ -63,6 +63,7 @@ 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, "Max time to wait for a device to be unsed 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")
@@ -175,6 +176,7 @@ func handle() {
 	defer cancel()
 
 	// Initialize driver
+	driver.DeviceInUseTimeout = *deviceInUseTimeout
 	gceDriver := driver.GetGCEDriver()
 
 	// Initialize identity server

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

@@ -0,0 +1,71 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package gceGCEDriver
+
+import (
+	"sync"
+	"time"
+
+	"k8s.io/klog/v2"
+)
+
+// deviceErrMap is an atomic data datastructure for recording deviceInUseError times
+// for specified devices
+type deviceErrMap struct {
+	mux               sync.Mutex
+	deviceInUseErrors map[string]time.Time
+}
+
+var DeviceInUseTimeout = time.Second * 30
+
+// 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 {
+	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()
+}
+
+// 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) {
+	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()
+		klog.V(4).Infof("Recording in-use error for device %s at time %s", deviceName, now)
+		devErrMap.deviceInUseErrors[deviceName] = now
+	}
+}
+
+// deleteDevice removes a specified device name from the map
+func (devErrMap *deviceErrMap) deleteDevice(deviceName string) {
+	devErrMap.mux.Lock()
+	defer devErrMap.mux.Unlock()
+	delete(devErrMap.deviceInUseErrors, deviceName)
+}

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 map[string]time.Time
+	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
@@ -85,8 +85,6 @@ const (
 	fsTypeExt3                 = "ext3"
 
 	readAheadKBMountFlagRegexPattern = "^read_ahead_kb=(.+)$"
-
-	deviceInUseTimeout = 30
 )
 
 var (
@@ -101,32 +99,6 @@ func getDefaultFsType() string {
 	}
 }
 
-// checkDeviceErrorTimeout returns true an error was encounted for the specified deviceName,
-// where the error happened at least `deviceInUseTimeout` seconds ago.
-func (ns *GCENodeServer) checkDeviceErrorTimeout(deviceName string) bool {
-	if ns.deviceInUseErrors == nil {
-		ns.deviceInUseErrors = make(map[string]time.Time)
-	}
-
-	lastErrTime, exists := ns.deviceInUseErrors[deviceName]
-	return exists && time.Now().Sub(lastErrTime).Seconds() >= deviceInUseTimeout
-}
-
-// markDeviceError updates the internal `deviceInUseErrors` map to denote an error was encounted
-// for the specified deviceName at the current time
-func (ns *GCENodeServer) markDeviceError(deviceName string) {
-	if ns.deviceInUseErrors == nil {
-		ns.deviceInUseErrors = make(map[string]time.Time)
-	}
-
-	// If an earlier error has already been recorded, do not overwrite it
-	if _, exists := ns.deviceInUseErrors[deviceName]; !exists {
-		now := time.Now()
-		klog.V(4).Infof("Recording in-use error for device %s at time %s", deviceName, now)
-		ns.deviceInUseErrors[deviceName] = now
-	}
-}
-
 func (ns *GCENodeServer) isVolumePathMounted(path string) bool {
 	notMnt, err := ns.Mounter.Interface.IsLikelyNotMountPoint(path)
 	klog.V(4).Infof("Checking volume path %s is mounted %t: error %v", path, !notMnt, err)
@@ -489,18 +461,18 @@ func (ns *GCENodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUns
 	}
 
 	if err := ns.confirmDeviceUnused(volumeID); err != nil {
-		var targetErr *ignoreableError
-		if errors.As(err, &targetErr) {
+		var ignoreableErr *ignoreableError
+		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.checkDeviceErrorTimeout(volumeID) {
+		} 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)
 		} else {
-			ns.markDeviceError(volumeID)
+			ns.deviceInUseErrors.markDeviceError(volumeID)
 			return nil, status.Errorf(codes.Internal, "NodeUnstageVolume for volume %s failed: %v", volumeID, err)
 		}
 	}
 
-	delete(ns.deviceInUseErrors, volumeID)
+	ns.deviceInUseErrors.deleteDevice(volumeID)
 	klog.V(4).Infof("NodeUnstageVolume succeeded on %v from %s", volumeID, stagingTargetPath)
 	return &csi.NodeUnstageVolumeResponse{}, nil
 }

test/e2e/utils/utils.go

@@ -68,6 +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=10", // Set lower than the usual value to expidite tests
 		fmt.Sprintf("--fallback-requisite-zones=%s", strings.Join(driverConfig.Zones, ",")),
 	}
 	extra_flags = append(extra_flags, fmt.Sprintf("--compute-endpoint=%s", driverConfig.ComputeEndpoint))