k8s源码分析-创建Pod流程

本文从源码层面解释kubernetes从接收创建Pod的指令到实际创建Pod的整个过程。

1.1 监听用户请求

监听的任务是由kube-apiserver这个组件来完成的，它实际上是一个WEB服务，一般是以双向TLS认证方式启动的，所以在启动时需要提供证书、私钥、客户端的CA证书和CA私钥，当然也支持HTTP的方式，启动后就开始监听用户请求了。

1.2 对请求分类

kube-apiserver中的WEB服务在启动时注册了很多的Handler，golang中的Handler相当于Java中的servlet或者是Spring中的Controller，是对某一业务逻辑的封装，通俗点说，一个Handler负责对一个URI请求的处理，而在kube-apiserver中，Handler被封装成了一个叫Store的对象，怎么封装的呢？比如/api/v1/namespaces/{namespace}/pods这个URI对应了一个叫PodStorage的Store，这个Store中包含了对/api/v1/namespaces/{namespace}/pods的多个Handler，这些Handler有的是处理创建请求，有的是处理删除请求等等，代表了对一种资源的操作集。

我们来看看这个Store的定义： staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go

type Store struct {
    CreateStrategy rest.RESTCreateStrategy
    AfterCreate ObjectFunc
    UpdateStrategy rest.RESTUpdateStrategy
    AfterUpdate ObjectFunc
    DeleteStrategy rest.RESTDeleteStrategy
    AfterDelete ObjectFunc
...
}

func (e *Store) Create(ctx genericapirequest.Context, obj runtime.Object, includeUninitialized bool) (runtime.Object, error) {

func (e *Store) Update(ctx genericapirequest.Context, name string, objInfo rest.UpdatedObjectInfo) (runtime.Object, bool, error) {
...

因为每种资源所需要的操作不一样，所以Store中只包含了基本的通用的操作，作为一个基础类。

1.3 处理请求

在Store的定义中有一个Storage storage.Interface字段，Store中的创建、更新、删除等操作，比如上面的Create()函数中会调用这个对象中的Create()方法，也就是说这个Storage对象包含了一组更低级的操作，可以看作是数据的持久化层，这些操作都是通用的，而Store可以用Storage中的功能组合出具有不同功能的控制层对象，也就是Store对象啊，，好吧我 们距离真相又进了一步，那这个Storage storage.Interface对象又是怎样实现的呢？

1.4 数据存储到ETCD

Storage字段是一个叫Interface的类型，里面定义了一些数据持久层的操作，这里就不贴出来了，我们更关心它的实现，我们先来看看Interface实例的创建吧，它的创 建工作是由一个工厂类负责的： staging/src/k8s.io/apiserver/pkg/storage/storagebackend/factory/factory.go

func Create(c storagebackend.Config) (storage.Interface, DestroyFunc, error) {
        switch c.Type {
        case storagebackend.StorageTypeETCD2:
                return newETCD2Storage(c)
        case storagebackend.StorageTypeUnset, storagebackend.StorageTypeETCD3:
                return newETCD3Storage(c)
        default:
                return nil, nil, fmt.Errorf("unknown storage type: %s", c.Type)
        }
}

好吧，看到这里就彻底明白了，根据配置中指定的存储服务创建不同的Storage对象，并且目前只支持ETCD2和ETCD3两种存储服务，持久层所做的增删改查就是对ETCD中 数据的增删改查，总结一下Storage对象也是就Interface被创建的过程：

1. 根据配置中提供的存储服务名称选择进入不相应的创建函数，也就是上面的逻辑。
2. 从配置对象中取出我们启动kube-apiserver时指定的ETCD的证书证书、私钥和CA证书（如果开启了双向证），用于apiserver和ETCD通信，用这些信自生成一个配置对象。
3. 用配置对象创建一个ETCD的客户端。
4. 创建一个k8s.io/apiserver/pkg/storage/etcd3.store对象并让其持有这个ETCD客户端
5. 将这个对象作为Interface返回。

1.5 调用ETCD客户端

那么这个k8s.io/apiserver/pkg/storage/etcd3.store对象就成了关键啦，因为它现了Interface的所有接口函数，我们看看它的Create()函数据是怎样实现的： vendor/k8s.io/apiserver/pkg/storage/etcd3/store.go

// Create implements storage.Interface.Create.
func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error {
        if version, err := s.versioner.ObjectResourceVersion(obj); err == nil && version != 0 {
                return errors.New("resourceVersion should not be set on objects to be created")
        }
        data, err := runtime.Encode(s.codec, obj)
        if err != nil {
                return err
        }
        key = path.Join(s.pathPrefix, key)

        opts, err := s.ttlOpts(ctx, int64(ttl))
        if err != nil {
                return err
        }

        newData, err := s.transformer.TransformToStorage(data, authenticatedDataString(key))
        if err != nil {
                return storage.NewInternalError(err.Error())
        }

        txnResp, err := s.client.KV.Txn(ctx).If(
                notFound(key),
        ).Then(
                clientv3.OpPut(key, string(newData), opts...),
        ).Commit()
        if err != nil {
                return err
        }
        if !txnResp.Succeeded {
                return storage.NewKeyExistsError(key, 0)
        }

        if out != nil {
                putResp := txnResp.Responses[0].GetResponsePut()
                return decode(s.codec, s.versioner, data, out, putResp.Header.Revision)
        }
        return nil
}

其中txnResp, err := s.client.KV.Txn(ctx).If(这一行就是调用ETCD客户端了。

2.1 调度Pod

调度工作是由kube-scheduler负责的，而且不受kube-apiserver控制，kube-scheduler通过kube-apiserver的REST API不断地检查是否有新的且还没有被调度的Pod，如果如有则根据配置中的调度算法为其绑定到某个节点，绑定的过程也是通过REST API将信息写入到kube-apiserver中的，对应的REST API定义如下： pkg/registry/core/rest/storage_core.go

func (c LegacyRESTStorageProvider) NewLegacyRESTStorage(restOptionsGetter generic.RESTOptionsGetter) (LegacyRESTStorage, genericapiserver.APIGroupInfo, error) {
...
        restStorageMap := map[string]rest.Storage{
...
                "pods/binding":     podStorage.Binding,
                "bindings":         podStorage.Binding,
...

"pods/binding"和"bindings"两个API就是了。

下面是REST API/api/v1/namespaces/{my-ns}/pods对应的Store对象的定义： pkg/registry/core/pod/storage/storage.go

func NewStorage(optsGetter generic.RESTOptionsGetter, k client.ConnectionInfoGetter, proxyTransport http.RoundTripper, podDisruptionBudgetClient policyclient.PodDisruptionBudgetsGetter) PodStorage {
...
        return PodStorage{
                Pod:         &REST{store, proxyTransport},
                Binding:     &BindingREST{store: store},

其中的Binding字段就是关于绑定逻辑的Store了。

3.1 创建Pod

最后的创建工作由各个节点上的kubelet组件负责，工作原理同kube-scheduler一样，通过REST API从kube-apiserver循环监听是否有新创建的并且已经被绑定到 自己节点的Pod，如果有则在自己的节点上创建相应的Docker容器并设置容器的网络、端口转发、内存和CPU限制等，然后把结果用REST API通知给kube-apiserver。

kubelet创建Pod的过程，主要逻辑在k8s源码的pkg/kubelet/kuberuntime/kuberuntime_manager.go文件的SyncPod()函数据中：

1. 检查Pod信息是否有更改，如果是则杀死现有的Pod，包括Pod中的所有容器。

2. 创建SandBox（沙箱），其实就是启动pause容器，这个过程中包含了一些重要操作：

   • 创建SandBox的函数：pkg/kubelet/dockershim/docker_sandbox.go:79

     func (ds *dockerService) RunPodSandbox(ctx context.Context, r *runtimeapi.RunPodSandboxRequest) (*runtimeapi.RunPodSandboxResponse, error) {
     

   • 拉取pause容器的镜像。

   • 启动pause容器。

   • 写入resolv.conf文件，覆盖docker默认创建的resolv.conf文件：pkg/kubelet/dockershim/docker_sandbox.go:140

     containerInfo, err := ds.client.InspectContainer(createResp.ID)
     

   • 设置Pod网络，也就是调用CNI来为容器添加网卡：pkg/kubelet/dockershim/docker_sandbox.go:163

     err = ds.network.SetUpPod(config.GetMetadata().Namespace, config.GetMetadata().Name, cID, config.Annotations)
     

3. 启动init容器。

4. 最后启动所有业务容器：pkg/kubelet/kuberuntime/kuberuntime_manager.go:712

   for _, idx := range podContainerChanges.ContainersToStart {
   		if msg, err := m.startContainer(podSandboxID, podSandboxConfig, container, pod, podStatus, pullSecrets, podIP, kubecontainer.ContainerTypeRegular); err != nil {
   ...
   }
   

至此Pod就算创建完成了，如果创建了相应的service，那么proxy组件会为service在物理机上通过iptables工具创建数据转发规则，将集群内的请求Pod的数据转发到相应的Pod中，如果Pod有多个副本，则以负载均衡的方式分别转发给这个Pod。

相关资料

• kubelet源码分析