func (c DockerRegistryClient) PullLayer(layerDigest image.Digest) (os.FileInfo, error) {
	return c.pullLayerHelper(layerDigest, false)
}
PullLayer pulls image layer from the registry, and verifies that the contents
of that layer match the digest of the manifest.
If the layer already exists in the imagestore, the download is skipped.
package registry
import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"net/url"
	"os"
	"regexp"
	"time"
	"github.com/cenkalti/backoff"
	"github.com/juju/ratelimit"
	"github.com/uber/makisu/lib/concurrency"
	"github.com/uber/makisu/lib/docker/image"
	"github.com/uber/makisu/lib/log"
	"github.com/uber/makisu/lib/storage"
	"github.com/uber/makisu/lib/utils"
	"github.com/uber/makisu/lib/utils/httputil"
)
const (
	baseManifestQuery = "http://%s/v2/%s/manifests/%s"
	baseLayerQuery    = "http://%s/v2/%s/blobs/%s"
	baseStartQuery    = "http://%s/v2/%s/blobs/uploads/"
)
type Client interface {
	Pull(tag string) (*image.DistributionManifest, error)
	Push(tag string) error
	PullManifest(tag string) (*image.DistributionManifest, error)
	PushManifest(tag string, manifest *image.DistributionManifest) error
	PullLayer(layerDigest image.Digest) (os.FileInfo, error)
	PushLayer(layerDigest image.Digest) error
	PullImageConfig(layerDigest image.Digest) (os.FileInfo, error)
	PushImageConfig(layerDigest image.Digest) error
}
var _ Client = (*DockerRegistryClient)(nil)
type DockerRegistryClient struct {
	config     Config
	store      *storage.ImageStore
	registry   string
	repository string
	client *http.Client
}
func New(store *storage.ImageStore, registry, repository string) *DockerRegistryClient {
	return newClient(store, registry, repository, nil)
}
func NewWithClient(store *storage.ImageStore, registry, repository string, client *http.Client) *DockerRegistryClient {
	return newClient(store, registry, repository, client)
}
func newClient(store *storage.ImageStore, registry, repository string, client *http.Client) *DockerRegistryClient {
	config := Config{}
	if registry == image.DockerHubRegistry {
		config = DefaultDockerHubConfiguration
	}
	repoConfig, ok := ConfigurationMap[registry]
	if ok {
		for repo, c := range repoConfig {
			r := regexp.MustCompile(repo)
			if r.MatchString(repository) {
				config = c
				break
			}
		}
	}
	return &DockerRegistryClient{
		config:     config.applyDefaults(),
		registry:   registry,
		repository: repository,
		store:      store,
		client:     client,
	}
}
func (c DockerRegistryClient) Pull(tag string) (*image.DistributionManifest, error) {
	name := image.NewImageName(c.registry, c.repository, tag)
	log.Infof("* Started pulling image %s", name)
	starttime := time.Now()
	manifest, err := c.PullManifest(tag)
	if err != nil {
		return nil, fmt.Errorf("pull manifest: %s", err)
	}
	multiError := utils.NewMultiErrors()
	workers := concurrency.NewWorkerPool(c.config.Concurrency)
	layerSet := make(map[string]interface{})
	for _, layer := range manifest.GetLayerDigests() {
		l := layer
		if _, ok := layerSet[l.Hex()]; ok {
			continue
		} else {
			layerSet[l.Hex()] = struct{}{}
		}
		workers.Do(func() {
			if _, err := c.PullLayer(l); err != nil {
				multiError.Add(fmt.Errorf("pull layer %s: %s", l, err))
				workers.Stop()
				return
			}
		})
	}
	l := manifest.GetConfigDigest()
	workers.Do(func() {
		if _, err := c.PullLayer(l); err != nil {
			multiError.Add(fmt.Errorf("pull image config %s: %s", l, err))
			workers.Stop()
			return
		}
	})
	workers.Wait()
	if err := multiError.Collect(); err != nil {
		return nil, err
	}
	if err := c.saveManifest(tag, manifest); err != nil {
		return nil, fmt.Errorf("save manifest: %s", err)
	}
	log.Infow(fmt.Sprintf("* Pulled image %s", name), "duration", time.Since(starttime))
	return manifest, nil
}
func (c DockerRegistryClient) Push(tag string) error {
	name := image.NewImageName(c.registry, c.repository, tag)
	log.Infof("* Started pushing image %s", name)
	starttime := time.Now()
	if found, err := c.manifestExists(tag); err != nil {
		return fmt.Errorf("check manifest exists for image %s: %s", name, err)
	} else if found {
		log.Infof("* Image %s already exists, overwriting", name)
	}
	manifest, err := c.loadManifest(tag)
	if err != nil {
		return fmt.Errorf("load manifest: %s", err)
	}
	multiError := utils.NewMultiErrors()
	workers := concurrency.NewWorkerPool(c.config.Concurrency)
	layerSet := make(map[string]interface{})
	for _, layer := range manifest.GetLayerDigests() {
		l := layer
		if _, ok := layerSet[l.Hex()]; ok {
			continue
		} else {
			layerSet[l.Hex()] = struct{}{}
		}
		workers.Do(func() {
			if err := c.PushLayer(l); err != nil {
				multiError.Add(fmt.Errorf("push layer %s: %s", l, err))
				workers.Stop()
				return
			}
		})
	}
	l := manifest.GetConfigDigest()
	workers.Do(func() {
		if err := c.PushImageConfig(l); err != nil {
			multiError.Add(fmt.Errorf("push image config %s: %s", l, err))
			workers.Stop()
			return
		}
	})
	workers.Wait()
	if err := multiError.Collect(); err != nil {
		return err
	}
	if err := c.PushManifest(tag, manifest); err != nil {
		return fmt.Errorf("push manifest: %s", err)
	}
	log.Infow(fmt.Sprintf("* Pushed image %s", name), "duration", time.Since(starttime))
	return nil
}
func (c DockerRegistryClient) PullManifest(tag string) (*image.DistributionManifest, error) {
	opt, err := c.config.Security.GetHTTPOption(c.registry, c.repository)
	if err != nil {
		return nil, fmt.Errorf("get security opt: %s", err)
	}
	URL := fmt.Sprintf(baseManifestQuery, c.registry, c.repository, tag)
	resp, err := httputil.Send(
		"GET",
		URL,
		httputil.SendClient(c.client),
		opt,
		httputil.SendTimeout(c.config.Timeout),
		c.config.sendRetry(),
		httputil.SendAcceptedCodes(http.StatusOK, http.StatusNotFound, http.StatusBadRequest),
		httputil.SendHeaders(map[string]string{"Accept": image.MediaTypeManifest}))
	if err != nil {
		return nil, fmt.Errorf("http send error: %s", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusBadRequest {
		return nil, fmt.Errorf("manifest not found")
	} else if resp.StatusCode != 200 {
		return nil, fmt.Errorf("bad pull manifest request resp code: %d", resp.StatusCode)
	}
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return nil, fmt.Errorf("read resp body: %s", err)
	}
	ctHeader := resp.Header.Get("Content-Type")
	manifest, _, err := image.UnmarshalDistributionManifest(ctHeader, body)
	if err != nil {
		return nil, fmt.Errorf("unmarshal distribution manifest: %s", err)
	}
	return &manifest, nil
}
func (c DockerRegistryClient) PushManifest(tag string, manifest *image.DistributionManifest) error {
	payload, err := json.MarshalIndent(manifest, "", "   ")
	if err != nil {
		return fmt.Errorf("marshal manifest: %s", err)
	}
	headers := map[string]string{
		"Content-Type": manifest.MediaType,
		"Host":         c.registry,
	}
	opt, err := c.config.Security.GetHTTPOption(c.registry, c.repository)
	if err != nil {
		return fmt.Errorf("get security opt: %s", err)
	}
	URL := fmt.Sprintf(baseManifestQuery, c.registry, c.repository, tag)
	resp, err := httputil.Send(
		"PUT",
		URL,
		httputil.SendClient(c.client),
		opt,
		httputil.SendTimeout(c.config.Timeout),
		c.config.sendRetry(),
		httputil.SendAcceptedCodes(http.StatusOK, http.StatusCreated),
		httputil.SendHeaders(headers),
		httputil.SendBody(bytes.NewReader(payload)))
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	return nil
}