pull_v1.go 11 KB

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