【kubernetes/k8s源码分析】istio pilot discovery源码分析

忘是亡心i 2022-03-25 07:24 181阅读 0赞

istio 1.0:  [https://github.com/istio/istio][https_github.com_istio_istio]

Pilot负责Envoy实例在Istio网格服务部署的生命周期，Pilot暴露的API是为了服务发现，动态更新负载均衡池和路由表

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70][]

# pilot总体架构 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70 1][]

官网pilot的架构图

### pilot-discovery功能 ###

>  *   从k8s list/watch `service`、`endpoint`、`pod`、`node`等资源，监听istio控制平面配置（K8s CRD）
>  *  翻译为Envoy可以理解的配置格式

Usage:  
  pilot-discovery discovery \[flags\]

## Server结构体 ##

pilot discovery运行的基本配置

// Server contains the runtime configuration for the Pilot discovery service.
    type Server struct {
    	HTTPListeningAddr       net.Addr
    	GRPCListeningAddr       net.Addr
    	SecureGRPCListeningAddr net.Addr
    	MonitorListeningAddr    net.Addr
    
    	// TODO(nmittler): Consider alternatives to exposing these directly
    	EnvoyXdsServer    *envoyv2.DiscoveryServer
    	ServiceController *aggregate.Controller
    
    	mesh             *meshconfig.MeshConfig
    	configController model.ConfigStoreCache
    	mixerSAN         []string
    	kubeClient       kubernetes.Interface
    	startFuncs       []startFunc
    	multicluster     *clusterregistry.Multicluster
    	httpServer       *http.Server
    	grpcServer       *grpc.Server
    	secureGRPCServer *grpc.Server
    	discoveryService *envoy.DiscoveryService
    	istioConfigStore model.IstioConfigStore
    	mux              *http.ServeMux
    	kubeRegistry     *kube.Controller
    }

# 1. main函数 #

路径pilot/cmd/pilot-discovery/main.go

func main() {
    	if err := rootCmd.Execute(); err != nil {
    		log.Errora(err)
    		os.Exit(-1)
    	}
    }

###    1.1 使用第三方命令行包cobra定义command ###

rootCmd = &cobra.Command{
    		Use:          "pilot-discovery",
    		Short:        "Istio Pilot.",
    		Long:         "Istio Pilot provides fleet-wide traffic management capabilities in the Istio Service Mesh.",
    		SilenceUsage: true,
    	}

###    1.2 定义rootCmd的子命令discoveryCmd ###

*      包括配置日志，discovery默认监听端口8080，monitor默认监听端口9093
 *      NewServer函数创建server实例（第2章节讲解）
 *      Start启动server实例运行主要执行体（第3章节讲解）

discoveryCmd = &cobra.Command{
    		Use:   "discovery",
    		Short: "Start Istio proxy discovery service.",
    		Args:  cobra.ExactArgs(0),
    		RunE: func(c *cobra.Command, args []string) error {
    			cmd.PrintFlags(c.Flags())
    			if err := log.Configure(loggingOptions); err != nil {
    				return err
    			}
    
    			// Create the stop channel for all of the servers.
    			stop := make(chan struct{})
    
    			// Apply deprecated flags if set to a value other than the default.
    			if serverArgs.DiscoveryOptions.HTTPAddr == ":8080" && httpPort != 8080 {
    				serverArgs.DiscoveryOptions.HTTPAddr = fmt.Sprintf(":%d", httpPort)
    			}
    			if serverArgs.DiscoveryOptions.MonitoringAddr == ":9093" && monitoringPort != 9093 {
    				serverArgs.DiscoveryOptions.MonitoringAddr = fmt.Sprintf(":%d", monitoringPort)
    			}
    
    			// Create the server for the discovery service.
    			discoveryServer, err := bootstrap.NewServer(serverArgs)
    			if err != nil {
    				return fmt.Errorf("failed to create discovery service: %v", err)
    			}
    
    			// Start the server
    			if err := discoveryServer.Start(stop); err != nil {
    				return fmt.Errorf("failed to start discovery service: %v", err)
    			}
    
    			cmd.WaitSignal(stop)
    			return nil
    		},

# 2. NewServer函数 #

NewServer创建一个Server实例

// NewServer creates a new Server instance based on the provided arguments.
    func NewServer(args PilotArgs) (*Server, error) {
    	// If the namespace isn't set, try looking it up from the environment.
    	if args.Namespace == "" {
    		args.Namespace = os.Getenv("POD_NAMESPACE")
    	}
    	if args.Config.ClusterRegistriesNamespace == "" {
    		if args.Namespace != "" {
    			args.Config.ClusterRegistriesNamespace = args.Namespace
    		} else {
    			args.Config.ClusterRegistriesNamespace = model.IstioSystemNamespace
    		}
    	}
    
    	s := &Server{}

###    2.1 initKubeClient函数    ###

serverArgs.Service.Registries默认设置为kubernetes，如果运行在k8s环境，则创建k8s client（默认kubeconfig文件为空，使用in cluster方式建立客户端连接）

// initKubeClient creates the k8s client if running in an k8s environment.
    func (s *Server) initKubeClient(args *PilotArgs) error {
    	if hasKubeRegistry(args) && args.Config.FileDir == "" {
    		client, kuberr := kubelib.CreateClientset(s.getKubeCfgFile(args), "")
    		if kuberr != nil {
    			return multierror.Prefix(kuberr, "failed to connect to Kubernetes API.")
    		}
    		s.kubeClient = client
    
    	}
    
    	return nil
    }

###    2.2 initMesh函数 ###

mesh默认配置/etc/istio/config/mesh

// initMesh creates the mesh in the pilotConfig from the input arguments.
    func (s *Server) initMesh(args *PilotArgs) error {
    	// If a config file was specified, use it.
    	if args.MeshConfig != nil {
    		s.mesh = args.MeshConfig
    		return nil
    	}
    	var mesh *meshconfig.MeshConfig
    	var err error
    
    	if args.Mesh.ConfigFile != "" {
    		mesh, err = cmd.ReadMeshConfig(args.Mesh.ConfigFile)
    		if err != nil {
    			log.Warnf("failed to read mesh configuration, using default: %v", err)
    		}
    	}
    
    	if mesh == nil {
    		// Config file either wasn't specified or failed to load - use a default mesh.
    		if _, mesh, err = GetMeshConfig(s.kubeClient, kube.IstioNamespace, kube.IstioConfigMap); err != nil {
    			log.Warnf("failed to read mesh configuration: %v", err)
    			return err
    		}
    
    		// Allow some overrides for testing purposes.
    		if args.Mesh.MixerAddress != "" {
    			mesh.MixerCheckServer = args.Mesh.MixerAddress
    			mesh.MixerReportServer = args.Mesh.MixerAddress
    		}
    	}
    
    	s.mesh = mesh
    	return nil
    }

###    2.3 initMixerSan函数 ###

前提是mesh已经被配置，配置mixerSAN

// initMixerSan configures the mixerSAN configuration item. The mesh must already have been configured.
    func (s *Server) initMixerSan(args *PilotArgs) error {
    	if s.mesh == nil {
    		return fmt.Errorf("the mesh has not been configured before configuring mixer san")
    	}
    	if s.mesh.DefaultConfig.ControlPlaneAuthPolicy == meshconfig.AuthenticationPolicy_MUTUAL_TLS {
    		s.mixerSAN = envoy.GetMixerSAN(args.Config.ControllerOptions.DomainSuffix, args.Namespace)
    	}
    	return nil
    }

###    2.4 initConfigController函数 ###

默认未空，makeKubeConfigController调用2.4.1讲解，可以使用文件保存，crd保存存于k8s的etcd中

// initConfigController creates the config controller in the pilotConfig.
    func (s *Server) initConfigController(args *PilotArgs) error {
    	if len(args.MCPServerAddrs) > 0 {
    		if err := s.initMCPConfigController(args); err != nil {
    			return err
    		}
    	} else if args.Config.Controller != nil {
    		s.configController = args.Config.Controller
    	} else if args.Config.FileDir != "" {
    		store := memory.Make(model.IstioConfigTypes)
    		configController := memory.NewController(store)
    
    		err := s.makeFileMonitor(args, configController)
    		if err != nil {
    			return err
    		}
    
    		s.configController = configController
    	} else {
    		controller, err := s.makeKubeConfigController(args)
    		if err != nil {
    			return err
    		}
    
    		s.configController = controller
    	}

**    2.4.1 makeKubeConfigController函数**

创建kube client，注册crd资源如下IstioConfigTypes定义

func (s *Server) makeKubeConfigController(args *PilotArgs) (model.ConfigStoreCache, error) {
    	kubeCfgFile := s.getKubeCfgFile(args)
    	configClient, err := crd.NewClient(kubeCfgFile, "", model.IstioConfigTypes, args.Config.ControllerOptions.DomainSuffix)
    	if err != nil {
    		return nil, multierror.Prefix(err, "failed to open a config client.")
    	}
    
    	if !args.Config.DisableInstallCRDs {
    		if err = configClient.RegisterResources(); err != nil {
    			return nil, multierror.Prefix(err, "failed to register custom resources.")
    		}
    	}
    
    	return crd.NewController(configClient, args.Config.ControllerOptions), nil
    }

**       IstioConfigTypes包含的资源**

路径istio/pilot/pkg/model/config.go

// IstioConfigTypes lists all Istio config types with schemas and validation
    	IstioConfigTypes = ConfigDescriptor{
    		VirtualService,
    		Gateway,
    		ServiceEntry,
    		DestinationRule,
    		EnvoyFilter,
    		HTTPAPISpec,
    		HTTPAPISpecBinding,
    		QuotaSpec,
    		QuotaSpecBinding,
    		AuthenticationPolicy,
    		AuthenticationMeshPolicy,
    		ServiceRole,
    		ServiceRoleBinding,
    		RbacConfig,
    		ClusterRbacConfig,
    	}

**     2.4.1.1 NewController函数**

创建controller，设置informer的watch机制，包括Add，Update，Delete回调函数，触发将入queue等待处理

>     informer.AddEventHandler(
>        cache.ResourceEventHandlerFuncs{
>           // TODO: filtering functions to skip over un-referenced resources (perf)
>           AddFunc: func(obj interface{}) {
>              k8sEvents.With(prometheus.Labels{"type": otype, "event": "add"}).Add(1)
>              c.queue.Push(kube.NewTask(handler.Apply, obj, model.EventAdd))
>           },
>           UpdateFunc: func(old, cur interface{}) {
>              if !reflect.DeepEqual(old, cur) {
>                 k8sEvents.With(prometheus.Labels{"type": otype, "event": "update"}).Add(1)
>                 c.queue.Push(kube.NewTask(handler.Apply, cur, model.EventUpdate))
>              } else {
>                 k8sEvents.With(prometheus.Labels{"type": otype, "event": "updateSame"}).Add(1)
>              }
>           },
>           DeleteFunc: func(obj interface{}) {
>              k8sEvents.With(prometheus.Labels{"type": otype, "event": "add"}).Add(1)
>              c.queue.Push(kube.NewTask(handler.Apply, obj, model.EventDelete))
>           },
>        })

// NewController creates a new Kubernetes controller for CRDs
    // Use "" for namespace to listen for all namespace changes
    func NewController(client *Client, options kube.ControllerOptions) model.ConfigStoreCache {
    	log.Infof("CRD controller watching namespaces %q", options.WatchedNamespace)
    
    	// Queue requires a time duration for a retry delay after a handler error
    	out := &controller{
    		client: client,
    		queue:  kube.NewQueue(1 * time.Second),
    		kinds:  make(map[string]cacheHandler),
    	}
    
    	// add stores for CRD kinds
    	for _, schema := range client.ConfigDescriptor() {
    		out.addInformer(schema, options.WatchedNamespace, options.ResyncPeriod)
    	}
    
    	return out
    }

2.4.2 待分析

// Defer starting the controller until after the service is created.
    	s.addStartFunc(func(stop <-chan struct{}) error {
    		go s.configController.Run(stop)
    		return nil
    	})
    
    	// If running in ingress mode (requires k8s), wrap the config controller.
    	if hasKubeRegistry(args) && s.mesh.IngressControllerMode != meshconfig.MeshConfig_OFF {
    		// Wrap the config controller with a cache.
    		configController, err := configaggregate.MakeCache([]model.ConfigStoreCache{
    			s.configController,
    			ingress.NewController(s.kubeClient, s.mesh, args.Config.ControllerOptions),
    		})
    		if err != nil {
    			return err
    		}
    
    		// Update the config controller
    		s.configController = configController
    
    		if ingressSyncer, errSyncer := ingress.NewStatusSyncer(s.mesh, s.kubeClient,
    			args.Namespace, args.Config.ControllerOptions); errSyncer != nil {
    			log.Warnf("Disabled ingress status syncer due to %v", errSyncer)
    		} else {
    			s.addStartFunc(func(stop <-chan struct{}) error {
    				go ingressSyncer.Run(stop)
    				return nil
    			})
    		}
    	}
    
    	// Create the config store.
    	s.istioConfigStore = model.MakeIstioStore(s.configController)

###    2.5 initServiceControllers ###

创建以及初始化service controller，根据不通的平台方式进行处理，主要分析k8s平台的处理2.5.1分析

*       Config则调用initConfigRegistry函数初始化
 *       Mock则调用initConfigRegistry函数初始化
 *       kubernetes调用createK8sServiceControllers处理
 *       consul调用initConsulRegistry

// initServiceControllers creates and initializes the service controllers
    func (s *Server) initServiceControllers(args *PilotArgs) error {
    	serviceControllers := aggregate.NewController()
    	registered := make(map[serviceregistry.ServiceRegistry]bool)
    	for _, r := range args.Service.Registries {
    		serviceRegistry := serviceregistry.ServiceRegistry(r)
    		if _, exists := registered[serviceRegistry]; exists {
    			log.Warnf("%s registry specified multiple times.", r)
    			continue
    		}
    		registered[serviceRegistry] = true
    		log.Infof("Adding %s registry adapter", serviceRegistry)
    		switch serviceRegistry {
    		case serviceregistry.ConfigRegistry:
    			s.initConfigRegistry(serviceControllers)
    		case serviceregistry.MockRegistry:
    			s.initMemoryRegistry(serviceControllers)
    		case serviceregistry.KubernetesRegistry:
    			if err := s.createK8sServiceControllers(serviceControllers, args); err != nil {
    				return err
    			}
    		case serviceregistry.ConsulRegistry:
    			if err := s.initConsulRegistry(serviceControllers, args); err != nil {
    				return err
    			}
    		case serviceregistry.MCPRegistry:
    			log.Infof("no-op: get service info from MCP ServiceEntries.")
    		default:
    			return multierror.Prefix(nil, "Service registry "+r+" is not supported.")
    		}
    	}

**    2.5.1 createK8sServiceControllers**

**    NewController函数创建Controller实例，创建informer机制，主要关注service endpoint node pod**

// createK8sServiceControllers creates all the k8s service controllers under this pilot
    func (s *Server) createK8sServiceControllers(serviceControllers *aggregate.Controller, args *PilotArgs) (err error) {
    	clusterID := string(serviceregistry.KubernetesRegistry)
    	log.Infof("Primary Cluster name: %s", clusterID)
    	kubectl := kube.NewController(s.kubeClient, args.Config.ControllerOptions)
    	s.kubeRegistry = kubectl
    	serviceControllers.AddRegistry(
    		aggregate.Registry{
    			Name:             serviceregistry.KubernetesRegistry,
    			ClusterID:        clusterID,
    			ServiceDiscovery: kubectl,
    			ServiceAccounts:  kubectl,
    			Controller:       kubectl,
    		})
    
    	return
    }

2.5.2 NewServiceDiscovery函数

创建serviceEntry服务，ServiceEntryStore与serviceEntry通讯以及监视change

// NewServiceDiscovery creates a new ServiceEntry discovery service
    func NewServiceDiscovery(callbacks model.ConfigStoreCache, store model.IstioConfigStore) *ServiceEntryStore {
    	c := &ServiceEntryStore{
    		serviceHandlers:  make([]serviceHandler, 0),
    		instanceHandlers: make([]instanceHandler, 0),
    		store:            store,
    		callbacks:        callbacks,
    		ip2instance:      map[string][]*model.ServiceInstance{},
    		instances:        map[string][]*model.ServiceInstance{},
    		updateNeeded:     true,
    	}
    	if callbacks != nil {
    		callbacks.RegisterEventHandler(model.ServiceEntry.Type, func(config model.Config, event model.Event) {
    			serviceEntry := config.Spec.(*networking.ServiceEntry)
    
    			// Recomputing the index here is too expensive.
    			c.changeMutex.Lock()
    			c.lastChange = time.Now()
    			c.updateNeeded = true
    			c.changeMutex.Unlock()
    
    			services := convertServices(serviceEntry, config.CreationTimestamp)
    			for _, handler := range c.serviceHandlers {
    				for _, service := range services {
    					go handler(service, event)
    				}
    			}
    
    			instances := convertInstances(serviceEntry, config.CreationTimestamp)
    			for _, handler := range c.instanceHandlers {
    				for _, instance := range instances {
    					go handler(instance, event)
    				}
    			}
    		})
    	}
    
    	return c
    }

**    2.5.3 service controller注册service entry，启动service controller的Run方法**

// add service entry registry to aggregator by default
    	serviceEntryRegistry := aggregate.Registry{
    		Name:             "ServiceEntries",
    		Controller:       serviceEntryStore,
    		ServiceDiscovery: serviceEntryStore,
    		ServiceAccounts:  serviceEntryStore,
    	}
    	serviceControllers.AddRegistry(serviceEntryRegistry)
    
    	s.ServiceController = serviceControllers
    
    	// Defer running of the service controllers.
    	s.addStartFunc(func(stop <-chan struct{}) error {
    		go s.ServiceController.Run(stop)
    		return nil
    	})

###    2.6 initDiscoveryService函数 ###

func (s *Server) initDiscoveryService(args *PilotArgs) error {
    	environment := &model.Environment{
    		Mesh:             s.mesh,
    		IstioConfigStore: s.istioConfigStore,
    		ServiceDiscovery: s.ServiceController,
    		ServiceAccounts:  s.ServiceController,
    		MixerSAN:         s.mixerSAN,
    	}
    
    	// Set up discovery service
    	discovery, err := envoy.NewDiscoveryService(
    		s.ServiceController,
    		s.configController,
    		environment,
    		args.DiscoveryOptions,
    	)

###    2.7 initMonitor函数 ###

// initMonitor initializes the configuration for the pilot monitoring server.
    func (s *Server) initMonitor(args *PilotArgs) error {
    	s.addStartFunc(func(stop <-chan struct{}) error {
    		monitor, addr, err := startMonitor(args.DiscoveryOptions.MonitoringAddr, s.mux)
    		if err != nil {
    			return err
    		}
    		s.MonitorListeningAddr = addr
    
    		go func() {
    			<-stop
    			err := monitor.Close()
    			log.Debugf("Monitoring server terminated: %v", err)
    		}()
    		return nil
    	})
    	return nil
    }

###     2.8 initClusterRegistries函数 ###

// initClusterRegistries starts the secret controller to watch for remote
    // clusters and initialize the multicluster structures.
    func (s *Server) initClusterRegistries(args *PilotArgs) (err error) {
    	if hasKubeRegistry(args) {
    
    		mc, err := clusterregistry.NewMulticluster(s.kubeClient,
    			args.Config.ClusterRegistriesNamespace,
    			args.Config.ControllerOptions.WatchedNamespace,
    			args.Config.ControllerOptions.DomainSuffix,
    			args.Config.ControllerOptions.ResyncPeriod,
    			s.ServiceController,
    			s.EnvoyXdsServer.ClearCacheFunc())
    
    		if err != nil {
    			log.Info("Unable to create new Multicluster object")
    			return err
    		}
    
    		s.multicluster = mc
    	}
    	return nil
    }

# 3. Start函数 #

启动使用addStartFunc注册的函数

// Start starts all components of the Pilot discovery service on the port specified in DiscoveryServiceOptions.
    // If Port == 0, a port number is automatically chosen. Content serving is started by this method,
    // but is executed asynchronously. Serving can be cancelled at any time by closing the provided stop channel.
    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
    		}
    	}
    
    	return nil
    }

### 总结： ###

*  从 Service provider（如kubernetes或者consul）获取 service 信息
 *  从 K8S API Server 中获取流量规则(定义的 CRD Resource)
 *  将服务信息和流量规则转化为数据面可以理解的格式，通过标准的数据面 API 下发到网格中的各个 sidecar 中

### 参考： ###

[https://istio.io/docs/concepts/traffic-management/][https_istio.io_docs_concepts_traffic-management]

[https_github.com_istio_istio]: https://github.com/istio/istio
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70]: /images/20220325/a649268abb734760ab5d2cfac229dd17.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70 1]: /images/20220325/f82e5b9fafba4e07bccecdcfaccfbc01.png
[https_istio.io_docs_concepts_traffic-management]: https://istio.io/docs/concepts/traffic-management/