自己實作Controller——標準型

標準Controller

上一篇通過一個簡單的例子，撰寫了一個controller-manager，以及一個極簡單的controller，從而對controller的開發有個最基本的認識，但是細心觀察前一篇實作的controller僅僅是每次全量獲取了所有資源，雖然都是從快取中獲取速度是比較快的，如果單次處理一個資源時的時間比較長，而且沒必要每次都把所有資源都掃描一遍，上一篇實作的controller顯然不符合使用場景了，本篇將繼續用另一種方式開發一個結構較為標準的Controller,并介紹支撐controller功能實作的Informer架構，

先介紹一下本次實作的controller需要實作的功能，本controller通過監控Node節點的新增，發現新增是通過特定方式加入集群的，就給該node打上label，

看下本次controller的結構包含的成員

type NodeController struct {
	kubeClient       *kubernetes.Clientset		//用于給符合條件的node打標記用
	nodeLister       corelisters.NodeLister		//用于獲取被監控的node資源
	nodeListerSynced cache.InformerSynced		
	nodesQueue       workqueue.DelayingInterface	//一個延時佇列，用于記錄需要controller的node的key
	cloudProvider    cloudproviders.CloudProvider	//用于判定node是否符合條件打標記，此成員并非controller關鍵結構的成員
}

下面是Controller的建構式

func NewNodeController(kubeClient *kubernetes.Clientset, nodeInformer coreinformers.NodeInformer, cp cloudproviders.CloudProvider) *NodeController {

	n := &NodeController{
		kubeClient:       kubeClient,
		nodeLister:       nodeInformer.Lister(),
		nodeListerSynced: nodeInformer.Informer().HasSynced,
		nodesQueue:       workqueue.NewNamedDelayingQueue("nodes"),
		cloudProvider:    cp,
	}

	nodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(cur interface{}) {
			node := cur.(*v1.Node)
			fmt.Printf("controller: Add event, nodes [%s]\n",  node.Name)
			n.syncNodes(node)
		},
		DeleteFunc: func(cur interface{}) {
			node := cur.(*v1.Node)
			fmt.Printf("controller: Delete event, nodes [%s]\n",  node.Name)
			n.syncNodes(node)
		},
	})

	return n
}

傳入建構式的Informer除了提供Lister和HasSynced函式以外，多了一個事件注冊的操作，注冊了node資源的Add事件和Delete事件，這個AddEventHandler除了注冊這兩種事件外，還可以注冊Update事件，由于在這次例子中不需要處理這方面的事件，因此沒有用上，此處通過Informer的事件回呼實作了哪個資源有變更，就會觸發事件，通知到controller來，與前一個controller相比這種就能立馬定位到有變更的資源，無需將資源全量掃描才找到對應的資源，代碼中兩個事件處理函式都是獲取到變更的node,列印出node的名字和變更事件，最后呼叫syncNodes方法

func (n *NodeController) syncNodes(node *v1.Node) {
	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(node)
	if err != nil {
		fmt.Printf("Couldn't get key for object %#v: %v \n", node, err)
		return
	}
	fmt.Printf("working queue add node %s\n", key)
	n.nodesQueue.Add(key)
}

syncNodes單純給資源求了一個key，再把這個key塞到workqueue中，接收informer傳遞過來的變更事件處理也就此結束，

然后到啟動controller的Run方法

func (n *NodeController) Run(stopCh <-chan struct{}) {

	defer runtime.HandleCrash()
	defer n.nodesQueue.ShutDown()

	fmt.Println("Starting service controller")
	defer fmt.Println("Shutting down service controller")

	if !cache.WaitForCacheSync(stopCh, n.nodeListerSynced) {
		runtime.HandleError(fmt.Errorf("time out waiting for caches to sync"))
		return
	}

	for i := 0; i < WorkerCount; i++ {
		go wait.Until(n.worker, time.Second, stopCh)
	}

	<-stopCh
}
func (n *NodeController) worker() {
	for {
		func() {
			key, quit := n.nodesQueue.Get()
			if quit {
				return
			}
			defer n.nodesQueue.Done(key)
			err := n.handleNode(key.(string))
			if err != nil {
				fmt.Printf("controller: error syncing node, %v \n", err)
			}
		}()
	}
}

接收終止信號和快取處理與之前的controller沒多大區別，區別就在于啟動了若干個worker協程，worker方法就從workqueue中取出key,然后處理node

func (n *NodeController) handleNode(key string) error {
	_, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		return err
	}
	node, err := n.nodeLister.Get(name)
	switch {
	case errors.IsNotFound(err):
		fmt.Printf("Node has been deleted %v \n", key)
		return nil
	case err != nil:
		fmt.Printf("Unable to retrieve node %v from store: %v \n", key, err)
		n.nodesQueue.AddAfter(key, time.Second*30)
		return err
	default:
		err := n.processNodeAddIntoCluster(node)
		if err != nil {
			n.nodesQueue.AddAfter(key, time.Second*30)
		}
		return err
	}
	return nil
}

由于workqueue里存的也只是代表資源的key,此時需要將key轉換回資源的name，還是通過lister從cache中把資源取出，當取不出來，就表明資源已經被洗掉了，如果有相應的洗掉相關的邏輯就可以在此處執行；如果能取出來，當前的資源已經是最新版本的資源（更新時會有新舊兩個版本的資源，這里拿到的是新版本的），當然在本例中沒有更新這個事件的處理；整個程序中出現了非期望的錯誤（如因資源被洗掉導致的IsNotFound error）,處理邏輯需要重試，重試的方式就是將延時重新入隊，延時佇列的作用在此處得以發揮，因為有可能出現這種極端場景：當前只有一個資源需要被處理，而且該資源在剛創建的時候因為其狀態未就緒確實會一直處理失敗，假設當其被處理失敗時馬上又重新入隊，那整個controller就會陷入一個類似于死回圈的狀態中，直到資源狀態就緒，這樣會浪費不少計算資源，

最后的processNodeAddIntoCluster就是給node打上label并呼叫kubeclient更新之，這里就不展示了，

看到上面使用了workqueue,頗有生產者消費者的模式,從informer的事件處理函式中獲取到變更的資源將其放入佇列，這個是生產者；worker處理方法從workqueue里取出變更資源處理，這個是消費者，使用了這個workqueue而不是直接在事件處理函式直接處理的目的在于事件觸發的速度與資源處理的速度不一致，有workqueue在其中起到緩沖作用，免除了因實際處理變更的邏輯造成了事件觸發方的阻塞導致影響了事件的實時性，

類似的佇列和生產者消費者模式在informer中也有出現，作為controller處理資源變更調諧資源狀態的上游，它又如何提供及時的變更事件給下游的controller,以及給下游的controller提供與apiserver一致的資源快取，

Informer機制

Informer機制架構如下圖所示

從該架構中還涉及了幾個相關組件：

• Reflector:用于監控apiserver相關資源的變化，及時把變更獲取回來，觸發相關的變更事件，變更的資源存放到Delta FIFO里面；
• Delta FIFO:一個存放資源變更的FIFO佇列，里面元素是以 Add/Update/Delete 為key,變更的資源為value這樣的一個物件；
• Indexder:用于存放從Delta FIFO佇列取出后，經過處理好的值，是apiserver的一個快取，無需每次從apiserver以及Etcd中獲取，減輕兩者的壓力，里面的快取應與apiserver(實質是Etcd)保持一致，

從上面的結構中，尤其是Delta FIFO，模式與上面使用了WorkQueue的controller是一致的，生產者消費者模式：

• 生產者方面：主要操作由Reflector執行，它主要依靠ListAndWatch方法對apiserver的資源進行監控，ListAndWatch方法中就呼叫了由Informer提供的List方法和Watch方法，一旦獲得變更的資源，就將資源及其變更的方式（Add,Update,Delete）一同存入Delta FIFO,如上面介紹Delta FIFO時所述；
• 消費者方面：主要操作由Controller執行，它主要呼叫Delta FIFO的Pop方法，從佇列中取出變更的資源及其變更方式，呼叫之前注冊到Informer里面相關的事件（即例子中NewNodeController函式中AddEventHandler方法注冊進去的事件處理函式），將變更分發出去，在這里Delta FIFO除了記錄變更的資源本身資料，也附帶記錄變更方式的作用體現出來了，最后將這個資源快取到Indexer中，快取時也是通過變更方式對快取進行操作，比如是Add變更，則直接往快取中Add一個記錄；是Update則把快取記錄更新；是Delete則洗掉記錄，這也是另一個體現記錄變更方式的地方，

Informer和自定義的Controller(這個controller并非消費者的controller)的互動主要是兩處，其一是AddEventHandler時，把變更資源收到自己的workerqueue中供后續執行調諧操作；其二是在調諧期間從Indexer(這個并非是直接呼叫，是通過呼叫Informer時嵌套呼叫的)獲取快取的資源值，整個程序所有涉及的成員如下圖所示

小結Informer

Informer在整個程序中起到銜接各方的作用，Reflector對apiserver的監控是由Informer提供的ListAndWatch方法，資源從apiserver介面拿回來資源資料由注冊進去Informer的反序列化器進行反序列化（這部分跟Schema有關），而資料進入Indexer快取前，的事件分發也是通過Informer呼叫各個已經注冊上來的事件處理函式，

了解了Informer機制，后面則進行自定義資源的Controller開發，里面還需要給自定義資源擴展一個Informer,及封裝跟apiserver訪問自定義資源的client,實作完整的手捏Controller.

標籤：Go

上一篇：Python中的元組介紹

下一篇：系統呼叫跟蹤——ls功能實作(二)