Use original error code when backing off

Change-Id: If514a26bf513fc47de6e74615f352a83268f5ecf

Author: mattcary

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

@@ -92,10 +92,12 @@ type GCEControllerServer struct {
 	errorBackoff *csiErrorBackoff
 }
 
+type csiErrorBackoffId string
+
 type csiErrorBackoff struct {
-	backoff *flowcontrol.Backoff
+	backoff    *flowcontrol.Backoff
+	errorCodes map[csiErrorBackoffId]codes.Code
 }
-type csiErrorBackoffId string
 
 type workItem struct {
 	ctx          context.Context
@@ -496,13 +498,13 @@ func (gceCS *GCEControllerServer) ControllerPublishVolume(ctx context.Context, r
 
 	backoffId := gceCS.errorBackoff.backoffId(req.NodeId, req.VolumeId)
 	if gceCS.errorBackoff.blocking(backoffId) {
-		return nil, status.Errorf(codes.Unavailable, "ControllerPublish not permitted on node %q due to backoff condition", req.NodeId)
+		return nil, status.Errorf(gceCS.errorBackoff.code(backoffId), "ControllerPublish not permitted on node %q due to backoff condition", req.NodeId)
 	}
 
 	resp, err, diskTypeForMetric := gceCS.executeControllerPublishVolume(ctx, req)
 	if err != nil {
-		klog.Infof("For node %s adding backoff due to error for volume %s: %v", req.NodeId, req.VolumeId, err.Error())
-		gceCS.errorBackoff.next(backoffId)
+		klog.Infof("For node %s adding backoff due to error for volume %s: %v", req.NodeId, req.VolumeId, err)
+		gceCS.errorBackoff.next(backoffId, common.CodeForError(err))
 	} else {
 		klog.Infof("For node %s clear backoff due to successful publish of volume %v", req.NodeId, req.VolumeId)
 		gceCS.errorBackoff.reset(backoffId)
@@ -661,12 +663,12 @@ func (gceCS *GCEControllerServer) ControllerUnpublishVolume(ctx context.Context,
 	// Only valid requests will be queued
 	backoffId := gceCS.errorBackoff.backoffId(req.NodeId, req.VolumeId)
 	if gceCS.errorBackoff.blocking(backoffId) {
-		return nil, status.Errorf(codes.Unavailable, "ControllerUnpublish not permitted on node %q due to backoff condition", req.NodeId)
+		return nil, status.Errorf(gceCS.errorBackoff.code(backoffId), "ControllerUnpublish not permitted on node %q due to backoff condition", req.NodeId)
 	}
 	resp, err, diskTypeForMetric := gceCS.executeControllerUnpublishVolume(ctx, req)
 	if err != nil {
-		klog.Infof("For node %s adding backoff due to error for volume %s", req.NodeId, req.VolumeId)
-		gceCS.errorBackoff.next(backoffId)
+		klog.Infof("For node %s adding backoff due to error for volume %s: %v", req.NodeId, req.VolumeId, err)
+		gceCS.errorBackoff.next(backoffId, common.CodeForError(err))
 	} else {
 		klog.Infof("For node %s clear backoff due to successful unpublish of volume %v", req.NodeId, req.VolumeId)
 		gceCS.errorBackoff.reset(backoffId)
@@ -1816,7 +1818,7 @@ func pickRandAndConsecutive(slice []string, n int) ([]string, error) {
 }
 
 func newCsiErrorBackoff(initialDuration, errorBackoffMaxDuration time.Duration) *csiErrorBackoff {
-	return &csiErrorBackoff{flowcontrol.NewBackOff(initialDuration, errorBackoffMaxDuration)}
+	return &csiErrorBackoff{flowcontrol.NewBackOff(initialDuration, errorBackoffMaxDuration), make(map[csiErrorBackoffId]codes.Code)}
 }
 
 func (_ *csiErrorBackoff) backoffId(nodeId, volumeId string) csiErrorBackoffId {
@@ -1828,10 +1830,27 @@ func (b *csiErrorBackoff) blocking(id csiErrorBackoffId) bool {
 	return blk
 }
 
-func (b *csiErrorBackoff) next(id csiErrorBackoffId) {
+func (b *csiErrorBackoff) code(id csiErrorBackoffId) codes.Code {
+	if code, ok := b.errorCodes[id]; ok {
+		return code
+	}
+	// If we haven't recorded a code, return unavailable, which signals a problem with the driver
+	// (ie, it's not filtered out as a user problem).
+	klog.Errorf("using default code for %s", id)
+	return codes.Unavailable
+}
+
+func (b *csiErrorBackoff) next(id csiErrorBackoffId, code *codes.Code) {
 	b.backoff.Next(string(id), b.backoff.Clock.Now())
+	if code != nil {
+		b.errorCodes[id] = *code
+	} else {
+		// Use unavailable, which is our default for a backoff error.
+		b.errorCodes[id] = codes.Unavailable
+	}
 }
 
 func (b *csiErrorBackoff) reset(id csiErrorBackoffId) {
 	b.backoff.Reset(string(id))
+	delete(b.errorCodes, id)
 }

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

@@ -257,7 +257,6 @@ func TestDeleteSnapshot(t *testing.T) {
 		_, err := gceDriver.cs.DeleteSnapshot(context.Background(), tc.req)
 		if err != nil {
 			serverError, ok := status.FromError(err)
-			t.Logf("get server error %v", serverError)
 			if !ok {
 				t.Fatalf("Could not get error status code from err: %v", serverError)
 			}
@@ -2970,7 +2969,7 @@ type backoffDriverConfig struct {
 
 func newFakeCSIErrorBackoff(tc *clock.FakeClock) *csiErrorBackoff {
 	backoff := flowcontrol.NewFakeBackOff(errorBackoffInitialDuration, errorBackoffMaxDuration, tc)
-	return &csiErrorBackoff{backoff}
+	return &csiErrorBackoff{backoff, make(map[csiErrorBackoffId]codes.Code)}
 }
 
 func TestControllerUnpublishBackoff(t *testing.T) {
@@ -3010,7 +3009,8 @@ func TestControllerUnpublishBackoff(t *testing.T) {
 			}
 
 			// Mock an active backoff condition on the node.
-			driver.cs.errorBackoff.next(backoffId)
+			errorCode := codes.Unavailable
+			driver.cs.errorBackoff.next(backoffId, &errorCode)
 
 			tc.config.clock.Step(step)
 			// A requst for a a different volume should succeed. This volume is not
@@ -3028,7 +3028,9 @@ func TestControllerUnpublishBackoff(t *testing.T) {
 				VolumeId: testVolumeID,
 				NodeId:   testNodeID,
 			}
-			// For the first 199 ms, the backoff condition is true. All controller publish request will be denied with 'Unavailable' error code.
+			// For the first 199 ms, the backoff condition is true. All controller publish
+			// request will be denied with the same unavailable error code as was set on
+			// the original error.
 			for i := 0; i < 199; i++ {
 				var err error
 				_, err = driver.cs.ControllerUnpublishVolume(context.Background(), unpubreq)
@@ -3057,13 +3059,16 @@ func TestControllerUnpublishBackoff(t *testing.T) {
 				t.Errorf("expected error")
 			}
 
-			// The above failure should cause driver to call Backoff.Next() again and a backoff duration of 400 ms duration is set starting at the 200th millisecond.
-			// For the 200-599 ms, the backoff condition is true, and new controller publish requests will be deined.
+			// The above failure should cause driver to call backoff.next() again and a
+			// backoff duration of 400 ms duration is set starting at the 200th
+			// millisecond.  For the 200-599 ms, the backoff condition is true, with an
+			// internal error this time, and new controller publish requests will be
+			// deined.
 			for i := 0; i < 399; i++ {
 				tc.config.clock.Step(step)
 				var err error
 				_, err = driver.cs.ControllerUnpublishVolume(context.Background(), unpubreq)
-				if !isUnavailableError(err) {
+				if !isInternalError(err) {
 					t.Errorf("unexpected error %v", err)
 				}
 			}
@@ -3139,8 +3144,9 @@ func TestControllerPublishBackoff(t *testing.T) {
 			backoffId := driver.cs.errorBackoff.backoffId(testNodeID, testVolumeID)
 			step := 1 * time.Millisecond
 
-			// Mock an active backoff condition on the node.
-			driver.cs.errorBackoff.next(backoffId)
+			// Mock an active bakcoff condition on the node.
+			errorCode := codes.Unavailable
+			driver.cs.errorBackoff.next(backoffId, &errorCode)
 
 			// A detach request for a different disk should succeed. As this disk is not
 			// on the instance, the detach will succeed without calling the gce detach
@@ -3213,13 +3219,16 @@ func TestControllerPublishBackoff(t *testing.T) {
 				t.Errorf("expected error")
 			}
 
-			// The above failure should cause driver to call Backoff.Next() again and a backoff duration of 400 ms duration is set starting at the 200th millisecond.
-			// For the 200-599 ms, the backoff condition is true, and new controller publish requests will be deined.
+			// The above failure should cause driver to call backoff.next() again and a
+			// backoff duration of 400 ms duration is set starting at the 200th
+			// millisecond.  For the 200-599 ms, the backoff condition is true, with an
+			// internal error this time, and new controller publish requests will be
+			// deined.
 			for i := 0; i < 399; i++ {
 				tc.config.clock.Step(step)
 				var err error
 				_, err = driver.cs.ControllerPublishVolume(context.Background(), pubreq)
-				if !isUnavailableError(err) {
+				if !isInternalError(err) {
 					t.Errorf("unexpected error %v", err)
 				}
 			}
@@ -3363,3 +3372,16 @@ func isUnavailableError(err error) bool {
 
 	return st.Code().String() == "Unavailable"
 }
+
+func isInternalError(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	st, ok := status.FromError(err)
+	if !ok {
+		return false
+	}
+
+	return st.Code().String() == "Internal"
+}