heapster源码分析——kubelet的api调用分析

Posted 2020-10-14 SonoFelice——棣琦

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了heapster源码分析——kubelet的api调用分析相关的知识，希望对你有一定的参考价值。

一、heapster简介

什么是Heapster？

Heapster是容器集群监控和性能分析工具，天然的支持Kubernetes和CoreOS。
Kubernetes有个出名的监控agent---cAdvisor。在每个kubernetes Node上都会运行cAdvisor，它会收集本机以及容器的监控数据(cpu,memory,filesystem,network,uptime)。
在较新的版本中，K8S已经将cAdvisor功能集成到kubelet组件中。每个Node节点可以直接进行web访问。

cAdvisor web界面访问： http://< Node-IP >:4194
cAdvisor也提供Restful API: https://github.com/google/cadv ... pi.md

Heapster是一个收集者，将每个Node上的cAdvisor的数据进行汇总，然后导到第三方工具(如InfluxDB)。

二、heapster调用kubelet源码分析

1、整体源码

heapster整体的源码分析可以参考文章：https://segmentfault.com/a/1190000008863353#articleHeader2，

上面链接中的文章中会对heapster从启动开始进行分析，主要讲述了下面内容：

main（）
创建数据源
创建后端服务
创建数据processors
获取源数据并存储
heapster API创建

给上述文章的作者点个赞，思路真的非常清晰。不过笔者最关注的点是heapster如何进行的kubelet调用。

2、调用kubelet源码分析

从上述文章提到的 NewKubeletProvider 函数中，我们找到了创建kubeClient和kubeletClient的地方，见下面代码的飘黄部分：

func NewKubeletProvider(uri *url.URL) (MetricsSourceProvider, error) {
    // 创建kubernetes master及kubelet client相关的配置
    kubeConfig, kubeletConfig, err := GetKubeConfigs(uri)
    if err != nil {
        return nil, err
    }
    // 创建kubeClient及kubeletClient
    kubeClient := kube_client.NewOrDie(kubeConfig)
    kubeletClient, err := NewKubeletClient(kubeletConfig)
    if err != nil {
        return nil, err
    }

    // 获取下所有的Nodes，测试下创建的client是否能正常通讯
    if _, err := kubeClient.Nodes().List(kube_api.ListOptions{
        LabelSelector: labels.Everything(),
        FieldSelector: fields.Everything()}); err != nil {
        glog.Errorf("Failed to load nodes: %v", err)
    }

    // 监控k8s的nodes变更
    // 这里会创建协程进行watch，便于后面调用nodeLister.List()列出所有的nodes。
    // 该Watch的实现，需要看下apiServer中的实现，后面会进行讲解
    lw := cache.NewListWatchFromClient(kubeClient, "nodes", kube_api.NamespaceAll, fields.Everything())
    nodeLister := &cache.StoreToNodeLister{Store: cache.NewStore(cache.MetaNamespaceKeyFunc)}
    reflector := cache.NewReflector(lw, &kube_api.Node{}, nodeLister.Store, time.Hour)
    reflector.Run()
    // 结构在前面介绍过
    return &kubeletProvider{
        nodeLister:    nodeLister,
        reflector:     reflector,
        kubeletClient: kubeletClient,
    }, nil
}

1）heapster源码目录

我们发现这个函数所在的文件为kubelet.go，所在的包为metrics/sources/kubelet

我们可以看到，kubelet调用相关的代码都在这个里面：

2）heapster调用kubelet的API

我们进入到NewKubeletClient函数看一下：

// 传入的是KubeletClientConfig，通过读取相关配置，来初始化client

func NewKubeletClient(kubeletConfig *kubelet_client.KubeletClientConfig) (*KubeletClient, error) {

    transport, err := kubelet_client.MakeTransport(kubeletConfig)
    if err != nil {
        return nil, err
    }
    c := &http.Client{ // 此处可以看到，是http调用
        Transport: transport,
        Timeout:   kubeletConfig.HTTPTimeout,
    }
    return &KubeletClient{
        config: kubeletConfig,
        client: c,
    }, nil
}

接下来我们看一下kubelet_client.go中KubeletClient有哪些方法：

(i)获取所有的containersInfo

第一个，获取所有的container统计信息，如果对cAdvisor比较了解的话，可以进入到该方法的返回值的ContainerInfo里面看下，是非常全面的容器监控信息（后续会有cAdvisor的源码分析，敬请期待）

func (self *KubeletClient) getAllContainers(url string, start, end time.Time) ([]cadvisor.ContainerInfo, error) {
    // Request data from all subcontainers.此处是构造requestBody
    request := statsRequest{
        ContainerName: "/",
        NumStats:      1,
        Start:         start, // 2017-11-10T06:46:17Z 这种utc格式的时间戳
        End:           end,
        Subcontainers: true,
    }
    body, err := json.Marshal(request)
    if err != nil {
        return nil, err
    }
    req, err := http.NewRequest("POST", url, bytes.NewBuffer(body))
    if err != nil {
        return nil, err
    }
// 设置请求头
    req.Header.Set("Content-Type", "application/json")

    var containers map[string]cadvisor.ContainerInfo
    client := self.client
    if client == nil {
        client = http.DefaultClient
    }
    err = self.postRequestAndGetValue(client, req, &containers)
    if err != nil {
        return nil, fmt.Errorf("failed to get all container stats from Kubelet URL %q: %v", url, err)
    }

    result := make([]cadvisor.ContainerInfo, 0, len(containers))
    for _, containerInfo := range containers {
        cont := self.parseStat(&containerInfo)
        if cont != nil {
            result = append(result, *cont)
        }
    }
    return result, nil
}

我们可以看到上面的源码其实是构造了一个http的post请求。那么我们在本地通过restClient模拟一下：

http://ip:10255/stats/container/（为什么用10255端口呢？在上面的源码中初始化NewKubeletClient的地方，我们进入到GetKubeClientConfig看一下就会发现，用的是kubelet的默认开放端口，在configs.go文件中）

返回值部分摘录如下（内容太多，折叠一下）：

"/": { "name": "/", "subcontainers": [ { "name": "/docker" }, { "name": "/init.scope" }, { "name": "/kube-proxy" }, { "name": "/kubepods" }, { "name": "/system.slice" }, { "name": "/user.slice" } ], "spec": { "creation_time": "2017-10-14T17:16:59.086488213+08:00", "has_cpu": true, "cpu": { "limit": 1024, "max_limit": 0, "mask": "0-3", "period": 100000 }, "has_memory": true, "memory": { "limit": 33604153344, "reservation": 9223372036854771712 }, "has_network": true, "has_filesystem": true, "has_diskio": true, "has_custom_metrics": false }, "stats": [ { "timestamp": "2017-11-10T14:29:25.105117933+08:00", "cpu": { "usage": { "total": 916886132160774, "per_cpu_usage": [ 225713500943530, 230866928282148, 229896967301966, 230408735633130 ], "user": 443962220000000, "system": 438380440000000 }, "cfs": { "periods": 0, "throttled_periods": 0, "throttled_time": 0 }, "load_average": 0 }, "diskio": { "io_service_bytes": [ { "major": 7, "minor": 4, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 1, "minor": 15, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } }, { "major": 1, "minor": 14, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } }, { "major": 1, "minor": 12, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } }, { "major": 1, "minor": 0, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } }, { "major": 7, "minor": 0, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 1, "minor": 11, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } }, { "major": 1, "minor": 4, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } }, { "major": 253, "minor": 0, "stats": { "Async": 61989669888, "Read": 23649280, "Sync": 17028616192, "Total": 79018286080, "Write": 78994636800 } }, { "major": 7, "minor": 7, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 9, "minor": 0, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 1, "minor": 13, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } }, { "major": 1, "minor": 6, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } }, { "major": 7, "minor": 6, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 7, "minor": 5, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 1, "minor": 5, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } }, { "major": 1, "minor": 3, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } }, { "major": 1, "minor": 7, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } }, { "major": 253, "minor": 16, "stats": { "Async": 258048, "Read": 21504, "Sync": 4773888, "Total": 5031936, "Write": 5010432 } }, { "major": 7, "minor": 3, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 7, "minor": 2, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 7, "minor": 1, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 1, "minor": 10, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } }, { "major": 1, "minor": 9, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } }, { "major": 1, "minor": 8, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } }, { "major": 1, "minor": 2, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } }, { "major": 1, "minor": 1, "stats": { "Async": 12288, "Read": 12288, "Sync": 0, "Total": 12288, "Write": 0 } } ], "io_serviced": [ { "major": 7, "minor": 4, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 1, "minor": 9, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 1, "minor": 8, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 9, "minor": 0, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 1, "minor": 1, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 1, "minor": 15, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 7, "minor": 7, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 7, "minor": 1, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 1, "minor": 4, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 1, "minor": 0, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 253, "minor": 0, "stats": { "Async": 2738092, "Read": 2965, "Sync": 4157380, "Total": 6895472, "Write": 6892507 } }, { "major": 1, "minor": 13, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 7, "minor": 5, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 1, "minor": 14, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 1, "minor": 5, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 253, "minor": 16, "stats": { "Async": 57, "Read": 7, "Sync": 32, "Total": 89, "Write": 82 } }, { "major": 1, "minor": 3, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 1, "minor": 2, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 7, "minor": 6, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 7, "minor": 2, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 1, "minor": 11, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 1, "minor": 10, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 1, "minor": 6, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 1, "minor": 12, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 1, "minor": 7, "stats": { "Async": 3, "Read": 3, "Sync": 0, "Total": 3, "Write": 0 } }, { "major": 7, "minor": 3, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } }, { "major": 7, "minor": 0, "stats": { "Async": 0, "Read": 0, "Sync": 0, "Total": 0, "Write": 0 } } ] }, "memory": { "usage": 12189630464, "cache": 288108544, "rss": 3631652864, "swap": 0, "working_set": 7756312576, "failcnt": 0, "container_data": { "pgfault": 8947531, "pgmajfault": 87 }, "hierarchical_data": { "pgfault": 8947531, "pgmajfault": 87 } }, "network": { "name": "cbr0", "rx_bytes": 1885776219, "rx_packets": 8199780, "rx_errors": 0, "rx_dropped": 0, "tx_bytes": 1854690043, "tx_packets": 8765796, "tx_errors": 0, "tx_dropped": 0, "interfaces": [ { "name": "cbr0", "rx_bytes": 1885776219, "rx_packets": 8199780, "rx_errors": 0, "rx_dropped": 0, "tx_bytes": 1854690043, "tx_packets": 8765796, "tx_errors": 0, "tx_dropped": 0 }, { "name": "ens3", "rx_bytes": 25108832715, "rx_packets": 74352449, "rx_errors": 0, "rx_dropped": 0, "tx_bytes": 93157559917, "tx_packets": 75983196, "tx_errors": 0, "tx_dropped": 0 } ], "tcp": { "Established": 0, "SynSent": 0, "SynRecv": 0, "FinWait1": 0, "FinWait2": 0, "TimeWait": 0, "Close": 0, "CloseWait": 0, "LastAck": 0, "Listen": 0, "Closing": 0 }, "tcp6": { "Established": 0, "SynSent": 0, "SynRecv": 0, "FinWait1": 0, "FinWait2": 0, "TimeWait": 0, "Close": 0, "CloseWait": 0, "LastAck": 0, "Listen": 0, "Closing": 0 }, "udp": { "Listen": 0, "Dropped": 0, "RxQueued": 0, "TxQueued": 0 }, "udp6": { "Listen": 0, "Dropped": 0, "RxQueued": 0, "TxQueued": 0 } }, "filesystem": [ { "device": "/dev/root", "type": "vfs", "capacity": 20749852672, "usage": 98007326

以上是关于heapster源码分析——kubelet的api调用分析的主要内容，如果未能解决你的问题，请参考以下文章

用 Heapster 监控集群 - 每天5分钟玩转 Docker 容器技术（176）

09-3.部署 heapster 插件

Kubelet源码分析之volume manager源码分析

kubelet源码分析: CNI 插件处理流程

kubelet分析