Envoy Ingress：Contour基本原理和源码分析

发表于

概述

  • contour 是一个开源的 k8s ingress controller
  • 使用 envoy 提供反向代理
  • 提供动态配置更新
  • 支持多集群网络代理

特性

  • 基于高性能的L7代理和负载均衡Envoy作为数据面
  • 灵活的架构
  • TLS安全代理

为什么选择Contour

  • 动态配置更新，最大限度减少连接中断
  • 安全的支持多集群，多配置
  • 使用 HTTPProxy CRD 增强 ingress 核心配置

安装

kubectl

1
2
3
4
bash复制代码# yaml 安装
kubectl apply -f https://projectcontour.io/quickstart/contour.yaml
# 验证
kubectl get pod -n projectcontour

helm

1
2
3
4
bash复制代码# 添加仓库
helm repo add bitnami https://charts.bitnami.com/bitnami  
helm repo update
helm install contour --namespace projectcontour bitnami/contour

资源

部署完成后查看安装的资源信息，包括：

  • contour deployment：counter controller，作为控制面
  • envoy daemonset：envoy，作为数据面
1
2
3
4
5
6
7
8
9
10
bash复制代码$ kubectl get deploy,svc,ds,job -n projectcontour
NAME                      READY   UP-TO-DATE   AVAILABLE   AGE
deployment.apps/contour   2/2     2            2           12d

NAME              TYPE           CLUSTER-IP       EXTERNAL-IP 
service/contour   ClusterIP      172.20.126.255   <none>    8001/TCP                     12d
service/envoy     LoadBalancer   172.20.12.164    ...

NAME                   DESIRED   CURRENT   READY   UP-TO-DATE   AVAILABLE   NODE SELECTOR   AGE
daemonset.apps/envoy   10        10        10      10           10          <none>          12d

CRD

contour 安装完成后，查看 crd 有哪些：

1
2
3
4
5
6
bash复制代码$ kubectl get crd|grep contour
contourconfigurations.projectcontour.io                      2021-11-04T13:30:06Z
contourdeployments.projectcontour.io                         2021-11-04T13:30:06Z
extensionservices.projectcontour.io                          2021-11-04T13:30:06Z
httpproxies.projectcontour.io                                2021-11-04T13:30:06Z
tlscertificatedelegations.projectcontour.io                  2021-11-04T13:30:06Z

其中最核心的 CRD 是 HTTPProxy，等效于 k8s 提供的原生 Ingress，提供路由配置功能。

看一个简单的 httpproxy 配置示例，当访问 foo1.bar.com 时，将代理到后端 s1 服务的 80 端口

将 foo1.bar.com 域名和任意一个Node节点（daemonset部署）的ip做域名绑定后，即可通过域名访问服务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
yaml复制代码apiVersion: projectcontour.io/v1
kind: HTTPProxy
metadata:
  name: name-example-foo
  namespace: default
spec:
  virtualhost:
    fqdn: foo1.bar.com
  routes:
    - conditions:
      - prefix: /
      services:
        - name: s1
          port: 80

架构

概述

contour 主要包含两个组件：

  • Envoy：提供高性能的反向代理
  • Contour：充当 Envoy 的后台管理服务，提供相关配置

两个组件是分开部署的，Contour 以 Deployment 的方式部署，Envoy 以 DaemonSet 的方式部署

架构图

原理

数据面：envoy侧

envoy pod 中包含两个 container

  • initContainer container：执行 contour bootstrap 命令，生成一个配置文件到临时目录，envoy container共享这个目录，并作为启动的配置文件，里面配置了 contour server 地址，作为 envoy 的管理服务
  • envoy container：根据配置文件启动envoy，建立 GRPC连接，并从服务端接收配置请求

控制面：contour 侧

  • contour 是 k8s api 的一个客户端，通过 informer 机制监听 ingress，service，endpoint，secret等k8s原生资源，和httpproxy这个CRD
  • contour 缓存k8s的资源信息，并最终转换为 Envoy 需要的 XDS 信息。

原理图

其他

从图中我们还看到其他的一些信息

  • contour job：自动生成证书，使得grpc上的xds安全传输
  • contour 内部还提供了选主模式

HTTPProxy配置

概述

Ingress 资源在 k8s 1.1 版本引入，从那以后，Ingress API 保持相对稳定，需要使用特殊的能力需要借助于 Annotation 实现。

HttpProxy CRD 的目标是扩展 Ingress API 的能力，为用户提供更丰富的功能体验，以及解决 Ingress 在多租户环境下的限制。

对比 Ingress 的优势

  • 安全地支持多 Kubernetes 集群，能够限制哪些命名空间可以配置虚拟主机和 TLS 凭据。
  • 能够包括来自另一个 HTTPProxy 的路径或域的路由配置（这个HTTPProxy可能在另一个命名空间中）
  • 允许单个路由中接受多个服务并负载均衡它们之间的流量
  • 天生支持定义服务权重和负载策略，不需要使用注解
  • 支持创建时校验配置的合法性

配置文件对比

Ingress

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
yaml复制代码# ingress.yaml
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: basic
spec:
  rules:
  - host: foo-basic.bar.com
    http:
      paths:
      - backend:
          service:
            name: s1
            port:
              number: 80

HTTPProxy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
yaml复制代码# httpproxy.yaml
apiVersion: projectcontour.io/v1
kind: HTTPProxy
metadata:
  name: basic
spec:
  virtualhost:
    fqdn: foo-basic.bar.com
  routes:
    - conditions:
      - prefix: /
      services:
        - name: s1
          port: 80

源码分析

原理回顾

通过前面 contour 的基本原理和架构的介绍，总结如下：

数据流转原理

  • contour 通过 informer 机制监听 k8s 资源
  • 将k8s资源转换为 envoy 需要的 xds 资源
  • 通过grpc将xds配置下发到envoy节点

流量流转原理

  • 带有域名的请求经过任意一个Node节点，进入Envoy（以 hostNetWork=true 部署方式为例）
  • envoy将请求路由到 HTTPProxy 中配置的后端 service

数据流转实现

概述

实际上在将 k8s 资源转换为最终 xds 资源的过程中，还经过了其他的数据转换。

高清图

contour数据流转图

  • k8s资源变化，触发 informer 通知机制，注册的回调函数将事件放入 channel
  • EventHandler协程消费 channel，将channel中的每种资源以一个map形式保存在KubernetesCache中，map的key是该资源的namespace + name唯一确定的
  • KubenetesCache中的资源，经过一序列 processor 的处理之后，生成 DAG
  • DAG变更，会通知一序列 Observer，依次调用 OnChange方法，传入DAG，每个 Observer对应一个 XDS协议，将DAG中需要的资源转换为XDS配置，保存在Cache中
  • GRPC读取Cache中的数据，通过Stream将XDS下发到Envoy。下发这一步使用了go-control-plane这个开源框架

源码调用图

源码调用高清图

源码调用图

ResourceEventHandler

k8s 提供了 informer 机制，可以 watch kubernetes资源的变更。contour中 watch 的资源包括两类：

k8s原生资源：

  • ingress
  • service
  • endpoint
  • namespace

crd资源：

  • httpproxy
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
go复制代码// contour/cmd/contour/serve.go
func doServe(log logrus.FieldLogger, ctx *serveContext) error {
  // 这里监听 contour 的 crd 资源, 比如：httpproxy
	for _, r := range k8s.DefaultResources() {
		inf, err := clients.InformerForResource(r)
		......
		inf.AddEventHandler(&dynamicHandler)
	}
  // 监听 ingress 资源
	for _, r := range k8s.IngressV1Resources() {
		if err := informOnResource(clients, r, &dynamicHandler); err != nil {
			log.WithError(err).WithField("resource", r).Fatal("failed to create informer")
		}
	}
  ......
  // 监听 secret 资源
	for _, r := range k8s.SecretsResources() {
    ......
		if err := informOnResource(clients, r, handler); err != nil {
			log.WithError(err).WithField("resource", r).Fatal("failed to create informer")
		}
	}

	// 监听 endpoint 资源
	for _, r := range k8s.EndpointsResources() {
		if err := informOnResource(clients, r, &k8s.DynamicClientHandler{
			Next: &contour.EventRecorder{
				Next:    endpointHandler,
				Counter: contourMetrics.EventHandlerOperations,
			},
			Converter: converter,
			Logger:    log.WithField("context", "endpointstranslator"),
		}); err != nil {
			log.WithError(err).WithField("resource", r).Fatal("failed to create informer")
		}
	}
  ......
}

不管是 原生资源，还是 crd 资源，Informer 监听的原理都是一样的，用户只需要编写事件处理函数，对应的接口是 ResourceEventHandler，该接口提供了增加、删除、更新回调函数。

1
2
3
4
5
6
go复制代码// k8s.io/client-go/tools/cache/controller.go
type ResourceEventHandler interface {
	OnAdd(obj interface{})
	OnUpdate(oldObj, newObj interface{})
	OnDelete(obj interface{})
}

Contour 中所有的资源复用了两个 ResourceEventHandler 的实现类：

  • EndpointsTranslator：处理 endpoint 资源
  • EventHandler：处理除了 endpoint 外的其他资源。

EventHandler 将得到的资源信息，放入channel中，由单独的协程去消费以解耦

1
2
3
4
5
6
7
8
9
10
11
go复制代码func (e *EventHandler) OnAdd(obj interface{}) {
	e.update <- opAdd{obj: obj}
}

func (e *EventHandler) OnUpdate(oldObj, newObj interface{}) {
	e.update <- opUpdate{oldObj: oldObj, newObj: newObj}
}

func (e *EventHandler) OnDelete(obj interface{}) {
	e.update <- opDelete{obj: obj}
}

放入channel的信息，由谁去消费呢？

监听 channel 的协程，拿到事件后，取出资源信息，保存到 KubernetesCache 中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
go复制代码// contour/cmd/contour/server.go
func doServe(log logrus.FieldLogger, ctx *serveContext) error {
	......
  // 启动协程
  g.Add(eventHandler.Start())
  ......
}

// counter/internal/contour/handler.go
func (e *EventHandler) Start() func(<-chan struct{}) error {
	e.update = make(chan interface{})
	return e.run
}

func (e *EventHandler) run(stop <-chan struct{}) error {
	......
  // 死循环，一直监听 channel 中的事件
	for {
		select {
		case op := <-e.update:
			if e.onUpdate(op) {
				......
			}
      ......
 }
  
// 处理事件
// 将资源保存到 e.Builder.Source
func (e *EventHandler) onUpdate(op interface{}) bool {
	switch op := op.(type) {
	case opAdd:
		return e.Builder.Source.Insert(op.obj)
	case opUpdate:
		......
		remove := e.Builder.Source.Remove(op.oldObj)
		insert := e.Builder.Source.Insert(op.newObj)
		return remove || insert
	case opDelete:
		return e.Builder.Source.Remove(op.obj)
	case bool:
		return op
	default:
		return false
	}
}

EndpointsTranslator 监听的 endpoint 资源，处理比较麻烦一些，以 add 为例分析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
go复制代码// contour/internal/xdscache/v3/endpointstranslator.go
func (e *EndpointsTranslator) OnAdd(obj interface{}) {
	switch obj := obj.(type) {
	case *v1.Endpoints:
    // 更新本地缓存
		e.cache.UpdateEndpoint(obj)
    // 合并资源
		e.Merge(e.cache.Recalculate())
		e.Notify()
		if e.Observer != nil {
      // 刷新保存快照
			e.Observer.Refresh()
		}
	default:
		e.Errorf("OnAdd unexpected type %T: %#v", obj, obj)
	}
}

KubernetesCache

类中定义了多个map类型的资源对象，分别存储不同的资源，key是 namespace 和 name 组成的 type.NamespacedName，将原生k8s中不同 namespace 的同一资源都存储到一个map中，多种资源的多个map，共同组成 KubernetesCache

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
go复制代码// contour/internal/dag/cache.go
type KubernetesCache struct {
	......
	ingresses                 map[types.NamespacedName]*networking_v1.Ingress
	ingressclass              *networking_v1.IngressClass
	httpproxies               map[types.NamespacedName]*contour_api_v1.HTTPProxy
	secrets                   map[types.NamespacedName]*v1.Secret
	tlscertificatedelegations map[types.NamespacedName]*contour_api_v1.TLSCertificateDelegation
	services                  map[types.NamespacedName]*v1.Service
	namespaces                map[string]*v1.Namespace
	gatewayclass              *gatewayapi_v1alpha1.GatewayClass
	gateway                   *gatewayapi_v1alpha1.Gateway
	httproutes                map[types.NamespacedName]*gatewayapi_v1alpha1.HTTPRoute
	tlsroutes                 map[types.NamespacedName]*gatewayapi_v1alpha1.TLSRoute
	tcproutes                 map[types.NamespacedName]*gatewayapi_v1alpha1.TCPRoute
	udproutes                 map[types.NamespacedName]*gatewayapi_v1alpha1.UDPRoute
	backendpolicies           map[types.NamespacedName]*gatewayapi_v1alpha1.BackendPolicy
	extensions                map[types.NamespacedName]*contour_api_v1alpha1.ExtensionService
  ......
}

// map中key的数据类型：由 namespace 和 name 共同决定
type NamespacedName struct {
	Namespace string
	Name      string
}

KubernetesCache 插入流程：

  • 首先判断资源类型
  • 将不同的资源放入不同的map
  • 有些资源的变更，还需要触发其他资源变更。比如：service变化了，对应的 ingress或者httpproxy也需要联动变化
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
go复制代码// contour/internal/dag/cache.go
func (kc *KubernetesCache) Insert(obj interface{}) bool {
  // 初始化 map，确保执行一次
	kc.initialize.Do(kc.init)
  // 根据不同的资源类型，放入不同的 map
	switch obj := obj.(type) {
	case *v1.Secret:
		......
    // 保存 secret 资源
		kc.secrets[k8s.NamespacedNameOf(obj)] = obj
		return kc.secretTriggersRebuild(obj)
	case *v1.Service:
		kc.services[k8s.NamespacedNameOf(obj)] = obj
		return kc.serviceTriggersRebuild(obj)
	case *v1.Namespace:
		kc.namespaces[obj.Name] = obj
		return true
	......
	}

	return false
}

DAG

构造好 KubernetesCache 之后，contour 通过 一些列的 processor，将 KubernetesCache 变成 DAG，DAG 代表了 k8s 不同资源关系的一个有向无环图。核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
go复制代码// contour/internal/contour/handler.go
func (e *EventHandler) rebuildDAG() {
  // 将 KubernetesCache 构建成 DAG
	latestDAG := e.Builder.Build()
  // DAG 变更时，通知所有的 Observer，
  // 每种 Observer 是一个 XDS 对象，取出 DAG 资源转换为 XDS 配置
	e.Observer.OnChange(latestDAG)
  // 更新 snapshot 信息
	for _, upd := range latestDAG.StatusCache.GetStatusUpdates() {
		e.StatusUpdater.Send(upd)
	}

}

DAG 数据结构如下，根节点 roots 是一个 Vertex 的列表，Vertex 只定义了Visit接口。即：所有实现了 Visit

1
2
3
4
5
6
7
8
9
10
11
12
13
go复制代码// contour/internal/dag/dag.go
type DAG struct {
	// StatusCache holds a cache of status updates to send.
	StatusCache status.Cache

	// roots are the root vertices of this DAG.
	roots []Vertex
}

// Vertex is a node in the DAG that can be visited.
type Vertex interface {
	Visit(func(Vertex))
}

build过程分析：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
go复制代码// contour/internal/dag/builder.go
func (b *Builder) Build() *DAG {
	// 先构造一个空的 DAG
  dag := DAG{
		StatusCache: status.NewCache(b.Source.ConfiguredGateway),
	}
 
  // 遍历所有的 processor，每个 processor 把不同的 k8s 资源变成 DAG 的一部分信息
  // 所有的 processor 共同构成完整的 DAG
	for _, p := range b.Processors {
		p.Run(&dag, &b.Source)
	}
	return &dag
}

processor 列表包括：

  • IngressProcessor
  • ExtensionServiceProcessor
  • HTTPProxyProcessor
  • GatewayAPIProcessor
  • ListenerProcessor
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
go复制代码func getDAGBuilder(ctx *serveContext, clients *k8s.Clients, clientCert, fallbackCert *types.NamespacedName, 	log logrus.FieldLogger) dag.Builder {

  dagProcessors := []dag.Processor{
      &dag.IngressProcessor{
        ...
      },
      &dag.ExtensionServiceProcessor{
        ...
      },
      &dag.HTTPProxyProcessor{
       ...
      },
    }
    if ctx.Config.GatewayConfig != nil && clients.ResourcesExist(k8s.GatewayAPIResources()...) {
      dagProcessors = append(dagProcessors, &dag.GatewayAPIProcessor{
        FieldLogger: log.WithField("context", "GatewayAPIProcessor"),
      })
    }
    ...
    dagProcessors = append(dagProcessors, &dag.ListenerProcessor{})
  ...
}

每个processor 实现 Run 方法，接收 DAG 和 KubernetesCache 参数，通过 KubernetesCache 修改 DAG，以 HTTPProxyProcessor 为例分析。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
go复制代码func (p *HTTPProxyProcessor) Run(dag *DAG, source *KubernetesCache) {
	...
  // 校验合法性
	for _, proxy := range p.validHTTPProxies() {
    // 计算路由规则
    // 入参为 crd 资源 HttpProxy
    // 内部经过各种计算，构造路由关系，最终保存在DAG中
		p.computeHTTPProxy(proxy)
	}

	for meta := range p.orphaned {
		proxy, ok := p.source.httpproxies[meta]
		if ok {
			pa, commit := p.dag.StatusCache.ProxyAccessor(proxy)
			pa.ConditionFor(status.ValidCondition).AddError(contour_api_v1.ConditionTypeOrphanedError,
				"Orphaned",
				"this HTTPProxy is not part of a delegation chain from a root HTTPProxy")
			commit()
		}
	}
}

DAG 中的节点（Vertex）有:

  • VirtualHost：对应 LDS
  • SecureVirtualHost：对应 LDS 和 CDS
  • ExtensionCluster：对应 CDS
  • ServiceCluster：对应 EDS
  • Cluster：对应 CDS，一部分SDS
  • Listener：对应 RDS

其中某些 Vectex 又包含子 Vectex，因此用图这种数据结构串联起来。

ResourceCache

前面分析 DAG 代码时，在生成 DAG 后，会通知所有的 Observer，即e.Observer.OnChange

1
2
3
4
5
6
7
8
9
10
11
12
go复制代码// contour/internal/contour/handler.go
func (e *EventHandler) rebuildDAG() {
  // 将 KubernetesCache 构建成 DAG
	latestDAG := e.Builder.Build()
  // DAG 变更时，通知所有的 Observer，
  // 每种 Observer 是一个 XDS 对象，取出 DAG 资源转换为 XDS 配置
	e.Observer.OnChange(latestDAG)
  // 更新 snapshot 信息
	for _, upd := range latestDAG.StatusCache.GetStatusUpdates() {
		e.StatusUpdater.Send(upd)
	}
}

这里的 e.Observer 是 一堆 Observer的组合

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
go复制代码// contour/cmd/contour/serve.go
func doServe(log logrus.FieldLogger, ctx *serveContext) error {
	 ...
  // ResoureCache 列表
  resources := []xdscache.ResourceCache{
		xdscache_v3.NewListenerCache(listenerConfig, ctx.statsAddr, ctx.statsPort),
		&xdscache_v3.SecretCache{},
		&xdscache_v3.RouteCache{},
		&xdscache_v3.ClusterCache{},
		endpointHandler,
	}
  ...
	eventHandler := &contour.EventHandler{
		...
    // 注册一个组合的 Observer
		Observer:        dag.ComposeObservers(append(xdscache.ObserversOf(resources), snapshotHandler)...),
		...
	}
}

// 组合多个Observer
func ComposeObservers(observers ...Observer) Observer {
	return ObserverFunc(func(d *DAG) {
		for _, o := range observers {
			o.OnChange(d)
		}
	})
}

ResourceCache 是一个接口，聚合了两个接口：dag.Observer 和 xds.Resource。该接口定义了如何处理 DAG，以及处理完成后如何返回给调用方。

  • Observer接口：
    • OnChange：参数为DAG，当 DAG 有任何变更，都会触发这个函数，可以拿到最新的DAG做处理
  • Resource接口：
    • Contents：返回xds资源的接口
    • Query：查询xds资源的接口
    • Register：注册xds资源的接口
    • TypeURL：xds资源类型
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
go复制代码// contour/internal/xdscache/resources.go
type ResourceCache interface {
	dag.Observer
	xds.Resource
}

// contour/internal/dag/dag.go
type Observer interface {
  // 当 DAG变更时，触发相应业务逻辑
	OnChange(*DAG)
}

// contour/internal/xds/resource.go
type Resource interface {
  // 返回全部资源内容
	// Contents returns the contents of this resource.
	Contents() []proto.Message
  
  // 通过给定的名称，返回对应的资源内容
	// Query returns an entry for each resource name supplied.
	Query(names []string) []proto.Message

  // 注册资源
	// Register registers ch to receive a value when Notify is called.
	Register(chan int, int, ...string)

  // 资源类型
	// TypeURL returns the typeURL of messages returned from Values.
	TypeURL() string
}

不同的 xds 资源类型都实现了ResourceCache 接口：

  • RouteCache：对应 RDS
  • ClusterCache：对应 CDS
  • ListenerCache：对应 LDS
  • SecretCache：对应 SDS
  • EndpointsTranslator：对应 EDS

转换后的 xds 资源都保存在 values 字段中。

  • Observer的OnChange接口：将 DAG 转换为 Envoy xds 配置保存在 values 中
  • Resource的接口：提供查询 values 的方法，即可得到 envoy 的xds配置
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
go复制代码// contour/internal/xdscache/v3/cluster.go
// ClusterCache manages the contents of the gRPC CDS cache.
type ClusterCache struct {
	mu     sync.Mutex
	values map[string]*envoy_cluster_v3.Cluster
	contour.Cond
}

// contour/internal/xdscache/v3/listener.go
// ListenerCache manages the contents of the gRPC LDS cache.
type ListenerCache struct {
	mu           sync.Mutex
	values       map[string]*envoy_listener_v3.Listener
	staticValues map[string]*envoy_listener_v3.Listener

	Config ListenerConfig
	contour.Cond
}

// contour/internal/xdscache/v3/route.go
// RouteCache manages the contents of the gRPC RDS cache.
type RouteCache struct {
	mu     sync.Mutex
	values map[string]*envoy_route_v3.RouteConfiguration
	contour.Cond
}

// contour/internal/xdscache/v3/secret.go
// SecretCache manages the contents of the gRPC SDS cache.
type SecretCache struct {
	mu     sync.Mutex
	values map[string]*envoy_tls_v3.Secret
	contour.Cond
}

// contour/internal/xdscache/v3/endpointstranslator.go
// A EndpointsTranslator translates Kubernetes Endpoints objects into Envoy
// ClusterLoadAssignment resources.
type EndpointsTranslator struct {
	// Observer notifies when the endpoints cache has been updated.
	Observer contour.Observer

	contour.Cond
	logrus.FieldLogger

	cache EndpointsCache

	mu      sync.Mutex // Protects entries.
	entries map[string]*envoy_endpoint_v3.ClusterLoadAssignment
}

以 RouteCache 的 OnChange 为例，查看实现逻辑

1
2
3
4
5
6
7
go复制代码// contour/internal/xdscache/v3/route.go
func (c *RouteCache) OnChange(root *dag.DAG) {
  // 读取 DAG 中的信息，生成 envoy 的 RDS 资源 envoy_route_v3.RouteConfiguration
	routes := visitRoutes(root)
  // 将配置文件放到 value 字段中，供 Query、Context 等方法调用时查看
	c.Update(routes)
}

DAG 转换为 RDS 的逻辑实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
go复制代码// contour/internal/xdscache/v3/route.go
func visitRoutes(root dag.Vertex) map[string]*envoy_route_v3.RouteConfiguration {
	// Collect the route configurations for all the routes we can
	// find. For HTTP hosts, the routes will all be collected on the
	// well-known ENVOY_HTTP_LISTENER, but for HTTPS hosts, we will
	// generate a per-vhost collection. This lets us keep different
	// SNI names disjoint when we later configure the listener.
  
  // http 和 https 需要做不同的处理
	rv := routeVisitor{
		routes: map[string]*envoy_route_v3.RouteConfiguration{
			ENVOY_HTTP_LISTENER: envoy_v3.RouteConfiguration(ENVOY_HTTP_LISTENER),
		},
	}

	rv.visit(root)

	for _, v := range rv.routes {
		sort.Stable(sorter.For(v.VirtualHosts))
	}

	return rv.routes
}

// contour/internal/xdscache/v3/route.go
// 从根节点开始查找所有的 Route，查找顺序： Listener -> VirtualHost(SecureVirtualHost) -> Route
func (v *routeVisitor) visit(vertex dag.Vertex) {
	switch l := vertex.(type) {
   // 处理 DAG 中的 Listener
	case *dag.Listener:
		l.Visit(func(vertex dag.Vertex) {
			switch vh := vertex.(type) {
      // 处理 VirtualHost
			case *dag.VirtualHost:
				v.onVirtualHost(vh)
      // 处理 SecureVirtualHost
			case *dag.SecureVirtualHost:
				v.onSecureVirtualHost(vh)
			default:
				// recurse
				vertex.Visit(v.visit)
			}
		})
	default:
		// recurse
		vertex.Visit(v.visit)
	}
}

// 查找 Listener.VirtualHost.Route
func (v *routeVisitor) onVirtualHost(vh *dag.VirtualHost) {
	var routes []*dag.Route

	vh.Visit(func(v dag.Vertex) {
		route, ok := v.(*dag.Route)
		if !ok {
			return
		}
    // 将所有的 route 收集到列表中
		routes = append(routes, route)
	})

	if len(routes) == 0 {
		return
	}

  // 定义了把 dag.Route 对象转换为 envoy_route_v3.Route 对象的方法
  // 将两种数据格式进行转换
	toEnvoyRoute := func(route *dag.Route) *envoy_route_v3.Route {
		...
		rt := &envoy_route_v3.Route{
			Match:  envoy_v3.RouteMatch(route),
			Action: envoy_v3.RouteRoute(route),
		}
		...
		return rt
	}

  // 对 route 排序
	sortRoutes(routes)
  // 这里生成的配置，存储到 routes 这个map中，key 是 ingress_http
	v.routes[ENVOY_HTTP_LISTENER].VirtualHosts = append(v.routes[ENVOY_HTTP_LISTENER].VirtualHosts, toEnvoyVirtualHost(vh, routes, toEnvoyRoute))
}

GRPC 数据下发

从前面的分析我们知道，数据最终被转换为多个 ResourceCache 的列表，这些数据最终有两个用途：

  • 通过 GRPC 下发给数据面侧的 Envoy，最终达到流量路由的效果
  • 传给 SnapshotHander，用于生成快照
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
go复制代码func doServe(log logrus.FieldLogger, ctx *serveContext) error {
	...
  // 用于生成快照，这部分比较简单，不做介绍
  snapshotHandler := xdscache.NewSnapshotHandler(resources, log.WithField("context", "snapshotHandler"))

  ...
  // 注册服务到 GRPC
  // 先通过 NewContourServer 生成一个 GRPC 服务，再注册到 grpcServer
  switch ctx.Config.Server.XDSServerType {
    // 根据启动参数的不同配置，实例化不同的对象
    
    // serverType = "envoy"
		case config.EnvoyServerType:
			v3cache := contour_xds_v3.NewSnapshotCache(false, log)
			snapshotHandler.AddSnapshotter(v3cache)
      // 调用 NewServer 构造对象
      // 这个方法是 go-control-plane 内置的方法
			contour_xds_v3.RegisterServer(envoy_server_v3.NewServer(taskCtx, v3cache, contour_xds_v3.NewRequestLoggingCallbacks(log)), grpcServer)
    // serverType = "contour"，这个是默认配置
		case config.ContourServerType:
      // 调用 NewContourServer 构造对象
			contour_xds_v3.RegisterServer(contour_xds_v3.NewContourServer(log, xdscache.ResourcesOf(resources)...), grpcServer)
  ...
}

下面分析数据经过 grpc 下发的流程，这部分主要是使用 go-control-plane sdk

注册 ADS、SDS、CDS、EDS、LDS、RDS 等都使用了 srv 对象，可以推断 Server 继承了 各种 XDS的接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
go复制代码// contour/internal/xds/v3/server.go

// 以下注册方法均调用的 go-control-plane sdk
// 包括注册 ADS、SDS、CDS、EDS、LDS、RDS
func RegisterServer(srv Server, g *grpc.Server) {
	// register services
	envoy_service_discovery_v3.RegisterAggregatedDiscoveryServiceServer(g, srv)
	envoy_service_secret_v3.RegisterSecretDiscoveryServiceServer(g, srv)
	envoy_service_cluster_v3.RegisterClusterDiscoveryServiceServer(g, srv)
	envoy_service_endpoint_v3.RegisterEndpointDiscoveryServiceServer(g, srv)
	envoy_service_listener_v3.RegisterListenerDiscoveryServiceServer(g, srv)
	envoy_service_route_v3.RegisterRouteDiscoveryServiceServer(g, srv)
}

// 真正实例化的 Server 对象是 `contourServer`
// 将前面准备好的 ResourceCache 列表传进来
func NewContourServer(log logrus.FieldLogger, resources ...xds.Resource) Server {
	c := contourServer{
		FieldLogger: log,
		resources:   map[string]xds.Resource{},
	}
  // 按照 TypeURL 作为 key，保存资源
	for i, r := range resources {
		c.resources[r.TypeURL()] = resources[i]
	}

	return &c
}

contourServer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
go复制代码// contour/internal/xds/v3/contour.go
type contourServer struct {
   // Since we only implement the streaming state of the world
   // protocol, embed the default null implementations to handle
   // the unimplemented gRPC endpoints.
   envoy_service_discovery_v3.UnimplementedAggregatedDiscoveryServiceServer
   envoy_service_secret_v3.UnimplementedSecretDiscoveryServiceServer
   envoy_service_route_v3.UnimplementedRouteDiscoveryServiceServer
   envoy_service_endpoint_v3.UnimplementedEndpointDiscoveryServiceServer
   envoy_service_cluster_v3.UnimplementedClusterDiscoveryServiceServer
   envoy_service_listener_v3.UnimplementedListenerDiscoveryServiceServer

   logrus.FieldLogger
   // 所有生成的 Envoy XDS 信息都传进来保存到这个字段
   resources   map[string]xds.Resource
   connections xds.Counter
}

把 ResourceCache 传给 contourServer，contourServer作为GRPC的server注册到GRPC上，go-control-plane 框架封装了各个XDS数据传输的接口，contourServer 实现这些接口，从而实现数据下发

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
go复制代码func (s *contourServer) StreamClusters(srv envoy_service_cluster_v3.ClusterDiscoveryService_StreamClustersServer) error {
	return s.stream(srv)
}

func (s *contourServer) StreamEndpoints(srv envoy_service_endpoint_v3.EndpointDiscoveryService_StreamEndpointsServer) error {
	return s.stream(srv)
}

func (s *contourServer) StreamListeners(srv envoy_service_listener_v3.ListenerDiscoveryService_StreamListenersServer) error {
	return s.stream(srv)
}

func (s *contourServer) StreamRoutes(srv envoy_service_route_v3.RouteDiscoveryService_StreamRoutesServer) error {
	return s.stream(srv)
}

func (s *contourServer) StreamSecrets(srv envoy_service_secret_v3.SecretDiscoveryService_StreamSecretsServer) error {
	return s.stream(srv)
}

所有的 XDS Stream方法内部都调用了 s.stream 方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
go复制代码func (s *contourServer) stream(st grpcStream) error {
	...
  // 持续监听连接，并处理请求
	// now stick in this loop until the client disconnects.
	for {
		// 获取 envoy 发送来的请求
		req, err := st.Recv()
		...
    // 根据请求的 TypeURL，找到匹配的 ResourceCache
		r, ok := s.resources[req.GetTypeUrl()]
		...
    // 注册监听器
		r.Register(ch, last, req.ResourceNames...)
		select {
		case last = <-ch:
			
			var resources []proto.Message
			switch len(req.ResourceNames) {
      // 如果没有指定资源名称，返回所有资源
			case 0:
				// 调用前面介绍过的 Contents 方法，从 ResourceCache 的 value 中取到 envoy 配置
				resources = r.Contents()
      // 如果指定资源名称，查找特定资源
			default:
				// resource hints supplied, return exactly those
				resources = r.Query(req.ResourceNames)
			}
      // 将 ResourceCache 转换成 PB 数据
			any := make([]*any.Any, 0, len(resources))
			for _, r := range resources {
				a, err := anypb.New(proto.MessageV2(r))
				if err != nil {
					return done(log, err)
				}
				any = append(any, a)
			}
      // 构造 grpc response 对象
			resp := &envoy_service_discovery_v3.DiscoveryResponse{
				VersionInfo: strconv.Itoa(last),
				Resources:   any,
				TypeUrl:     req.GetTypeUrl(),
				Nonce:       strconv.Itoa(last),
			}

      // 将 response 通过 grpc 发送给客户端 envoy
			if err := st.Send(resp); err != nil {
				return done(log, err)
			}

		case <-ctx.Done():
			return done(log, ctx.Err())
		}
	}
}

框架server VS contourServer

contour会根据 ctx.Config.Server.XDSServerType 属性的值，初始化不同的 server。

  • envoy：初始化框架默认的server
  • contour：初始化 contourServer

Go-control-plane 提供的 pb 接口包括三类：

  • StreamXXX
  • DeltaXXX
  • FetchXXX

XXX 代表不同的 XDS，比如 StreamClusters。

contourServer 只实现了所有 XDS 的 Stream 相关方法，并没有实现 Delta，Fetch 相关方法

Envoy 连接到 Contour服务端

前面介绍了 contour 下发数据到 envoy 的流程，那最开始 envoy是如何连接上 contour的？

initContainer 生成配置文件

查看 envoy daemonset yaml文件，可以看到，内部配置了一个 initContainer，用于生成本地配置文件，并和 envoy 业务容器共享，envoy 启动时根据配置文件启动。启动命令中还指定了 envoy 连接的 grpc server 地址，即 contour的地址，这里配置的是 k8s svc 名称

1
2
3
4
5
6
7
8
9
10
11
yaml复制代码initContainers:
- args:
  - bootstrap
  - /config/envoy.json
  - --xds-address=contour
  - --xds-port=8001
  - --xds-resource-version=v3
  - --resources-dir=/config/resources
  - --envoy-cafile=/certs/ca.crt
  - --envoy-cert-file=/certs/tls.crt
  - --envoy-key-file=/certs/tls.key

envoy启动配置文件内容

进入 envoy 容器可以查看生成的配置内容

1
bash复制代码kubectl -n projectcontour exec -it envoy-65qfq -c envoy -- cat /config/envoy.json
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
yaml复制代码{
    "static_resources": {
        "clusters": [
            {
                "name": "contour",
                "alt_stat_name": "projectcontour_contour_8001",
                "type": "STRICT_DNS",
                "connect_timeout": "5s",
                "load_assignment": {
                    "cluster_name": "contour",
                    "endpoints": [
                        {
                            "lb_endpoints": [
                                {
                                    "endpoint": {
                                        "address": {
                                            "socket_address": {
                                                "address": "contour",
                                                "port_value": 8001
                                            }
                                        }
                                    }
                                }
                            ]
                        }
                    ]
                },
                "circuit_breakers": {
                    "thresholds": [
                        {
                            "priority": "HIGH",
                            "max_connections": 100000,
                            "max_pending_requests": 100000,
                            "max_requests": 60000000,
                            "max_retries": 50
                        },
                        {
                            "max_connections": 100000,
                            "max_pending_requests": 100000,
                            "max_requests": 60000000,
                            "max_retries": 50
                        }
                    ]
                },
                "typed_extension_protocol_options": {
                    "envoy.extensions.upstreams.http.v3.HttpProtocolOptions": {
                        "@type": "type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions",
                        "explicit_http_config": {
                            "http2_protocol_options": {}
                        }
                    }
                },
                "transport_socket": {
                    "name": "envoy.transport_sockets.tls",
                    "typed_config": {
                        "@type": "type.googleapis.com/envoy.extensions.transport_sockets.tls.v3.UpstreamTlsContext",
                        "common_tls_context": {
                            "tls_params": {
                                "tls_maximum_protocol_version": "TLSv1_3"
                            },
                            "tls_certificate_sds_secret_configs": [
                                {
                                    "name": "contour_xds_tls_certificate",
                                    "sds_config": {
                                        "path": "/config/resources/sds/xds-tls-certificate.json",
                                        "resource_api_version": "V3"
                                    }
                                }
                            ],
                            "validation_context_sds_secret_config": {
                                "name": "contour_xds_tls_validation_context",
                                "sds_config": {
                                    "path": "/config/resources/sds/xds-validation-context.json",
                                    "resource_api_version": "V3"
                                }
                            }
                        }
                    }
                },
                "upstream_connection_options": {
                    "tcp_keepalive": {
                        "keepalive_probes": 3,
                        "keepalive_time": 30,
                        "keepalive_interval": 5
                    }
                }
            },
            {
                "name": "envoy-admin",
                "alt_stat_name": "projectcontour_envoy-admin_9001",
                "type": "STATIC",
                "connect_timeout": "0.250s",
                "load_assignment": {
                    "cluster_name": "envoy-admin",
                    "endpoints": [
                        {
                            "lb_endpoints": [
                                {
                                    "endpoint": {
                                        "address": {
                                            "pipe": {
                                                "path": "/admin/admin.sock",
                                                "mode": 420
                                            }
                                        }
                                    }
                                }
                            ]
                        }
                    ]
                }
            }
        ]
    },
    "dynamic_resources": {
        "lds_config": {
            "api_config_source": {
                "api_type": "GRPC",
                "transport_api_version": "V3",
                "grpc_services": [
                    {
                        "envoy_grpc": {
                            "cluster_name": "contour"
                        }
                    }
                ]
            },
            "resource_api_version": "V3"
        },
        "cds_config": {
            "api_config_source": {
                "api_type": "GRPC",
                "transport_api_version": "V3",
                "grpc_services": [
                    {
                        "envoy_grpc": {
                            "cluster_name": "contour"
                        }
                    }
                ]
            },
            "resource_api_version": "V3"
        }
    },
    "admin": {
        "access_log": [
            {
                "name": "envoy.access_loggers.file",
                "typed_config": {
                    "@type": "type.googleapis.com/envoy.extensions.access_loggers.file.v3.FileAccessLog",
                    "path": "/dev/null"
                }
            }
        ],
        "address": {
            "pipe": {
                "path": "/admin/admin.sock",
                "mode": 420
            }
        }
    }
}

总结

k8s本身没有内置 ingress 控制器，ingress 控制器呈现百花齐放的状态，参考官方文档：ingress 控制器。

不同的控制器，底层使用的网络代理不同，主要有：nginx、envoy、kong、haproxy等。envoy 作为云原生时代的网络代理，被越来越多的开源项目使用。比如：istio、contour、gloo、。这些项目的本质，都是简化 envoy 的配置，封装更简单的路由规则，通过内部的逻辑转换，变成envoy 的xds配置。

go-control-plane 框架封装了基于grpc下发xds配置，可以方便开发者开发自己的控制面。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%