docker: handle http 429 status codes

Consolidate checking the http-status codes to allow for a more uniform
error handling.  Also treat code 429 (too many requests) as a known
error instead of an invalid status code.

When hitting 429, perform an exponential back off starting a 2 seconds
for at most 5 iterations.  If the http.Response set the `Retry-Header`
then use the provided value or date to compute the delay until the
next attempt.  Note that the maximum delay is 60 seconds.

Signed-off-by: Valentin Rothberg <[email protected]>

docker/docker_client.go

@@ -21,7 +21,7 @@ import (
 	"github.com/containers/image/pkg/sysregistriesv2"
 	"github.com/containers/image/pkg/tlsclientconfig"
 	"github.com/containers/image/types"
-	"github.com/docker/distribution/registry/client"
+	clientLib "github.com/docker/distribution/registry/client"
 	"github.com/docker/go-connections/tlsconfig"
 	digest "github.com/opencontainers/go-digest"
 	"github.com/pkg/errors"
@@ -47,14 +47,7 @@ const (
 	extensionSignatureTypeAtomic    = "atomic" // extensionSignature.Type
 )
 
-var (
-	// ErrV1NotSupported is returned when we're trying to talk to a
-	// docker V1 registry.
-	ErrV1NotSupported = errors.New("can't talk to a V1 docker registry")
-	// ErrUnauthorizedForCredentials is returned when the status code returned is 401
-	ErrUnauthorizedForCredentials = errors.New("unable to retrieve auth token: invalid username/password")
-	systemPerHostCertDirPaths     = [2]string{"/etc/containers/certs.d", "/etc/docker/certs.d"}
-)
+var systemPerHostCertDirPaths = [2]string{"/etc/containers/certs.d", "/etc/docker/certs.d"}
 
 // extensionSignature and extensionSignatureList come from github.com/openshift/origin/pkg/dockerregistry/server/signaturedispatcher.go:
 // signature represents a Docker image signature.
@@ -284,14 +277,7 @@ func CheckAuth(ctx context.Context, sys *types.SystemContext, username, password
 	}
 	defer resp.Body.Close()
 
-	switch resp.StatusCode {
-	case http.StatusOK:
-		return nil
-	case http.StatusUnauthorized:
-		return ErrUnauthorizedForCredentials
-	default:
-		return errors.Errorf("error occured with status code %d (%s)", resp.StatusCode, http.StatusText(resp.StatusCode))
-	}
+	return httpResponseToError(resp)
 }
 
 // SearchResult holds the information of each matching image
@@ -364,8 +350,8 @@ func SearchRegistry(ctx context.Context, sys *types.SystemContext, registry, ima
 			logrus.Debugf("error getting search results from v1 endpoint %q: %v", registry, err)
 		} else {
 			defer resp.Body.Close()
-			if resp.StatusCode != http.StatusOK {
-				logrus.Debugf("error getting search results from v1 endpoint %q, status code %d (%s)", registry, resp.StatusCode, http.StatusText(resp.StatusCode))
+			if resp.StatusCode == http.StatusOK {
+				logrus.Debugf("error getting search results from v1 endpoint %q: %v", registry, httpResponseToError(resp))
 			} else {
 				if err := json.NewDecoder(resp.Body).Decode(v1Res); err != nil {
 					return nil, err
@@ -381,8 +367,8 @@ func SearchRegistry(ctx context.Context, sys *types.SystemContext, registry, ima
 		logrus.Debugf("error getting search results from v2 endpoint %q: %v", registry, err)
 	} else {
 		defer resp.Body.Close()
-		if resp.StatusCode != http.StatusOK {
-			logrus.Errorf("error getting search results from v2 endpoint %q, status code %d (%s)", registry, resp.StatusCode, http.StatusText(resp.StatusCode))
+		if resp.StatusCode == http.StatusOK {
+			logrus.Errorf("error getting search results from v2 endpoint %q: %v", registry, httpResponseToError(resp))
 		} else {
 			if err := json.NewDecoder(resp.Body).Decode(v2Res); err != nil {
 				return nil, err
@@ -417,8 +403,77 @@ func (c *dockerClient) makeRequest(ctx context.Context, method, path string, hea
 // makeRequestToResolvedURL creates and executes a http.Request with the specified parameters, adding authentication and TLS options for the Docker client.
 // streamLen, if not -1, specifies the length of the data expected on stream.
 // makeRequest should generally be preferred.
+// In case of an http 429 status code in the response, it performs an exponential back off starting at 2 seconds for at most 5 iterations.
+// If the `Retry-After` header is set in the response, the specified value or date is
+// If the stream is non-nil, no back off will be performed.
 // TODO(runcom): too many arguments here, use a struct
 func (c *dockerClient) makeRequestToResolvedURL(ctx context.Context, method, url string, headers map[string][]string, stream io.Reader, streamLen int64, auth sendAuth, extraScope *authScope) (*http.Response, error) {
+	var (
+		res   *http.Response
+		err   error
+		delay int64
+	)
+	delay = 2
+	const numIterations = 5
+	const maxDelay = 60
+
+	// math.Min() only supports float64, so have an anonymous func to avoid
+	// casting.
+	min := func(a int64, b int64) int64 {
+		if a < b {
+			return a
+		}
+		return b
+	}
+
+	nextDelay := func(r *http.Response, delay int64) int64 {
+		after := res.Header.Get("Retry-After")
+		if after == "" {
+			return min(delay, maxDelay)
+		}
+		logrus.Debugf("detected 'Retry-After' header %q", after, delay)
+		// First check if we have a numerical value.
+		if num, err := strconv.ParseInt(after, 10, 64); err == nil {
+			return min(num, maxDelay)
+		}
+		// Secondly check if we have an http date.
+		// If the delta between the date and now is positive, use it.
+		// Otherwise, fall back to using the default exponential back off.
+		if t, err := http.ParseTime(after); err == nil {
+			delta := int64(t.Sub(time.Now()).Seconds())
+			if delta > 0 {
+				return min(delta, maxDelay)
+			}
+			logrus.Debugf("negative date: falling back to using %d seconds", after, delay)
+			return min(delay, maxDelay)
+		}
+		// If the header contains bogus, fall back to using the default
+		// exponential back off.
+		logrus.Debugf("invalid format: falling back to using %d seconds", after, delay)
+		return min(delay, maxDelay)
+	}
+
+	for i := 0; i < numIterations; i++ {
+		res, err = c.makeRequestToResolvedURLOnce(ctx, method, url, headers, stream, streamLen, auth, extraScope)
+		if stream == nil && res != nil && res.StatusCode == http.StatusTooManyRequests {
+			if i < numIterations-1 {
+				delay = nextDelay(res, delay) // compute next delay - does NOT exceed maxDelay
+				logrus.Debugf("too many request to %s: sleeping for %d seconds before next attempt", url, delay)
+				time.Sleep(time.Duration(delay) * time.Second)
+				delay = delay * 2 // exponential back off
+			}
+			continue
+		}
+		break
+	}
+	return res, err
+}
+
+// makeRequestToResolvedURLOnce creates and executes a http.Request with the specified parameters, adding authentication and TLS options for the Docker client.
+// streamLen, if not -1, specifies the length of the data expected on stream.
+// makeRequest should generally be preferred.
+// Note that no exponential back off is performed when receiving an http 429 status code.
+func (c *dockerClient) makeRequestToResolvedURLOnce(ctx context.Context, method, url string, headers map[string][]string, stream io.Reader, streamLen int64, auth sendAuth, extraScope *authScope) (*http.Response, error) {
 	req, err := http.NewRequest(method, url, stream)
 	if err != nil {
 		return nil, err
@@ -569,7 +624,7 @@ func (c *dockerClient) detectPropertiesHelper(ctx context.Context) error {
 		defer resp.Body.Close()
 		logrus.Debugf("Ping %s status %d", url, resp.StatusCode)
 		if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusUnauthorized {
-			return errors.Errorf("error pinging registry %s, response code %d (%s)", c.registry, resp.StatusCode, http.StatusText(resp.StatusCode))
+			return httpResponseToError(resp)
 		}
 		c.challenges = parseAuthHeader(resp.Header)
 		c.scheme = scheme
@@ -581,7 +636,7 @@ func (c *dockerClient) detectPropertiesHelper(ctx context.Context) error {
 		err = ping("http")
 	}
 	if err != nil {
-		err = errors.Wrap(err, "pinging docker registry returned")
+		err = errors.Wrapf(err, "error pinging docker registry %s", c.registry)
 		if c.sys != nil && c.sys.DockerDisableV1Ping {
 			return err
 		}
@@ -627,9 +682,11 @@ func (c *dockerClient) getExtensionsSignatures(ctx context.Context, ref dockerRe
 		return nil, err
 	}
 	defer res.Body.Close()
+
 	if res.StatusCode != http.StatusOK {
-		return nil, errors.Wrapf(client.HandleErrorResponse(res), "Error downloading signatures for %s in %s", manifestDigest, ref.ref.Name())
+		return nil, errors.Wrapf(clientLib.HandleErrorResponse(res), "Error downloading signatures for %s in %s", manifestDigest, ref.ref.Name())
 	}
+
 	body, err := ioutil.ReadAll(res.Body)
 	if err != nil {
 		return nil, err

docker/docker_image.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"net/http"
 	"net/url"
 	"strings"
 
@@ -71,9 +70,8 @@ func GetRepositoryTags(ctx context.Context, sys *types.SystemContext, ref types.
 			return nil, err
 		}
 		defer res.Body.Close()
-		if res.StatusCode != http.StatusOK {
-			// print url also
-			return nil, errors.Errorf("Invalid status code returned when fetching tags list %d (%s)", res.StatusCode, http.StatusText(res.StatusCode))
+		if err := httpResponseToError(res); err != nil {
+			return nil, err
 		}
 
 		var tagsHolder struct {

docker/docker_image_dest.go

@@ -19,7 +19,7 @@ import (
 	"github.com/containers/image/pkg/blobinfocache/none"
 	"github.com/containers/image/types"
 	"github.com/docker/distribution/registry/api/errcode"
-	"github.com/docker/distribution/registry/api/v2"
+	v2 "github.com/docker/distribution/registry/api/v2"
 	"github.com/docker/distribution/registry/client"
 	"github.com/opencontainers/go-digest"
 	imgspecv1 "github.com/opencontainers/image-spec/specs-go/v1"

docker/docker_image_src.go

@@ -232,9 +232,8 @@ func (s *dockerImageSource) GetBlob(ctx context.Context, info types.BlobInfo, ca
 	if err != nil {
 		return nil, 0, err
 	}
-	if res.StatusCode != http.StatusOK {
-		// print url also
-		return nil, 0, errors.Errorf("Invalid status code returned when fetching blob %d (%s)", res.StatusCode, http.StatusText(res.StatusCode))
+	if err := httpResponseToError(res); err != nil {
+		return nil, 0, err
 	}
 	cache.RecordKnownLocation(s.ref.Transport(), bicTransportScope(s.ref), info.Digest, newBICLocationReference(s.ref))
 	return res.Body, getBlobSize(res), nil

docker/errors.go

@@ -0,0 +1,32 @@
+package docker
+
+import (
+	"net/http"
+
+	"github.com/pkg/errors"
+)
+
+var (
+	// ErrV1NotSupported is returned when we're trying to talk to a
+	// docker V1 registry.
+	ErrV1NotSupported = errors.New("can't talk to a V1 docker registry")
+	// ErrUnauthorizedForCredentials is returned when the status code returned is 401
+	ErrUnauthorizedForCredentials = errors.New("unable to retrieve auth token: invalid username/password")
+	// ErrTooManyRequests is returned when the status code returned is 429
+	ErrTooManyRequests = errors.New("too many request to registry")
+)
+
+// httpResponseToError translates the https.Response into an error. It returns
+// nil if the response is not considered an error.
+func httpResponseToError(res *http.Response) error {
+	switch res.StatusCode {
+	case http.StatusOK:
+		return nil
+	case http.StatusTooManyRequests:
+		return ErrTooManyRequests
+	case http.StatusUnauthorized:
+		return ErrUnauthorizedForCredentials
+	default:
+		return errors.Errorf("invalid status code from registry %d (%s)", res.StatusCode, http.StatusText(res.StatusCode))
+	}
+}