寫在最前
controller-manager作為K8S master的其中一個組件，負責眾多controller的啟動和終止，這些controller負責監控著k8s中各種資源，執行調諧，使他們的實際狀態能不斷趨近與期望狀態，這些controller包括servercontroller,nodecontroller,deploymentcontroller等，對于自定義資源（CRD）也需要為之配備controller,CRD的controller也需要有controller-manager啟動之，停止它，整個程序篇幅較長，故鄙人將其拆分成多篇，通過本系列，首篇將介紹如何定義一個controller-manager去啟動它所管轄的controller，并實作一個最簡單的controller，第二篇再實作一個較為標準的controller,并介紹informer的結構；最后一篇將介紹不借助腳手架如何實作一個CRD的controller，
介紹一下整個專案的結構
controller-demo
|---api //用于放定義CRD各個屬性的struct
|---v1
|---client
|---versiond
|----scheme //用于存放CRD的scheme
|----typed //用于存放CRD對應的client
|---controller //用于存放各個controller
|---informers //用于存放informer,包含各個apiGroup各個version及一個factory
|---ecsbind/v1 //其中一個apiGroup,其中一個version的informer,當然也是唯一一個
|---internalinterfaces //informer的interface介面
|---listers //用于存放lister,包含各個apiGroup各個version
|---ecsbind/v1 //其中一個apiGroup,其中一個version的informer,同樣也是唯一一個
controller-manager
controller有兩個函式，一個負責供main函式呼叫啟動controller-manager,作為controller-manager的入口；另一個是用于啟動他所管理的所有controller，
供main函式呼叫的Run函式定義如下
func Run(stopCh <-chan struct{}) error {
run :=func(stopCh <-chan struct{}){
err := StartController(stopCh)
if err != nil {
glog.Fatalf("error running service controllers: %v", err)
}
select {}
}
///忽略leader選舉的相關邏輯
......
run(stopCh)
panic("unreachable")
}
上述函式傳入一個通道，用于傳遞給各個controller一個終止的信號，函式里定義了一個run的函式，用于呼叫StartController，之所以需要定義一個run函式，是因為一般這類的組件雖然為了高可用會運行多個副本，但是僅有一個副本是真正運行，其他的副本是作為待命狀態運行，而這個真正運行的副本稱為leader,從普通副本中通過資源爭奪稱為leader的程序稱為leader選舉,僅有leader掛掉了，剩余的副本再進行一次leader選舉成為新leader，當然也可進行leader選舉模式運行，因此Run函式中應該包含是否進行leader選舉，若是則執行leader選舉的邏輯，當選成leader才執行run函式；如果不進行leader選舉則直接執行run，不過這段邏輯被省略了，
controller-manager的另一個函式是真正啟動各個controller，StartController同樣接收了從Run函式傳過來的通道，這個通道最終轉給各個controller，傳遞停止的信號，在函式中會構造各個controller,通過開辟一個協程呼叫controller的Run方法將controller啟動，代碼如下所示
unc StartController(stopCh <-chan struct{}) error {
cfg, err := clientcmd.BuildConfigFromFlags("", "/root/.kube/config")
if err != nil {
glog.Fatalf("error building kubernetes config:%s", err.Error())
}
kubeClient, err := kubernetes.NewForConfig(cfg)
factory := informers.NewSharedInformerFactory(kubeClient, 0)
podInformer:=factory.Core().V1().Pods()
pc:=controller.NewPodController(kubeClient,podInformer,"k8s-cluster")
go pc.Run(stopCh)
factory.Start(stopCh)
return nil
}
controller.NewPodController是構造了一個PodController,構造PodController時所需要的kubeClient,informer需要預先構造，對于k8s原有的資源，其informer都可以通過SharedInformerFactory獲得，通過協程執行 pc.Run(stopCh)后，也需要執行factory.Start(stopCh)，factroy.Start需要等各個controller Run了之后方可執行，否則對應Controlle則會沒有運行效果，
一個精簡的Controller
這個controller的作用是統計集群中所有pod的數量，然后將pod的總數寫到master的某個label上，且被統計過的pod都會在它的event中產生一條新記錄來表明此pod被統計過，
羅列一下這個podcontroller結構的欄位
type PodController struct {
kubeClient kubernetes.Interface //用于給master打label
clusterName string
podLister corelisters.PodLister //用于獲取被監控的pod資源
podListerSynced cache.InformerSynced //用于同步cache
broadcaster record.EventBroadcaster //用于廣播事件
recorder record.EventRecorder //用于記錄pod的event
}
Controller的建構式如下
func NewPodController(kubeClient kubernetes.Interface,podInformer coreinformers.PodInformer,clusterName string)*PodController {
eventBroadcaster := record.NewBroadcaster()
eventBroadcaster.StartLogging(glog.Infof)
recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "pod_controller"})
rc:=&PodController{
kubeClient:kubeClient,
clusterName:clusterName,
podLister:podInformer.Lister(),
podListerSynced:podInformer.Informer().HasSynced,
broadcaster:eventBroadcaster,
recorder:recorder,
}
return rc
}
controller的各個屬性中，除了broadcaster和recorder是自身構造外，其余都是通過引數傳入，由于controller中需要用到事件記錄，提供這一功能的是recorder，然而觸發了事件需要將其散播出去給訂閱者的需要一個broadcaster，這里涉及到k8s的事件機制，并不打算在細述，在建構式中已經把事件廣播系結到glog.Infof，也就是說recorder觸發了事件，會在日志中輸出事件的資訊，后面在Run controller的時候還會用到這個廣播器，
整個Controller的啟動方法如下
func (p *PodController)Run(stopCh <-chan struct{}) {
glog.Info("Starting pod controller\n")
defer glog.Info("Shutting down pod controller\n")
if !controller.WaitForCacheSync("pod", stopCh, p.podListerSynced) {
return
}
if p.broadcaster != nil {
p.broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: v1core.New(p.kubeClient.CoreV1().RESTClient()).Events("")})
}
go wait.NonSlidingUntil(func() {
if err := p.reconcilePods(); err != nil {
glog.Errorf("Couldn't reconcile pod: %v", err)
}
}, metav1.Duration{Duration: 10 * time.Second}.Duration, stopCh)
<-stopCh
}
WaitForCacheSync用于同步指定資源的快取，這個快取后續會使用到，萬一同步失敗的話controller將不能運行
StartRecordingToSink用于把事件廣播到apiserver中，如果此處不執行，即便是recorder觸發了事件，apiserver沒收到這個事件，最終事件資訊沒保存到對應pod中，我們通過kubectl describe po時就會看不到相應的event記錄
通過啟動一個協程去定期執行reconcilePods()方法，NonSlidingUntil函式被呼叫后馬上執行傳進去的func，后續每隔10秒就重復執行一次，直到收到stopCh通道傳來的終止信號才停止，
最后通過等待接收stopCh傳來的信號而阻塞當前協程，從而阻止了完成本函式的呼叫
reconcilePods方法的大致邏輯如前所述，經過多次呼叫從lister中獲取所有pod,遍歷每個pod把pod的命名空間和pod的名稱列印出來,然后通過labelselector找出集群中的master節點將pod的數量打到名為hopegi/pod-count的label上，最后給每個pod的event事件添加一條pod count is n(這個n是pod的總數)這樣的記錄，
func (p *PodController)reconcilePods()error {
glog.Infof("reconcilePods ")
pods,err:= p.podLister.List(labels.Everything())
if err!=nil{
return fmt.Errorf("error listing pods: %v", err)
}
return p.reconcile(pods)
}
func (p *PodController)reconcile(pods []*v1.Pod)error {
glog.Infof("reconcile pods")
for _,pod :=range pods{
fmt.Printf("pod name is %s.%s \n",(*pod).Namespace,(*pod).Name)
}
nodes,err:= p.kubeClient.CoreV1().Nodes().List(metav1.ListOptions{LabelSelector:"node-role.kubernetes.io/master"})
if err!=nil{
glog.Infof("get master error %v\n",err)
return err
}
for _,n:=range nodes.Items{
n.Labels["hopegi/pod-count"]=fmt.Sprintf("%d",len(pods))
_,err= p.kubeClient.CoreV1().Nodes().Update(&n)
if err!=nil{
glog.Infof("label node error:%v ",err)
}
}
if p.recorder!=nil {
msg:=fmt.Sprintf("pod count is %d",len(pods))
for _, pod := range pods {
p.recorder.Eventf(&v1.ObjectReference{
Kind:"Pod",
Name:pod.Name,
UID:pod.UID,
Namespace:pod.Namespace,
},v1.EventTypeNormal,"SuccessCalculatePod",msg)
}
}
return nil
}
獲取集群里所有的pod用的是lister而不是通過kubeclient去獲取差別在于，作為某個K8S資源的informer(本例中是pod),它內部都有一個對應資源的快取，這個快取在listAndWatch機制的作用下與集群中存盤的pod資料保持一致，這個listAndWatch將在下一篇中介紹；后者是每訪問一次都會往apiserver中發一次請求，在眾多controller頻繁地跟apiserver通訊，apiserver會不堪重負，且消耗大量網路資源，獲取效率也低下，
小結
本篇簡單的實作了一個controller,雖然它并沒有如開篇所說的那樣對資源的實際狀態與期望狀態的差異進行調諧，通過最簡單的方式周期性地檢查pod的狀態，引入了informer,獲取了它listAndWatch的結果，后續將介紹這個informer的機制，它如何執行listAndWatch,一個較為常見的標準的Controller是如何實作的，
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/301635.html
標籤：Go