网站开发和合同范本,卫生间装修效果图,临清做网站,可以显示一张图片的网站怎么搭建kubernetes scheduler 浅析
什么是kubernetes scheduler#xff1f;
小到运行着几十个工作负载的 kubernetes 集群#xff0c;大到运行成千上万个工作负载 kubernetes 集群#xff0c;每个工作负载到底应该在哪里运行#xff0c;这需要一个聪明的大脑进行指挥#xff0c…kubernetes scheduler 浅析
什么是kubernetes scheduler
小到运行着几十个工作负载的 kubernetes 集群大到运行成千上万个工作负载 kubernetes 集群每个工作负载到底应该在哪里运行这需要一个聪明的大脑进行指挥kubernetes scheduler 就是这个聪明的大脑。从结果看他的工作很简单只是为 pod.spec.nodeName 填充上一个 node 的名字而已从过程看他又是极其复杂的因为到底要选哪个节点才最合理答案往往是和场景强相关的几乎找不到一套适应各种场景的调度算法。因此各式各样的算法插件也层出不穷公司对调度算法的定制化开发也成了常见需求。
调度器如何运行的
一个调度器主要是由这样两个大循环构成 第一个控制循环负责从 etcd 里读取未被调度的 pod添加到调度队列。第二个控制循环的主要逻辑就是不断地从调度队列里出队一个 pod。然后调用 Predicates 算法进行“过滤”。这一步“过滤”得到的一组 Node就是所有可以运行这个 Pod 的宿主机列表。接下来调度器就会再调用 Priorities 算法为上述列表里的 Node 打分分数从 0 到 100。得分最高的 Node就会作为这次调度的结果。调度算法执行完成后调度器就需要将 Pod 对象的 nodeName 字段的值修改为上述 Node 的名字。
调度器是如何保障同一个控制器的不同pod副本尽量不要在同一个node上的
结合我们上面分析的调度器的结构很容易可以联想到调度器应该就是在打分阶段根据节点的不同情况是否存在相同副本存在几个进行打分来保证尽量分散 pod 的。没错调度器确实就是这么做的不过这个打分的过程又被分成了三个阶段我们可以通过分析这部分的代码来了解其原理。 源码传送门
第一阶段PreScore
func (pl *SelectorSpread) PreScore(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, nodes []*v1.Node) *framework.Status {if skipSelectorSpread(pod) {return nil}var selector labels.Selectorselector helper.DefaultSelector(pod,pl.services,pl.replicationControllers,pl.replicaSets,pl.statefulSets,)state : preScoreState{selector: selector,}cycleState.Write(preScoreStateKey, state)return nil
}
复制代码
func DefaultSelector(pod *v1.Pod, sl corelisters.ServiceLister, cl corelisters.ReplicationControllerLister, rsl appslisters.ReplicaSetLister, ssl appslisters.StatefulSetLister) labels.Selector {labelSet : make(labels.Set)// Since services, RCs, RSs and SSs match the pod, they wont have conflicting// labels. Merging is safe.if services, err : GetPodServices(sl, pod); err nil {for _, service : range services {labelSet labels.Merge(labelSet, service.Spec.Selector)}}if rcs, err : cl.GetPodControllers(pod); err nil {for _, rc : range rcs {labelSet labels.Merge(labelSet, rc.Spec.Selector)}}selector : labels.NewSelector()if len(labelSet) ! 0 {selector labelSet.AsSelector()}if rss, err : rsl.GetPodReplicaSets(pod); err nil {for _, rs : range rss {if other, err : metav1.LabelSelectorAsSelector(rs.Spec.Selector); err nil {if r, ok : other.Requirements(); ok {selector selector.Add(r...)}}}}if sss, err : ssl.GetPodStatefulSets(pod); err nil {for _, ss : range sss {if other, err : metav1.LabelSelectorAsSelector(ss.Spec.Selector); err nil {if r, ok : other.Requirements(); ok {selector selector.Add(r...)}}}}return selector
}
复制代码
在方法签名里的 cycleState 作为调度不同阶段传递数据的一个仓库这一阶段做的事情并不多就是把所有包含 pod label 的控制器找出来并保存到 cycleState供下一阶段使用。需要注意的是 pod 的控制器必须是 Services、ReplicaSets 和 StatefulSetsreplicationControllers 其中之一。
第二阶段Score
func (pl *SelectorSpread) Score(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {if skipSelectorSpread(pod) {return 0, nil}c, err : state.Read(preScoreStateKey)if err ! nil {return 0, framework.NewStatus(framework.Error, fmt.Sprintf(Error reading %q from cycleState: %v, preScoreStateKey, err))}s, ok : c.(*preScoreState)if !ok {return 0, framework.NewStatus(framework.Error, fmt.Sprintf(%v convert to tainttoleration.preScoreState error, c))}nodeInfo, err : pl.sharedLister.NodeInfos().Get(nodeName)if err ! nil {return 0, framework.NewStatus(framework.Error, fmt.Sprintf(getting node %q from Snapshot: %v, nodeName, err))}count : countMatchingPods(pod.Namespace, s.selector, nodeInfo)return int64(count), nil
}
复制代码
func skipSelectorSpread(pod *v1.Pod) bool {return len(pod.Spec.TopologySpreadConstraints) ! 0
}
复制代码
func countMatchingPods(namespace string, selector labels.Selector, nodeInfo *framework.NodeInfo) int {if len(nodeInfo.Pods) 0 || selector.Empty() {return 0}count : 0for _, p : range nodeInfo.Pods {// Ignore pods being deleted for spreading purposes// Similar to how it is done for SelectorSpreadPriorityif namespace p.Pod.Namespace p.Pod.DeletionTimestamp nil {if selector.Matches(labels.Set(p.Pod.Labels)) {count}}}return count
}
复制代码
首先判断了 pod 是否设置了拓扑约束如果设置了直接就得0分。如果没设置就交给一个叫做 countMatchingPods 的函数来计算分数。计算方法就是看在一个 node 上同命名空间下并且不是处于删除状态的 pod与当前要调度的 Pod 的 label 相同的有多少个1个1分累积起来为当前节点的分。
第三阶段NormalizeScore
这一阶段的作用为对 Score 阶段的得分做一个修正修正后的结果才会作为节点的最终得分。具体代码如下。
func (pl *SelectorSpread) NormalizeScore(ctx context.Context, state *framework.CycleState, pod *v1.Pod, scores framework.NodeScoreList) *framework.Status {if skipSelectorSpread(pod) {return nil}countsByZone : make(map[string]int64, 10)maxCountByZone : int64(0)maxCountByNodeName : int64(0)for i : range scores {if scores[i].Score maxCountByNodeName {maxCountByNodeName scores[i].Score}nodeInfo, err : pl.sharedLister.NodeInfos().Get(scores[i].Name)if err ! nil {return framework.NewStatus(framework.Error, fmt.Sprintf(getting node %q from Snapshot: %v, scores[i].Name, err))}zoneID : utilnode.GetZoneKey(nodeInfo.Node())if zoneID {continue}countsByZone[zoneID] scores[i].Score}for zoneID : range countsByZone {if countsByZone[zoneID] maxCountByZone {maxCountByZone countsByZone[zoneID]}}haveZones : len(countsByZone) ! 0maxCountByNodeNameFloat64 : float64(maxCountByNodeName)maxCountByZoneFloat64 : float64(maxCountByZone)MaxNodeScoreFloat64 : float64(framework.MaxNodeScore)for i : range scores {// initializing to the default/max node score of maxPriorityfScore : MaxNodeScoreFloat64if maxCountByNodeName 0 {fScore MaxNodeScoreFloat64 * (float64(maxCountByNodeName-scores[i].Score) / maxCountByNodeNameFloat64)}// If there is zone information present, incorporate itif haveZones {nodeInfo, err : pl.sharedLister.NodeInfos().Get(scores[i].Name)if err ! nil {return framework.NewStatus(framework.Error, fmt.Sprintf(getting node %q from Snapshot: %v, scores[i].Name, err))}zoneID : utilnode.GetZoneKey(nodeInfo.Node())if zoneID ! {zoneScore : MaxNodeScoreFloat64if maxCountByZone 0 {zoneScore MaxNodeScoreFloat64 * (float64(maxCountByZone-countsByZone[zoneID]) / maxCountByZoneFloat64)}fScore (fScore * (1.0 - zoneWeighting)) (zoneWeighting * zoneScore)}}scores[i].Score int64(fScore)}return nil
}
复制代码
首先还是判断了 pod 是否设置了拓扑约束如果设置了得分为0这一点和上一阶段完全相同。然后找出得分最高的节点和 zone节点的最终得分为100 *最大节点得分-当前节点得分/ 最大节点得分所以当前节点得分越高最终得分就越低这是符合尽量分散的预期的。如果是 zone 的情况 当前 zone 的得分 100*(最大 zone 的得分-当前 zone 的得分)/最大 zone 的得分最终节点维度得分 上一步的节点得分 * (1- zone 维度的权重) zone 维度的得分* zone 维度的权重
什么是zone
这个概念在调度的过程中是为了适配某几个节点视为一个整体的情况比如 nodeA 和 nodeB 同属于一个 zone, 那么在调度算法中会在一定程度是同时考虑这两个节点是否存在相同副本。那么一个 node 到底属于哪个 zone 呢我在源码中找到了这样一段代码。
func GetZoneKey(node *v1.Node) string {labels : node.Labelsif labels nil {return }// TODO: prefer stable labels for zone in v1.18zone, ok : labels[v1.LabelZoneFailureDomain]if !ok {zone, _ labels[v1.LabelZoneFailureDomainStable]}// TODO: prefer stable labels for region in v1.18region, ok : labels[v1.LabelZoneRegion]if !ok {region, _ labels[v1.LabelZoneRegionStable]}if region zone {return }// We include the null character just in case region or failureDomain has a colon// (We do assume theres no null characters in a region or failureDomain)// As a nice side-benefit, the null character is not printed by fmt.Print or glogreturn region :\x00: zone
}
复制代码
这说明两个 node 是否是在同一个 zone取决于它的 label。
资源碎片问题怎么解决
举个例子来说明一下这个问题如同下图这样我有两个节点分别剩余1GPU但是当我想要创建一个 2GPU 的 pod 出来却找不到能调度的节点。 解决思路
如果我们优先把一个 node 占满再去往另一个 pod 上调度就能在一定程度上解决这个问题。 具体算法就是如下。 实现方式
和调度器自身解决同一控制器下的 pod 尽量分散的方式一样我们这个算法也是对 node 的打分进行干预。因此可以很容易想到把我们的算法插入进调度器就好了。那么这项工作该怎么进行呢下面将介绍业内主流的方案。
扩展调度器的标准流程Scheduling Framework
调度器扩展的方式可以总结为一下四种
直接改 kube-scheduler 的源码额外部署完全独立的调度器调度器扩展程序Scheduling Framework 前面三种已经不推荐使用了因此这里只介绍最后一种。 还记得我们在文章开头的那两个控制循环吧扩展的突破口就在第二个循环内部。它可以放大成下面这张图。 ○ 大致可分为两部分 ■ 调度过程为 pod 选择出 node ● 绿色区域是同步的同一时刻只能有一个 pod 在被调度 ■ 绑定过程把 pod 放到选出的 node 上 ● 黄色区域是异步的同一时刻可以有多个 pod 在被绑定 ○ 调度过程和绑定过程遇到如下情况时会中途退出这个时候该 Pod 将被放回到 待调度队列并等待下次重试 ■ 调度程序认为当前没有该 Pod 的可选节点 ■ 内部错误 调度器一共为我们提供了12个接口实现一个接口对应的是完成某一阶段的自定义调度逻辑这12个接口的作用分别是。QueueSort定义哪个 pod 先调度Pre-filter强制筛选的一部分信息预处理。Filter强制筛选掉绝对不能调度的 node。Post-filter可以用来做一些通知因为此时已经筛选掉了绝对不能调度的节点。留下的都是能运行该 pod 的节点只是此时还不能确定最佳节点是谁。PreScore打分前的预处理上面调度器源码的分析中实现者在这里保存了 pod 控制器的数据供后面的阶段使用。Score为每个 node 进行打分。NormalizeScoring修改上一步的得分作为最终得分。需要注意的是如果你在上一步的得分不在[0-100]这个区间那么你必须实现这个接口。Reserve在绑定之前为pod预留资源因为 pod 的绑定过程是异步的上图黄色部分这样可以避免资源超出的情况。如果绑定失败会触发 UnReserve 释放资源。Permit阻止或者延迟 pod 的绑定。Pre-bind绑定前的预处理比如处理 pod 依赖的网络资源。bind扩展用于将 pod 绑定到节点上。Unreserve因为在 reserve 阶段为 pod 预留了资源所以如果 pod 绑定失败需要在该节点释放预留的资源。
代码里如何体现
下面我基于kube-scheduler 1.19做一个简单的demo
第一步实现调度器插件
package pkgimport (v1 k8s.io/api/core/v1k8s.io/apimachinery/pkg/runtimecontextframework k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1
)type demoPlugin struct {}func NewDemoPlugin(_ runtime.Unknown, _ framework.FrameworkHandle) (framework.Plugin, error) {return demoPlugin{}, nil
}
// Name returns name of the plugin. It is used in logs, etc.
func (d *demoPlugin) Name() string {return demo-plugin
}
func (d *demoPlugin) Score(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {return 100,nil
}
// ScoreExtensions of the Score plugin.
func (d *demoPlugin) ScoreExtensions() framework.ScoreExtensions {return nil
}
复制代码
实现了 Name Score ScoreExtensions 这三个接口就完成了一个最简单的调度器插件这里我默认的就返回每个节点都是100分。
第二步把这个插件注册到调度器
package mainimport (git.cai-inc.com/devops/zcy-scheduler/pkg/demomath/randostimek8s.io/component-base/logsk8s.io/kubernetes/cmd/kube-scheduler/app)func main() {rand.Seed(time.Now().UnixNano())command : app.NewSchedulerCommand(app.WithPlugin(noderesources.AllocatableName, demo.NewDemoPlugin),)logs.InitLogs()defer logs.FlushLogs()if err : command.Execute(); err ! nil {os.Exit(1)}
}
复制代码
这里我们可以直接引用 kubernetres/cmd/kube-scheduler/app 下的代码里面有一个 NewSchedulerCommand 在这里就可以把我们的自定义插件注册进去。
第三步编写配置文件。
apiVersion: kubescheduler.config.k8s.io/v1beta1
kind: KubeSchedulerConfiguration
leaderElection:leaderElect: false
clientConnection:kubeconfig: /Users/zcy/newgo/zcy-scheduler/scheduler.conf
profiles:
- schedulerName: zcy-schedulerplugins:score:enabled:- name: demo-plugindisabled:- name: *
复制代码
在这里给你的调度器起一个名字并且设定了在 score 阶段使用 demo-plugin 插件。以上就是实现一个调度器的基本方式了。
总结
今天的文章介绍了scheduler 的结构主要就是两个循环。怎么保证 pod 的副本分散到不同节点上以及怎么解决资源碎片的问题。最后还介绍了如何开发自己的调度器希望能给大家开发自定义调度器时提供一些思路。
参考资料
www.qikqiak.com/post/custom…github.com/kubernetes-…time.geekbang.org/column/arti…
推荐阅读
Guava Cache实战—从场景使用到原理分析
详解 HTTP2.0 及 HTTPS 协议
初识 JVM带你从不同的视角认识 JVM
招贤纳士
政采云技术团队Zero一个富有激情、创造力和执行力的团队Base 在风景如画的杭州。团队现有300多名研发小伙伴既有来自阿里、华为、网易的“老”兵也有来自浙大、中科大、杭电等校的新人。团队在日常业务开发之外还分别在云原生、区块链、人工智能、低代码平台、中间件、大数据、物料体系、工程平台、性能体验、可视化等领域进行技术探索和实践推动并落地了一系列的内部技术产品持续探索技术的新边界。此外团队还纷纷投身社区建设目前已经是 google flutter、scikit-learn、Apache Dubbo、Apache Rocketmq、Apache Pulsar、CNCF Dapr、Apache DolphinScheduler、alibaba Seata 等众多优秀开源社区的贡献者。如果你想改变一直被事折腾希望开始折腾事如果你想改变一直被告诫需要多些想法却无从破局如果你想改变你有能力去做成那个结果却不需要你如果你想改变你想做成的事需要一个团队去支撑但没你带人的位置如果你想改变本来悟性不错但总是有那一层窗户纸的模糊……如果你相信相信的力量相信平凡人能成就非凡事相信能遇到更好的自己。如果你希望参与到随着业务腾飞的过程亲手推动一个有着深入的业务理解、完善的技术体系、技术创造价值、影响力外溢的技术团队的成长过程我觉得我们该聊聊。任何时间等着你写点什么发给 zcy-tccai-inc.com
