distribution/pull_v1.go

   1 package distribution
   2
   3 import (
   4         "errors"
   5         "fmt"
   6         "io"
   7         "io/ioutil"
   8         "net"
   9         "net/url"
  10         "os"
  11         "strings"
  12         "time"
  13
  14         "github.com/Sirupsen/logrus"
  15         "github.com/docker/distribution/reference"
  16         "github.com/docker/distribution/registry/client/transport"
  17         "github.com/docker/docker/distribution/metadata"
  18         "github.com/docker/docker/distribution/xfer"
  19         "github.com/docker/docker/dockerversion"
  20         "github.com/docker/docker/image"
  21         "github.com/docker/docker/image/v1"
  22         "github.com/docker/docker/layer"
  23         "github.com/docker/docker/pkg/ioutils"
  24         "github.com/docker/docker/pkg/progress"
  25         "github.com/docker/docker/pkg/stringid"
  26         "github.com/docker/docker/registry"
  27         "golang.org/x/net/context"
  28 )
  29
  30 type v1Puller struct {
  31         v1IDService *metadata.V1IDService
  32         endpoint    registry.APIEndpoint
  33         config      *ImagePullConfig
  34         repoInfo    *registry.RepositoryInfo
  35         session     *registry.Session
  36 }
  37
  38 func (p *v1Puller) Pull(ctx context.Context, ref reference.Named) error {
  39         if _, isCanonical := ref.(reference.Canonical); isCanonical {
  40                 // Allowing fallback, because HTTPS v1 is before HTTP v2
  41                 return fallbackError{err: ErrNoSupport{Err: errors.New("Cannot pull by digest with v1 registry")}}
  42         }
  43
  44         tlsConfig, err := p.config.RegistryService.TLSConfig(p.repoInfo.Index.Name)
  45         if err != nil {
  46                 return err
  47         }
  48         // Adds Docker-specific headers as well as user-specified headers (metaHeaders)
  49         tr := transport.NewTransport(
  50                 // TODO(tiborvass): was ReceiveTimeout
  51                 registry.NewTransport(tlsConfig),
  52                 registry.DockerHeaders(dockerversion.DockerUserAgent(ctx), p.config.MetaHeaders)...,
  53         )
  54         client := registry.HTTPClient(tr)
  55         v1Endpoint, err := p.endpoint.ToV1Endpoint(dockerversion.DockerUserAgent(ctx), p.config.MetaHeaders)
  56         if err != nil {
  57                 logrus.Debugf("Could not get v1 endpoint: %v", err)
  58                 return fallbackError{err: err}
  59         }
  60         p.session, err = registry.NewSession(client, p.config.AuthConfig, v1Endpoint)
  61         if err != nil {
  62                 // TODO(dmcgowan): Check if should fallback
  63                 logrus.Debugf("Fallback from error: %s", err)
  64                 return fallbackError{err: err}
  65         }
  66         if err := p.pullRepository(ctx, ref); err != nil {
  67                 // TODO(dmcgowan): Check if should fallback
  68                 return err
  69         }
  70         progress.Message(p.config.ProgressOutput, "", p.repoInfo.Name.Name()+": this image was pulled from a legacy registry.  Important: This registry version will not be supported in future versions of docker.")
  71
  72         return nil
  73 }
  74
  75 func (p *v1Puller) pullRepository(ctx context.Context, ref reference.Named) error {
  76         progress.Message(p.config.ProgressOutput, "", "Pulling repository "+p.repoInfo.Name.Name())
  77
  78         tagged, isTagged := ref.(reference.NamedTagged)
  79
  80         repoData, err := p.session.GetRepositoryData(p.repoInfo.Name)
  81         if err != nil {
  82                 if strings.Contains(err.Error(), "HTTP code: 404") {
  83                         if isTagged {
  84                                 return fmt.Errorf("Error: image %s:%s not found", reference.Path(p.repoInfo.Name), tagged.Tag())
  85                         }
  86                         return fmt.Errorf("Error: image %s not found", reference.Path(p.repoInfo.Name))
  87                 }
  88                 // Unexpected HTTP error
  89                 return err
  90         }
  91
  92         logrus.Debug("Retrieving the tag list")
  93         var tagsList map[string]string
  94         if !isTagged {
  95                 tagsList, err = p.session.GetRemoteTags(repoData.Endpoints, p.repoInfo.Name)
  96         } else {
  97                 var tagID string
  98                 tagsList = make(map[string]string)
  99                 tagID, err = p.session.GetRemoteTag(repoData.Endpoints, p.repoInfo.Name, tagged.Tag())
 100                 if err == registry.ErrRepoNotFound {
 101                         return fmt.Errorf("Tag %s not found in repository %s", tagged.Tag(), p.repoInfo.Name.Name())
 102                 }
 103                 tagsList[tagged.Tag()] = tagID
 104         }
 105         if err != nil {
 106                 logrus.Errorf("unable to get remote tags: %s", err)
 107                 return err
 108         }
 109
 110         for tag, id := range tagsList {
 111                 repoData.ImgList[id] = &registry.ImgData{
 112                         ID:       id,
 113                         Tag:      tag,
 114                         Checksum: "",
 115                 }
 116         }
 117
 118         layersDownloaded := false
 119         for _, imgData := range repoData.ImgList {
 120                 if isTagged && imgData.Tag != tagged.Tag() {
 121                         continue
 122                 }
 123
 124                 err := p.downloadImage(ctx, repoData, imgData, &layersDownloaded)
 125                 if err != nil {
 126                         return err
 127                 }
 128         }
 129
 130         writeStatus(reference.FamiliarString(ref), p.config.ProgressOutput, layersDownloaded)
 131         return nil
 132 }
 133
 134 func (p *v1Puller) downloadImage(ctx context.Context, repoData *registry.RepositoryData, img *registry.ImgData, layersDownloaded *bool) error {
 135         if img.Tag == "" {
 136                 logrus.Debugf("Image (id: %s) present in this repository but untagged, skipping", img.ID)
 137                 return nil
 138         }
 139
 140         localNameRef, err := reference.WithTag(p.repoInfo.Name, img.Tag)
 141         if err != nil {
 142                 retErr := fmt.Errorf("Image (id: %s) has invalid tag: %s", img.ID, img.Tag)
 143                 logrus.Debug(retErr.Error())
 144                 return retErr
 145         }
 146
 147         if err := v1.ValidateID(img.ID); err != nil {
 148                 return err
 149         }
 150
 151         progress.Updatef(p.config.ProgressOutput, stringid.TruncateID(img.ID), "Pulling image (%s) from %s", img.Tag, p.repoInfo.Name.Name())
 152         success := false
 153         var lastErr error
 154         for _, ep := range p.repoInfo.Index.Mirrors {
 155                 ep += "v1/"
 156                 progress.Updatef(p.config.ProgressOutput, stringid.TruncateID(img.ID), fmt.Sprintf("Pulling image (%s) from %s, mirror: %s", img.Tag, p.repoInfo.Name.Name(), ep))
 157                 if err = p.pullImage(ctx, img.ID, ep, localNameRef, layersDownloaded); err != nil {
 158                         // Don't report errors when pulling from mirrors.
 159                         logrus.Debugf("Error pulling image (%s) from %s, mirror: %s, %s", img.Tag, p.repoInfo.Name.Name(), ep, err)
 160                         continue
 161                 }
 162                 success = true
 163                 break
 164         }
 165         if !success {
 166                 for _, ep := range repoData.Endpoints {
 167                         progress.Updatef(p.config.ProgressOutput, stringid.TruncateID(img.ID), "Pulling image (%s) from %s, endpoint: %s", img.Tag, p.repoInfo.Name.Name(), ep)
 168                         if err = p.pullImage(ctx, img.ID, ep, localNameRef, layersDownloaded); err != nil {
 169                                 // It's not ideal that only the last error is returned, it would be better to concatenate the errors.
 170                                 // As the error is also given to the output stream the user will see the error.
 171                                 lastErr = err
 172                                 progress.Updatef(p.config.ProgressOutput, stringid.TruncateID(img.ID), "Error pulling image (%s) from %s, endpoint: %s, %s", img.Tag, p.repoInfo.Name.Name(), ep, err)
 173                                 continue
 174                         }
 175                         success = true
 176                         break
 177                 }
 178         }
 179         if !success {
 180                 err := fmt.Errorf("Error pulling image (%s) from %s, %v", img.Tag, p.repoInfo.Name.Name(), lastErr)
 181                 progress.Update(p.config.ProgressOutput, stringid.TruncateID(img.ID), err.Error())
 182                 return err
 183         }
 184         return nil
 185 }
 186
 187 func (p *v1Puller) pullImage(ctx context.Context, v1ID, endpoint string, localNameRef reference.Named, layersDownloaded *bool) (err error) {
 188         var history []string
 189         history, err = p.session.GetRemoteHistory(v1ID, endpoint)
 190         if err != nil {
 191                 return err
 192         }
 193         if len(history) < 1 {
 194                 return fmt.Errorf("empty history for image %s", v1ID)
 195         }
 196         progress.Update(p.config.ProgressOutput, stringid.TruncateID(v1ID), "Pulling dependent layers")
 197
 198         var (
 199                 descriptors []xfer.DownloadDescriptor
 200                 newHistory  []image.History
 201                 imgJSON     []byte
 202                 imgSize     int64
 203         )
 204
 205         // Iterate over layers, in order from bottom-most to top-most. Download
 206         // config for all layers and create descriptors.
 207         for i := len(history) - 1; i >= 0; i-- {
 208                 v1LayerID := history[i]
 209                 imgJSON, imgSize, err = p.downloadLayerConfig(v1LayerID, endpoint)
 210                 if err != nil {
 211                         return err
 212                 }
 213
 214                 // Create a new-style config from the legacy configs
 215                 h, err := v1.HistoryFromConfig(imgJSON, false)
 216                 if err != nil {
 217                         return err
 218                 }
 219                 newHistory = append(newHistory, h)
 220
 221                 layerDescriptor := &v1LayerDescriptor{
 222                         v1LayerID:        v1LayerID,
 223                         indexName:        p.repoInfo.Index.Name,
 224                         endpoint:         endpoint,
 225                         v1IDService:      p.v1IDService,
 226                         layersDownloaded: layersDownloaded,
 227                         layerSize:        imgSize,
 228                         session:          p.session,
 229                 }
 230
 231                 descriptors = append(descriptors, layerDescriptor)
 232         }
 233
 234         rootFS := image.NewRootFS()
 235         resultRootFS, release, err := p.config.DownloadManager.Download(ctx, *rootFS, "", descriptors, p.config.ProgressOutput)
 236         if err != nil {
 237                 return err
 238         }
 239         defer release()
 240
 241         config, err := v1.MakeConfigFromV1Config(imgJSON, &resultRootFS, newHistory)
 242         if err != nil {
 243                 return err
 244         }
 245
 246         imageID, err := p.config.ImageStore.Put(config)
 247         if err != nil {
 248                 return err
 249         }
 250
 251         if p.config.ReferenceStore != nil {
 252                 if err := p.config.ReferenceStore.AddTag(localNameRef, imageID, true); err != nil {
 253                         return err
 254                 }
 255         }
 256
 257         return nil
 258 }
 259
 260 func (p *v1Puller) downloadLayerConfig(v1LayerID, endpoint string) (imgJSON []byte, imgSize int64, err error) {
 261         progress.Update(p.config.ProgressOutput, stringid.TruncateID(v1LayerID), "Pulling metadata")
 262
 263         retries := 5
 264         for j := 1; j <= retries; j++ {
 265                 imgJSON, imgSize, err := p.session.GetRemoteImageJSON(v1LayerID, endpoint)
 266                 if err != nil && j == retries {
 267                         progress.Update(p.config.ProgressOutput, stringid.TruncateID(v1LayerID), "Error pulling layer metadata")
 268                         return nil, 0, err
 269                 } else if err != nil {
 270                         time.Sleep(time.Duration(j) * 500 * time.Millisecond)
 271                         continue
 272                 }
 273
 274                 return imgJSON, imgSize, nil
 275         }
 276
 277         // not reached
 278         return nil, 0, nil
 279 }
 280
 281 type v1LayerDescriptor struct {
 282         v1LayerID        string
 283         indexName        string
 284         endpoint         string
 285         v1IDService      *metadata.V1IDService
 286         layersDownloaded *bool
 287         layerSize        int64
 288         session          *registry.Session
 289         tmpFile          *os.File
 290 }
 291
 292 func (ld *v1LayerDescriptor) Key() string {
 293         return "v1:" + ld.v1LayerID
 294 }
 295
 296 func (ld *v1LayerDescriptor) ID() string {
 297         return stringid.TruncateID(ld.v1LayerID)
 298 }
 299
 300 func (ld *v1LayerDescriptor) DiffID() (layer.DiffID, error) {
 301         return ld.v1IDService.Get(ld.v1LayerID, ld.indexName)
 302 }
 303
 304 func (ld *v1LayerDescriptor) Download(ctx context.Context, progressOutput progress.Output) (io.ReadCloser, int64, error) {
 305         progress.Update(progressOutput, ld.ID(), "Pulling fs layer")
 306         layerReader, err := ld.session.GetRemoteImageLayer(ld.v1LayerID, ld.endpoint, ld.layerSize)
 307         if err != nil {
 308                 progress.Update(progressOutput, ld.ID(), "Error pulling dependent layers")
 309                 if uerr, ok := err.(*url.Error); ok {
 310                         err = uerr.Err
 311                 }
 312                 if terr, ok := err.(net.Error); ok && terr.Timeout() {
 313                         return nil, 0, err
 314                 }
 315                 return nil, 0, xfer.DoNotRetry{Err: err}
 316         }
 317         *ld.layersDownloaded = true
 318
 319         ld.tmpFile, err = ioutil.TempFile("", "GetImageBlob")
 320         if err != nil {
 321                 layerReader.Close()
 322                 return nil, 0, err
 323         }
 324
 325         reader := progress.NewProgressReader(ioutils.NewCancelReadCloser(ctx, layerReader), progressOutput, ld.layerSize, ld.ID(), "Downloading")
 326         defer reader.Close()
 327
 328         _, err = io.Copy(ld.tmpFile, reader)
 329         if err != nil {
 330                 ld.Close()
 331                 return nil, 0, err
 332         }
 333
 334         progress.Update(progressOutput, ld.ID(), "Download complete")
 335
 336         logrus.Debugf("Downloaded %s to tempfile %s", ld.ID(), ld.tmpFile.Name())
 337
 338         ld.tmpFile.Seek(0, 0)
 339
 340         // hand off the temporary file to the download manager, so it will only
 341         // be closed once
 342         tmpFile := ld.tmpFile
 343         ld.tmpFile = nil
 344
 345         return ioutils.NewReadCloserWrapper(tmpFile, func() error {
 346                 tmpFile.Close()
 347                 err := os.RemoveAll(tmpFile.Name())
 348                 if err != nil {
 349                         logrus.Errorf("Failed to remove temp file: %s", tmpFile.Name())
 350                 }
 351                 return err
 352         }), ld.layerSize, nil
 353 }
 354
 355 func (ld *v1LayerDescriptor) DeltaBase() io.ReadSeeker {
 356         return nil
 357 }
 358
 359 func (ld *v1LayerDescriptor) Close() {
 360         if ld.tmpFile != nil {
 361                 ld.tmpFile.Close()
 362                 if err := os.RemoveAll(ld.tmpFile.Name()); err != nil {
 363                         logrus.Errorf("Failed to remove temp file: %s", ld.tmpFile.Name())
 364                 }
 365                 ld.tmpFile = nil
 366         }
 367 }
 368
 369 func (ld *v1LayerDescriptor) Registered(diffID layer.DiffID) {
 370         // Cache mapping from this layer's DiffID to the blobsum
 371         ld.v1IDService.Set(ld.v1LayerID, ld.indexName, diffID)
 372 }
 373
 374 func (ld *v1LayerDescriptor) Size() int64 {
 375         return ld.layerSize
 376 }