Set expiration for keys in deviceInUse cache. Add unit test.

Author: davis-haba

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

@@ -15,15 +15,15 @@ limitations under the License.
 package gceGCEDriver
 
 import (
-	"fmt"
 	"sync"
 	"time"
 
-	lru "github.com/hashicorp/golang-lru/v2"
+	"github.com/hashicorp/golang-lru/v2/expirable"
 	"k8s.io/klog/v2"
 )
 
 // maxDeviceCacheSize specifies the maximum number if in-use devices to cache.
+// 256 was selected since it is twice the number of max PDs per VM (128)
 const maxDeviceCacheSize = 256
 
 // currentTime is used to stub time.Now in unit tests
@@ -34,14 +34,11 @@ var currentTime = time.Now
 type deviceErrMap struct {
 	timeout time.Duration
 	mux     sync.Mutex
-	cache   *lru.Cache[string, time.Time]
+	cache   *expirable.LRU[string, time.Time]
 }
 
 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))
-	}
+	c := expirable.NewLRU[string, time.Time](maxDeviceCacheSize, nil, timeout*2)
 
 	return &deviceErrMap{
 		cache:   c,
@@ -55,8 +52,12 @@ func (devErrMap *deviceErrMap) checkDeviceErrorTimeout(deviceName string) bool {
 	devErrMap.mux.Lock()
 	defer devErrMap.mux.Unlock()
 
-	lastErrTime, exists := devErrMap.cache.Get(deviceName)
-	return exists && currentTime().Sub(lastErrTime).Seconds() >= devErrMap.timeout.Seconds()
+	firstEncounteredErrTime, exists := devErrMap.cache.Get(deviceName)
+	if !exists {
+		return false
+	}
+	expirationTime := firstEncounteredErrTime.Add(devErrMap.timeout)
+	return currentTime().After(expirationTime)
 }
 
 // markDeviceError updates the internal `cache` map to denote an error was encounted

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

@@ -0,0 +1,48 @@
+/*
+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 (
+	"testing"
+	"time"
+)
+
+func TestDeviceErrorMap(t *testing.T) {
+	timeout := time.Second * 10
+	dName := "fake-device"
+	eMap := newDeviceErrMap(timeout)
+	defer func() { currentTime = time.Now }()
+
+	// Register an error. Checking the timeout right after should return false
+	stubCurrentTime(0)
+	eMap.markDeviceError(dName)
+	isTimedOut := eMap.checkDeviceErrorTimeout(dName)
+	if isTimedOut {
+		t.Errorf("checkDeviceErrorTimeout expected to be false if called immediately after marking an error")
+	}
+
+	// Advance time. Checking the timeout should now return true
+	stubCurrentTime(int64(timeout.Seconds()) + 1)
+	isTimedOut = eMap.checkDeviceErrorTimeout(dName)
+	if !isTimedOut {
+		t.Errorf("checkDeviceErrorTimeout expected to be true after waiting for timeout")
+	}
+}
+
+func stubCurrentTime(unixTime int64) {
+	currentTime = func() time.Time {
+		return time.Unix(unixTime, 0)
+	}
+}

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

@@ -471,19 +471,21 @@ func (ns *GCENodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUns
 		return nil, status.Error(codes.Internal, fmt.Sprintf("NodeUnstageVolume failed: %v\nUnmounting arguments: %s\n", err.Error(), stagingTargetPath))
 	}
 
-	if err := ns.confirmDeviceUnused(volumeID); err != nil {
-		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.deviceInUseErrors.checkDeviceErrorTimeout(volumeID) {
-			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)
+	if ns.enableDeviceInUseCheck {
+		if err := ns.confirmDeviceUnused(volumeID); err != nil {
+			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.deviceInUseErrors.checkDeviceErrorTimeout(volumeID) {
+				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)
+			}
 		}
+		ns.deviceInUseErrors.deleteDevice(volumeID)
 	}
 
-	ns.deviceInUseErrors.deleteDevice(volumeID)
 	klog.V(4).Infof("NodeUnstageVolume succeeded on %v from %s", volumeID, stagingTargetPath)
 	return &csi.NodeUnstageVolumeResponse{}, nil
 }
@@ -492,10 +494,6 @@ 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())}