从 iptables 看 k8s service 的实现机制

概述

k8s service 可以看做是多个 Pod 的负载均衡。有以下几种 service:

  • LoadBalancer
  • ClusterIP
  • NodePort
  • ExternalName

在 service 的演进中,从最初的 userspace 的方案变成 iptables 和 ipvs 的方案,其中,ipvs 主要是解决了 iptables 的性能问题。这篇文章主要分析 iptables 如何实现 service 的负载均衡。

ClusterIP

ClusterIP 是提供在集群中访问 Service 的方案,通常每个 Service 都会分配一个 VIP,然后为多个 Pod 提供负载均衡。这里我们创建两个副本的 nginx 部署,以及一个 nginx service。具体信息如下:

$ kubectl get endpoints nginx
NAME    ENDPOINTS                     AGE
nginx   172.17.0.4:80,172.17.0.5:80   65m

$ kubectl get service nginx
NAME    TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)   AGE
nginx   ClusterIP   10.111.67.225   <none>        80/TCP    65m

在集群中访问 nginx.default.svc.cluster.local 时,DNS 会将这个地址解析到 Service 的 IP 上,也就是 10.111.67.225。下面我们看看 iptables 是如何将访问这个地址的流量转到真实的 Pod 上的。

首先看一下 nat 表上的 OUTPUT 链:

$ iptables -t nat -nL OUTPUT
Chain OUTPUT (policy ACCEPT)
target     prot opt source               destination
KUBE-SERVICES  all  --  0.0.0.0/0            0.0.0.0/0            /* kubernetes service portals */
DOCKER     all  --  0.0.0.0/0           !127.0.0.0/8          ADDRTYPE match dst-type LOCAL

第一条规则会匹配所有的流量,然后跳到 KUBE-SERVICES 这条链上。我们看一下 KUBE-SERVICES 的具体内容:

$ iptables -t nat -nL KUBE-SERVICES
Chain KUBE-SERVICES (2 references)
target     prot opt source               destination
KUBE-SVC-NPX46M4PTMTKRN6Y  tcp  --  0.0.0.0/0            10.96.0.1            /* default/kubernetes:https cluster IP */ tcp dpt:443
KUBE-SVC-P4Q3KNUAWJVP4ILH  tcp  --  0.0.0.0/0            10.111.67.225        /* default/nginx:http cluster IP */ tcp dpt:80
KUBE-SVC-TCOU7JCQXEZGVUNU  udp  --  0.0.0.0/0            10.96.0.10           /* kube-system/kube-dns:dns cluster IP */ udp dpt:53
KUBE-SVC-ERIFXISQEP7F7OF4  tcp  --  0.0.0.0/0            10.96.0.10           /* kube-system/kube-dns:dns-tcp cluster IP */ tcp dpt:53
KUBE-SVC-JD5MR3NA4I4DYORP  tcp  --  0.0.0.0/0            10.96.0.10           /* kube-system/kube-dns:metrics cluster IP */ tcp dpt:9153
KUBE-NODEPORTS  all  --  0.0.0.0/0            0.0.0.0/0            /* kubernetes service nodeports; NOTE: this must be the last rule in this chain */ ADDRTYPE match dst-type LOCAL

这里前面的 KUBE-SVC-* 都是根据 destination, protocol 和目的端口号来匹配的,根据我们的 service 地址和端口号以及协议,可以定位到 KUBE-SVC-P4Q3KNUAWJVP4ILH 这条规则可以匹配,然后跳到这条链上。我们接着看这条链定义了什么:

$ iptables -t nat -nL KUBE-SVC-P4Q3KNUAWJVP4ILH
Chain KUBE-SVC-P4Q3KNUAWJVP4ILH (1 references)
target     prot opt source               destination
KUBE-SEP-GL7IUDQTUTXSADHR  all  --  0.0.0.0/0            0.0.0.0/0            /* default/nginx:http */ statistic mode random probability 0.50000000000
KUBE-SEP-VMO3WCKZND6ZICDD  all  --  0.0.0.0/0            0.0.0.0/0            /* default/nginx:http */

有两条规则,根据第一条规则后面的内容,我们可以知道这就是使用 iptables 实现负载均衡的地方了。第一条规则有 50% 的匹配几率。如果匹配到了其中一条,就会跳到另外一个链上。比如:

$ iptables -t nat -nL KUBE-SEP-GL7IUDQTUTXSADHR
Chain KUBE-SEP-GL7IUDQTUTXSADHR (1 references)
target     prot opt source               destination
KUBE-MARK-MASQ  all  --  172.17.0.4           0.0.0.0/0            /* default/nginx:http */
DNAT       tcp  --  0.0.0.0/0            0.0.0.0/0            /* default/nginx:http */ tcp to:172.17.0.4:80

其中第一条规则的 source 是 Pod 的 IP,在访问 Service 时目前还不会匹配,于是我们看第二条规则,将目的 IP 和 Port 改写成 172.17.0.4:80,也就是我们的 Pod IP,这样流量就经过负载均衡指向了我们的 Pod了。

NodePort

我们将上面的 Service 改成 NodePort

nginx        NodePort    10.111.67.225   <none>        80:30000/TCP   34h

然后查询机器上的 30000 端口。

$ ss -lp | grep 30000
tcp               LISTEN              0                    0                                                                                            0.0.0.0:30000                                                 0.0.0.0:*                  users:(("kube-proxy",pid=4006,fd=8))

可以看到, kube-proxy 监听了 30000 端口,同时我们看 nat 表上的 PREROUTING 链。

KUBE-SERVICES  all  --  0.0.0.0/0            0.0.0.0/0            /* kubernetes service portals */

再看 KUBE-SERVICES

KUBE-SVC-TCOU7JCQXEZGVUNU  udp  --  0.0.0.0/0            10.96.0.10           /* kube-system/kube-dns:dns cluster IP */ udp dpt:53
KUBE-SVC-ERIFXISQEP7F7OF4  tcp  --  0.0.0.0/0            10.96.0.10           /* kube-system/kube-dns:dns-tcp cluster IP */ tcp dpt:53
KUBE-SVC-JD5MR3NA4I4DYORP  tcp  --  0.0.0.0/0            10.96.0.10           /* kube-system/kube-dns:metrics cluster IP */ tcp dpt:9153
KUBE-SVC-NPX46M4PTMTKRN6Y  tcp  --  0.0.0.0/0            10.96.0.1            /* default/kubernetes:https cluster IP */ tcp dpt:443
KUBE-SVC-P4Q3KNUAWJVP4ILH  tcp  --  0.0.0.0/0            10.111.67.225        /* default/nginx:http cluster IP */ tcp dpt:80
KUBE-NODEPORTS  all  --  0.0.0.0/0            0.0.0.0/0            /* kubernetes service nodeports; NOTE: this must be the last rule in this chain */ ADDRTYPE match dst-type LOCAL

最后一条 KUBE-NODEPORTS 可以匹配到,这里有个匹配条件,那就是 ADDRTYPE match dst-type LOCAL。注意这里的 LOCAL 指的是本机网卡上存在的地址,也就是这条数据是发到本机,那么就能匹配。

KUBE-NODEPORTS 的规则如下:

KUBE-MARK-MASQ  tcp  --  0.0.0.0/0            0.0.0.0/0            /* default/nginx:http */ tcp dpt:30000
KUBE-SVC-P4Q3KNUAWJVP4ILH  tcp  --  0.0.0.0/0            0.0.0.0/0            /* default/nginx:http */ tcp dpt:30000

第一条规则是替换源地址为本机出口的网卡地址。第二条规则如下:

KUBE-SEP-F3MS6OIYSABTYGOY  all  --  0.0.0.0/0            0.0.0.0/0            /* default/nginx:http */ statistic mode random probability 0.50000000000
KUBE-SEP-VMO3WCKZND6ZICDD  all  --  0.0.0.0/0            0.0.0.0/0            /* default/nginx:http */

这里我们在 ClusterIP 中就分析了实现方法,因此这里忽略。

LoadBalancer

LoadBalancer 本身不是由 Kubernetes 提供的,其原理说起来也不难,我们先创建一个 LoadBalancer 的 Service 看看:

nginx        LoadBalancer   10.111.67.225   <pending>     80:32014/TCP   34h

这里因为我的本地集群没有 LoadBalancer,所以一直处于 Pending 状态。但是我们可以看到,这里还有一个 80:32014。和上面的 NodePort 输出一致。也就是说创建 LoadBalancer 时,会在 Pod 所在的机器上开启 NodePort,然后由外部的 LoadBalancer 将负载均衡过的流量带到机器的指定的 NodePort 上。

一些有意思的参数

这里顺便多提几个有意思的Service 参数

externalTrafficPolicy:可选值有 LocalCluster

  • Local: 流量只会被导向本机的 Pod,这样就少一次包的转发,提高性能。但是缺点是如果容易导致负载不均衡。
  • Cluster: 在集群范围内转发流量

如果能保证 Pod 均匀的分布在不同的节点上,那么外部的 LoadBalancer 配合 Local 的 externalTrafficPolicy 可以带来更好的性能。

sessionAffinity: 会话亲和性,可以设置为 ClientIP,来达到将同一个 IP 的会话转发到相同的 Pod 上。其也是通过 iptables 实现的。

KUBE-SEP-Q7ZFI57LOFFPF3HN  all  --  0.0.0.0/0            0.0.0.0/0            /* test/nginx-session-affinity:http */ recent: CHECK seconds: 10800 reap name: KUBE-SEP-Q7ZFI57LOFFPF3HN side: source mask: 255.255.255.255
KUBE-SEP-LWUZWBNY6M3CYJ2M  all  --  0.0.0.0/0            0.0.0.0/0            /* test/nginx-session-affinity:http */ recent: CHECK seconds: 10800 reap name: KUBE-SEP-LWUZWBNY6M3CYJ2M side: source mask: 255.255.255.255
KUBE-SEP-Q7ZFI57LOFFPF3HN  all  --  0.0.0.0/0            0.0.0.0/0            /* test/nginx-session-affinity:http */ statistic mode random probability 0.50000000000
KUBE-SEP-LWUZWBNY6M3CYJ2M  all  --  0.0.0.0/0            0.0.0.0/0            /* test/nginx-session-affinity:http */

这个 iptables 的前两条规则就是在做 iptables 的检查。

kubernetes 的 taints 和 tolerations 的理解和实践

概述

taints 和 tolerations 是一个比较好理解的概念,taints 可以翻译为污点,给 node 打上 taints,就可以用来驱逐 pod,并防止 pod 调度到该节点上。就像是某个人有了一个坏习惯(taints),那么其他人的就会远离这个人。但是有些人可以容忍别人的坏习惯,那么就会不受影响。就像 pod 拥有了 tolerations,就可以免疫节点上对应的 taints。

taints 的官方说明为:

The node this Taint is attached to has the “effect” on any pod that does not tolerate the Taint.

也就是说,taint 会为所有不能忍受该 taint 的 pod 添加副作用(不调度,偏好不调度,不执行)

如果要让某些 pod 免疫这些 taints,可以使用 tolerations。

taints 的使用

在使用 taints 的时候很简单,我们只需要指定节点 taints 的 key 和 value 即可。比如:

$ kubectl taint nodes minikube onlyNginxPod=true:NoSchedule

其中,onlyNginxPod 是 taints 的 key,true 是 value,NoSchedule 是 effect。另外还有 PreferNoScheduler 和 NoExecute。这里顺便总结一下这三种 effect 的区别:

  • NoSchedule: 表示不要将 Pod 向该节点调度。如果 Pod 已经调度到该节点了,则不受影响。
  • PreferNoScheduler: 表示尽量不要往该节点调度,但是如果没有其他选择,还是会将 Pod 调度到该节点。
  • NoExecute: Pod 不仅不能往上调度,所有已经运行在该节点上的 Pod 将会被驱逐。

这个时候,我们尝试创建一个普通的 pod,看看调度情况。

pod.yaml

apiVersion: v1
kind: Pod
metadata:
    name: nginx
    namespace: default
    labels:
        app: nginx
spec:
    containers:
    - name: nginx
      image: nginx:latest
$ kubectl apply -f pod-nginx.yaml 
$ kubectl describe pods nginx

Warning  FailedScheduling  <unknown>  default-scheduler  0/1 nodes are available: 1 node(s) had taint {onlyNginxPod: true}, that the pod didn't tolerate.

会出现上面的警告信息。表示因为 pod 没有容忍该 taint,所以没有办法调度上去。

我们可以用以下语句来删除 taint

$ kubectl taint node minikube onlyNginxPod=true:NoSchedule-

tolerations 的使用

某些情况下,我们仍然希望 Pod 可以调度到有 taint 的节点上,这时候就可以为 Pod 指定 tolerations。比如将上面的 Pod 改写成如下:

pod.yaml

apiVersion: v1
kind: Pod
metadata:
    name: nginx
    namespace: default
    labels:
        app: nginx
spec:
    containers:
    - name: nginx
      image: nginx:latest
    tolerations:
      - key: onlyNginxPod
        operator: Equal
        value: "true"
        effect: NoSchedule

然后创建这个 Pod,

···yaml
$ kubectl apply -f pod.yaml
$ kubectl get pods
NAME READY STATUS RESTARTS AGE
nginx 0/1 ContainerCreating 0 4s
···

说明该 pod 调度成功,tolerations 生效了。tolerations 会匹配 key 和 effect,只有一样的时候才会生效。tolerations 的 operator 字段除了 Equal 之外,还有 Exists

  • Equal: 要求 value 也相同。
  • Exists: 不需要设置 value 字段。

kubernetes 中使用 taints 和 tolerations 的常用场景

taints 和 tolerations 不仅仅是提供给用户使用的特性,kubernetes 本身也大量使用了 taints 和 tolerations。比如:

  • node.kubernetes.io/not-ready: Node 还没准备好,对应的 NodeCondition 的 ReadyFalse。比如在创建集群时,还没有安装 CNI 的话,节点就会有该 taint。对应的 effect 为 NoSchedule。

  • node.kubernetes.io/unreachable: node controller 无法连接到 Node,此时 NodeCondition 的 ReadyUnknown。对应的 effect 为 NoSchedule。

  • node.kubernetes.io/out-of-disk: 磁盘用尽。对应的 effect 为 NoSchedule。

  • node.kubernetes.io/memory-pressure: 节点有内存的压力。对应的 effect 为 NoSchedule。

  • node.kubernetes.io/disk-pressure: 节点有磁盘压力。和四个参数有关:nodefs.available, nodefs.inodesFree, imagefs.available, imagefs.inodesFree。nodefs 是用来存储卷和 daemon 日志的,imagefs 是容器运行时用来存储镜像和容器可写层的。当这些值到达某个阈值,就会出现 disk-pressure。对应的 effect 为 NoSchedule。

  • node.kubernetes.io/network-unavailable: 节点的网络还不可用。对应的 effect 为 NoSchedule。

  • node.kubernetes.io/unschedulable: 节点是不可调度的。对应的 effect 为 NoSchedule。

  • node.kubernetes.io/pid-pressure: 节点上的进程太多。对应的 effect 为 NoSchedule。

[Gaia Scheduler] gpu-manager 的虚拟化 gpu 分配流程

概述

在之前的一篇文章主要是分析了 gpu-manager 的启动流程。关于 gpu-manager 应该会有一系列的文章,一是觉得这是一个很有价值的项目,二是为这个项目花了好几天去看代码,想通过写文章的方式对内容进行梳理。

这篇文章主要分析 gpu-manager 的虚拟 gpu 分配原理,我认为将虚拟 gpu 分配给容器主要有两个重点:

  • gpu-manager 作为 device plugin 的工作流程
  • 虚拟 gpu 分配的最优方案,分配需要保证最少碎片,同时性能最好

从 pod 调度到虚拟 gpu 分配

这一部分会涉及到 device plugin 的工作机制,因此不熟悉的话可以看一下之前的一篇文章:Kubernetes开发知识–device-plugin的实现。下面附上一张这篇文章中 device plugin 的工作时序图:

device plugin

在之前的启动流程分析文章中,说到 gpu-manager 向 kubelet 注册。在这之后, gpu-manager 就正式作为一个 device plugin 来工作了。这个时候,我们可以创建如下的 pod:

apiVersion: v1
kind: Pod
metadata:
  name: tf-training-example-10
  namespace: test
  labels:
    name: tf-training-example
spec:
  restartPolicy: Never
  containers:
  - name: tf-training-example
    image: joyme/tf_training_example:1.5
    resources:
      requests:
        tencent.com/vcuda-core: 20
        tencent.com/vcuda-memory: 15
      limits:
        tencent.com/vcuda-core: 20
        tencent.com/vcuda-memory: 15

这个创建 pod 的请求会到达 kubernetes 的 API Server,然后由 kube-scheduler 进行调度。kube-scheduler 的调度经过预选和优选两个阶段,确定了最佳的目标节点。这时候 kubelet 就上场了。因为我们的 pod 中的容器请求了 vcuda-corevcuda-memory 这两个资源,但是 kubelet 并没有能力去给容器分配这些资源,于是它就找是谁注册了这些资源类型,然后发现是 vcore 和 vmemory 这两个服务注册的,于是使用 grpc 和 /var/lib/kubelet/device-plugins/vcore.sock 以及 /var/lib/kubelet/device-plugins/vmemory.sock 通过 unix socket 通讯。

vcore 和 vmemory 是两种资源,因此这里其实相当于注册了两个 device plugin。

对于 vcuda-memory,kubelet 调用的 Allocate 方法如下:

/** device plugin interface */
func (vr *vmemoryResourceServer) Allocate(ctx context.Context, reqs *pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error) {
    glog.V(2).Infof("%+v allocation request for vmemory", reqs)
    fakeData := make([]*pluginapi.ContainerAllocateResponse, 0)
    fakeData = append(fakeData, &pluginapi.ContainerAllocateResponse{})

    return &pluginapi.AllocateResponse{
        ContainerResponses: fakeData,
    }, nil
}

这里其实并没有做任何实际分配操作,我们可以认为 vcuda-core 和 vcuda-memory 必然是同时申请分配的,因此我们只需要处理二者之一即可。

对于 vcuda-core,kubelet 会调用的 Allocate 方法代码如下:

func (vr *vcoreResourceServer) Allocate(ctx context.Context, reqs *pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error) {
    glog.V(2).Infof("%+v allocation request for vcore", reqs)
    return vr.mgr.Allocate(ctx, reqs)
}

最终会走到 pkg/services/allocator/nvidia/allocator.go 的 Allcate 方法中。下面就来到这篇文章最复杂的部分了。

func (ta *NvidiaTopoAllocator) Allocate(_ context.Context, reqs *pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error) {
}

我们先看一下函数原型,reqs *pluginapi.AllocateRequest 这个参数是分配请求,然后返回了一个分配响应 *pluginapi.AllocateResponse。这里看一下 AllocateRequest:

// - Allocate is expected to be called during pod creation since allocation
//   failures for any container would result in pod startup failure.
// - Allocate allows kubelet to exposes additional artifacts in a pod's
//   environment as directed by the plugin.
// - Allocate allows Device Plugin to run device specific operations on
//   the Devices requested
type AllocateRequest struct {
    ContainerRequests []*ContainerAllocateRequest `protobuf:"bytes,1,rep,name=container_requests,json=containerRequests" json:"container_requests,omitempty"`
}

type ContainerAllocateRequest struct {
    DevicesIDs []string `protobuf:"bytes,1,rep,name=devicesIDs" json:"devicesIDs,omitempty"`
}

很明显,请求里包含了每个容器需要的设备数组。同时通过 AllocateRequest 上的注释可以得出以下信息:

  • Allocate 是在 pod 创建时被调用的,因此任何容器分配失败都会造成pod启动失败。
  • Allocate 允许 kubelet 在 pod 环境中引入更多的 artifacts,这部分工作由我们的 device plugin 主导。对于 gpu manager 来说就是,覆盖容器内的 LD_LIBRARY_PATH,挂载 cuda 库文件等等。
  • Allocate 允许 device plugin 在设备上运行特定的操作。

然后再来看一下 AllocateResponse:

// AllocateResponse includes the artifacts that needs to be injected into
// a container for accessing 'deviceIDs' that were mentioned as part of
// 'AllocateRequest'.
// Failure Handling:
// if Kubelet sends an allocation request for dev1 and dev2.
// Allocation on dev1 succeeds but allocation on dev2 fails.
// The Device plugin should send a ListAndWatch update and fail the
// Allocation request
type AllocateResponse struct {
    ContainerResponses []*ContainerAllocateResponse `protobuf:"bytes,1,rep,name=container_responses,json=containerResponses" json:"container_responses,omitempty"`
}

type ContainerAllocateResponse struct {
    // List of environment variable to be set in the container to access one of more devices.
    Envs map[string]string `protobuf:"bytes,1,rep,name=envs" json:"envs,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
    // Mounts for the container.
    Mounts []*Mount `protobuf:"bytes,2,rep,name=mounts" json:"mounts,omitempty"`
    // Devices for the container.
    Devices []*DeviceSpec `protobuf:"bytes,3,rep,name=devices" json:"devices,omitempty"`
    // Container annotations to pass to the container runtime
    Annotations map[string]string `protobuf:"bytes,4,rep,name=annotations" json:"annotations,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
}

这里我们又可以看到一些关键信息,AllocateResponse 为每个容器返回了 ContainerAllocateResponse,包括容器的环境变量,容器的挂载,容器的设备信息,容器的 annotations 信息。其中,容器的设备信息如下:

// DeviceSpec specifies a host device to mount into a container.
type DeviceSpec struct {
    // Path of the device within the container.
    ContainerPath string `protobuf:"bytes,1,opt,name=container_path,json=containerPath,proto3" json:"container_path,omitempty"`
    // Path of the device on the host.
    HostPath string `protobuf:"bytes,2,opt,name=host_path,json=hostPath,proto3" json:"host_path,omitempty"`
    // Cgroups permissions of the device, candidates are one or more of
    // * r - allows container to read from the specified device.
    // * w - allows container to write to the specified device.
    // * m - allows container to create device files that do not yet exist.
    Permissions string `protobuf:"bytes,3,opt,name=permissions,proto3" json:"permissions,omitempty"`
}

即在容器中挂载设备需要:

  • 设备相对于容器的地址
  • 设备在宿主机上的地址
  • 设备的 Cgroups 信息

这时候我们再来重新思考 gpu-manager 的 gpu 虚拟化原理。如果你看过腾讯关于 Gaia Scheduler 的论文,就会知道 gpu-manager 需要做以下工作:

  • 为容器挂载 cuda 相关的库,包括 vcuda-control 这个项目的拦截库
  • 通过覆盖容器中的 LD_LIBRARY_PATH 来将 cuda 调用指向 libcuda-control.so 这个库,这个库里面对显存和计算 api 做了拦截。
  • 为容器挂载 vcuda.sock,在容器调用特定的 cuda api 时,会触发 grpc 调用,通过 vcuda.sock 和 virtual manager 通信,virtual manager 下发容器配置。这样拦截库就知道自己应该怎么限制容器了。这里留一个问题 A,为什么要大费周章的通过 grpc,直接挂载容器配置文件可行吗?

这些 gpu-manager 要做的工作都是 device plugin 的 Allocate 调用提供的能力。所以 gpu-manager 需要在 Allocate 期间完成这么多的工作。这也是这部分比较复杂的原因。下面我们带着这些信息去看代码,会更容易懂一些。下面的代码都是来自于 pkg/services/allocator/nvidia/allocator.go 中的 Allocate 方法,但是因为很长,所以我会截取出来分析。

// k8s send allocate request for one container at a time
req := reqs.ContainerRequests[0]
resps := &pluginapi.AllocateResponse{}
reqCount = uint(len(req.DevicesIDs))

这部分取了 Allocate 中的第一个 ContainerRequest,通过注释知道,k8s 一次只为一个容器发送分配请求。

    if ta.unfinishedPod != nil {

    } else {

    }

接下来有一个对 unfinishedPod 的判断,因为 k8s 一次请求只针对一个容器,因此这里的 unfinishedPod 指的是只分配了部分容器,还有其他容器没有分配的 pod。这里我们需要仔细思考一下,使用 unfinishedPod 的目的是什么?看到这里我有两个猜测:

  1. 因为 k8s 一次请求只针对一个容器,所以为了优先分配完一个 pod,就需要标记 unfinishedPod 了。但是仔细想想,因为 Allocate 的请求和响应中都没有容器的信息,所以本次请求分配的容器是由 kubelet 决定的。device plugin 并没有能力改变容器的分配顺序,这个想法是错的。
  2. 为了性能考虑。因为 gpu-manager 有两个 device plugin:vmemory 和 vcore。但是之前说到 vmemory 的分配没有做任何工作。所以我们不得不在分配 vcore 的时候,把 vmemory 的分配工作也做了。可是我怎么知道当前正在给哪个容器分配资源?那我也更不知道分配多少 vmemory 了。但是天无绝人之路啊,我可以遍历当前节点上的所有 pod,然后挑出需要 gpu 资源的 pod。然后再从这些 pod 中挑出符合这次请求的容器。这里如果使用 unfinishedPod 就避免了重复的大规模查找操作。

那么,假设现在还有一个未完成的 pod,会执行下面的代码

// 候选pod
candidatePod = ta.unfinishedPod
// 从已分配的pod中查找
cache := ta.allocatedPod.GetCache(string(candidatePod.UID))
if cache == nil {
    msg := fmt.Sprintf("failed to find pod %s in cache", candidatePod.UID)
    glog.Infof(msg)
    return nil, fmt.Errorf(msg)
}
for i, c := range candidatePod.Spec.Containers {
    if _, ok := cache[c.Name]; ok {
        continue
    }

    if !utils.IsGPURequiredContainer(&c) {
        continue
    }

    if reqCount != utils.GetGPUResourceOfContainer(&candidatePod.Spec.Containers[i], types.VCoreAnnotation) {
        msg := fmt.Sprintf("allocation request mismatch for pod %s, reqs %v", candidatePod.UID, reqs)
        glog.Infof(msg)
        return nil, fmt.Errorf(msg)
    }
    // 候选的容器(应该就是待分配资源的容器)
    candidateContainer = &candidatePod.Spec.Containers[i]
    found = true
    break
}

上面这段代码遍历这个 pod 的容器列表,然后和缓存中的容器对比,如果没有分配并且需要 gpu 资源,并且容器请求的资源量和当前的分配请求一致,就认定这个容器是我们接下来要为之分配的候选人了。这里我们又有一个问题 B,如果一个 Pod 中有多个 vcore 请求一致,但是 vmemory 不同的容器,这里只通过 vcore 的请求量来判断,可以保证这个分配请求和我们的候选容器能对的上吗?这个问题我们可以产生如下的猜测:

  1. AllocateRequest 是按照 Pod 中的容器顺序来的,这样我们在做 reqCount 对比的时候,因为顺序一致就能保证请求和候选容器是对应关系了。那么,AllocateRequest 是按照 Pod 中容器顺序来的吗?这是一个新的问题 C。
  2. 其实请求和候选容器不对应也没关系,因为容器中进行 cuda 调用拦截的时候,才会请求 virtual manager,拿到容器的资源限制配置信息。只要这个环节能保证容器和其请求的资源量对应上,就不会有任何问题?这也是我们的问题 E:cuda 调用拦截的时候,如何保证容器和配置的对应关系。这也和问题 A 相呼应,如果这个猜测成立,那就是为什么问题 A 中要大费周折的使用 grpc 调用下发配置,而不是直接把配置信息挂载或写到容器的变量中。

接下来我们继续看,如果没有未完成的容器,就执行以下代码:


// 获取候选的pod,候选的pod是当前节点上的需要GPU,没有分配并且不应该删除的pod pods, err := getCandidatePods(ta.k8sClient, ta.config.Hostname) if err != nil { msg := fmt.Sprintf("Failed to find candidate pods due to %v", err) glog.Infof(msg) return nil, fmt.Errorf(msg) } for _, pod := range pods { if found { break } for i, c := range pod.Spec.Containers { if !utils.IsGPURequiredContainer(&c) { continue } podCache := ta.allocatedPod.GetCache(string(pod.UID)) if podCache != nil { if _, ok := podCache[c.Name]; ok { glog.Infof("container %s of pod %s has been allocate, continue to next", c.Name, pod.UID) continue } } if utils.GetGPUResourceOfContainer(&pod.Spec.Containers[i], types.VCoreAnnotation) == reqCount { glog.Infof("Found candidate Pod %s(%s) with device count %d", pod.UID, c.Name, reqCount) candidatePod = pod candidateContainer = &pod.Spec.Containers[i] found = true break } } }

和上面的不同之处,就是在获取候选 pod 这里。获取候选 pod 的代码如下:

    candidatePods := []*v1.Pod{}
    allPods, err := getPodsOnNode(client, hostname, string(v1.PodPending))

    for _, pod := range allPods {
        current := pod
        if utils.IsGPURequiredPod(&current) && !utils.IsGPUAssignedPod(&current) && !utils.ShouldDelete(&current) {
            candidatePods = append(candidatePods, &current)
        }
    }

    return OrderPodsdByPredicateTime(candidatePods), nil

先是获取节点上的所有 pod,然后从节点上的 pod 中选取需要 GPU,并且没有分配 GPU,并且不应该删除的 pod。最后得到一个候选 pod 列表。最后对这个列表根据时间排序。这样就可以拿到最先被调度的 pod 了。这里其实也默认了一个前提,最先调度的 pod 会最先发出分配请求。这里还有一个需要注意的地方,排序依据的时间有两个选择:预选时间或创建时间。

    if predicateTimeStr, ok := pod.ObjectMeta.Annotations[types.PredicateTimeAnnotation]; ok {
        u64, err := strconv.ParseUint(predicateTimeStr, 10, 64)
        if err != nil {
            glog.Warningf("Failed to parse predicate Timestamp %s due to %v", predicateTimeStr, err)
        } else {
            predicateTime = u64
        }
    } else {
        // If predicate time not found, use createionTimestamp instead
        predicateTime = uint64(pod.ObjectMeta.CreationTimestamp.UnixNano())
    }

    return predicateTime

其中,预选时间并不是 kube-scheduler 添加的,而是和 gpu-manager 配合使用的 gpu-admission 这个项目。如果没有预选时间,就会使用 pod 的创建时间。这也就是说,我们不使用 gpu-admission 这个项目,也可以正常使用 gpu-manager。其实这里还有一个问题 D,我怎么保证挑出来的容器就是这次分配请求的呢?这个问题还要留在后面的分析中。

现在我们拿到了候选容器,就需要进行真正的分配工作了。

// get vmemory info from container spec
vmemory := utils.GetGPUResourceOfContainer(candidateContainer, types.VMemoryAnnotation)
for i := 0; i < int(vmemory); i++ {
    req.DevicesIDs = append(req.DevicesIDs, types.VMemoryAnnotation)
}

resp, err := ta.allocateOne(candidatePod, candidateContainer, req)
if err != nil {
    glog.Errorf(err.Error())
    return nil, err
}
resps.ContainerResponses = append(resps.ContainerResponses, resp)

这段代码中,我们拿到容器的 vmemory 信息。因为 vmemory 是根据数量划分的。1 个 vmemory 相当于 256M 的 memory,也就是一个 deviceID。这里请求多少的 vmemory,就存多少个 deviceID。然后调用 allocateOne 为单个容器进行真正的分配工作。下面我们开始分析 allocateOne 的分配逻辑。

var (
    nodes                       []*nvtree.NvidiaNode
    needCores, needMemoryBlocks int64
    predicateMissed             bool
    allocated                   bool
)

// 是否是 gpu 预选 pod
predicateMissed = !utils.IsGPUPredicatedPod(pod)
// 单节点的总内存
singleNodeMemory := int64(ta.tree.Leaves()[0].Meta.TotalMemory)
for _, v := range req.DevicesIDs {
    if strings.HasPrefix(v, types.VCoreAnnotation) {
        // 请求 core
        needCores++
    } else if strings.HasPrefix(v, types.VMemoryAnnotation) {
        // 请求 memory
        needMemoryBlocks++
    }
}

首先就是根据 deviceID 来计算需要多少 core 和 memory。接下来会调用 ta.recycle() 回收资源。回收的逻辑如下:

func (ta *NvidiaTopoAllocator) recycle() {
    activePods := watchdog.GetActivePods()

    lastActivePodUids := sets.NewString()
    activePodUids := sets.NewString()
    for _, uid := range ta.allocatedPod.Pods() {
        lastActivePodUids.Insert(uid)
    }
    for uid := range activePods {
        activePodUids.Insert(uid)
    }

    // difference 出来的就是已经运行结束的pod,可以回收分配的gpu资源
    podsToBeRemoved := lastActivePodUids.Difference(activePodUids)

    glog.V(5).Infof("Pods to be removed: %v", podsToBeRemoved.List())

    // 释放资源
    ta.freeGPU(podsToBeRemoved.List())
}

对已分配的 pod 和 正在运行的 pod 集合取差集,差集就是分配了资源但是已经停止运行的 pod 。然后对这部分 pod 释放 GPU 资源。具体的释放逻辑放在后面分析。现在继续向下看,这里我们直接跳到尝试分配资源的逻辑上。分配 gpu 资源分为三种情况:

  1. 如果需要的核心数大于 100,也就是说超过一个物理 GPU,就使用 link 评估器来选出 GPU 节点
  2. 如果正好是一个 100 核心,则使用 fragment 评估器
  3. 如果小于 100 核心,则使用 share 评估器。

情况 1 的代码如下:

eval, ok := ta.evaluators["link"]
if !ok {
    return nil, fmt.Errorf("can not find evaluator link")
}
if needCores%nvtree.HundredCore > 0 {
    return nil, fmt.Errorf("cores are greater than %d, must be multiple of %d", nvtree.HundredCore, nvtree.HundredCore)
}
nodes = eval.Evaluate(needCores, 0)

注意到这里还要求请求的核心数必须是 100 的整数,也就是说必须是整数个物理 GPU,你不能请求 1.5 个 物理 GPU 这种。

情况 2 的代码如下:

eval, ok := ta.evaluators["fragment"]
if !ok {
    return nil, fmt.Errorf("can not find evaluator fragment")
}
nodes = eval.Evaluate(needCores, 0)

情况 3 的代码如下:

// EnableShare 是在启动时指定的参数,代表是否允许多个容器共享一个gpu
if !ta.config.EnableShare {
    return nil, fmt.Errorf("share mode is not enabled")
}
if needCores == 0 || needMemory == 0 {
    return nil, fmt.Errorf("that cores or memory is zero is not permitted in share mode")
}

// evaluate in share mode
shareMode = true
// 使用 share 评估
eval, ok := ta.evaluators["share"]
if !ok {
    return nil, fmt.Errorf("can not find evaluator share")
}
// 评估出来的合适的 nvidia gpu 节点
nodes = eval.Evaluate(needCores, needMemory)
if len(nodes) == 0 {
    if shareMode && needMemory > singleNodeMemory {
        return nil, fmt.Errorf("request memory %d is larger than %d", needMemory, singleNodeMemory)
    }

    return nil, fmt.Errorf("no free node")
}

在评估出来节点之后,会先判断这个这个 pod 是否真的经过预选阶段?判断方法如下:

func IsGPUPredicatedPod(pod *v1.Pod) (predicated bool) {
    glog.V(4).Infof("Determine if the pod %s needs GPU resource", pod.Name)
    var ok bool

    // Check if pod request for GPU resource
    if GetGPUResourceOfPod(pod, types.VCoreAnnotation) <= 0 || GetGPUResourceOfPod(pod, types.VMemoryAnnotation) <= 0 {
        glog.V(4).Infof("Pod %s in namespace %s does not Request for GPU resource",
            pod.Name,
            pod.Namespace)
        return predicated
    }

    // Check if pod already has predicate time
    // tencent.com/predicate-time 是 gpu-admission 中添加的。
    if _, ok = pod.ObjectMeta.Annotations[types.PredicateTimeAnnotation]; !ok {
        glog.V(4).Infof("No predicate time for pod %s in namespace %s",
            pod.Name,
            pod.Namespace)
        return predicated
    }

    // Check if pod has already been assigned
    if assigned, ok := pod.ObjectMeta.Annotations[types.GPUAssigned]; !ok {
        glog.V(4).Infof("No assigned flag for pod %s in namespace %s",
            pod.Name,
            pod.Namespace)
        return predicated
    } else if assigned == "true" {
        glog.V(4).Infof("pod %s in namespace %s has already been assigned",
            pod.Name,
            pod.Namespace)
        return predicated
    }
    predicated = true
    return predicated
}

共有三个要求才算经过了预选:

  • resource 字段请求了 vcore 和 vgpu,并且大于 0。
  • 必须有 tencent.com/predicate-time 字段。这点要求必须经过 gpu-admission 的预选阶段。
  • 没有被分配 gpu 资源

如果经过预选的话,就需要执行以下的逻辑:

// get predicate node by annotation
containerIndex, err := utils.GetContainerIndexByName(pod, container.Name)
if err != nil {
    return nil, err
}
var devStr string
if idxStr, ok := pod.ObjectMeta.Annotations[types.PredicateGPUIndexPrefix+strconv.Itoa(containerIndex)]; ok {
    if _, err := strconv.Atoi(idxStr); err != nil {
        return nil, fmt.Errorf("predicate idx %s invalid for pod %s ", idxStr, pod.UID)
    }
    devStr = types.NvidiaDevicePrefix + idxStr
    if !utils.IsValidGPUPath(devStr) {
        return nil, fmt.Errorf("predicate idx %s invalid", devStr)
    }
} else {
    return nil, fmt.Errorf("failed to find predicate idx for pod %s", pod.UID)
}

predicateNode := ta.tree.Query(devStr)
if predicateNode == nil {
    return nil, fmt.Errorf("failed to get predicate node %s", devStr)
}

// check if we choose the same node as scheduler
if predicateNode.MinorName() != nodes[0].MinorName() {
    return nil, fmt.Errorf("Nvidia node mismatch for pod %s(%s), pick up:%s  predicate: %s",
        pod.Name, container.Name, nodes[0].MinorName(), predicateNode.MinorName())
}

也就是说,经过预选阶段的 Pod 都会根据容器的顺序在 Annotations 为该容器写上配置信息。这说明在 gpu-admission 这个项目中会为容器分配 gpu 设备。最后还要检查一下在 gpu-manager 中分配的 gpu 设备和 gpu-admission 中是否一致,不一致的话也会返回分配失败。

现在我们已经知道要为当前请求的容器分配哪个 gpu 设备,以及分配的资源数量。这样就可以构建 ContainerAllocateResponse 了。先把已分配的设备放到响应中:

    for _, n := range nodes {
        name := n.MinorName()
        glog.V(2).Infof("Allocate %s for %s(%s), Meta (%d:%d)", name, pod.UID, container.Name, n.Meta.ID, n.Meta.MinorID)

        ctntResp.Annotations[types.VCoreAnnotation] = fmt.Sprintf("%d", needCores)
        ctntResp.Annotations[types.VMemoryAnnotation] = fmt.Sprintf("%d", needMemory)

        ctntResp.Devices = append(ctntResp.Devices, &pluginapi.DeviceSpec{
            ContainerPath: name,
            HostPath:      name,
            Permissions:   "rwm",
        })
        deviceList = append(deviceList, n.Meta.UUID)

        if !allocated {
            // 在 gpu tree 中标记设备已占用
            ta.tree.MarkOccupied(n, needCores, needMemory)
        }
        allocatedDevices.Insert(name)
    }

更改响应的 Annotations:

ctntResp.Annotations[types.VDeviceAnnotation] = vDeviceAnnotationStr(nodes)

检查 pod 的所有容器是否都完成了分配,并把新的分配信息写入到 checkpoint:

unfinished := false
for _, c := range pod.Spec.Containers {
    if !utils.IsGPURequiredContainer(&c) {
        continue
    }
    podCache := ta.allocatedPod.GetCache(string(pod.UID))
    if podCache != nil {
        if _, ok := podCache[c.Name]; !ok {
            unfinished = true
            break
        }
    }
}
if unfinished {
    ta.unfinishedPod = pod
} else {
    ta.unfinishedPod = nil
}
ta.writeCheckpoint()

在响应中为容器添加 /dev/nvidiactl/dev/nvidia-uvm,如果配置了 extraConfig,还会把里面要默认添加的设备加进去:

// Append control device
ctntResp.Devices = append(ctntResp.Devices, &pluginapi.DeviceSpec{
    ContainerPath: types.NvidiaCtlDevice,
    HostPath:      types.NvidiaCtlDevice,
    Permissions:   "rwm",
})

ctntResp.Devices = append(ctntResp.Devices, &pluginapi.DeviceSpec{
    ContainerPath: types.NvidiaUVMDevice,
    HostPath:      types.NvidiaUVMDevice,
    Permissions:   "rwm",
})

// Append default device
if cfg, found := ta.extraConfig["default"]; found {
    for _, dev := range cfg.Devices {
        ctntResp.Devices = append(ctntResp.Devices, &pluginapi.DeviceSpec{
            ContainerPath: dev,
            HostPath:      dev,
            Permissions:   "rwm",
        })
    }
}

此时,响应中的设备信息已经处理结束,接下来处理容器中的环境变量,gpu manager 需要通过修改 LD_LIBRARY_PATH 来劫持程序对 cuda 的调用,然后通过 NVIDIA_VISIBLE_DEVICES 来让挂载的设备可见。

// LD_LIBRARY_PATH
ctntResp.Envs["LD_LIBRARY_PATH"] = "/usr/local/nvidia/lib64"
for _, env := range container.Env {
    if env.Name == "compat32" && strings.ToLower(env.Value) == "true" {
        ctntResp.Envs["LD_LIBRARY_PATH"] = "/usr/local/nvidia/lib"
    }
}

// NVIDIA_VISIBLE_DEVICES
ctntResp.Envs["NVIDIA_VISIBLE_DEVICES"] = strings.Join(deviceList, ",")

接着根据是否处于 shareMode,也就是单个 gpu 能否被共享来挂载不同的 host 目录。

if shareMode {
    // nvidia 是劫持的库,用在shareMode这种情况
    ctntResp.Mounts = append(ctntResp.Mounts, &pluginapi.Mount{
        ContainerPath: "/usr/local/nvidia",
        HostPath:      types.DriverLibraryPath,
        ReadOnly:      true,
    })
} else {
    // 非shareMode用正常的库即可
    ctntResp.Mounts = append(ctntResp.Mounts, &pluginapi.Mount{
        ContainerPath: "/usr/local/nvidia",
        HostPath:      types.DriverOriginLibraryPath,
        ReadOnly:      true,
    })
}

shareMode 下,会挂载的 host 目录是 /etc/gpu-manager/vdriver/nvidia,这里面是被劫持的库。否则挂载 /etc/gpu-manager/vdriver/origin,里面是原始的 CUDA 库。

紧接着,将 host 上的 /etc/gpu-manager/vm/{podUID} 挂载到容器中,这个是为了容器内可以通过 vcuda.sockvirtual-manager 通信。

// 将host上的/etc/gpu-manager/vm/podUID挂载进去(vcuda.sock),这个目录是在PreStartContainer期间由VirtualManager创建的
ctntResp.Mounts = append(ctntResp.Mounts, &pluginapi.Mount{
    ContainerPath: types.VCUDA_MOUNTPOINT,
    HostPath:      filepath.Join(ta.config.VirtualManagerPath, string(pod.UID)),
    ReadOnly:      true,
})

如果当前请求的容器所属 Pod 没有经过 gpu-admission,还会被放到一个处理队列中:

if predicateMissed {
    ar := &allocateResult{
        pod:     pod,
        result:  PREDICATE_MISSING,
        resChan: make(chan struct{}),
    }

    // 这个 queue 的处理是在virtualmanager里面的process方法
    ta.queue.AddRateLimited(ar)
    <-ar.resChan
}

这个队列会在 pkg/service/allocator/nvidia/allocator.goproccessResult 中处理。

这样,kubelet 调用 Allocate 方法就结束了。这里再来回顾一下上面遗留的问题和相关逻辑:

问题:

  • 问题 A: 为什么要大费周章的通过 grpc,直接挂载容器配置文件可行吗?

    总结: 这一点在上面的阅读中可以发现,这时候容器本身对自己应该限制多少的 gpu 资源调用并不知道。这个问题得和 B/C/D 问题结合来看。因为做 Allocate 调用时,kubelet 并没有告知此时在为哪个容器请求配置。因此只能根据请求的资源量以及 Pod 的 predicateTime 或 createTime 来判断。这个是无法保证一定准确的,因此此时容器的具体资源配置也无法确定。可能这就是要通过 grpc 而不是挂载容器配置文件的原因吧。

  • 问题 B: 如果一个 Pod 中有多个 vcore 请求一致,但是 vmemory 不同的容器,这里只通过 vcore 的请求量来判断,可以保证这个分配请求和我们的候选容器能对的上吗?
    总结:问题 B 是在 unfinisedPod 中查找当前请求的容器。只要能保证 unfinishedPod 是正确的(问题 D 说明不能保证),那么就可以保证容器是对的上的(问题 C 保证了这个结论)。

  • 问题 C: AllocateRequest 是按照 Pod 中的容器顺序来的?
    总结:对于这个问题,最好的回答方式是去看 kubelet 的源代码。

    for _, container := range pod.Spec.Containers {
        if err := m.allocateContainerResources(pod, &container, devicesToReuse); err != nil {
            return err
        }
        m.podDevices.removeContainerAllocatedResources(string(pod.UID), container.Name, devicesToReuse)
    }
    

    这边做 Allocate 的时候,是顺序遍历 Pod 中的容器,因此这个问题的答案是肯定的。

  • 问题 D: 遍历当前节点上的所有 pod,然后挑出需要 gpu 资源的 pod,根据 predicatedTimecreateTime 排序。然后再从这些 pod 中, 按顺序挑出符合这次请求的容器,怎么保证挑出来的容器就是这次分配请求的呢?

    总结:我觉得回答这个问题,需要确定两个大前提,一是 Pod 从创建到发起 Allocate 的过程,都是顺序的。这样就能保证当调用 Allocate 对应的 Pod 永远是尚未分配到资源的第一个。二是在一个 Pod 中,为每个容器 Allocate 时,也是顺序的,这一点在问题 C 中得到确认。

    但是实际上,第一个前提是不能保证的,在 Pod bind 到节点时,这个是并发执行的。因此可以得出一个结论:在这个阶段无法保证 Allocate 请求和我们的候选容器是对应关系。关于这一点我也提了个 issue:a question about Allocate for a container?。官方也给了回答,因为这个原因 gpu manager 有时候会报 UnexpectedAdmissionError 错误。

    所以根据问题 4,我们还要使用 gpu-admission 这个项目,来保证该阶段的正确性,具体机制还得等到看 gpu-admission 的时候才能知道了。

其实以上四个问题都是因为 kubelet 的 Allocate 请求不会带上正在分配的容器。所以需要一系列的查找方式来确定具体的容器。

因为篇幅问题,关于 gpu 的最佳分配策略会作为下一篇文章的内容。

参考资料

从零开始入门 K8s:调度器的调度流程和算法介绍

kubernetes 的挂载传播(mount propagation)机制

概述

今天在看 kubectl-debug 这个项目的时候,看到其部署文件的 volumeMounuts 中使用了一个 mountPropagation 字段,因为不清楚这个字段的作用,就做了一下了解。mount propagation 背后的东西还是很多的,因此整理了这篇文章,顺便梳理一下知识点。

kubernetes 的 mount propagation 翻译成中文就是挂载传播。挂载传播提供了共享卷挂载的能力,它允许在同一个 Pod,甚至同一个节点内,在多个容器之间共享卷的挂载。

kubernetes 的挂载传播

卷的挂载传播由 Container.volumeMounts 的 mountPropagation 字段控制。它的值有:

  • None: 这种卷挂载将不会收到任何后续由 host 创建的在这个卷上或其子目录上的挂载。同样的,由容器创建的挂载在 host 上也是不可见的。这是默认的模式。这个其实很好理解,就是容器内和 host 的后续挂载完全隔离。

  • HostToContainer: 这种卷挂载将会收到之后所有的由 host 创建在该卷上或其子目录上的挂载。换句话说,如果 host 在卷挂载内挂载的任何内容,在容器中都是可见的。同样,如果任何具有 Bidirectional 的 Pod 挂载传播到该卷挂载上,具有 HostToContainer 的挂载传播都可以看见。整个挂载传播的流程如下:

  • Bidirectional: 这种挂载机制和 HostToContainer 类似。此外,任何在容器中创建的挂载都会传播到 host,然后传播到使用相同卷的所有 Pod 的所有容器。注意:Bidirectional 挂载传播是很危险的。可能会危害到 host 的操作系统。因此只有特权容器在允许使用它。

在了解了这几种挂载传播之后,我们可以做一些实验来验证一下,首先验证的是 None 的挂载传播类型,我们创建一个 nginx 的Pod:

apiVersion: v1
kind: Pod
metadata:
    name: mount-a
    namespace: default
    label:
      app: mount
spec:
    containers:
    - name: main
      image: nginx:latest
      volumeMounts:
      - name: testmount
        mountPath: /home
        mountPropagation: None
    volumes:
    - name: testmount
      hostPath:
        path: /mnt/

然后我们分别向 host 的 /mnt 和容器的 /home 下挂载目录并查看容器和 host 的情况:

容器中:

$ kubectl exec -it mount-a sh
$ cd /home
$ ls 
sda1

host 上:

$ cd /mnt
$ ls 
sda1

然后在 host 上创建挂载:

$ mkdir /mnt/none
$ sudo mount --bind /var /mnt/none
$ ls none
cache  empty  lib  lock  log  run  spool  tmp

这个时候,我们再看容器中的文件:

$ ls none
# 无输出

这说明 host 上在该卷下的挂载并不会改变容器中的文件。接下来我们可以在容器中按照上面的方案来验证容器中的挂载也不会影响 host 中的目录视图。这里就不展示了。接下来看一下 HostToContainer 的挂载传播,我们将上面的 Pod 的 mountPropagation 字段改成 HostToContainer,然后先取消 host 上的挂载:

sudo umoint /mnt/none

然后重新创建 Pod,和上面一样,在 host 上创建挂载,查看容器中的挂载情况:

$ ls none
cache  empty  lib  lock  log  run  spool  tmp

host 上的挂载因为 HostToContainer 机制传播到了容器中。我们继续看最后一种 Bidirectional 机制。这次我们要创建两个 Pod: mount-a, mount-b,并把 mountPropagation 字段改成 Bidirectional。注意,因为 Bidirectional 是危险的,所以只有特权容器才可以使用。因此这里还需要把容器改成特权模式,最后在 mount-a 中的容器执行挂载,验证挂载是否传播到 host 和 mount-b 的容器中。

apiVersion: v1
kind: Pod
metadata:
    name: mount-a
    namespace: default
    labels:
      app: mount
spec:
    containers:
    - name: main
      image: nginx:latest
      securityContext:
        privileged: true
      volumeMounts:
      - name: testmount
        mountPath: /home
        mountPropagation: Bidirectional
    volumes:
    - name: testmount
      hostPath:
        path: /mnt/
---
apiVersion: v1
kind: Pod
metadata:
    name: mount-b
    namespace: default
    labels:
      app: mount
spec:
    containers:
    - name: main
      image: nginx:latest
      securityContext:
        privileged: true
      volumeMounts:
      - name: testmount
        mountPath: /home
        mountPropagation: Bidirectional
    volumes:
    - name: testmount
      hostPath:
        path: /mnt/

然后进入 mount-a,创建挂载:

$ kubectl exec -it mount-a sh
$ su
$ mount --bind /var /home/none
$ ls /home/none
backups  lib    lock  mail  run    tmp
cache    local  log   opt   spool

这时候查看 host 下的 /mnt/none:

$ ls /mnt/none
backups  lib    lock  mail  run    tmp
cache    local  log   opt   spool

可以发现,容器中的挂载传播到了 host 上。这时候再查看 mount-b 中的容器。

$ ls /home/none
backups  lib    lock  mail  run    tmp
cache    local  log   opt   spool

挂载也传播到了 mount-b 的容器中。

linux mount 的几种类型

上面分析了 kubernetes 的挂载传播机制,在 linux mount 中,也有类似的概念。mount 分为下面几种:

  • shared mount: 相当于上面所说的 Bidirectional 的挂载传播
  • slave mount: 每个 slave mount 都有一个 shared master mount,挂载传播只能从 master -> slave,等同于上面的 HostToContainer, host 是 master,container 是 slave。
  • private mount: 很明显,private 就是相当于 None,挂载不会向任何一方传播。
  • unbindable mount:unbindable mount 其实就是 unbindable private mount,也就是不允许使用 --bind 的挂载。

mount namespace 的机制

kubernetes 的挂载传播不是其本身实现的,也不是 docker 之类的容器运行时提供的。这是由容器化技术的基础:linux namespace 提供的,linux namespace 当前共有 6 种:

  • cgroup namespace: 隔离 cgroup 根目录
  • pid namespace: 隔离进程 id
  • ipc namespace: 隔离 System V IPC, POSIX message queues
  • uts namespace: 隔离 Hostname 和 NIS domain name
  • user namespace: 隔离用户和用户组 ID
  • mount namespace: 隔离挂载点
  • network namespace: 隔离网络设备,网络栈,端口等

其中,mount namespace 是这篇文章的重点。我们可以通过 clone 调用来看看 mount namespace 的使用:

#define _GNU_SOURCE
#include <stdio.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/mount.h>
#include <sched.h>
#include <signal.h>
#include <unistd.h>

#define STACK_SIZE (1024*1024)
static char container_stack[STACK_SIZE];

char* const container_args[] = {
    "/bin/bash",
    NULL
};

int container_main(void* arg)
{
    printf("Container [%5d] - inside the container!\n", getpid());
    mount("none", "/", NULL, MS_REC|MS_PRIVATE, NULL);
    execv(container_args[0], container_args);
    printf("Something's wrong!\n");
    return 1;
}
int main()
{
    printf("Parent [%5d] - start a container!\n", getpid());
    /* 启用Mount Namespace - 增加CLONE_NEWNS参数 */
    int container_pid = clone(container_main, container_stack+STACK_SIZE, CLONE_NEWNS | SIGCHLD, NULL);
    waitpid(container_pid, NULL, 0);
    printf("Parent - container stopped!\n");
    return 0;
}

编译运行:

$ gcc main.c -o mount
$ sudo ./mount

然后尝试挂载,来验证挂载 MS_PRIVATE 的挂载传播问题。MS_PRIVATE 下 namespace 内和 host 应该是隔离的。MS_PRIVATE 还可以替换成 MS_UNBINDABLEMS_SLAVEMS_SHARED

关于更多的 namespace 的资料,建议看这两篇文章:

参考资料

[Gaia Scheduler] gpu-manager 启动流程分析

概述

Gaia scheduler 是腾讯开源的在 Kubernetes 集群中做 GPU 虚拟化的方案,实现了为容器分配虚拟化 GPU 资源并加以限制,它的最大的优势就是不需要特殊的硬件支持,并且性能损耗很小。关于它的论文,地址在这里:Gaia Scheduler: A Kubernetes-Based Scheduler Framework。如果想要理解这个项目,强烈建议先读这篇论文。

Gaia Scheduler 可以分为 4 个组件:

  • GPU Manager: 作为 device plugin 向 kubelet 注册。共注册了两个设备,包括 vcore 和 vmemory,支持两种计算资源:tencent.com/vcuda-coretencent.com/vcuda-memory,分别用来做 GPU 计算资源和 GPU 内存资源的请求和限制。

  • GPU Scheduler: 这里的 scheduler 并不是 kubernetes 的调度器,是 GPU Manager 在收到 kubelet 的 Allocate 调用后,它需求将设备挂载给容器。为了实现最佳的 GPU 挂载,就有这样一个专门的 Scheduler 来根据节点上当前的 GPU 拓扑和资源占用情况进行调度。

  • vGPU Manager: vGPU Manager 是具体负责管理容器的组件,包括监控容器状态,传递配置,和容器内的vGPU Library通信,以及在容器死亡后进行回收操作。

  • vGPU Library: vGPU Library 虽然相关的代码量不多,但它是 Gaia Scheduler 最重要的部分。因为它是实现 GPU 虚拟化的核心。通过覆盖容器中的 LD_LIBRARY_PATH 以及自定义了 libcuda-control.so 实现对 CUDA API 的拦截。

Gaia Scheduler 主要由三个项目组成: gpu-managervcuda-controllergpu-admission。但是这里的 gpu-manager 是 Gaia Scheduler 的主要实现,包含了上述的 4 个组件,vcuda-controller 就是 vGPU Library,已经被打包到了 gpu-manager 这个项目中。gpu-manager 需要配合 gpu-admission 项目来完成 GPU Scheduler 的工作。不要因此产生误解。下文中我们主要就 gpu-manager 这个项目进行分析。

启动流程分析

gpu-manager 本身主要作为 kubernetes 的 device plugin 来实现的,定义了两种设备: vcuda-corevcuda-memory,我们的应用通过 pod 的资源字段进行申请,然后 kube-scheduler 会根据节点上的资源状态进行调度。因此,你最好还需要了解 kubernetes 的 device plugin 的开发知识。关于 device plugin 的开发,可以看之前的一篇文章:Kubernetes开发知识–device-plugin的实现

启动参数

分析一个项目从启动参数开始,可以帮助我们快速了解:

  • driver: 这个是 GPU 的驱动,当前的默认值是 nvidia,很显然该项目可以扩展支持其他类型的 GPU。
  • extra-config: 额外的配置,这个参数暂时看不出来有什么特别
  • volume-config: 这里的 volume 指的是一些动态链接库和可执行文件的位置。也就是 gpu-manager 需要拦截调用的一些库
  • docker-endpoint: 用来挂载到容器中和 docker 做通信的,默认位置是 unix:////var/run/docker.sock
  • query-port: 统计信息服务的查询接口
  • query-port: 统计信息服务的监听地址
  • kubeconfig: 用来授权的配置文件
  • standalone: 暂时还不清楚的参数
  • sample-period: gpu-manager 会查询 gpu 设备的使用情况,这个参数用来设定采样周期
  • node-labels: 给节点自动打标签
  • hostname-override: gpu-manager 在运行时,只关注自己节点上的 pod,这主要是通过 hostname 来辨认的
  • virtual-manager-path: gpu-manager 会为所有需要虚拟 gpu 资源的 pod 创建唯一的文件夹,文件夹的路径就在这个地址下。
  • device-plugin-path: kubernetes 默认的 device plugin 的目录地址
  • checkpoint-path: gpu-manager 会产生 checkpoint 来当缓存用
  • share-mode: gpu-manager 最大的特点就是将一个物理 gpu 分成多个虚拟 gpu,也就是共享模式
  • allocation-check-period: 检查分配了虚拟 gpu 资源的 pod 的状态,及时回收资源
  • incluster-mode: 是否在集群内运行

服务启动

gpu-manager 推荐的部署方案是通过 kubernetes 的 daemonset,然后配置 node selector 调度到指定的节点上。然后 gpu-manager 就开始在指定节点上启动了。

srv := server.NewManager(cfg)
go srv.Run()

这里,我们需要看一下这个 srv 的具体实现,首先是它的结构体:

type managerImpl struct {
    config *config.Config

    allocator      allocFactory.GPUTopoService     // gpu 容器调度分配
    displayer      *display.Display                // gpu 使用情况可视化服务
    virtualManager *vitrual_manager.VirtualManager // 负责管理 vgpu

    bundleServer map[string]ResourceServer
    srv          *grpc.Server
}

config 包含了我们上面的所有参数,就不进去细看了。

allocator 负责在容器调度到节点上后,为其分配具体的设备资源。allocator 实现了探测节点上的 gpu 拓扑架构,然后以最佳性能,最少碎片为目的使用最优的方案进行资源分配。

displayer 是将 gpu 的使用情况输出,方便我们查看。

virtualManager 负责 vgpu 分配后的管理工作。

bundleServer 包含 vcore,vmemory,我们上面提到这两种资源以 device plugin 的方式进行注册,因此他们需要启动 grpc server。

srv: 将 gpu display server 注册到这个 grpc server 中。

接下来,我们就可以分析 srv.Run() 方法具体执行了哪些内容。为了先对整个流程有个大概的印象,我将内容整理成以下条目:

  • 启动 volumeManager,将节点上和 nvidia gpu (包括cuda) 的所有可执行文件和库移动到 /etc/gpu-manager/vdriver 中。并且将关键的库替换成 vcuda-control,实现 cuda 调用的拦截。
  • watchdog 创建 pod 缓存并监控 pod,之后所有关于 pod 的操作都来源于这里。
  • watchdog 给节点打上标签
  • 启动 virtualManager
  • gpu 拓扑结构感知。
  • 初始化资源分配器
  • 设置 vcuda, vmemory, display 的 grpc 服务
  • 启动 metrics 的 http 服务,主要是提供给 prometheus
  • 启动 vcuda,vmemory 的 grpc 服务
  • 启动 display 的 grpc 服务

接下来,我们具体来分析每一步是如何做的。当然,这里只会挑一些重点的部分。

volumeManager 的启动

func (vm *VolumeManager) Run() (err error) {
    // ldcache 是动态链接库的缓存信息
    cache, err := ldcache.Open()
    defer func() {
        if e := cache.Close(); err == nil {
            err = e
        }
    }()
    vols := make(VolumeMap)
    for _, cfg := range vm.Config {
        vol := &Volume{
            Path: path.Join(cfg.BasePath, cfg.Name),
        }

        if cfg.Name == "nvidia" {
            // nvidia 库的位置
            types.DriverLibraryPath = filepath.Join(cfg.BasePath, cfg.Name)
        } else {
            // origin 库的位置
            types.DriverOriginLibraryPath = filepath.Join(cfg.BasePath, cfg.Name)
        }

        for t, c := range cfg.Components {
            switch t {
            case "binaries":
                // 调用 which 来查找可执行文件的位置
                bins, err := which(c...)
                // 将实际位置存起来
                vol.dirs = append(vol.dirs, volumeDir{binDir, bins})
            case "libraries":
                // 是库的话,就从 ldcache 里面去找
                libs32, libs64 := cache.Lookup(c...)
                // 将 library 位置存起来
                vol.dirs = append(vol.dirs, volumeDir{lib32Dir, libs32}, volumeDir{lib64Dir, libs64})
            }
            vols[cfg.Name] = vol
        }
    }
    // 找到了需要的库位置之后,做 mirror 处理
    if err := vm.mirror(vols); err != nil {
        return err
    }
    return nil
}

这段代码的前半部分都是在查找指定的动态链接库和可执行文件,这些文件是在 volume.conf 这个配置文件中指定的,通过参数传进来。查找动态链接库时,使用的是 ldcache,查找可执行文件时,使用了系统的 which 指令。找到之后会将其所在位置记录下来。接着就是对找到的库做 mirror 处理。

func (vm *VolumeManager) mirror(vols VolumeMap) error {
    // nvidia 和 origin
    for driver, vol := range vols {
        if exist, _ := vol.exist(); !exist {
            // 这里的path是/etc/gpu-manager/vdriver下面
            if err := os.MkdirAll(vol.Path, 0755); err != nil {
                return err
            }
        }
        for _, d := range vol.dirs {
            vpath := path.Join(vol.Path, d.name)
            // 创建 bin lib lib64
            if err := os.MkdirAll(vpath, 0755); err != nil {
                return err
            }

            // For each file matching the volume components (blacklist excluded), create a hardlink/copy
            // of it inside the volume directory. We also need to create soname symlinks similar to what
            // ldconfig does since our volume will only show up at runtime.
            for _, f := range d.files {
                glog.V(2).Infof("Mirror %s to %s", f, vpath)
                if err := vm.mirrorFiles(driver, vpath, f); err != nil {
                    return err
                }

                if strings.HasPrefix(path.Base(f), "libcuda.so") {
                    driverStr := strings.SplitN(strings.TrimPrefix(path.Base(f), "libcuda.so."), ".", 2)
                    types.DriverVersionMajor, _ = strconv.Atoi(driverStr[0]) // 驱动版本号
                    types.DriverVersionMinor, _ = strconv.Atoi(driverStr[1])
                    glog.V(2).Infof("Driver version: %d.%d", types.DriverVersionMajor, types.DriverVersionMinor)
                }

                if strings.HasPrefix(path.Base(f), "libcuda-control.so") {
                    vm.cudaControlFile = f
                }
            }
        }
    }

    vCudaFileFn := func(soFile string) error {
        if err := os.Remove(soFile); err != nil {
            if !os.IsNotExist(err) {
                return err
            }
        }
        if err := clone(vm.cudaControlFile, soFile); err != nil {
            return err
        }

        glog.V(2).Infof("Vcuda %s to %s", vm.cudaControlFile, soFile)

        l := strings.TrimRight(soFile, ".0123456789")
        if err := os.Remove(l); err != nil {
            if !os.IsNotExist(err) {
                return err
            }
        }
        if err := clone(vm.cudaControlFile, l); err != nil {
            return err
        }
        glog.V(2).Infof("Vcuda %s to %s", vm.cudaControlFile, l)
        return nil
    }

    if vm.share && len(vm.cudaControlFile) > 0 {
        if len(vm.cudaSoname) > 0 {
            for _, f := range vm.cudaSoname {
                if err := vCudaFileFn(f); err != nil {
                    return err
                }
            }
        }

        if len(vm.mlSoName) > 0 {
            for _, f := range vm.mlSoName {
                if err := vCudaFileFn(f); err != nil {
                    return err
                }
            }
        }
    }

    return nil
}

这段代码先会对所有上面查找到的库或可执行文件调用 mirrorFiles,但是记录下来了 libcuda.so 的版本号和 libcuda-control.so 的位置。注意,这个 libcuda-control 就是 vcuda-control 项目生成的用来拦截 cuda 调用的库。

然后将 cudaControlFile clone到所有 cudaSonamemlSoName 中库的位置。这个 clone 方法会先尝试硬链接过去,如果失败就直接复制过去。这里的 cudaControlFile 就是我们上面所说的 libcuda-control.so 啦。cudaSonamemlSoName 包含了所有需要被拦截调用的库。这样子就实现了拦截所有的 cuda 调用。下面我们在看一下 mirrorFiles 这个方法就可以了。

// driver 是配置文件中的 "nvidia" 或 "origin"
// vpath 是要 mirror 到的位置,在 /etc/gpu-manager/vdriver 下面
func (vm *VolumeManager) mirrorFiles(driver, vpath string, file string) error {
    // In computing, the Executable and Linkable Format (ELF, formerly named Extensible Linking Format), is a common standard file format for executable files, object code, shared libraries, and core dumps
    obj, err := elf.Open(file)
    defer obj.Close()

    // 黑名单机制,具体用处还不清楚,跟 nvidia 的驱动相关
    ok, err := blacklisted(file, obj)
    if ok {
        return nil
    }
    l := path.Join(vpath, path.Base(file))
    // 不管有没有,先尝试把 gpu-manager 里面的移除
    if err := removeFile(l); err != nil {
        return err
    }
    // clone 优先硬连接,其次是复制文件到指定位置
    if err := clone(file, l); err != nil {
        return err
    }
    // 从 elf 中获取当前库的 soname
    soname, err := obj.DynString(elf.DT_SONAME)
    if len(soname) > 0 {
        // 将获取到 soname 组成路径
        l = path.Join(vpath, soname[0])
        // 如果文件和它的soname不一致(是否可以认为这个文件是软链接过去的)
        if err := linkIfNotSameName(path.Base(file), l); err != nil && !os.IsExist(err) {
            return err
        }

        // XXX Many applications (wrongly) assume that libcuda.so exists (e.g. with dlopen)
        // Hardcode the libcuda symlink for the time being.
        if strings.Contains(driver, "nvidia") {
            // 这里为什么要移除 libcuda.so 和 libnvidia-ml.so 的软链接
            // 因为gpu调用会涉及到这两个库,这两个库会软链接到真实的库上。移除后替换成拦截的库
            // Remove libcuda symbol link
            if vm.share && driver == "nvidia" && strings.HasPrefix(soname[0], "libcuda.so") {
                os.Remove(l)
                vm.cudaSoname[l] = l
            }

            // Remove libnvidia-ml symbol link
            if vm.share && driver == "nvidia" && strings.HasPrefix(soname[0], "libnvidia-ml.so") {
                os.Remove(l)
                vm.mlSoName[l] = l
            }

            // XXX GLVND requires this symlink for indirect GLX support
            // It won't be needed once we have an indirect GLX vendor neutral library.
            if strings.HasPrefix(soname[0], "libGLX_nvidia") {
                l = strings.Replace(l, "GLX_nvidia", "GLX_indirect", 1)
                if err := linkIfNotSameName(path.Base(file), l); err != nil && !os.IsExist(err) {
                    return err
                }
            }
        }
    }

    return nil
}

这段代码中,先使用 blacklisted 排除一些不需要处理的库,然后尝试将库或可执行文件 clone 到我们的 /etc/gpu-manager/vdriver 下面。/etc/gpu-manager/vdriver 下面有两个文件夹,一个是 nvidia,保存了已经被我们拦截的库,一个是 origin,这里面是原始的未处理的库。同时,还将 libcuda.so 和 libnvidia-ml.so 移除了,这样就调用不到真实的库了,转而在之后用我们拦截的库来替换这几个文件。

至此,volumeManager 分析结束。

gpu 拓扑结构感知

关于 gpu 拓扑结构这一块,主要是为了在之后做资源分配时选择最优方案用的。腾讯也有分享过这一块的资料(腾讯基于 Kubernetes 的企业级容器云实践):

gpu 拓扑结构

这里不影响我们理解整个工作机制,所以先不分析。

初始化资源分配器

// 分配器,根据driver调用相应的分配器
initAllocator := allocFactory.NewFuncForName(m.config.Driver)
if initAllocator == nil {
    return fmt.Errorf("can not find allocator for %s", m.config.Driver)
}

m.allocator = initAllocator(m.config, tree, client)

这里的 initAllocator 对应的方法是:

//NewNvidiaTopoAllocator returns a new NvidiaTopoAllocator
func NewNvidiaTopoAllocator(config *config.Config, tree device.GPUTree, k8sClient kubernetes.Interface) allocator.GPUTopoService {
    runtimeRequestTimeout := metav1.Duration{Duration: 2 * time.Minute}
    imagePullProgressDeadline := metav1.Duration{Duration: 1 * time.Minute}
    dockerClientConfig := &dockershim.ClientConfig{
        DockerEndpoint:            config.DockerEndpoint,
        RuntimeRequestTimeout:     runtimeRequestTimeout.Duration,
        ImagePullProgressDeadline: imagePullProgressDeadline.Duration,
    }

    _tree, _ := tree.(*nvtree.NvidiaTree)
    cm, err := checkpoint.NewManager(config.CheckpointPath, checkpointFileName)
    if err != nil {
        glog.Fatalf("Failed to create checkpoint manager due to %s", err.Error())
    }
    alloc := &NvidiaTopoAllocator{
        tree:              _tree,
        config:            config,
        evaluators:        make(map[string]Evaluator),
        dockerClient:      dockershim.NewDockerClientFromConfig(dockerClientConfig),
        allocatedPod:      cache.NewAllocateCache(),
        k8sClient:         k8sClient,
        queue:             workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
        stopChan:          make(chan struct{}),
        checkpointManager: cm,
    }

    // Load kernel module if it's not loaded
    alloc.loadModule()

    // Initialize evaluator
    alloc.initEvaluator(_tree)

    // Read extra config if it's given
    alloc.loadExtraConfig(config.ExtraConfigPath)

    // Process allocation results in another goroutine
    go wait.Until(alloc.runProcessResult, time.Second, alloc.stopChan)

    // Recover
    alloc.recoverInUsed()

    // Check allocation in another goroutine periodically
    go alloc.checkAllocationPeriodically(alloc.stopChan)

    return alloc
}

allocator 调用 loadModule() 来启用 nvidia 的内核模块。

调用 initEvaluator(_tree) 来初始化评估器,这里的 _tree 就是感知到的 gpu 拓扑结构。

调用 loadExtraConfig(config.ExtraConfigPath) 来加载启动时传入的额外参数配置文件。

go wait.Until(alloc.runProcessResult, time.Second, alloc.stopChan) 创建了新的协程来处理分配结果。

recoverInUsed() 是恢复 gpu 分配结果。比如在 gpu-manager 重启之后,之前的 gpu 分配结果都丢失了,但是节点上还有大量的容器正在占用 gpu,这个方法会通过查找节点上存活的容器,通过 docker endpoint, 调用 InspectContainer 获取容器中占用的 device id,然后标记该设备和容器之间的占用关系。

go alloc.checkAllocationPeriodically(alloc.stopChan) 创建新的协程来周期性的检查资源分配情况。如果是 Failed 和 Pending 状态的容器,就根据错误信息检查是否应该删除它们,然后如果这些 pod 的控制器是 deployment 类似的,就尝试删除它们,这样控制器会重新创建这些 pod 进行调度,让这些 pod 恢复到正常运行状态。

启动各种服务

vcuda,vmemory 的 grpc 服务是 device plugin 的机制。metrics service 是提供给 prometheus 调用的,以监控该节点的相关信息。display 服务会打印 gpu 拓扑结构的相关信息。

Device plugin 的注册

Device plugin

这张图是 device plugin 注册的时序图。gpu-manager 的注册方法是:

func (m *managerImpl) RegisterToKubelet() error {
    socketFile := filepath.Join(m.config.DevicePluginPath, types.KubeletSocket)
    dialOptions := []grpc.DialOption{grpc.WithInsecure(), grpc.WithDialer(utils.UnixDial), grpc.WithBlock(), grpc.WithTimeout(time.Second * 5)}

    conn, err := grpc.Dial(socketFile, dialOptions...)
    if err != nil {
        return err
    }
    defer conn.Close()

    client := pluginapi.NewRegistrationClient(conn)

    for _, srv := range m.bundleServer {
        req := &pluginapi.RegisterRequest{
            Version:      pluginapi.Version,
            Endpoint:     path.Base(srv.SocketName()),
            ResourceName: srv.ResourceName(),
            Options:      &pluginapi.DevicePluginOptions{PreStartRequired: true},
        }

        glog.V(2).Infof("Register to kubelet with endpoint %s", req.Endpoint)
        _, err = client.Register(context.Background(), req)
        if err != nil {
            return err
        }
    }

    return nil
}

这里分别注册了 vcuda 和 vmemory。vcuda 和 vmemory 的 Allocate 方法都指向了同一个方法,写在了 service/allocator/nvidia/allocator.go 中。

至此,gpu-manager 的启动流程结束。接下来的 gpu-manager 的职责就是等待 kubelet 通过 grpc 的调用,在容器调度到节点的时候进行资源设备的分配,必要目录的挂载等工作了。具体的可以见下一篇文章

最后,提供一个简单的脑图帮助理解:

gpu-manager-arch

Kubernetes开发知识–device-plugin的实现

什么是 device plugin

Kubernetes 作为一个自动化容器编排系统,在调度 pod 的时候会根据容器需要的资源进行节点的选择,节点的选择会分为预选和优选阶段。预选阶段会根据所有节点上剩余的资源量与 pod 需要的资源量进行对比,选出能够满足需求的节点。通常情况下,这里的资源都会包括 CPU 和 Memory,就像下面这样:

resources:
  requests:
    memory: "64Mi"
    cpu: "250m"
  limits:
    memory: "128Mi"
    cpu: "500m"

但是现在我们有了新的需求,我们有一个 tensorflow 的模型需要借助 tensorflow serving 来部署,同时我们希望采用 GPU 部署的方案以加快模型的在线推算速度。这样我们就需要一个 GPU 的资源并借助 Kubernetes 的调度器将容器调度到有空余 GPU 资源的方案。

在 1.11 版本之前的 Kubernetes 中,提供了 alpha.kubernetes.io/nvidia-gpu 的资源名称来帮助我们根据 GPU 资源调度。但是这也带来了一些问题:

  • Kubernetes 需要维护 NVIDIA GPU 相关的代码,增加了维护成本。
  • NVIDIA GPU 方面的专家不一定熟悉 Kubernetes,这不符合让最擅长的人做最擅长的事的原则。
  • 除了 NVIDIA GPU,还会有其他的计算资源需要支持。

因此,Kubernetes 在 1.8 版本引入了 device plugin 机制,将第三方的计算资源通过插件的方式引入 Kubernetes,并且由第三方厂商自行维护。Kubernetes 社区的活瞬间就轻松了,第三方厂商也开心了。

通过以上的说明,可以总结出 device plugin 主要用来解耦第三方计算资源和 kubernetes 系统,将第三方的计算资源通过插件的方式引入 Kubernetes。当然 cpu 和 memory 除外,毕竟谁还能少了 CPU 和 Memory 呢。

device plugin 能做什么

目前一些常用的 device plugin 有:

我了解的还有腾讯的 Gaia Scheduler,通过 device plugin 实现的 GPU 虚拟化方案。如果 NVIDIA 的 GPU 方案还不够适合你,可以看看腾讯的这个方案:tkestack/gpu-manager

我觉得腾讯的 GPU 虚拟化方案是最能说明 device plugin 使用场景的例子,通过定义 tencent.com/vcuda-coretencent.com/vcuda-memory 这两个计算资源,来将一个物理 GPU 划分成多个虚拟 GPU 进行调度,这样可以实现一个 GPU 上部署多个 tensorflow serving。你不需要购买特殊的硬件或者修改任何 Kubernetes 的代码,就有了 GPU 虚拟化的能力。

下面我们开个脑洞,现在我们有了一种叫做 cola 的计算资源,可以提高程序的 IO 能力。但是 cola 的资源有限,只分配给特定的容器使用。这时候我们就可以通过实现自己的 device plugin 来满足这个需求。我们的计算资源名就叫做: myway5.com/cola,在下面一小节做具体的实现。

device plugin 的实现方案

device plugin 的工作原理其实不复杂。主要有以下步骤:

  • 首先 device plugin 可以通过手动或 daemonset 部署到需要的节点上。
  • 为了让 Kubernetes 发现 device plugin,需要向 kubelet 的 unix socket。 进行注册,注册的信息包括 device plugin 的 unix socket,API Version,ResourceName。
  • kubelet 通过 grpc 向 device plugin 调用 ListAndWatch, 获取当前节点上的资源。
  • kubelet 向 api server 更新节点状态来通知资源变更。
  • 用户创建 pod,请求资源并调度到节点上后,kubelet 调用 device plugin 的 Allocate 进行资源分配。

时序图如下:

device plugins

在 device plugin 的实现中,最关键的两个要实现的方法是 ListAndWatchAllocate。除此之外,还要注意监控 kubelet 的重启,一般是使用 fsnotify 类似的库监控 kubelet.sock 的重新创建事件。如果重新创建了,则认为 kubelet 是重启了,我们需要重新向 kubelet 注册 device plugin。

ListAndWatch

我们上面定义的 myway5.com/cola 资源用 /etc/colas 下的文件代表。每一个文件代表一个可用的资源。因此实现 ListAndWatch 就是查找该文件夹下的文件,然后添加到设备列表发送给 kubelet,之后调用 fsnotify 去监控文件的 CREATEREMOVE 事件。每次设备列表发生变更都重新向 kubelet 发送更新过的设备列表。

// ListAndWatch returns a stream of List of Devices
// Whenever a Device state change or a Device disappears, ListAndWatch
// returns the new list
func (s *ColaServer) ListAndWatch(e *pluginapi.Empty, srv pluginapi.DevicePlugin_ListAndWatchServer) error {
    log.Infoln("ListAndWatch called")
    devs := make([]*pluginapi.Device, len(s.devices))

    i := 0
    for _, dev := range s.devices {
        devs[i] = dev
        i++
    }

    err := srv.Send(&pluginapi.ListAndWatchResponse{Devices: devs})
    if err != nil {
        log.Errorf("ListAndWatch send device error: %v", err)
        return err
    }

    // 更新 device list
    for {
        log.Infoln("waiting for device change")
        select {
        case <-s.notify:
            log.Infoln("开始更新device list, 设备数:", len(s.devices))
            devs := make([]*pluginapi.Device, len(s.devices))

            i := 0
            for _, dev := range s.devices {
                devs[i] = dev
                i++
            }

            srv.Send(&pluginapi.ListAndWatchResponse{Devices: devs})
        }
    }
}

Allocate

在用户创建的 Pod 请求资源时,Kubernetes 的调度器会进行调度,并通过 kubelet 向 device plugin 发出 Allocate 调用,这一步的调用主要是为了让 device plugin 为容器调度资源。 在调度成功后向 kubelet 返回调度结果即可。

// Allocate is called during container creation so that the Device
// Plugin can run device specific operations and instruct Kubelet
// of the steps to make the Device available in the container
func (s *ColaServer) Allocate(ctx context.Context, reqs *pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error) {
    log.Infoln("Allocate called")
    resps := &pluginapi.AllocateResponse{}
    for _, req := range reqs.ContainerRequests {
        log.Infof("received request: %v", strings.Join(req.DevicesIDs, ","))
        resp := pluginapi.ContainerAllocateResponse{
            Envs: map[string]string{
                "COLA_DEVICES": strings.Join(req.DevicesIDs, ","),
            },
        }
        resps.ContainerResponses = append(resps.ContainerResponses, &resp)
    }
    return resps, nil
}

部署

device plugin 可以手动部署到机器上,也可以通过 Daemonset 进行部署。这里当然是 Daemonset 进行部署了。部署的时候有几个注意事项:

  • 需要挂载 hostPath,其中 /var/lib/kubelet/device-plugins 是必须的。这个文件夹下有 kubelet.sock,以及我们也需要将 device plugin 的 unix socket 文件存在这里。使得 kubelet 可以和我们的应用通信。
  • 为 device plugin 的 Pod 设置调度优先级别,通常设置成 priorityClassName: "system-node-critical"。这样可以保证不会因为节点利用率过高被逐出。
  • 如果资源设备不是每台机器都有,建议使用 nodeSelector 将 device plugin 调度到指定的机器上。

device plugin 的开发源代码可以参考上面的 cola 例子:cola device plugin

部署结束之后,可以查看一下节点的资源情况:

$ kubectl describe nodes test
Capacity:
 cpu:                2
 ephemeral-storage:  17784752Ki
 hugepages-2Mi:      0
 memory:             1986740Ki
 myway5.com/cola:    2
 pods:               110
Allocatable:
 cpu:                2
 ephemeral-storage:  17784752Ki
 hugepages-2Mi:      0
 memory:             1986740Ki
 myway5.com/cola:    2
 pods:               110

创建一个 pod,请求 myway5.com/cola 资源:

$ kubectl apply -f e2e/pod-with-cola.yaml

然后查看一下 cola pod 的日志来了解设备发现和调度情况:

$ kubectl -n kube-system logs cola-thtm9 
time="2020-03-24T08:16:53Z" level=info msg="cola device plugin starting"
time="2020-03-24T08:16:53Z" level=info msg="find device 'cocacola'"
time="2020-03-24T08:16:53Z" level=info msg="find device 'peisicola'"
time="2020-03-24T08:16:53Z" level=info msg="watching devices"
time="2020-03-24T08:16:53Z" level=info msg="start GPPC server for 'myway5.com/cola'"
time="2020-03-24T08:16:53Z" level=info msg="Register to kubelet with endpoint cola.sock"
time="2020-03-24T08:16:53Z" level=info msg="register to kubelet successfully"
time="2020-03-24T08:16:53Z" level=info msg="ListAndWatch called"
time="2020-03-24T08:16:53Z" level=info msg="waiting for device change"
time="2020-03-24T08:17:10Z" level=info msg="Allocate called"

Kubernetes 开发知识–Kubernetes 准入控制与 admission webhook 的使用

一、什么是准入控制

我们都知道 Kubernetes 的最核心组件就是它的 API Server,所有资源的创建、更新和删除都是通过 API Server 进行的。我们可以通过 HTTP 请求或者 kubectl 这样的客户端来和 APIServer 通信,在我们操作对象的请求到达 API Server 之前,会先到达准入控制器这里。准入控制器可以执行“验证”和“变更”操作。因此我们可以认为有两种准入控制器:变更(mutating)准入控制器验证(validating)准入控制器

准入控制过程也同样分为两个阶段。第一阶段,运行变更准入控制器,对对象进行修改操作;第二阶段,运行验证准入控制器。如果任何一个阶段的任何控制器拒绝了该请求,则整个请求将立即被拒绝,并向终端用户返回一个错误。

我们可以通过下面的图来直观的感受一下:

Kubernetes 开发知识--Kubernetes 准入控制与 admission webhook 的使用

二、准入控制能做什么

准入控制存在的目的就是提高 Kubernetes 架构的灵活性。因此 Kubernetes 提供了一些准入控制器,并且允许你打开或关闭一部分,同时你也可以自定义准入控制器,来完成你自己的一些特殊需求。可以自定义的准入控制器也分为两种:一个叫 MutatingAdmissionWebhook,一个叫 ValidatingAdmissionWebhook,其实也是我们上面说的变更准入控制器验证准入控制器

Kubernetes 提供了一些常用的准入控制器,下面举几个例子:

  • AlwaysPullImages: 该准入控制器会修改每一个新创建的 Pod 的镜像拉取策略为 Always。这样在多租户集群里,用户就能保证自己的私有镜像只会在有凭证的情况下使用,而不会出现因为机器上缓存了镜像而被其他用户直接使用的情况。

  • NamespaceLifecycle:该准入控制器禁止在一个正在被终止的 Namespace 中创建新对象,并且确保使用不存在的 Namespace 的请求被拒绝。该准入控制器还会禁止删除三个系统保留的命名空间,即 defaultkube-systemkube-public

除了 Kubernetes 提供的这些准入控制器,我们可以编写 MutatingAdmissionWebhookValidatingAdmissionWebhook。比如我之前有写一个叫做 [lazykube](https://github.com/joyme123/lazykube)MutatingAdmissionWebhook,它的作用是对每一个创建的 Pod 的镜像源都进行校验,如果是 docker hub, gcr.io, quay.io 等地址的镜像,就修改成国内的代理源,这样就可以自动解决镜像需要翻墙下载的问题了。

三、如何使用 admission webhook 编写一个准入控制器

使用 admission webhook 编写一个准入控制器其实很简单,它就和你日常写 web server 一样:

// Start 启动服务
func (whsrv *WebhookServer) Start() error {
    mux := http.NewServeMux()
    mux.HandleFunc("/mutate", whsrv.serve)
    whsrv.server.Handler = mux

    if err := whsrv.server.ListenAndServeTLS("", ""); err != nil {
        return fmt.Errorf("Failed to listen and serve webhook server: %v", err)
    }

    return nil
}

在收到一个 AdmissionReview 请求的时候,将其中的 AdmissionRequest 对象携带的资源对象取出来进行校验或变更:

var body []byte
if r.Body != nil {
    if data, err := ioutil.ReadAll(r.Body); err == nil {
        body = data
    }
}

if len(body) == 0 {
    log.Error("empty body")
    http.Error(w, "empty body", http.StatusBadRequest)
    return
}

// verify the content type is accurate
contentType := r.Header.Get("Content-Type")
if contentType != "application/json" {
    log.Errorf("Content-Type=%s, expect application/json", contentType)
    http.Error(w, "invalid Content-Type, expect `application/json`", http.StatusUnsupportedMediaType)
    return
}

var admissionResponse *v1beta1.AdmissionResponse
ar := v1beta1.AdmissionReview{}
if _, _, err := deserializer.Decode(body, nil, &ar); err != nil {
    log.Errorf("Can't decode body: %v", err)
    admissionResponse = &v1beta1.AdmissionResponse{
        Result: &metav1.Status{
            Message: err.Error(),
        },
    }
} else {
    admissionResponse = whsrv.mutate(&ar)
}

然后生成 AdmissionResponse, 通过 AdmissionReview 进行响应即可:

admissionReview := v1beta1.AdmissionReview{}
if admissionResponse != nil {
    admissionReview.Response = admissionResponse
    if ar.Request != nil {
        admissionReview.Response.UID = ar.Request.UID
    }
}

resp, err := json.Marshal(admissionReview)
if err != nil {
    log.Errorf("Can't encode response: %v", err)
    http.Error(w, fmt.Sprintf("could not encode response: %v", err), http.StatusInternalServerError)
}
log.Infoln("Ready to write response ...")
if _, err := w.Write(resp); err != nil {
    log.Errorf("Can't write response: %v", err)
    http.Error(w, fmt.Sprintf("could not write response: %v", err), http.StatusInternalServerError)
}

完整的代码可以参考上面的 lazykube 项目。

这样我们写好之后就可以开始部署了。我们需要告诉 Kubernetes 我们的 admission webhook 的地址,以及我们要操作的资源对象,因此我们创建一个 MutatingWebhookConfiguration 或者 ValidatingWebhookConfiguration 对象,比如 lazykube 就是对创建 Pod 进行变更操作:

apiVersion: admissionregistration.k8s.io/v1beta1
kind: MutatingWebhookConfiguration
metadata:
  name: lazykube-webhook-cfg
  namespace: kube-system
  labels:
    app: lazykube
webhooks:
  - name: lazykube.myway5.com
    clientConfig:
      service:
        name: lazykube-webhook-svc
        namespace: kube-system
        path: "/mutate"
      caBundle: ××××××××
    rules:
      - operations: [ "CREATE" ]
        apiGroups: [""]
        apiVersions: ["v1"]
        resources: ["pods"]

然后为了让 Kubernetes 通过 HTTPS 进行通信,并信任我们的证书,我们需要让 Kubernetes 集群为我们的签发证书。具体的方法可以参考这里:kubernetes 集群中的证书签发。最后使用 Deployment 部署我们的服务即可。

kubernetes 集群中的证书签发

场景

在我们使用 kubernetes 的 admission webhook 机制实现一些集群资源认证、修改的方案时,会涉及到集群内部的 https 通信。这就涉及到我们的服务需要配置证书,并且要让 kubernetes 的组件信任该证书。我们都知道 kubernetes 集群中所有的证书都是由一个自定义的 CA 签发的,并且 kubernetes 集群都信任该 CA,因此基于该原理,使用 kubernetes 提供的 CertificateSigningRequest 来为我们的证书签名即可。

具体流程

kubernetes 官方的文档上提供了比较详细的说明,文档地址在这里:管理集群中的 TLS 认证

大致流程如下:

  • 使用 cfssl 为我们的 service 地址创建证书
  • 使用 CertificateSigningRequest 请求 kubernetes 的 CA 来为该证书签名
  • 管理员通过 kubectl certificate approve 来批准请求
  • 通过 kubectl get csr 获取签名后的证书,并使用到我们的服务中。

当然这个过程还不够自动化,我们可以使用一些很好的脚本来帮助我们完成这个工作:

这个脚本使用 openssl 来创建公私钥,创建 CertificateSigningRequest 请求对公钥签名,然后自动批准并将私钥和签名后的公钥存到 secret 中。使用方式如下:

./webhook-create-signed-cert.sh \
    --service lazykube-webhook-svc \
    --secret lazykube-webhook-certs \
    --namespace kube-system

这个脚本其实跟证书签发没有太大关系,但是如果你使用 mutatingwebhook 的话正好可以使用它,同理 validatingwebhook 也是如此

cat mutatingwebhook.yaml | \
    ./webhook-patch-ca-bundle.sh > \
    mutatingwebhook-ca-bundle.yaml

一些问题

我在使用上述脚本的时候,发现使用 rancher 创建的 kubernetes 集群有问题。这是因为 rancher 创建的 kubernetes 集群没有默认开启 kubernetes controller manager 的签名选项。具体的选项可以参考这里:给集群管理员的一个建议

修改的方案就是打开该选项,rancher 中可以在界面上编辑集群的 yaml 文件,加上以下参数:

services:
  kube-controller: 
    extra_args: 
      cluster-signing-cert-file: "/etc/kubernetes/ssl/kube-ca.pem"
      cluster-signing-key-file: "/etc/kubernetes/ssl/kube-ca-key.pem"

kubernetes存储–FlexVolume

简介

kubernetes 使用 volume 来满足它的存储需求,它支持很多的存储系统,比如 nfs、 glusterfs、cephfs等等,但是这些存储的实现方式有一个问题,就是它们的实现代码都必须合并到 Kubernetes 的代码中(称为 in-tree),这为 kubernetes 社区带来了维护上的成本。因此,kubernetes 提出了两种 out-of-tree 的方案: FlexVolume 和 csi。通过这两种方案实现的存储功能不必合并到 kubernetes 的代码仓库,由存储系统的供应商单独维护。

FlexVolume 是这篇文章主要关注的点,FlexVolume 自 1.2 版本开始就支持了。它使用基于 exec 的模型来与驱动程序对接。用户必须在每个节点(有些情况下包括主节点)上的预定义卷插件路径中安装 FlexVolume 驱动程序的可执行文件。当需要挂载 volume 的时候,由 kubelet 执行挂载命令来挂载即可。

基于 nfs 实现 FlexVolume

在探究 FlexVolume 的实现原理之前,我们可以先看一下官方提供的基于 nfs 的例子

注: 我这里是用 minikube 启动的本地 kubernetes 集群。

为了部署基于 nfs 实现的 FlexVolume,我们首先将目录下的 nfs 复制到 deploy 文件夹下

$ cp nfs deploy

然后将 deploy/deploy.sh 中的 dummy 修改成 nfs,表示我们使用的插件脚本是 nfs 这个可执行文件。

接着在 deploy 文件夹下构建 docker 镜像,这里要修改 Dockerfile,将 nfs COPY 到镜像中。然后执行下面的命令(镜像标签需要修改成你自己的):

$ docker build -t joyme/nfs-flexvolume:1.0 .
$ docker push joyme/nfs-flexvolume:1.0

镜像构建并推送完成之后,我们就开始部署了。因为 FlexVolume 要求将驱动文件放在指定的目录下,最粗暴的方式就是手动将文件 scp 到集群的每个节点上。这里为了方便,我们还可以使用 kubernetes 的 Daemenset,然后使用 hostPath 将文件放到主机之上。我们修改 deploy 文件夹下的 ds.yaml 这个部署文件。将我们刚刚推送的镜像填进去。然后执行以下命令进行部署。

$ kubectl apply -f ds.yaml

这里有个地方要注意, 默认的插件安装地址是 /usr/libexec/kubernetes/kubelet-plugins/volume/exec/, 但是 kubelet 的参数 --volume-plugin-dir 和 controller manager 的参数 --flex-volume-plugin-dir 都可以修改这个值,如果你启动这些组件是指定了这些参数,那就需要修改 ds.yaml 中的路径。

在集群中部署完成之后,我们可以到某个节点上检查一下/usr/libexec/kubernetes/kubelet-plugins/volume/exec/是否存在我们刚刚部署的文件。

最后我们创建一个 nginx,挂载一个 FlexVolume。在创建之前,我们需要先启动一个 nfs server,这里为了方便,可以使用容器启动一个。

$ docker run -d --privileged --restart=always \
-v /tmp:/dws_nas_scratch \
-e NFS_EXPORT_DIR_1=/dws_nas_scratch \
-e NFS_EXPORT_DOMAIN_1=\* \
-e NFS_EXPORT_OPTIONS_1=ro,insecure,no_subtree_check,no_root_squash,fsid=1 \
-p 111:111 -p 111:111/udp \
-p 2049:2049 -p 2049:2049/udp \
-p 32765:32765 -p 32765:32765/udp \
-p 32766:32766 -p 32766:32766/udp \
-p 32767:32767 -p 32767:32767/udp \
fuzzle/docker-nfs-server:latest

使用官方提供的 nginx-nfs.yaml 文件,然后把其中的 server 地址修改一下,使用以下命令创建:

$ kubectl apply -f nginx-nfs.yaml

注意:如果出现错误,可以检查 node 上是否安装了 jq, nfs-common 等必要的依赖包。

实现原理

在完成上面例子的过程中,关于 FlexVolume 的大多数问题都比较好解答了。我们来看一下 nfs 的实现代码:

usage() {
    err "Invalid usage. Usage: "
    err "\t$0 init"
    err "\t$0 mount <mount dir> <json params>"
    err "\t$0 unmount <mount dir>"
    exit 1
}

err() {
    echo -ne $* 1>&2
}

log() {
    echo -ne $* >&1
}

ismounted() {
    MOUNT=`findmnt -n ${MNTPATH} 2>/dev/null | cut -d' ' -f1`
    if [ "${MOUNT}" == "${MNTPATH}" ]; then
        echo "1"
    else
        echo "0"
    fi
}

domount() {
    MNTPATH=$1

    NFS_SERVER=$(echo $2 | jq -r '.server')
    SHARE=$(echo $2 | jq -r '.share')

    if [ $(ismounted) -eq 1 ] ; then
        log '{"status": "Success"}'
        exit 0
    fi

    mkdir -p ${MNTPATH} &> /dev/null

    mount -t nfs ${NFS_SERVER}:/${SHARE} ${MNTPATH} &> /dev/null
    if [ $? -ne 0 ]; then
        err "{ \"status\": \"Failure\", \"message\": \"Failed to mount ${NFS_SERVER}:${SHARE} at ${MNTPATH}\"}"
        exit 1
    fi
    log '{"status": "Success"}'
    exit 0
}

unmount() {
    MNTPATH=$1
    if [ $(ismounted) -eq 0 ] ; then
        log '{"status": "Success"}'
        exit 0
    fi

    umount ${MNTPATH} &> /dev/null
    if [ $? -ne 0 ]; then
        err "{ \"status\": \"Failed\", \"message\": \"Failed to unmount volume at ${MNTPATH}\"}"
        exit 1
    fi

    log '{"status": "Success"}'
    exit 0
}

op=$1

if ! command -v jq >/dev/null 2>&1; then
    err "{ \"status\": \"Failure\", \"message\": \"'jq' binary not found. Please install jq package before using this driver\"}"
    exit 1
fi

if [ "$op" = "init" ]; then
    log '{"status": "Success", "capabilities": {"attach": false}}'
    exit 0
fi

if [ $# -lt 2 ]; then
    usage
fi

shift

case "$op" in
    mount)
        domount $*
        ;;
    unmount)
        unmount $*
        ;;
    *)
        log '{"status": "Not supported"}'
        exit 0
esac

exit 1

其实就是一段 shell 脚本,支持三个命令: init、mount、unmount。当我们在集群中为某个 pod 挂载 FlexVolume时,该 pod 所在节点的 kubelet 会调用其指定的插件脚本执行 mount 命令,然后挂载给 pod 使用。当然了,FlexVolume 还支持更复杂的插件。这个可以看官方的文档: flexvolume

部署方案

关于如何部署 FlexVolume 的插件,其实在例子中也有提到,这里可以总结一下:

  • 手动部署到每个节点的指定目录下,比如我们刚刚部署的 nfs ,其实际路径是: /usr/libexec/kubernetes/kubelet-plugins/volume/exec/k8s~nfs。其中 /usr/libexec/kubernetes/kubelet-plugins/volume/exec 是默认路径,也可以通过 kubelet 的参数 --volume-plugin-dir 和 controller manager 的参数 --flex-volume-plugin-dir 来指定。k8s~nfs 这个路径中,k8s 是供应商, nfs 是驱动名称,在使用的时候可以这样指定: `driver: “k8s/nfs”。

  • 使用 kubernetes 的 deamonset 配合 hostPath 来部署,因为 daemonset 会在每个节点上都启动 pod,然后通过 hostPath 将插件放在指定的位置即可。kubernetes 集群中 master 节点可能被设置成不允许调度。这种情况下 daemonset 默认不调度到 master 节点上,可以使用 tolerations 来解决这个问题. 具体可参考: Scheduler is not scheduling Pod for DaemonSet in Master node

  • 其实除了 kubelet 要调用插件之外,controller-manager 也要调用。比如执行 init, attach, detach, waitforattach, isattached 等命令。

argo的输入输出源代码分析

简介

argo是一个工作流的调度引擎,支持 Steps 和 DAG 这两种工作流。

  • Steps: 是按照步骤,从前往后的工作流调度方案。工作流中的每一步都只依赖上一步的结果
  • DAG: 全称是 directed acyclic graph,译为有向无环图。与 Steps 的区别在于每一步可能依赖之前的多步输出,但是不会循环依赖(也就是无环)

不论是在什么类型的工作流上,argo都抽象出了两种输入输出:

  • parameters: 通常情况下都是字符串,该字符串可以来源于标准输出,也可以来源于文件的内容
  • artifacts: 可以理解成文件

输入输出是连接整个工作流的核心。每一步都可以看作是一次函数调用。那么在argo中,它是如何实现在多步之间输入输出的传输呢?下面会通过源代码进行分析。

在看代码之前,可以看一个 argo 的工作流中的一个pod,为了查看更方便,我删除一些不需要关注的字段:

$ kubectl -n workflow describe pods custom-workflow-111-2fw2f-2639432629

Name:           custom-workflow-111-2fw2f-2639432629
Namespace:      workflow
Labels:         pipeline.starx.com/nodeID=743
                workflows.argoproj.io/completed=true
                workflows.argoproj.io/workflow=custom-workflow-111-2fw2f
Annotations:    cni.projectcalico.org/podIP: 10.42.0.83/32
                workflows.argoproj.io/node-name: custom-workflow-111-2fw2f.yolov3-evaluate-743
                workflows.argoproj.io/outputs:
                  {"result":...
                workflows.argoproj.io/template:
                  {"name":"yolov3-evaluate-743","inputs":{"parameters":[{"name":"userParam","value":"eyJTY29yZVRocmVzaG9sZCI6MC41LCJJb3VfVGhyZXNob2xkIjowLjQ...
Controlled By:  Workflow/custom-workflow-111-2fw2f
Init Containers:
  init:
    Image:         argoproj/argoexec:v2.3.0
    Command:
      argoexec
      init
    Environment:
      ARGO_POD_NAME:  custom-workflow-111-2fw2f-2639432629 (v1:metadata.name)
    Mounts:
      /argo/inputs/artifacts from input-artifacts (rw)
      /argo/podmetadata from podmetadata (rw)
      /argo/staging from argo-staging (rw)
      /var/run/secrets/kubernetes.io/serviceaccount from default-token-lfk5b (ro)
Containers:
  wait:
    Image:         argoproj/argoexec:v2.3.0
    Command:
      argoexec
      wait
    Environment:
      ARGO_POD_NAME:  custom-workflow-111-2fw2f-2639432629 (v1:metadata.name)
    Mounts:
      /argo/podmetadata from podmetadata (rw)
      /mainctrfs/argo/staging from argo-staging (rw)
      /mainctrfs/tmp/artifacts/artifact-input0 from input-artifacts (rw,path="artifact0")
      /mainctrfs/tmp/artifacts/artifact-input1 from input-artifacts (rw,path="artifact1")
      /var/run/docker.sock from docker-sock (ro)
      /var/run/secrets/kubernetes.io/serviceaccount from default-token-lfk5b (ro)
  main:
    Image:         registry.cn-shanghai.aliyuncs.com/xinhuodev/wt:0.4
    Command:
      sh
    Args:
      /argo/staging/script
    Mounts:
      /argo/staging from argo-staging (rw)
      /tmp/artifacts/artifact-input0 from input-artifacts (rw,path="artifact0")
      /tmp/artifacts/artifact-input1 from input-artifacts (rw,path="artifact1")
Volumes:
  podmetadata:
    Type:  DownwardAPI (a volume populated by information about the pod)
    Items:
      metadata.annotations -> annotations
  docker-sock:
    Type:          HostPath (bare host directory volume)
    Path:          /var/run/docker.sock
    HostPathType:  Socket
  input-artifacts:
    Type:       EmptyDir (a temporary directory that shares a pod's lifetime)
    Medium:     
    SizeLimit:  <unset>
  argo-staging:
    Type:       EmptyDir (a temporary directory that shares a pod's lifetime)
    Medium:     
    SizeLimit:  <unset>
  default-token-lfk5b:
    Type:        Secret (a volume populated by a Secret)
    SecretName:  default-token-lfk5b
    Optional:    false

我们需要关注的信息有:

  • Pod 的 Annotations
  • Init Containers 启动的初始化容器
  • Containers 中的 wait 容器和 main 容器
  • Pod 的 Volumes 和每个容器的 Mounts

Init 容器

argo 创建的 Pod 的初始化容器执行了 argoexec init 命令,从名字上可以猜测出,这个容器负责初始化 Pod 中的环境,比如获取来上一步的输入等等,对应的代码是 cmd/argoexec/commands/init.go, 我们的分析也从这里开始。在执行 argo exec init之后,第一个调用的函数应该是loadArtifacts()。这个方法中做了三件事: initExecutor()wfExecutor.StageFiles()wfExecutor.LoadArtifacts()

initExecutor:

initExecutor 的代码如下(删除了不重要的代码):

func initExecutor() *executor.WorkflowExecutor {
    tmpl, err := executor.LoadTemplate(podAnnotationsPath)

    var cre executor.ContainerRuntimeExecutor
    switch os.Getenv(common.EnvVarContainerRuntimeExecutor) {
    case common.ContainerRuntimeExecutorK8sAPI:
        cre, err = k8sapi.NewK8sAPIExecutor(clientset, config, podName, namespace)
    case common.ContainerRuntimeExecutorKubelet:
        cre, err = kubelet.NewKubeletExecutor()
    case common.ContainerRuntimeExecutorPNS:
        cre, err = pns.NewPNSExecutor(clientset, podName, namespace, tmpl.Outputs.HasOutputs())
    default:
        cre, err = docker.NewDockerExecutor()
    }

    wfExecutor := executor.NewExecutor(clientset, podName, namespace, podAnnotationsPath, cre, *tmpl)
    yamlBytes, _ := json.Marshal(&wfExecutor.Template)
    return &wfExecutor
}

podAnnotationsPath加载模板,这个模板其实就是 Argo 中单步的执行模板,默认情况下它的值是 /argo/podmetadata/annotations,这正好是 init 容器的挂载,而这个挂载对应的卷是:

 podmetadata:
    Type:  DownwardAPI (a volume populated by information about the pod)
    Items:
      metadata.annotations -> annotations

这里的 DownwardAPI 也解释一下,它是一种 volume 的类型,可以将 Pod 和 Container 的字段通过挂载文件的方式提供给容器内的进程方案。那么这里就是将 Pod 的 Annotations 字段通过上面的路径提供给 init 容器,init 容器根据其中的 template 获取该 Pod 的输入输出。

接下来判断根据容器运行时进行判断,这里我们只考虑 docker 作为容器运行时的情况。最后调用NewExecutor实例化了一个 wfExecutor

StageFiles()

源代码如下:

func (we *WorkflowExecutor) StageFiles() error {
    var filePath string
    var body []byte
    switch we.Template.GetType() {
    case wfv1.TemplateTypeScript:
        log.Infof("Loading script source to %s", common.ExecutorScriptSourcePath)
        filePath = common.ExecutorScriptSourcePath
        body = []byte(we.Template.Script.Source)
    case wfv1.TemplateTypeResource:
        log.Infof("Loading manifest to %s", common.ExecutorResourceManifestPath)
        filePath = common.ExecutorResourceManifestPath
        body = []byte(we.Template.Resource.Manifest)
    default:
        return nil
    }
    err := ioutil.WriteFile(filePath, body, 0644)
    if err != nil {
        return errors.InternalWrapError(err)
    }
    return nil
}

职责很简单,根据 template 的类型,写入到不同的文件中,比如 script 就写入到 /argo/staging/script。这就是我们在 main 容器中执行的脚本了。

LoadArtifacts

// LoadArtifacts loads artifacts from location to a container path
func (we *WorkflowExecutor) LoadArtifacts() error {
    for _, art := range we.Template.Inputs.Artifacts {
        artDriver, err := we.InitDriver(art)

        var artPath string
        mnt := common.FindOverlappingVolume(&we.Template, art.Path)
        if mnt == nil {
            artPath = path.Join(common.ExecutorArtifactBaseDir, art.Name)
        } else {
            // If we get here, it means the input artifact path overlaps with an user specified
            // volumeMount in the container. Because we also implement input artifacts as volume
            // mounts, we need to load the artifact into the user specified volume mount,
            // as opposed to the `input-artifacts` volume that is an implementation detail
            // unbeknownst to the user.
            log.Infof("Specified artifact path %s overlaps with volume mount at %s. Extracting to volume mount", art.Path, mnt.MountPath)
            artPath = path.Join(common.ExecutorMainFilesystemDir, art.Path)
        }

        // The artifact is downloaded to a temporary location, after which we determine if
        // the file is a tarball or not. If it is, it is first extracted then renamed to
        // the desired location. If not, it is simply renamed to the location.
        tempArtPath := artPath + ".tmp"
        err = artDriver.Load(&art, tempArtPath)
        if err != nil {
            return err
        }
        if isTarball(tempArtPath) {
            err = untar(tempArtPath, artPath)
            _ = os.Remove(tempArtPath)
        } else {
            err = os.Rename(tempArtPath, artPath)
        }

        if art.Mode != nil {
            err = os.Chmod(artPath, os.FileMode(*art.Mode))
        }
    }
    return nil
}

InitDriver是初始化 Artifacts 的驱动。Argo 支持多种类型的存储系统,在 v2.3.0 这个版本支持: s3, http, git, artifactory, hdfs, raw。

FindOverlappingVolume 是检查 artifacts 的路径和用户挂载的路径是否有重合。如果有,则返回深度最深的路径,如果没有,则返回 nil。如果返回 nil, 则使用 /argo/inputs/artifacts 作为 artifacts 的基础路径。否则使用 /mainctrfs 作为路径。

下面就是下载文件,解压文件并修改权限了。

注意在这里,init、wait和main容器都挂载了input-artifactsargo-staging,并且 init 将输入和script放在了这两个卷中,所以其他几个卷都可以共享这些文件。

wait 容器

wait容器的职责有以下几点:

  • 等待 main 容器结束
  • 杀死 sidecar
  • 保存日志
  • 保存 parameters
  • 保存 artifacts
  • 获取脚本的输出流
  • 将输出放在 Annotations 上

下面我们看这些功能点的实现:

等待 main 容器结束

// Wait is the sidecar container logic which waits for the main container to complete.
// Also monitors for updates in the pod annotations which may change (e.g. terminate)
// Upon completion, kills any sidecars after it finishes.
func (we *WorkflowExecutor) Wait() error {
    // WaitInit() 是初始化操作,只有 PSN 需要
    err := we.RuntimeExecutor.WaitInit()
    if err != nil {
        return err
    }
    log.Infof("Waiting on main container")
    // waitMainContainerStart的主要原理是周期轮询Pod中的所有容器,检查main容器的ContainerID字段
    // 不为空说明启动了
    mainContainerID, err := we.waitMainContainerStart()
    if err != nil {
        return err
    }
    log.Infof("main container started with container ID: %s", mainContainerID)
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // monitorAnnotations是因为pod的annotations会更改
    annotationUpdatesCh := we.monitorAnnotations(ctx)
    // 超时会杀死
    go we.monitorDeadline(ctx, annotationUpdatesCh)

    // 这里是直接用ContainerRuntime去等待容器结束的,比如docker,直接调用docker wait
    err = we.RuntimeExecutor.Wait(mainContainerID)
    if err != nil {
        return err
    }
    log.Infof("Main container completed")
    return nil
}

杀死 sidecar

main 容器运行结束后,wait 容器会负责杀死其他容器(这个让我发现了之前用 sidecar 做 main 容器运行结束后的清理工作一直无效的原因)。

// KillSidecars kills any sidecars to the main container
func (we *WorkflowExecutor) KillSidecars() error {
    if len(we.Template.Sidecars) == 0 {
        log.Infof("No sidecars")
        return nil
    }
    log.Infof("Killing sidecars")
    pod, err := we.getPod()
    if err != nil {
        return err
    }
    sidecarIDs := make([]string, 0)
    // 遍历pod中的容器,排除main和wait,然后调用runtime来杀死容器
    for _, ctrStatus := range pod.Status.ContainerStatuses {
        if ctrStatus.Name == common.MainContainerName || ctrStatus.Name == common.WaitContainerName {
            continue
        }
        if ctrStatus.State.Terminated != nil {
            continue
        }
        containerID := containerID(ctrStatus.ContainerID)
        log.Infof("Killing sidecar %s (%s)", ctrStatus.Name, containerID)
        sidecarIDs = append(sidecarIDs, containerID)
    }
    if len(sidecarIDs) == 0 {
        return nil
    }
    return we.RuntimeExecutor.Kill(sidecarIDs)
}

保存日志

argo 是支持将 main 容器中的日志持久化并保存到指定的地方的(s3, hdfs, Artifactory)。这在 argo 的文档上好像没有提到过。这一部分的逻辑比较简单,就是通过 ContainerRuntime 获取获取容器中的输出流,然后存成文件,通过 argo 中的 storage driver 保存下来。

保存 parameters

// SaveParameters will save the content in the specified file path as output parameter value
func (we *WorkflowExecutor) SaveParameters() error {
    if len(we.Template.Outputs.Parameters) == 0 {
        log.Infof("No output parameters")
        return nil
    }
    log.Infof("Saving output parameters")
    mainCtrID, err := we.GetMainContainerID()
    if err != nil {
        return err
    }

    // 遍历模板参数
    for i, param := range we.Template.Outputs.Parameters {
        log.Infof("Saving path output parameter: %s", param.Name)
        // Determine the file path of where to find the parameter
        if param.ValueFrom == nil || param.ValueFrom.Path == "" {
            continue
        }

        var output string
        if we.isBaseImagePath(param.ValueFrom.Path) {
            log.Infof("Copying %s from base image layer", param.ValueFrom.Path)
            // 容器内,通过 runtime 获取
            output, err = we.RuntimeExecutor.GetFileContents(mainCtrID, param.ValueFrom.Path)
            if err != nil {
                return err
            }
        } else {
            log.Infof("Copying %s from from volume mount", param.ValueFrom.Path)
            mountedPath := filepath.Join(common.ExecutorMainFilesystemDir, param.ValueFrom.Path)
            // 容器的挂载卷,直接获取
            out, err := ioutil.ReadFile(mountedPath)
            if err != nil {
                return err
            }
            output = string(out)
        }

        outputLen := len(output)
        // Trims off a single newline for user convenience
        if outputLen > 0 && output[outputLen-1] == '\n' {
            output = output[0 : outputLen-1]
        }
        // 保存下来
        we.Template.Outputs.Parameters[i].Value = &output
        log.Infof("Successfully saved output parameter: %s", param.Name)
    }
    return nil
}

保存 artifacts

保存 artifacts 和 保存 parameters 的操作是一样的。

// SaveArtifacts uploads artifacts to the archive location
func (we *WorkflowExecutor) SaveArtifacts() error {
    if len(we.Template.Outputs.Artifacts) == 0 {
        log.Infof("No output artifacts")
        return nil
    }
    log.Infof("Saving output artifacts")
    mainCtrID, err := we.GetMainContainerID()
    if err != nil {
        return err
    }

    err = os.MkdirAll(tempOutArtDir, os.ModePerm)
    if err != nil {
        return errors.InternalWrapError(err)
    }

    for i, art := range we.Template.Outputs.Artifacts {
        err := we.saveArtifact(mainCtrID, &art)
        if err != nil {
            return err
        }
        we.Template.Outputs.Artifacts[i] = art
    }
    return nil
}

获取脚本的输出流

直接调用 runtime 去获取 main 容器的输出流,然后保存到 template.outputs 中

func (we *WorkflowExecutor) CaptureScriptResult() error {
    if we.Template.Script == nil {
        return nil
    }
    log.Infof("Capturing script output")
    mainContainerID, err := we.GetMainContainerID()
    if err != nil {
        return err
    }
    reader, err := we.RuntimeExecutor.GetOutputStream(mainContainerID, false)
    if err != nil {
        return err
    }
    defer func() { _ = reader.Close() }()
    bytes, err := ioutil.ReadAll(reader)
    if err != nil {
        return errors.InternalWrapError(err)
    }
    out := string(bytes)
    // Trims off a single newline for user convenience
    outputLen := len(out)
    if outputLen > 0 && out[outputLen-1] == '\n' {
        out = out[0 : outputLen-1]
    }
    we.Template.Outputs.Result = &out
    return nil
}

将输出放在 Annotations 上

将 outputs 存在 pod 的 annotations 上。

func (we *WorkflowExecutor) AnnotateOutputs(logArt *wfv1.Artifact) error {
    outputs := we.Template.Outputs.DeepCopy()
    if logArt != nil {
        outputs.Artifacts = append(outputs.Artifacts, *logArt)
    }

    if !outputs.HasOutputs() {
        return nil
    }
    log.Infof("Annotating pod with output")
    outputBytes, err := json.Marshal(outputs)
    if err != nil {
        return errors.InternalWrapError(err)
    }
    return we.AddAnnotation(common.AnnotationKeyOutputs, string(outputBytes))
}

总结

init 容器做了 pod 的初始化,包括存储 script,下载 artifacts等等,这样我们的 main 容器就不用关心输入的来源,只需要在指定地方使用即可。wait 容器负责监控 main 容器的生命周期,在 main 容器中的主要逻辑运行结束之后,负责将输出部分读取,持久化,这样 main 容器就不用操心如何将该步产生的结果传到后面的步骤上的问题。