pull.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724
  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. for _, mirror := range mirrors {
  130. mirrorRepoInfo := makeMirrorRepoInfo(repoInfo, mirror)
  131. endpoint, err := registry.NewEndpoint(mirrorRepoInfo.Index, nil)
  132. if err != nil {
  133. logrus.Errorf("Unable to create endpoint for %s: %s", mirror, err)
  134. continue
  135. }
  136. if endpoint.Version == 2 {
  137. if v2MirrorEndpoint == nil {
  138. v2MirrorEndpoint = endpoint
  139. v2MirrorRepoInfo = mirrorRepoInfo
  140. } else {
  141. // > 1 v2 mirrors given
  142. return nil, nil, fmt.Errorf("multiple v2 mirrors configured")
  143. }
  144. } else {
  145. v1MirrorCount++
  146. }
  147. }
  148. if v1MirrorCount == len(mirrors) {
  149. // OK, but mirrors are v1
  150. return nil, nil, nil
  151. }
  152. if v2MirrorEndpoint != nil && v1MirrorCount == 0 {
  153. // OK, 1 v2 mirror specified
  154. return v2MirrorEndpoint, v2MirrorRepoInfo, nil
  155. }
  156. if v2MirrorEndpoint != nil && v1MirrorCount > 0 {
  157. return nil, nil, fmt.Errorf("v1 and v2 mirrors configured")
  158. }
  159. // No endpoint could be established with the given mirror configurations
  160. // Fallback to pulling from the hub as per v1 behavior.
  161. return nil, nil, nil
  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. func (s *TagStore) pullV2Repository(r *registry.Session, out io.Writer, repoInfo *registry.RepositoryInfo, tag string, sf *streamformatter.StreamFormatter) error {
  411. endpoint, err := r.V2RegistryEndpoint(repoInfo.Index)
  412. if err != nil {
  413. if repoInfo.Index.Official {
  414. logrus.Debugf("Unable to pull from V2 registry, falling back to v1: %s", err)
  415. return ErrV2RegistryUnavailable
  416. }
  417. return fmt.Errorf("error getting registry endpoint: %s", err)
  418. }
  419. auth, err := r.GetV2Authorization(endpoint, repoInfo.RemoteName, true)
  420. if err != nil {
  421. return fmt.Errorf("error getting authorization: %s", err)
  422. }
  423. var layersDownloaded bool
  424. if tag == "" {
  425. logrus.Debugf("Pulling tag list from V2 registry for %s", repoInfo.CanonicalName)
  426. tags, err := r.GetV2RemoteTags(endpoint, repoInfo.RemoteName, auth)
  427. if err != nil {
  428. return err
  429. }
  430. if len(tags) == 0 {
  431. return registry.ErrDoesNotExist
  432. }
  433. for _, t := range tags {
  434. if downloaded, err := s.pullV2Tag(r, out, endpoint, repoInfo, t, sf, auth); err != nil {
  435. return err
  436. } else if downloaded {
  437. layersDownloaded = true
  438. }
  439. }
  440. } else {
  441. if downloaded, err := s.pullV2Tag(r, out, endpoint, repoInfo, tag, sf, auth); err != nil {
  442. return err
  443. } else if downloaded {
  444. layersDownloaded = true
  445. }
  446. }
  447. requestedTag := repoInfo.CanonicalName
  448. if len(tag) > 0 {
  449. requestedTag = utils.ImageReference(repoInfo.CanonicalName, tag)
  450. }
  451. WriteStatus(requestedTag, out, sf, layersDownloaded)
  452. return nil
  453. }
  454. 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) {
  455. logrus.Debugf("Pulling tag from V2 registry: %q", tag)
  456. remoteDigest, manifestBytes, err := r.GetV2ImageManifest(endpoint, repoInfo.RemoteName, tag, auth)
  457. if err != nil {
  458. return false, err
  459. }
  460. // loadManifest ensures that the manifest payload has the expected digest
  461. // if the tag is a digest reference.
  462. localDigest, manifest, verified, err := s.loadManifest(manifestBytes, tag, remoteDigest)
  463. if err != nil {
  464. return false, fmt.Errorf("error verifying manifest: %s", err)
  465. }
  466. if verified {
  467. logrus.Printf("Image manifest for %s has been verified", utils.ImageReference(repoInfo.CanonicalName, tag))
  468. }
  469. out.Write(sf.FormatStatus(tag, "Pulling from %s", repoInfo.CanonicalName))
  470. // downloadInfo is used to pass information from download to extractor
  471. type downloadInfo struct {
  472. imgJSON []byte
  473. img *image.Image
  474. digest digest.Digest
  475. tmpFile *os.File
  476. length int64
  477. downloaded bool
  478. err chan error
  479. }
  480. downloads := make([]downloadInfo, len(manifest.FSLayers))
  481. for i := len(manifest.FSLayers) - 1; i >= 0; i-- {
  482. var (
  483. sumStr = manifest.FSLayers[i].BlobSum
  484. imgJSON = []byte(manifest.History[i].V1Compatibility)
  485. )
  486. img, err := image.NewImgJSON(imgJSON)
  487. if err != nil {
  488. return false, fmt.Errorf("failed to parse json: %s", err)
  489. }
  490. downloads[i].img = img
  491. // Check if exists
  492. if s.graph.Exists(img.ID) {
  493. logrus.Debugf("Image already exists: %s", img.ID)
  494. continue
  495. }
  496. dgst, err := digest.ParseDigest(sumStr)
  497. if err != nil {
  498. return false, err
  499. }
  500. downloads[i].digest = dgst
  501. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Pulling fs layer", nil))
  502. downloadFunc := func(di *downloadInfo) error {
  503. logrus.Debugf("pulling blob %q to V1 img %s", sumStr, img.ID)
  504. if c, err := s.poolAdd("pull", "img:"+img.ID); err != nil {
  505. if c != nil {
  506. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Layer already being pulled by another client. Waiting.", nil))
  507. <-c
  508. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Download complete", nil))
  509. } else {
  510. logrus.Debugf("Image (id: %s) pull is already running, skipping: %v", img.ID, err)
  511. }
  512. } else {
  513. defer s.poolRemove("pull", "img:"+img.ID)
  514. tmpFile, err := ioutil.TempFile("", "GetV2ImageBlob")
  515. if err != nil {
  516. return err
  517. }
  518. r, l, err := r.GetV2ImageBlobReader(endpoint, repoInfo.RemoteName, di.digest, auth)
  519. if err != nil {
  520. return err
  521. }
  522. defer r.Close()
  523. verifier, err := digest.NewDigestVerifier(di.digest)
  524. if err != nil {
  525. return err
  526. }
  527. if _, err := io.Copy(tmpFile, progressreader.New(progressreader.Config{
  528. In: ioutil.NopCloser(io.TeeReader(r, verifier)),
  529. Out: out,
  530. Formatter: sf,
  531. Size: int(l),
  532. NewLines: false,
  533. ID: stringid.TruncateID(img.ID),
  534. Action: "Downloading",
  535. })); err != nil {
  536. return fmt.Errorf("unable to copy v2 image blob data: %s", err)
  537. }
  538. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Verifying Checksum", nil))
  539. if !verifier.Verified() {
  540. return fmt.Errorf("image layer digest verification failed for %q", di.digest)
  541. }
  542. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Download complete", nil))
  543. logrus.Debugf("Downloaded %s to tempfile %s", img.ID, tmpFile.Name())
  544. di.tmpFile = tmpFile
  545. di.length = l
  546. di.downloaded = true
  547. }
  548. di.imgJSON = imgJSON
  549. return nil
  550. }
  551. downloads[i].err = make(chan error)
  552. go func(di *downloadInfo) {
  553. di.err <- downloadFunc(di)
  554. }(&downloads[i])
  555. }
  556. var tagUpdated bool
  557. for i := len(downloads) - 1; i >= 0; i-- {
  558. d := &downloads[i]
  559. if d.err != nil {
  560. if err := <-d.err; err != nil {
  561. return false, err
  562. }
  563. }
  564. if d.downloaded {
  565. // if tmpFile is empty assume download and extracted elsewhere
  566. defer os.Remove(d.tmpFile.Name())
  567. defer d.tmpFile.Close()
  568. d.tmpFile.Seek(0, 0)
  569. if d.tmpFile != nil {
  570. err = s.graph.Register(d.img,
  571. progressreader.New(progressreader.Config{
  572. In: d.tmpFile,
  573. Out: out,
  574. Formatter: sf,
  575. Size: int(d.length),
  576. ID: stringid.TruncateID(d.img.ID),
  577. Action: "Extracting",
  578. }))
  579. if err != nil {
  580. return false, err
  581. }
  582. if err := s.graph.SetDigest(d.img.ID, d.digest); err != nil {
  583. return false, err
  584. }
  585. // FIXME: Pool release here for parallel tag pull (ensures any downloads block until fully extracted)
  586. }
  587. out.Write(sf.FormatProgress(stringid.TruncateID(d.img.ID), "Pull complete", nil))
  588. tagUpdated = true
  589. } else {
  590. out.Write(sf.FormatProgress(stringid.TruncateID(d.img.ID), "Already exists", nil))
  591. }
  592. }
  593. // Check for new tag if no layers downloaded
  594. if !tagUpdated {
  595. repo, err := s.Get(repoInfo.LocalName)
  596. if err != nil {
  597. return false, err
  598. }
  599. if repo != nil {
  600. if _, exists := repo[tag]; !exists {
  601. tagUpdated = true
  602. }
  603. } else {
  604. tagUpdated = true
  605. }
  606. }
  607. if verified && tagUpdated {
  608. 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."))
  609. }
  610. if localDigest != remoteDigest { // this is not a verification check.
  611. // NOTE(stevvooe): This is a very defensive branch and should never
  612. // happen, since all manifest digest implementations use the same
  613. // algorithm.
  614. logrus.WithFields(
  615. logrus.Fields{
  616. "local": localDigest,
  617. "remote": remoteDigest,
  618. }).Debugf("local digest does not match remote")
  619. out.Write(sf.FormatStatus("", "Remote Digest: %s", remoteDigest))
  620. }
  621. out.Write(sf.FormatStatus("", "Digest: %s", localDigest))
  622. if tag == localDigest.String() {
  623. // TODO(stevvooe): Ideally, we should always set the digest so we can
  624. // use the digest whether we pull by it or not. Unfortunately, the tag
  625. // store treats the digest as a separate tag, meaning there may be an
  626. // untagged digest image that would seem to be dangling by a user.
  627. if err = s.SetDigest(repoInfo.LocalName, localDigest.String(), downloads[0].img.ID); err != nil {
  628. return false, err
  629. }
  630. }
  631. if !utils.DigestReference(tag) {
  632. // only set the repository/tag -> image ID mapping when pulling by tag (i.e. not by digest)
  633. if err = s.Tag(repoInfo.LocalName, tag, downloads[0].img.ID, true); err != nil {
  634. return false, err
  635. }
  636. }
  637. return tagUpdated, nil
  638. }