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. Signed-off-by: Valentin Rothberg <[email protected]>
containers · Sep 17, 2019 · 994ca9c · 994ca9c
1 parent 9bafe00
commit 994ca9c
Show file tree

Hide file tree

Showing 5 changed files with 74 additions and 33 deletions.
diff --git a/docker/docker_client.go b/docker/docker_client.go
@@ -21,7 +21,6 @@ import (
 	"github.com/containers/image/pkg/sysregistriesv2"
 	"github.com/containers/image/pkg/tlsclientconfig"
 	"github.com/containers/image/types"
-	"github.com/docker/distribution/registry/client"
 	"github.com/docker/go-connections/tlsconfig"
 	digest "github.com/opencontainers/go-digest"
 	"github.com/pkg/errors"
@@ -47,14 +46,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 +276,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 +349,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 err := httpResponseToError(resp); err != nil {
+				logrus.Debugf("error getting search results from v1 endpoint %q: %v", err)
 			} else {
 				if err := json.NewDecoder(resp.Body).Decode(v1Res); err != nil {
 					return nil, err
@@ -381,8 +366,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 err := httpResponseToError(resp); err != nil {
+			logrus.Errorf("error getting search results from v2 endpoint %q: %v", registry, err)
 		} else {
 			if err := json.NewDecoder(resp.Body).Decode(v2Res); err != nil {
 				return nil, err
@@ -419,6 +404,30 @@ func (c *dockerClient) makeRequest(ctx context.Context, method, path string, hea
 // makeRequest should generally be preferred.
 // 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 time.Duration
+	)
+	// Do an exponential back off starting at 2 seconds for at most 5
+	// iterations.
+	delay = 2
+	for i := 0; i < 5; i++ {
+		res, err = c.makeRequestToResolvedURLHelper(ctx, method, url, headers, stream, streamLen, auth, extraScope)
+		if res != nil && res.StatusCode == http.StatusTooManyRequests {
+			logrus.Debugf("%s: sleeping for %d seconds before next attempt", http.StatusText(res.StatusCode), delay)
+			if i < 4 {
+				time.Sleep(delay * time.Second)
+				delay *= 2
+			}
+			continue
+		}
+		break
+	}
+	return res, err
+}
+
+func (c *dockerClient) makeRequestToResolvedURLHelper(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 +578,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 +590,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 +636,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())
+
+	if err := httpResponseToError(res); err != nil {
+		return nil, errors.Wrapf(err, "Error downloading signatures for %s in %s", manifestDigest, ref.ref.Name())
 	}
+
 	body, err := ioutil.ReadAll(res.Body)
 	if err != nil {
 		return nil, err

diff --git a/docker/docker_image.go b/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 {

diff --git a/docker/docker_image_dest.go b/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"
@@ -203,6 +203,7 @@ func (d *dockerImageDestination) blobExists(ctx context.Context, repo reference.
 		return false, -1, err
 	}
 	defer res.Body.Close()
+
 	switch res.StatusCode {
 	case http.StatusOK:
 		logrus.Debugf("... already exists")

diff --git a/docker/docker_image_src.go b/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

diff --git a/docker/errors.go b/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))
+	}
+}