【 K8s 源码之调度学习】Pod 间亲和性和反亲和性的源码分析

查看案例

字段	含义
podAffinity	Pod 间的亲和性定义
podAntiAffinity	Pod 间的反亲和性定义
requiredDuringSchedulingIgnoredDuringExecution	硬性要求，必须满足条件，保证分散部署的效果最好使用用此方式
preferredDuringSchedulingIgnoredDuringExecution	软性要求，可以不完全满足，即有可能同一node上可以跑多个副本
requiredDuringSchedulingIgnoredDuringExecution	labelSelector
	topologyKey
preferredDuringSchedulingIgnoredDuringExecution	weight
	podAffinityTerm	labelSelector
		topologyKey
topologyKey	可以理解为 Node 的 Label，具有相同的 Label 的 Node，视为同一拓扑
	如三个节点打上 Label ： - Node1 —— zone：beijing - Node2 —— zone：shanghai - Node3 —— zone：beijing 那么 Node1 和 Node3 为同一拓扑，Node2 为另一拓扑
	topologyKey: kubernetes.io/hostname 上面为常见的配置，可以通过 kubectl get nodes --show-labels看到 节点上的 Lable，就具有此 kubernetes.io/hostname Label 因此就是将每个节点，作为一个独立的拓扑

apiVersion: v1
kind: Pod
metadata:name: test-pod
spec:affinity:# 首先根据 labelSelector 选择具有 service.cpaas.io/name: deployment-nginx  Label 的 所有 Pod# 接下来根据 podAffinity 亲和性，将此 pod 调度到与选中 Pod 中具有 topologyKey 的 Node 上podAffinity:requiredDuringSchedulingIgnoredDuringExecution:- labelSelector:matchLabels:service.cpaas.io/name: deployment-nginxtopologyKey: kubernetes.io/hostname- labelSelector:matchLabels:service.cpaas.io/name: deployment-busyboxtopologyKey: kubernetes.io/hostname# 首先根据 labelSelector 选择具有 key 为 a ，value为 b 或 c 的 Label 的 Pod# 接下来根据 podAntiAffinity，将此 pod 调度到与选中 Pod 中都不相同的 Node 上，该节点需要具有 topologyKey labelpodAntiAffinity:preferredDuringSchedulingIgnoredDuringExecution:- weight: 100podAffinityTerm:labelSelector:matchExpressions:- key: aoperator: Invalues: ["b", "c"]topologyKey: kubernetes.io/hostnamecontainers:- name: test-podimage: nginx:1.18

代码分析

代码路径：pkg/scheduler/framework/plugins/interpodaffinity

首先根据调度器框架，观察源码，可以看出实现了一下四个接口：

1. PreFilter
2. Filter
3. PreScore
4. Score

首先明确几点

• 该插件是考虑 Pod 间的亲和性和反亲和性（就是新Pod 和 现存 Pod 的关系）
• 但最终结果是将 Pod 调度到合适的 Node 上（因此要记录 Node 的信息）

1 | PreFilter

此步骤作用：

1. 梳理出【现存哪些 Pod】 讨厌【新 Pod】，记录【满足条件的现存 Pod】 对应 Node 信息为 existingPodAntiAffinityMap
2. 梳理出【新 Pod】喜欢【哪些现存Pod】，记录【满足条件的现存 Pod】 对应 Node 信息为 incomingPodAffinityMap
3. 梳理出【新 Pod】讨厌【哪些现存Pod】，记录【满足条件的现存 Pod】 对应 Node 信息为 incomingPodAntiAffinityMap

所以可以小总结一下

• existingPodAntiAffinityMap 和 incomingPodAntiAffinityMap 这些记录的节点，新 Pod 不喜欢
• incomingPodAffinityMap 记录的节点，Pod 喜欢

问题 —— 为什么不梳理 【现存哪些 Pod】 喜欢【新 Pod】？

• 因为现在是调度【新 Pod】，只要不被讨厌，不影响【现存 Pod 】就行，因此只需要可能会影响的【现存 Pod】

注意上面所说的【条件】—— 指的是【硬性要求 requiredDuringSchedulingIgnoredDuringExecution 】 —— 因此才考虑这么详细

// 这里只截取了 PreFilter 部分重要函数
// pkg/scheduler/framework/plugins/interpodaffinity/filtering.go// 考虑现存 Pod 的 反亲和性 anti-affinity
// 简单理解：就是用现存 Pod 的 anti-affinity Terms 配置，要求 NewPod，记录下满足的 Node，说明这些节点不能调度（因为现存 Pod 排斥新 Pod）
// 这里的 anti-affinity Terms 是指 requiredDuringSchedulingIgnoredDuringExecution 定义的硬性要求
// 问题：为什么不考虑现存 Pod 的亲和性？ —— 因为现存 Pod 的亲和性（是亲和他之前 Pod），在其调度的时候早已考虑，现在只需要考虑其反感的
// 代码级理解：
// 1. 遍历所有具有 anti-affinity 现存 Pod
// 2. 若即将调度的 NewPod 满足该 Pod 的 anti-affnity Terms，
// 3. 就记录到 existingPodAntiAffinityMap 中，key 为该 Pod 所在的 node 信息（topologyKey、topologyValue），value 为满足的 Terms 次数
// 例如 map{（hostname：node01）：1}
// existingPodAntiAffinityMap will be used later for efficient check on existing pods' anti-affinity
existingPodAntiAffinityMap := getTPMapMatchingExistingAntiAffinity(pod, nodesWithRequiredAntiAffinityPods)// 考虑新 NewPod 的亲和性和反亲和性
// 简单理解： 就是用 NewPod 的 anti-affinity 和 affinity Terms 配置，要求现存的 Pod，记录下满足的 Node
// incomingPodAffinityMap will be used later for efficient check on incoming pod's affinity
// incomingPodAntiAffinityMap will be used later for efficient check on incoming pod's anti-affinity
incomingPodAffinityMap, incomingPodAntiAffinityMap := getTPMapMatchingIncomingAffinityAntiAffinity(podInfo, allNodes)

2 | Filter

• *framework.CycleState 将上面统计的信息传递过来
• 现在的工作就是：
  • 传来了一个 Node 信息
  • 判断该 Node 与上面的 existingPodAntiAffinityMap、incomingPodAntiAffinityMap 、incomingPodAffinityMap 的关系
  • 若该 Node 满足条件，那么可以进入到下面的【打分阶段】

// pkg/scheduler/framework/plugins/interpodaffinity/filtering.go
func (pl *InterPodAffinity) Filter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {if nodeInfo.Node() == nil {return framework.NewStatus(framework.Error, "node not found")}state, err := getPreFilterState(cycleState)if err != nil {return framework.NewStatus(framework.Error, err.Error())}if !satisfyPodAffinity(state, nodeInfo) {return framework.NewStatus(framework.UnschedulableAndUnresolvable, ErrReasonAffinityNotMatch, ErrReasonAffinityRulesNotMatch)}if !satisfyPodAntiAffinity(state, nodeInfo) {return framework.NewStatus(framework.Unschedulable, ErrReasonAffinityNotMatch, ErrReasonAntiAffinityRulesNotMatch)}if !satisfyExistingPodsAntiAffinity(state, nodeInfo) {return framework.NewStatus(framework.Unschedulable, ErrReasonAffinityNotMatch, ErrReasonExistingAntiAffinityRulesNotMatch)}return nil
}

3 | PreScore

这部分主要看 processExistingPod 函数

• 可以看出根据【现存 Pod】 和【新 Pod】的【软性要求preferredDuringSchedulingIgnoredDuringExecution】，对节点进行打分

// pkg/scheduler/framework/plugins/interpodaffinity/scoring.go
// PreScore builds and writes cycle state used by Score and NormalizeScore.
func (pl *InterPodAffinity) PreScore(pCtx context.Context,cycleState *framework.CycleState,pod *v1.Pod,nodes []*v1.Node,
) *framework.Status {// ... ...topoScores := make([]scoreMap, len(allNodes))index := int32(-1)processNode := func(i int) {nodeInfo := allNodes[i]if nodeInfo.Node() == nil {return}// Unless the pod being scheduled has affinity terms, we only// need to process pods with affinity in the node.podsToProcess := nodeInfo.PodsWithAffinityif hasAffinityConstraints || hasAntiAffinityConstraints {// We need to process all the pods.podsToProcess = nodeInfo.Pods}topoScore := make(scoreMap)for _, existingPod := range podsToProcess {pl.processExistingPod(state, existingPod, nodeInfo, pod, topoScore)}if len(topoScore) > 0 {topoScores[atomic.AddInt32(&index, 1)] = topoScore}}parallelize.Until(context.Background(), len(allNodes), processNode)for i := 0; i <= int(index); i++ {state.topologyScore.append(topoScores[i])}cycleState.Write(preScoreStateKey, state)return nil
}func (pl *InterPodAffinity) processExistingPod(state *preScoreState,existingPod *framework.PodInfo,existingPodNodeInfo *framework.NodeInfo,incomingPod *v1.Pod,topoScore scoreMap,
) {existingPodNode := existingPodNodeInfo.Node()// For every soft pod affinity term of <pod>, if <existingPod> matches the term,// increment <p.counts> for every node in the cluster with the same <term.TopologyKey>// value as that of <existingPods>`s node by the term`s weight.topoScore.processTerms(state.podInfo.PreferredAffinityTerms, existingPod.Pod, existingPodNode, 1)// For every soft pod anti-affinity term of <pod>, if <existingPod> matches the term,// decrement <p.counts> for every node in the cluster with the same <term.TopologyKey>// value as that of <existingPod>`s node by the term`s weight.topoScore.processTerms(state.podInfo.PreferredAntiAffinityTerms, existingPod.Pod, existingPodNode, -1)// For every hard pod affinity term of <existingPod>, if <pod> matches the term,// increment <p.counts> for every node in the cluster with the same <term.TopologyKey>// value as that of <existingPod>'s node by the constant <args.hardPodAffinityWeight>if pl.args.HardPodAffinityWeight > 0 {for _, term := range existingPod.RequiredAffinityTerms {t := framework.WeightedAffinityTerm{AffinityTerm: term, Weight: pl.args.HardPodAffinityWeight}topoScore.processTerm(&t, incomingPod, existingPodNode, 1)}}// For every soft pod affinity term of <existingPod>, if <pod> matches the term,// increment <p.counts> for every node in the cluster with the same <term.TopologyKey>// value as that of <existingPod>'s node by the term's weight.topoScore.processTerms(existingPod.PreferredAffinityTerms, incomingPod, existingPodNode, 1)// For every soft pod anti-affinity term of <existingPod>, if <pod> matches the term,// decrement <pm.counts> for every node in the cluster with the same <term.TopologyKey>// value as that of <existingPod>'s node by the term's weight.topoScore.processTerms(existingPod.PreferredAntiAffinityTerms, incomingPod, existingPodNode, -1)
}

4 | Score

这部分就是，将节点的得分进行累计计算，返回此符合条件的节点的得分数

• 注意，所有符合条件都会调用此函数，得到自己对应的分数

// pkg/scheduler/framework/plugins/interpodaffinity/scoring.go
// Score invoked at the Score extension point.
// The "score" returned in this function is the sum of weights got from cycleState which have its topologyKey matching with the node's labels.
// it is normalized later.
// Note: the returned "score" is positive for pod-affinity, and negative for pod-antiaffinity.
func (pl *InterPodAffinity) Score(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {nodeInfo, err := pl.sharedLister.NodeInfos().Get(nodeName)if err != nil || nodeInfo.Node() == nil {return 0, framework.NewStatus(framework.Error, fmt.Sprintf("getting node %q from Snapshot: %v, node is nil: %v", nodeName, err, nodeInfo.Node() == nil))}node := nodeInfo.Node()s, err := getPreScoreState(cycleState)if err != nil {return 0, framework.NewStatus(framework.Error, err.Error())}var score int64for tpKey, tpValues := range s.topologyScore {if v, exist := node.Labels[tpKey]; exist {score += tpValues[v]}}return score, nil
}