docker 源码分析(5) -- 镜像拉取及存储

1
2
3
4
5
6
7
8
9

if err := cli.Cmd(flag.Args()...); err != nil {
    if sterr, ok := err.(*utils.StatusError); ok {
        if sterr.Status != "" {
	        log.Println(sterr.Status)
        }
        os.Exit(sterr.StatusCode)
    }
    log.Fatal(err)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

func (cli *DockerCli) getMethod(args ...string) (func(...string) error, bool) {
    camelArgs := make([]string, len(args))
    for i, s := range args {
        if len(s) == 0 {
            return nil, false
        }
        camelArgs[i] = strings.ToUpper(s[:1]) + strings.ToLower(s[1:])
    }
    methodName := "Cmd" + strings.Join(camelArgs, "")
    method := reflect.ValueOf(cli).MethodByName(methodName)
    if !method.IsValid() {
        return nil, false
    }
    return method.Interface().(func(...string) error), true
}

1
2
3
4
5
6
7
8

taglessRemote, tag := parsers.ParseRepositoryTag(remote)
if tag == "" && !*allTags {
    newRemote = taglessRemote + ":" + graph.DEFAULTTAG
}

if tag != "" && *allTags {
    return fmt.Errorf("tag can't be used with --all-tags/-a")
}

1
2
3
4
5
6
7
8

hostname, _, err := registry.ResolveRepositoryName(taglessRemote)
if err != nil {
	return err
}

cli.LoadConfigFile()

authConfig := cli.configFile.ResolveAuthConfig(hostname)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

v  = url.Values{}

v.Set("fromImage", newRemote)

pull := func(authConfig registry.AuthConfig) error {
    buf, err := json.Marshal(authConfig)
    if err != nil {
        return err
    }
    registryAuthHeader := []string{
        base64.URLEncoding.EncodeToString(buf),
    }
        
    return cli.stream("POST", "/images/create?"+v.Encode(), nil, cli.out, map[string][]string{
        "X-Registry-Auth": registryAuthHeader,
    })
}

if err := pull(authConfig); err != nil {
    ...
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

image = r.Form.Get("fromImage")
repo  = r.Form.Get("repo")
tag   = r.Form.Get("tag")


if image != "" { //pull
    if tag == "" {
        image, tag = parsers.ParseRepositoryTag(image)
    }
    metaHeaders := map[string][]string{}
    for k, v := range r.Header {
        if strings.HasPrefix(k, "X-Meta-") {
            metaHeaders[k] = v
        }
    }
    job = eng.Job("pull", image, tag)
    job.SetenvBool("parallel", version.GreaterThan("1.3"))
    job.SetenvJson("metaHeaders", metaHeaders)
    job.SetenvJson("authConfig", authConfig)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

c, err := s.poolAdd("pull", localName+":"+tag)
if err != nil {
    if c != nil {
        job.Stdout.Write(sf.FormatStatus("", "Repository %s already being pulled by another client. Waiting.", localName))
        <-c
        return engine.StatusOK
    }
    return job.Error(err)
}
defer s.poolRemove("pull", localName+":"+tag)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

hostname, remoteName, err := registry.ResolveRepositoryName(localName)

endpoint, err := registry.NewEndpoint(hostname, s.insecureRegistries)

r, err := registry.NewSession(authConfig, registry.HTTPRequestFactory(metaHeaders), endpoint, true)

if endpoint.VersionString(1) == registry.IndexServerAddress() {
    localName = remoteName

    isOfficial = isOfficialName(remoteName)
    if isOfficial && strings.IndexRune(remoteName, '/') == -1 {
        remoteName = "library/" + remoteName
    }

    mirrors = s.mirrors
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

if len(mirrors) == 0 && (isOfficial || endpoint.Version == registry.APIVersion2) {
    j := job.Eng.Job("trust_update_base")
    if err = j.Run(); err != nil {
        return job.Errorf("error updating trust base graph: %s", err)
    }

    if err := s.pullV2Repository(job.Eng, r, job.Stdout, localName, remoteName, tag, sf, job.GetenvBool("parallel")); err == nil {
        if err = job.Eng.Job("log", "pull", logName, "").Run(); err != nil {
            log.Errorf("Error logging event 'pull' for %s: %s", logName, err)
        }
        return engine.StatusOK
    } else if err != registry.ErrDoesNotExist {
        log.Errorf("Error from V2 registry: %s", err)
    }
}

if err = s.pullRepository(r, job.Stdout, localName, remoteName, tag, sf, job.GetenvBool("parallel"), mirrors); err != nil {
    return job.Error(err)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

tags, err := r.GetV2RemoteTags(remoteName, nil)
if err != nil {
    return err
}
for _, t := range tags {
    if downloaded, err := s.pullV2Tag(eng, r, out, localName, remoteName, t, sf, parallel); err != nil {
        return err
    } else if downloaded {
        layersDownloaded = true
    }
}

1
2
3
4
5
6
7
8

type ManifestData struct {
    Name          string             `json:"name"`
    Tag           string             `json:"tag"`
    Architecture  string             `json:"architecture"`
    FSLayers      []*FSLayer         `json:"fsLayers"`
    History       []*ManifestHistory `json:"history"`
    SchemaVersion int                `json:"schemaVersion"`
}

1
2
3
4
5
6

log.Debugf("Pulling tag from V2 registry: %q", tag)
manifestBytes, err := r.GetV2ImageManifest(remoteName, tag, nil)
// 验证各项信息是否正确
manifest, verified, err := s.verifyManifest(eng, manifestBytes)

downloads := make([]downloadInfo, len(manifest.FSLayers))

1
2
3
4
5
6
7
8

type downloadInfo struct {
    imgJSON    []byte
    img        *image.Image
    tmpFile    *os.File
    length     int64
    downloaded bool
    err        chan error
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

for i := len(manifest.FSLayers) - 1; i >= 0; i-- {
    var (
        sumStr  = manifest.FSLayers[i].BlobSum
        imgJSON = []byte(manifest.History[i].V1Compatibility)
    )

    img, err := image.NewImgJSON(imgJSON)
    if err != nil {
        return false, fmt.Errorf("failed to parse json: %s", err)
    }
    downloads[i].img = img

1
2
3
4

if s.graph.Exists(img.ID) {
    log.Debugf("Image already exists: %s", img.ID)
    continue
}

1
2
3
4
5
6
7
8
9

if c, err := s.poolAdd("pull", "img:"+img.ID); err != nil {
    if c != nil {
        out.Write(sf.FormatProgress(utils.TruncateID(img.ID), "Layer already being pulled by another client. Waiting.", nil))
        <-c
        out.Write(sf.FormatProgress(utils.TruncateID(img.ID), "Download complete", nil))
    } else {
        log.Debugf("Image (id: %s) pull is already running, skipping: %v", img.ID, err)
    }
} 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

// 本地文件名
tmpFile, err := ioutil.TempFile("", "GetV2ImageBlob")
if err != nil {
    return err
}

// 下载
r, l, err := r.GetV2ImageBlobReader(remoteName, sumType, checksum, nil)
if err != nil {
    return err
}
defer r.Close()
io.Copy(tmpFile, utils.ProgressReader(r, int(l), out, sf, false, utils.TruncateID(img.ID), "Downloading"))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

if parallel {
    downloads[i].err = make(chan error)
    go func(di *downloadInfo) {
        di.err <- downloadFunc(di)
    }(&downloads[i])
} else {
    err := downloadFunc(&downloads[i])
    if err != nil {
        return false, err
    }
}

1
2
3
4
5
6
7

if d.tmpFile != nil {
    err = s.graph.Register(d.img,
        utils.ProgressReader(d.tmpFile, int(d.length), out, sf, false, utils.TruncateID(d.img.ID), "Extracting"))
    if err != nil {
        return false, err
    }
}

1
2
3
4
5

type Graph struct {
    Root    string
    idIndex *truncindex.TruncIndex
    driver  graphdriver.Driver
}

1
2
3
4
5

type TruncIndex struct {
    sync.RWMutex
    trie *patricia.Trie
    ids  map[string]struct{}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

type Driver interface {
    ProtoDriver
    // Diff produces an archive of the changes between the specified
    // layer and its parent layer which may be "".
    Diff(id, parent string) (archive.Archive, error)
    // Changes produces a list of changes between the specified layer
    // and its parent layer. If parent is "", then all changes will be ADD changes.
    Changes(id, parent string) ([]archive.Change, error)
    // ApplyDiff extracts the changeset from the given diff into the
    // layer with the specified id and parent, returning the size of the
    // new layer in bytes.
    ApplyDiff(id, parent string, diff archive.ArchiveReader) (bytes int64, err error)
    // DiffSize calculates the changes between the specified id
    // and its parent and returns the size in bytes of the changes
    // relative to its base filesystem directory.
    DiffSize(id, parent string) (bytes int64, err error)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

type ProtoDriver interface {
    // String returns a string representation of this driver.
    String() string
    // Create creates a new, empty, filesystem layer with the
    // specified id and parent. Parent may be "".
    Create(id, parent string) error
    // Remove attempts to remove the filesystem layer with this id.
    Remove(id string) error
    // Get returns the mountpoint for the layered filesystem referred
    // to by this id. You can optionally specify a mountLabel or "".
    // Returns the absolute path to the mounted layered filesystem.
    Get(id, mountLabel string) (dir string, err error)
    // Put releases the system resources for the specified id,
    // e.g, unmounting layered filesystem.
    Put(id string)
    // Exists returns whether a filesystem layer with the specified
    // ID exists on this driver.
    Exists(id string) bool
    // Status returns a set of key-value pairs which give low
    // level diagnostic status about this driver.
    Status() [][2]string
    // Cleanup performs necessary tasks to release resources
    // held by the driver, e.g., unmounting all layered filesystems
    // known to this driver.
    Cleanup() error
}

1
2
3

if err := utils.ValidateID(img.ID); err != nil {
    return err
}

1
2
3
4
5
6
7
8
9

if graph.Exists(img.ID) {
    return fmt.Errorf("Image %s already exists", img.ID)
}

if err := os.RemoveAll(graph.ImageRoot(img.ID)); err != nil && !os.IsNotExist(err) {
    return err
}

graph.driver.Remove(img.ID)

1
2
3

if err := graph.driver.Create(img.ID, img.Parent); err != nil {
	return fmt.Errorf("Driver %s failed to create image rootfs %s: %s", graph.driver, img.ID, err)
}

1
2
3

if err := a.createDirsFor(id); err != nil {
	return err
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

func (a *Driver) createDirsFor(id string) error {
    paths := []string{
        "mnt",
        "diff",
    }

    for _, p := range paths {
        if err := os.MkdirAll(path.Join(a.rootPath(), p, id), 0755); err != nil {
            return err
        }
    }
    return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

f, err := os.Create(path.Join(a.rootPath(), "layers", id))
if err != nil {
    return err
}

if parent != "" {
    ids, err := getParentIds(a.rootPath(), parent)
    if err != nil {
        return err
    }

    if _, err := fmt.Fprintln(f, parent); err != nil {
        return err
    }
    for _, i := range ids {
        if _, err := fmt.Fprintln(f, i); err != nil {
            return err
        }
    }
}

1
2
3
4

img.SetGraph(graph)
if err := image.StoreImage(img, layerData, tmp); err != nil {
	return err
}

1
2
3
4
5
6
7
8
9

layerDataDecompressed, err := archive.DecompressStream(layerData)

if layerTarSum, err = tarsum.NewTarSum(layerDataDecompressed, true, tarsum.VersionDev); err != nil {
    return err
}

if size, err = driver.ApplyDiff(img.ID, img.Parent, layerTarSum); err != nil {
    return err
}

1
2
3
4
5
6
7
8

func (a *Driver) ApplyDiff(id, parent string, diff archive.ArchiveReader) (bytes int64, err error) {
    // AUFS doesn't need the parent id to apply the diff.
    if err = a.applyDiff(id, diff); err != nil {
        return
    }

    return a.DiffSize(id, parent)
}

1
2
3
4

// 解压数据
func (a *Driver) applyDiff(id string, diff archive.ArchiveReader) error {
    return chrootarchive.Untar(diff, path.Join(a.rootPath(), "diff", id), nil)
}

1
2
3
4
5

// 遍历目录计算大小
func (a *Driver) DiffSize(id, parent string) (bytes int64, err error) {
    // AUFS doesn't need the parent layer to calculate the diff size.
    return utils.TreeSize(path.Join(a.rootPath(), "diff", id))
}

1
2
3
4

img.Size = size
if err := img.SaveSize(root); err != nil {
    return err
}

1
2
3
4
5
6
7
8

f, err := os.OpenFile(jsonPath(root), os.O_CREATE|os.O_WRONLY|os.O_TRUNC, os.FileMode(0600))
if err != nil {
    return err
}

defer f.Close()

return json.NewEncoder(f).Encode(img)

1
2
3
4
5

checksum := layerTarSum.Sum(nil)

if img.Checksum != "" && img.Checksum != checksum {
    log.Warnf("image layer checksum mismatch: computed %q, expected %q", checksum, img.Checksum)
}

1
2
3

if err := os.Rename(tmp, graph.ImageRoot(img.ID)); err != nil {
    return err
}

1

graph.idIndex.Add(img.ID)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

type TagStore struct {
    path               string
    graph              *Graph
    mirrors            []string
    insecureRegistries []string
    Repositories       map[string]Repository
    sync.Mutex
    // FIXME: move push/pull-related fields
    // to a helper type
    pullingPool map[string]chan struct{}
    pushingPool map[string]chan struct{}
}

type Repository map[string]string

1
2
3

if err = s.Set(localName, tag, downloads[0].img.ID, true); err != nil {
    return false, err
}

1
2
3
4
5
6

img, err := store.LookupImage(imageName)
store.Lock()
defer store.Unlock()
if err != nil {
    return err
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

img, err := store.GetImage(repos, tag)
store.Lock()
defer store.Unlock()
if err != nil {
    return nil, err
} else if img == nil {
    if img, err = store.graph.Get(name); err != nil {
        return nil, err
    }
}
return img,nil

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

if err := store.reload(); err != nil {
    return err
}
var repo Repository
if r, exists := store.Repositories[repoName]; exists {
    repo = r
    if old, exists := store.Repositories[repoName][tag]; exists && !force {
        return fmt.Errorf("Conflict: Tag %s is already set to image %s, if you want to replace it, please use -f option", tag, old)
    }
} else {
    repo = make(map[string]string)
    store.Repositories[repoName] = repo
}
repo[tag] = img.ID
return store.save()