Merge pull request #531 from abhi/registry-mirror

Adding Registry Mirror support

cmd/cri-containerd/options/options.go

@@ -72,6 +72,8 @@ type PluginConfig struct {
 	ContainerdConfig `toml:"containerd" json:"containerd,omitempty"`
 	// CniConfig contains config related to cni
 	CniConfig `toml:"cni" json:"cni,omitempty"`
+	// Registry contains config related to the registry
+	Registry `toml:"registry" json:"registry,omitempty"`
 	// StreamServerAddress is the ip address streaming server is listening on.
 	StreamServerAddress string `toml:"stream_server_address" json:"streamServerAddress,omitempty"`
 	// StreamServerPort is the port streaming server is listening on.
@@ -191,6 +193,8 @@ func (c *CRIContainerdOptions) AddFlags(fs *pflag.FlagSet) {
 		defaults.SkipImageFSUUID, "Skip retrieval of imagefs uuid. When turned on, kubelet will not be able to get imagefs capacity or perform imagefs disk eviction.")
 	fs.BoolVar(&c.EnableIPv6DAD, "enable-ipv6-dad",
 		defaults.EnableIPv6DAD, "Enable IPv6 DAD (duplicate address detection) for pod sandbox network. Enabling this will increase pod sandbox start latency by several seconds.")
+	fs.Var(&c.Registry, "registry",
+		"Registry config for image pull eg --registry=myregistry.io=https://mymirror.io/ --registry=myregistry2.io=https://mymirror2.io/")
 }
 
 // InitFlags load configurations from config file, and then overwrite with flags.
@@ -259,6 +263,13 @@ func DefaultConfig() Config {
 			SystemdCgroup:       false,
 			SkipImageFSUUID:     false,
 			EnableIPv6DAD:       false,
+			Registry: Registry{
+				Mirrors: map[string]Mirror{
+					"docker.io": {
+						Endpoints: []string{"https://registry-1.docker.io"},
+					},
+				},
+			},
 		},
 		ContainerdRootDir:  "/var/lib/containerd",
 		ContainerdEndpoint: "/run/containerd/containerd.sock",

cmd/cri-containerd/options/registry.go

@@ -0,0 +1,88 @@
+/*
+Copyright 2018 The Containerd Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package options
+
+import (
+	"fmt"
+	"net/url"
+	"strings"
+)
+
+// Mirror contains the config related to the registry mirror
+type Mirror struct {
+	Endpoints []string `toml:"endpoint" json:"endpoint,omitempty"`
+	// TODO (Abhi) We might need to add auth per namespace. Looks like
+	// image auth information is passed by kube itself.
+}
+
+// Registry is registry settings configured
+type Registry struct {
+	Mirrors map[string]Mirror `toml:"mirrors" json:"mirrors,omitempty"`
+}
+
+// String returns the string format of registry type
+func (r *Registry) String() string {
+	// Its not used hence return empty string
+	return ""
+}
+
+// Set validates and converts into the internal registry struct
+func (r *Registry) Set(s string) error {
+	// --registry docker.io=https://mymirror.io,http://mymirror2.io
+	// If no option is set then return format error
+	if len(s) == 0 {
+		return fmt.Errorf("incomplete registry mirror option")
+	}
+	var mirrors []string
+	host := "docker.io"
+	opt := strings.Split(s, "=")
+	if len(opt) > 1 {
+		// If option is set in the format "mynamespace.io=https://mymirror.io,https://mymirror2.io"
+		// Then associate the mirror urls for the namespace only"
+		host = opt[0]
+		mirrors = strings.Split(opt[1], ",")
+	} else {
+		// If option is set in the format "https://mymirror.io,https://mymirror.io"
+		// Then associate mirror against default docker.io namespace
+		mirrors = strings.Split(opt[0], ",")
+	}
+
+	// Validate the format of the urls passed
+	for _, u := range mirrors {
+		_, err := url.Parse(u)
+		if err != nil {
+			return fmt.Errorf("invalid registry mirror url format %v: %v", u, err)
+		}
+	}
+
+	if r.Mirrors == nil {
+		r.Mirrors = make(map[string]Mirror)
+	}
+	if _, ok := r.Mirrors[host]; !ok {
+		r.Mirrors[host] = Mirror{}
+	}
+	m := r.Mirrors[host]
+	m.Endpoints = append(m.Endpoints, mirrors...)
+	r.Mirrors[host] = m
+
+	return nil
+}
+
+// Type returns a string name for the option type
+func (r *Registry) Type() string {
+	return "list"
+}

pkg/containerd/resolver/auth.go

@@ -0,0 +1,202 @@
+/*
+Copyright 2018 The Containerd Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// The corresponding file is in containerd/remote/docker/
+// This package can be removed once a more feasible and hollistic resolver
+// is finalized in containerd.
+
+package resolver
+
+import (
+	"net/http"
+	"sort"
+	"strings"
+)
+
+type authenticationScheme byte
+
+const (
+	basicAuth  authenticationScheme = 1 << iota // Defined in RFC 7617
+	digestAuth                                  // Defined in RFC 7616
+	bearerAuth                                  // Defined in RFC 6750
+)
+
+// challenge carries information from a WWW-Authenticate response header.
+// See RFC 2617.
+type challenge struct {
+	// scheme is the auth-scheme according to RFC 2617
+	scheme authenticationScheme
+
+	// parameters are the auth-params according to RFC 2617
+	parameters map[string]string
+}
+
+type byScheme []challenge
+
+func (bs byScheme) Len() int      { return len(bs) }
+func (bs byScheme) Swap(i, j int) { bs[i], bs[j] = bs[j], bs[i] }
+
+// Sort in priority order: token > digest > basic
+func (bs byScheme) Less(i, j int) bool { return bs[i].scheme > bs[j].scheme }
+
+// Octet types from RFC 2616.
+type octetType byte
+
+var octetTypes [256]octetType
+
+const (
+	isToken octetType = 1 << iota
+	isSpace
+)
+
+func init() {
+	// OCTET      = <any 8-bit sequence of data>
+	// CHAR       = <any US-ASCII character (octets 0 - 127)>
+	// CTL        = <any US-ASCII control character (octets 0 - 31) and DEL (127)>
+	// CR         = <US-ASCII CR, carriage return (13)>
+	// LF         = <US-ASCII LF, linefeed (10)>
+	// SP         = <US-ASCII SP, space (32)>
+	// HT         = <US-ASCII HT, horizontal-tab (9)>
+	// <">        = <US-ASCII double-quote mark (34)>
+	// CRLF       = CR LF
+	// LWS        = [CRLF] 1*( SP | HT )
+	// TEXT       = <any OCTET except CTLs, but including LWS>
+	// separators = "(" | ")" | "<" | ">" | "@" | "," | ";" | ":" | "\" | <">
+	//              | "/" | "[" | "]" | "?" | "=" | "{" | "}" | SP | HT
+	// token      = 1*<any CHAR except CTLs or separators>
+	// qdtext     = <any TEXT except <">>
+
+	for c := 0; c < 256; c++ {
+		var t octetType
+		isCtl := c <= 31 || c == 127
+		isChar := 0 <= c && c <= 127
+		isSeparator := strings.ContainsRune(" \t\"(),/:;<=>?@[]\\{}", rune(c))
+		if strings.ContainsRune(" \t\r\n", rune(c)) {
+			t |= isSpace
+		}
+		if isChar && !isCtl && !isSeparator {
+			t |= isToken
+		}
+		octetTypes[c] = t
+	}
+}
+
+func parseAuthHeader(header http.Header) []challenge {
+	challenges := []challenge{}
+	for _, h := range header[http.CanonicalHeaderKey("WWW-Authenticate")] {
+		v, p := parseValueAndParams(h)
+		var s authenticationScheme
+		switch v {
+		case "basic":
+			s = basicAuth
+		case "digest":
+			s = digestAuth
+		case "bearer":
+			s = bearerAuth
+		default:
+			continue
+		}
+		challenges = append(challenges, challenge{scheme: s, parameters: p})
+	}
+	sort.Stable(byScheme(challenges))
+	return challenges
+}
+
+func parseValueAndParams(header string) (value string, params map[string]string) {
+	params = make(map[string]string)
+	value, s := expectToken(header)
+	if value == "" {
+		return
+	}
+	value = strings.ToLower(value)
+	for {
+		var pkey string
+		pkey, s = expectToken(skipSpace(s))
+		if pkey == "" {
+			return
+		}
+		if !strings.HasPrefix(s, "=") {
+			return
+		}
+		var pvalue string
+		pvalue, s = expectTokenOrQuoted(s[1:])
+		if pvalue == "" {
+			return
+		}
+		pkey = strings.ToLower(pkey)
+		params[pkey] = pvalue
+		s = skipSpace(s)
+		if !strings.HasPrefix(s, ",") {
+			return
+		}
+		s = s[1:]
+	}
+}
+
+func skipSpace(s string) (rest string) {
+	i := 0
+	for ; i < len(s); i++ {
+		if octetTypes[s[i]]&isSpace == 0 {
+			break
+		}
+	}
+	return s[i:]
+}
+
+func expectToken(s string) (token, rest string) {
+	i := 0
+	for ; i < len(s); i++ {
+		if octetTypes[s[i]]&isToken == 0 {
+			break
+		}
+	}
+	return s[:i], s[i:]
+}
+
+func expectTokenOrQuoted(s string) (value string, rest string) {
+	if !strings.HasPrefix(s, "\"") {
+		return expectToken(s)
+	}
+	s = s[1:]
+	for i := 0; i < len(s); i++ {
+		switch s[i] {
+		case '"':
+			return s[:i], s[i+1:]
+		case '\\':
+			p := make([]byte, len(s)-1)
+			j := copy(p, s[:i])
+			escape := true
+			for i = i + 1; i < len(s); i++ {
+				b := s[i]
+				switch {
+				case escape:
+					escape = false
+					p[j] = b
+					j++
+				case b == '\\':
+					escape = true
+				case b == '"':
+					return string(p[:j]), s[i+1:]
+				default:
+					p[j] = b
+					j++
+				}
+			}
+			return "", ""
+		}
+	}
+	return "", ""
+}

pkg/containerd/resolver/fetcher.go

@@ -0,0 +1,142 @@
+/*
+Copyright 2018 The Containerd Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// The corresponding file is in containerd/remote/docker/
+// This package can be removed once a more feasible and hollistic resolver
+// is finalized in containerd.
+
+package resolver
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"path"
+	"strings"
+
+	"github.com/containerd/containerd/errdefs"
+	"github.com/containerd/containerd/images"
+	"github.com/containerd/containerd/log"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+)
+
+type dockerFetcher struct {
+	*dockerBase
+}
+
+func (r dockerFetcher) Fetch(ctx context.Context, desc ocispec.Descriptor) (io.ReadCloser, error) {
+	ctx = log.WithLogger(ctx, log.G(ctx).WithFields(
+		logrus.Fields{
+			//"base":   r.base.String(),
+			"digest": desc.Digest,
+		},
+	))
+
+	urls, err := r.getV2URLPaths(ctx, desc)
+	if err != nil {
+		return nil, err
+	}
+
+	ctx, err = contextWithRepositoryScope(ctx, r.refspec, false)
+	if err != nil {
+		return nil, err
+	}
+
+	return newHTTPReadSeeker(desc.Size, func(offset int64) (io.ReadCloser, error) {
+		for _, u := range urls {
+			rc, err := r.open(ctx, u, desc.MediaType, offset)
+			if err != nil {
+				if errdefs.IsNotFound(err) {
+					continue // try one of the other urls.
+				}
+
+				return nil, err
+			}
+
+			return rc, nil
+		}
+
+		return nil, errors.Wrapf(errdefs.ErrNotFound,
+			"could not fetch content descriptor %v (%v) from remote",
+			desc.Digest, desc.MediaType)
+
+	})
+}
+
+func (r dockerFetcher) open(ctx context.Context, u, mediatype string, offset int64) (io.ReadCloser, error) {
+	req, err := http.NewRequest(http.MethodGet, u, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	req.Header.Set("Accept", strings.Join([]string{mediatype, `*`}, ", "))
+
+	if offset > 0 {
+		// TODO(stevvooe): Only set this header in response to the
+		// "Accept-Ranges: bytes" header.
+		req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset))
+	}
+
+	resp, err := r.doRequestWithRetries(ctx, req, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.StatusCode > 299 {
+		// TODO(stevvooe): When doing a offset specific request, we should
+		// really distinguish between a 206 and a 200. In the case of 200, we
+		// can discard the bytes, hiding the seek behavior from the
+		// implementation.
+
+		resp.Body.Close()
+		if resp.StatusCode == http.StatusNotFound {
+			return nil, errors.Wrapf(errdefs.ErrNotFound, "content at %v not found", u)
+		}
+		return nil, errors.Errorf("unexpected status code %v: %v", u, resp.Status)
+	}
+
+	return resp.Body, nil
+}
+
+// getV2URLPaths generates the candidate urls paths for the object based on the
+// set of hints and the provided object id. URLs are returned in the order of
+// most to least likely succeed.
+func (r *dockerFetcher) getV2URLPaths(ctx context.Context, desc ocispec.Descriptor) ([]string, error) {
+	var urls []string
+
+	if len(desc.URLs) > 0 {
+		// handle fetch via external urls.
+		for _, u := range desc.URLs {
+			log.G(ctx).WithField("url", u).Debug("adding alternative url")
+			urls = append(urls, u)
+		}
+	}
+
+	switch desc.MediaType {
+	case images.MediaTypeDockerSchema2Manifest, images.MediaTypeDockerSchema2ManifestList,
+		images.MediaTypeDockerSchema1Manifest,
+		ocispec.MediaTypeImageManifest, ocispec.MediaTypeImageIndex:
+		urls = append(urls, r.url(path.Join("manifests", desc.Digest.String())))
+	}
+
+	// always fallback to attempting to get the object out of the blobs store.
+	urls = append(urls, r.url(path.Join("blobs", desc.Digest.String())))
+
+	return urls, nil
+}

pkg/containerd/resolver/httpreadseeker.go

@@ -0,0 +1,148 @@
+/*
+Copyright 2018 The Containerd Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// The corresponding file is in containerd/remote/docker/.
+// This package can be removed once a more feasible and hollistic resolver
+// is finalized in containerd.
+
+package resolver
+
+import (
+	"bytes"
+	"io"
+	"io/ioutil"
+
+	"github.com/containerd/containerd/errdefs"
+	"github.com/containerd/containerd/log"
+	"github.com/pkg/errors"
+)
+
+type httpReadSeeker struct {
+	size   int64
+	offset int64
+	rc     io.ReadCloser
+	open   func(offset int64) (io.ReadCloser, error)
+	closed bool
+}
+
+func newHTTPReadSeeker(size int64, open func(offset int64) (io.ReadCloser, error)) (io.ReadCloser, error) {
+	return &httpReadSeeker{
+		size: size,
+		open: open,
+	}, nil
+}
+
+func (hrs *httpReadSeeker) Read(p []byte) (n int, err error) {
+	if hrs.closed {
+		return 0, io.EOF
+	}
+
+	rd, err := hrs.reader()
+	if err != nil {
+		return 0, err
+	}
+
+	n, err = rd.Read(p)
+	hrs.offset += int64(n)
+	return
+}
+
+func (hrs *httpReadSeeker) Close() error {
+	if hrs.closed {
+		return nil
+	}
+	hrs.closed = true
+	if hrs.rc != nil {
+		return hrs.rc.Close()
+	}
+
+	return nil
+}
+
+func (hrs *httpReadSeeker) Seek(offset int64, whence int) (int64, error) {
+	if hrs.closed {
+		return 0, errors.Wrap(errdefs.ErrUnavailable, "Fetcher.Seek: closed")
+	}
+
+	abs := hrs.offset
+	switch whence {
+	case io.SeekStart:
+		abs = offset
+	case io.SeekCurrent:
+		abs += offset
+	case io.SeekEnd:
+		if hrs.size == -1 {
+			return 0, errors.Wrap(errdefs.ErrUnavailable, "Fetcher.Seek: unknown size, cannot seek from end")
+		}
+		abs = hrs.size + offset
+	default:
+		return 0, errors.Wrap(errdefs.ErrInvalidArgument, "Fetcher.Seek: invalid whence")
+	}
+
+	if abs < 0 {
+		return 0, errors.Wrapf(errdefs.ErrInvalidArgument, "Fetcher.Seek: negative offset")
+	}
+
+	if abs != hrs.offset {
+		if hrs.rc != nil {
+			if err := hrs.rc.Close(); err != nil {
+				log.L.WithError(err).Errorf("Fetcher.Seek: failed to close ReadCloser")
+			}
+
+			hrs.rc = nil
+		}
+
+		hrs.offset = abs
+	}
+
+	return hrs.offset, nil
+}
+
+func (hrs *httpReadSeeker) reader() (io.Reader, error) {
+	if hrs.rc != nil {
+		return hrs.rc, nil
+	}
+
+	if hrs.size == -1 || hrs.offset < hrs.size {
+		// only try to reopen the body request if we are seeking to a value
+		// less than the actual size.
+		if hrs.open == nil {
+			return nil, errors.Wrapf(errdefs.ErrNotImplemented, "cannot open")
+		}
+
+		rc, err := hrs.open(hrs.offset)
+		if err != nil {
+			return nil, errors.Wrapf(err, "httpReaderSeeker: failed open")
+		}
+
+		if hrs.rc != nil {
+			if err := hrs.rc.Close(); err != nil {
+				log.L.WithError(err).Errorf("httpReadSeeker: failed to close ReadCloser")
+			}
+		}
+		hrs.rc = rc
+	} else {
+		// There is an edge case here where offset == size of the content. If
+		// we seek, we will probably get an error for content that cannot be
+		// sought (?). In that case, we should err on committing the content,
+		// as the length is already satisified but we just return the empty
+		// reader instead.
+
+		hrs.rc = ioutil.NopCloser(bytes.NewReader([]byte{}))
+	}
+
+	return hrs.rc, nil
+}

pkg/containerd/resolver/resolver.go

@@ -0,0 +1,606 @@
+/*
+Copyright 2018 The Containerd Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package resolver
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"path"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/containerd/containerd/images"
+	"github.com/containerd/containerd/log"
+	"github.com/containerd/containerd/reference"
+	"github.com/containerd/containerd/remotes"
+	digest "github.com/opencontainers/go-digest"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"golang.org/x/net/context/ctxhttp"
+)
+
+// This file is a modified copy of containerd/remote/docker/resolver.go.
+// The changes carried over by this file includes support for image ref
+// resolution and fetching image using multiple registry mirror urls. A new
+// ResolverOption called Registry is added. Registry will contain a map
+// of namespace relevant mirror urls. The client will use the ResolverOptions
+// to set the urls associated with the namespace of the image reference.
+// This package can be removed once a more feasible and hollistic resolver
+// is finalized in containerd. The specific changes are made to the base
+// function to calculate the base urls for the image location.  urls() is
+// added to fetch the mirror urls associated with the namespace of the image
+// ResolverOptions are changed for client to set the namespace and mirror urls
+// for to pull the image.
+
+var (
+	// ErrNoToken is returned if a request is successful but the body does not
+	// contain an authorization token.
+	ErrNoToken = errors.New("authorization server did not include a token in the response")
+
+	// ErrInvalidAuthorization is used when credentials are passed to a server but
+	// those credentials are rejected.
+	ErrInvalidAuthorization = errors.New("authorization failed")
+)
+
+type containerdResolver struct {
+	credentials func(string) (string, string, error)
+	plainHTTP   bool
+	client      *http.Client
+	tracker     StatusTracker
+	registry    map[string][]string
+}
+
+// Options are used to configured a new Docker register resolver
+type Options struct {
+	// Credentials provides username and secret given a host.
+	// If username is empty but a secret is given, that secret
+	// is interpretted as a long lived token.
+	Credentials func(string) (string, string, error)
+
+	// PlainHTTP specifies to use plain http and not https
+	PlainHTTP bool
+
+	// Client is the http client to used when making registry requests
+	Client *http.Client
+
+	// Tracker is used to track uploads to the registry. This is used
+	// since the registry does not have upload tracking and the existing
+	// mechanism for getting blob upload status is expensive.
+	Tracker StatusTracker
+
+	Registry map[string][]string
+}
+
+// NewResolver returns a new resolver to a Docker registry
+func NewResolver(options Options) remotes.Resolver {
+	tracker := options.Tracker
+	if tracker == nil {
+		tracker = NewInMemoryTracker()
+	}
+
+	return &containerdResolver{
+		credentials: options.Credentials,
+		plainHTTP:   options.PlainHTTP,
+		client:      options.Client,
+		tracker:     tracker,
+		registry:    options.Registry,
+	}
+}
+
+var _ remotes.Resolver = &containerdResolver{}
+
+func (r *containerdResolver) Resolve(ctx context.Context, ref string) (string, ocispec.Descriptor, error) {
+	refspec, err := reference.Parse(ref)
+	if err != nil {
+		return "", ocispec.Descriptor{}, err
+	}
+
+	if refspec.Object == "" {
+		return "", ocispec.Descriptor{}, reference.ErrObjectRequired
+	}
+
+	base, err := r.base(refspec)
+	if err != nil {
+		return "", ocispec.Descriptor{}, err
+	}
+
+	fetcher := dockerFetcher{
+		dockerBase: base,
+	}
+
+	var (
+		urls []string
+		dgst = refspec.Digest()
+	)
+
+	if dgst != "" {
+		if err := dgst.Validate(); err != nil {
+			// need to fail here, since we can't actually resolve the invalid
+			// digest.
+			return "", ocispec.Descriptor{}, err
+		}
+
+		// turns out, we have a valid digest, make a url.
+		urls = append(urls, fetcher.urls("manifests", dgst.String())...)
+
+		// fallback to blobs on not found.
+		urls = append(urls, fetcher.urls("blobs", dgst.String())...)
+	} else {
+		urls = append(urls, fetcher.urls("manifests", refspec.Object)...)
+	}
+
+	ctx, err = contextWithRepositoryScope(ctx, refspec, false)
+	if err != nil {
+		return "", ocispec.Descriptor{}, err
+	}
+	for _, u := range urls {
+		log.G(ctx).WithFields(logrus.Fields{
+			"url": u,
+		}).Debug("Trying to fetch from url")
+		req, err := http.NewRequest(http.MethodHead, u, nil)
+		if err != nil {
+			return "", ocispec.Descriptor{}, err
+		}
+
+		// set headers for all the types we support for resolution.
+		req.Header.Set("Accept", strings.Join([]string{
+			images.MediaTypeDockerSchema2Manifest,
+			images.MediaTypeDockerSchema2ManifestList,
+			ocispec.MediaTypeImageManifest,
+			ocispec.MediaTypeImageIndex, "*"}, ", "))
+
+		log.G(ctx).Info("resolving")
+		resp, err := fetcher.doRequestWithRetries(ctx, req, nil)
+		if err != nil {
+			return "", ocispec.Descriptor{}, err
+		}
+		resp.Body.Close() // don't care about body contents.
+
+		if resp.StatusCode > 299 {
+			if resp.StatusCode == http.StatusNotFound {
+				continue
+			}
+			return "", ocispec.Descriptor{}, errors.Errorf("unexpected status code %v: %v", u, resp.Status)
+		}
+
+		// this is the only point at which we trust the registry. we use the
+		// content headers to assemble a descriptor for the name. when this becomes
+		// more robust, we mostly get this information from a secure trust store.
+		dgstHeader := digest.Digest(resp.Header.Get("Docker-Content-Digest"))
+
+		if dgstHeader != "" {
+			if err := dgstHeader.Validate(); err != nil {
+				return "", ocispec.Descriptor{}, errors.Wrapf(err, "%q in header not a valid digest", dgstHeader)
+			}
+			dgst = dgstHeader
+		}
+
+		if dgst == "" {
+			return "", ocispec.Descriptor{}, errors.Errorf("could not resolve digest for %v", ref)
+		}
+
+		var (
+			size       int64
+			sizeHeader = resp.Header.Get("Content-Length")
+		)
+
+		size, err = strconv.ParseInt(sizeHeader, 10, 64)
+		if err != nil {
+
+			return "", ocispec.Descriptor{}, errors.Wrapf(err, "invalid size header: %q", sizeHeader)
+		}
+		if size < 0 {
+			return "", ocispec.Descriptor{}, errors.Errorf("%q in header not a valid size", sizeHeader)
+		}
+
+		desc := ocispec.Descriptor{
+			Digest:    dgst,
+			MediaType: resp.Header.Get("Content-Type"), // need to strip disposition?
+			Size:      size,
+		}
+
+		log.G(ctx).WithField("desc.digest", desc.Digest).Debug("resolved")
+		return ref, desc, nil
+	}
+
+	return "", ocispec.Descriptor{}, errors.Errorf("%v not found", ref)
+}
+
+func (r *containerdResolver) Fetcher(ctx context.Context, ref string) (remotes.Fetcher, error) {
+	refspec, err := reference.Parse(ref)
+	if err != nil {
+		return nil, err
+	}
+
+	base, err := r.base(refspec)
+	if err != nil {
+		return nil, err
+	}
+
+	return dockerFetcher{
+		dockerBase: base,
+	}, nil
+}
+
+func (r *containerdResolver) Pusher(ctx context.Context, ref string) (remotes.Pusher, error) {
+	return nil, nil
+}
+
+type dockerBase struct {
+	refspec reference.Spec
+	base    []url.URL
+	token   string
+
+	client   *http.Client
+	useBasic bool
+	username string
+	secret   string
+}
+
+func (r *containerdResolver) base(refspec reference.Spec) (*dockerBase, error) {
+	var (
+		err              error
+		base             []url.URL
+		username, secret string
+	)
+
+	host := refspec.Hostname()
+	prefix := strings.TrimPrefix(refspec.Locator, host+"/")
+
+	if urls, ok := r.registry[host]; ok {
+		urls, err := r.getV2Urls(urls, prefix)
+		if err != nil {
+			return nil, fmt.Errorf("failed to fetch v2 urls: %v", err)
+		}
+		base = append(base, urls...)
+	} else if host == "docker.io" {
+		base = append(base, []url.URL{{Host: "registry-1.docker.io", Scheme: "https", Path: path.Join("/v2", prefix)}}...)
+	} else {
+		scheme := "https"
+		if r.plainHTTP || strings.HasPrefix(host, "localhost:") {
+			scheme = "http"
+		}
+		base = append(base, []url.URL{{Host: host, Scheme: scheme, Path: path.Join("/v2", prefix)}}...)
+	}
+
+	if r.credentials != nil {
+		username, secret, err = r.credentials(base[0].Host)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return &dockerBase{
+		refspec:  refspec,
+		base:     base,
+		client:   r.client,
+		username: username,
+		secret:   secret,
+	}, nil
+}
+
+func (r *dockerBase) url(ps ...string) string {
+	url := r.base[0]
+	url.Path = path.Join(url.Path, path.Join(ps...))
+	return url.String()
+}
+
+func (r *dockerBase) urls(ps ...string) []string {
+	urls := []string{}
+	for _, url := range r.base {
+		url.Path = path.Join(url.Path, path.Join(ps...))
+		urls = append(urls, url.String())
+	}
+	return urls
+}
+
+func (r *dockerBase) authorize(req *http.Request) {
+	if r.useBasic {
+		req.SetBasicAuth(r.username, r.secret)
+	} else if r.token != "" {
+		req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", r.token))
+	}
+}
+
+func (r *dockerBase) doRequest(ctx context.Context, req *http.Request) (*http.Response, error) {
+	ctx = log.WithLogger(ctx, log.G(ctx).WithField("url", req.URL.String()))
+	log.G(ctx).WithField("request.headers", req.Header).WithField("request.method", req.Method).Debug("do request")
+	r.authorize(req)
+	resp, err := ctxhttp.Do(ctx, r.client, req)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to do request")
+	}
+	log.G(ctx).WithFields(logrus.Fields{
+		"status":           resp.Status,
+		"response.headers": resp.Header,
+	}).Debug("fetch response received")
+	return resp, nil
+}
+
+func (r *dockerBase) doRequestWithRetries(ctx context.Context, req *http.Request, responses []*http.Response) (*http.Response, error) {
+	resp, err := r.doRequest(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+
+	responses = append(responses, resp)
+	req, err = r.retryRequest(ctx, req, responses)
+	if err != nil {
+		resp.Body.Close()
+		return nil, err
+	}
+	if req != nil {
+		resp.Body.Close()
+		return r.doRequestWithRetries(ctx, req, responses)
+	}
+	return resp, err
+}
+
+func (r *dockerBase) retryRequest(ctx context.Context, req *http.Request, responses []*http.Response) (*http.Request, error) {
+	if len(responses) > 5 {
+		return nil, nil
+	}
+	last := responses[len(responses)-1]
+	if last.StatusCode == http.StatusUnauthorized {
+		log.G(ctx).WithField("header", last.Header.Get("WWW-Authenticate")).Debug("Unauthorized")
+		for _, c := range parseAuthHeader(last.Header) {
+			if c.scheme == bearerAuth {
+				if err := invalidAuthorization(c, responses); err != nil {
+					r.token = ""
+					return nil, err
+				}
+				if err := r.setTokenAuth(ctx, c.parameters); err != nil {
+					return nil, err
+				}
+				return copyRequest(req)
+			} else if c.scheme == basicAuth {
+				if r.username != "" && r.secret != "" {
+					r.useBasic = true
+				}
+				return copyRequest(req)
+			}
+		}
+		return nil, nil
+	} else if last.StatusCode == http.StatusMethodNotAllowed && req.Method == http.MethodHead {
+		// Support registries which have not properly implemented the HEAD method for
+		// manifests endpoint
+		if strings.Contains(req.URL.Path, "/manifests/") {
+			// TODO: copy request?
+			req.Method = http.MethodGet
+			return copyRequest(req)
+		}
+	}
+
+	// TODO: Handle 50x errors accounting for attempt history
+	return nil, nil
+}
+
+func invalidAuthorization(c challenge, responses []*http.Response) error {
+	errStr := c.parameters["error"]
+	if errStr == "" {
+		return nil
+	}
+
+	n := len(responses)
+	if n == 1 || (n > 1 && !sameRequest(responses[n-2].Request, responses[n-1].Request)) {
+		return nil
+	}
+
+	return errors.Wrapf(ErrInvalidAuthorization, "server message: %s", errStr)
+}
+
+func sameRequest(r1, r2 *http.Request) bool {
+	if r1.Method != r2.Method {
+		return false
+	}
+	if *r1.URL != *r2.URL {
+		return false
+	}
+	return true
+}
+
+func copyRequest(req *http.Request) (*http.Request, error) {
+	ireq := *req
+	if ireq.GetBody != nil {
+		var err error
+		ireq.Body, err = ireq.GetBody()
+		if err != nil {
+			return nil, err
+		}
+	}
+	return &ireq, nil
+}
+
+func (r *dockerBase) setTokenAuth(ctx context.Context, params map[string]string) error {
+	realm, ok := params["realm"]
+	if !ok {
+		return errors.New("no realm specified for token auth challenge")
+	}
+
+	realmURL, err := url.Parse(realm)
+	if err != nil {
+		return fmt.Errorf("invalid token auth challenge realm: %s", err)
+	}
+
+	to := tokenOptions{
+		realm:   realmURL.String(),
+		service: params["service"],
+	}
+
+	to.scopes = getTokenScopes(ctx, params)
+	if len(to.scopes) == 0 {
+		return errors.Errorf("no scope specified for token auth challenge")
+	}
+	if r.secret != "" {
+		// Credential information is provided, use oauth POST endpoint
+		r.token, err = r.fetchTokenWithOAuth(ctx, to)
+		if err != nil {
+			return errors.Wrap(err, "failed to fetch oauth token")
+		}
+	} else {
+		// Do request anonymously
+		r.token, err = r.getToken(ctx, to)
+		if err != nil {
+			return errors.Wrap(err, "failed to fetch anonymous token")
+		}
+	}
+
+	return nil
+}
+
+type tokenOptions struct {
+	realm   string
+	service string
+	scopes  []string
+}
+
+type postTokenResponse struct {
+	AccessToken  string    `json:"access_token"`
+	RefreshToken string    `json:"refresh_token"`
+	ExpiresIn    int       `json:"expires_in"`
+	IssuedAt     time.Time `json:"issued_at"`
+	Scope        string    `json:"scope"`
+}
+
+func (r *dockerBase) fetchTokenWithOAuth(ctx context.Context, to tokenOptions) (string, error) {
+	form := url.Values{}
+	form.Set("scope", strings.Join(to.scopes, " "))
+	form.Set("service", to.service)
+	// TODO: Allow setting client_id
+	form.Set("client_id", "containerd-dist-tool")
+
+	if r.username == "" {
+		form.Set("grant_type", "refresh_token")
+		form.Set("refresh_token", r.secret)
+	} else {
+		form.Set("grant_type", "password")
+		form.Set("username", r.username)
+		form.Set("password", r.secret)
+	}
+
+	resp, err := ctxhttp.PostForm(ctx, r.client, to.realm, form)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+
+	// Registries without support for POST may return 404 for POST /v2/token.
+	// As of September 2017, GCR is known to return 404.
+	if (resp.StatusCode == 405 && r.username != "") || resp.StatusCode == 404 {
+		return r.getToken(ctx, to)
+	} else if resp.StatusCode < 200 || resp.StatusCode >= 400 {
+		b, _ := ioutil.ReadAll(io.LimitReader(resp.Body, 64000)) // 64KB
+		log.G(ctx).WithFields(logrus.Fields{
+			"status": resp.Status,
+			"body":   string(b),
+		}).Debugf("token request failed")
+		// TODO: handle error body and write debug output
+		return "", errors.Errorf("unexpected status: %s", resp.Status)
+	}
+
+	decoder := json.NewDecoder(resp.Body)
+
+	var tr postTokenResponse
+	if err = decoder.Decode(&tr); err != nil {
+		return "", fmt.Errorf("unable to decode token response: %s", err)
+	}
+
+	return tr.AccessToken, nil
+}
+
+type getTokenResponse struct {
+	Token        string    `json:"token"`
+	AccessToken  string    `json:"access_token"`
+	ExpiresIn    int       `json:"expires_in"`
+	IssuedAt     time.Time `json:"issued_at"`
+	RefreshToken string    `json:"refresh_token"`
+}
+
+// getToken fetches a token using a GET request
+func (r *dockerBase) getToken(ctx context.Context, to tokenOptions) (string, error) {
+	req, err := http.NewRequest("GET", to.realm, nil)
+	if err != nil {
+		return "", err
+	}
+
+	reqParams := req.URL.Query()
+
+	if to.service != "" {
+		reqParams.Add("service", to.service)
+	}
+
+	for _, scope := range to.scopes {
+		reqParams.Add("scope", scope)
+	}
+
+	if r.secret != "" {
+		req.SetBasicAuth(r.username, r.secret)
+	}
+
+	req.URL.RawQuery = reqParams.Encode()
+
+	resp, err := ctxhttp.Do(ctx, r.client, req)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode < 200 || resp.StatusCode >= 400 {
+		// TODO: handle error body and write debug output
+		return "", errors.Errorf("unexpected status: %s", resp.Status)
+	}
+
+	decoder := json.NewDecoder(resp.Body)
+
+	var tr getTokenResponse
+	if err = decoder.Decode(&tr); err != nil {
+		return "", fmt.Errorf("unable to decode token response: %s", err)
+	}
+
+	// `access_token` is equivalent to `token` and if both are specified
+	// the choice is undefined.  Canonicalize `access_token` by sticking
+	// things in `token`.
+	if tr.AccessToken != "" {
+		tr.Token = tr.AccessToken
+	}
+
+	if tr.Token == "" {
+		return "", ErrNoToken
+	}
+
+	return tr.Token, nil
+}
+
+func (r *containerdResolver) getV2Urls(urls []string, imagePath string) ([]url.URL, error) {
+	v2Urls := []url.URL{}
+	for _, u := range urls {
+		v2Url, err := url.Parse(u)
+		if err != nil {
+			return nil, fmt.Errorf("Failed to parse url during getv2 urls: %+v, err:%s", u, err)
+		}
+		v2Url.Path = path.Join("/v2", imagePath)
+		v2Urls = append(v2Urls, *v2Url)
+	}
+	return v2Urls, nil
+}

pkg/containerd/resolver/scope.go

@@ -0,0 +1,80 @@
+/*
+Copyright 2018 The Containerd Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// The corresponding file is in containerd/remote/docker/.
+// This package can be removed once a more feasible and hollistic resolver
+// is finalized in containerd.
+
+package resolver
+
+import (
+	"context"
+	"net/url"
+	"sort"
+	"strings"
+
+	"github.com/containerd/containerd/reference"
+)
+
+// repositoryScope returns a repository scope string such as "repository:foo/bar:pull"
+// for "host/foo/bar:baz".
+// When push is true, both pull and push are added to the scope.
+func repositoryScope(refspec reference.Spec, push bool) (string, error) {
+	u, err := url.Parse("dummy://" + refspec.Locator)
+	if err != nil {
+		return "", err
+	}
+	s := "repository:" + strings.TrimPrefix(u.Path, "/") + ":pull"
+	if push {
+		s += ",push"
+	}
+	return s, nil
+}
+
+// tokenScopesKey is used for the key for context.WithValue().
+// value: []string (e.g. {"registry:foo/bar:pull"})
+type tokenScopesKey struct{}
+
+// contextWithRepositoryScope returns a context with tokenScopesKey{} and the repository scope value.
+func contextWithRepositoryScope(ctx context.Context, refspec reference.Spec, push bool) (context.Context, error) {
+	s, err := repositoryScope(refspec, push)
+	if err != nil {
+		return nil, err
+	}
+	return context.WithValue(ctx, tokenScopesKey{}, []string{s}), nil
+}
+
+// getTokenScopes returns deduplicated and sorted scopes from ctx.Value(tokenScopesKey{}) and params["scope"].
+func getTokenScopes(ctx context.Context, params map[string]string) []string {
+	var scopes []string
+	if x := ctx.Value(tokenScopesKey{}); x != nil {
+		scopes = append(scopes, x.([]string)...)
+	}
+	if scope, ok := params["scope"]; ok {
+		for _, s := range scopes {
+			// Note: this comparison is unaware of the scope grammar (https://docs.docker.com/registry/spec/auth/scope/)
+			// So, "repository:foo/bar:pull,push" != "repository:foo/bar:push,pull", although semantically they are equal.
+			if s == scope {
+				// already appended
+				goto Sort
+			}
+		}
+		scopes = append(scopes, scope)
+	}
+Sort:
+	sort.Strings(scopes)
+	return scopes
+}

pkg/containerd/resolver/scope_test.go

@@ -0,0 +1,58 @@
+/*
+Copyright 2018 The Containerd Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// The corresponding file is in containerd/remote/docker/.
+// This package can be removed once a more feasible and hollistic resolver
+// is finalized in containerd.
+
+package resolver
+
+import (
+	"testing"
+
+	"github.com/containerd/containerd/reference"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestRepositoryScope(t *testing.T) {
+	testCases := []struct {
+		refspec  reference.Spec
+		push     bool
+		expected string
+	}{
+		{
+			refspec: reference.Spec{
+				Locator: "host/foo/bar",
+				Object:  "ignored",
+			},
+			push:     false,
+			expected: "repository:foo/bar:pull",
+		},
+		{
+			refspec: reference.Spec{
+				Locator: "host:4242/foo/bar",
+				Object:  "ignored",
+			},
+			push:     true,
+			expected: "repository:foo/bar:pull,push",
+		},
+	}
+	for _, x := range testCases {
+		actual, err := repositoryScope(x.refspec, x.push)
+		assert.NoError(t, err)
+		assert.Equal(t, x.expected, actual)
+	}
+}

pkg/containerd/resolver/status.go

@@ -0,0 +1,71 @@
+/*
+Copyright 2018 The Containerd Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// The corresponding file is in containerd/remote/docker/.
+// This package can be removed once a more feasible and hollistic resolver
+// is finalized in containerd.
+
+package resolver
+
+import (
+	"sync"
+
+	"github.com/containerd/containerd/content"
+	"github.com/containerd/containerd/errdefs"
+	"github.com/pkg/errors"
+)
+
+// Status of a content operation
+type Status struct {
+	content.Status
+
+	// UploadUUID is used by the Docker registry to reference blob uploads
+	UploadUUID string
+}
+
+// StatusTracker to track status of operations
+type StatusTracker interface {
+	GetStatus(string) (Status, error)
+	SetStatus(string, Status)
+}
+
+type memoryStatusTracker struct {
+	statuses map[string]Status
+	m        sync.Mutex
+}
+
+// NewInMemoryTracker returns a StatusTracker that tracks content status in-memory
+func NewInMemoryTracker() StatusTracker {
+	return &memoryStatusTracker{
+		statuses: map[string]Status{},
+	}
+}
+
+func (t *memoryStatusTracker) GetStatus(ref string) (Status, error) {
+	t.m.Lock()
+	defer t.m.Unlock()
+	status, ok := t.statuses[ref]
+	if !ok {
+		return Status{}, errors.Wrapf(errdefs.ErrNotFound, "status for ref %v", ref)
+	}
+	return status, nil
+}
+
+func (t *memoryStatusTracker) SetStatus(ref string, status Status) {
+	t.m.Lock()
+	t.statuses[ref] = status
+	t.m.Unlock()
+}

pkg/server/helpers.go

@@ -210,7 +210,6 @@ func (c *criContainerdService) localResolve(ctx context.Context, refOrID string)
 		if _, err := imagedigest.Parse(refOrID); err == nil {
 			return refOrID
 		}
-
 		return func(ref string) string {
 			// ref is not image id, try to resolve it locally.
 			normalized, err := util.NormalizeImageRef(ref)

pkg/server/image_pull.go

@@ -25,12 +25,12 @@ import (
 	"github.com/containerd/containerd"
 	"github.com/containerd/containerd/errdefs"
 	containerdimages "github.com/containerd/containerd/images"
-	"github.com/containerd/containerd/remotes/docker"
 	imagespec "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/sirupsen/logrus"
 	"golang.org/x/net/context"
 	"k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1/runtime"
 
+	containerdresolver "github.com/containerd/cri-containerd/pkg/containerd/resolver"
 	imagestore "github.com/containerd/cri-containerd/pkg/store/image"
 	"github.com/containerd/cri-containerd/pkg/util"
 )
@@ -83,15 +83,14 @@ func (c *criContainerdService) PullImage(ctx context.Context, r *runtime.PullIma
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse image reference %q: %v", imageRef, err)
 	}
-	// TODO(random-liu): [P0] Avoid concurrent pulling/removing on the same image reference.
 	ref := namedRef.String()
 	if ref != imageRef {
 		logrus.Debugf("PullImage using normalized image ref: %q", ref)
 	}
-
+	resolver := containerdresolver.NewResolver(containerdresolver.Options{
-	resolver := docker.NewResolver(docker.ResolverOptions{
 		Credentials: func(string) (string, string, error) { return ParseAuth(r.GetAuth()) },
 		Client:      http.DefaultClient,
+		Registry:    c.getResolverOptions(),
 	})
 	_, desc, err := resolver.Resolve(ctx, ref)
 	if err != nil {
@@ -212,3 +211,11 @@ func (c *criContainerdService) createImageReference(ctx context.Context, name st
 	_, err = c.client.ImageService().Update(ctx, img, "target")
 	return err
 }
+
+func (c *criContainerdService) getResolverOptions() map[string][]string {
+	options := make(map[string][]string)
+	for ns, mirror := range c.config.Mirrors {
+		options[ns] = append(options[ns], mirror.Endpoints...)
+	}
+	return options
+}