Istio Pilot 源码分析（二）

// istio/pilot/pkg/serviceregistry/serviceentry/servicediscovery.go:61// ServiceEntryStore communicates with ServiceEntry CRDs and monitors for changestype ServiceEntryStore struct {  XdsUpdater model.XDSUpdater  // 用来接收 EnvouXdsServer 的接口，主要用来 Push 相应的 xDS 更新请求  store      model.IstioConfigStore // 保存 ServiceEntry 实例的地方  storeMutex sync.RWMutex  // 读写 store 时需要的锁  // 以 hostname/namespace 以及类型（是服务还是实例）等作为索引的服务实例表  instances map[instancesKey]map[configKey][]*model.ServiceInstance  // seWithSelectorByNamespace 保存了每个 namespace 里所有的 ServiceEntry，也是作为一个索引供 handler 使用  seWithSelectorByNamespace map[string][]servicesWithEntry  refreshIndexes            bool  ...}

// WorkloadEntry 变化时的处理逻辑func (s *ServiceEntryStore) workloadEntryHandler(old, curr model.Config, event model.Event) {}// ServiceEntry 变化时的处理逻辑func (s *ServiceEntryStore) serviceEntryHandler(old, curr model.Config, event model.Event) {}

func NewServiceDiscovery(configController model.ConfigStoreCache, store model.IstioConfigStore, xdsUpdater model.XDSUpdater) *ServiceEntryStore {  s := &ServiceEntryStore{    XdsUpdater:            xdsUpdater,    store:                 store,    ip2instance:           map[string][]*model.ServiceInstance{},    instances:             map[instancesKey]map[configKey][]*model.ServiceInstance{},    workloadInstancesByIP: map[string]*model.WorkloadInstance{},    refreshIndexes:        true,  }  if configController != nil {    configController.RegisterEventHandler(gvk.ServiceEntry, s.serviceEntryHandler)    configController.RegisterEventHandler(gvk.WorkloadEntry, s.workloadEntryHandler)  }  return s}

type ServiceInstance struct {  Service     *Service       `json:"service,omitempty"`  ServicePort *Port          `json:"servicePort,omitempty"`  Endpoint    *IstioEndpoint `json:"endpoint,omitempty"`}

if event != model.EventDelete {  s.updateExistingInstances(key, instances)} else {  s.deleteExistingInstances(key, instances)}

// edsUpdate triggers an EDS update for the given instancesfunc (s *ServiceEntryStore) edsUpdate(instances []*model.ServiceInstance) {  allInstances := []*model.ServiceInstance{}  // Find all keys we need to lookup  keys := map[instancesKey]struct{}{}  for _, i := range instances {    keys[makeInstanceKey(i)] = struct{}{}  }  s.maybeRefreshIndexes()  s.storeMutex.RLock()  for key := range keys {    for _, i := range s.instances[key] {      allInstances = append(allInstances, i...)    }  }  s.storeMutex.RUnlock()  // This was a delete  if len(allInstances) == 0 {    for k := range keys {      _ = s.XdsUpdater.EDSUpdate(s.Cluster(), string(k.hostname), k.namespace, nil)    }    return  }  ...}

// edsUpdate triggers an EDS update for the given instancesfunc (s *ServiceEntryStore) edsUpdate(instances []*model.ServiceInstance) {  ...  endpoints := make(map[instancesKey][]*model.IstioEndpoint)  for _, instance := range allInstances {    port := instance.ServicePort    key := makeInstanceKey(instance)    endpoints[key] = append(endpoints[key],    &model.IstioEndpoint{      Address:         instance.Endpoint.Address,      EndpointPort:    instance.Endpoint.EndpointPort,      ServicePortName: port.Name,      Labels:          instance.Endpoint.Labels,      UID:             instance.Endpoint.UID,      ServiceAccount:  instance.Endpoint.ServiceAccount,      Network:         instance.Endpoint.Network,      Locality:        instance.Endpoint.Locality,      LbWeight:        instance.Endpoint.LbWeight,      TLSMode:         instance.Endpoint.TLSMode,    })  }  for k, eps := range endpoints {    _ = s.XdsUpdater.EDSUpdate(s.Cluster(), string(k.hostname), k.namespace, eps)  }}

func (s *ServiceEntryStore) workloadEntryHandler(old, curr model.Config, event model.Event) {  wle := curr.Spec.(*networking.WorkloadEntry)  key := configKey{    kind:      workloadEntryConfigType,    name:      curr.Name,    namespace: curr.Namespace,  }  ...  s.storeMutex.RLock()  // We will only select entries in the same namespace  entries := s.seWithSelectorByNamespace[curr.Namespace]  s.storeMutex.RUnlock()  // if there are no service entries, return now to avoid taking unnecessary locks  if len(entries) == 0 {    return  }  log.Debugf("Handle event %s for workload entry %s in namespace %s", event, curr.Name, curr.Namespace)  instances := []*model.ServiceInstance{}  for _, se := range entries {    workloadLabels := labels.Collection{wle.Labels}    if !workloadLabels.IsSupersetOf(se.entry.WorkloadSelector.Labels) {      // Not a match, skip this one      continue    }    instance := convertWorkloadEntryToServiceInstances(wle, se.services, se.entry)    instances = append(instances, instance...)  }  if event != model.EventDelete {    s.updateExistingInstances(key, instances)  } else {    s.deleteExistingInstances(key, instances)  }  s.edsUpdate(instances)}

func (s *DiscoveryServer) EDSUpdate(clusterID, serviceName string, namespace string,  istioEndpoints []*model.IstioEndpoint) error {  inboundEDSUpdates.Increment()  // 判断是否是全量下发  fp := s.edsUpdate(clusterID, serviceName, namespace, istioEndpoints)  s.ConfigUpdate(&model.PushRequest{    Full: fp,    ConfigsUpdated: map[model.ConfigKey]struct{}{{      Kind:      gvk.ServiceEntry,      Name:      serviceName,      Namespace: namespace,    }: {}},    Reason: []model.TriggerReason{model.EndpointUpdate},  })  return nil}

func (s *ServiceEntryStore) serviceEntryHandler(old, curr model.Config, event model.Event) {  cs := convertServices(curr)  configsUpdated := map[model.ConfigKey]struct{}{}  var addedSvcs, deletedSvcs, updatedSvcs, unchangedSvcs []*model.Service  ...}

switch event {case model.EventUpdate:  os := convertServices(old)  if selectorChanged(old, curr) {    // Consider all services are updated.    mark := make(map[host.Name]*model.Service, len(cs))    for _, svc := range cs {      mark[svc.Hostname] = svc      updatedSvcs = append(updatedSvcs, svc)    }    for _, svc := range os {      if _, f := mark[svc.Hostname]; !f {        updatedSvcs = append(updatedSvcs, svc)      }    }  } else {    addedSvcs, deletedSvcs, updatedSvcs, unchangedSvcs = servicesDiff(os, cs)  }case model.EventDelete:  deletedSvcs = cscase model.EventAdd:  addedSvcs = csdefault:  // this should not happen  unchangedSvcs = cs}

for _, svcs := range [][]*model.Service{addedSvcs, deletedSvcs, updatedSvcs} {  for _, svc := range svcs {    configsUpdated[model.ConfigKey{      Kind:      gvk.ServiceEntry,      Name:      string(svc.Hostname),      Namespace: svc.Attributes.Namespace}] = struct{}{}  }}

if len(unchangedSvcs) > 0 {  // If this service entry had endpoints with IPs (i.e. resolution STATIC), then we do EDS update.  // If the service entry had endpoints with FQDNs (i.e. resolution DNS), then we need to do  // full push (as fqdn endpoints go via strict_dns clusters in cds).  currentServiceEntry := curr.Spec.(*networking.ServiceEntry)  oldServiceEntry := old.Spec.(*networking.ServiceEntry)  if currentServiceEntry.Resolution == networking.ServiceEntry_DNS {    if !reflect.DeepEqual(currentServiceEntry.Endpoints, oldServiceEntry.Endpoints) {      // fqdn endpoints have changed. Need full push      for _, svc := range unchangedSvcs {        configsUpdated[model.ConfigKey{          Kind:      gvk.ServiceEntry,          Name:      string(svc.Hostname),          Namespace: svc.Attributes.Namespace}] = struct{}{}      }    }  }}

if len(unchangedSvcs) > 0 && !fullPush {  // IP endpoints in a STATIC service entry has changed. We need EDS update  // If will do full-push, leave the edsUpdate to that.  // XXX We should do edsUpdate for all unchangedSvcs since we begin to calculate service  // data according to this "configsUpdated" and thus remove the "!willFullPush" condition.  instances := convertInstances(curr, unchangedSvcs)  key := configKey{    kind:      serviceEntryConfigType,    name:      curr.Name,    namespace: curr.Namespace,  }  // If only instances have changed, just update the indexes for the changed instances.  s.updateExistingInstances(key, instances)  s.edsUpdate(instances)  return}

if fullPush {  // When doing a full push, for added and updated services trigger an eds update  // so that endpoint shards are updated.  var instances []*model.ServiceInstance  if len(addedSvcs) > 0 {    instances = append(instances, convertInstances(curr, addedSvcs)...)  }  if len(updatedSvcs) > 0 {    instances = append(instances, convertInstances(curr, updatedSvcs)...)  }  if len(unchangedSvcs) > 0 {    currentServiceEntry := curr.Spec.(*networking.ServiceEntry)    oldServiceEntry := old.Spec.(*networking.ServiceEntry)    // Non DNS service entries are sent via EDS. So we should compare and update if such endpoints change.    if currentServiceEntry.Resolution != networking.ServiceEntry_DNS {      if !reflect.DeepEqual(currentServiceEntry.Endpoints, oldServiceEntry.Endpoints) {        instances = append(instances, convertInstances(curr, unchangedSvcs)...)      }    }  }  s.edsUpdate(instances)  // If service entry is deleted, cleanup endpoint shards for services.  for _, svc := range deletedSvcs {    s.XdsUpdater.SvcUpdate(s.Cluster(), string(svc.Hostname), svc.Attributes.Namespace, model.EventDelete)  }  pushReq := &model.PushRequest{    Full:           true,    ConfigsUpdated: configsUpdated,    Reason:         []model.TriggerReason{model.ServiceUpdate},  }  s.XdsUpdater.ConfigUpdate(pushReq)}

// 相关源码目录kube: pilot/pkg/serviceregistry/kubemcp: pilot/pkg/serviceregistry/mcp