共計 11465 個字符，預計需要花費 29 分鐘才能閱讀完成。
本篇內容主要講解“如何實例化一個 Taint Manager”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓丸趣 TV 小編來帶大家學習“如何實例化一個 Taint Manager”吧!
NewNoExecuteTaintManager
PodInformer 添加 Event Handler 時，通過調用 taintManager.PodUpdated(oldPod *v1.Pod, newPod *v1.Pod) 往 tc.podUpdateQueue 添加 updateItem。
NodeInformer 添加 Event Handler 時，通過調用 taintManager.NodeUpdated(oldNode *v1.Node, newNode *v1.Node) 往 tc.nodeUpdateQueue 添加 updateItem。
當創建 NodeController 時，如果 runTaintManager 為 true(通過 kube-controller-manager 的 –enable-taint-manager 中指定，默認為 true)，則會通過 NewNoExecuteTaintManager 來實例化一個 Taint Manager。
pkg/controller/node/nodecontroller.go:195
func NewNodeController(..) (*NodeController, error) { podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{AddFunc: func(obj interface{}) {
if nc.taintManager != nil {nc.taintManager.PodUpdated(nil, pod)
} else {
nodeEventHandlerFuncs = cache.ResourceEventHandlerFuncs{AddFunc: func(originalObj interface{}) {
if nc.taintManager != nil {nc.taintManager.NodeUpdated(nil, node)
if nc.runTaintManager {nc.taintManager = NewNoExecuteTaintManager(kubeClient)
...
return nc, nil
}因此，創建 NodeController 時已經配置了監聽 pod 和 node 的事件，并會將相關數據發送到 tc.podUpdateQueue 和 tc.nodeUpdateQueue，然后由 Taint Manager 從中取出數據進行處理。在此之前，我們先來看看 NewNoExecuteTaintManager 是如何實例化一個 Taint Manager 的。
pkg/controller/node/taint_controller.go:152
func NewNoExecuteTaintManager(c clientset.Interface) *NoExecuteTaintManager {
tm := NoExecuteTaintManager{
client: c,
recorder: recorder,
// taintedNodes 記錄每個 Node 對應的 Taint 信息。taintedNodes: make(map[string][]v1.Taint),
// nodeUpdateQueue 中取出的 updateItem 會發送到 nodeUpdateChannel，Tait Manager 從該 Channel 中取出對應的 node update info。nodeUpdateChannel: make(chan *nodeUpdateItem, nodeUpdateChannelSize),
// podUpdateQueue 中取出的 updateItem 會發送到 podUpdateChannel，Tait Manager 從該 Channel 中取出對應的 pod update info。podUpdateChannel: make(chan *podUpdateItem, podUpdateChannelSize),
// Node Controller 監聽到的 node update info 會發送到 nodeUpdateQueue。nodeUpdateQueue: workqueue.New(),
// Node Controller 監聽到的 pod update info 會發送到 podUpdateQueue。podUpdateQueue: workqueue.New(),
// CreateWorkerQueue creates a new TimedWorkerQueue for workers that will execute deletePodHandler.
tm.taintEvictionQueue = CreateWorkerQueue(deletePodHandler(c, tm.emitPodDeletionEvent))
return tm
}相關的代碼分析見里面的代碼注釋。需要強調的是，我們在這里給 tm.taintEvictionQueue 注冊了函數 deletePodHandler，用來通過 Taint Eviction 時刪除 pod 時調用。Taint Manager Run 的時候會通過 tc.taintEvictionQueue.AddWork() 時創建 Worker 來執行 deletePodHandler。
func deletePodHandler(c clientset.Interface, emitEventFunc func(types.NamespacedName)) func(args *WorkArgs) error {return func(args *WorkArgs) error {
ns := args.NamespacedName.Namespace
name := args.NamespacedName.Name
glog.V(0).Infof(NoExecuteTaintManager is deleting Pod: %v , args.NamespacedName.String())
if emitEventFunc != nil {emitEventFunc(args.NamespacedName)
var err error
// 按照失敗重試 5 次，每次間隔 10s 的重試機制，調用 apiserver 的 api 刪除對應的 Pod。for i := 0; i retries; i++ {err = c.Core().Pods(ns).Delete(name, metav1.DeleteOptions{})
if err == nil {
break
time.Sleep(10 * time.Millisecond)
return err
}Run
在 Kubernetes Node Controller 源碼分析之執行篇中提到，在 Node Controller Run 的時候，如果 runTaintManager 為 true，則會調用 nc.taintManager.Run 啟動 Taint Manager loop。
pkg/controller/node/nodecontroller.go:550
func (nc *NodeController) Run() {go func() {
if nc.runTaintManager {go nc.taintManager.Run(wait.NeverStop)
}接下來，我們來看 Taint Manager 的 Run 方法。Node Controller 啟動的 Taint Manager 實例其實就是 NoExecuteTaintManager，其對應的 Run 方法代碼如下。
pkg/controller/node/taint_controller.go:179
// Run starts NoExecuteTaintManager which will run in loop until `stopCh` is closed.
func (tc *NoExecuteTaintManager) Run(stopCh -chan struct{}) {glog.V(0).Infof(Starting NoExecuteTaintManager)
// Functions that are responsible for taking work items out of the workqueues and putting them into channels.
// 從 tc.nodeUpdateQueue 中獲取 updateItem，并發送到 tc.nodeUpdateChannel。go func(stopCh -chan struct{}) {
for {item, shutdown := tc.nodeUpdateQueue.Get()
if shutdown {
break
nodeUpdate := item.(*nodeUpdateItem)
select {
case -stopCh:
break
case tc.nodeUpdateChannel - nodeUpdate:
}(stopCh)
// 從 tc.podUpdateQueue 中獲取 updateItem，并發送到 tc.podUpdateChannel。go func(stopCh -chan struct{}) {
for {item, shutdown := tc.podUpdateQueue.Get()
if shutdown {
break
podUpdate := item.(*podUpdateItem)
select {
case -stopCh:
break
case tc.podUpdateChannel - podUpdate:
}(stopCh)
// When processing events we want to prioritize Node updates over Pod updates,
// as NodeUpdates that interest NoExecuteTaintManager should be handled as soon as possible -
// we don t want user (or system) to wait until PodUpdate queue is drained before it can
// start evicting Pods from tainted Nodes.
for {
select {
case -stopCh:
break
// 從 tc.nodeUpdateChannel 獲取 nodeUpdate 數據，然后 invoke tc.handleNodeUpdate 進行處理。case nodeUpdate := -tc.nodeUpdateChannel:
tc.handleNodeUpdate(nodeUpdate)
// 從 tc.podUpdateChannel 獲取 podUpdate 數據，在 invoke tc.handlePodUpdate 進行處理之前，先確保 tc.nodeUpdateQueue 中的數據已經被處理完。case podUpdate := -tc.podUpdateChannel:
// If we found a Pod update we need to empty Node queue first.
priority:
for {
select {
case nodeUpdate := -tc.nodeUpdateChannel:
tc.handleNodeUpdate(nodeUpdate)
default:
break priority
// After Node queue is emptied we process podUpdate.
tc.handlePodUpdate(podUpdate)
}可見, Run 方法中分別從對應的 queue 中取出數據，然后調用 tc.handleNodeUpdate 和 tc.handlePodUpdate 進行處理。
// pkg/controller/node/taint_controller.go:365
func (tc *NoExecuteTaintManager) handleNodeUpdate(nodeUpdate *nodeUpdateItem) {
// Delete
// 如果 nodeUpdate.newNode == nil，則表明該 Node 被刪除了，那么將該 Node 的 Taints 信息從 tc.taintedNodes 緩存中刪除。if nodeUpdate.newNode == nil {
node := nodeUpdate.oldNode
glog.V(4).Infof(Noticed node deletion: %#v , node.Name)
tc.taintedNodesLock.Lock()
defer tc.taintedNodesLock.Unlock()
delete(tc.taintedNodes, node.Name)
return
// Create or Update
// 如果是 Node Create 或者 Node Update Event，則更新 tc.taintedNodes 緩存中記錄的該 Node 的 Taints 信息。glog.V(4).Infof(Noticed node update: %#v , nodeUpdate)
node := nodeUpdate.newNode
taints := nodeUpdate.newTaints
func() {tc.taintedNodesLock.Lock()
defer tc.taintedNodesLock.Unlock()
glog.V(4).Infof(Updating known taints on node %v: %v , node.Name, taints)
if len(taints) == 0 {delete(tc.taintedNodes, node.Name)
} else {tc.taintedNodes[node.Name] = taints
// 然后，獲取該 Node 上所有 pods list。pods, err := getPodsAssignedToNode(tc.client, node.Name)
if err != nil {glog.Errorf(err.Error())
return
if len(pods) == 0 {
return
// Short circuit, to make this controller a bit faster.
// 如果該 Node 上的 Taints 被刪除了，則取消所有該 node 上的 pod evictions。if len(taints) == 0 {glog.V(4).Infof(All taints were removed from the Node %v. Cancelling all evictions... , node.Name)
for i := range pods {tc.cancelWorkWithEvent(types.NamespacedName{Namespace: pods[i].Namespace, Name: pods[i].Name})
return
// 否則，調用 tc.processPodOnNode 根據 Node Taints info 和 Pod Tolerations info 處理該 Node 上的 Pod Eviction。now := time.Now()
for i := range pods {pod := pods[i]
podNamespacedName := types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}
tc.processPodOnNode(podNamespacedName, node.Name, pod.Spec.Tolerations, taints, now)
}handleNodeUpdate 的邏輯為：
如果 nodeUpdate.newNode == nil，則表明該 Node 被刪除了，那么將該 Node 的 Taints 信息從 tc.taintedNodes 緩存中刪除。
如果是 Node Create 或者 Node Update Event，則更新 tc.taintedNodes 緩存中記錄的該 Node 的 Taints 信息。
獲取該 Node 上所有 pods list。
如果該 Node 上的 Taints 被刪除了，則取消所有該 node 上的 pod evictions。
否則，遍歷 pods list 中的每個 pod，分別調用 tc.processPodOnNode 根據 Node Taints info 和 Pod Tolerations info 處理該 Node 上的 Pod Eviction。
// pkg/controller/node/taint_controller.go:334
func (tc *NoExecuteTaintManager) handlePodUpdate(podUpdate *podUpdateItem) {
// Delete
// 如果 podUpdate.newPod == nil，則表明該 Pod 被刪除了，那么取消該 Pod Evictions。if podUpdate.newPod == nil {
pod := podUpdate.oldPod
podNamespacedName := types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}
glog.V(4).Infof(Noticed pod deletion: %#v , podNamespacedName)
tc.cancelWorkWithEvent(podNamespacedName)
return
// Create or Update
// 如果是 Pod Create 或者 Pod Update Event，則取出該 pod 的 node 上的 Taints info。pod := podUpdate.newPod
podNamespacedName := types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}
glog.V(4).Infof(Noticed pod update: %#v , podNamespacedName)
nodeName := pod.Spec.NodeName
if nodeName == {
return
taints, ok := func() ([]v1.Taint, bool) {tc.taintedNodesLock.Lock()
defer tc.taintedNodesLock.Unlock()
taints, ok := tc.taintedNodes[nodeName]
return taints, ok
// It s possible that Node was deleted, or Taints were removed before, which triggered
// eviction cancelling if it was needed.
if !ok {
return
// 然后，調用 tc.processPodOnNode 根據 Node Taints info 和 Pod Tolerations info 處理該 Node 上的 Pod Eviction。tc.processPodOnNode(podNamespacedName, nodeName, podUpdate.newTolerations, taints, time.Now())
}handlePodUpdate 的邏輯為：
如果 podUpdate.newPod == nil，則表明該 Pod 被刪除了，那么取消該 Pod Evictions。
如果是 Pod Create 或者 Pod Update Event，則取出該 pod 的 node 上的 Taints info。
如果 node 上的 Taints info 信息為空，表明 Taints info 被刪除了或者 Node 被刪除了，那么就不需要處理該 node 上的 pod eviction 了, 流程結束。
否則，調用 tc.processPodOnNode 根據 Node Taints info 和 Pod Tolerations info 處理該 Node 上的 Pod Eviction。
因此，不管是 handlePodUpdate 還是 handleNodeUpdate, 最終都是通過 processPodOnNode 來處理 Pod Eviction 的。
pkg/controller/node/taint_controller.go:295
func (tc *NoExecuteTaintManager) processPodOnNode(
podNamespacedName types.NamespacedName,
nodeName string,
tolerations []v1.Toleration,
taints []v1.Taint,
now time.Time,
) {
// 如果該 node 的 taints info 為空，則取消 Taint Eviction Pods。if len(taints) == 0 {tc.cancelWorkWithEvent(podNamespacedName)
// 對比 node 的 taints info 和 pod tolerations info，判斷出 node 的 taints 是否都能被 pod 所能容忍。allTolerated, usedTolerations := v1.GetMatchingTolerations(taints, tolerations)
// 如果不是全部都能容忍，那么調用立刻調用 AddWork 來創建 worker，啟動 tc.taintEvictionQueue 注冊的 deletePodHandler 來刪除該 pod。if !allTolerated {glog.V(2).Infof(Not all taints are tolerated after update for Pod %v on %v , podNamespacedName.String(), nodeName)
// We re canceling scheduled work (if any), as we re going to delete the Pod right away.
tc.cancelWorkWithEvent(podNamespacedName)
tc.taintEvictionQueue.AddWork(NewWorkArgs(podNamespacedName.Name, podNamespacedName.Namespace), time.Now(), time.Now())
return
// 否則，取 pod 的所有 tolerations 的 TolerationSeconds 的最小值作為 minTolerationTime。如果某個 Toleration 沒有設置 TolerationSeconds，則表示 0，如果設置的值為負數，則用 0 替代。minTolerationTime := getMinTolerationTime(usedTolerations)
// getMinTolerationTime returns negative value to denote infinite toleration.
if minTolerationTime 0 {glog.V(4).Infof(New tolerations for %v tolerate forever. Scheduled deletion won t be cancelled if already scheduled. , podNamespacedName.String())
return
startTime := now
triggerTime := startTime.Add(minTolerationTime)
// 從 tc.taintEvictionQueue 中獲取 Worker-scheduledEviction
scheduledEviction := tc.taintEvictionQueue.GetWorkerUnsafe(podNamespacedName.String())
// 如果獲取到不為空的 scheduledEviction，則判斷 worker 創建時間加上 minTolerationTime 是否達到觸發時間要求，如果沒達到，則不進行 Taint Pod Eviction，流程結束。if scheduledEviction != nil {
startTime = scheduledEviction.CreatedAt
if startTime.Add(minTolerationTime).Before(triggerTime) {return} else {tc.cancelWorkWithEvent(podNamespacedName)
// 如果達到觸發時間要求，則取消 worker，并立刻調用 AddWork 來創建 worker，啟動 tc.taintEvictionQueue 注冊的 deletePodHandler 來刪除該 pod。tc.taintEvictionQueue.AddWork(NewWorkArgs(podNamespacedName.Name, podNamespacedName.Namespace), startTime, triggerTime)
}processPodOnNode 的邏輯為：
如果該 node 的 taints info 為空，則取消 Taint Eviction Pods。
對比 node 的 taints info 和 pod tolerations info，判斷出 node 的 taints 是否都能被 pod 所能容忍。
如果不是全部都能容忍，那么調用立刻調用 AddWork 來創建 worker，啟動 tc.taintEvictionQueue 注冊的 deletePodHandler 來刪除該 pod。
否則，取 pod 的所有 tolerations 的 TolerationSeconds 的最小值作為 minTolerationTime。如果某個 Toleration 沒有設置 TolerationSeconds，表示不作驅逐。
如果獲取到不為空的 scheduledEviction，則判斷 worker 創建時間加上 minTolerationTime 是否達到觸發時間要求，如果沒達到，則不進行 Taint Pod Eviction，流程結束。
如果達到觸發時間要求，則取消 worker，并立刻調用 AddWork 來創建 worker，啟動 tc.taintEvictionQueue 注冊的 deletePodHandler 來刪除該 pod。
如果 minTolerationTime 小于 0，則永遠容忍，流程結束。
從 tc.taintEvictionQueue 中獲取 Worker-scheduledEviction。
到此，相信大家對“如何實例化一個 Taint Manager”有了更深的了解，不妨來實際操作一番吧！這里是丸趣 TV 網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！
