docker pull命令实现与镜像存储（3）_context.cancelfunc docker

在《pull命令实现与镜像存储（1）》和《pull命令实现与镜像存储（2）》我们分析pull命令在docker客户端的实现部分，最后我们了解到客户端将结构化参数发送到服务端的URL：/images/create。接下来我们将分析在服务端的实现部分，将从该URL入手。
我们在《dockerd路由和初始化》中了解了docker的API是如何初始化的，实现在docker\cmd\dockerd\daemon.go我们回顾下：

func initRouter(s *apiserver.Server, d *daemon.Daemon, c *cluster.Cluster) {
    decoder := runconfig.ContainerDecoder{}

    routers := []router.Router{}

    // we need to add the checkpoint router before the container router or the DELETE gets masked
    routers = addExperimentalRouters(routers, d, decoder)

    routers = append(routers, []router.Router{
        container.NewRouter(d, decoder), //关于容器命令的API
        image.NewRouter(d, decoder),     //关于镜命令的API
        systemrouter.NewRouter(d, c),    //关于系命令的API api/server/router/system被重命名了
        volume.NewRouter(d),             //关于卷命令的API
        build.NewRouter(dockerfile.NewBuildManager(d)),//关于构建命令的API
        swarmrouter.NewRouter(c),
    }...)

    if d.NetworkControllerEnabled() {
        routers = append(routers, network.NewRouter(d, c))
    }

    s.InitRouter(utils.IsDebugEnabled(), routers...)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

可以看到有关于镜像的API “ image.NewRouter(d, decoder)”，实现在docker\api\server\router\image\image.go：

// NewRouter initializes a new image router
func NewRouter(backend Backend, decoder httputils.ContainerDecoder) router.Router {
    r := &imageRouter{
        backend: backend,
        decoder: decoder,
    }
    r.initRoutes()
    return r
}

// Routes returns the available routes to the image controller
func (r *imageRouter) Routes() []router.Route {
    return r.routes
}

// initRoutes initializes the routes in the image router
func (r *imageRouter) initRoutes() {
    r.routes = []router.Route{
        // GET
        router.NewGetRoute("/images/json", r.getImagesJSON),
        router.NewGetRoute("/images/search", r.getImagesSearch),
        router.NewGetRoute("/images/get", r.getImagesGet),
        router.NewGetRoute("/images/{name:.*}/get", r.getImagesGet),
        router.NewGetRoute("/images/{name:.*}/history", r.getImagesHistory),
        router.NewGetRoute("/images/{name:.*}/json", r.getImagesByName),
        // POST
        router.NewPostRoute("/commit", r.postCommit),
        router.NewPostRoute("/images/load", r.postImagesLoad),
        router.Cancellable(router.NewPostRoute("/images/create", r.postImagesCreate)),
        router.Cancellable(router.NewPostRoute("/images/{name:.*}/push", r.postImagesPush)),
        router.NewPostRoute("/images/{name:.*}/tag", r.postImagesTag),
        router.NewPostRoute("/images/prune", r.postImagesPrune),
        // DELETE
        router.NewDeleteRoute("/images/{name:.*}", r.deleteImages),
    }
}
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
27
28
29
30
31
32
33
34
35
36

可以看到函数调用过程：NewRouter->initRoutes，且我们上面提到的pull命令的API：/images/create赫然在列。这里已经很明了了，pull命令在服务端将由r.postImagesCreate处理，实现在docker\api\server\router\image\image_routes.go，我们分析下该函数：

// Creates an image from Pull or from Import
func (s *imageRouter) postImagesCreate(ctx context.Context, w http.ResponseWriter, r *http.Request, vars map[string]string) error {
    if err := httputils.ParseForm(r); err != nil {
        return err
    }
    // Calling POST /v1.25/images/create?fromImage=gplang&tag=latest
    var (
        image   = r.Form.Get("fromImage")
        repo    = r.Form.Get("repo")
        tag     = r.Form.Get("tag")
        message = r.Form.Get("message")
        err     error
        output  = ioutils.NewWriteFlusher(w)
    )
    defer output.Close()
    //设置回应的http头，说明数据是json
    w.Header().Set("Content-Type", "application/json")
      //镜像名不存在
    if image != "" { //pull
        metaHeaders := map[string][]string{}
        for k, v := range r.Header {
            if strings.HasPrefix(k, "X-Meta-") {
                metaHeaders[k] = v
            }
        }

        authEncoded := r.Header.Get("X-Registry-Auth")
        authConfig := &types.AuthConfig{}
        if authEncoded != "" {
            authJSON := base64.NewDecoder(base64.URLEncoding, strings.NewReader(authEncoded))
            if err := json.NewDecoder(authJSON).Decode(authConfig); err != nil {
                // for a pull it is not an error if no auth was given
                // to increase compatibility with the existing api it is defaulting to be empty
                authConfig = &types.AuthConfig{}
            }
        }
        // Backend is all the methods that need to be implemented
        // to provide image specific functionality（功能）.
        //在Daemon类型实现了该API接口，在docker/daemon/image_pull.go
        err = s.backend.PullImage(ctx, image, tag, metaHeaders, authConfig, output)


    } else { //import
        src := r.Form.Get("fromSrc")
        // 'err' MUST NOT be defined within this block, we need any error
        // generated from the download to be available to the output
        // stream processing below
        err = s.backend.ImportImage(src, repo, tag, message, r.Body, output, r.Form["changes"])
    }
    if err != nil {
        if !output.Flushed() {
            return err
        }
        sf := streamformatter.NewJSONStreamFormatter()
        output.Write(sf.FormatError(err))
    }

    return nil
}
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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

—————————————2016.12.09 22:03 更新—————————————————

可以看到主要是从http参数中解析出镜像名和tag，分了有镜像名和无镜像名两个分支。我们拉取镜像时我们传入了ubuntu这个镜像名，所以走if分支（image 为空的情况不知是什么情况，暂时不去深究）。从上面的代码中我们可以看到以镜像名，tag以及授权信息等参数调用函数s.backend.PullImage。可是backend这个是什么呢？backend是接口Backend的实例，我们要找其实现类。

type Backend interface {
    containerBackend
    imageBackend
    importExportBackend
    registryBackend
}
1
2
3
4
5
6

我们回到镜像相关的API初始化的代码：

// NewRouter initializes a new image router
func NewRouter(backend Backend, decoder httputils.ContainerDecoder) router.Router {
    r := &imageRouter{
        backend: backend,
        decoder: decoder,
    }
    r.initRoutes()
    return r
}
1
2
3
4
5
6
7
8
9

可以看到是NewRouter的时候传入的，我们看下调用代码，在docker\cmd\dockerd\daemon.go的 initRouter(s *apiserver.Server, d *daemon.Daemon, c *cluster.Cluster) 函数，有：

image.NewRouter(d, decoder),
1

我们再往上看initRouter的调用代码，在文件docker\cmd\dockerd\daemon.go的star函数：

    initRouter(api, d, c)
1

原来是这里的d，再看下d是如何来的：

    d, err := daemon.NewDaemon(cli.Config, registryService, containerdRemote)
1

返回的是一个Daemon对象指针。这下我们我们可以知道s.backend.PullImage实际上调用的是Daemon的成员PullImage函数。实际上Daemon不仅实现了image相关的接口，而是实现了所有docker的操作的接口。往后我们分析的接口都可以在那里找到实现。我现在去看下PullImage函数的实现，在文件docker\daemon\image_pull.go：

// PullImage initiates a pull operation. image is the repository name to pull, and
// tag may be either empty, or indicate a specific tag to pull.
func (daemon *Daemon) PullImage(ctx context.Context, image, tag string, metaHeaders map[string][]string, authConfig *types.AuthConfig, outStream io.Writer) error {
    // Special case: "pull -a" may send an image name with a
    // trailing :. This is ugly, but let's not break API
    // compatibility.
    image = strings.TrimSuffix(image, ":")
    //fromImage=gplang&tag=latest
    //name格式: xxx:yyy | @zzz  xxx 代表镜像名,如果没有加上仓库地址:docker.io，会使用默认的仓库地址， yyy ：代表版本 zzz： 代表摘要
    ref, err := reference.ParseNamed(image)
    if err != nil {
        return err
    }
    //如果tag不为空，则要看标签还是摘要，或者什么也不是
    if tag != "" {
        // The "tag" could actually be a digest.
        var dgst digest.Digest
        dgst, err = digest.ParseDigest(tag)
        if err == nil {
            ref, err = reference.WithDigest(ref, dgst)
        } else {
            ref, err = reference.WithTag(ref, tag)
        }
        if err != nil {
            return err
        }
    }

    return daemon.pullImageWithReference(ctx, ref, metaHeaders, authConfig, outStream)
}
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
27
28
29
30

是不是看到熟悉的东西，对这里又将镜像名等解析了一遍，如果我们传入的是tag就得到一个reference.NamedTagged对象ref。然后交给pullImageWithReference：

func (daemon *Daemon) pullImageWithReference(ctx context.Context, ref reference.Named, metaHeaders map[string][]string, authConfig *types.AuthConfig, outStream io.Writer) error {
    // Include a buffer so that slow client connections don't affect
    // transfer performance.
    progressChan := make(chan progress.Progress, 100)

    writesDone := make(chan struct{})

    ctx, cancelFunc := context.WithCancel(ctx)

    go func() {
        writeDistributionProgress(cancelFunc, outStream, progressChan)
        close(writesDone)
    }()
        //注意这里有很多重要的接口
    imagePullConfig := &distribution.ImagePullConfig{
        MetaHeaders:      metaHeaders,
        AuthConfig:       authConfig,
        ProgressOutput:   progress.ChanOutput(progressChan),
        RegistryService:  daemon.RegistryService,//默认regist服务接口实现的实例
        ImageEventLogger: daemon.LogImageEvent,
        MetadataStore:    daemon.distributionMetadataStore,
        ImageStore:       daemon.imageStore,
        ReferenceStore:   daemon.referenceStore,
        DownloadManager:  daemon.downloadManager,
    }

    err := distribution.Pull(ctx, ref, imagePullConfig)
    close(progressChan)
    <-writesDone
    return err
}
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
27
28
29
30
31

这里再调用distribution.Pull，还有就是要注意这里构造了一个imagePullConfig 对象，里面包含了很多拉取镜像时要用到的接口（我们暂且先记下，后面分析到的时候再回过头来看）。如此绕来绕去，想必是有点晕头转向了。在继续之前我们先说下docker的代码风格，如果了解了docker的代码风格，我想我们就知道docker解决问题的套路，这样即使我们没有完全掌握docker源码，我们也可以根据我们看过的docker源码推测出其他逻辑。我们先就以即将要分析的distribution.Pull中的Service为例。

可以看到在文件docker\registry\service.goService中定义了Service接口，接口中有一些镜像仓库相关的方法。接着在接口定义的文件中定义了Service接口的默认实现。他们是怎么关联在一起的呢（不是指go语法上的关联）。一般在这个文件中为定义NewXXX的方法，该方法返回的就是了接口实现对象的指针：

// NewService returns a new instance of DefaultService ready to be
// installed into an engine.
func NewService(options ServiceOptions) *DefaultService {
    return &DefaultService{
        config: newServiceConfig(options),
    }
}
1
2
3
4
5
6
7

明白了这个套路，我们接着分析distribution.Pull：

/ Pull initiates a pull operation. image is the repository name to pull, and
// tag may be either empty, or indicate a specific tag to pull.
func Pull(ctx context.Context, ref reference.Named, imagePullConfig *ImagePullConfig) error {
    // Resolve the Repository name from fqn to RepositoryInfo

    //在/docker/registry/config.go的 newServiceConfig初始化仓库地址和仓库镜像地址，其中有官方的和通过选项insecure-registry自定义的私有仓库,实质是通过IndexName找到IndexInfo，有用的也只有IndexName
    //这里的imagePullConfig.RegistryService为daemon.RegistryService，也即是docker\registry\service.go的DefaultService
    //初始化时,会将insecure-registry选项和registry-mirrors存入ServiceOptions,在NewService函数被调用时,作为参入传入

    //repoInfo为RepositoryInfo对象,其实是对reference.Named对象的封装,添加了镜像成员和官方标示
    repoInfo, err := imagePullConfig.RegistryService.ResolveRepository(ref)
    if err != nil {
        return err
    }

    // makes sure name is not empty or `scratch`
    //为了确保不为空或?
    if err := ValidateRepoName(repoInfo.Name()); err != nil {
        return err
    }
    // APIEndpoint represents a remote API endpoint
    // /docker/cmd/dockerddaemon.go----大约125 和 248
    //如果没有镜像仓库服务器地址，默认使用V2仓库地址registry-1.docker.io
    //Hostname()函数来源于Named

    //实质上如果Hostname()返回的是官方仓库地址,则endpoint的URL将是registry-1.docker.io,如果有镜像则会添加镜像作为endpoint
    // 否则就是私有地址的两种类型:http和https

    //V2的接口具体代码在Zdocker\registry\service_v2.go的函数lookupV2Endpoints
    //
    logrus.Debugf("Get endpoint from:%s", repoInfo.Hostname())
    endpoints, err := imagePullConfig.RegistryService.LookupPullEndpoints(repoInfo.Hostname())
    if err != nil {
        return err
    }

    var (
        lastErr error

        // discardNoSupportErrors is used to track whether an endpoint encountered an error of type registry.ErrNoSupport
        // By default it is false, which means that if an ErrNoSupport error is encountered, it will be saved in lastErr.
        // As soon as another kind of error is encountered, discardNoSupportErrors is set to true, avoiding the saving of
        // any subsequent ErrNoSupport errors in lastErr.
        // It's needed for pull-by-digest on v1 endpoints: if there are only v1 endpoints configured, the error should be
        // returned and displayed, but if there was a v2 endpoint which supports pull-by-digest, then the last relevant
        // error is the ones from v2 endpoints not v1.
        discardNoSupportErrors bool

        // confirmedV2 is set to true if a pull attempt managed to
        // confirm that it was talking to a v2 registry. This will
        // prevent fallback to the v1 protocol.
        confirmedV2 bool

        // confirmedTLSRegistries is a map indicating which registries
        // are known to be using TLS. There should never be a plaintext
        // retry for any of these.
        confirmedTLSRegistries = make(map[string]struct{})
    )
    //如果设置了镜像服务器地址,且使用了官方默认的镜像仓库,则endpoints包含官方仓库地址和镜像服务器地址,否则就是私有仓库地址的http和https形式
    for _, endpoint := range endpoints {
        logrus.Debugf("Endpoint API version:%d", endpoint.Version)
        if confirmedV2 && endpoint.Version == registry.APIVersion1 {
            logrus.Debugf("Skipping v1 endpoint %s because v2 registry was detected", endpoint.URL)
            continue
        }

        if endpoint.URL.Scheme != "https" {
            if _, confirmedTLS := confirmedTLSRegistries[endpoint.URL.Host]; confirmedTLS {
                logrus.Debugf("Skipping non-TLS endpoint %s for host/port that appears to use TLS", endpoint.URL)
                continue
            }
        }

        logrus.Debugf("Trying to pull %s from %s %s", repoInfo.Name(), endpoint.URL, endpoint.Version)
        //针对每一个endpoint，建立一个Puller,newPuller会根据endpoint的形式（endpoint应该遵循restful api的设计，url中含有版本号），决定采用version1还是version2版本
        //imagePullConfig是个很重要的对象,包含了很多镜像操作相关的对象
        puller, err := newPuller(endpoint, repoInfo, imagePullConfig)
        if err != nil {
            lastErr = err
            continue
        }
        if err := puller.Pull(ctx, ref); err != nil {
            // Was this pull cancelled? If so, don't try to fall
            // back.
            fallback := false
            select {
            case <-ctx.Done():
            default:
                if fallbackErr, ok := err.(fallbackError); ok {
                    fallback = true
                    confirmedV2 = confirmedV2 || fallbackErr.confirmedV2
                    if fallbackErr.transportOK && endpoint.URL.Scheme == "https" {
                        confirmedTLSRegistries[endpoint.URL.Host] = struct{}{}
                    }
                    err = fallbackErr.err
                }
            }
            if fallback {
                if _, ok := err.(ErrNoSupport); !ok {
                    // Because we found an error that's not ErrNoSupport, discard all subsequent ErrNoSupport errors.
                    discardNoSupportErrors = true
                    // append subsequent errors
                    lastErr = err
                } else if !discardNoSupportErrors {
                    // Save the ErrNoSupport error, because it's either the first error or all encountered errors
                    // were also ErrNoSupport errors.
                    // append subsequent errors
                    lastErr = err
                }
                logrus.Errorf("Attempting next endpoint for pull after error: %v", err)
                continue
            }
            logrus.Errorf("Not continuing with pull after error: %v", err)
            return err
        }

        imagePullConfig.ImageEventLogger(ref.String(), repoInfo.Name(), "pull")
        return nil
    }

    if lastErr == nil {
        lastErr = fmt.Errorf("no endpoints found for %s", ref.String())
    }

    return lastErr
}
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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

代码比较多，总结起来就是镜像仓库信息repoInfo–>端点信息endpoints–>puller拉取镜像。这是应该有很多疑问，镜像仓库信息是个什么东西？端点信息是什么？如何拉取？我们逐个分析。首先我们看下镜像仓库信息的定义以及例子(在docker\api\types\registry\registry.go)：

type RepositoryInfo struct {
    reference.Named
    // Index points to registry information
    Index *registrytypes.IndexInfo
    // Official indicates whether the repository is considered official.
    // If the registry is official, and the normalized name does not
    // contain a '/' (e.g. "foo"), then it is considered an official repo.
    //表示是否官方的地址,实际上只要拉取镜像时只传入镜像的信息
    //而没有仓库的信息,就会使用官方默认的仓库地址,这时Official成员就是true
    Official bool
}

// RepositoryInfo Examples:
// {
//   "Index" : {
//     "Name" : "docker.io",
//     "Mirrors" : ["https://registry-2.docker.io/v1/", "https://registry-3.docker.io/v1/"],
//     "Secure" : true,
//     "Official" : true,
//   },
//   "RemoteName" : "library/debian",
//   "LocalName" : "debian",
//   "CanonicalName" : "docker.io/debian"
//   "Official" : true,
// }
//
// {
//   "Index" : {
//     "Name" : "127.0.0.1:5000",
//     "Mirrors" : [],
//     "Secure" : false,
//     "Official" : false,
//   },
//   "RemoteName" : "user/repo",
//   "LocalName" : "127.0.0.1:5000/user/repo",
//   "CanonicalName" : "127.0.0.1:5000/user/repo",
//   "Official" : false,
// }
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
27
28
29
30
31
32
33
34
35
36
37
38

结合代码中的注释,我想我们可以知道RepositoryInfo其实是就是包含了所有可用仓库地址(仓库镜像地址也算)的结构.
———————2016.12.10 22:31更新————————————-
好了,现在我们看下这个结构式如何被填充的.RegistryService实际上是DefaultService.看下imagePullConfig.RegistryService.ResolveRepository(ref),实现在docker\registry\service.go:

func (s *DefaultService) ResolveRepository(name reference.Named) (*RepositoryInfo, error) {
    return newRepositoryInfo(s.config, name)
}

// newRepositoryInfo validates and breaks down a repository name into a RepositoryInfo
func newRepositoryInfo(config *serviceConfig, name reference.Named) (*RepositoryInfo, error) {
    // newIndexInfo returns IndexInfo configuration from indexName
    index, err := newIndexInfo(config, name.Hostname())
    if err != nil {
        return nil, err
    }
    official := !strings.ContainsRune(name.Name(), '/')
    return &RepositoryInfo{name, index, official}, nil
}

// newIndexInfo returns IndexInfo configuration from indexName
func newIndexInfo(config *serviceConfig, indexName string) (*registrytypes.IndexInfo, error) {
    var err error
    indexName, err = ValidateIndexName(indexName)
    if err != nil {
        return nil, err
    }

    // Return any configured index info, first.
    //config是在上面NewService函数中通过传入的ServiceOptions选项生成的
    //serviceConfig，在docker\registry\config.go的InstallCliFlags被初始化
    //index其实就是镜像的仓库地址,或仓库的镜像地址
        //
    if index, ok := config.IndexConfigs[indexName]; ok {
        return index, nil
    }

    // Construct a non-configured index info.
    index := &registrytypes.IndexInfo{
        Name:     indexName,
        Mirrors:  make([]string, 0),
        Official: false,
    }
    index.Secure = isSecureIndex(config, indexName)
    return index, nil
}
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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

三个成员,Name就是根据参数(ubuntu:latest)解析出来的Named对象,Official 如果我们只传入类似ubuntu:latest则使用官方默认的,该成员就是true,就剩下Index了.可以看到Index来源于config.IndexConfigs.那config.IndexConfigs是什么呢?容易发现config.IndexConfigs来源于DefaultService的config。DefaultService的config则来源于NewService时的ServiceOptions。先看下ServiceOptions，实现在docker\registry\config.go：

// ServiceOptions holds command line options.
type ServiceOptions struct {
    Mirrors            []string `json:"registry-mirrors,omitempty"`
    InsecureRegistries []string `json:"insecure-registries,omitempty"`

    // V2Only controls access to legacy registries.  If it is set to true via the
    // command line flag the daemon will not attempt to contact v1 legacy registries
    V2Only bool `json:"disable-legacy-registry,omitempty"`
}
1
2
3
4
5
6
7
8
9