Envoy Proxy 动态API的使用总结

Envoy Proxy和其它L4/L7反向搭理工具最大的区别就是原生支持动态配置。首先来看一下Envoy的大致架构

从上图可以简单理解：Listener负责接受外部的请求，然后经过Filter/Router处理之后，在转发到具体的Cluster。其中Listener，Router，Cluster和Host地址都是可以动态配置的，配置这些数据的服务就称之为X Discovery Services，简称XDS。

本文主要描述如何编写XDS Server更新逻辑。

Envoy Porxy XDS Service通过GRPC服务进行数据更新，所有Proto文件可以参考 https://github.com/envoyproxy/envoy/tree/master/api/envoy/api/v2 。用户可以根据proto文件自行生成相对应语言的GRPC代码文件。如果使用golang来实现的话，Envoy已经提供了一份编译好的GRPC代码，地址在这里: https://github.com/envoyproxy/go-control-plane/tree/master/envoy/api/v2

每个XDS Service都有两种GRPC服务， Stream和Delta。 Stream用来更新全量数据，Delta用来更新增量数据。下面以RDS Service为例来看看如何实现一个 XDS Service。

RDS Service可以提供所有的Route信息，一个简化后的典型Route配置如下:

# 完整的Route API定义参考 https://www.envoyproxy.io/docs/envoy/latest/api-v2/api/v2/rds.proto#envoy-api-msg-routeconfiguration

                route_config:

                  name: local_route

                  virtual_hosts:

                    - name: local_service

                      domains: ["*"]

                      require_tls: NONE

                      routes:

                        - match:

                            prefix: "/MyService"

                          route: { cluster: my-grpc-svc_cluster }

上面的配置语义为：当收到一个Path前缀为/MyService的请求后，将此请求转发到my-grpc-svc_cluster. (my-grpc-svc_cluster表示的是后端Upstream信息，可以是STATIC类型也可以由CDS Service动态提供)

RDS Service的作用就是动态生成类似上面的语义配置。先来看相对简单的StreamRoutes如何实现。

GRPC描述文件中，对此函数的定义如下：

service RouteDiscoveryService {

  rpc StreamRoutes(stream DiscoveryRequest) returns (stream DiscoveryResponse) {

  }

  rpc DeltaRoutes(stream DeltaDiscoveryRequest) returns (stream DeltaDiscoveryResponse) {

  }

  rpc FetchRoutes(DiscoveryRequest) returns (DiscoveryResponse) {

    option (google.api.http) = {

      post: "/v2/discovery:routes"

      body: "*"

    };

  }

}

从返回值可以看出StreamRoutes是一个流函数，RDS会通过这个流实时将数据推送给Envoy。所以大致的实现模型就是如下的样子:

func (r rds) StreamRoutes(ls envoy_api_v2.RouteDiscoveryService_StreamRoutesServer) error {

    for{

        select{

            case x <- c>:

                ls.Send(xxx)

        }

    }

}

Send函数接受的是DiscoveryResponse指针，而这个DiscoveryResponse从定义来看是自解释动态结构体。具体数据类型由typeUrl属性来决定。具体到Route来说，typeURL是"type.googleapis.com/envoy.api.v2.RouteConfiguration". （类型说明参见 https://www.envoyproxy.io/docs/envoy/latest/api-docs/xds_protocol.html?highlight=type url#resource-types ）

数据则由Resource来保存。

Resource是[]*any.Any类型，说白了就是万能的Interface{}。所以创建any.Any时需要指定具体的数据类型("type.googleapis.com/envoy.api.v2.RouteConfiguration"). data则是经过ProtoMessage编码后的二进制数据。所以创建any.Any应该是下面的样子:



            data, err := proto.Marshal(xxxx)

			if err != nil {

				logrus.Errorf("Marshal Error. %s", err)

				continue

			}

			any :=  &any.Any{

				TypeUrl: "type.googleapis.com/envoy.api.v2.Cluster",

				Value:   data,

			})

xxxx是RDS需要返回给Envoy的路由数据，也就是RouteConfiguration。所以下面来看如何构建RouteConfiguration。通过API定义可知，有一些数据是必输项(通过proto校验描述文件也可以获取必输项，但不如看API文档来的直接)。假设我们要实现开篇简单的Route配置，那么 RouteConfiguration 应该这样定义:

      r:=&envoy_api_v2.RouteConfiguration{

			Name: "local_route",

			VirtualHosts: []*route.VirtualHost{

				&route.VirtualHost{

					Name: "local_service",

					Domains: []string{

						"*",

					},

					Routes: []*route.Route{

						&route.Route{

							Match: &route.RouteMatch{

								PathSpecifier: &route.RouteMatch_Prefix{Prefix: "/MyService"},

							},

							Action: &route.Route_Route{Route: &route.RouteAction{

								ClusterSpecifier: &route.RouteAction_Cluster{Cluster: "my-grpc-svc_cluster"},

							}},

						},

					},

				},

			},

		},

需要注意两个地方：

Name: "local_route"。这里的Name一定要和Listener中定义的RouteConfig Name保持一致。如果不一致，Listener不会加载这段Route配置(换言之，这个Name就是双方的关联主键)
Cluster 名称也要保持一致。同理，如果不一致，后续请求转发时就会找不到UPstream

经过这些步骤，一个近似完整的Route DiscoveryResponse就定义完成了。而后就可以通过调用Send来发送给Envoy。

然而此时事情并没有结束，开篇说过Stream同步全量，Delta同步增量。再详细一点，在StreamRoutes中每次都需要传输当前所有的Route配置，而不仅仅是发生过变更的数据 . 个人感觉这种处理方式，对于数据组织来说很麻烦，但对于Envoy数据更新来说确很方便(每次都是全量数据，不用做merge了)。 merge总是一件耗时费力的事情，就看事情谁来做，这次envoy决定让用户来做了。

所以我们需要调整一下StreamRoutes实现模型:

func (r rds) StreamRoutes(ls envoy_api_v2.RouteDiscoveryService_StreamRoutesServer) error {

    for{

        select{

            case x <- c>:

                // x表示变动的数据

                n := merge(x) //对x进行merge操作，返回当前最新全量数据n

                var srvRoute []*route.Route

                for _, d := range n{

                    srvRoute = append(srvRoute, &route.Route{

                                        Match: &route.RouteMatch{

                                            PathSpecifier: &route.RouteMatch_Prefix{Prefix: xxxx},

                                        },

                                        Action: &route.Route_Route{Route: &route.RouteAction{

                                            ClusterSpecifier: &route.RouteAction_Cluster{Cluster: xxxx},

                                        }},

                                    })

                }

                rc := []*envoy_api_v2.RouteConfiguration{

                    &envoy_api_v2.RouteConfiguration{

                        Name: "local_route",

                        VirtualHosts: []*route.VirtualHost{

                            &route.VirtualHost{

                                Name: "local_service",

                                Domains: []string{

                                    "*",

                                },

                                Routes: srvRoute,

                            },

                        },

                    },

                }

                var resource []*any.Any

                for _, rca := range rc {

                    data, err := proto.Marshal(rca)

                    if err != nil {

                        return err

                    }

                    resource = append(resource, &any.Any{

                        TypeUrl: "type.googleapis.com/envoy.api.v2.RouteConfiguration",

                        Value:   data,

                    })

                }

                ls.Send(&envoy_api_v2.DiscoveryResponse{

                        VersionInfo: xxx,

                        Resources:   resource,

                        Canary:      false,

                        TypeUrl:     "type.googleapis.com/envoy.api.v2.RouteConfiguration",

                        Nonce:       time.Now().String(),

                    })

        }

    }

}

调整之后，每次就会返回Envoy最新的Route数据。上面的模型仅考虑了单Envoy实例的情况，并未考虑多实例。当多实例链接RDS Service时，从c获取数据，就会变成非幂等事件，从而无法保证所有Envoy实例数据保持一致。

实现StreamRoutes之后，在来看如何实现DeltaRoutes。

Delta是用来同步增量数据的，从函数原型来看，入参也是一个Stream，所以函数原型应该和StreamRoutes差不多。 如果你也这样想，就错了

Delta的stream只是用来传输数据的(猜测是为了提高数据传输效率，而并不是为了保持长连接)。每次传输完成之后，Envoy都会主动断开这个链接。也就是说，Envoy是定时调用DeltaRoutes来获取增量更新数据的。如果按照stream的实现模型来编写逻辑，将会发现经过一段时间后，这个stream会莫名的变成closed状态。原因就是envoy接收到此次事件后，主动关闭了stream。

所以如果要使用Delta模式，那么会无法保证Envoy无法实时响应数据变化(因为这个定时调用的存在)。而如果使用Stream模式，那么用户需要自行维护数据正确性(如果merge很复杂，正确性就会下降)。

所以选择Stream还是Delta对于用户来说是个问题。

巴特西

新手学分布式 - Envoy Proxy XDS Server动态配置的一点使用心得

Envoy Proxy 动态API的使用总结

最新文章

热门文章