pull.go 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730
  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 := make(map[string]string)
  193. if askedTag == "" {
  194. tagsList, err = r.GetRemoteTags(repoData.Endpoints, repoInfo.RemoteName)
  195. } else {
  196. var tagId string
  197. tagId, err = r.GetRemoteTag(repoData.Endpoints, repoInfo.RemoteName, askedTag)
  198. tagsList[askedTag] = tagId
  199. }
  200. if err != nil {
  201. if err == registry.ErrRepoNotFound && askedTag != "" {
  202. return fmt.Errorf("Tag %s not found in repository %s", askedTag, repoInfo.CanonicalName)
  203. }
  204. logrus.Errorf("unable to get remote tags: %s", err)
  205. return err
  206. }
  207. for tag, id := range tagsList {
  208. repoData.ImgList[id] = &registry.ImgData{
  209. ID: id,
  210. Tag: tag,
  211. Checksum: "",
  212. }
  213. }
  214. logrus.Debugf("Registering tags")
  215. // If no tag has been specified, pull them all
  216. if askedTag == "" {
  217. for tag, id := range tagsList {
  218. repoData.ImgList[id].Tag = tag
  219. }
  220. } else {
  221. // Otherwise, check that the tag exists and use only that one
  222. id, exists := tagsList[askedTag]
  223. if !exists {
  224. return fmt.Errorf("Tag %s not found in repository %s", askedTag, repoInfo.CanonicalName)
  225. }
  226. repoData.ImgList[id].Tag = askedTag
  227. }
  228. errors := make(chan error)
  229. layersDownloaded := false
  230. for _, image := range repoData.ImgList {
  231. downloadImage := func(img *registry.ImgData) {
  232. if askedTag != "" && img.Tag != askedTag {
  233. errors <- nil
  234. return
  235. }
  236. if img.Tag == "" {
  237. logrus.Debugf("Image (id: %s) present in this repository but untagged, skipping", img.ID)
  238. errors <- nil
  239. return
  240. }
  241. // ensure no two downloads of the same image happen at the same time
  242. if c, err := s.poolAdd("pull", "img:"+img.ID); err != nil {
  243. if c != nil {
  244. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Layer already being pulled by another client. Waiting.", nil))
  245. <-c
  246. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Download complete", nil))
  247. } else {
  248. logrus.Debugf("Image (id: %s) pull is already running, skipping: %v", img.ID, err)
  249. }
  250. errors <- nil
  251. return
  252. }
  253. defer s.poolRemove("pull", "img:"+img.ID)
  254. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), fmt.Sprintf("Pulling image (%s) from %s", img.Tag, repoInfo.CanonicalName), nil))
  255. success := false
  256. var lastErr, err error
  257. var isDownloaded bool
  258. for _, ep := range repoInfo.Index.Mirrors {
  259. // Ensure endpoint is v1
  260. ep = ep + "v1/"
  261. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), fmt.Sprintf("Pulling image (%s) from %s, mirror: %s", img.Tag, repoInfo.CanonicalName, ep), nil))
  262. if isDownloaded, err = s.pullImage(r, out, img.ID, ep, repoData.Tokens, sf); err != nil {
  263. // Don't report errors when pulling from mirrors.
  264. logrus.Debugf("Error pulling image (%s) from %s, mirror: %s, %s", img.Tag, repoInfo.CanonicalName, ep, err)
  265. continue
  266. }
  267. layersDownloaded = layersDownloaded || isDownloaded
  268. success = true
  269. break
  270. }
  271. if !success {
  272. for _, ep := range repoData.Endpoints {
  273. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), fmt.Sprintf("Pulling image (%s) from %s, endpoint: %s", img.Tag, repoInfo.CanonicalName, ep), nil))
  274. if isDownloaded, err = s.pullImage(r, out, img.ID, ep, repoData.Tokens, sf); err != nil {
  275. // It's not ideal that only the last error is returned, it would be better to concatenate the errors.
  276. // As the error is also given to the output stream the user will see the error.
  277. lastErr = err
  278. 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))
  279. continue
  280. }
  281. layersDownloaded = layersDownloaded || isDownloaded
  282. success = true
  283. break
  284. }
  285. }
  286. if !success {
  287. err := fmt.Errorf("Error pulling image (%s) from %s, %v", img.Tag, repoInfo.CanonicalName, lastErr)
  288. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), err.Error(), nil))
  289. errors <- err
  290. return
  291. }
  292. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Download complete", nil))
  293. errors <- nil
  294. }
  295. go downloadImage(image)
  296. }
  297. var lastError error
  298. for i := 0; i < len(repoData.ImgList); i++ {
  299. if err := <-errors; err != nil {
  300. lastError = err
  301. }
  302. }
  303. if lastError != nil {
  304. return lastError
  305. }
  306. for tag, id := range tagsList {
  307. if askedTag != "" && tag != askedTag {
  308. continue
  309. }
  310. if err := s.Tag(repoInfo.LocalName, tag, id, true); err != nil {
  311. return err
  312. }
  313. }
  314. requestedTag := repoInfo.CanonicalName
  315. if len(askedTag) > 0 {
  316. requestedTag = utils.ImageReference(repoInfo.CanonicalName, askedTag)
  317. }
  318. WriteStatus(requestedTag, out, sf, layersDownloaded)
  319. return nil
  320. }
  321. func (s *TagStore) pullImage(r *registry.Session, out io.Writer, imgID, endpoint string, token []string, sf *streamformatter.StreamFormatter) (bool, error) {
  322. history, err := r.GetRemoteHistory(imgID, endpoint)
  323. if err != nil {
  324. return false, err
  325. }
  326. out.Write(sf.FormatProgress(stringid.TruncateID(imgID), "Pulling dependent layers", nil))
  327. // FIXME: Try to stream the images?
  328. // FIXME: Launch the getRemoteImage() in goroutines
  329. layersDownloaded := false
  330. for i := len(history) - 1; i >= 0; i-- {
  331. id := history[i]
  332. // ensure no two downloads of the same layer happen at the same time
  333. if c, err := s.poolAdd("pull", "layer:"+id); err != nil {
  334. logrus.Debugf("Image (id: %s) pull is already running, skipping: %v", id, err)
  335. <-c
  336. }
  337. defer s.poolRemove("pull", "layer:"+id)
  338. if !s.graph.Exists(id) {
  339. out.Write(sf.FormatProgress(stringid.TruncateID(id), "Pulling metadata", nil))
  340. var (
  341. imgJSON []byte
  342. imgSize int
  343. err error
  344. img *image.Image
  345. )
  346. retries := 5
  347. for j := 1; j <= retries; j++ {
  348. imgJSON, imgSize, err = r.GetRemoteImageJSON(id, endpoint)
  349. if err != nil && j == retries {
  350. out.Write(sf.FormatProgress(stringid.TruncateID(id), "Error pulling dependent layers", nil))
  351. return layersDownloaded, err
  352. } else if err != nil {
  353. time.Sleep(time.Duration(j) * 500 * time.Millisecond)
  354. continue
  355. }
  356. img, err = image.NewImgJSON(imgJSON)
  357. layersDownloaded = true
  358. if err != nil && j == retries {
  359. out.Write(sf.FormatProgress(stringid.TruncateID(id), "Error pulling dependent layers", nil))
  360. return layersDownloaded, fmt.Errorf("Failed to parse json: %s", err)
  361. } else if err != nil {
  362. time.Sleep(time.Duration(j) * 500 * time.Millisecond)
  363. continue
  364. } else {
  365. break
  366. }
  367. }
  368. for j := 1; j <= retries; j++ {
  369. // Get the layer
  370. status := "Pulling fs layer"
  371. if j > 1 {
  372. status = fmt.Sprintf("Pulling fs layer [retries: %d]", j)
  373. }
  374. out.Write(sf.FormatProgress(stringid.TruncateID(id), status, nil))
  375. layer, err := r.GetRemoteImageLayer(img.ID, endpoint, int64(imgSize))
  376. if uerr, ok := err.(*url.Error); ok {
  377. err = uerr.Err
  378. }
  379. if terr, ok := err.(net.Error); ok && terr.Timeout() && j < retries {
  380. time.Sleep(time.Duration(j) * 500 * time.Millisecond)
  381. continue
  382. } else if err != nil {
  383. out.Write(sf.FormatProgress(stringid.TruncateID(id), "Error pulling dependent layers", nil))
  384. return layersDownloaded, err
  385. }
  386. layersDownloaded = true
  387. defer layer.Close()
  388. err = s.graph.Register(img,
  389. progressreader.New(progressreader.Config{
  390. In: layer,
  391. Out: out,
  392. Formatter: sf,
  393. Size: imgSize,
  394. NewLines: false,
  395. ID: stringid.TruncateID(id),
  396. Action: "Downloading",
  397. }))
  398. if terr, ok := err.(net.Error); ok && terr.Timeout() && j < retries {
  399. time.Sleep(time.Duration(j) * 500 * time.Millisecond)
  400. continue
  401. } else if err != nil {
  402. out.Write(sf.FormatProgress(stringid.TruncateID(id), "Error downloading dependent layers", nil))
  403. return layersDownloaded, err
  404. } else {
  405. break
  406. }
  407. }
  408. }
  409. out.Write(sf.FormatProgress(stringid.TruncateID(id), "Download complete", nil))
  410. }
  411. return layersDownloaded, nil
  412. }
  413. func WriteStatus(requestedTag string, out io.Writer, sf *streamformatter.StreamFormatter, layersDownloaded bool) {
  414. if layersDownloaded {
  415. out.Write(sf.FormatStatus("", "Status: Downloaded newer image for %s", requestedTag))
  416. } else {
  417. out.Write(sf.FormatStatus("", "Status: Image is up to date for %s", requestedTag))
  418. }
  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. var layersDownloaded bool
  434. if tag == "" {
  435. logrus.Debugf("Pulling tag list from V2 registry for %s", repoInfo.CanonicalName)
  436. tags, err := r.GetV2RemoteTags(endpoint, repoInfo.RemoteName, auth)
  437. if err != nil {
  438. return err
  439. }
  440. if len(tags) == 0 {
  441. return registry.ErrDoesNotExist
  442. }
  443. for _, t := range tags {
  444. if downloaded, err := s.pullV2Tag(r, out, endpoint, repoInfo, t, sf, auth); err != nil {
  445. return err
  446. } else if downloaded {
  447. layersDownloaded = true
  448. }
  449. }
  450. } else {
  451. if downloaded, err := s.pullV2Tag(r, out, endpoint, repoInfo, tag, sf, auth); err != nil {
  452. return err
  453. } else if downloaded {
  454. layersDownloaded = true
  455. }
  456. }
  457. requestedTag := repoInfo.CanonicalName
  458. if len(tag) > 0 {
  459. requestedTag = utils.ImageReference(repoInfo.CanonicalName, tag)
  460. }
  461. WriteStatus(requestedTag, out, sf, layersDownloaded)
  462. return nil
  463. }
  464. 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) {
  465. logrus.Debugf("Pulling tag from V2 registry: %q", tag)
  466. remoteDigest, manifestBytes, err := r.GetV2ImageManifest(endpoint, repoInfo.RemoteName, tag, auth)
  467. if err != nil {
  468. return false, err
  469. }
  470. // loadManifest ensures that the manifest payload has the expected digest
  471. // if the tag is a digest reference.
  472. localDigest, manifest, verified, err := s.loadManifest(manifestBytes, tag, remoteDigest)
  473. if err != nil {
  474. return false, fmt.Errorf("error verifying manifest: %s", err)
  475. }
  476. if verified {
  477. logrus.Printf("Image manifest for %s has been verified", utils.ImageReference(repoInfo.CanonicalName, tag))
  478. }
  479. out.Write(sf.FormatStatus(tag, "Pulling from %s", repoInfo.CanonicalName))
  480. // downloadInfo is used to pass information from download to extractor
  481. type downloadInfo struct {
  482. imgJSON []byte
  483. img *image.Image
  484. digest digest.Digest
  485. tmpFile *os.File
  486. length int64
  487. downloaded bool
  488. err chan error
  489. }
  490. downloads := make([]downloadInfo, len(manifest.FSLayers))
  491. for i := len(manifest.FSLayers) - 1; i >= 0; i-- {
  492. var (
  493. sumStr = manifest.FSLayers[i].BlobSum
  494. imgJSON = []byte(manifest.History[i].V1Compatibility)
  495. )
  496. img, err := image.NewImgJSON(imgJSON)
  497. if err != nil {
  498. return false, fmt.Errorf("failed to parse json: %s", err)
  499. }
  500. downloads[i].img = img
  501. // Check if exists
  502. if s.graph.Exists(img.ID) {
  503. logrus.Debugf("Image already exists: %s", img.ID)
  504. continue
  505. }
  506. dgst, err := digest.ParseDigest(sumStr)
  507. if err != nil {
  508. return false, err
  509. }
  510. downloads[i].digest = dgst
  511. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Pulling fs layer", nil))
  512. downloadFunc := func(di *downloadInfo) error {
  513. logrus.Debugf("pulling blob %q to V1 img %s", sumStr, img.ID)
  514. if c, err := s.poolAdd("pull", "img:"+img.ID); err != nil {
  515. if c != nil {
  516. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Layer already being pulled by another client. Waiting.", nil))
  517. <-c
  518. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Download complete", nil))
  519. } else {
  520. logrus.Debugf("Image (id: %s) pull is already running, skipping: %v", img.ID, err)
  521. }
  522. } else {
  523. defer s.poolRemove("pull", "img:"+img.ID)
  524. tmpFile, err := ioutil.TempFile("", "GetV2ImageBlob")
  525. if err != nil {
  526. return err
  527. }
  528. r, l, err := r.GetV2ImageBlobReader(endpoint, repoInfo.RemoteName, di.digest, auth)
  529. if err != nil {
  530. return err
  531. }
  532. defer r.Close()
  533. verifier, err := digest.NewDigestVerifier(di.digest)
  534. if err != nil {
  535. return err
  536. }
  537. if _, err := io.Copy(tmpFile, progressreader.New(progressreader.Config{
  538. In: ioutil.NopCloser(io.TeeReader(r, verifier)),
  539. Out: out,
  540. Formatter: sf,
  541. Size: int(l),
  542. NewLines: false,
  543. ID: stringid.TruncateID(img.ID),
  544. Action: "Downloading",
  545. })); err != nil {
  546. return fmt.Errorf("unable to copy v2 image blob data: %s", err)
  547. }
  548. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Verifying Checksum", nil))
  549. if !verifier.Verified() {
  550. return fmt.Errorf("image layer digest verification failed for %q", di.digest)
  551. }
  552. out.Write(sf.FormatProgress(stringid.TruncateID(img.ID), "Download complete", nil))
  553. logrus.Debugf("Downloaded %s to tempfile %s", img.ID, tmpFile.Name())
  554. di.tmpFile = tmpFile
  555. di.length = l
  556. di.downloaded = true
  557. }
  558. di.imgJSON = imgJSON
  559. return nil
  560. }
  561. downloads[i].err = make(chan error)
  562. go func(di *downloadInfo) {
  563. di.err <- downloadFunc(di)
  564. }(&downloads[i])
  565. }
  566. var tagUpdated bool
  567. for i := len(downloads) - 1; i >= 0; i-- {
  568. d := &downloads[i]
  569. if d.err != nil {
  570. if err := <-d.err; err != nil {
  571. return false, err
  572. }
  573. }
  574. if d.downloaded {
  575. // if tmpFile is empty assume download and extracted elsewhere
  576. defer os.Remove(d.tmpFile.Name())
  577. defer d.tmpFile.Close()
  578. d.tmpFile.Seek(0, 0)
  579. if d.tmpFile != nil {
  580. err = s.graph.Register(d.img,
  581. progressreader.New(progressreader.Config{
  582. In: d.tmpFile,
  583. Out: out,
  584. Formatter: sf,
  585. Size: int(d.length),
  586. ID: stringid.TruncateID(d.img.ID),
  587. Action: "Extracting",
  588. }))
  589. if err != nil {
  590. return false, err
  591. }
  592. // FIXME: Pool release here for parallel tag pull (ensures any downloads block until fully extracted)
  593. }
  594. out.Write(sf.FormatProgress(stringid.TruncateID(d.img.ID), "Pull complete", nil))
  595. tagUpdated = true
  596. } else {
  597. out.Write(sf.FormatProgress(stringid.TruncateID(d.img.ID), "Already exists", nil))
  598. }
  599. }
  600. // Check for new tag if no layers downloaded
  601. if !tagUpdated {
  602. repo, err := s.Get(repoInfo.LocalName)
  603. if err != nil {
  604. return false, err
  605. }
  606. if repo != nil {
  607. if _, exists := repo[tag]; !exists {
  608. tagUpdated = true
  609. }
  610. } else {
  611. tagUpdated = true
  612. }
  613. }
  614. if verified && tagUpdated {
  615. 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."))
  616. }
  617. if localDigest != remoteDigest { // this is not a verification check.
  618. // NOTE(stevvooe): This is a very defensive branch and should never
  619. // happen, since all manifest digest implementations use the same
  620. // algorithm.
  621. logrus.WithFields(
  622. logrus.Fields{
  623. "local": localDigest,
  624. "remote": remoteDigest,
  625. }).Debugf("local digest does not match remote")
  626. out.Write(sf.FormatStatus("", "Remote Digest: %s", remoteDigest))
  627. }
  628. out.Write(sf.FormatStatus("", "Digest: %s", localDigest))
  629. if tag == localDigest.String() {
  630. // TODO(stevvooe): Ideally, we should always set the digest so we can
  631. // use the digest whether we pull by it or not. Unfortunately, the tag
  632. // store treats the digest as a separate tag, meaning there may be an
  633. // untagged digest image that would seem to be dangling by a user.
  634. if err = s.SetDigest(repoInfo.LocalName, localDigest.String(), downloads[0].img.ID); err != nil {
  635. return false, err
  636. }
  637. }
  638. if !utils.DigestReference(tag) {
  639. // only set the repository/tag -> image ID mapping when pulling by tag (i.e. not by digest)
  640. if err = s.Tag(repoInfo.LocalName, tag, downloads[0].img.ID, true); err != nil {
  641. return false, err
  642. }
  643. }
  644. return tagUpdated, nil
  645. }