摘要：這樣就會有一個問題一個任務(wù)要起個，每個需要一張卡，總共需要張卡，而集群中只有張空閑的卡，這樣默認(rèn)的調(diào)度器會如何處理因?yàn)槟J(rèn)調(diào)度器是一個一個調(diào)度的，只會檢查單個資源夠不夠，這樣前個都能成功，最后一個調(diào)度失敗。

kubernetes集群三步安裝

深度學(xué)習(xí)中經(jīng)常會出現(xiàn)多機(jī)多卡的任務(wù)，也就是同事會起多個pod，但是這多個pod屬于同一個任務(wù)。

這樣就會有一個問題

一個任務(wù)要起100個pod，每個pod需要一張卡，總共需要100張GPU卡，而集群中只有99張空閑的GPU卡，這樣默認(rèn)的k8s調(diào)度器會如何處理？

因?yàn)槟J(rèn)調(diào)度器是一個一個pod調(diào)度的，只會檢查單個pod資源夠不夠，這樣前99個都能成功，最后一個pod調(diào)度失敗。

這樣非常有可能造成

任務(wù)跑不了

前99個占著GPU不釋放，新的任務(wù)無法調(diào)度

嚴(yán)重時整個集群死鎖，都“占著茅坑不拉屎”

所以需要在調(diào)度時對整個task所需所有資源進(jìn)行檢查，當(dāng)集群總體資源不夠時，一個pod都得不到調(diào)度。

社區(qū)提供了一個能支持這種特性的調(diào)度器
但是這個調(diào)度器是沒辦法和原生調(diào)度器很好的配合工作的

最大的問題在于兩個調(diào)度器都有cache，這樣cache的內(nèi)容會沖突，導(dǎo)致調(diào)度混亂

這個調(diào)度器沒法和原生調(diào)度器同時起作用，這樣用了這個batch調(diào)度器后就沒法用親和性什么的特性了

所以我們做的事是將兩者特性融合，選擇的方法是定制化開發(fā)kube-scheduler

其實(shí)scheduler是可以通過extender擴(kuò)展的，但是extender還是太弱了，它僅能在預(yù)選和優(yōu)選過程中加入自己的過濾策略，而這對于批處理任務(wù)遠(yuǎn)遠(yuǎn)不夠。

需要優(yōu)選時加batch任務(wù)檢查
拿到一個pod ---> 如果是一個batchpod ---> 查詢集群資源是否滿足batch任務(wù)--->否調(diào)度失敗
需要保障batch任務(wù)中其它pod能得到調(diào)度

如果集群資源能滿足這個batch任務(wù)直接去bind有個問題：
假設(shè)調(diào)度隊(duì)列是這樣,假設(shè)集群中有三個GPU，而batch任務(wù)需要三個GPU：

所以最終結(jié)果是A批任務(wù)占用了一個GPU但是整個任務(wù)是調(diào)度失敗的，那一個GPU還得不到釋放

所以需要修改pod調(diào)度隊(duì)列里的順序?讓A batch pod連續(xù)調(diào)度? 沒這么簡單，

pod調(diào)度是創(chuàng)建協(xié)程并發(fā)調(diào)度的，這樣即便去調(diào)整任務(wù)隊(duì)列里pod的順序也不一定能保證batch任務(wù)其它pod能得到優(yōu)先調(diào)度。

go wait.Until(sched.scheduleOne, 0, sched.config.StopEverything)

只要batch pod走到Bind邏輯了就沒有回頭路了

batch任務(wù)中所有pod先進(jìn)行assume調(diào)度，其中任意一個失敗就清理掉其它已經(jīng)bind但是還沒實(shí)際進(jìn)行調(diào)度的pod。 并把所有pod扔回隊(duì)列，或者直接返回調(diào)度失敗清理改任務(wù)的pod，讓上層重新觸發(fā)?

scheduler流程 scheduler/sheduler.go scheduleOne邏輯：

選節(jié)點(diǎn)->cache assume pod on node-> 創(chuàng)建協(xié)程bind

所以在assume時去檢查，不滿足退還已經(jīng)調(diào)度的pod是不可行的，因?yàn)橹癰atch任務(wù)中的pod可能已經(jīng)bind過了， 所以只能batch任務(wù)中最后一個pod得到確認(rèn)才能去bind前面的pod

預(yù)占用策略
預(yù)占用策略： 第一個batch pod任務(wù)來時，檢查集群資源是不是夠，如果夠進(jìn)行預(yù)占，把其它幾個node打上標(biāo)記，讓接下來pod無法占用其它的node，這樣batch任務(wù)其實(shí)pod過來就有節(jié)點(diǎn)可用。

回到了不能bind的問題。。。

這種問題有兩點(diǎn)：

如何知道batch任務(wù)中其它pod需要什么樣的節(jié)點(diǎn)，如果pod都一樣問題可簡化
如果后面的pod失敗了，第一個pod還是已經(jīng)bind，還是會出現(xiàn)一樣的問題
最終還是在所有pod assume之前不能bind單個pod

綜上，需要在幾個地方處理

隊(duì)列最好用優(yōu)先級隊(duì)列，把正在調(diào)度的pod的關(guān)聯(lián)pod優(yōu)先級提高
選節(jié)點(diǎn)時做判斷，看集群資源是否夠
選好節(jié)點(diǎn)assume pod時檢查，如果自己不夠或者pod組不夠就不去bind
問題是之前的pod已經(jīng)走了bind流程，所以最重要的是如何解決讓之前的pod不去bind，延遲bind

最終方案 - 延遲綁定

方案：在batch任務(wù)bind時進(jìn)行特殊處理

如果是batch任務(wù)扔進(jìn)task cache，不進(jìn)行binding

如果batch任務(wù)最后一個pod扔進(jìn)task cache,該task ready，放進(jìn)bind隊(duì)列

在bind隊(duì)列里取task 進(jìn)行bind，task互斥鎖與普通pod bind時互斥

使用
batch任務(wù)使用，pod增加兩個注解：

      annotations:
        scheduling.k8s.io/group-name: qj-1
        scheduling.k8s.io/group-pod-num: 3

pod加上這兩個注解表示屬于同一個task, num表示task里有多少pod。

本來是再定義一個CRD去描述這個task，耦合會小一些，但是實(shí)現(xiàn)麻煩些，需要多監(jiān)聽一個CRD，偷懶就沒這樣做

延遲綁定流程：

如果是普通的pod，找到節(jié)點(diǎn)后assume就直接bind

如果是批處理任務(wù)，直接扔到批處理緩存中返回

有個協(xié)程一直檢查批緩存中是否有成功的task (pod都齊了)

成功的task扔進(jìn)binding隊(duì)列，worker取成功的task進(jìn)行批量綁定，綁定時與普通pod互斥

batch scheduler接口與成員

Run 起一個協(xié)程檢查成功的task并塞入隊(duì)列
RunBind 起一個task綁定協(xié)程
PodQuePriority 去動態(tài)修改pod隊(duì)列的優(yōu)先級，讓同task的pod優(yōu)先調(diào)度

執(zhí)行流程：

scheduler/scheduler.go:

    //fanux if it is a batch pod, return
    if sched.Config.BatchScheduler.IsBatchPod(assumedPod) {
        err = sched.Config.BatchScheduler.HandleBatchPod(assumedPod)
        if err != nil {
            glog.Errorf("schedule batch pod failed: %v", assumedPod.Namespace, assumedPod.Name)
        }
        return
    }

增加綁定互斥，防止batch任務(wù)和普通pod同事binding:

    go func() {
        //fanux add bind mutex
        sched.Config.BatchScheduler.Lock()
        defer sched.Config.BatchScheduler.UnLock()

        err := sched.bind(assumedPod, &v1.Binding{

should"t use filterFunc, needs nodelist

scheduler/util/batch.go

package util

import "api/core/v1"

//CheckResourceIsEnough is
func CheckResourceIsEnough(pod *v1.Pod, nodes []*v1.Node) (bool, error) {
    return false, nil
}

scheduler/core/generic_scheduler.go

    //fanux add checkBatchPodResource
    flag, err := util.CheckResourceIsEnough(pod, filteredNodes)
    if !flag || err != nil {
        return "", err
    }

    trace.Step("Prioritizing")

處理資源不足時的情況

    suggestedHost, err := sched.schedule(pod)

    //fanux add handle if resource not enough
    if strings.Contains(err.Error(), common.BatchResourceNotEnough) {
        sched.Config.BatchScheduler.HandleResourceNotEnough(pod)
    } else if err != nil {

nodeInfo allocatableResource - requestedResource is avaliavle resource

    requestedResource *Resource
    nonzeroRequest    *Resource
    allocatableResource *Resource

GPU 是 ScalarResources, 資源名稱叫 : NVIDIAGPUResourceName = "nvidia.com/gpu"

type Resource struct {
    MilliCPU         int64
    Memory           int64
    EphemeralStorage int64
    // We store allowedPodNumber (which is Node.Status.Allocatable.Pods().Value())
    // explicitly as int, to avoid conversions and improve performance.
    AllowedPodNumber int
    // ScalarResources
    ScalarResources map[v1.ResourceName]int64
}

    batchScheduler := batch.NewBatchScheduler(c.schedulerCache, c.podQueue, &binder{c.client}, &podConditionUpdater{c.client})

需要知道已經(jīng)有哪些pod已經(jīng)assume過了，把這個數(shù)量減掉才是batch任務(wù)還需要多少GPU

core/generic_scheduler.go

    //fanux add batch Cache
    //check batch pod resource is enough need batch scheduler cache
    BatchCache common.TaskCache

    //fanux add checkBatchPodResource
    flag, err := common.CheckResourceIsEnough(pod, filteredNodes, g.cachedNodeInfoMap, g.BatchCache)

factory.go

    //fanux check batch resource is enough need batch scheduler cache
    batchCache := batchScheduler.GetTaskCache()

    algo := core.NewGenericScheduler(
        ...
        batchCache,
    )

then checkresource :

    //shoud not use metadata, need use metadata - assumed pod num in batch cache
    _, podNum := GetPodBathMeta(pod)
    podNum -= batchCache.GetTaskAssumedPodNum(pod)

有很多細(xì)節(jié)

//獲取pod需要多少GPU，這個需要把pod里容器配額加起來
func GetPodGPUCount(pod *v1.Pod) (count int) {
    for _, c := range pod.Spec.Containers {
        limit, ok := c.Resources.Limits[NVIDIAGPUResourceName]
        l, okay := limit.AsInt64()
        if !ok || !okay {
            continue
        }
        count += int(l)
    }

    glog.Infof("Pod [%s] need GPU [%d]", pod.GetName(), count)

    return
}

//獲取節(jié)點(diǎn)空閑GPU，需要把可分配的減去已經(jīng)申請的
func GetNodeFreeGPU(nodeInfo *cache.NodeInfo) int {
    if nodeInfo == nil {
        return 0
    }

    allocatable, ok := nodeInfo.AllocatableResource().ScalarResources[NVIDIAGPUResourceName]
    if !ok {
        glog.Errorf("can"t fetch allocatable GPU : %v", nodeInfo)
        return 0
    }
    glog.Infof("node [%s] allocatable GPU [%d]", nodeInfo.Node().Name, allocatable)

    requested, ok := nodeInfo.RequestedResource().ScalarResources[NVIDIAGPUResourceName]
    if !ok {
        //glog.Errorf("can"t fetch requested GPU : %v", nodeInfo)
        //return 0
        requested = 0
    }
    glog.Infof("node [%s] requested GPU [%d]", nodeInfo.Node().Name, requested)

    available := allocatable - requested

    glog.Infof("available node [%s] GPU : [%d]", nodeInfo.Node().Name, available)

    return int(available)
}

//這里最關(guān)鍵的點(diǎn)是需要把a(bǔ)nnotations里面獲取的task pod總數(shù)減去已經(jīng)assume過的batch pod，這樣才是真實(shí)所需
func CheckResourceIsEnough(pod *v1.Pod, nodes []*v1.Node, cachedNodeInfoMap map[string]*cache.NodeInfo, batchCache TaskCache) (bool, error) {
    //if is not batch pod, return true,nil
    if !IsBatch(pod) {
        glog.Infof("pod %s is not batch pod", pod.GetName())
        return true, nil
    }

    //shoud not use metadata, need use metadata - ready pod num in batch cache
    _, podNum := GetPodBathMeta(pod)
    podNum -= batchCache.GetTaskAssumedPodNum(pod)

    everyPodNeedsGPU := GetPodGPUCount(pod)
    if everyPodNeedsGPU == 0 {
        glog.Infof("pod %s require 0 GPU", pod.GetName())
        return true, nil
    }

    // TODO maybe check nodes[1:], node[0] already allocate a pod, CPU and other metric may reach limit
    for _, node := range nodes {
        nodeInfo, ok := cachedNodeInfoMap[node.Name]
        if !ok {
            continue
        }
        nodeFree := GetNodeFreeGPU(nodeInfo)
        podNum -= nodeFree / everyPodNeedsGPU
        glog.Infof("pod: [%s] node: [%s] podNum [%d] nodeFree [%d] podNeed [%d]", pod.GetName(), node.Name, podNum, nodeFree, everyPodNeedsGPU)
        if podNum <= 0 {
            return true, nil
        }
    }

    return false, fmt.Errorf("BatchResourceNotEnough : pod name is %s", pod.GetName())
}

//判斷是不是batch pod
func IsBatch(pod *v1.Pod) bool {
    g, n := GetPodBathMeta(pod)
    if g == "" || n == 0 {
        glog.Infof("The pod"s group name is empty string,pod name is %v.", pod.GetName())
        return false
    }
    return true
}

資源包

這里包含docker nv-docker GPU-device plugin
install.sh...

/etc/docker/daemon.json

[root@compute-gpu006 ~]# cat /etc/docker/daemon.json
{
    "default-runtime":"nvidia",
    "runtimes": {
        "nvidia": {
            "path": "/usr/bin/nvidia-container-runtime",
            "runtimeArgs": []
        }
    }
}

kubectl describe node xxx:

Capacity:
 cpu:                72
 ephemeral-storage:  222779Mi
 hugepages-1Gi:      0
 hugepages-2Mi:      2Gi
 memory:             791014684Ki
 nvidia.com/gpu:     2                # 這里就能看到GPU了
 pods:               110
Allocatable:
 cpu:                72
 ephemeral-storage:  210240641086
 hugepages-1Gi:      0
 hugepages-2Mi:      2Gi
 memory:             788815132Ki
 nvidia.com/gpu:     2
 pods:               110

原生調(diào)度器的設(shè)計就是pod one by one，所以做這個功能的開發(fā)還是改動非常大的，也是比較困難的，工作量不大，但是需要找到一個優(yōu)雅的方案，
合理的架構(gòu)比較麻煩,想了很久做了這個侵入不太大的實(shí)現(xiàn)方案，歡迎大家一起討論