s3: introduce retrying on known errors

s3/s3.go

 //
 // goamz - Go packages to interact with the Amazon Web Services.
 //
 //   https://wiki.ubuntu.com/goamz
 //
 // Copyright (c) 2011 Canonical Ltd.
 //
 // Written by Gustavo Niemeyer <gustavo.niemeyer@canonical.com>
 //
 
 package s3
 
 import (
         "bytes"
         "encoding/xml"
         "fmt"
         "io"
         "io/ioutil"
         "launchpad.net/goamz/aws"
         "log"
+        "net"
         "net/http"
         "net/http/httputil"
         "net/url"
         "strconv"
         "strings"
         "time"
 )
 
 const debug = false
 
 // The S3 type encapsulates operations with an S3 region.
 type S3 struct {
         aws.Auth
         aws.Region
         private byte // Reserve the right of using private data.
 }
 
 // The Bucket type encapsulates operations with an S3 bucket.
 type Bucket struct {
         *S3
         Name string
 }
 
 // The Owner type represents the owner of the object in an S3 bucket.
 type Owner struct {
         ID          string
         DisplayName string
 }
 
+var attempts = aws.AttemptStrategy{
+        Total: 5 * time.Second,
+        Delay: 200 * time.Millisecond,
+}
+
 // New creates a new S3.
 func New(auth aws.Auth, region aws.Region) *S3 {
         return &S3{auth, region, 0}
 }
 
 // Bucket returns a Bucket with the given name.
 func (s3 *S3) Bucket(name string) *Bucket {
         if s3.Region.S3BucketEndpoint != "" || s3.Region.S3LowercaseBucket {
                 name = strings.ToLower(name)
         }
@@ skipping 41 matching lines @@
                 headers: headers,
                 payload: b.locationConstraint(),
         }
         return b.S3.query(req, nil)
 }
 
 // DelBucket removes an existing S3 bucket. All objects in the bucket must
 // be removed before the bucket itself can be removed.
 //
 // See http://goo.gl/GoBrY for details.
-func (b *Bucket) DelBucket() error {
+func (b *Bucket) DelBucket() (err error) {
         req := &request{
                 method: "DELETE",
                 bucket: b.Name,
                 path:   "/",
         }
-        return b.S3.query(req, nil)
-}
-
-func hasCode(err error, code string) bool {
-        s3err, ok := err.(*Error)
-        return ok && s3err.Code == code
+        for attempt := attempts.Start(); attempt.Next(); {
+                err = b.S3.query(req, nil)
+                if !shouldRetry(err) || hasCode(err, "NoSuchBucket") {
+                        break
+                }
+        }
+        return err
 }
 
 // Get retrieves an object from an S3 bucket.
 //
 // See http://goo.gl/isCO7 for details.
 func (b *Bucket) Get(path string) (data []byte, err error) {
         body, err := b.GetReader(path)
         if err != nil {
                 return nil, err
         }
         data, err = ioutil.ReadAll(body)
         body.Close()
         return data, err
 }
 
 // GetReader retrieves an object from an S3 bucket.
 // It is the caller's responsibility to call Close on rc when
 // finished reading.
 func (b *Bucket) GetReader(path string) (rc io.ReadCloser, err error) {
         req := &request{
                 bucket: b.Name,
                 path:   path,
         }
         err = b.S3.prepare(req)
         if err != nil {
                 return nil, err
         }
-        resp, err := b.S3.run(req, nil)
-        if err != nil {
-                return nil, err
+        for attempt := attempts.Start(); attempt.Next(); {
+                resp, err := b.S3.run(req, nil)
+                if shouldRetry(err) && attempt.HasNext() {

[rog, 2013/02/01 10:04:01]

nice idiom.

+                        continue
+                }
+                if err != nil {
+                        return nil, err
+                }
+                return resp.Body, nil
         }
-        return resp.Body, nil
+        panic("unreachable")
 }
 
 // Put inserts an object into the S3 bucket.
 //
 // See http://goo.gl/FEBPD for details.
 func (b *Bucket) Put(path string, data []byte, contType string, perm ACL) error {
         body := bytes.NewBuffer(data)
         return b.PutReader(path, body, int64(len(data)), contType, perm)
 }
 
@@ skipping 118 matching lines @@
                 "marker":    {marker},
         }
         if max != 0 {
                 params["max-keys"] = []string{strconv.FormatInt(int64(max), 10)}
         }
         req := &request{
                 bucket: b.Name,
                 params: params,
         }
         result = &ListResp{}
-        err = b.S3.query(req, result)
+        for attempt := attempts.Start(); attempt.Next(); {
+                err = b.S3.query(req, result)
+                if !shouldRetry(err) {
+                        break
+                }
+                println("Retrying List.")

[rog, 2013/02/01 10:04:01]

d

[niemeyer, 2013/02/01 20:21:58]

Done.

+        }
         if err != nil {
                 return nil, err
         }
         return result, nil
 }
 
 // URL returns a non-signed URL that allows retriving the
 // object at path. It only works if the object is publicly
 // readable (see SignedURL).
 func (b *Bucket) URL(path string) string {
@@ skipping 190 matching lines @@
         r.Body.Close()
         err.StatusCode = r.StatusCode
         if err.Message == "" {
                 err.Message = r.Status
         }
         if debug {
                 log.Printf("err: %#v\n", err)
         }
         return &err
 }
+
+func shouldRetry(err error) bool {
+        if err == nil {
+                return false
+        }
+        switch e := err.(type) {

[rog, 2013/02/01 10:04:01]

io.ErrUnexpectedEOF is one i've seen in the past, although i haven't seen it for
a while.

[niemeyer, 2013/02/01 20:21:58]

Done.

+        case *net.DNSError:
+                return true
+        case *net.OpError:

[rog, 2013/02/01 10:04:01]

this seems very broad. we don't want to retry for 5 seconds if we've got an
address parsing error, for example.
i'd be tempted to restrict it to Op=="remote error" and see how it goes.

[niemeyer, 2013/02/01 20:21:58]

True. I'll restrict to "read" and "write".

+                return true
+        case *Error:
+                switch e.Code {
+                case "InternalError", "NoSuchUpload", "NoSuchBucket":
+                        return true
+                }
+        }
+        return false
+}
+
+func hasCode(err error, code string) bool {
+        s3err, ok := err.(*Error)
+        return ok && s3err.Code == code
+}
+