4.深入Istio原始碼：Pilot的Discovery Server如何執行xDS非同步分發
阿新 • • 發佈：2020-12-05
> 轉載請宣告出處哦~，本篇文章釋出於luozhiyun的部落格：https://www.luozhiyun.com
>
> 本文使用的Istio原始碼是 release 1.5。
## 介紹
Discovery Service主要為資料面（執行在 sidecar 中的 Envoy 等 proxy 元件）提供控制資訊服務。Discovery Service為資料面提供的資訊叫做xds ，這裡的 `x` 是一個代詞，在 Istio 中，xds 包括 `cds`(cluster discovery service)、`lds`(listener discovery service)、`rds`(route discovery service)、`eds`(endpoint discovery service)，而 `ads`(aggregated discovery service) 是對這些服務的一個統一封裝。
Discovery Service中主要包含下述邏輯：
* 啟動GRPC Server並接收來自Envoy端的連線請求；
* 接收Envoy端的xDS請求，從Config Controller和Service Controller中獲取配置和服務資訊，生成響應訊息傳送給Envoy；
* 監聽來自Config Controller的配置變化和來自Service Controller的服務變化訊息，並將配置和服務變化內容通過xDS介面推送到Envoy。
## Discovery Service初始化
從上面的流程圖可以知道，在呼叫NewServer建立XdsServer的時候會做很多初始化的工作。如初始化Pilot Server、網格初始化、初始化Istio Config的控制器、初始化Service Discovery的控制器等。我們下面列出和Discovery Service初始化相關的程式碼：
```go
func NewServer(args *PilotArgs) (*Server, error) {
//建立Pilot Server
s := &Server{
basePort: args.BasePort,
clusterID: getClusterID(args),
environment: e,
//初始化XdsServer
EnvoyXdsServer: envoyv2.NewDiscoveryServer(e, args.Plugins),
forceStop: args.ForceStop,
mux: http.NewServeMux(),
}
...
//初始化xDS服務端
if err := s.initDiscoveryService(args); err != nil {
return nil, fmt.Errorf("discovery service: %v", err)
}
...
}
```
從上面的程式碼可以看出XdsServer是通過呼叫NewDiscoveryServer方法來進行初始化的，返回的是一個DiscoveryServer例項，具體欄位的使用會在後面說到。
```go
type DiscoveryServer struct {
...
// Endpoint 的快取，以服務名和 namespace 作為索引，主要用於 EDS 更新
EndpointShardsByService map[string]map[string]*EndpointShards
//統一接收其他元件發來的 PushRequest 的 channel
pushChannel chan *model.PushRequest
updateMutex sync.RWMutex
//pushQueue 主要是在真正 xDS 推送前做防抖快取
pushQueue *PushQueue
}
```
建立完Server後會呼叫initDiscoveryService方法：
```go
func (s *Server) initDiscoveryService(args *PilotArgs) error {
...
//初始化Service Controller和Config Controller的Handler，用於informer回撥
if err := s.initEventHandlers(); err != nil {
return err
}
...
// 會在初始化完畢之後呼叫Start方法，啟動XdsServer
s.addStartFunc(func(stop <-chan struct{}) error {
s.EnvoyXdsServer.Start(stop)
return nil
})
//初始化Grpc Server服務，並註冊到XdsServer中
s.initGrpcServer(args.KeepaliveOptions)
s.httpServer = &http.Server{
Addr: args.DiscoveryOptions.HTTPAddr,
Handler: s.mux,
}
...
}
```
這個方法主要做了這麼幾件事：
1. 初始化各種回撥處理器；
2. 將XdsServer啟動函式新增到Server的startFuncs佇列中，會在初始化完畢之後呼叫；
3. 呼叫initGrpcServer方法初始化Grpc Server服務，並註冊到XdsServer中。
在初始化 grpcServer 的時候，呼叫了 `DiscoveryServer.Register()` 方法，向 grpcServer 註冊了以下幾個服務：
```go
func (s *DiscoveryServer) Register(rpcs *grpc.Server) {
//註冊的時候傳入grpc server 和 DiscoveryServer
ads.RegisterAggregatedDiscoveryServiceServer(rpcs, s)
}
```
DiscoveryServer實際上實現了AggregatedDiscoveryServiceServer介面：
```go
type AggregatedDiscoveryServiceServer interface {
// 全量 ADS Stream 介面
StreamAggregatedResources(AggregatedDiscoveryService_StreamAggregatedResourcesServer) error
// 增量 ADS Stream 介面
DeltaAggregatedResources(AggregatedDiscoveryService_DeltaAggregatedResourcesServer) error
}
}
```
StreamAggregatedResources 接收DiscoveryRequest ，返回 DiscoveryResponse 流，包含全量的 xDS 資料，DeltaAggregatedResources方法目前沒有具體實現。
大致呼叫流程如下：
## Discovery Service啟動
discoveryServer.Start方法還是在pilot discovery的main方法中被呼叫。main方法會在呼叫完bootstrap.NewServer方法後，進行Start方法的呼叫：
```go
discoveryCmd = &cobra.Command{
...
RunE: func(c *cobra.Command, args []string) error {
...
stop := make(chan struct{})
// 建立xDs伺服器
discoveryServer, err := bootstrap.NewServer(&serverArgs)
if err != nil {
return fmt.Errorf("failed to create discovery service: %v", err)
}
// 啟動伺服器
if err := discoveryServer.Start(stop); err != nil {
return fmt.Errorf("failed to start discovery service: %v", err)
}
...
return nil
},
}
```
在呼叫Start方法的時候會獲取到Server的startFuncs集合，然後依次執行裡面設定的函式：
```go
func (s *Server) Start(stop <-chan struct{}) error {
// Now start all of the components.
for _, fn := range s.startFuncs {
if err := fn(stop); err != nil {
return err
}
}
...
}
```
遍歷呼叫完畢後會分別啟動server Controller和config Controller的run函式，以及呼叫xdsServer的Start方法，Start方法主要分別啟動了三個執行緒：
```go
func (s *DiscoveryServer) Start(stopCh <-chan struct{}) {
adsLog.Infof("Starting ADS server")
go s.handleUpdates(stopCh)
go s.periodicRefreshMetrics(stopCh)
go s.sendPushes(stopCh)
}
```
比較重要的是handleUpdates方法和sendPushes方法。
handleUpdates方法主要是處理 pushChannel 中收到的推送請求，最後會呼叫startPush將資料推入到DiscoveryServer的pushQueue管道中；sendPushes方法則是獲取pushQueue管道中的資料，封裝成XdsEvent推入到XdsConnection的pushChannel進行非同步處理。
### handleUpdates
```go
func (s *DiscoveryServer) handleUpdates(stopCh <-chan struct{}) {
debounce(s.pushChannel, stopCh, s.Push)
}
func debounce(ch chan *model.PushRequest, stopCh <-chan struct{}, pushFn func(req *model.PushRequest)) {
var timeChan <-chan time.Time
var startDebounce time.Time
var lastConfigUpdateTime time.Time
pushCounter := 0
debouncedEvents := 0
var req *model.PushRequest
free := true
freeCh := make(chan struct{}, 1)
push := func(req *model.PushRequest) {
pushFn(req)
freeCh <- struct{}{}
}
pushWorker := func() {
eventDelay := time.Since(startDebounce)
quietTime := time.Since(lastConfigUpdateTime)
// debounceMax為10s ，debounceAfter為100毫秒
//延遲時間大於等於最大延遲時間 或者 靜默時間大於等於最小靜默時間
if eventDelay >