node backoff

Author: saikat-royc

go.mod

@@ -34,6 +34,7 @@ require (
 	github.com/Microsoft/go-winio v0.4.16 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/blang/semver v3.5.1+incompatible // indirect
+	github.com/cenkalti/backoff/v4 v4.1.2 // indirect
 	github.com/cespare/xxhash/v2 v2.1.1 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/fsnotify/fsnotify v1.4.9 // indirect

go.sum

@@ -117,6 +117,8 @@ github.com/blang/semver v3.5.1+incompatible h1:cQNTCjp13qL8KC3Nbxr/y2Bqb63oX6wdn
 github.com/blang/semver v3.5.1+incompatible/go.mod h1:kRBLl5iJ+tD4TcOOxsy/0fnwebNt5EWlYSAyrTnjyyk=
 github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps=
 github.com/bwmarrin/snowflake v0.0.0/go.mod h1:NdZxfVWX+oR6y2K0o6qAYv6gIOP9rjG0/E9WsDpxqwE=
+github.com/cenkalti/backoff/v4 v4.1.2 h1:6Yo7N8UP2K6LWZnW94DLVSSrbobcWdVzAYOisuDPIFo=
+github.com/cenkalti/backoff/v4 v4.1.2/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/certifi/gocertifi v0.0.0-20191021191039-0944d244cd40/go.mod h1:sGbDF6GwGcLpkNXPUTkMRoywsNa/ol15pxFe6ERfguA=
 github.com/certifi/gocertifi v0.0.0-20200922220541-2c3bb06c6054/go.mod h1:sGbDF6GwGcLpkNXPUTkMRoywsNa/ol15pxFe6ERfguA=

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

@@ -22,11 +22,11 @@ import (
 	"testing"
 	"time"
 
+	"context"
+
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
 	"github.com/golang/protobuf/ptypes"
 
-	"context"
-
 	compute "google.golang.org/api/compute/v1"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -1872,20 +1872,16 @@ func TestCreateVolumeDiskReady(t *testing.T) {
 	}
 }
 
-func TestControllerPublishUnpublishVolume(t *testing.T) {
+func TestControllerPublishUnpublishBackoff(t *testing.T) {
 	testCases := []struct {
-		name              string
-		seedDisks         []*gce.CloudDisk
-		pubReq            *csi.ControllerPublishVolumeRequest
-		unpubReq          *csi.ControllerUnpublishVolumeRequest
-		errorSeenOnNode   bool
-		fakeCloudProvider bool
+		name     string
+		node     string
+		pubReq   *csi.ControllerPublishVolumeRequest
+		unpubReq *csi.ControllerUnpublishVolumeRequest
 	}{
 		{
-			name: "queue up publish requests if node has publish error",
-			seedDisks: []*gce.CloudDisk{
-				createZonalCloudDisk(name),
-			},
+			name: "controller publish backoff",
+			node: testNodeID,
 			pubReq: &csi.ControllerPublishVolumeRequest{
 				VolumeId: testVolumeID,
 				NodeId:   testNodeID,
@@ -1898,159 +1894,58 @@ func TestControllerPublishUnpublishVolume(t *testing.T) {
 					},
 				},
 			},
-			errorSeenOnNode:   true,
-			fakeCloudProvider: false,
 		},
 		{
-			name: "queue up and process publish requests if node has publish error",
-			seedDisks: []*gce.CloudDisk{
-				createZonalCloudDisk(name),
-			},
-			pubReq: &csi.ControllerPublishVolumeRequest{
-				VolumeId: testVolumeID,
-				NodeId:   testNodeID,
-				VolumeCapability: &csi.VolumeCapability{
-					AccessType: &csi.VolumeCapability_Mount{
-						Mount: &csi.VolumeCapability_MountVolume{},
-					},
-					AccessMode: &csi.VolumeCapability_AccessMode{
-						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
-					},
-				},
-			},
-			errorSeenOnNode:   true,
-			fakeCloudProvider: true,
-		},
-		{
-			name: "do not queue up publish requests if node doesn't have publish error",
-			seedDisks: []*gce.CloudDisk{
-				createZonalCloudDisk(name),
-			},
-			pubReq: &csi.ControllerPublishVolumeRequest{
-				VolumeId: testVolumeID,
-				NodeId:   testNodeID,
-				VolumeCapability: &csi.VolumeCapability{
-					AccessType: &csi.VolumeCapability_Mount{
-						Mount: &csi.VolumeCapability_MountVolume{},
-					},
-					AccessMode: &csi.VolumeCapability_AccessMode{
-						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
-					},
-				},
-			},
-			errorSeenOnNode:   false,
-			fakeCloudProvider: false,
-		},
-		{
-			name: "queue up unpublish requests if node has publish error",
-			seedDisks: []*gce.CloudDisk{
-				createZonalCloudDisk(name),
-			},
-			unpubReq: &csi.ControllerUnpublishVolumeRequest{
-				VolumeId: testVolumeID,
-				NodeId:   testNodeID,
-			},
-			errorSeenOnNode:   true,
-			fakeCloudProvider: false,
-		},
-		{
-			name: "queue up and process unpublish requests if node has publish error",
-			seedDisks: []*gce.CloudDisk{
-				createZonalCloudDisk(name),
-			},
+			name: "controller unpublish backoff",
+			node: testNodeID,
 			unpubReq: &csi.ControllerUnpublishVolumeRequest{
 				VolumeId: testVolumeID,
 				NodeId:   testNodeID,
 			},
-			errorSeenOnNode:   true,
-			fakeCloudProvider: true,
-		},
-		{
-			name: "do not queue up unpublish requests if node doesn't have publish error",
-			seedDisks: []*gce.CloudDisk{
-				createZonalCloudDisk(name),
-			},
-			unpubReq: &csi.ControllerUnpublishVolumeRequest{
-				VolumeId: testVolumeID,
-				NodeId:   testNodeID,
-			},
-			errorSeenOnNode:   false,
-			fakeCloudProvider: false,
 		},
 	}
 	for _, tc := range testCases {
-		t.Logf("test case: %s", tc.name)
-
-		var gceDriver *GCEDriver
-
-		if tc.fakeCloudProvider {
-			fcp, err := gce.CreateFakeCloudProvider(project, zone, tc.seedDisks)
+		t.Run(tc.name, func(t *testing.T) {
+			fcp, err := gce.CreateFakeCloudProvider(project, zone, nil)
 			if err != nil {
-				t.Fatalf("Failed to create fake cloud provider: %v", err)
-			}
-
-			instance := &compute.Instance{
-				Name:  node,
-				Disks: []*compute.AttachedDisk{},
+				t.Errorf("Failed to create fake cloud provider: %v", err)
 			}
-			fcp.InsertInstance(instance, zone, node)
-
-			// Setup new driver each time so no interference
-			gceDriver = initGCEDriverWithCloudProvider(t, fcp)
-		} else {
-			gceDriver = initGCEDriver(t, tc.seedDisks)
-		}
-		gceDriver.cs.opsManager.ready = true
-		// mark the node in the map
-		if tc.errorSeenOnNode {
-			gceDriver.cs.publishErrorsSeenOnNode[testNodeID] = true
-		}
-
-		requestCount := 50
-		for i := 0; i < requestCount; i++ {
+			driver := initGCEDriverWithCloudProvider(t, fcp)
+			driver.cs.opsManager.ready = true
+			// Simulate node with active backoff condition.
+			driver.cs.nodeBackoff.AddOrUpdateNodeBackoffEntry(tc.node)
+			boff := driver.cs.nodeBackoff.backoff[tc.node]
+			if boff.expBackoff.MaxElapsedTime != 0 {
+				t.Errorf("unexpected backoff object %+v", boff)
+			}
+			// Setup a large next retry time, to allow sufficient time for the controller publish/unpublish retries in this test case.
+			mockNextRetryTime := time.Now().Add(time.Minute)
+			boff.nextRetryTime = mockNextRetryTime
+			driver.cs.nodeBackoff.backoff[tc.node].nextRetryTime = mockNextRetryTime
 			if tc.pubReq != nil {
-				gceDriver.cs.ControllerPublishVolume(context.Background(), tc.pubReq)
+				for i := 0; i < 10; i++ {
+					_, err := driver.cs.ControllerPublishVolume(context.Background(), tc.pubReq)
+					if err == nil {
+						t.Errorf("expected error but found none")
+					}
+					if !isUnavailableError(err) {
+						t.Errorf("got error %v found, want Unavailable", err)
+					}
+				}
 			}
 
 			if tc.unpubReq != nil {
-				gceDriver.cs.ControllerUnpublishVolume(context.Background(), tc.unpubReq)
-			}
-		}
-
-		queued := false
-
-		if tc.errorSeenOnNode {
-			if err := wait.Poll(10*time.Nanosecond, 1*time.Second, func() (bool, error) {
-				if gceDriver.cs.queue.Len() > 0 {
-					queued = true
-
-					if tc.fakeCloudProvider {
-						gceDriver.cs.Run()
+				for i := 0; i < 10; i++ {
+					_, err := driver.cs.ControllerUnpublishVolume(context.Background(), tc.unpubReq)
+					if err == nil {
+						t.Errorf("expected error but found none")
+					}
+					if !isUnavailableError(err) {
+						t.Errorf("got error %v found, want Unavailable", err)
 					}
-				}
-
-				// Items are queued up and eventually all processed
-				if tc.fakeCloudProvider {
-					return queued && gceDriver.cs.queue.Len() == 0, nil
-				}
-
-				return gceDriver.cs.queue.Len() == requestCount, nil
-			}); err != nil {
-				if tc.fakeCloudProvider {
-					t.Fatalf("%v requests not processed for node has seen error", gceDriver.cs.queue.Len())
-				} else {
-					t.Fatalf("Only %v requests queued up for node has seen error", gceDriver.cs.queue.Len())
 				}
 			}
-		}
-
-		if !tc.errorSeenOnNode {
-			if err := wait.Poll(10*time.Nanosecond, 10*time.Millisecond, func() (bool, error) {
-				return gceDriver.cs.queue.Len() != 0, nil
-			}); err == nil {
-				t.Fatalf("%v requests queued up for node hasn't seen error", gceDriver.cs.queue.Len())
-			}
-		}
+		})
 	}
 }
 
@@ -3112,3 +3007,68 @@ func TestHydrateCache(t *testing.T) {
 		})
 	}
 }
+
+func TestBackoff(t *testing.T) {
+	driver := initGCEDriver(t, nil)
+	cs := driver.cs
+	testNodeID = "testnode"
+	// testcase1: Add a new node key to the backoff map and check that retry time is not updated on failures within the next retry time.
+	// Empty backoff map.
+	safe, _ := cs.nodeBackoff.IsSafetoRetry(testNodeID)
+	if !safe {
+		t.Errorf("unexpected retry %v", safe)
+	}
+
+	cs.nodeBackoff.AddOrUpdateNodeBackoffEntry(testNodeID)
+	boff := cs.nodeBackoff.backoff[testNodeID]
+	if boff.expBackoff.MaxElapsedTime != 0 {
+		t.Errorf("unexpected backoff object %+v", boff)
+	}
+
+	// Setup a large next retry time, and try to update the backoff object. this should be a no-op.
+	mockNextRetryTime := time.Now().Add(time.Minute)
+	boff.nextRetryTime = mockNextRetryTime
+	cs.nodeBackoff.backoff[testNodeID] = boff
+	// Simulate a scenario where after setting a retry time due to the first failure, all subsequent failures within the next retry time should not update the retry time again.
+	for i := 0; i < 20; i++ {
+		cs.nodeBackoff.AddOrUpdateNodeBackoffEntry(testNodeID)
+	}
+	if !boff.nextRetryTime.Equal(mockNextRetryTime) {
+		t.Errorf("unexpected retry time, got %s, want %s", boff.nextRetryTime.String(), mockNextRetryTime.String())
+	}
+
+	// Simulate success and remove the node key from the backoff map.
+	cs.nodeBackoff.DeleteNodeBackoffEntry(testNodeID)
+	if _, ok := cs.nodeBackoff.backoff[testNodeID]; ok {
+		t.Errorf("unexpected key %s", testNodeID)
+	}
+
+	// testcase 2: Retry updating the backoff map after the next retry time has elapsed.
+	cs.nodeBackoff.AddOrUpdateNodeBackoffEntry(testNodeID)
+	initialRetryTime := cs.nodeBackoff.backoff[testNodeID].nextRetryTime
+	// Mock a failure after the retry time has elapsed. This should update the retry time.
+	time.Sleep(2 * time.Second)
+	cs.nodeBackoff.AddOrUpdateNodeBackoffEntry(testNodeID)
+	newRetryTime := cs.nodeBackoff.backoff[testNodeID].nextRetryTime
+	if !newRetryTime.After(initialRetryTime) {
+		t.Errorf("unexpected retry time: boff1 %s, boff2 %s", initialRetryTime.String(), newRetryTime.String())
+	}
+	// Simulate success and remove the node key from the backoff map.
+	cs.nodeBackoff.DeleteNodeBackoffEntry(testNodeID)
+	if _, ok := cs.nodeBackoff.backoff[testNodeID]; ok {
+		t.Errorf("unexpected key %s", testNodeID)
+	}
+}
+
+func isUnavailableError(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	st, ok := status.FromError(err)
+	if !ok {
+		return false
+	}
+
+	return st.Code().String() == "Unavailable"
+}

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

@@ -20,7 +20,6 @@ import (
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
-	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog"
 	"k8s.io/mount-utils"
 	common "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
@@ -151,13 +150,12 @@ func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, devi
 
 func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute) *GCEControllerServer {
 	return &GCEControllerServer{
-		Driver:                  gceDriver,
-		CloudProvider:           cloudProvider,
-		seen:                    map[string]int{},
-		volumeLocks:             common.NewVolumeLocks(),
-		queue:                   workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "controllerserver"),
-		publishErrorsSeenOnNode: map[string]bool{},
-		opsManager:              NewOpsManager(cloudProvider),
+		Driver:        gceDriver,
+		CloudProvider: cloudProvider,
+		seen:          map[string]int{},
+		volumeLocks:   common.NewVolumeLocks(),
+		nodeBackoff:   NewNodeBackoffInfo(),
+		opsManager:    NewOpsManager(cloudProvider),
 	}
 }
 

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

@@ -23,7 +23,9 @@ import (
 	"context"
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
+
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/status"
 	"k8s.io/klog"
 )
 
@@ -199,3 +201,16 @@ func collectMountOptions(fsType string, mntFlags []string) []string {
 	}
 	return options
 }
+
+func isInternalError(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	st, ok := status.FromError(err)
+	if !ok {
+		return false
+	}
+
+	return st.Code().String() == "Internal"
+}