Istio 服务注册插件机制代码解析

type Service struct {
        // Hostn/服务器名
        Hostname Hostname `json:"hostname"`

        // 虚拟 IP / 负载均衡器 IP
        Address string `json:"address,omitempty"`

        // 如果服务部署在多个集群中，ClusterVIPs 会保存不同集群中该服务对应的 VIP
        ClusterVIPs map[string]string `json:"cluster-vips,omitempty"`

        // 服务端口列表
        Ports PortList `json:"ports,omitempty"`

        // 运行该服务的服务账号
        ServiceAccounts []string `json:"serviceaccounts,omitempty"`

        // 该服务是否为一个“外部服务”，采用 ServiceEntry 定义的服务该标志为 true
        MeshExternal bool

        // 服务解析规则：包括 
        // ClientSideLB: 由 Envoy 代理根据其本地的 LB pool 进行请求路由
        // DNSLB: 查询 DNS 服务器得到 IP 地址，并将请求发到该 IP
        // Passthrough：将请求发转发到其原始目的地
        Resolution Resolution

        // 服务创建时间
        CreationTime time.Time `json:"creationTime,omitempty"`

        // 服务的一些附加属性
        Attributes ServiceAttributes
}

type ServiceInstance struct {
        // Endpoint 中包括服务实例的 IP：Port，UID 等
        Endpoint       NetworkEndpoint `json:"endpoint,omitempty"`
        // 对应的服务
        Service        *Service        `json:"service,omitempty"`
        // 该实例上的标签，例如版本号
        Labels         Labels          `json:"labels,omitempty"`
        // 运行该服务的服务账号
        ServiceAccount string          `json:"serviceaccount,omitempty"`
}

type Registry struct {
        // 注册表的类型，例如 Kubernetes, Consul, 等等。
        Name serviceregistry.ServiceRegistry
        // 某些类型的服务注册表支持多集群，例如 Kubernetes，在这种情况下需要用 CluterID 来区分同一类型下不同集群的服务注册表
        ClusterID string
        // 控制器，负责向外发送该 Registry 相关的 Service 变化消息
        model.Controller
        // 服务发现接口，用于获取注册表中的服务信息
        model.ServiceDiscovery
}

// ServiceRegistry defines underlying platform supporting service registry
type ServiceRegistry string

const (
        // MockRegistry，用于测试的服务注册表，包含两个硬编码的 test services
        MockRegistry ServiceRegistry = "Mock"
        // ConfigRegistry，可以从 Configstore 中获取定义的 service registry，加入到 Istio 的服务列表中
        KubernetesRegistry ServiceRegistry = "Kubernetes"
        // 从 Consul 获取服务数据的服务注册表
        ConsulRegistry ServiceRegistry = "Consul"
        // 采用“Mesh Configuration Protocol”的服务注册表
        MCPRegistry ServiceRegistry = "MCP"
)

type Controller interface {
        // 添加一个 Service Handler，服务的变化会通知到该 Handler
        AppendServiceHandler(f func(*Service, Event)) error

        // 添加一个 Service Instance Handler，服务实例的变化会通知到该 Handler
        AppendInstanceHandler(f func(*ServiceInstance, Event)) error

        // 启动 Controller 的主循环，对 Service Catalog 的变化进行分发
        Run(stop <-chan struct{})
}

type ServiceDiscovery interface {
        // 列出该 Service Registry 中的所有服务
        Services() ([]*Service, error)

        // 根据主机名查询服务
        // 该接口已废弃
        GetService(hostname Hostname) (*Service, error)

        // 根据主机名，服务端点和标签查询服务实例
        InstancesByPort(hostname Hostname, servicePort int, labels LabelsCollection) ([]*ServiceInstance, error)
		
		// 查询边车代理所在节点上的服务实例 
		GetProxyServiceInstances(*Proxy) ([]*ServiceInstance, error)
		
		// 获取边车代理所在的 Region,Zone 和 SubZone
        GetProxyLocality(*Proxy) string
        
		// 管理端口，Istio 生成的配置会将管理端口的流量排除，不进行路由处理
        ManagementPorts(addr string) PortList

        // 列出用于监控检查的探针
        WorkloadHealthCheckInfo(addr string) ProbeList
}

▼+Controller : struct
    [fields]
   -client : *api.Client
   -monitor : Monitor
    [methods]
   +AppendInstanceHandler(f func(*model.ServiceInstance, model.Event)) : error
   +AppendServiceHandler(f func(*model.Service, model.Event)) : error
   +GetIstioServiceAccounts(hostname model.Hostname, ports []int) : []string
   +GetProxyServiceInstances(node *model.Proxy) : []*model.ServiceInstance, error
   +GetService(hostname model.Hostname) : *model.Service, error
   +InstancesByPort(hostname model.Hostname, port int, labels model.LabelsCollection) : []*model.ServiceInstance, error
   +ManagementPorts(addr string) : model.PortList
   +Run(stop chan )
   +Services() : []*model.Service, error
   +WorkloadHealthCheckInfo(addr string) : model.ProbeList
   -getCatalogService(name string, q *api.QueryOptions) : []*api.CatalogService, error
   -getServices() : map[string][]string, error
    [functions]
   +NewController(addr string, interval time.Duration) : *Controller, error

//定时轮询 Consul Server Rest 接口，以获取服务数据变化
func (m *consulMonitor) run(stop <-chan struct{}) {
        ticker := time.NewTicker(m.period)
        for {
                select {
                case <-stop:
                        ticker.Stop()
                        return
                case <-ticker.C:
                        m.updateServiceRecord()
                        m.updateInstanceRecord()
                }
        }
}

//比较这一次和上一次的服务数据，如有变化则回调 ServiceHandler 进行通知
func (m *consulMonitor) updateServiceRecord() {
        svcs, _, err := m.discovery.Catalog().Services(nil)
        if err != nil {
                log.Warnf("Could not fetch services: %v", err)
                return
        }
        newRecord := consulServices(svcs)
        
        if !reflect.DeepEqual(newRecord, m.serviceCachedRecord) {
                // This is only a work-around solution currently
                // Since Handler functions generally act as a refresher
                // regardless of the input, thus passing in meaningless
                // input should make functionalities work
                //TODO
                obj := []*api.CatalogService{}
                var event model.Event
                for _, f := range m.serviceHandlers {
                        go func(handler ServiceHandler) {
                                if err := handler(obj, event); err != nil {
                                        log.Warnf("Error executing service handler function: %v", err)
                                }
                        }(f)
                }
                m.serviceCachedRecord = newRecord
        }
}