Use LRU for deviceInUse map. Disable device in use checks entirely if specified to do so

Author: davis-haba

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

@@ -64,8 +64,8 @@ var (
 	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")
 
-	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")
+	enableDeviceInUseCheck = flag.Bool("enable-device-in-use-check-on-node-unstage", true, "If set to true, block NodeUnstageVolume requests until the specified device is not in use")
+	deviceInUseTimeout     = flag.Duration("device-in-use-timeout", 30*time.Second, "Max time to wait for a device to be unused when attempting to unstage. Exceeding the timeout will cause an unstage request to return success and ignore the device in use check.")
 
 	maxProcs                = flag.Int("maxprocs", 1, "GOMAXPROCS override")
 	maxConcurrentFormat     = flag.Int("max-concurrent-format", 1, "The maximum number of concurrent format exec calls")
@@ -244,8 +244,8 @@ func handle() {
 			klog.Fatalf("Failed to set up metadata service: %v", err.Error())
 		}
 		nsArgs := driver.NodeServerArgs{
-			EnableDeviceInUseTimeout: *enableDeviceInUseTimeout,
-			DeviceInUseTimeout:       *deviceInUseTimeout,
+			EnableDeviceInUseCheck: *enableDeviceInUseCheck,
+			DeviceInUseTimeout:     *deviceInUseTimeout,
 		}
 		nodeServer = driver.NewNodeServer(gceDriver, mounter, deviceUtils, meta, statter, nsArgs)
 		if *maxConcurrentFormatAndMount > 0 {

go.mod

@@ -76,6 +76,7 @@ require (
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 // indirect
 	github.com/hashicorp/errwrap v1.0.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.0 // indirect
+	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/imdario/mergo v0.3.12 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect

go.sum

@@ -1550,6 +1550,8 @@ github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.3/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
+github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
+github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64=
 github.com/hashicorp/mdns v1.0.0/go.mod h1:tL+uN++7HEJ6SQLQ2/p+z2pH24WQKWjBPkE0mNTz8vQ=

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

@@ -15,69 +15,68 @@ limitations under the License.
 package gceGCEDriver
 
 import (
+	"fmt"
 	"sync"
 	"time"
 
+	lru "github.com/hashicorp/golang-lru/v2"
 	"k8s.io/klog/v2"
 )
 
+// maxDeviceCacheSize specifies the maximum number if in-use devices to cache.
+const maxDeviceCacheSize = 256
+
+// currentTime is used to stub time.Now in unit tests
+var currentTime = time.Now
+
 // 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
+	timeout time.Duration
+	mux     sync.Mutex
+	cache   *lru.Cache[string, time.Time]
 }
 
-func newDeviceErrMap(shouldEnable bool, timeout time.Duration) *deviceErrMap {
+func newDeviceErrMap(timeout time.Duration) *deviceErrMap {
+	c, err := lru.New[string, time.Time](maxDeviceCacheSize)
+	if err != nil {
+		panic(fmt.Sprintf("Could not initialize deviceInUse LRU cache: %s", err))
+	}
+
 	return &deviceErrMap{
-		deviceInUseErrors: make(map[string]time.Time),
-		enabled:           shouldEnable,
-		timeout:           timeout,
+		cache:   c,
+		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()
 
-	lastErrTime, exists := devErrMap.deviceInUseErrors[deviceName]
-	return exists && time.Now().Sub(lastErrTime).Seconds() >= devErrMap.timeout.Seconds()
+	lastErrTime, exists := devErrMap.cache.Get(deviceName)
+	return exists && currentTime().Sub(lastErrTime).Seconds() >= devErrMap.timeout.Seconds()
 }
 
-// markDeviceError updates the internal `deviceInUseErrors` map to denote an error was encounted
+// markDeviceError updates the internal `cache` 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 an earlier error has already been recorded, do not overwrite it
-	if _, exists := devErrMap.deviceInUseErrors[deviceName]; !exists {
-		now := time.Now()
+	if _, exists := devErrMap.cache.Get(deviceName); !exists {
+		now := currentTime()
 		klog.V(4).Infof("Recording in-use error for device %s at time %s", deviceName, now)
-		devErrMap.deviceInUseErrors[deviceName] = now
+		devErrMap.cache.Add(deviceName, now)
 	}
 }
 
 // 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)
+	devErrMap.cache.Remove(deviceName)
 }

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

@@ -147,13 +147,14 @@ func NewIdentityServer(gceDriver *GCEDriver) *GCEIdentityServer {
 
 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,
-		deviceInUseErrors: newDeviceErrMap(args.EnableDeviceInUseTimeout, args.DeviceInUseTimeout),
+		Driver:                 gceDriver,
+		Mounter:                mounter,
+		DeviceUtils:            deviceUtils,
+		MetadataService:        meta,
+		volumeLocks:            common.NewVolumeLocks(),
+		VolumeStatter:          statter,
+		enableDeviceInUseCheck: args.EnableDeviceInUseCheck,
+		deviceInUseErrors:      newDeviceErrMap(args.DeviceInUseTimeout),
 	}
 }
 

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

@@ -51,6 +51,9 @@ type GCENodeServer struct {
 	// for that same volume (as defined by VolumeID) return an Aborted error
 	volumeLocks *common.VolumeLocks
 
+	// enableDeviceInUseCheck, if true, will block NodeUnstageVolume requests if the specified
+	// device is still in use (or until --device-in-use-timeout is reached, if specified)
+	enableDeviceInUseCheck bool
 	// deviceInUseErrors keeps tracks of device names and a timestamp for when an error is
 	// encounted for that device
 	deviceInUseErrors *deviceErrMap
@@ -70,10 +73,9 @@ type GCENodeServer struct {
 }
 
 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
+	// EnableDeviceInUseCheck enables functionality which will block NodeUnstageVolume request
+	// until the device is not in use
+	EnableDeviceInUseCheck bool
 
 	DeviceInUseTimeout time.Duration
 }
@@ -490,6 +492,10 @@ func (ns *GCENodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUns
 type ignoreableError struct{ error }
 
 func (ns *GCENodeServer) confirmDeviceUnused(volumeID string) error {
+	if !ns.enableDeviceInUseCheck {
+		return nil
+	}
+
 	devicePath, err := getDevicePath(ns, volumeID, "" /* partition, which is unused */)
 	if err != nil {
 		return &ignoreableError{fmt.Errorf("failed to find device path for volume %s: %v", volumeID, err.Error())}

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), NodeServerArgs{false, 0})
+	nodeServer := NewNodeServer(gceDriver, mounter, deviceUtils, metaService, mountmanager.NewFakeStatter(mounter), NodeServerArgs{true, 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), NodeServerArgs{false, 0})
+	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(false), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter), NodeServerArgs{true, 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), NodeServerArgs{false, 0}).WithSerializedFormatAndMount(5*time.Second, 1)
+	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(false), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter), NodeServerArgs{true, 0}).WithSerializedFormatAndMount(5*time.Second, 1)
 
 	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, nil, nil, nodeServer)
 	if err != nil {

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, driver.NodeServerArgs{EnableDeviceInUseTimeout: false, DeviceInUseTimeout: 0})
+	nodeServer := driver.NewNodeServer(gceDriver, mounter, deviceUtils, metadataservice.NewFakeService(), fakeStatter, driver.NodeServerArgs{EnableDeviceInUseCheck: true, 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())