|
@@ -1,43 +1,42 @@
|
|
package distribution
|
|
package distribution
|
|
|
|
|
|
import (
|
|
import (
|
|
- "encoding/json"
|
|
|
|
"errors"
|
|
"errors"
|
|
"fmt"
|
|
"fmt"
|
|
"io"
|
|
"io"
|
|
|
|
+ "io/ioutil"
|
|
"net"
|
|
"net"
|
|
"net/url"
|
|
"net/url"
|
|
"strings"
|
|
"strings"
|
|
- "sync"
|
|
|
|
"time"
|
|
"time"
|
|
|
|
|
|
"github.com/Sirupsen/logrus"
|
|
"github.com/Sirupsen/logrus"
|
|
"github.com/docker/distribution/reference"
|
|
"github.com/docker/distribution/reference"
|
|
"github.com/docker/distribution/registry/client/transport"
|
|
"github.com/docker/distribution/registry/client/transport"
|
|
"github.com/docker/docker/distribution/metadata"
|
|
"github.com/docker/docker/distribution/metadata"
|
|
|
|
+ "github.com/docker/docker/distribution/xfer"
|
|
"github.com/docker/docker/image"
|
|
"github.com/docker/docker/image"
|
|
"github.com/docker/docker/image/v1"
|
|
"github.com/docker/docker/image/v1"
|
|
"github.com/docker/docker/layer"
|
|
"github.com/docker/docker/layer"
|
|
- "github.com/docker/docker/pkg/archive"
|
|
|
|
- "github.com/docker/docker/pkg/progressreader"
|
|
|
|
- "github.com/docker/docker/pkg/streamformatter"
|
|
|
|
|
|
+ "github.com/docker/docker/pkg/ioutils"
|
|
|
|
+ "github.com/docker/docker/pkg/progress"
|
|
"github.com/docker/docker/pkg/stringid"
|
|
"github.com/docker/docker/pkg/stringid"
|
|
"github.com/docker/docker/registry"
|
|
"github.com/docker/docker/registry"
|
|
|
|
+ "golang.org/x/net/context"
|
|
)
|
|
)
|
|
|
|
|
|
type v1Puller struct {
|
|
type v1Puller struct {
|
|
v1IDService *metadata.V1IDService
|
|
v1IDService *metadata.V1IDService
|
|
endpoint registry.APIEndpoint
|
|
endpoint registry.APIEndpoint
|
|
config *ImagePullConfig
|
|
config *ImagePullConfig
|
|
- sf *streamformatter.StreamFormatter
|
|
|
|
repoInfo *registry.RepositoryInfo
|
|
repoInfo *registry.RepositoryInfo
|
|
session *registry.Session
|
|
session *registry.Session
|
|
}
|
|
}
|
|
|
|
|
|
-func (p *v1Puller) Pull(ref reference.Named) (fallback bool, err error) {
|
|
|
|
|
|
+func (p *v1Puller) Pull(ctx context.Context, ref reference.Named) (fallback bool, err error) {
|
|
if _, isDigested := ref.(reference.Digested); isDigested {
|
|
if _, isDigested := ref.(reference.Digested); isDigested {
|
|
// Allowing fallback, because HTTPS v1 is before HTTP v2
|
|
// Allowing fallback, because HTTPS v1 is before HTTP v2
|
|
- return true, registry.ErrNoSupport{errors.New("Cannot pull by digest with v1 registry")}
|
|
|
|
|
|
+ return true, registry.ErrNoSupport{Err: errors.New("Cannot pull by digest with v1 registry")}
|
|
}
|
|
}
|
|
|
|
|
|
tlsConfig, err := p.config.RegistryService.TLSConfig(p.repoInfo.Index.Name)
|
|
tlsConfig, err := p.config.RegistryService.TLSConfig(p.repoInfo.Index.Name)
|
|
@@ -62,19 +61,17 @@ func (p *v1Puller) Pull(ref reference.Named) (fallback bool, err error) {
|
|
logrus.Debugf("Fallback from error: %s", err)
|
|
logrus.Debugf("Fallback from error: %s", err)
|
|
return true, err
|
|
return true, err
|
|
}
|
|
}
|
|
- if err := p.pullRepository(ref); err != nil {
|
|
|
|
|
|
+ if err := p.pullRepository(ctx, ref); err != nil {
|
|
// TODO(dmcgowan): Check if should fallback
|
|
// TODO(dmcgowan): Check if should fallback
|
|
return false, err
|
|
return false, err
|
|
}
|
|
}
|
|
- out := p.config.OutStream
|
|
|
|
- out.Write(p.sf.FormatStatus("", "%s: this image was pulled from a legacy registry. Important: This registry version will not be supported in future versions of docker.", p.repoInfo.CanonicalName.Name()))
|
|
|
|
|
|
+ progress.Message(p.config.ProgressOutput, "", p.repoInfo.CanonicalName.Name()+": this image was pulled from a legacy registry. Important: This registry version will not be supported in future versions of docker.")
|
|
|
|
|
|
return false, nil
|
|
return false, nil
|
|
}
|
|
}
|
|
|
|
|
|
-func (p *v1Puller) pullRepository(ref reference.Named) error {
|
|
|
|
- out := p.config.OutStream
|
|
|
|
- out.Write(p.sf.FormatStatus("", "Pulling repository %s", p.repoInfo.CanonicalName.Name()))
|
|
|
|
|
|
+func (p *v1Puller) pullRepository(ctx context.Context, ref reference.Named) error {
|
|
|
|
+ progress.Message(p.config.ProgressOutput, "", "Pulling repository "+p.repoInfo.CanonicalName.Name())
|
|
|
|
|
|
repoData, err := p.session.GetRepositoryData(p.repoInfo.RemoteName)
|
|
repoData, err := p.session.GetRepositoryData(p.repoInfo.RemoteName)
|
|
if err != nil {
|
|
if err != nil {
|
|
@@ -112,46 +109,18 @@ func (p *v1Puller) pullRepository(ref reference.Named) error {
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
- errors := make(chan error)
|
|
|
|
- layerDownloaded := make(chan struct{})
|
|
|
|
-
|
|
|
|
layersDownloaded := false
|
|
layersDownloaded := false
|
|
- var wg sync.WaitGroup
|
|
|
|
for _, imgData := range repoData.ImgList {
|
|
for _, imgData := range repoData.ImgList {
|
|
if isTagged && imgData.Tag != tagged.Tag() {
|
|
if isTagged && imgData.Tag != tagged.Tag() {
|
|
continue
|
|
continue
|
|
}
|
|
}
|
|
|
|
|
|
- wg.Add(1)
|
|
|
|
- go func(img *registry.ImgData) {
|
|
|
|
- p.downloadImage(out, repoData, img, layerDownloaded, errors)
|
|
|
|
- wg.Done()
|
|
|
|
- }(imgData)
|
|
|
|
- }
|
|
|
|
-
|
|
|
|
- go func() {
|
|
|
|
- wg.Wait()
|
|
|
|
- close(errors)
|
|
|
|
- }()
|
|
|
|
-
|
|
|
|
- var lastError error
|
|
|
|
-selectLoop:
|
|
|
|
- for {
|
|
|
|
- select {
|
|
|
|
- case err, ok := <-errors:
|
|
|
|
- if !ok {
|
|
|
|
- break selectLoop
|
|
|
|
- }
|
|
|
|
- lastError = err
|
|
|
|
- case <-layerDownloaded:
|
|
|
|
- layersDownloaded = true
|
|
|
|
|
|
+ err := p.downloadImage(ctx, repoData, imgData, &layersDownloaded)
|
|
|
|
+ if err != nil {
|
|
|
|
+ return err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
- if lastError != nil {
|
|
|
|
- return lastError
|
|
|
|
- }
|
|
|
|
-
|
|
|
|
localNameRef := p.repoInfo.LocalName
|
|
localNameRef := p.repoInfo.LocalName
|
|
if isTagged {
|
|
if isTagged {
|
|
localNameRef, err = reference.WithTag(localNameRef, tagged.Tag())
|
|
localNameRef, err = reference.WithTag(localNameRef, tagged.Tag())
|
|
@@ -159,194 +128,143 @@ selectLoop:
|
|
localNameRef = p.repoInfo.LocalName
|
|
localNameRef = p.repoInfo.LocalName
|
|
}
|
|
}
|
|
}
|
|
}
|
|
- writeStatus(localNameRef.String(), out, p.sf, layersDownloaded)
|
|
|
|
|
|
+ writeStatus(localNameRef.String(), p.config.ProgressOutput, layersDownloaded)
|
|
return nil
|
|
return nil
|
|
}
|
|
}
|
|
|
|
|
|
-func (p *v1Puller) downloadImage(out io.Writer, repoData *registry.RepositoryData, img *registry.ImgData, layerDownloaded chan struct{}, errors chan error) {
|
|
|
|
|
|
+func (p *v1Puller) downloadImage(ctx context.Context, repoData *registry.RepositoryData, img *registry.ImgData, layersDownloaded *bool) error {
|
|
if img.Tag == "" {
|
|
if img.Tag == "" {
|
|
logrus.Debugf("Image (id: %s) present in this repository but untagged, skipping", img.ID)
|
|
logrus.Debugf("Image (id: %s) present in this repository but untagged, skipping", img.ID)
|
|
- return
|
|
|
|
|
|
+ return nil
|
|
}
|
|
}
|
|
|
|
|
|
localNameRef, err := reference.WithTag(p.repoInfo.LocalName, img.Tag)
|
|
localNameRef, err := reference.WithTag(p.repoInfo.LocalName, img.Tag)
|
|
if err != nil {
|
|
if err != nil {
|
|
retErr := fmt.Errorf("Image (id: %s) has invalid tag: %s", img.ID, img.Tag)
|
|
retErr := fmt.Errorf("Image (id: %s) has invalid tag: %s", img.ID, img.Tag)
|
|
logrus.Debug(retErr.Error())
|
|
logrus.Debug(retErr.Error())
|
|
- errors <- retErr
|
|
|
|
|
|
+ return retErr
|
|
}
|
|
}
|
|
|
|
|
|
if err := v1.ValidateID(img.ID); err != nil {
|
|
if err := v1.ValidateID(img.ID); err != nil {
|
|
- errors <- err
|
|
|
|
- return
|
|
|
|
|
|
+ return err
|
|
}
|
|
}
|
|
|
|
|
|
- out.Write(p.sf.FormatProgress(stringid.TruncateID(img.ID), fmt.Sprintf("Pulling image (%s) from %s", img.Tag, p.repoInfo.CanonicalName.Name()), nil))
|
|
|
|
|
|
+ progress.Updatef(p.config.ProgressOutput, stringid.TruncateID(img.ID), "Pulling image (%s) from %s", img.Tag, p.repoInfo.CanonicalName.Name())
|
|
success := false
|
|
success := false
|
|
var lastErr error
|
|
var lastErr error
|
|
- var isDownloaded bool
|
|
|
|
for _, ep := range p.repoInfo.Index.Mirrors {
|
|
for _, ep := range p.repoInfo.Index.Mirrors {
|
|
ep += "v1/"
|
|
ep += "v1/"
|
|
- out.Write(p.sf.FormatProgress(stringid.TruncateID(img.ID), fmt.Sprintf("Pulling image (%s) from %s, mirror: %s", img.Tag, p.repoInfo.CanonicalName.Name(), ep), nil))
|
|
|
|
- if isDownloaded, err = p.pullImage(out, img.ID, ep, localNameRef); err != nil {
|
|
|
|
|
|
+ progress.Updatef(p.config.ProgressOutput, stringid.TruncateID(img.ID), fmt.Sprintf("Pulling image (%s) from %s, mirror: %s", img.Tag, p.repoInfo.CanonicalName.Name(), ep))
|
|
|
|
+ if err = p.pullImage(ctx, img.ID, ep, localNameRef, layersDownloaded); err != nil {
|
|
// Don't report errors when pulling from mirrors.
|
|
// Don't report errors when pulling from mirrors.
|
|
logrus.Debugf("Error pulling image (%s) from %s, mirror: %s, %s", img.Tag, p.repoInfo.CanonicalName.Name(), ep, err)
|
|
logrus.Debugf("Error pulling image (%s) from %s, mirror: %s, %s", img.Tag, p.repoInfo.CanonicalName.Name(), ep, err)
|
|
continue
|
|
continue
|
|
}
|
|
}
|
|
- if isDownloaded {
|
|
|
|
- layerDownloaded <- struct{}{}
|
|
|
|
- }
|
|
|
|
success = true
|
|
success = true
|
|
break
|
|
break
|
|
}
|
|
}
|
|
if !success {
|
|
if !success {
|
|
for _, ep := range repoData.Endpoints {
|
|
for _, ep := range repoData.Endpoints {
|
|
- out.Write(p.sf.FormatProgress(stringid.TruncateID(img.ID), fmt.Sprintf("Pulling image (%s) from %s, endpoint: %s", img.Tag, p.repoInfo.CanonicalName.Name(), ep), nil))
|
|
|
|
- if isDownloaded, err = p.pullImage(out, img.ID, ep, localNameRef); err != nil {
|
|
|
|
|
|
+ progress.Updatef(p.config.ProgressOutput, stringid.TruncateID(img.ID), "Pulling image (%s) from %s, endpoint: %s", img.Tag, p.repoInfo.CanonicalName.Name(), ep)
|
|
|
|
+ if err = p.pullImage(ctx, img.ID, ep, localNameRef, layersDownloaded); err != nil {
|
|
// It's not ideal that only the last error is returned, it would be better to concatenate the errors.
|
|
// It's not ideal that only the last error is returned, it would be better to concatenate the errors.
|
|
// As the error is also given to the output stream the user will see the error.
|
|
// As the error is also given to the output stream the user will see the error.
|
|
lastErr = err
|
|
lastErr = err
|
|
- out.Write(p.sf.FormatProgress(stringid.TruncateID(img.ID), fmt.Sprintf("Error pulling image (%s) from %s, endpoint: %s, %s", img.Tag, p.repoInfo.CanonicalName.Name(), ep, err), nil))
|
|
|
|
|
|
+ progress.Updatef(p.config.ProgressOutput, stringid.TruncateID(img.ID), "Error pulling image (%s) from %s, endpoint: %s, %s", img.Tag, p.repoInfo.CanonicalName.Name(), ep, err)
|
|
continue
|
|
continue
|
|
}
|
|
}
|
|
- if isDownloaded {
|
|
|
|
- layerDownloaded <- struct{}{}
|
|
|
|
- }
|
|
|
|
success = true
|
|
success = true
|
|
break
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if !success {
|
|
if !success {
|
|
err := fmt.Errorf("Error pulling image (%s) from %s, %v", img.Tag, p.repoInfo.CanonicalName.Name(), lastErr)
|
|
err := fmt.Errorf("Error pulling image (%s) from %s, %v", img.Tag, p.repoInfo.CanonicalName.Name(), lastErr)
|
|
- out.Write(p.sf.FormatProgress(stringid.TruncateID(img.ID), err.Error(), nil))
|
|
|
|
- errors <- err
|
|
|
|
- return
|
|
|
|
|
|
+ progress.Update(p.config.ProgressOutput, stringid.TruncateID(img.ID), err.Error())
|
|
|
|
+ return err
|
|
}
|
|
}
|
|
- out.Write(p.sf.FormatProgress(stringid.TruncateID(img.ID), "Download complete", nil))
|
|
|
|
|
|
+ progress.Update(p.config.ProgressOutput, stringid.TruncateID(img.ID), "Download complete")
|
|
|
|
+ return nil
|
|
}
|
|
}
|
|
|
|
|
|
-func (p *v1Puller) pullImage(out io.Writer, v1ID, endpoint string, localNameRef reference.Named) (layersDownloaded bool, err error) {
|
|
|
|
|
|
+func (p *v1Puller) pullImage(ctx context.Context, v1ID, endpoint string, localNameRef reference.Named, layersDownloaded *bool) (err error) {
|
|
var history []string
|
|
var history []string
|
|
history, err = p.session.GetRemoteHistory(v1ID, endpoint)
|
|
history, err = p.session.GetRemoteHistory(v1ID, endpoint)
|
|
if err != nil {
|
|
if err != nil {
|
|
- return false, err
|
|
|
|
|
|
+ return err
|
|
}
|
|
}
|
|
if len(history) < 1 {
|
|
if len(history) < 1 {
|
|
- return false, fmt.Errorf("empty history for image %s", v1ID)
|
|
|
|
|
|
+ return fmt.Errorf("empty history for image %s", v1ID)
|
|
}
|
|
}
|
|
- out.Write(p.sf.FormatProgress(stringid.TruncateID(v1ID), "Pulling dependent layers", nil))
|
|
|
|
- // FIXME: Try to stream the images?
|
|
|
|
- // FIXME: Launch the getRemoteImage() in goroutines
|
|
|
|
|
|
+ progress.Update(p.config.ProgressOutput, stringid.TruncateID(v1ID), "Pulling dependent layers")
|
|
|
|
|
|
var (
|
|
var (
|
|
- referencedLayers []layer.Layer
|
|
|
|
- parentID layer.ChainID
|
|
|
|
- newHistory []image.History
|
|
|
|
- img *image.V1Image
|
|
|
|
- imgJSON []byte
|
|
|
|
- imgSize int64
|
|
|
|
|
|
+ descriptors []xfer.DownloadDescriptor
|
|
|
|
+ newHistory []image.History
|
|
|
|
+ imgJSON []byte
|
|
|
|
+ imgSize int64
|
|
)
|
|
)
|
|
|
|
|
|
- defer func() {
|
|
|
|
- for _, l := range referencedLayers {
|
|
|
|
- layer.ReleaseAndLog(p.config.LayerStore, l)
|
|
|
|
- }
|
|
|
|
- }()
|
|
|
|
-
|
|
|
|
- layersDownloaded = false
|
|
|
|
-
|
|
|
|
- // Iterate over layers from top-most to bottom-most, checking if any
|
|
|
|
- // already exist on disk.
|
|
|
|
- var i int
|
|
|
|
- for i = 0; i != len(history); i++ {
|
|
|
|
- v1LayerID := history[i]
|
|
|
|
- // Do we have a mapping for this particular v1 ID on this
|
|
|
|
- // registry?
|
|
|
|
- if layerID, err := p.v1IDService.Get(v1LayerID, p.repoInfo.Index.Name); err == nil {
|
|
|
|
- // Does the layer actually exist
|
|
|
|
- if l, err := p.config.LayerStore.Get(layerID); err == nil {
|
|
|
|
- for j := i; j >= 0; j-- {
|
|
|
|
- logrus.Debugf("Layer already exists: %s", history[j])
|
|
|
|
- out.Write(p.sf.FormatProgress(stringid.TruncateID(history[j]), "Already exists", nil))
|
|
|
|
- }
|
|
|
|
- referencedLayers = append(referencedLayers, l)
|
|
|
|
- parentID = layerID
|
|
|
|
- break
|
|
|
|
- }
|
|
|
|
- }
|
|
|
|
- }
|
|
|
|
-
|
|
|
|
- needsDownload := i
|
|
|
|
-
|
|
|
|
// Iterate over layers, in order from bottom-most to top-most. Download
|
|
// Iterate over layers, in order from bottom-most to top-most. Download
|
|
- // config for all layers, and download actual layer data if needed.
|
|
|
|
- for i = len(history) - 1; i >= 0; i-- {
|
|
|
|
|
|
+ // config for all layers and create descriptors.
|
|
|
|
+ for i := len(history) - 1; i >= 0; i-- {
|
|
v1LayerID := history[i]
|
|
v1LayerID := history[i]
|
|
- imgJSON, imgSize, err = p.downloadLayerConfig(out, v1LayerID, endpoint)
|
|
|
|
|
|
+ imgJSON, imgSize, err = p.downloadLayerConfig(v1LayerID, endpoint)
|
|
if err != nil {
|
|
if err != nil {
|
|
- return layersDownloaded, err
|
|
|
|
- }
|
|
|
|
-
|
|
|
|
- img = &image.V1Image{}
|
|
|
|
- if err := json.Unmarshal(imgJSON, img); err != nil {
|
|
|
|
- return layersDownloaded, err
|
|
|
|
- }
|
|
|
|
-
|
|
|
|
- if i < needsDownload {
|
|
|
|
- l, err := p.downloadLayer(out, v1LayerID, endpoint, parentID, imgSize, &layersDownloaded)
|
|
|
|
-
|
|
|
|
- // Note: This needs to be done even in the error case to avoid
|
|
|
|
- // stale references to the layer.
|
|
|
|
- if l != nil {
|
|
|
|
- referencedLayers = append(referencedLayers, l)
|
|
|
|
- }
|
|
|
|
- if err != nil {
|
|
|
|
- return layersDownloaded, err
|
|
|
|
- }
|
|
|
|
-
|
|
|
|
- parentID = l.ChainID()
|
|
|
|
|
|
+ return err
|
|
}
|
|
}
|
|
|
|
|
|
// Create a new-style config from the legacy configs
|
|
// Create a new-style config from the legacy configs
|
|
h, err := v1.HistoryFromConfig(imgJSON, false)
|
|
h, err := v1.HistoryFromConfig(imgJSON, false)
|
|
if err != nil {
|
|
if err != nil {
|
|
- return layersDownloaded, err
|
|
|
|
|
|
+ return err
|
|
}
|
|
}
|
|
newHistory = append(newHistory, h)
|
|
newHistory = append(newHistory, h)
|
|
|
|
+
|
|
|
|
+ layerDescriptor := &v1LayerDescriptor{
|
|
|
|
+ v1LayerID: v1LayerID,
|
|
|
|
+ indexName: p.repoInfo.Index.Name,
|
|
|
|
+ endpoint: endpoint,
|
|
|
|
+ v1IDService: p.v1IDService,
|
|
|
|
+ layersDownloaded: layersDownloaded,
|
|
|
|
+ layerSize: imgSize,
|
|
|
|
+ session: p.session,
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ descriptors = append(descriptors, layerDescriptor)
|
|
}
|
|
}
|
|
|
|
|
|
rootFS := image.NewRootFS()
|
|
rootFS := image.NewRootFS()
|
|
- l := referencedLayers[len(referencedLayers)-1]
|
|
|
|
- for l != nil {
|
|
|
|
- rootFS.DiffIDs = append([]layer.DiffID{l.DiffID()}, rootFS.DiffIDs...)
|
|
|
|
- l = l.Parent()
|
|
|
|
|
|
+ resultRootFS, release, err := p.config.DownloadManager.Download(ctx, *rootFS, descriptors, p.config.ProgressOutput)
|
|
|
|
+ if err != nil {
|
|
|
|
+ return err
|
|
}
|
|
}
|
|
|
|
+ defer release()
|
|
|
|
|
|
- config, err := v1.MakeConfigFromV1Config(imgJSON, rootFS, newHistory)
|
|
|
|
|
|
+ config, err := v1.MakeConfigFromV1Config(imgJSON, &resultRootFS, newHistory)
|
|
if err != nil {
|
|
if err != nil {
|
|
- return layersDownloaded, err
|
|
|
|
|
|
+ return err
|
|
}
|
|
}
|
|
|
|
|
|
imageID, err := p.config.ImageStore.Create(config)
|
|
imageID, err := p.config.ImageStore.Create(config)
|
|
if err != nil {
|
|
if err != nil {
|
|
- return layersDownloaded, err
|
|
|
|
|
|
+ return err
|
|
}
|
|
}
|
|
|
|
|
|
if err := p.config.TagStore.AddTag(localNameRef, imageID, true); err != nil {
|
|
if err := p.config.TagStore.AddTag(localNameRef, imageID, true); err != nil {
|
|
- return layersDownloaded, err
|
|
|
|
|
|
+ return err
|
|
}
|
|
}
|
|
|
|
|
|
- return layersDownloaded, nil
|
|
|
|
|
|
+ return nil
|
|
}
|
|
}
|
|
|
|
|
|
-func (p *v1Puller) downloadLayerConfig(out io.Writer, v1LayerID, endpoint string) (imgJSON []byte, imgSize int64, err error) {
|
|
|
|
- out.Write(p.sf.FormatProgress(stringid.TruncateID(v1LayerID), "Pulling metadata", nil))
|
|
|
|
|
|
+func (p *v1Puller) downloadLayerConfig(v1LayerID, endpoint string) (imgJSON []byte, imgSize int64, err error) {
|
|
|
|
+ progress.Update(p.config.ProgressOutput, stringid.TruncateID(v1LayerID), "Pulling metadata")
|
|
|
|
|
|
retries := 5
|
|
retries := 5
|
|
for j := 1; j <= retries; j++ {
|
|
for j := 1; j <= retries; j++ {
|
|
imgJSON, imgSize, err := p.session.GetRemoteImageJSON(v1LayerID, endpoint)
|
|
imgJSON, imgSize, err := p.session.GetRemoteImageJSON(v1LayerID, endpoint)
|
|
if err != nil && j == retries {
|
|
if err != nil && j == retries {
|
|
- out.Write(p.sf.FormatProgress(stringid.TruncateID(v1LayerID), "Error pulling layer metadata", nil))
|
|
|
|
|
|
+ progress.Update(p.config.ProgressOutput, stringid.TruncateID(v1LayerID), "Error pulling layer metadata")
|
|
return nil, 0, err
|
|
return nil, 0, err
|
|
} else if err != nil {
|
|
} else if err != nil {
|
|
time.Sleep(time.Duration(j) * 500 * time.Millisecond)
|
|
time.Sleep(time.Duration(j) * 500 * time.Millisecond)
|
|
@@ -360,95 +278,66 @@ func (p *v1Puller) downloadLayerConfig(out io.Writer, v1LayerID, endpoint string
|
|
return nil, 0, nil
|
|
return nil, 0, nil
|
|
}
|
|
}
|
|
|
|
|
|
-func (p *v1Puller) downloadLayer(out io.Writer, v1LayerID, endpoint string, parentID layer.ChainID, layerSize int64, layersDownloaded *bool) (l layer.Layer, err error) {
|
|
|
|
- // ensure no two downloads of the same layer happen at the same time
|
|
|
|
- poolKey := "layer:" + v1LayerID
|
|
|
|
- broadcaster, found := p.config.Pool.add(poolKey)
|
|
|
|
- broadcaster.Add(out)
|
|
|
|
- if found {
|
|
|
|
- logrus.Debugf("Image (id: %s) pull is already running, skipping", v1LayerID)
|
|
|
|
- if err = broadcaster.Wait(); err != nil {
|
|
|
|
- return nil, err
|
|
|
|
- }
|
|
|
|
- layerID, err := p.v1IDService.Get(v1LayerID, p.repoInfo.Index.Name)
|
|
|
|
- if err != nil {
|
|
|
|
- return nil, err
|
|
|
|
- }
|
|
|
|
- // Does the layer actually exist
|
|
|
|
- l, err := p.config.LayerStore.Get(layerID)
|
|
|
|
- if err != nil {
|
|
|
|
- return nil, err
|
|
|
|
- }
|
|
|
|
- return l, nil
|
|
|
|
- }
|
|
|
|
|
|
+type v1LayerDescriptor struct {
|
|
|
|
+ v1LayerID string
|
|
|
|
+ indexName string
|
|
|
|
+ endpoint string
|
|
|
|
+ v1IDService *metadata.V1IDService
|
|
|
|
+ layersDownloaded *bool
|
|
|
|
+ layerSize int64
|
|
|
|
+ session *registry.Session
|
|
|
|
+}
|
|
|
|
|
|
- // This must use a closure so it captures the value of err when
|
|
|
|
- // the function returns, not when the 'defer' is evaluated.
|
|
|
|
- defer func() {
|
|
|
|
- p.config.Pool.removeWithError(poolKey, err)
|
|
|
|
- }()
|
|
|
|
|
|
+func (ld *v1LayerDescriptor) Key() string {
|
|
|
|
+ return "v1:" + ld.v1LayerID
|
|
|
|
+}
|
|
|
|
|
|
- retries := 5
|
|
|
|
- for j := 1; j <= retries; j++ {
|
|
|
|
- // Get the layer
|
|
|
|
- status := "Pulling fs layer"
|
|
|
|
- if j > 1 {
|
|
|
|
- status = fmt.Sprintf("Pulling fs layer [retries: %d]", j)
|
|
|
|
- }
|
|
|
|
- broadcaster.Write(p.sf.FormatProgress(stringid.TruncateID(v1LayerID), status, nil))
|
|
|
|
- layerReader, err := p.session.GetRemoteImageLayer(v1LayerID, endpoint, layerSize)
|
|
|
|
|
|
+func (ld *v1LayerDescriptor) ID() string {
|
|
|
|
+ return stringid.TruncateID(ld.v1LayerID)
|
|
|
|
+}
|
|
|
|
+
|
|
|
|
+func (ld *v1LayerDescriptor) DiffID() (layer.DiffID, error) {
|
|
|
|
+ return ld.v1IDService.Get(ld.v1LayerID, ld.indexName)
|
|
|
|
+}
|
|
|
|
+
|
|
|
|
+func (ld *v1LayerDescriptor) Download(ctx context.Context, progressOutput progress.Output) (io.ReadCloser, int64, error) {
|
|
|
|
+ progress.Update(progressOutput, ld.ID(), "Pulling fs layer")
|
|
|
|
+ layerReader, err := ld.session.GetRemoteImageLayer(ld.v1LayerID, ld.endpoint, ld.layerSize)
|
|
|
|
+ if err != nil {
|
|
|
|
+ progress.Update(progressOutput, ld.ID(), "Error pulling dependent layers")
|
|
if uerr, ok := err.(*url.Error); ok {
|
|
if uerr, ok := err.(*url.Error); ok {
|
|
err = uerr.Err
|
|
err = uerr.Err
|
|
}
|
|
}
|
|
- if terr, ok := err.(net.Error); ok && terr.Timeout() && j < retries {
|
|
|
|
- time.Sleep(time.Duration(j) * 500 * time.Millisecond)
|
|
|
|
- continue
|
|
|
|
- } else if err != nil {
|
|
|
|
- broadcaster.Write(p.sf.FormatProgress(stringid.TruncateID(v1LayerID), "Error pulling dependent layers", nil))
|
|
|
|
- return nil, err
|
|
|
|
- }
|
|
|
|
- *layersDownloaded = true
|
|
|
|
- defer layerReader.Close()
|
|
|
|
-
|
|
|
|
- reader := progressreader.New(progressreader.Config{
|
|
|
|
- In: layerReader,
|
|
|
|
- Out: broadcaster,
|
|
|
|
- Formatter: p.sf,
|
|
|
|
- Size: layerSize,
|
|
|
|
- NewLines: false,
|
|
|
|
- ID: stringid.TruncateID(v1LayerID),
|
|
|
|
- Action: "Downloading",
|
|
|
|
- })
|
|
|
|
-
|
|
|
|
- inflatedLayerData, err := archive.DecompressStream(reader)
|
|
|
|
- if err != nil {
|
|
|
|
- return nil, fmt.Errorf("could not get decompression stream: %v", err)
|
|
|
|
- }
|
|
|
|
-
|
|
|
|
- l, err := p.config.LayerStore.Register(inflatedLayerData, parentID)
|
|
|
|
- if err != nil {
|
|
|
|
- return nil, fmt.Errorf("failed to register layer: %v", err)
|
|
|
|
|
|
+ if terr, ok := err.(net.Error); ok && terr.Timeout() {
|
|
|
|
+ return nil, 0, err
|
|
}
|
|
}
|
|
- logrus.Debugf("layer %s registered successfully", l.DiffID())
|
|
|
|
|
|
+ return nil, 0, xfer.DoNotRetry{Err: err}
|
|
|
|
+ }
|
|
|
|
+ *ld.layersDownloaded = true
|
|
|
|
|
|
- if terr, ok := err.(net.Error); ok && terr.Timeout() && j < retries {
|
|
|
|
- time.Sleep(time.Duration(j) * 500 * time.Millisecond)
|
|
|
|
- continue
|
|
|
|
- } else if err != nil {
|
|
|
|
- broadcaster.Write(p.sf.FormatProgress(stringid.TruncateID(v1LayerID), "Error downloading dependent layers", nil))
|
|
|
|
- return nil, err
|
|
|
|
- }
|
|
|
|
|
|
+ tmpFile, err := ioutil.TempFile("", "GetImageBlob")
|
|
|
|
+ if err != nil {
|
|
|
|
+ layerReader.Close()
|
|
|
|
+ return nil, 0, err
|
|
|
|
+ }
|
|
|
|
|
|
- // Cache mapping from this v1 ID to content-addressable layer ID
|
|
|
|
- if err := p.v1IDService.Set(v1LayerID, p.repoInfo.Index.Name, l.ChainID()); err != nil {
|
|
|
|
- return nil, err
|
|
|
|
- }
|
|
|
|
|
|
+ reader := progress.NewProgressReader(ioutils.NewCancelReadCloser(ctx, layerReader), progressOutput, ld.layerSize, ld.ID(), "Downloading")
|
|
|
|
+ defer reader.Close()
|
|
|
|
|
|
- broadcaster.Write(p.sf.FormatProgress(stringid.TruncateID(v1LayerID), "Download complete", nil))
|
|
|
|
- broadcaster.Close()
|
|
|
|
- return l, nil
|
|
|
|
|
|
+ _, err = io.Copy(tmpFile, reader)
|
|
|
|
+ if err != nil {
|
|
|
|
+ return nil, 0, err
|
|
}
|
|
}
|
|
|
|
|
|
- // not reached
|
|
|
|
- return nil, nil
|
|
|
|
|
|
+ progress.Update(progressOutput, ld.ID(), "Download complete")
|
|
|
|
+
|
|
|
|
+ logrus.Debugf("Downloaded %s to tempfile %s", ld.ID(), tmpFile.Name())
|
|
|
|
+
|
|
|
|
+ tmpFile.Seek(0, 0)
|
|
|
|
+ return ioutils.NewReadCloserWrapper(tmpFile, tmpFileCloser(tmpFile)), ld.layerSize, nil
|
|
|
|
+}
|
|
|
|
+
|
|
|
|
+func (ld *v1LayerDescriptor) Registered(diffID layer.DiffID) {
|
|
|
|
+ // Cache mapping from this layer's DiffID to the blobsum
|
|
|
|
+ ld.v1IDService.Set(ld.v1LayerID, ld.indexName, diffID)
|
|
}
|
|
}
|