added retry loop to GCS client

storage/gcs/errors.go

@@ -0,0 +1,68 @@
+package gcs
+
+import (
+	"log"
+	"math/rand"
+	"net/url"
+	"time"
+
+	"google.golang.org/api/googleapi"
+)
+
+const (
+	maxSleep      = 15 * time.Second
+	maxRetryCount = 20
+)
+
+func retry(desc string, f func() (interface{}, error)) (interface{}, error) {
+	v, err := f()
+	nextSleep := 500 * time.Millisecond
+	multiplier := 1.5 + rand.Float32()/2
+	retryCount := 0
+	for shouldRetry(err) && retryCount < maxRetryCount {
+		retryCount++
+		log.Printf("Got error when calling %v. Retrying (%v/%v)...", desc, retryCount, maxRetryCount)
+		time.Sleep(nextSleep)
+		nextSleep = time.Duration(float32(nextSleep) * multiplier)
+		if nextSleep > maxSleep {
+			nextSleep = maxSleep
+		}
+		v, err = f()
+	}
+	return v, err
+}
+
+func shouldRetry(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	if _, ok := err.(*url.Error); ok {
+		return true
+	}
+
+	if ge, ok := err.(*googleapi.Error); ok {
+		if ge.Code >= 500 {
+			return true
+		}
+
+		return false
+	}
+
+	log.Printf("Got non-retriable error: %#+v", err)
+
+	return false
+}
+
+func isGoogleAPIError(err error, code int) bool {
+	if err == nil {
+		return false
+	}
+	if err, ok := err.(*googleapi.Error); ok {
+		if err.Code == code {
+			return true
+		}
+	}
+
+	return false
+}

storage/gcs/gcs_storage.go

@@ -40,35 +40,44 @@ type gcsStorage struct {
 }
 
 func (gcs *gcsStorage) BlockExists(b string) (bool, error) {
-	_, err := gcs.objectsService.Get(gcs.BucketName, gcs.getObjectNameString(b)).Do()
+	call := gcs.objectsService.Get(gcs.BucketName, gcs.getObjectNameString(b))
+	_, err := retry(
+		"Get",
+		func() (interface{}, error) {
+			return call.Do()
+		})
 
 	if err == nil {
 		return true, nil
 	}
-	if err, ok := err.(*googleapi.Error); ok {
-		if err.Code == http.StatusNotFound {
-			return false, nil
-		}
+
+	if isGoogleAPIError(err, http.StatusNotFound) {
+		return false, nil
 	}
 
 	return false, err
 }
 
 func (gcs *gcsStorage) GetBlock(b string) ([]byte, error) {
-	v, err := gcs.objectsService.Get(gcs.BucketName, gcs.getObjectNameString(b)).Download()
-	if err != nil {
-		if err, ok := err.(*googleapi.Error); ok {
-			if err.Code == http.StatusNotFound {
-				return nil, storage.ErrBlockNotFound
-			}
-		}
+	call := gcs.objectsService.Get(gcs.BucketName, gcs.getObjectNameString(b))
+	v, err := retry(
+		"Get",
+		func() (interface{}, error) {
+			return call.Download()
+		})
+	if isGoogleAPIError(err, http.StatusNotFound) {
+		return nil, storage.ErrBlockNotFound
+	}
 
+	if err != nil {
 		return nil, fmt.Errorf("unable to get block '%s': %v", b, err)
 	}
 
-	defer v.Body.Close()
+	dl := v.(*http.Response)
 
-	return ioutil.ReadAll(v.Body)
+	defer dl.Body.Close()
+
+	return ioutil.ReadAll(dl.Body)
 }
 
 func (gcs *gcsStorage) PutBlock(b string, data []byte, options storage.PutOptions) error {
@@ -86,22 +95,28 @@ func (gcs *gcsStorage) PutBlock(b string, data []byte, options storage.PutOption
 		// To avoid the race, check this server-side.
 		call = call.IfGenerationMatch(0)
 	}
-	_, err := call.Do()
 
-	if err != nil {
-		if err, ok := err.(*googleapi.Error); ok {
-			if err.Code == http.StatusPreconditionFailed {
-				// Condition not met indicates that the block already exists.
-				return nil
-			}
-		}
+	_, err := retry(
+		"Insert",
+		func() (interface{}, error) {
+			return call.Do()
+		})
+
+	if isGoogleAPIError(err, http.StatusPreconditionFailed) {
+		// Condition not met indicates that the block already exists.
+		return nil
 	}
 
 	return err
 }
 
 func (gcs *gcsStorage) DeleteBlock(b string) error {
-	err := gcs.objectsService.Delete(gcs.BucketName, string(b)).Do()
+	call := gcs.objectsService.Delete(gcs.BucketName, string(b))
+	_, err := retry(
+		"Delete",
+		func() (interface{}, error) {
+			return call.Do(), nil
+		})
 	if err != nil {
 		return fmt.Errorf("unable to delete block %s: %v", b, err)
 	}
@@ -118,13 +133,24 @@ func (gcs *gcsStorage) ListBlocks(prefix string) chan (storage.BlockMetadata) {
 
 	go func() {
 		ps := gcs.getObjectNameString(prefix)
-		page, _ := gcs.objectsService.List(gcs.BucketName).
-			Prefix(ps).Do()
+		page, err := retry(
+			"List",
+			func() (interface{}, error) {
+				return gcs.objectsService.List(gcs.BucketName).
+					Prefix(ps).Do()
+			})
+
 		for {
+			if err != nil {
+				ch <- storage.BlockMetadata{Error: err}
+				break
+			}
+
 			if page == nil {
 				break
 			}
-			for _, o := range page.Items {
+			objects := page.(*gcsclient.Objects)
+			for _, o := range objects.Items {
 				t, e := time.Parse(time.RFC3339, o.TimeCreated)
 				if e != nil {
 					ch <- storage.BlockMetadata{
@@ -139,10 +165,14 @@ func (gcs *gcsStorage) ListBlocks(prefix string) chan (storage.BlockMetadata) {
 				}
 			}
 
-			if page.NextPageToken != "" {
-				page, _ = gcs.objectsService.List(gcs.BucketName).
-					PageToken(ps).
-					Prefix(gcs.getObjectNameString(prefix)).Do()
+			if objects.NextPageToken != "" {
+				page, err = retry(
+					"List",
+					func() (interface{}, error) {
+						return gcs.objectsService.List(gcs.BucketName).
+							PageToken(objects.NextPageToken).
+							Prefix(ps).Do()
+					})
 			} else {
 				break
 			}