资讯专栏INFORMATION COLUMN

Kubelet源码分析(二): DockerClient

李世赞 / 2913人阅读

摘要:接口分析源码目录实现的接口如下可以看到结构体实现了所有的接口。这些接口其实是对的操作接口进行了封装,下面取一个接口进行分析该接口的关键就是调用了所以关键对象还是,继续回到上面讲初始化时介绍到的结构体。

源码版本

kubernetes version: v1.3.0

DockerClient初始化

DockerClient是KubeletConfig的成员之一。
KubeletConfig结构介绍:

type KubeletConfig struct {
    Address                        net.IP
    AllowPrivileged                bool
...
    DockerClient                   dockertools.DockerInterface
    RuntimeCgroups                 string
    DockerExecHandler              dockertools.ExecHandler
...
}

而kubeletConfig的初始化是在UnsecuredKubeletConfig()接口中进行的,需要依赖最开始组建的kubeletServer配置结构,该kubeletServer结构中有DockerEndpoint字符串成员:

type KubeletServer struct {
    componentconfig.KubeletConfiguration

    AuthPath      util.StringFlag // Deprecated -- use KubeConfig instead
    KubeConfig    util.StringFlag
    APIServerList []string

    RunOnce bool

    // Insert a probability of random errors during calls to the master.
    ChaosChance float64
    // Crash immediately, rather than eating panics.
    ReallyCrashForTesting bool
    SystemReserved        config.ConfigurationMap
    KubeReserved          config.ConfigurationMap
}

type KubeletConfiguration struct {
    // config is the path to the config file or directory of files
    Config string `json:"config"`
...
    DockerEndpoint string `json:"dockerEndpoint"`
...    

实际上如果没有指定该参数的话,会默认使用端点"unix:///var/run/docker.sock"做为DockerEndpoint。可以查看NewEnvClient()接口。

回到kubeletConfig的初始化接口UnsecuredKubeletConfig():

func UnsecuredKubeletConfig(s *options.KubeletServer) (*KubeletConfig, error) {
    hostNetworkSources, err := kubetypes.GetValidatedSources(strings.Split(s.HostNetworkSources, ","))
    if err != nil {
        return nil, err
    }
...
    return &KubeletConfig{
        Address:                      net.ParseIP(s.Address),
        AllowPrivileged:              s.AllowPrivileged,
        ...
        DockerClient:                 dockertools.ConnectToDockerOrDie(s.DockerEndpoint, s.RuntimeRequestTimeout.Duration), // TODO(random-liu): Set RuntimeRequestTimeout for rkt.
...
    }

接着继续查看dockertools.ConnectToDockerOrDie(s.DockerEndpoint, s.RuntimeRequestTimeout.Duration)。

func ConnectToDockerOrDie(dockerEndpoint string, requestTimeout time.Duration) DockerInterface {
    if dockerEndpoint == "fake://" {
        return NewFakeDockerClient()
    }
    client, err := getDockerClient(dockerEndpoint)
    if err != nil {
        glog.Fatalf("Couldn"t connect to docker: %v", err)
    }
    glog.Infof("Start docker client with request timeout=%v", requestTimeout)
    return newKubeDockerClient(client, requestTimeout)
}

先前我们了解了如果在kubelet启动时没有传入"docker-endpoint"参数的话,s.DockerEndpoint即为空。
s.RuntimeRequestTimeout.Duration值可以查看NewKubeletServer()函数的初始化,是2min。
getDockerClient()接口比较简单:
getDockerClient --> dockerapi.NewEnvClient() --> NewClient().
NewClient()接口如下:

func NewClient(host string, version string, client *http.Client, httpHeaders map[string]string) (*Client, error) {
    proto, addr, basePath, err := ParseHost(host)
    if err != nil {
        return nil, err
    }

    transport, err := transport.NewTransportWithHTTP(proto, addr, client)
    if err != nil {
        return nil, err
    }

    return &Client{
        proto:             proto,
        addr:              addr,
        basePath:          basePath,
        transport:         transport,
        version:           version,
        customHTTPHeaders: httpHeaders,
    }, nil
}

之前讲了如果没有传入"docker-endpoint"参数的话,默认值就是"unix:///var/run/docker.sock".即host参数为该值。
ParseHost()先根据host进行解析,然后创建transport-->Client。
Client结构如下:

type Client struct {
    // proto holds the client protocol i.e. unix.
    proto string
    // addr holds the client address.
    addr string
    // basePath holds the path to prepend to the requests.
    basePath string
    // transport is the interface to send request with, it implements transport.Client.
    transport transport.Client
    // version of the server to talk to.
    version string
    // custom http headers configured by users.
    customHTTPHeaders map[string]string
}

创建Client成功之后,最终开始提到的ConnectToDockerOrDie()接口会调用newKubeDockerClient()生成pkg/kubelet/dockertools/kube_docker_client.go里的kubeDockerClient结构:

type kubeDockerClient struct {
    // timeout is the timeout of short running docker operations.
    timeout time.Duration
    client  *dockerapi.Client
}

初始化到这里就结束了,那我们回到最初,介绍下DockerClient定义:
dockertools.DockerInterface如下:

type DockerInterface interface {
    ListContainers(options dockertypes.ContainerListOptions) ([]dockertypes.Container, error)
    InspectContainer(id string) (*dockertypes.ContainerJSON, error)
    CreateContainer(dockertypes.ContainerCreateConfig) (*dockertypes.ContainerCreateResponse, error)
    StartContainer(id string) error
    StopContainer(id string, timeout int) error
    RemoveContainer(id string, opts dockertypes.ContainerRemoveOptions) error
    InspectImage(image string) (*dockertypes.ImageInspect, error)
    ListImages(opts dockertypes.ImageListOptions) ([]dockertypes.Image, error)
    PullImage(image string, auth dockertypes.AuthConfig, opts dockertypes.ImagePullOptions) error
    RemoveImage(image string, opts dockertypes.ImageRemoveOptions) ([]dockertypes.ImageDelete, error)
    ImageHistory(id string) ([]dockertypes.ImageHistory, error)
    Logs(string, dockertypes.ContainerLogsOptions, StreamOptions) error
    Version() (*dockertypes.Version, error)
    Info() (*dockertypes.Info, error)
    CreateExec(string, dockertypes.ExecConfig) (*dockertypes.ContainerExecCreateResponse, error)
    StartExec(string, dockertypes.ExecStartCheck, StreamOptions) error
    InspectExec(id string) (*dockertypes.ContainerExecInspect, error)
    AttachToContainer(string, dockertypes.ContainerAttachOptions, StreamOptions) error
}

而我们最终初始化返回了结构体kubeDockerClient,所以DockerInterface接口的实现,我们可以回到kubeDockerClient结构体所在文件pkg/kubelet/dockertools/kube_docker_client.go查看接口实现。

DockeClient接口分析

源码目录: pkg/kubelet/dockertools/kube_docker_client.go
实现的接口如下:

可以看到kubeDockerClient结构体实现了所有的DockerInterface接口。
这些接口其实是对docker的操作接口进行了封装,下面取一个接口进行分析:

func (d *kubeDockerClient) ListContainers(options dockertypes.ContainerListOptions) ([]dockertypes.Container, error) {
    ctx, cancel := d.getTimeoutContext()
    defer cancel()
    containers, err := d.client.ContainerList(ctx, options)
    if ctxErr := contextError(ctx); ctxErr != nil {
        return nil, ctxErr
    }
    if err != nil {
        return nil, err
    }
    return containers, nil
}

该ListContainers()接口的关键就是调用了d.client.ContainerList(ctx, options).
所以关键对象还是client,继续回到上面讲初始化时介绍到的Client结构体。
Client结构所在文件: vendor/github.com/docker/engine-api/client/client.go
Client package结构:

操作docker API的接口都封装在这些文件中,有空可以深入了解下,这里就不一一介绍了,我们继续回到d.client.ContainerList(ctx, options),实现如下:

func (cli *Client) ContainerList(ctx context.Context, options types.ContainerListOptions) ([]types.Container, error) {
    query := url.Values{}

    if options.All {
        query.Set("all", "1")
    }

    if options.Limit != -1 {
        query.Set("limit", strconv.Itoa(options.Limit))
    }

    if options.Since != "" {
        query.Set("since", options.Since)
    }

    if options.Before != "" {
        query.Set("before", options.Before)
    }

    if options.Size {
        query.Set("size", "1")
    }

    if options.Filter.Len() > 0 {
        filterJSON, err := filters.ToParamWithVersion(cli.version, options.Filter)

        if err != nil {
            return nil, err
        }

        query.Set("filters", filterJSON)
    }

    resp, err := cli.get(ctx, "/containers/json", query, nil)
    if err != nil {
        return nil, err
    }

    var containers []types.Container
    err = json.NewDecoder(resp.body).Decode(&containers)
    ensureReaderClosed(resp)
    return containers, err
}

前面都是一些参数初始化,其实就是构建一个GET请求,然后调用cli.get(),该get就是一个httpRequest:

func (cli *Client) get(ctx context.Context, path string, query url.Values, headers map[string][]string) (*serverResponse, error) {
    return cli.sendRequest(ctx, "GET", path, query, nil, headers)
}

func (cli *Client) sendRequest(ctx context.Context, method, path string, query url.Values, obj interface{}, headers map[string][]string) (*serverResponse, error) {
    var body io.Reader

    if obj != nil {
        var err error
        body, err = encodeData(obj)
        if err != nil {
            return nil, err
        }
        if headers == nil {
            headers = make(map[string][]string)
        }
        headers["Content-Type"] = []string{"application/json"}
    }

    return cli.sendClientRequest(ctx, method, path, query, body, headers)
}

func (cli *Client) sendClientRequest(ctx context.Context, method, path string, query url.Values, body io.Reader, headers map[string][]string) (*serverResponse, error) {
    serverResp := &serverResponse{
        body:       nil,
        statusCode: -1,
    }

...
    req, err := cli.newRequest(method, path, query, body, headers)
    if cli.proto == "unix" || cli.proto == "npipe" {
        // For local communications, it doesn"t matter what the host is. We just
        // need a valid and meaningful host name. (See #189)
        req.Host = "docker"
    }
    req.URL.Host = cli.addr
    req.URL.Scheme = cli.transport.Scheme()

    if expectedPayload && req.Header.Get("Content-Type") == "" {
        req.Header.Set("Content-Type", "text/plain")
    }

    resp, err := cancellable.Do(ctx, cli.transport, req)
    if resp != nil {
        serverResp.statusCode = resp.StatusCode
    }

...

    if serverResp.statusCode < 200 || serverResp.statusCode >= 400 {
        body, err := ioutil.ReadAll(resp.Body)
        if err != nil {
            return serverResp, err
        }
        if len(body) == 0 {
            return serverResp, fmt.Errorf("Error: request returned %s for API route and version %s, check if the server supports the requested API version", http.StatusText(serverResp.statusCode), req.URL)
        }
        return serverResp, fmt.Errorf("Error response from daemon: %s", bytes.TrimSpace(body))
    }

    serverResp.body = resp.Body
    serverResp.header = resp.Header
    return serverResp, nil
}

func Do(ctx context.Context, client transport.Sender, req *http.Request) (*http.Response, error) {
...
    result := make(chan responseAndError, 1)

    go func() {
        resp, err := client.Do(req)
        testHookDoReturned()
        result <- responseAndError{resp, err}
    }()

    var resp *http.Response

    select {
    case <-ctx.Done():
        testHookContextDoneBeforeHeaders()
        cancel()
        // Clean up after the goroutine calling client.Do:
        go func() {
            if r := <-result; r.resp != nil && r.resp.Body != nil {
                testHookDidBodyClose()
                r.resp.Body.Close()
            }
        }()
        return nil, ctx.Err()
    case r := <-result:
        var err error
        resp, err = r.resp, r.err
        if err != nil {
            return resp, err
        }
    }

...

    return resp, nil
}

上面列出了httpRequest的整个调用过程,最终调用client.Do(),该client对象需要回到之前的初始化过程中去,实际就是调用vemdor/github.com/docker/engine-api/client/client.go中的Client.transport,而该对象初始化时设置为apiTransport对象:

type apiTransport struct {
    *http.Client
    *tlsInfo
    transport *http.Transport
}

所以client.Do()实际就是调用http.Client.Do()。
OK,到此算是分析结束,具体的各个接口实现,还是需要花时间查看源码,但也都是大同小异。

学习源码的过程中,可以看到很多经典的实现,比如上面介绍的cancellable.Do()接口实现,golang非常推崇的"协程+channel"的方式,通过select case的方式循环等待协程处理的结果,确实很方便。

文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。

转载请注明本文地址:https://www.ucloud.cn/yun/32537.html

相关文章

  • Kubelet源码分析(一):启动流程分析

    摘要:源码版本简介在急群众,在每个节点上都会启动一个服务进程。该进程用于处理节点下发到本节点的任务,管理及中的容器。每个进程会在上注册节点自身信息,定期向节点汇报节点资源的使用情况,并通过监控容器和节点资源。最后运行健康检测服务。 源码版本 kubernetes version: v1.3.0 简介 在Kubernetes急群众,在每个Node节点上都会启动一个kubelet服务进程。该进程...

    mindwind 评论0 收藏0
  • Kubelet源码分析(四) diskSpaceManager

    摘要:顾名思义就是管理磁盘空间的,实际它的实现较为简单,就是给所在的节点预留磁盘空间的,当该节点磁盘空间低于该值时,将拒绝的创建。其实就是中的接口该接口很简单,就是分别调用实现的两个接口,然后判断磁盘空间是否够用。 源码版本 kubernetes version: v1.3.0 简介 前一节介绍了Garbage Collection,涉及到的策略基本与磁盘资源有关。对于k8s集群如何高效的利...

    hlcfan 评论0 收藏0
  • Kubelet源码分析(三):Garbage Collection

    摘要:源码版本介绍在分析启动流程时,老是会碰到各类,这里单独提出来做下较详细的分析。主要由两部分组成使用指定的回收策略,删除那些已经结束的所有的生命周期管理就是通过来实现的,其实该也是依赖了。相关配置该值表示磁盘占用率达到该值后会触发。 源码版本 kubernetes version: v1.3.0 kubelet GC介绍 在分析kubelet启动流程时,老是会碰到各类GC,这里单独提出来...

    siberiawolf 评论0 收藏0
  • Kubelet无法访问rancher-metadata问题分析

    摘要:引言能够支持,可以快速几乎无障碍的拉起一套环境,这对刚入门的小白来说简直是一大利器。本文就分析一下关于无法访问问题。检查一下有问题的节点的系统,果然会发现安装了服务服务名为。 引言 Rancher能够支持Kubernetes,可以快速几乎无障碍的拉起一套K8s环境,这对刚入门K8s的小白来说简直是一大利器。当然由于系统特性五花八门,系统内置软件也相互影响,所以有时候伙伴们会碰到比较难缠...

    hiYoHoo 评论0 收藏0
  • Kubelet无法访问rancher-metadata问题分析

    摘要:引言能够支持,可以快速几乎无障碍的拉起一套环境,这对刚入门的小白来说简直是一大利器。本文就分析一下关于无法访问问题。检查一下有问题的节点的系统,果然会发现安装了服务服务名为。 引言 Rancher能够支持Kubernetes,可以快速几乎无障碍的拉起一套K8s环境,这对刚入门K8s的小白来说简直是一大利器。当然由于系统特性五花八门,系统内置软件也相互影响,所以有时候伙伴们会碰到比较难缠...

    bigdevil_s 评论0 收藏0

发表评论

0条评论

李世赞

|高级讲师

TA的文章

阅读更多
最新活动
阅读需要支付1元查看
<