pull.go 23 KB

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