pull.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712
  1. package graph
  2. import (
  3. "fmt"
  4. "io"
  5. "io/ioutil"
  6. "net"
  7. "net/url"
  8. "os"
  9. "strings"
  10. "time"
  11. "github.com/Sirupsen/logrus"
  12. "github.com/docker/distribution/digest"
  13. "github.com/docker/docker/cliconfig"
  14. "github.com/docker/docker/image"
  15. "github.com/docker/docker/pkg/progressreader"
  16. "github.com/docker/docker/pkg/streamformatter"
  17. "github.com/docker/docker/pkg/stringid"
  18. "github.com/docker/docker/pkg/transport"
  19. "github.com/docker/docker/registry"
  20. "github.com/docker/docker/utils"
  21. )
  22. type ImagePullConfig struct {
  23. MetaHeaders map[string][]string
  24. AuthConfig *cliconfig.AuthConfig
  25. OutStream io.Writer
  26. }
  27. func (s *TagStore) Pull(image string, tag string, imagePullConfig *ImagePullConfig) error {
  28. var (
  29. sf = streamformatter.NewJSONStreamFormatter()
  30. )
  31. // Resolve the Repository name from fqn to RepositoryInfo
  32. repoInfo, err := s.registryService.ResolveRepository(image)
  33. if err != nil {
  34. return err
  35. }
  36. if err := validateRepoName(repoInfo.LocalName); err != nil {
  37. return err
  38. }
  39. c, err := s.poolAdd("pull", utils.ImageReference(repoInfo.LocalName, tag))
  40. if err != nil {
  41. if c != nil {
  42. // Another pull of the same repository is already taking place; just wait for it to finish
  43. imagePullConfig.OutStream.Write(sf.FormatStatus("", "Repository %s already being pulled by another client. Waiting.", repoInfo.LocalName))
  44. <-c
  45. return nil
  46. }
  47. return err
  48. }
  49. defer s.poolRemove("pull", utils.ImageReference(repoInfo.LocalName, tag))
  50. logName := repoInfo.LocalName
  51. if tag != "" {
  52. logName = utils.ImageReference(logName, tag)
  53. }
  54. // Attempt pulling official content from a provided v2 mirror
  55. if repoInfo.Index.Official {
  56. v2mirrorEndpoint, v2mirrorRepoInfo, err := configureV2Mirror(repoInfo, s.registryService)
  57. if err != nil {
  58. logrus.Errorf("Error configuring mirrors: %s", err)
  59. return err
  60. }
  61. if v2mirrorEndpoint != nil {
  62. logrus.Debugf("Attempting to pull from v2 mirror: %s", v2mirrorEndpoint.URL)
  63. return s.pullFromV2Mirror(v2mirrorEndpoint, v2mirrorRepoInfo, imagePullConfig, tag, sf, logName)
  64. }
  65. }
  66. logrus.Debugf("pulling image from host %q with remote name %q", repoInfo.Index.Name, repoInfo.RemoteName)
  67. endpoint, err := repoInfo.GetEndpoint(imagePullConfig.MetaHeaders)
  68. if err != nil {
  69. return err
  70. }
  71. // TODO(tiborvass): reuse client from endpoint?
  72. // Adds Docker-specific headers as well as user-specified headers (metaHeaders)
  73. tr := transport.NewTransport(
  74. registry.NewTransport(registry.ReceiveTimeout, endpoint.IsSecure),
  75. registry.DockerHeaders(imagePullConfig.MetaHeaders)...,
  76. )
  77. client := registry.HTTPClient(tr)
  78. r, err := registry.NewSession(client, imagePullConfig.AuthConfig, endpoint)
  79. if err != nil {
  80. return err
  81. }
  82. if len(repoInfo.Index.Mirrors) == 0 && (repoInfo.Index.Official || endpoint.Version == registry.APIVersion2) {
  83. if repoInfo.Official {
  84. s.trustService.UpdateBase()
  85. }
  86. logrus.Debugf("pulling v2 repository with local name %q", repoInfo.LocalName)
  87. if err := s.pullV2Repository(r, imagePullConfig.OutStream, repoInfo, tag, sf); err == nil {
  88. s.eventsService.Log("pull", logName, "")
  89. return nil
  90. } else if err != registry.ErrDoesNotExist && err != ErrV2RegistryUnavailable {
  91. logrus.Errorf("Error from V2 registry: %s", err)
  92. }
  93. logrus.Debug("image does not exist on v2 registry, falling back to v1")
  94. }
  95. if utils.DigestReference(tag) {
  96. return fmt.Errorf("pulling with digest reference failed from v2 registry")
  97. }
  98. logrus.Debugf("pulling v1 repository with local name %q", repoInfo.LocalName)
  99. if err = s.pullRepository(r, imagePullConfig.OutStream, repoInfo, tag, sf); err != nil {
  100. return err
  101. }
  102. s.eventsService.Log("pull", logName, "")
  103. return nil
  104. }
  105. func makeMirrorRepoInfo(repoInfo *registry.RepositoryInfo, mirror string) *registry.RepositoryInfo {
  106. mirrorRepo := &registry.RepositoryInfo{
  107. RemoteName: repoInfo.RemoteName,
  108. LocalName: repoInfo.LocalName,
  109. CanonicalName: repoInfo.CanonicalName,
  110. Official: false,
  111. Index: &registry.IndexInfo{
  112. Official: false,
  113. Secure: repoInfo.Index.Secure,
  114. Name: mirror,
  115. Mirrors: []string{},
  116. },
  117. }
  118. return mirrorRepo
  119. }
  120. func configureV2Mirror(repoInfo *registry.RepositoryInfo, s *registry.Service) (*registry.Endpoint, *registry.RepositoryInfo, error) {
  121. mirrors := repoInfo.Index.Mirrors
  122. if len(mirrors) == 0 {
  123. // no mirrors configured
  124. return nil, nil, nil
  125. }
  126. v1MirrorCount := 0
  127. var v2MirrorEndpoint *registry.Endpoint
  128. var v2MirrorRepoInfo *registry.RepositoryInfo
  129. var lastErr error
  130. for _, mirror := range mirrors {
  131. mirrorRepoInfo := makeMirrorRepoInfo(repoInfo, mirror)
  132. endpoint, err := registry.NewEndpoint(mirrorRepoInfo.Index, nil)
  133. if err != nil {
  134. logrus.Errorf("Unable to create endpoint for %s: %s", mirror, err)
  135. lastErr = err
  136. continue
  137. }
  138. if endpoint.Version == 2 {
  139. if v2MirrorEndpoint == nil {
  140. v2MirrorEndpoint = endpoint
  141. v2MirrorRepoInfo = mirrorRepoInfo
  142. } else {
  143. // > 1 v2 mirrors given
  144. return nil, nil, fmt.Errorf("multiple v2 mirrors configured")
  145. }
  146. } else {
  147. v1MirrorCount++
  148. }
  149. }
  150. if v1MirrorCount == len(mirrors) {
  151. // OK, but mirrors are v1
  152. return nil, nil, nil
  153. }
  154. if v2MirrorEndpoint != nil && v1MirrorCount == 0 {
  155. // OK, 1 v2 mirror specified
  156. return v2MirrorEndpoint, v2MirrorRepoInfo, nil
  157. }
  158. if v2MirrorEndpoint != nil && v1MirrorCount > 0 {
  159. lastErr = fmt.Errorf("v1 and v2 mirrors configured")
  160. }
  161. return nil, nil, lastErr
  162. }
  163. func (s *TagStore) pullFromV2Mirror(mirrorEndpoint *registry.Endpoint, repoInfo *registry.RepositoryInfo,
  164. imagePullConfig *ImagePullConfig, tag string, sf *streamformatter.StreamFormatter, logName string) error {
  165. tr := transport.NewTransport(
  166. registry.NewTransport(registry.ReceiveTimeout, mirrorEndpoint.IsSecure),
  167. registry.DockerHeaders(imagePullConfig.MetaHeaders)...,
  168. )
  169. client := registry.HTTPClient(tr)
  170. mirrorSession, err := registry.NewSession(client, &cliconfig.AuthConfig{}, mirrorEndpoint)
  171. if err != nil {
  172. return err
  173. }
  174. logrus.Debugf("Pulling v2 repository with local name %q from %s", repoInfo.LocalName, mirrorEndpoint.URL)
  175. if err := s.pullV2Repository(mirrorSession, imagePullConfig.OutStream, repoInfo, tag, sf); err != nil {
  176. return err
  177. }
  178. s.eventsService.Log("pull", logName, "")
  179. return nil
  180. }
  181. func (s *TagStore) pullRepository(r *registry.Session, out io.Writer, repoInfo *registry.RepositoryInfo, askedTag string, sf *streamformatter.StreamFormatter) error {
  182. out.Write(sf.FormatStatus("", "Pulling repository %s", repoInfo.CanonicalName))
  183. repoData, err := r.GetRepositoryData(repoInfo.RemoteName)
  184. if err != nil {
  185. if strings.Contains(err.Error(), "HTTP code: 404") {
  186. return fmt.Errorf("Error: image %s not found", utils.ImageReference(repoInfo.RemoteName, askedTag))
  187. }
  188. // Unexpected HTTP error
  189. return err
  190. }
  191. logrus.Debugf("Retrieving the tag list")
  192. tagsList, err := r.GetRemoteTags(repoData.Endpoints, repoInfo.RemoteName)
  193. if err != nil {
  194. logrus.Errorf("unable to get remote tags: %s", err)
  195. return err
  196. }
  197. for tag, id := range tagsList {
  198. repoData.ImgList[id] = &registry.ImgData{
  199. ID: id,
  200. Tag: tag,
  201. Checksum: "",
  202. }
  203. }
  204. logrus.Debugf("Registering tags")
  205. // If no tag has been specified, pull them all
  206. if askedTag == "" {
  207. for tag, id := range tagsList {
  208. repoData.ImgList[id].Tag = tag
  209. }
  210. } else {
  211. // Otherwise, check that the tag exists and use only that one
  212. id, exists := tagsList[askedTag]
  213. if !exists {
  214. return fmt.Errorf("Tag %s not found in repository %s", askedTag, repoInfo.CanonicalName)
  215. }
  216. repoData.ImgList[id].Tag = askedTag
  217. }
  218. errors := make(chan error)
  219. layersDownloaded := false
  220. for _, image := range repoData.ImgList {
  221. downloadImage := func(img *registry.ImgData) {
  222. if askedTag != "" && img.Tag != askedTag {
  223. errors <- nil
  224. return
  225. }
  226. if img.Tag == "" {
  227. logrus.Debugf("Image (id: %s) present in this repository but untagged, skipping", img.ID)
  228. errors <- nil
  229. return
  230. }
  231. // ensure no two downloads of the same image happen at the same time
  232. if c, err := s.poolAdd("pull", "img:"+img.ID); err != nil {
  233. if c != nil {
  234. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Layer already being pulled by another client. Waiting.", nil))
  235. <-c
  236. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Download complete", nil))
  237. } else {
  238. logrus.Debugf("Image (id: %s) pull is already running, skipping: %v", img.ID, err)
  239. }
  240. errors <- nil
  241. return
  242. }
  243. defer s.poolRemove("pull", "img:"+img.ID)
  244. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), fmt.Sprintf("Pulling image (%s) from %s", img.Tag, repoInfo.CanonicalName), nil))
  245. success := false
  246. var lastErr, err error
  247. var isDownloaded bool
  248. for _, ep := range repoInfo.Index.Mirrors {
  249. // Ensure endpoint is v1
  250. ep = ep + "v1/"
  251. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), fmt.Sprintf("Pulling image (%s) from %s, mirror: %s", img.Tag, repoInfo.CanonicalName, ep), nil))
  252. if isDownloaded, err = s.pullImage(r, out, img.ID, ep, repoData.Tokens, sf); err != nil {
  253. // Don't report errors when pulling from mirrors.
  254. logrus.Debugf("Error pulling image (%s) from %s, mirror: %s, %s", img.Tag, repoInfo.CanonicalName, ep, err)
  255. continue
  256. }
  257. layersDownloaded = layersDownloaded || isDownloaded
  258. success = true
  259. break
  260. }
  261. if !success {
  262. for _, ep := range repoData.Endpoints {
  263. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), fmt.Sprintf("Pulling image (%s) from %s, endpoint: %s", img.Tag, repoInfo.CanonicalName, ep), nil))
  264. if isDownloaded, err = s.pullImage(r, out, img.ID, ep, repoData.Tokens, sf); err != nil {
  265. // It's not ideal that only the last error is returned, it would be better to concatenate the errors.
  266. // As the error is also given to the output stream the user will see the error.
  267. lastErr = err
  268. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), fmt.Sprintf("Error pulling image (%s) from %s, endpoint: %s, %s", img.Tag, repoInfo.CanonicalName, ep, err), nil))
  269. continue
  270. }
  271. layersDownloaded = layersDownloaded || isDownloaded
  272. success = true
  273. break
  274. }
  275. }
  276. if !success {
  277. err := fmt.Errorf("Error pulling image (%s) from %s, %v", img.Tag, repoInfo.CanonicalName, lastErr)
  278. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), err.Error(), nil))
  279. errors <- err
  280. return
  281. }
  282. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Download complete", nil))
  283. errors <- nil
  284. }
  285. go downloadImage(image)
  286. }
  287. var lastError error
  288. for i := 0; i < len(repoData.ImgList); i++ {
  289. if err := <-errors; err != nil {
  290. lastError = err
  291. }
  292. }
  293. if lastError != nil {
  294. return lastError
  295. }
  296. for tag, id := range tagsList {
  297. if askedTag != "" && tag != askedTag {
  298. continue
  299. }
  300. if err := s.Tag(repoInfo.LocalName, tag, id, true); err != nil {
  301. return err
  302. }
  303. }
  304. requestedTag := repoInfo.CanonicalName
  305. if len(askedTag) > 0 {
  306. requestedTag = utils.ImageReference(repoInfo.CanonicalName, askedTag)
  307. }
  308. WriteStatus(requestedTag, out, sf, layersDownloaded)
  309. return nil
  310. }
  311. func (s *TagStore) pullImage(r *registry.Session, out io.Writer, imgID, endpoint string, token []string, sf *streamformatter.StreamFormatter) (bool, error) {
  312. history, err := r.GetRemoteHistory(imgID, endpoint)
  313. if err != nil {
  314. return false, err
  315. }
  316. out.Write(sf.FormatProgress(stringid.TruncateID(imgID), "Pulling dependent layers", nil))
  317. // FIXME: Try to stream the images?
  318. // FIXME: Launch the getRemoteImage() in goroutines
  319. layersDownloaded := false
  320. for i := len(history) - 1; i >= 0; i-- {
  321. id := history[i]
  322. // ensure no two downloads of the same layer happen at the same time
  323. if c, err := s.poolAdd("pull", "layer:"+id); err != nil {
  324. logrus.Debugf("Image (id: %s) pull is already running, skipping: %v", id, err)
  325. <-c
  326. }
  327. defer s.poolRemove("pull", "layer:"+id)
  328. if !s.graph.Exists(id) {
  329. out.Write(sf.FormatProgress(stringid.TruncateID(id), "Pulling metadata", nil))
  330. var (
  331. imgJSON []byte
  332. imgSize int
  333. err error
  334. img *image.Image
  335. )
  336. retries := 5
  337. for j := 1; j <= retries; j++ {
  338. imgJSON, imgSize, err = r.GetRemoteImageJSON(id, endpoint)
  339. if err != nil && j == retries {
  340. out.Write(sf.FormatProgress(stringid.TruncateID(id), "Error pulling dependent layers", nil))
  341. return layersDownloaded, err
  342. } else if err != nil {
  343. time.Sleep(time.Duration(j) * 500 * time.Millisecond)
  344. continue
  345. }
  346. img, err = image.NewImgJSON(imgJSON)
  347. layersDownloaded = true
  348. if err != nil && j == retries {
  349. out.Write(sf.FormatProgress(stringid.TruncateID(id), "Error pulling dependent layers", nil))
  350. return layersDownloaded, fmt.Errorf("Failed to parse json: %s", err)
  351. } else if err != nil {
  352. time.Sleep(time.Duration(j) * 500 * time.Millisecond)
  353. continue
  354. } else {
  355. break
  356. }
  357. }
  358. for j := 1; j <= retries; j++ {
  359. // Get the layer
  360. status := "Pulling fs layer"
  361. if j > 1 {
  362. status = fmt.Sprintf("Pulling fs layer [retries: %d]", j)
  363. }
  364. out.Write(sf.FormatProgress(stringid.TruncateID(id), status, nil))
  365. layer, err := r.GetRemoteImageLayer(img.ID, endpoint, int64(imgSize))
  366. if uerr, ok := err.(*url.Error); ok {
  367. err = uerr.Err
  368. }
  369. if terr, ok := err.(net.Error); ok && terr.Timeout() && j < retries {
  370. time.Sleep(time.Duration(j) * 500 * time.Millisecond)
  371. continue
  372. } else if err != nil {
  373. out.Write(sf.FormatProgress(stringid.TruncateID(id), "Error pulling dependent layers", nil))
  374. return layersDownloaded, err
  375. }
  376. layersDownloaded = true
  377. defer layer.Close()
  378. err = s.graph.Register(img,
  379. progressreader.New(progressreader.Config{
  380. In: layer,
  381. Out: out,
  382. Formatter: sf,
  383. Size: imgSize,
  384. NewLines: false,
  385. ID: stringid.TruncateID(id),
  386. Action: "Downloading",
  387. }))
  388. if terr, ok := err.(net.Error); ok && terr.Timeout() && j < retries {
  389. time.Sleep(time.Duration(j) * 500 * time.Millisecond)
  390. continue
  391. } else if err != nil {
  392. out.Write(sf.FormatProgress(stringid.TruncateID(id), "Error downloading dependent layers", nil))
  393. return layersDownloaded, err
  394. } else {
  395. break
  396. }
  397. }
  398. }
  399. out.Write(sf.FormatProgress(stringid.TruncateID(id), "Download complete", nil))
  400. }
  401. return layersDownloaded, nil
  402. }
  403. func WriteStatus(requestedTag string, out io.Writer, sf *streamformatter.StreamFormatter, layersDownloaded bool) {
  404. if layersDownloaded {
  405. out.Write(sf.FormatStatus("", "Status: Downloaded newer image for %s", requestedTag))
  406. } else {
  407. out.Write(sf.FormatStatus("", "Status: Image is up to date for %s", requestedTag))
  408. }
  409. }
  410. // downloadInfo is used to pass information from download to extractor
  411. type downloadInfo struct {
  412. imgJSON []byte
  413. img *image.Image
  414. digest digest.Digest
  415. tmpFile *os.File
  416. length int64
  417. downloaded bool
  418. err chan error
  419. }
  420. func (s *TagStore) pullV2Repository(r *registry.Session, out io.Writer, repoInfo *registry.RepositoryInfo, tag string, sf *streamformatter.StreamFormatter) error {
  421. endpoint, err := r.V2RegistryEndpoint(repoInfo.Index)
  422. if err != nil {
  423. if repoInfo.Index.Official {
  424. logrus.Debugf("Unable to pull from V2 registry, falling back to v1: %s", err)
  425. return ErrV2RegistryUnavailable
  426. }
  427. return fmt.Errorf("error getting registry endpoint: %s", err)
  428. }
  429. auth, err := r.GetV2Authorization(endpoint, repoInfo.RemoteName, true)
  430. if err != nil {
  431. return fmt.Errorf("error getting authorization: %s", err)
  432. }
  433. if !auth.CanAuthorizeV2() {
  434. return ErrV2RegistryUnavailable
  435. }
  436. var layersDownloaded bool
  437. if tag == "" {
  438. logrus.Debugf("Pulling tag list from V2 registry for %s", repoInfo.CanonicalName)
  439. tags, err := r.GetV2RemoteTags(endpoint, repoInfo.RemoteName, auth)
  440. if err != nil {
  441. return err
  442. }
  443. if len(tags) == 0 {
  444. return registry.ErrDoesNotExist
  445. }
  446. for _, t := range tags {
  447. if downloaded, err := s.pullV2Tag(r, out, endpoint, repoInfo, t, sf, auth); err != nil {
  448. return err
  449. } else if downloaded {
  450. layersDownloaded = true
  451. }
  452. }
  453. } else {
  454. if downloaded, err := s.pullV2Tag(r, out, endpoint, repoInfo, tag, sf, auth); err != nil {
  455. return err
  456. } else if downloaded {
  457. layersDownloaded = true
  458. }
  459. }
  460. requestedTag := repoInfo.CanonicalName
  461. if len(tag) > 0 {
  462. requestedTag = utils.ImageReference(repoInfo.CanonicalName, tag)
  463. }
  464. WriteStatus(requestedTag, out, sf, layersDownloaded)
  465. return nil
  466. }
  467. func (s *TagStore) pullV2Tag(r *registry.Session, out io.Writer, endpoint *registry.Endpoint, repoInfo *registry.RepositoryInfo, tag string, sf *streamformatter.StreamFormatter, auth *registry.RequestAuthorization) (bool, error) {
  468. logrus.Debugf("Pulling tag from V2 registry: %q", tag)
  469. manifestBytes, manifestDigest, err := r.GetV2ImageManifest(endpoint, repoInfo.RemoteName, tag, auth)
  470. if err != nil {
  471. return false, err
  472. }
  473. // loadManifest ensures that the manifest payload has the expected digest
  474. // if the tag is a digest reference.
  475. manifest, verified, err := s.loadManifest(manifestBytes, manifestDigest, tag)
  476. if err != nil {
  477. return false, fmt.Errorf("error verifying manifest: %s", err)
  478. }
  479. if err := checkValidManifest(manifest); err != nil {
  480. return false, err
  481. }
  482. if verified {
  483. logrus.Printf("Image manifest for %s has been verified", utils.ImageReference(repoInfo.CanonicalName, tag))
  484. }
  485. out.Write(sf.FormatStatus(tag, "Pulling from %s", repoInfo.CanonicalName))
  486. downloads := make([]downloadInfo, len(manifest.FSLayers))
  487. for i := len(manifest.FSLayers) - 1; i >= 0; i-- {
  488. var (
  489. sumStr = manifest.FSLayers[i].BlobSum
  490. imgJSON = []byte(manifest.History[i].V1Compatibility)
  491. )
  492. img, err := image.NewImgJSON(imgJSON)
  493. if err != nil {
  494. return false, fmt.Errorf("failed to parse json: %s", err)
  495. }
  496. downloads[i].img = img
  497. // Check if exists
  498. if s.graph.Exists(img.ID) {
  499. logrus.Debugf("Image already exists: %s", img.ID)
  500. continue
  501. }
  502. dgst, err := digest.ParseDigest(sumStr)
  503. if err != nil {
  504. return false, err
  505. }
  506. downloads[i].digest = dgst
  507. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Pulling fs layer", nil))
  508. downloadFunc := func(di *downloadInfo) error {
  509. logrus.Debugf("pulling blob %q to V1 img %s", sumStr, img.ID)
  510. if c, err := s.poolAdd("pull", "img:"+img.ID); err != nil {
  511. if c != nil {
  512. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Layer already being pulled by another client. Waiting.", nil))
  513. <-c
  514. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Download complete", nil))
  515. } else {
  516. logrus.Debugf("Image (id: %s) pull is already running, skipping: %v", img.ID, err)
  517. }
  518. } else {
  519. defer s.poolRemove("pull", "img:"+img.ID)
  520. tmpFile, err := ioutil.TempFile("", "GetV2ImageBlob")
  521. if err != nil {
  522. return err
  523. }
  524. r, l, err := r.GetV2ImageBlobReader(endpoint, repoInfo.RemoteName, di.digest, auth)
  525. if err != nil {
  526. return err
  527. }
  528. defer r.Close()
  529. verifier, err := digest.NewDigestVerifier(di.digest)
  530. if err != nil {
  531. return err
  532. }
  533. if _, err := io.Copy(tmpFile, progressreader.New(progressreader.Config{
  534. In: ioutil.NopCloser(io.TeeReader(r, verifier)),
  535. Out: out,
  536. Formatter: sf,
  537. Size: int(l),
  538. NewLines: false,
  539. ID: stringid.TruncateID(img.ID),
  540. Action: "Downloading",
  541. })); err != nil {
  542. return fmt.Errorf("unable to copy v2 image blob data: %s", err)
  543. }
  544. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Verifying Checksum", nil))
  545. if !verifier.Verified() {
  546. logrus.Infof("Image verification failed: checksum mismatch for %q", di.digest.String())
  547. verified = false
  548. }
  549. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Download complete", nil))
  550. logrus.Debugf("Downloaded %s to tempfile %s", img.ID, tmpFile.Name())
  551. di.tmpFile = tmpFile
  552. di.length = l
  553. di.downloaded = true
  554. }
  555. di.imgJSON = imgJSON
  556. return nil
  557. }
  558. downloads[i].err = make(chan error)
  559. go func(di *downloadInfo) {
  560. di.err <- downloadFunc(di)
  561. }(&downloads[i])
  562. }
  563. var tagUpdated bool
  564. for i := len(downloads) - 1; i >= 0; i-- {
  565. d := &downloads[i]
  566. if d.err != nil {
  567. if err := <-d.err; err != nil {
  568. return false, err
  569. }
  570. }
  571. if d.downloaded {
  572. // if tmpFile is empty assume download and extracted elsewhere
  573. defer os.Remove(d.tmpFile.Name())
  574. defer d.tmpFile.Close()
  575. d.tmpFile.Seek(0, 0)
  576. if d.tmpFile != nil {
  577. err = s.graph.Register(d.img,
  578. progressreader.New(progressreader.Config{
  579. In: d.tmpFile,
  580. Out: out,
  581. Formatter: sf,
  582. Size: int(d.length),
  583. ID: stringid.TruncateID(d.img.ID),
  584. Action: "Extracting",
  585. }))
  586. if err != nil {
  587. return false, err
  588. }
  589. // FIXME: Pool release here for parallel tag pull (ensures any downloads block until fully extracted)
  590. }
  591. out.Write(sf.FormatProgress(stringid.TruncateID(d.img.ID), "Pull complete", nil))
  592. tagUpdated = true
  593. } else {
  594. out.Write(sf.FormatProgress(stringid.TruncateID(d.img.ID), "Already exists", nil))
  595. }
  596. }
  597. // Check for new tag if no layers downloaded
  598. if !tagUpdated {
  599. repo, err := s.Get(repoInfo.LocalName)
  600. if err != nil {
  601. return false, err
  602. }
  603. if repo != nil {
  604. if _, exists := repo[tag]; !exists {
  605. tagUpdated = true
  606. }
  607. } else {
  608. tagUpdated = true
  609. }
  610. }
  611. if verified && tagUpdated {
  612. out.Write(sf.FormatStatus(utils.ImageReference(repoInfo.CanonicalName, tag), "The image you are pulling has been verified. Important: image verification is a tech preview feature and should not be relied on to provide security."))
  613. }
  614. if manifestDigest != "" {
  615. out.Write(sf.FormatStatus("", "Digest: %s", manifestDigest))
  616. }
  617. if utils.DigestReference(tag) {
  618. if err = s.SetDigest(repoInfo.LocalName, tag, downloads[0].img.ID); err != nil {
  619. return false, err
  620. }
  621. } else {
  622. // only set the repository/tag -> image ID mapping when pulling by tag (i.e. not by digest)
  623. if err = s.Tag(repoInfo.LocalName, tag, downloads[0].img.ID, true); err != nil {
  624. return false, err
  625. }
  626. }
  627. return tagUpdated, nil
  628. }