2.深入Istio：Pilot服务发现

luozhiyun

发布于 2020-11-24 14:51:07

1.4K0

发布于 2020-11-24 14:51:07

转载请声明出处哦~，本篇文章发布于luozhiyun的博客：https://www.luozhiyun.com 本文使用的Istio源码是 release 1.5。

介绍

pilot-discovery是在Pilot中的核心服务，在Pilot中名为pilot-discovery，主要功能是从注册中心（如 kubernetes 或者 consul）获取信息并汇集，从 Kubernetes API Server 中获取流量规则，并将服务信息和流量规则转化为数据面可以理解的格式，通过标准的数据面 API 下发到网格中的各个SideCar中。

pilot-discovery包含了服务发现、配置规则发现、xDS配置下发。总体上打算分三篇来进行讲解，这一篇主要看看服务发现部分的实现。文章中有涉及xDS协议的一些东西，大家可以看看这篇文章：深入解读Service Mesh背后的技术细节。

Pilot服务发现指通过监听底层平台的服务注册中心来缓存Istio服务模型，并且监视服务模型的变化，再服务模型更新时触发相关事件回调处理函数的执行。

服务发现工作机制

Pilot初始化

	discoveryCmd = &cobra.Command{
		Use:   "discovery",
		Short: "Start Istio proxy discovery service.",
		Args:  cobra.ExactArgs(0),
		RunE: func(c *cobra.Command, args []string) error {
			...
			//日志配置
			if err := log.Configure(loggingOptions); err != nil {
				return err
			} 
			... 
			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)
			}
			//等待进程推出
			cmd.WaitSignal(stop) 
			discoveryServer.WaitUntilCompletion()
			return nil
		},
	}

Pilot服务在初始化的时候首先会初始化日志配置，然后创建xDs服务器，这里的xDs指的是x Discovery Service的意思，x代表了一系列的组件如：Cluster、Endpoint、Listener、Route 等。

func NewServer(args *PilotArgs) (*Server, error) {
	 
	args.Default()
	e := &model.Environment{
		ServiceDiscovery: aggregate.NewController(),
		PushContext:      model.NewPushContext(),
	}

	s := &Server{
		basePort:       args.BasePort,
		clusterID:      getClusterID(args),
		environment:    e,
		EnvoyXdsServer: envoyv2.NewDiscoveryServer(e, args.Plugins),
		forceStop:      args.ForceStop,
		mux:            http.NewServeMux(),
	}
 
	// 初始化处理Istio Config的控制器
	if err := s.initConfigController(args); err != nil {
		return nil, fmt.Errorf("config controller: %v", err)
	}
	// 初始化处理Service Discovery的控制器
	if err := s.initServiceControllers(args); err != nil {
		return nil, fmt.Errorf("service controllers: %v", err)
	} 
	... 
	//初始化xDS服务端
	if err := s.initDiscoveryService(args); err != nil {
		return nil, fmt.Errorf("discovery service: %v", err)
	}
	... 
	// Webhook 回调服务
	if err := s.initHTTPSWebhookServer(args); err != nil {
		return nil, fmt.Errorf("injectionWebhook server: %v", err)
	}
    //sidecar注入相关
	if err := s.initSidecarInjector(args); err != nil {
		return nil, fmt.Errorf("sidecar injector: %v", err)
	}

	... 
	return s, nil
}

NewServer方法里面初始化了很多模块，这里挑相关的看看initConfigController是和配置服务相关的，我们之后再看，这里我们主要看initServiceControllers。

ServiceControllers

服务发现的主要逻辑在Pilot中由ServiceController（服务控制器）实现，通过监听底层平台的服务注册中心来缓存Istio服务模型，并监视服务模型的变化，在服务模型更新时触发相关事件回调处理函数的执行。

初始化

Controller的初始化执行流程很简单，这里用一张图来描述，initServiceControllers方法最后会调用到NewController方法来进行初始化。

func NewController(client kubernetes.Interface, options Options) *Controller {
	log.Infof("Service controller watching namespace %q for services, endpoints, nodes and pods, refresh %s",
		options.WatchedNamespace, options.ResyncPeriod)

	// The queue requires a time duration for a retry delay after a handler error
	// 初始化Controller
	c := &Controller{
		domainSuffix:               options.DomainSuffix,
		client:                     client,
		//控制器任务队列
		queue:                      queue.NewQueue(1 * time.Second),
		clusterID:                  options.ClusterID,
		xdsUpdater:                 options.XDSUpdater,
		servicesMap:                make(map[host.Name]*model.Service),
		externalNameSvcInstanceMap: make(map[host.Name][]*model.ServiceInstance),
		networksWatcher:            options.NetworksWatcher,
		metrics:                    options.Metrics,
	}
	//获取informer
	sharedInformers := informers.NewSharedInformerFactoryWithOptions(client, options.ResyncPeriod, informers.WithNamespace(options.WatchedNamespace))
	//注册 informer处理器
	c.services = sharedInformers.Core().V1().Services().Informer()
	//Services Handler
	registerHandlers(c.services, c.queue, "Services", c.onServiceEvent)
	//endpoints Handler
	switch options.EndpointMode {
	case EndpointsOnly:
		c.endpoints = newEndpointsController(c, sharedInformers)
	case EndpointSliceOnly:
		c.endpoints = newEndpointSliceController(c, sharedInformers)
	}
	//Nodes Handler
	c.nodes = sharedInformers.Core().V1().Nodes().Informer()
	registerHandlers(c.nodes, c.queue, "Nodes", c.onNodeEvent)

	podInformer := sharedInformers.Core().V1().Pods().Informer()
	c.pods = newPodCache(podInformer, c)
	//Pods Handler
	registerHandlers(podInformer, c.queue, "Pods", c.pods.onEvent)

	return c
}

NewController方法里面首先是初始化Controller，然后获取informer后分别注册Services Handler、endpoints Handler、Nodes Handler、Pods Handler。

核心功能就是监听k8s相关资源（Service、Endpoint、Pod、Node）的更新事件，执行相应的事件处理回调函数。

这里的Controller结构体实现了Controller接口：

type Controller interface {
	// AppendServiceHandler notifies about changes to the service catalog.
	AppendServiceHandler(f func(*Service, Event)) error

	// AppendInstanceHandler notifies about changes to the service instances
	// for a service.
	AppendInstanceHandler(f func(*ServiceInstance, Event)) error

	// Run until a signal is received
	Run(stop <-chan struct{})
}

再注册完毕后会调用其Run方法异步执行。

//异步调用Run方法
go serviceControllers.Run(stop)
//run方法里面会遍历GetRegistries列表，并异步执行其Run方法
func (c *Controller) Run(stop <-chan struct{}) {

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

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

到这里ServiceController为四种资源分别创建了一个监听器，用于监听K8s的资源更新，并注册EventHandler。

Service处理器

func (c *Controller) onServiceEvent(curr interface{}, event model.Event) error {
	if err := c.checkReadyForEvents(); err != nil {
		return err
	}

	svc, ok := curr.(*v1.Service)
	if !ok {
		tombstone, ok := curr.(cache.DeletedFinalStateUnknown)
		if !ok {
			log.Errorf("Couldn't get object from tombstone %#v", curr)
			return nil
		}
		svc, ok = tombstone.Obj.(*v1.Service)
		if !ok {
			log.Errorf("Tombstone contained object that is not a service %#v", curr)
			return nil
		}
	}

	log.Debugf("Handle event %s for service %s in namespace %s", event, svc.Name, svc.Namespace)
	//将k8s service 转换成 istio service
	svcConv := kube.ConvertService(*svc, c.domainSuffix, c.clusterID)
	//根据事件类型处理事件
	switch event {
	//删除事件
	case model.EventDelete:
		c.Lock()
		delete(c.servicesMap, svcConv.Hostname)
		delete(c.externalNameSvcInstanceMap, svcConv.Hostname)
		c.Unlock()
		// EDS needs to just know when service is deleted.
		//更新服务缓存
		c.xdsUpdater.SvcUpdate(c.clusterID, svc.Name, svc.Namespace, event)
	default:
		// instance conversion is only required when service is added/updated.
		instances := kube.ExternalNameServiceInstances(*svc, svcConv)
		c.Lock()
		c.servicesMap[svcConv.Hostname] = svcConv
		if instances == nil {
			delete(c.externalNameSvcInstanceMap, svcConv.Hostname)
		} else {
			c.externalNameSvcInstanceMap[svcConv.Hostname] = instances
		}
		c.Unlock()
		//更新服务缓存
		c.xdsUpdater.SvcUpdate(c.clusterID, svc.Name, svc.Namespace, event)
	}

	// Notify service handlers.
	// 触发XDS事件处理器
	for _, f := range c.serviceHandlers {
		f(svcConv, event)
	}

	return nil
}

Service事件处理器会将根据事件的类型更新缓存，然后调用serviceHandlers的事件处理器进行回调。serviceHandlers事件处理器是在初始化DiscoveryService的时候设置的。

serviceHandler := func(svc *model.Service, _ model.Event) {
		pushReq := &model.PushRequest{
			Full:               true,
			NamespacesUpdated:  map[string]struct{}{svc.Attributes.Namespace: {}},
			ConfigTypesUpdated: map[resource.GroupVersionKind]struct{}{collections.IstioNetworkingV1Alpha3Serviceentries.Resource().GroupVersionKind(): {}},
			Reason:             []model.TriggerReason{model.ServiceUpdate},
		}
        //配置更新
		s.EnvoyXdsServer.ConfigUpdate(pushReq)
	}

Endpoint处理器

Endpoint处理器会在调用newEndpointsController创建endpointsController的时候进行注册

func newEndpointsController(c *Controller, sharedInformers informers.SharedInformerFactory) *endpointsController {
	informer := sharedInformers.Core().V1().Endpoints().Informer()
	out := &endpointsController{
		kubeEndpoints: kubeEndpoints{
			c:        c,
			informer: informer,
		},
	}
	//注册处理器
	out.registerEndpointsHandler()
	return out
}

在回调的时候会调用到endpointsController的onEvent方法：

func (e *endpointsController) onEvent(curr interface{}, event model.Event) error {
	... 
	return e.handleEvent(ep.Name, ep.Namespace, event, curr, func(obj interface{}, event model.Event) {
		ep := obj.(*v1.Endpoints)
		//EDS更新处理
		e.c.updateEDS(ep, event)
	})
}

这里会调用updateEDS进行EDS（Endpoint Discovery service）更新处理。

func (c *Controller) updateEDS(ep *v1.Endpoints, event model.Event) {
	hostname := kube.ServiceHostname(ep.Name, ep.Namespace, c.domainSuffix)

	endpoints := make([]*model.IstioEndpoint, 0)
	if event != model.EventDelete {
		for _, ss := range ep.Subsets {
			for _, ea := range ss.Addresses {
				//获取Endpoint对应的Pod实例
				pod := c.pods.getPodByIP(ea.IP)
				...   
				// 将Endpoint转换成Istio模型IstioEndpoint
				for _, port := range ss.Ports {
					endpoints = append(endpoints, &model.IstioEndpoint{
						Address:         ea.IP,
						EndpointPort:    uint32(port.Port),
						ServicePortName: port.Name,
						Labels:          labelMap,
						UID:             uid,
						ServiceAccount:  sa,
						Network:         c.endpointNetwork(ea.IP),
						Locality:        locality,
						Attributes:      model.ServiceAttributes{Name: ep.Name, Namespace: ep.Namespace},
						TLSMode:         tlsMode,
					})
				}
			}
		}
	} 
	//使用xdsUpdater更新EDS
	_ = c.xdsUpdater.EDSUpdate(c.clusterID, string(hostname), ep.Namespace, endpoints)
}

在这里会重新封装endpoints然后调用EDSUpdate进行更新。

func (s *DiscoveryServer) EDSUpdate(clusterID, serviceName string, namespace string,
	istioEndpoints []*model.IstioEndpoint) error {
	inboundEDSUpdates.Increment()
	s.edsUpdate(clusterID, serviceName, namespace, istioEndpoints, false)
	return nil
}

func (s *DiscoveryServer) edsUpdate(clusterID, serviceName string, namespace string,
	istioEndpoints []*model.IstioEndpoint, internal bool) { 
	s.mutex.Lock()
	defer s.mutex.Unlock()
	requireFull := false
 
	...
	//找到之前缓存的服务
	if _, f := s.EndpointShardsByService[serviceName]; !f { 
		s.EndpointShardsByService[serviceName] = map[string]*EndpointShards{}
	}
	ep, f := s.EndpointShardsByService[serviceName][namespace]
	//不存在则初始化
	if !f { 
		ep = &EndpointShards{
			Shards:          map[string][]*model.IstioEndpoint{},
			ServiceAccounts: map[string]bool{},
		}
		s.EndpointShardsByService[serviceName][namespace] = ep
		if !internal {
			adsLog.Infof("Full push, new service %s", serviceName)
			requireFull = true
		}
	} 
	... 
	ep.mutex.Lock()
	ep.Shards[clusterID] = istioEndpoints
	ep.ServiceAccounts = serviceAccounts
	ep.mutex.Unlock() 
	
	if !internal {
		var edsUpdates map[string]struct{}
		if !requireFull {
			edsUpdates = map[string]struct{}{serviceName: {}}
		}
		//配置更新
		s.ConfigUpdate(&model.PushRequest{
			Full:               requireFull,
			NamespacesUpdated:  map[string]struct{}{namespace: {}},
			ConfigTypesUpdated: map[resource.GroupVersionKind]struct{}{collections.IstioNetworkingV1Alpha3Serviceentries.Resource().GroupVersionKind(): {}},
			EdsUpdates:         edsUpdates,
			Reason:             []model.TriggerReason{model.EndpointUpdate},
		})
	}
}

edsUpdate方法里面实际上就是做了两件事，一是更新缓存，二是调用ConfigUpdate进行配置更新。

ConfigUpdate资源更新实际上就是通过事件分发执行xDS分发，这块的细节我们稍后再讲。