Istio Pilot 源码分析（一）

初始化 Environment

什么是 Environment 呢？根据定义 Environment 为 Pilot 提供了一个汇总的、运行中所需的 API 集合。 Environment 中字段（接口）如下：

type Environment struct {
  ServiceDiscovery  // 服务发现的接口模型，主要列出 services 和 instances
  IstioConfigStore  // Istio 配置文件的存储器，主要列出 ServiceEntry 等配置
  mesh.Watcher      // mesh config 文件的监听器
  mesh.NetworksWatcher // mesh network config 文件的监听器
  PushContext *PushContext // 在推送（下发 xDS）生成期间保存信息的上下文
  DomainSuffix string // istio server 默认的后缀域名
}

其中 PushContext 是 Pilot 在推送 xDS 前，生成配置期间保存相关信息的上下文的地方，在全量推送配置和配置发生改变时重置。它会保存所有的错误和统计信息，并缓存一些配置的计算信息。 ServiceDiscovery 提供了枚举 Istio 中服务和实例的方法。 mesh.Watcher 和 mesh.NetworksWatcher 负责监听 istiod 启动时挂载的两个配置文件，这两个配置文件是通过 configmap 映射到 Pod 的文件系统中的，监听器将在监听到配置文件变化时运行预先注册的 Handler 。文件挂载参考 istiod 的配置文件：

apiVersion: v1
kind: Pod
metadata:
  name: istiod-56c488887d-z9k5c
  namespace: istio-system
spec:
  containers:
    volumeMounts:
    - mountPath: /etc/istio/config
      name: config-volume
  volumes:
  - configMap:
      defaultMode: 420
      name: istio
    name: config-volume

相应的配置存储在 istio-system/istio 这个 configmap 中，里面保存了 mesh 和 meshNetworks 两种配置，样例如下:

apiVersion: v1
kind: ConfigMap
metadata:
  name: istio
  namespace: istio-system
data:
  mesh: |-
    accessLogEncoding: TEXT
    accessLogFile: ""
    accessLogFormat: ""
    defaultConfig:
      binaryPath: /usr/local/bin/mosn
      concurrency: 2
      configPath: ./etc/istio/proxy
    ...    
  meshNetworks: 'networks: {}'

再回头看 Environment 的初始化：

e := &model.Environment{
  PushContext:  model.NewPushContext(),
  DomainSuffix: args.RegistryOptions.KubeOptions.DomainSuffix,
}
ac := aggregate.NewController(aggregate.Options{
  MeshHolder: e,
})
e.ServiceDiscovery = ac

首先是初始化了一份 PushContext ，创建 PushContext 所需的各种列表和 Map 。 其次是初始化了一个聚合所有注册中心的 Controller 作为 Environment 中的 ServiceDiscovery 。 该 Controller 提供从所有注册中心（如 Kubernetes, Consul, MCP 等）获取服务和实例列表的方法。 这里传入了一个参数 MeshHolder 是想利用 Environment 中的 mesh.Watcher 将 mesh 这个配置同步过去。

初始化 Server

Server 的结构之前分析过，这里将之前初始化的 Environment 传入后，开始初始化 XDSServer 。

s := &Server{
  clusterID:       getClusterID(args),
  environment:     e,
  XDSServer:       xds.NewDiscoveryServer(e, args.Plugins), // 初始化 XDSServer
  fileWatcher:     filewatcher.NewWatcher(),
  httpMux:         http.NewServeMux(),
  monitoringMux:   http.NewServeMux(),
  readinessProbes: make(map[string]readinessProbe),
}

XDSServer 相关的代码在 istio/pilot/pkg/xds/discovery.go 中，对应为 DiscoveryServer ，该服务为 Envoy xDS APIs 的 gRPC 实现。 DiscoveryServer 关键定义如下：

type DiscoveryServer struct {
  Env *model.Environment   // 即上述 pilot server 中的 Environment
  ConfigGenerator core.ConfigGenerator  // 控制面 Istio 配置的生成器，如 VirtualService 等
  Generators map[string]model.XdsResourceGenerator // 针对不同配置类型的定制化生成器
  concurrentPushLimit chan struct{}
  // 不同服务所有实例的集合，增量更新，key 为 service 和 namespace
  // EndpointShards 中是以不同的注册中心名为 key 分组保存实例
  EndpointShardsByService map[string]map[string]*EndpointShards 
  pushChannel chan *model.PushRequest  // 接收 push 请求的 channel
  pushQueue *PushQueue     // 防抖之后，真正 Push xDS 之前所用的缓冲队列
  adsClients      map[string]*Connection  // ADS 和 EDS 的 gRPC 连接
  StatusReporter DistributionStatusCache  // 监听 xDS ACK 和连接断开
  // xDS 状态更新的生成器（更新 connect, disconnect, nacks, acks）
  // 状态更新后向所有 connection 推送 DiscoveryResponse
  InternalGen *InternalGen 
  serverReady bool     // 表示缓存已同步，server 可以接受请求
  debounceOptions debounceOptions  // 防抖设置
  cache Cache // xDS 资源的缓存，目前仅适用于 EDS，线程安全
}

初始化 MeshConfig 、 KubeClient 、 MeshNetworks 和 MeshHandlers

s.initMeshConfiguration(args, s.fileWatcher)
if err := s.initKubeClient(args); err != nil {
  return nil, fmt.Errorf("error initializing kube client: %v", err)
}
s.initMeshNetworks(args, s.fileWatcher)
s.initMeshHandlers()

这几个初始化函数比较好理解， initMeshConfiguration 和 initMeshNetworks 都是通过 fileWatcher 对 istiod 从 configmap 中挂载的两个配置文件 mesh 和 meshNetworks 进行监听。当配置文件发生变化时重载配置并触发相应的 Handlers 。

filewatcher 的代码在另一个管理通用工具包的项目里： github.com/istio/pkg/filewatcher ，感兴趣的同学可以再详细研究下，底层使用到了 fsnotify 这个库来推送文件变化事件。

initMeshHandlers 为上述两个配置文件注册了两个 Handler ，当配置文件发生变化时触发全量 xDS 下发。

初始化 Controllers

这部分比较核心，初始化了三种控制器分别处理证书、配置信息和注册信息，证书及安全相关的内容本篇先暂不讨论。主要来看 initConfigController 和 initServiceControllers 。

func (s *Server) initControllers(args *PilotArgs) error {
  log.Info("initializing controllers")
  if err := s.initCertController(args); err != nil {
    return fmt.Errorf("error initializing certificate controller: %v", err)
  }
  if err := s.initConfigController(args); err != nil {
    return fmt.Errorf("error initializing config controller: %v", err)
  }
  if err := s.initServiceControllers(args); err != nil {
    return fmt.Errorf("error initializing service controllers: %v", err)
  }
  return nil
}

配置信息大都是 Istio 定义的一系列 CRD（如 VirtualService 、 DestinationRules 等），一个控制面可以通过 MCP 同时接入多个 Kubernetes 之外的配置数据源，也可通过文件目录（主要用来调试）挂载，默认是读取 Kubernetes 中的配置数据：

func (s *Server) initK8SConfigStore(args *PilotArgs) error {
  configController, err := s.makeKubeConfigController(args)
  ...
  s.initStatusController(args, features.EnableStatus) // 初始化上面提到的 StatusReporter
  return nil
}

配置数据包括以下类型，具体每个类型的含义 Istio 官网都有介绍及用例，这里不再赘述：

// PilotServiceApi contains only collections used by Pilot, including experimental Service Api.
PilotServiceApi = collection.NewSchemasBuilder().
    MustAdd(IstioNetworkingV1Alpha3Destinationrules).
    MustAdd(IstioNetworkingV1Alpha3Envoyfilters).
    MustAdd(IstioNetworkingV1Alpha3Gateways).
    MustAdd(IstioNetworkingV1Alpha3Serviceentries).
    MustAdd(IstioNetworkingV1Alpha3Sidecars).
    MustAdd(IstioNetworkingV1Alpha3Virtualservices).
    MustAdd(IstioNetworkingV1Alpha3Workloadentries).
    MustAdd(IstioNetworkingV1Alpha3Workloadgroups).
    MustAdd(IstioSecurityV1Beta1Authorizationpolicies).
    MustAdd(IstioSecurityV1Beta1Peerauthentications).
    MustAdd(IstioSecurityV1Beta1Requestauthentications).
    MustAdd(K8SServiceApisV1Alpha1Gatewayclasses).
    MustAdd(K8SServiceApisV1Alpha1Gateways).
    MustAdd(K8SServiceApisV1Alpha1Httproutes).
    MustAdd(K8SServiceApisV1Alpha1Tcproutes).
    Build()

详细看下 initK8SConfigStore 中的 makeKubeConfigController 方法，这里初始化了一个处理 Istio CRDs 的 Client ，实现 ConfigStoreCache 这个接口中增删改查等方法。

func (s *Server) makeKubeConfigController(args *PilotArgs) (model.ConfigStoreCache, error) {
  c, err := crdclient.New(s.kubeClient, buildLedger(args.RegistryOptions), args.Revision, args.RegistryOptions.KubeOptions)
  if err != nil {
    return nil, err
  }
  return c, nil
}

Client 定义如下：

type Client struct {
  schemas collection.Schemas  // Istio CRDs shemas
  domainSuffix string
  configLedger ledger.Ledger
  revision string
  kinds map[resource.GroupVersionKind]*cacheHandler // 跟踪已知类型的所有缓存 handler
  queue queue.Instance
  istioClient istioclient.Interface
  serviceApisClient serviceapisclient.Interface
}

再依次对这些类型创建 Informer 开启监听。回到 initConfigController ，创建好 ConfigStore 之后，再对其进一步包装：

// 将所有 ConfigStore 聚合并缓存
aggregateConfigController, err := configaggregate.MakeCache(s.ConfigStores)
// 通过 s.configController 统一操作上面聚合的 ConfigStores
s.configController = aggregateConfigController
// 将其包装为 IstioConfigStore 传入 environment，便于操作 ServiceEntry/Gateway 等资源
// IstioConfigStore 会在之后的 ServiceEntryStore 中用到
s.environment.IstioConfigStore = model.MakeIstioStore(s.configController)

最后将该 Controller 的启动函数注册到 startFuncs 中：

s.addStartFunc(func(stop <-chan struct{}) error {
  go s.configController.Run(stop)
  return nil
})

再来看 initServiceControllers 处理服务发现的 Controller 初始化:

func (s *Server) initServiceControllers(args *PilotArgs) error {
  serviceControllers := s.ServiceController()
  for _, r := range args.RegistryOptions.Registries {
    // ...
    switch serviceRegistry {
    case serviceregistry.Kubernetes:
      if err := s.initKubeRegistry(serviceControllers, args); err != nil {
        return err
      }
    // ...
  }
  // ...
}

从之前初始化的 environment.ServiceDiscovery 中获取已注册的服务中心，如果是 Kubernetes 则执行 initKubeRegistry:

// initKubeRegistry creates all the k8s service controllers under this pilot
func (s *Server) initKubeRegistry(serviceControllers *aggregate.Controller, args *PilotArgs) (err error) {
  // ...
  log.Infof("Initializing Kubernetes service registry %q", args.RegistryOptions.KubeOptions.ClusterID)
  kubeRegistry := kubecontroller.NewController(s.kubeClient, args.RegistryOptions.KubeOptions)
  s.kubeRegistry = kubeRegistry
  serviceControllers.AddRegistry(kubeRegistry)
  return
}

进一步初始化 Kubernetes 注册中心，方法为 NewController ，先看一下这个 Controller 的结构：

type Controller struct {
  client kubernetes.Interface
  queue queue.Instance
  serviceInformer cache.SharedIndexInformer
  serviceLister   listerv1.ServiceLister
  endpoints kubeEndpointsController
  nodeInformer cache.SharedIndexInformer
  nodeLister   listerv1.NodeLister
  pods *PodCache
  metrics         model.Metrics
  networksWatcher mesh.NetworksWatcher
  xdsUpdater      model.XDSUpdater
  domainSuffix    string
  clusterID       string
  serviceHandlers  []func(*model.Service, model.Event)
  instanceHandlers []func(*model.ServiceInstance, model.Event)
  workloadHandlers []func(*model.WorkloadInstance, model.Event)
  sync.RWMutex
  servicesMap map[host.Name]*model.Service
  nodeSelectorsForServices map[host.Name]labels.Instance
  nodeInfoMap map[string]kubernetesNode
  externalNameSvcInstanceMap map[host.Name][]*model.ServiceInstance
  workloadInstancesByIP map[string]*model.WorkloadInstance
  ranger cidranger.Ranger
  networkForRegistry string
  once sync.Once
}

可以看到 Controller 对 Services 、 Nodes 、 Pods 等资源各自初始化了 Informer 、 Lister 以及对应的 Map，各类 Handlers 在 Informer 监听到增删改查时推送相应的事件到 queue ，再由 onServiceEvent 、 onNodeEvent 、 c.pods.onEvent 中更新对应的 Map 。

回到 initServiceControllers ，初始化完 Kubernetes 注册中心之后，还需要关注 Kubernetes 集群之外的服务，这些服务基本都是通过 ServiceEntry 注册到控制面的，所有 ServiceEntry 配置数据目前还都在之前初始化的 configController 配置中心控制器中，这里将 ServiceEntry 数据单独拎出来初始化一个 ServicEntry 注册中心，加入到 serviceControllers 中：

s.serviceEntryStore = serviceentry.NewServiceDiscovery(
  s.configController, s.environment.IstioConfigStore, s.XDSServer)
serviceControllers.AddRegistry(s.serviceEntryStore)

serviceEntryStore 相关的逻辑会在后续 xDS 下发流程的分析中再阐述。

最后将 serviceControllers 中所有的服务注册中心的 Controller 的启动函数都注册到 startFuncs 中:

s.addStartFunc(func(stop <-chan struct{}) error {
  go serviceControllers.Run(stop)
  return nil
})

// Run starts all the controllers
func (c *Controller) Run(stop <-chan struct{}) {

  for _, r := range c.GetRegistries() {
    go r.Run(stop)
  }

  <-stop
  log.Info("Registry Aggregator terminated")
}

初始化 RegistryEventHandlers

initRegistryEventHandlers 设置了三个事件处理器 serviceHandler 、 instanceHandler 和 configHandler 分别响应服务、实例和配置数据的更新事件。

serviceHandler 如下：

serviceHandler := func(svc *model.Service, _ model.Event) {
  pushReq := &model.PushRequest{
    Full: true,
    ConfigsUpdated: map[model.ConfigKey]struct{}{{
      Kind:      gvk.ServiceEntry,
      Name:      string(svc.Hostname),
      Namespace: svc.Attributes.Namespace,
    }: {}},
    Reason: []model.TriggerReason{model.ServiceUpdate},
  }
  s.XDSServer.ConfigUpdate(pushReq)
}
if err := s.ServiceController().AppendServiceHandler(serviceHandler); err != nil {
  return fmt.Errorf("append service handler failed: %v", err)
}

可以看到当服务本身发生变化时，会触发 xDS 的全量下发，所有与该服务相关的代理都会收到推送。

实例的变动也会触发 xDS 的全量下发，不过仅在连接 Consul 时生效。Kubernetes 和 MCP 这两种服务发现的场景下，更新事件的 Handler 是在别的地方注册的。

instanceHandler := func(si *model.ServiceInstance, _ model.Event) {
  // TODO: This is an incomplete code. This code path is called for consul, etc.
  // In all cases, this is simply an instance update and not a config update. So, we need to update
  // EDS in all proxies, and do a full config push for the instance that just changed (add/update only).
  s.EnvoyXdsServer.ConfigUpdate(&model.PushRequest{
    Full: true,
    ConfigsUpdated: map[model.ConfigKey]struct{}{{
      Kind:      gvk.ServiceEntry,
      Name:      string(si.Service.Hostname),
      Namespace: si.Service.Attributes.Namespace,
    }: {}},
    Reason: []model.TriggerReason{model.ServiceUpdate},
  })
}
// 跳过 Kubernetes 和 MCP
for _, registry := range s.ServiceController().GetRegistries() {
  // Skip kubernetes and external registries as they are handled separately
  if registry.Provider() == serviceregistry.Kubernetes ||
    registry.Provider() == serviceregistry.External {
    continue
  }
  if err := registry.AppendInstanceHandler(instanceHandler); err != nil {
    return fmt.Errorf("append instance handler to registry %s failed: %v", registry.Provider(), err)
  }
}

上一步初始化了 configController ，它操作的对象主要是像 VirtualService 、 DestinationRules 这些 Istio 定义的配置，这些配置的变化也会触发 xDS 的全量下发，所有与该配置相关的代理都会收到推送。不过 ServiceEntry 和 WorkloadEntry 除外，这两个资源的配置下发是由 ServiceEntryStore 管理的，之前在初始化 ServiceController 时定义的 s.serviceEntryStore 会处理，之后的篇幅再做详细介绍。

configHandler := func(_, curr model.Config, event model.Event) {
  pushReq := &model.PushRequest{
    Full: true,
    ConfigsUpdated: map[model.ConfigKey]struct{}{{
      Kind:      curr.GroupVersionKind,
      Name:      curr.Name,
      Namespace: curr.Namespace,
    }: {}},
    Reason: []model.TriggerReason{model.ConfigUpdate},
  }
  s.EnvoyXdsServer.ConfigUpdate(pushReq)
}

下面是跳过 ServiceEntry 和 WorkloadEntry 的代码：

for _, schema := range schemas {
  // This resource type was handled in external/servicediscovery.go, no need to rehandle here.
  if schema.Resource().GroupVersionKind() == collections.IstioNetworkingV1Alpha3Serviceentries.
    Resource().GroupVersionKind() {
    continue
  }
  if schema.Resource().GroupVersionKind() == collections.IstioNetworkingV1Alpha3Workloadentries.
    Resource().GroupVersionKind() {
    continue
  }

  s.configController.RegisterEventHandler(schema.Resource().GroupVersionKind(), configHandler)
}

初始化 DiscoveryService

func (s *Server) initDiscoveryService(args *PilotArgs) error {
  log.Infof("starting discovery service")
  // Implement EnvoyXdsServer grace shutdown
  s.addStartFunc(func(stop <-chan struct{}) error {
    s.EnvoyXdsServer.Start(stop)
    return nil
  })

  s.initGrpcServer(args.KeepaliveOptions)
  grpcListener, err := net.Listen("tcp", args.ServerOptions.GRPCAddr)
  if err != nil {
    return err
  }
  s.GRPCListener = grpcListener

  return nil
}

这里将 EnvoyXdsServer 的启动添加至 startFuncs 中，便于后续统一启动。并初始化 gRPC 服务器，监听对应的端口。

初始化 gRPC 服务器，并注册 xDS V2 和 xDS V3 的 ADS 服务到 gRPC 服务器上:

func (s *Server) initGrpcServer(options *istiokeepalive.Options) {
  grpcOptions := s.grpcServerOptions(options)
  s.grpcServer = grpc.NewServer(grpcOptions...)
  s.EnvoyXdsServer.Register(s.grpcServer)
  reflection.Register(s.grpcServer)
}

func (s *DiscoveryServer) Register(rpcs *grpc.Server) {
  // Register v2 and v3 servers
  discovery.RegisterAggregatedDiscoveryServiceServer(rpcs, s)
  discoveryv2.RegisterAggregatedDiscoveryServiceServer(rpcs, s.createV2Adapter())
}

可以看到 ADS 的 gRPC 服务包含两个流式方法，一个是全量推送，一个是增量推送。

var _AggregatedDiscoveryService_serviceDesc = grpc.ServiceDesc{
  ServiceName: "envoy.service.discovery.v3.AggregatedDiscoveryService",
  HandlerType: (*AggregatedDiscoveryServiceServer)(nil),
  Methods:     []grpc.MethodDesc{},
  Streams: []grpc.StreamDesc{
    {
      StreamName:    "StreamAggregatedResources",
      Handler:       _AggregatedDiscoveryService_StreamAggregatedResources_Handler,
      ServerStreams: true,
      ClientStreams: true,
    },
    {
      StreamName:    "DeltaAggregatedResources",
      Handler:       _AggregatedDiscoveryService_DeltaAggregatedResources_Handler,
      ServerStreams: true,
      ClientStreams: true,
    },
  },
  Metadata: "envoy/service/discovery/v3/ads.proto",
}

注册 kubeClient.RunAndWait

将 kubeClient.RunAndWait 方法注册至 startFuncs 中， RunAndWait 启动后所有 Informer 将开始缓存，并等待它们同步完成。之所以在最后运行，可以保证所有的 Informer 都已经注册。

if s.kubeClient != nil {
  s.addStartFunc(func(stop <-chan struct{}) error {
    s.kubeClient.RunAndWait(stop)
    return nil
  })
}