3.深入Istio：Pilot配置規則ConfigController
轉載請聲明出處哦~，本篇文章發佈於luozhiyun的博客：//www.luozhiyun.com
本文使用的Istio源碼是 release 1.5。
Config Controller用於管理各種配置數據，包括用戶創建的流量管理規則和策略。Istio目前支持三種類型的Config Controller：
- MCP：是一種網絡配置協議，用於隔離Pilot和底層平台（文件系統、K8s），使得Pilot無須感知底層平台的差異，從而達到解耦的目的。
- File：通過監視器周期性地讀取本地配置文件，將配置規則緩存在內存中，並維護配置的增加、更新、刪除事件，當緩存由變化的時候，異步通知執行事件回調。
- Kubernetes：基於k8s的Config發現利用了k8s Informer的監聽能力。在k8s集群中，Config以CustomResource的形式存在。通過監聽apiserver配置規則資源，維護所有資源的緩存Store，並觸發事件處理回調函數。
ConfigController初始化
ConfigController是在initConfigController中被初始化的，在initConfigController方法中會調用makeKubeConfigController進行controller的初始化。
func (s *Server) makeKubeConfigController(args *PilotArgs) (model.ConfigStoreCache, error) {
//創建configClient
configClient, err := controller.NewClient(args.Config.KubeConfig, "", collections.Pilot,
args.Config.ControllerOptions.DomainSuffix, buildLedger(args.Config), args.Revision)
if err != nil {
return nil, multierror.Prefix(err, "failed to open a config client.")
}
//創建controller，並為config資源設置監聽
return controller.NewController(configClient, args.Config.ControllerOptions), nil
}
func NewController(client *Client, options controller2.Options) model.ConfigStoreCache {
log.Infof("CRD controller watching namespaces %q", options.WatchedNamespace)
// The queue requires a time duration for a retry delay after a handler error
out := &controller{
client: client,
queue: queue.NewQueue(1 * time.Second),
kinds: make(map[resource.GroupVersionKind]*cacheHandler),
}
// add stores for CRD kinds
//獲取所有的CRD類型
for _, s := range client.Schemas().All() {
//為每一種Config資源都創建一個informer，監聽所有的Config資源
out.addInformer(s, options.WatchedNamespace, options.ResyncPeriod)
}
return out
}
初始化完controller之後會獲取所有的CRD類型，為每一種Config資源都創建一個informer，監聽所有的Config資源。
Pilot = collection.NewSchemasBuilder().
//MeshPolicy
MustAdd(IstioAuthenticationV1Alpha1Meshpolicies).
MustAdd(IstioAuthenticationV1Alpha1Policies).
MustAdd(IstioConfigV1Alpha2Httpapispecbindings).
MustAdd(IstioConfigV1Alpha2Httpapispecs).
MustAdd(IstioMixerV1ConfigClientQuotaspecbindings).
MustAdd(IstioMixerV1ConfigClientQuotaspecs).
//DestinationRule
MustAdd(IstioNetworkingV1Alpha3Destinationrules).
//EnvoyFilter
MustAdd(IstioNetworkingV1Alpha3Envoyfilters).
//Gateway
MustAdd(IstioNetworkingV1Alpha3Gateways).
//ServiceEntry
MustAdd(IstioNetworkingV1Alpha3Serviceentries).
//Sidecar
MustAdd(IstioNetworkingV1Alpha3Sidecars).
//VirtualService
MustAdd(IstioNetworkingV1Alpha3Virtualservices).
MustAdd(IstioRbacV1Alpha1Clusterrbacconfigs).
MustAdd(IstioRbacV1Alpha1Rbacconfigs).
MustAdd(IstioRbacV1Alpha1Servicerolebindings).
MustAdd(IstioRbacV1Alpha1Serviceroles).
MustAdd(IstioSecurityV1Beta1Authorizationpolicies).
MustAdd(IstioSecurityV1Beta1Peerauthentications).
MustAdd(IstioSecurityV1Beta1Requestauthentications).
Build()
這裡定義好了所有要用到的Config資源類型，主要涉及網絡配置、認證、鑒權、策略管理等。
ConfigController事件處理
下面我們看一下controller定義：
type controller struct {
client *Client
queue queue.Instance
kinds map[resource.GroupVersionKind]*cacheHandler
}
client是調用controller.NewClient初始化的client；queue會在Informer監聽到資源的變動的時候將數據push到隊列中，controller在調用run方法的時候單獨運行一個線程運行queue中的函數；kinds在調用addInformer方法的時候初始化進去。
queue.Instance的定義如下：
type Task func() error
type Instance interface {
Push(task Task)
Run(<-chan struct{})
}
type queueImpl struct {
delay time.Duration
tasks []Task
cond *sync.Cond
closing bool
}
queueImpl繼承了Instance接口，在調用push方法的時候，會將Task放入到tasks數組中，並在調用Run方法的時候消費數組中的數據。
controller繼承了ConfigStoreCache接口：
type ConfigStoreCache interface {
ConfigStore
// 註冊規則事件處理函數
RegisterEventHandler(kind resource.GroupVersionKind, handler func(Config, Config, Event))
// 運行
Run(stop <-chan struct{})
// 配置緩存是否已同步
HasSynced() bool
}
ConfigStoreCache通過RegisterEventHandler接口為上面提到的配置資源都註冊事件處理函數，通過Run方法啟動控制器。
func (c *controller) Run(stop <-chan struct{}) {
log.Infoa("Starting Pilot K8S CRD controller")
go func() {
cache.WaitForCacheSync(stop, c.HasSynced)
//單獨啟動一個線程運行queue裏面的函數
c.queue.Run(stop)
}()
for _, ctl := range c.kinds {
go ctl.informer.Run(stop)
}
<-stop
log.Info("controller terminated")
}
在調用Run方法的時候會單獨的啟動一個線程調用queue的Run方法消費隊列中的數據，並遍歷所有的配置信息，調用informer的Run方法開啟監聽。
監聽器的EventHandler通過如下代碼註冊：
func (c *controller) newCacheHandler(
schema collection.Schema,
o runtime.Object,
otype string,
resyncPeriod time.Duration,
lf cache.ListFunc,
wf cache.WatchFunc) *cacheHandler {
informer := cache.NewSharedIndexInformer(
&cache.ListWatch{ListFunc: lf, WatchFunc: wf}, o,
resyncPeriod, cache.Indexers{})
h := &cacheHandler{
c: c,
schema: schema,
informer: informer,
}
informer.AddEventHandler(
cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
incrementEvent(otype, "add")
//將ADD事件發送至隊列
c.queue.Push(func() error {
return h.onEvent(nil, obj, model.EventAdd)
})
},
UpdateFunc: func(old, cur interface{}) {
if !reflect.DeepEqual(old, cur) {
incrementEvent(otype, "update")
//將Update事件發送至隊列
c.queue.Push(func() error {
return h.onEvent(old, cur, model.EventUpdate)
})
} else {
incrementEvent(otype, "updatesame")
}
},
DeleteFunc: func(obj interface{}) {
incrementEvent(otype, "delete")
//將Delete事件發送至隊列
c.queue.Push(func() error {
return h.onEvent(nil, obj, model.EventDelete)
})
},
})
return h
}
當Config資源創建、更新、刪除時，EventHandler創建任務對象並將其發送到任務隊列中，然後由任務處理線程處理。當對應的事件被調用的時候會觸發onEvent方法，會調用到cacheHandler的onEvent方法，最後設置完畢後將cacheHandler返回，controller會將此cacheHandler設置到kinds數組中存下來。
下面我們看一下cacheHandler的定義：
type cacheHandler struct {
c *controller
schema collection.Schema
informer cache.SharedIndexInformer
handlers []func(model.Config, model.Config, model.Event)
}
cacheHandler在上面初始化的時候，會傳入對應的controller、Schema、informer，然後在調用configController的RegisterEventHandler方法的時候會初始化對應的configHandler。
configController的RegisterEventHandler方法會在初始化DiscoveryService的時候調用initEventHandlers方法進行初始化：
func (s *Server) initEventHandlers() error {
...
if s.configController != nil {
configHandler := func(old, curr model.Config, _ model.Event) {
pushReq := &model.PushRequest{
Full: true,
ConfigTypesUpdated: map[resource.GroupVersionKind]struct{}{curr.GroupVersionKind(): {}},
Reason: []model.TriggerReason{model.ConfigUpdate},
}
s.EnvoyXdsServer.ConfigUpdate(pushReq)
}
//遍歷所有的資源
for _, schema := range collections.Pilot.All() {
// This resource type was handled in external/servicediscovery.go, no need to rehandle here.
//ServiceEntry 這個資源不在這裡註冊，感興趣的朋友可以自己找一下
if schema.Resource().GroupVersionKind() == collections.IstioNetworkingV1Alpha3Serviceentries.
Resource().GroupVersionKind() {
continue
}
//註冊configHandler到configController中
s.configController.RegisterEventHandler(schema.Resource().GroupVersionKind(), configHandler)
}
}
return nil
}
initEventHandlers會調用collections.Pilot.All方法獲取所有的資源配置，然後遍歷調用RegisterEventHandler方法將configHandler函數註冊到cacheHandler的handlers中，至於configHandler函數做了什麼，我們到下一篇講XdsServer的時候再講。
這一部分的代碼是比較繞的，這裡畫個圖理解一下吧。
整個執行流程為：
總結
至此，ConfigController的核心原理及工作流程就介紹完畢了。本篇主要講解了我們常用的Istio的Gateway、DestinationRule及VirtualService等配置是如何被Istio監聽到並作出相應改變的。希望大家能有所收穫。
Reference
//ruofeng.me/2018/11/08/how-does-istio-pilot-push-eds-config/
//zhaohuabing.com/post/2019-10-21-pilot-discovery-code-analysi
//www.servicemesher.com/blog/envoy-proxy-config-deep-dive/
//www.cnblogs.com/163yun/p/8962278.html
