Istio技术与实践02：源码解析之Istio on Kubernetes 统一服务发现

前言

从场景和架构上看Istio和Kubernetes都是非常契合的一种搭配。

首先从场景上看Kuberntes为应用负载的部署、运维、扩缩容等提供了强大的支持。通过Service机制提供了负载间访问机制，通过域名结合Kubeproxy提供的转发机制可以方便的访问到对端的服务实例。因此如上图可以认为Kubernetes提供了一定的服务发现和负载均衡能力，但是较深入细致的流量治理能力，因为Kubnernetes所处的基础位置并未提供，而Istio正是补齐了这部分能力，两者的结合提供了一个端到端的容器服务运行和治理的解决方案。

从架构看Istio和Kubernetes更是深度的结合。 得益于Kuberntes Pod的设计，数据面的Sidecar作为一种高性能轻量的代理自动注入到Pod中和业务容器部署在一起，接管业务容器的inbound和outbound的流量，从而实现对业务容器中服务访问的治理。在控制面上Istio基于其Adapter机制集成Kubernetes的域名，从而避免了两套名字服务的尴尬场景。

在本文中将结合Pilot的代码实现来重点描述图中上半部分的实现，下半部分的内容Pilot提供的通用的API给Envoy使用可参照上一篇文章的DiscoverServer部分的描述。

理解了Pilot的ServiceDiscovery的Adapter的主流程后，了解这部分内容比较容易。Pilot-discovery在initServiceControllers时，根据服务注册配置的方式，如果是Kubernetes，则会走到这个分支来构造K8sServiceController。

case serviceregistry.KubernetesRegistry:

   s.createK8sServiceControllers(serviceControllers, args); err != nil {
   return err
}

创建controller其实就是创建了一个Kubenernetes的controller，可以看到List/Watch了Service、Endpoints、Node、Pod几个资源对象。

// NewControllercreates a new Kubernetes controller
func NewController(clientkubernetes.Interface, options ControllerOptions) *Controller {
   out := &Controller{
      domainSuffix: options.DomainSuffix,
      client:       client,
      queue:        NewQueue(1 * time.Second),
   }
   out.services =out.createInformer(&v1.Service{}, "Service", options.ResyncPeriod,
      func(opts meta_v1.ListOptions) (runtime.Object, error) {
         return client.CoreV1().Services(options.WatchedNamespace).List(opts)
      },
      func(opts meta_v1.ListOptions) (watch.Interface, error) {
         return client.CoreV1().Services(options.WatchedNamespace).Watch(opts)
      })
   out.endpoints =out.createInformer(&v1.Endpoints{}, "Endpoints", options.ResyncPeriod,
      func(opts meta_v1.ListOptions) (runtime.Object, error) {
         return client.CoreV1().Endpoints(options.WatchedNamespace).List(opts)
      },
      func(opts meta_v1.ListOptions) (watch.Interface, error) {
         return client.CoreV1().Endpoints(options.WatchedNamespace).Watch(opts)
      })
   out.nodes =out.createInformer(&v1.Node{}, "Node", options.ResyncPeriod,
      func(opts meta_v1.ListOptions) (runtime.Object, error) {
         return client.CoreV1().Nodes().List(opts)
      },
      func(opts meta_v1.ListOptions) (watch.Interface, error) {
         return client.CoreV1().Nodes().Watch(opts)
      })
   out.pods =newPodCache(out.createInformer(&v1.Pod{}, "Pod", options.ResyncPeriod,
      func(opts meta_v1.ListOptions) (runtime.Object, error) {
         return client.CoreV1().Pods(options.WatchedNamespace).List(opts)
      },
      func(opts meta_v1.ListOptions) (watch.Interface, error) {
         return client.CoreV1().Pods(options.WatchedNamespace).Watch(opts)
      }))

   return out
}

在 createInformer 中其实就是创建了SharedIndexInformer。这种方式在Kubernetes的各种Controller中广泛使用。Informer调用 APIserver的 List 和 Watch 两种类型的 API。在初始化的时，先调用 List API 获得全部资源对象，缓存在内存中; 然后，调用 Watch API 去 Watch这种这种资源对象，维护缓存。

Service informer := cache.NewSharedIndexInformer(
   &cache.ListWatch{ListFunc: lf, WatchFunc: wf}, o,
   resyncPeriod, cache.Indexers{})

下面看下Kubernetes场景下对ServiceDiscovery接口的实现。我们看下Kubernetes下提供的服务发现的接口，包括获取服务列表和服务实例列表。

func (c *Controller) GetService(hostname model.Hostname) (*model.Service, error) {
   name, namespace, err := parseHostname(hostname)
   item, exists := c.serviceByKey(name, namespace)
   svc := convertService(*item, c.domainSuffix)
   return svc, nil
}

最终是从infromer的缓存中获取Service资源对象。

func (c *Controller) serviceByKey(name, namespace string) (*v1.Service, bool) {
   item, exists, err := c.services.informer.GetStore().GetByKey(KeyFunc(name, namespace))
   return item.(*v1.Service), true
}

获取服务实例列表也是类似，也是从Informer的缓存中获取对应资源，只是涉及的对象和处理过程比Service要复杂一些。

func (c *Controller) InstancesByPort(hostname model.Hostname, reqSvcPort int,
   labelsList model.LabelsCollection) ([]*model.ServiceInstance, error) {
   // Get actual service by name
   name, namespace, err := parseHostname(hostname)
   item, exists := c.serviceByKey(name, namespace)
   svc := convertService(*item, c.domainSuffix)
   svcPortEntry, exists := svc.Ports.GetByPort(reqSvcPort)
   for _, item := range c.endpoints.informer.GetStore().List() {
      ep := *item.(*v1.Endpoints)
      …
   }
...
}
}
return nil, nil
}

可以看到就是做了如下的转换,将Kubernetes的对一个服务发现的数据结构转换成Istio的抽象模型对应的数据结构。

其实在conversion.go中提供了多个convert的方法将Kubernetes的数据对象转换成Istio的标准格式。除了上面的对Service、Instance的convert外，还包含对port，label、protocol的convert。如下面protocol的convert就值得一看。

func ConvertProtocol(name string, proto v1.Protocol) model.Protocol {
   out := model.ProtocolTCP
   switch proto {
   case v1.ProtocolUDP:
      out = model.ProtocolUDP
   case v1.ProtocolTCP:
      prefix := name
      i := strings.Index(name, "-")
      if i >= 0 {
         prefix = name[:i]
      }
      protocol := model.ParseProtocol(prefix)
      if protocol != model.ProtocolUDP && protocol != model.ProtocolUnsupported {
         out = protocol
      }
   }
   return out
}

看过Istio文档的都知道在使用Istio和Kuberntes结合的场景下创建Pod时要求满足4个约束。其中重要的一个是Port必须要有名，且Port的名字名字的格式有严格要求：Service 的端口必须命名，且端口的名字必须满足格式 <protocol>[-<suffix>]，例如name: http2-foo 。在K8s场景下这部分我们一般可以不对Pod命名的，看这段解析的代码可以看服务的Protocol是从name中解析出来的。如果Service的protocol是UDP的，则协议UDP；如果是TCP的，则会从名字中继续解析协议。如果名称是不可识别的前缀或者端口上的流量就会作为普通的 TCP 流量来处理。

另外同时在Informer 中添加对add、delete和update事件的回调，分别对应 informer 监听到创建、更新和删除这三种事件类型。可以看到这里是将待执行的回调操作包装成一个task，再压到Queue中，然后在Queue的run()方法中拿出去挨个执行，这部分不细看了。

到这里Kuberntes特有的服务发现能力就介绍完了。即kube\controller也实现了ServiceDiscovery中规定的服务发现的接口中定义的全部发方法。除了初始化了一个kube controller来从Kubeapiserver中获取和维护服务发现数据外，在pilot server初始化的时候，还有一个重要的initDiscoveryService初始化DiscoveryServer，这个discoveryserver使用contrller，其实是ServiceDiscovery上的服务发现供。发布成通用协议的接口，V1是rest，V2是gRPC,进而提供服务发现的能力给Envoy调用，这部分是Pilot服务发现的通用机制，在上篇文章的adapter机制中有详细描述，这里不再赘述。

以上介绍了istio基于Kubernetes的名字服务实现服务发现的机制和流程。整个调用关系如下，可以看到和其他的Adapter实现其实类似。