|
@@ -1,10 +1,12 @@
|
|
|
package graph
|
|
|
|
|
|
import (
|
|
|
+ "errors"
|
|
|
"fmt"
|
|
|
"io"
|
|
|
"io/ioutil"
|
|
|
"os"
|
|
|
+ "sync"
|
|
|
|
|
|
"github.com/Sirupsen/logrus"
|
|
|
"github.com/docker/distribution"
|
|
@@ -73,7 +75,8 @@ func (p *v2Puller) pullV2Repository(tag string) (err error) {
|
|
|
|
|
|
}
|
|
|
|
|
|
- broadcaster, found := p.poolAdd("pull", taggedName)
|
|
|
+ poolKey := "v2:" + taggedName
|
|
|
+ broadcaster, found := p.poolAdd("pull", poolKey)
|
|
|
broadcaster.Add(p.config.OutStream)
|
|
|
if found {
|
|
|
// Another pull of the same repository is already taking place; just wait for it to finish
|
|
@@ -83,7 +86,7 @@ func (p *v2Puller) pullV2Repository(tag string) (err error) {
|
|
|
// This must use a closure so it captures the value of err when the
|
|
|
// function returns, not when the 'defer' is evaluated.
|
|
|
defer func() {
|
|
|
- p.poolRemoveWithError("pull", taggedName, err)
|
|
|
+ p.poolRemoveWithError("pull", poolKey, err)
|
|
|
}()
|
|
|
|
|
|
var layersDownloaded bool
|
|
@@ -104,7 +107,8 @@ func (p *v2Puller) pullV2Repository(tag string) (err error) {
|
|
|
|
|
|
// downloadInfo is used to pass information from download to extractor
|
|
|
type downloadInfo struct {
|
|
|
- img *image.Image
|
|
|
+ img contentAddressableDescriptor
|
|
|
+ imgIndex int
|
|
|
tmpFile *os.File
|
|
|
digest digest.Digest
|
|
|
layer distribution.ReadSeekCloser
|
|
@@ -114,12 +118,66 @@ type downloadInfo struct {
|
|
|
broadcaster *broadcaster.Buffered
|
|
|
}
|
|
|
|
|
|
+// contentAddressableDescriptor is used to pass image data from a manifest to the
|
|
|
+// graph.
|
|
|
+type contentAddressableDescriptor struct {
|
|
|
+ id string
|
|
|
+ parent string
|
|
|
+ strongID digest.Digest
|
|
|
+ compatibilityID string
|
|
|
+ config []byte
|
|
|
+ v1Compatibility []byte
|
|
|
+}
|
|
|
+
|
|
|
+func newContentAddressableImage(v1Compatibility []byte, blobSum digest.Digest, parent digest.Digest) (contentAddressableDescriptor, error) {
|
|
|
+ img := contentAddressableDescriptor{
|
|
|
+ v1Compatibility: v1Compatibility,
|
|
|
+ }
|
|
|
+
|
|
|
+ var err error
|
|
|
+ img.config, err = image.MakeImageConfig(v1Compatibility, blobSum, parent)
|
|
|
+ if err != nil {
|
|
|
+ return img, err
|
|
|
+ }
|
|
|
+ img.strongID, err = image.StrongID(img.config)
|
|
|
+ if err != nil {
|
|
|
+ return img, err
|
|
|
+ }
|
|
|
+
|
|
|
+ unmarshalledConfig, err := image.NewImgJSON(v1Compatibility)
|
|
|
+ if err != nil {
|
|
|
+ return img, err
|
|
|
+ }
|
|
|
+
|
|
|
+ img.compatibilityID = unmarshalledConfig.ID
|
|
|
+ img.id = img.strongID.Hex()
|
|
|
+
|
|
|
+ return img, nil
|
|
|
+}
|
|
|
+
|
|
|
+// ID returns the actual ID to be used for the downloaded image. This may be
|
|
|
+// a computed ID.
|
|
|
+func (img contentAddressableDescriptor) ID() string {
|
|
|
+ return img.id
|
|
|
+}
|
|
|
+
|
|
|
+// Parent returns the parent ID to be used for the image. This may be a
|
|
|
+// computed ID.
|
|
|
+func (img contentAddressableDescriptor) Parent() string {
|
|
|
+ return img.parent
|
|
|
+}
|
|
|
+
|
|
|
+// MarshalConfig renders the image structure into JSON.
|
|
|
+func (img contentAddressableDescriptor) MarshalConfig() ([]byte, error) {
|
|
|
+ return img.config, nil
|
|
|
+}
|
|
|
+
|
|
|
type errVerification struct{}
|
|
|
|
|
|
func (errVerification) Error() string { return "verification failed" }
|
|
|
|
|
|
func (p *v2Puller) download(di *downloadInfo) {
|
|
|
- logrus.Debugf("pulling blob %q to %s", di.digest, di.img.ID)
|
|
|
+ logrus.Debugf("pulling blob %q to %s", di.digest, di.img.id)
|
|
|
|
|
|
blobs := p.repo.Blobs(context.Background())
|
|
|
|
|
@@ -151,12 +209,12 @@ func (p *v2Puller) download(di *downloadInfo) {
|
|
|
Formatter: p.sf,
|
|
|
Size: di.size,
|
|
|
NewLines: false,
|
|
|
- ID: stringid.TruncateID(di.img.ID),
|
|
|
+ ID: stringid.TruncateID(di.img.id),
|
|
|
Action: "Downloading",
|
|
|
})
|
|
|
io.Copy(di.tmpFile, reader)
|
|
|
|
|
|
- di.broadcaster.Write(p.sf.FormatProgress(stringid.TruncateID(di.img.ID), "Verifying Checksum", nil))
|
|
|
+ di.broadcaster.Write(p.sf.FormatProgress(stringid.TruncateID(di.img.id), "Verifying Checksum", nil))
|
|
|
|
|
|
if !verifier.Verified() {
|
|
|
err = fmt.Errorf("filesystem layer verification failed for digest %s", di.digest)
|
|
@@ -165,9 +223,9 @@ func (p *v2Puller) download(di *downloadInfo) {
|
|
|
return
|
|
|
}
|
|
|
|
|
|
- di.broadcaster.Write(p.sf.FormatProgress(stringid.TruncateID(di.img.ID), "Download complete", nil))
|
|
|
+ di.broadcaster.Write(p.sf.FormatProgress(stringid.TruncateID(di.img.id), "Download complete", nil))
|
|
|
|
|
|
- logrus.Debugf("Downloaded %s to tempfile %s", di.img.ID, di.tmpFile.Name())
|
|
|
+ logrus.Debugf("Downloaded %s to tempfile %s", di.img.id, di.tmpFile.Name())
|
|
|
di.layer = layerDownload
|
|
|
|
|
|
di.err <- nil
|
|
@@ -193,6 +251,17 @@ func (p *v2Puller) pullV2Tag(out io.Writer, tag, taggedName string) (verified bo
|
|
|
logrus.Printf("Image manifest for %s has been verified", taggedName)
|
|
|
}
|
|
|
|
|
|
+ // remove duplicate layers and check parent chain validity
|
|
|
+ err = fixManifestLayers(&manifest.Manifest)
|
|
|
+ if err != nil {
|
|
|
+ return false, err
|
|
|
+ }
|
|
|
+
|
|
|
+ imgs, err := p.getImageInfos(manifest.Manifest)
|
|
|
+ if err != nil {
|
|
|
+ return false, err
|
|
|
+ }
|
|
|
+
|
|
|
out.Write(p.sf.FormatStatus(tag, "Pulling from %s", p.repo.Name()))
|
|
|
|
|
|
var downloads []*downloadInfo
|
|
@@ -213,26 +282,32 @@ func (p *v2Puller) pullV2Tag(out io.Writer, tag, taggedName string) (verified bo
|
|
|
}()
|
|
|
|
|
|
for i := len(manifest.FSLayers) - 1; i >= 0; i-- {
|
|
|
- img, err := image.NewImgJSON([]byte(manifest.History[i].V1Compatibility))
|
|
|
- if err != nil {
|
|
|
- logrus.Debugf("error getting image v1 json: %v", err)
|
|
|
- return false, err
|
|
|
- }
|
|
|
- p.graph.Retain(p.sessionID, img.ID)
|
|
|
- layerIDs = append(layerIDs, img.ID)
|
|
|
+ img := imgs[i]
|
|
|
+
|
|
|
+ p.graph.Retain(p.sessionID, img.id)
|
|
|
+ layerIDs = append(layerIDs, img.id)
|
|
|
+
|
|
|
+ p.graph.imageMutex.Lock(img.id)
|
|
|
|
|
|
// Check if exists
|
|
|
- if p.graph.Exists(img.ID) {
|
|
|
- logrus.Debugf("Image already exists: %s", img.ID)
|
|
|
- out.Write(p.sf.FormatProgress(stringid.TruncateID(img.ID), "Already exists", nil))
|
|
|
+ if p.graph.Exists(img.id) {
|
|
|
+ if err := p.validateImageInGraph(img.id, imgs, i); err != nil {
|
|
|
+ p.graph.imageMutex.Unlock(img.id)
|
|
|
+ return false, fmt.Errorf("image validation failed: %v", err)
|
|
|
+ }
|
|
|
+ logrus.Debugf("Image already exists: %s", img.id)
|
|
|
+ p.graph.imageMutex.Unlock(img.id)
|
|
|
continue
|
|
|
}
|
|
|
- out.Write(p.sf.FormatProgress(stringid.TruncateID(img.ID), "Pulling fs layer", nil))
|
|
|
+ p.graph.imageMutex.Unlock(img.id)
|
|
|
+
|
|
|
+ out.Write(p.sf.FormatProgress(stringid.TruncateID(img.id), "Pulling fs layer", nil))
|
|
|
|
|
|
d := &downloadInfo{
|
|
|
- img: img,
|
|
|
- poolKey: "layer:" + img.ID,
|
|
|
- digest: manifest.FSLayers[i].BlobSum,
|
|
|
+ img: img,
|
|
|
+ imgIndex: i,
|
|
|
+ poolKey: "v2layer:" + img.id,
|
|
|
+ digest: manifest.FSLayers[i].BlobSum,
|
|
|
// TODO: seems like this chan buffer solved hanging problem in go1.5,
|
|
|
// this can indicate some deeper problem that somehow we never take
|
|
|
// error from channel in loop below
|
|
@@ -274,26 +349,49 @@ func (p *v2Puller) pullV2Tag(out io.Writer, tag, taggedName string) (verified bo
|
|
|
}
|
|
|
|
|
|
d.tmpFile.Seek(0, 0)
|
|
|
- reader := progressreader.New(progressreader.Config{
|
|
|
- In: d.tmpFile,
|
|
|
- Out: d.broadcaster,
|
|
|
- Formatter: p.sf,
|
|
|
- Size: d.size,
|
|
|
- NewLines: false,
|
|
|
- ID: stringid.TruncateID(d.img.ID),
|
|
|
- Action: "Extracting",
|
|
|
- })
|
|
|
-
|
|
|
- err = p.graph.Register(d.img, reader)
|
|
|
- if err != nil {
|
|
|
- return false, err
|
|
|
- }
|
|
|
+ err := func() error {
|
|
|
+ reader := progressreader.New(progressreader.Config{
|
|
|
+ In: d.tmpFile,
|
|
|
+ Out: d.broadcaster,
|
|
|
+ Formatter: p.sf,
|
|
|
+ Size: d.size,
|
|
|
+ NewLines: false,
|
|
|
+ ID: stringid.TruncateID(d.img.id),
|
|
|
+ Action: "Extracting",
|
|
|
+ })
|
|
|
+
|
|
|
+ p.graph.imageMutex.Lock(d.img.id)
|
|
|
+ defer p.graph.imageMutex.Unlock(d.img.id)
|
|
|
+
|
|
|
+ // Must recheck the data on disk if any exists.
|
|
|
+ // This protects against races where something
|
|
|
+ // else is written to the graph under this ID
|
|
|
+ // after attemptIDReuse.
|
|
|
+ if p.graph.Exists(d.img.id) {
|
|
|
+ if err := p.validateImageInGraph(d.img.id, imgs, d.imgIndex); err != nil {
|
|
|
+ return fmt.Errorf("image validation failed: %v", err)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if err := p.graph.register(d.img, reader); err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+
|
|
|
+ if err := p.graph.setLayerDigest(d.img.id, d.digest); err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+
|
|
|
+ if err := p.graph.setV1CompatibilityConfig(d.img.id, d.img.v1Compatibility); err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
|
|
|
- if err := p.graph.SetDigest(d.img.ID, d.digest); err != nil {
|
|
|
+ return nil
|
|
|
+ }()
|
|
|
+ if err != nil {
|
|
|
return false, err
|
|
|
}
|
|
|
|
|
|
- d.broadcaster.Write(p.sf.FormatProgress(stringid.TruncateID(d.img.ID), "Pull complete", nil))
|
|
|
+ d.broadcaster.Write(p.sf.FormatProgress(stringid.TruncateID(d.img.id), "Pull complete", nil))
|
|
|
d.broadcaster.Close()
|
|
|
tagUpdated = true
|
|
|
}
|
|
@@ -424,3 +522,217 @@ func (p *v2Puller) validateManifest(m *manifest.SignedManifest, tag string) (ver
|
|
|
}
|
|
|
return verified, nil
|
|
|
}
|
|
|
+
|
|
|
+// fixManifestLayers removes repeated layers from the manifest and checks the
|
|
|
+// correctness of the parent chain.
|
|
|
+func fixManifestLayers(m *manifest.Manifest) error {
|
|
|
+ images := make([]*image.Image, len(m.FSLayers))
|
|
|
+ for i := range m.FSLayers {
|
|
|
+ img, err := image.NewImgJSON([]byte(m.History[i].V1Compatibility))
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ images[i] = img
|
|
|
+ if err := image.ValidateID(img.ID); err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if images[len(images)-1].Parent != "" {
|
|
|
+ return errors.New("Invalid parent ID in the base layer of the image.")
|
|
|
+ }
|
|
|
+
|
|
|
+ // check general duplicates to error instead of a deadlock
|
|
|
+ idmap := make(map[string]struct{})
|
|
|
+
|
|
|
+ var lastID string
|
|
|
+ for _, img := range images {
|
|
|
+ // skip IDs that appear after each other, we handle those later
|
|
|
+ if _, exists := idmap[img.ID]; img.ID != lastID && exists {
|
|
|
+ return fmt.Errorf("ID %+v appears multiple times in manifest", img.ID)
|
|
|
+ }
|
|
|
+ lastID = img.ID
|
|
|
+ idmap[lastID] = struct{}{}
|
|
|
+ }
|
|
|
+
|
|
|
+ // backwards loop so that we keep the remaining indexes after removing items
|
|
|
+ for i := len(images) - 2; i >= 0; i-- {
|
|
|
+ if images[i].ID == images[i+1].ID { // repeated ID. remove and continue
|
|
|
+ m.FSLayers = append(m.FSLayers[:i], m.FSLayers[i+1:]...)
|
|
|
+ m.History = append(m.History[:i], m.History[i+1:]...)
|
|
|
+ } else if images[i].Parent != images[i+1].ID {
|
|
|
+ return fmt.Errorf("Invalid parent ID. Expected %v, got %v.", images[i+1].ID, images[i].Parent)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+// getImageInfos returns an imageinfo struct for every image in the manifest.
|
|
|
+// These objects contain both calculated strongIDs and compatibilityIDs found
|
|
|
+// in v1Compatibility object.
|
|
|
+func (p *v2Puller) getImageInfos(m manifest.Manifest) ([]contentAddressableDescriptor, error) {
|
|
|
+ imgs := make([]contentAddressableDescriptor, len(m.FSLayers))
|
|
|
+
|
|
|
+ var parent digest.Digest
|
|
|
+ for i := len(imgs) - 1; i >= 0; i-- {
|
|
|
+ var err error
|
|
|
+ imgs[i], err = newContentAddressableImage([]byte(m.History[i].V1Compatibility), m.FSLayers[i].BlobSum, parent)
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ parent = imgs[i].strongID
|
|
|
+ }
|
|
|
+
|
|
|
+ p.attemptIDReuse(imgs)
|
|
|
+
|
|
|
+ return imgs, nil
|
|
|
+}
|
|
|
+
|
|
|
+var idReuseLock sync.Mutex
|
|
|
+
|
|
|
+// attemptIDReuse does a best attempt to match verified compatibilityIDs
|
|
|
+// already in the graph with the computed strongIDs so we can keep using them.
|
|
|
+// This process will never fail but may just return the strongIDs if none of
|
|
|
+// the compatibilityIDs exists or can be verified. If the strongIDs themselves
|
|
|
+// fail verification, we deterministically generate alternate IDs to use until
|
|
|
+// we find one that's available or already exists with the correct data.
|
|
|
+func (p *v2Puller) attemptIDReuse(imgs []contentAddressableDescriptor) {
|
|
|
+ // This function needs to be protected with a global lock, because it
|
|
|
+ // locks multiple IDs at once, and there's no good way to make sure
|
|
|
+ // the locking happens a deterministic order.
|
|
|
+ idReuseLock.Lock()
|
|
|
+ defer idReuseLock.Unlock()
|
|
|
+
|
|
|
+ idMap := make(map[string]struct{})
|
|
|
+ for _, img := range imgs {
|
|
|
+ idMap[img.id] = struct{}{}
|
|
|
+ idMap[img.compatibilityID] = struct{}{}
|
|
|
+
|
|
|
+ if p.graph.Exists(img.compatibilityID) {
|
|
|
+ if _, err := p.graph.GenerateV1CompatibilityChain(img.compatibilityID); err != nil {
|
|
|
+ logrus.Debugf("Migration v1Compatibility generation error: %v", err)
|
|
|
+ return
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ for id := range idMap {
|
|
|
+ p.graph.imageMutex.Lock(id)
|
|
|
+ defer p.graph.imageMutex.Unlock(id)
|
|
|
+ }
|
|
|
+
|
|
|
+ // continueReuse controls whether the function will try to find
|
|
|
+ // existing layers on disk under the old v1 IDs, to avoid repulling
|
|
|
+ // them. The hashes are checked to ensure these layers are okay to
|
|
|
+ // use. continueReuse starts out as true, but is set to false if
|
|
|
+ // the code encounters something that doesn't match the expected hash.
|
|
|
+ continueReuse := true
|
|
|
+
|
|
|
+ for i := len(imgs) - 1; i >= 0; i-- {
|
|
|
+ if p.graph.Exists(imgs[i].id) {
|
|
|
+ // Found an image in the graph under the strongID. Validate the
|
|
|
+ // image before using it.
|
|
|
+ if err := p.validateImageInGraph(imgs[i].id, imgs, i); err != nil {
|
|
|
+ continueReuse = false
|
|
|
+ logrus.Debugf("not using existing strongID: %v", err)
|
|
|
+
|
|
|
+ // The strong ID existed in the graph but didn't
|
|
|
+ // validate successfully. We can't use the strong ID
|
|
|
+ // because it didn't validate successfully. Treat the
|
|
|
+ // graph like a hash table with probing... compute
|
|
|
+ // SHA256(id) until we find an ID that either doesn't
|
|
|
+ // already exist in the graph, or has existing content
|
|
|
+ // that validates successfully.
|
|
|
+ for {
|
|
|
+ if err := p.tryNextID(imgs, i, idMap); err != nil {
|
|
|
+ logrus.Debug(err.Error())
|
|
|
+ } else {
|
|
|
+ break
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ continue
|
|
|
+ }
|
|
|
+
|
|
|
+ if continueReuse {
|
|
|
+ compatibilityID := imgs[i].compatibilityID
|
|
|
+ if err := p.validateImageInGraph(compatibilityID, imgs, i); err != nil {
|
|
|
+ logrus.Debugf("stopping ID reuse: %v", err)
|
|
|
+ continueReuse = false
|
|
|
+ } else {
|
|
|
+ // The compatibility ID exists in the graph and was
|
|
|
+ // validated. Use it.
|
|
|
+ imgs[i].id = compatibilityID
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // fix up the parents of the images
|
|
|
+ for i := 0; i < len(imgs); i++ {
|
|
|
+ if i == len(imgs)-1 { // Base layer
|
|
|
+ imgs[i].parent = ""
|
|
|
+ } else {
|
|
|
+ imgs[i].parent = imgs[i+1].id
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// validateImageInGraph checks that an image in the graph has the expected
|
|
|
+// strongID. id is the entry in the graph to check, imgs is the slice of
|
|
|
+// images being processed (for access to the parent), and i is the index
|
|
|
+// into this slice which the graph entry should be checked against.
|
|
|
+func (p *v2Puller) validateImageInGraph(id string, imgs []contentAddressableDescriptor, i int) error {
|
|
|
+ img, err := p.graph.Get(id)
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("missing: %v", err)
|
|
|
+ }
|
|
|
+ layerID, err := p.graph.getLayerDigest(id)
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("digest: %v", err)
|
|
|
+ }
|
|
|
+ var parentID digest.Digest
|
|
|
+ if i != len(imgs)-1 {
|
|
|
+ if img.Parent != imgs[i+1].id { // comparing that graph points to validated ID
|
|
|
+ return fmt.Errorf("parent: %v %v", img.Parent, imgs[i+1].id)
|
|
|
+ }
|
|
|
+ parentID = imgs[i+1].strongID
|
|
|
+ } else if img.Parent != "" {
|
|
|
+ return fmt.Errorf("unexpected parent: %v", img.Parent)
|
|
|
+ }
|
|
|
+
|
|
|
+ v1Config, err := p.graph.getV1CompatibilityConfig(img.ID)
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("v1Compatibility: %v %v", img.ID, err)
|
|
|
+ }
|
|
|
+
|
|
|
+ json, err := image.MakeImageConfig(v1Config, layerID, parentID)
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("make config: %v", err)
|
|
|
+ }
|
|
|
+
|
|
|
+ if dgst, err := image.StrongID(json); err == nil && dgst == imgs[i].strongID {
|
|
|
+ logrus.Debugf("Validated %v as %v", dgst, id)
|
|
|
+ } else {
|
|
|
+ return fmt.Errorf("digest mismatch: %v %v, error: %v", dgst, imgs[i].strongID, err)
|
|
|
+ }
|
|
|
+
|
|
|
+ // All clear
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+func (p *v2Puller) tryNextID(imgs []contentAddressableDescriptor, i int, idMap map[string]struct{}) error {
|
|
|
+ nextID, _ := digest.FromBytes([]byte(imgs[i].id))
|
|
|
+ imgs[i].id = nextID.Hex()
|
|
|
+
|
|
|
+ if _, exists := idMap[imgs[i].id]; !exists {
|
|
|
+ p.graph.imageMutex.Lock(imgs[i].id)
|
|
|
+ defer p.graph.imageMutex.Unlock(imgs[i].id)
|
|
|
+ }
|
|
|
+
|
|
|
+ if p.graph.Exists(imgs[i].id) {
|
|
|
+ if err := p.validateImageInGraph(imgs[i].id, imgs, i); err != nil {
|
|
|
+ return fmt.Errorf("not using existing strongID permutation %s: %v", imgs[i].id, err)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return nil
|
|
|
+}
|