1. use flowcontrol backoff lib for publish/unpublish backoff

2. backoff on resource exhausted error

Author: saikat-royc

pkg/gce-cloud-provider/compute/fake-gce.go

@@ -17,6 +17,7 @@ package gcecloudprovider
 import (
 	"context"
 	"fmt"
+	"net/http"
 	"strconv"
 	"strings"
 
@@ -419,8 +420,9 @@ func (cloud *FakeCloudProvider) UpdateDiskStatus(s string) {
 }
 
 type Signal struct {
-	ReportError   bool
-	ReportRunning bool
+	ReportError                bool
+	ReportRunning              bool
+	ReportTooManyRequestsError bool
 }
 
 type FakeBlockingCloudProvider struct {
@@ -446,6 +448,9 @@ func (cloud *FakeBlockingCloudProvider) WaitForZonalOp(ctx context.Context, proj
 	if val.ReportError {
 		return fmt.Errorf("force mock error of zonal op %s", opName)
 	}
+	if val.ReportTooManyRequestsError {
+		return tooManyRequestsError()
+	}
 	return nil
 }
 
@@ -483,6 +488,12 @@ func invalidError() *googleapi.Error {
 	}
 }
 
+func tooManyRequestsError() *googleapi.Error {
+	return &googleapi.Error{
+		Code: http.StatusTooManyRequests,
+	}
+}
+
 func (cloud *FakeCloudProvider) StartAttachDiskOp(ctx context.Context, volKey *meta.Key, readWrite, diskType, project, location, instanceName string) (*computev1.Operation, error) {
 	source := cloud.GetDiskSourceURI(project, volKey)
 	attachedDiskV1 := &computev1.AttachedDisk{

pkg/gce-cloud-provider/compute/gce-compute.go

@@ -716,7 +716,7 @@ func (cloud *CloudProvider) WaitForAttach(ctx context.Context, project string, v
 		klog.V(6).Infof("Polling for attach of disk %v to instance %v to complete for %v", volKey.Name, instanceName, time.Since(start))
 		disk, err := cloud.GetDisk(ctx, project, volKey, GCEAPIVersionV1)
 		if err != nil {
-			return false, fmt.Errorf("GetDisk failed to get disk: %v", err)
+			return false, err
 		}
 
 		if disk == nil {
@@ -953,7 +953,7 @@ func (cloud *CloudProvider) CheckZonalOpDoneStatus(ctx context.Context, project,
 	lastKnownOp, err := cloud.service.ZoneOperations.Get(project, location, opId).Context(ctx).Do()
 	if err != nil {
 		if !IsGCENotFoundError(err) {
-			return false, fmt.Errorf("failed to get operation %s: %v", opId, err)
+			return false, err
 		}
 		return true, nil
 	}

pkg/gce-cloud-provider/compute/gce.go

@@ -263,3 +263,12 @@ func IsGCENotFoundError(err error) bool {
 func IsGCEInvalidError(err error) bool {
 	return IsGCEError(err, "invalid")
 }
+
+// IsGCENotFoundError returns true if the error is a googleapi.Error with
+// resource exhausted error code.
+func IsTooManyRequestError(err error) bool {
+	if apierr, ok := err.(*googleapi.Error); ok && apierr.Code == http.StatusTooManyRequests {
+		return true
+	}
+	return false
+}