pull.go 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596
  1. package graph
  2. import (
  3. "bytes"
  4. "encoding/json"
  5. "fmt"
  6. "io"
  7. "io/ioutil"
  8. "net"
  9. "net/url"
  10. "os"
  11. "strings"
  12. "time"
  13. log "github.com/Sirupsen/logrus"
  14. "github.com/docker/docker/engine"
  15. "github.com/docker/docker/image"
  16. "github.com/docker/docker/registry"
  17. "github.com/docker/docker/utils"
  18. "github.com/docker/libtrust"
  19. )
  20. func (s *TagStore) verifyManifest(eng *engine.Engine, manifestBytes []byte) (*registry.ManifestData, bool, error) {
  21. sig, err := libtrust.ParsePrettySignature(manifestBytes, "signatures")
  22. if err != nil {
  23. return nil, false, fmt.Errorf("error parsing payload: %s", err)
  24. }
  25. keys, err := sig.Verify()
  26. if err != nil {
  27. return nil, false, fmt.Errorf("error verifying payload: %s", err)
  28. }
  29. payload, err := sig.Payload()
  30. if err != nil {
  31. return nil, false, fmt.Errorf("error retrieving payload: %s", err)
  32. }
  33. var manifest registry.ManifestData
  34. if err := json.Unmarshal(payload, &manifest); err != nil {
  35. return nil, false, fmt.Errorf("error unmarshalling manifest: %s", err)
  36. }
  37. if manifest.SchemaVersion != 1 {
  38. return nil, false, fmt.Errorf("unsupported schema version: %d", manifest.SchemaVersion)
  39. }
  40. var verified bool
  41. for _, key := range keys {
  42. job := eng.Job("trust_key_check")
  43. b, err := key.MarshalJSON()
  44. if err != nil {
  45. return nil, false, fmt.Errorf("error marshalling public key: %s", err)
  46. }
  47. namespace := manifest.Name
  48. if namespace[0] != '/' {
  49. namespace = "/" + namespace
  50. }
  51. stdoutBuffer := bytes.NewBuffer(nil)
  52. job.Args = append(job.Args, namespace)
  53. job.Setenv("PublicKey", string(b))
  54. // Check key has read/write permission (0x03)
  55. job.SetenvInt("Permission", 0x03)
  56. job.Stdout.Add(stdoutBuffer)
  57. if err = job.Run(); err != nil {
  58. return nil, false, fmt.Errorf("error running key check: %s", err)
  59. }
  60. result := engine.Tail(stdoutBuffer, 1)
  61. log.Debugf("Key check result: %q", result)
  62. if result == "verified" {
  63. verified = true
  64. }
  65. }
  66. return &manifest, verified, nil
  67. }
  68. func (s *TagStore) CmdPull(job *engine.Job) engine.Status {
  69. if n := len(job.Args); n != 1 && n != 2 {
  70. return job.Errorf("Usage: %s IMAGE [TAG]", job.Name)
  71. }
  72. var (
  73. localName = job.Args[0]
  74. tag string
  75. sf = utils.NewStreamFormatter(job.GetenvBool("json"))
  76. authConfig = &registry.AuthConfig{}
  77. metaHeaders map[string][]string
  78. )
  79. // Resolve the Repository name from fqn to RepositoryInfo
  80. repoInfo, err := registry.ResolveRepositoryInfo(job, localName)
  81. if err != nil {
  82. return job.Error(err)
  83. }
  84. if len(job.Args) > 1 {
  85. tag = job.Args[1]
  86. }
  87. job.GetenvJson("authConfig", authConfig)
  88. job.GetenvJson("metaHeaders", &metaHeaders)
  89. c, err := s.poolAdd("pull", repoInfo.LocalName+":"+tag)
  90. if err != nil {
  91. if c != nil {
  92. // Another pull of the same repository is already taking place; just wait for it to finish
  93. job.Stdout.Write(sf.FormatStatus("", "Repository %s already being pulled by another client. Waiting.", repoInfo.LocalName))
  94. <-c
  95. return engine.StatusOK
  96. }
  97. return job.Error(err)
  98. }
  99. defer s.poolRemove("pull", repoInfo.LocalName+":"+tag)
  100. endpoint, err := repoInfo.GetEndpoint()
  101. if err != nil {
  102. return job.Error(err)
  103. }
  104. r, err := registry.NewSession(authConfig, registry.HTTPRequestFactory(metaHeaders), endpoint, true)
  105. if err != nil {
  106. return job.Error(err)
  107. }
  108. logName := repoInfo.LocalName
  109. if tag != "" {
  110. logName += ":" + tag
  111. }
  112. if len(repoInfo.Index.Mirrors) == 0 && (repoInfo.Official || endpoint.Version == registry.APIVersion2) {
  113. j := job.Eng.Job("trust_update_base")
  114. if err = j.Run(); err != nil {
  115. return job.Errorf("error updating trust base graph: %s", err)
  116. }
  117. if err := s.pullV2Repository(job.Eng, r, job.Stdout, repoInfo, tag, sf, job.GetenvBool("parallel")); err == nil {
  118. if err = job.Eng.Job("log", "pull", logName, "").Run(); err != nil {
  119. log.Errorf("Error logging event 'pull' for %s: %s", logName, err)
  120. }
  121. return engine.StatusOK
  122. } else if err != registry.ErrDoesNotExist {
  123. log.Errorf("Error from V2 registry: %s", err)
  124. }
  125. }
  126. if err = s.pullRepository(r, job.Stdout, repoInfo, tag, sf, job.GetenvBool("parallel")); err != nil {
  127. return job.Error(err)
  128. }
  129. if err = job.Eng.Job("log", "pull", logName, "").Run(); err != nil {
  130. log.Errorf("Error logging event 'pull' for %s: %s", logName, err)
  131. }
  132. return engine.StatusOK
  133. }
  134. func (s *TagStore) pullRepository(r *registry.Session, out io.Writer, repoInfo *registry.RepositoryInfo, askedTag string, sf *utils.StreamFormatter, parallel bool) error {
  135. out.Write(sf.FormatStatus("", "Pulling repository %s", repoInfo.CanonicalName))
  136. repoData, err := r.GetRepositoryData(repoInfo.RemoteName)
  137. if err != nil {
  138. if strings.Contains(err.Error(), "HTTP code: 404") {
  139. return fmt.Errorf("Error: image %s:%s not found", repoInfo.RemoteName, askedTag)
  140. }
  141. // Unexpected HTTP error
  142. return err
  143. }
  144. log.Debugf("Retrieving the tag list")
  145. tagsList, err := r.GetRemoteTags(repoData.Endpoints, repoInfo.RemoteName, repoData.Tokens)
  146. if err != nil {
  147. log.Errorf("%v", err)
  148. return err
  149. }
  150. for tag, id := range tagsList {
  151. repoData.ImgList[id] = &registry.ImgData{
  152. ID: id,
  153. Tag: tag,
  154. Checksum: "",
  155. }
  156. }
  157. log.Debugf("Registering tags")
  158. // If no tag has been specified, pull them all
  159. var imageId string
  160. if askedTag == "" {
  161. for tag, id := range tagsList {
  162. repoData.ImgList[id].Tag = tag
  163. }
  164. } else {
  165. // Otherwise, check that the tag exists and use only that one
  166. id, exists := tagsList[askedTag]
  167. if !exists {
  168. return fmt.Errorf("Tag %s not found in repository %s", askedTag, repoInfo.CanonicalName)
  169. }
  170. imageId = id
  171. repoData.ImgList[id].Tag = askedTag
  172. }
  173. errors := make(chan error)
  174. layers_downloaded := false
  175. for _, image := range repoData.ImgList {
  176. downloadImage := func(img *registry.ImgData) {
  177. if askedTag != "" && img.Tag != askedTag {
  178. log.Debugf("(%s) does not match %s (id: %s), skipping", img.Tag, askedTag, img.ID)
  179. if parallel {
  180. errors <- nil
  181. }
  182. return
  183. }
  184. if img.Tag == "" {
  185. log.Debugf("Image (id: %s) present in this repository but untagged, skipping", img.ID)
  186. if parallel {
  187. errors <- nil
  188. }
  189. return
  190. }
  191. // ensure no two downloads of the same image happen at the same time
  192. if c, err := s.poolAdd("pull", "img:"+img.ID); err != nil {
  193. if c != nil {
  194. out.Write(sf.FormatProgress(utils.TruncateID(img.ID), "Layer already being pulled by another client. Waiting.", nil))
  195. <-c
  196. out.Write(sf.FormatProgress(utils.TruncateID(img.ID), "Download complete", nil))
  197. } else {
  198. log.Debugf("Image (id: %s) pull is already running, skipping: %v", img.ID, err)
  199. }
  200. if parallel {
  201. errors <- nil
  202. }
  203. return
  204. }
  205. defer s.poolRemove("pull", "img:"+img.ID)
  206. out.Write(sf.FormatProgress(utils.TruncateID(img.ID), fmt.Sprintf("Pulling image (%s) from %s", img.Tag, repoInfo.CanonicalName), nil))
  207. success := false
  208. var lastErr, err error
  209. var is_downloaded bool
  210. for _, ep := range repoInfo.Index.Mirrors {
  211. out.Write(sf.FormatProgress(utils.TruncateID(img.ID), fmt.Sprintf("Pulling image (%s) from %s, mirror: %s", img.Tag, repoInfo.CanonicalName, ep), nil))
  212. if is_downloaded, err = s.pullImage(r, out, img.ID, ep, repoData.Tokens, sf); err != nil {
  213. // Don't report errors when pulling from mirrors.
  214. log.Debugf("Error pulling image (%s) from %s, mirror: %s, %s", img.Tag, repoInfo.CanonicalName, ep, err)
  215. continue
  216. }
  217. layers_downloaded = layers_downloaded || is_downloaded
  218. success = true
  219. break
  220. }
  221. if !success {
  222. for _, ep := range repoData.Endpoints {
  223. out.Write(sf.FormatProgress(utils.TruncateID(img.ID), fmt.Sprintf("Pulling image (%s) from %s, endpoint: %s", img.Tag, repoInfo.CanonicalName, ep), nil))
  224. if is_downloaded, err = s.pullImage(r, out, img.ID, ep, repoData.Tokens, sf); err != nil {
  225. // It's not ideal that only the last error is returned, it would be better to concatenate the errors.
  226. // As the error is also given to the output stream the user will see the error.
  227. lastErr = err
  228. out.Write(sf.FormatProgress(utils.TruncateID(img.ID), fmt.Sprintf("Error pulling image (%s) from %s, endpoint: %s, %s", img.Tag, repoInfo.CanonicalName, ep, err), nil))
  229. continue
  230. }
  231. layers_downloaded = layers_downloaded || is_downloaded
  232. success = true
  233. break
  234. }
  235. }
  236. if !success {
  237. err := fmt.Errorf("Error pulling image (%s) from %s, %v", img.Tag, repoInfo.CanonicalName, lastErr)
  238. out.Write(sf.FormatProgress(utils.TruncateID(img.ID), err.Error(), nil))
  239. if parallel {
  240. errors <- err
  241. return
  242. }
  243. }
  244. out.Write(sf.FormatProgress(utils.TruncateID(img.ID), "Download complete", nil))
  245. if parallel {
  246. errors <- nil
  247. }
  248. }
  249. if parallel {
  250. go downloadImage(image)
  251. } else {
  252. downloadImage(image)
  253. }
  254. }
  255. if parallel {
  256. var lastError error
  257. for i := 0; i < len(repoData.ImgList); i++ {
  258. if err := <-errors; err != nil {
  259. lastError = err
  260. }
  261. }
  262. if lastError != nil {
  263. return lastError
  264. }
  265. }
  266. for tag, id := range tagsList {
  267. if askedTag != "" && id != imageId {
  268. continue
  269. }
  270. if err := s.Set(repoInfo.LocalName, tag, id, true); err != nil {
  271. return err
  272. }
  273. }
  274. requestedTag := repoInfo.CanonicalName
  275. if len(askedTag) > 0 {
  276. requestedTag = repoInfo.CanonicalName + ":" + askedTag
  277. }
  278. WriteStatus(requestedTag, out, sf, layers_downloaded)
  279. return nil
  280. }
  281. func (s *TagStore) pullImage(r *registry.Session, out io.Writer, imgID, endpoint string, token []string, sf *utils.StreamFormatter) (bool, error) {
  282. history, err := r.GetRemoteHistory(imgID, endpoint, token)
  283. if err != nil {
  284. return false, err
  285. }
  286. out.Write(sf.FormatProgress(utils.TruncateID(imgID), "Pulling dependent layers", nil))
  287. // FIXME: Try to stream the images?
  288. // FIXME: Launch the getRemoteImage() in goroutines
  289. layers_downloaded := false
  290. for i := len(history) - 1; i >= 0; i-- {
  291. id := history[i]
  292. // ensure no two downloads of the same layer happen at the same time
  293. if c, err := s.poolAdd("pull", "layer:"+id); err != nil {
  294. log.Debugf("Image (id: %s) pull is already running, skipping: %v", id, err)
  295. <-c
  296. }
  297. defer s.poolRemove("pull", "layer:"+id)
  298. if !s.graph.Exists(id) {
  299. out.Write(sf.FormatProgress(utils.TruncateID(id), "Pulling metadata", nil))
  300. var (
  301. imgJSON []byte
  302. imgSize int
  303. err error
  304. img *image.Image
  305. )
  306. retries := 5
  307. for j := 1; j <= retries; j++ {
  308. imgJSON, imgSize, err = r.GetRemoteImageJSON(id, endpoint, token)
  309. if err != nil && j == retries {
  310. out.Write(sf.FormatProgress(utils.TruncateID(id), "Error pulling dependent layers", nil))
  311. return layers_downloaded, err
  312. } else if err != nil {
  313. time.Sleep(time.Duration(j) * 500 * time.Millisecond)
  314. continue
  315. }
  316. img, err = image.NewImgJSON(imgJSON)
  317. layers_downloaded = true
  318. if err != nil && j == retries {
  319. out.Write(sf.FormatProgress(utils.TruncateID(id), "Error pulling dependent layers", nil))
  320. return layers_downloaded, fmt.Errorf("Failed to parse json: %s", err)
  321. } else if err != nil {
  322. time.Sleep(time.Duration(j) * 500 * time.Millisecond)
  323. continue
  324. } else {
  325. break
  326. }
  327. }
  328. for j := 1; j <= retries; j++ {
  329. // Get the layer
  330. status := "Pulling fs layer"
  331. if j > 1 {
  332. status = fmt.Sprintf("Pulling fs layer [retries: %d]", j)
  333. }
  334. out.Write(sf.FormatProgress(utils.TruncateID(id), status, nil))
  335. layer, err := r.GetRemoteImageLayer(img.ID, endpoint, token, int64(imgSize))
  336. if uerr, ok := err.(*url.Error); ok {
  337. err = uerr.Err
  338. }
  339. if terr, ok := err.(net.Error); ok && terr.Timeout() && j < retries {
  340. time.Sleep(time.Duration(j) * 500 * time.Millisecond)
  341. continue
  342. } else if err != nil {
  343. out.Write(sf.FormatProgress(utils.TruncateID(id), "Error pulling dependent layers", nil))
  344. return layers_downloaded, err
  345. }
  346. layers_downloaded = true
  347. defer layer.Close()
  348. err = s.graph.Register(img,
  349. utils.ProgressReader(layer, imgSize, out, sf, false, utils.TruncateID(id), "Downloading"))
  350. if terr, ok := err.(net.Error); ok && terr.Timeout() && j < retries {
  351. time.Sleep(time.Duration(j) * 500 * time.Millisecond)
  352. continue
  353. } else if err != nil {
  354. out.Write(sf.FormatProgress(utils.TruncateID(id), "Error downloading dependent layers", nil))
  355. return layers_downloaded, err
  356. } else {
  357. break
  358. }
  359. }
  360. }
  361. out.Write(sf.FormatProgress(utils.TruncateID(id), "Download complete", nil))
  362. }
  363. return layers_downloaded, nil
  364. }
  365. func WriteStatus(requestedTag string, out io.Writer, sf *utils.StreamFormatter, layers_downloaded bool) {
  366. if layers_downloaded {
  367. out.Write(sf.FormatStatus("", "Status: Downloaded newer image for %s", requestedTag))
  368. } else {
  369. out.Write(sf.FormatStatus("", "Status: Image is up to date for %s", requestedTag))
  370. }
  371. }
  372. // downloadInfo is used to pass information from download to extractor
  373. type downloadInfo struct {
  374. imgJSON []byte
  375. img *image.Image
  376. tmpFile *os.File
  377. length int64
  378. downloaded bool
  379. err chan error
  380. }
  381. func (s *TagStore) pullV2Repository(eng *engine.Engine, r *registry.Session, out io.Writer, repoInfo *registry.RepositoryInfo, tag string, sf *utils.StreamFormatter, parallel bool) error {
  382. var layersDownloaded bool
  383. if tag == "" {
  384. log.Debugf("Pulling tag list from V2 registry for %s", repoInfo.CanonicalName)
  385. tags, err := r.GetV2RemoteTags(repoInfo.RemoteName, nil)
  386. if err != nil {
  387. return err
  388. }
  389. for _, t := range tags {
  390. if downloaded, err := s.pullV2Tag(eng, r, out, repoInfo, t, sf, parallel); err != nil {
  391. return err
  392. } else if downloaded {
  393. layersDownloaded = true
  394. }
  395. }
  396. } else {
  397. if downloaded, err := s.pullV2Tag(eng, r, out, repoInfo, tag, sf, parallel); err != nil {
  398. return err
  399. } else if downloaded {
  400. layersDownloaded = true
  401. }
  402. }
  403. requestedTag := repoInfo.CanonicalName
  404. if len(tag) > 0 {
  405. requestedTag = repoInfo.CanonicalName + ":" + tag
  406. }
  407. WriteStatus(requestedTag, out, sf, layersDownloaded)
  408. return nil
  409. }
  410. func (s *TagStore) pullV2Tag(eng *engine.Engine, r *registry.Session, out io.Writer, repoInfo *registry.RepositoryInfo, tag string, sf *utils.StreamFormatter, parallel bool) (bool, error) {
  411. log.Debugf("Pulling tag from V2 registry: %q", tag)
  412. manifestBytes, err := r.GetV2ImageManifest(repoInfo.RemoteName, tag, nil)
  413. if err != nil {
  414. return false, err
  415. }
  416. manifest, verified, err := s.verifyManifest(eng, manifestBytes)
  417. if err != nil {
  418. return false, fmt.Errorf("error verifying manifest: %s", err)
  419. }
  420. if len(manifest.FSLayers) != len(manifest.History) {
  421. return false, fmt.Errorf("length of history not equal to number of layers")
  422. }
  423. if verified {
  424. out.Write(sf.FormatStatus(repoInfo.CanonicalName+":"+tag, "The image you are pulling has been verified"))
  425. } else {
  426. out.Write(sf.FormatStatus(tag, "Pulling from %s", repoInfo.CanonicalName))
  427. }
  428. if len(manifest.FSLayers) == 0 {
  429. return false, fmt.Errorf("no blobSums in manifest")
  430. }
  431. downloads := make([]downloadInfo, len(manifest.FSLayers))
  432. for i := len(manifest.FSLayers) - 1; i >= 0; i-- {
  433. var (
  434. sumStr = manifest.FSLayers[i].BlobSum
  435. imgJSON = []byte(manifest.History[i].V1Compatibility)
  436. )
  437. img, err := image.NewImgJSON(imgJSON)
  438. if err != nil {
  439. return false, fmt.Errorf("failed to parse json: %s", err)
  440. }
  441. downloads[i].img = img
  442. // Check if exists
  443. if s.graph.Exists(img.ID) {
  444. log.Debugf("Image already exists: %s", img.ID)
  445. continue
  446. }
  447. chunks := strings.SplitN(sumStr, ":", 2)
  448. if len(chunks) < 2 {
  449. return false, fmt.Errorf("expected 2 parts in the sumStr, got %#v", chunks)
  450. }
  451. sumType, checksum := chunks[0], chunks[1]
  452. out.Write(sf.FormatProgress(utils.TruncateID(img.ID), "Pulling fs layer", nil))
  453. downloadFunc := func(di *downloadInfo) error {
  454. log.Debugf("pulling blob %q to V1 img %s", sumStr, img.ID)
  455. if c, err := s.poolAdd("pull", "img:"+img.ID); err != nil {
  456. if c != nil {
  457. out.Write(sf.FormatProgress(utils.TruncateID(img.ID), "Layer already being pulled by another client. Waiting.", nil))
  458. <-c
  459. out.Write(sf.FormatProgress(utils.TruncateID(img.ID), "Download complete", nil))
  460. } else {
  461. log.Debugf("Image (id: %s) pull is already running, skipping: %v", img.ID, err)
  462. }
  463. } else {
  464. defer s.poolRemove("pull", "img:"+img.ID)
  465. tmpFile, err := ioutil.TempFile("", "GetV2ImageBlob")
  466. if err != nil {
  467. return err
  468. }
  469. r, l, err := r.GetV2ImageBlobReader(repoInfo.RemoteName, sumType, checksum, nil)
  470. if err != nil {
  471. return err
  472. }
  473. defer r.Close()
  474. io.Copy(tmpFile, utils.ProgressReader(r, int(l), out, sf, false, utils.TruncateID(img.ID), "Downloading"))
  475. out.Write(sf.FormatProgress(utils.TruncateID(img.ID), "Download complete", nil))
  476. log.Debugf("Downloaded %s to tempfile %s", img.ID, tmpFile.Name())
  477. di.tmpFile = tmpFile
  478. di.length = l
  479. di.downloaded = true
  480. }
  481. di.imgJSON = imgJSON
  482. return nil
  483. }
  484. if parallel {
  485. downloads[i].err = make(chan error)
  486. go func(di *downloadInfo) {
  487. di.err <- downloadFunc(di)
  488. }(&downloads[i])
  489. } else {
  490. err := downloadFunc(&downloads[i])
  491. if err != nil {
  492. return false, err
  493. }
  494. }
  495. }
  496. var layersDownloaded bool
  497. for i := len(downloads) - 1; i >= 0; i-- {
  498. d := &downloads[i]
  499. if d.err != nil {
  500. err := <-d.err
  501. if err != nil {
  502. return false, err
  503. }
  504. }
  505. if d.downloaded {
  506. // if tmpFile is empty assume download and extracted elsewhere
  507. defer os.Remove(d.tmpFile.Name())
  508. defer d.tmpFile.Close()
  509. d.tmpFile.Seek(0, 0)
  510. if d.tmpFile != nil {
  511. err = s.graph.Register(d.img,
  512. utils.ProgressReader(d.tmpFile, int(d.length), out, sf, false, utils.TruncateID(d.img.ID), "Extracting"))
  513. if err != nil {
  514. return false, err
  515. }
  516. // FIXME: Pool release here for parallel tag pull (ensures any downloads block until fully extracted)
  517. }
  518. out.Write(sf.FormatProgress(utils.TruncateID(d.img.ID), "Pull complete", nil))
  519. layersDownloaded = true
  520. } else {
  521. out.Write(sf.FormatProgress(utils.TruncateID(d.img.ID), "Already exists", nil))
  522. }
  523. }
  524. if err = s.Set(repoInfo.LocalName, tag, downloads[0].img.ID, true); err != nil {
  525. return false, err
  526. }
  527. return layersDownloaded, nil
  528. }