云原生 envoy xDS 动态配置 java控制平面开发 支持restful grpc实现 EDS 动态endpoint配置

envoy xDS 动态配置 java控制平面开发 支持restful grpc 动态endpoint配置

大纲

• 基础概念
• Envoy 动态配置API
• 配置方式
• 动静结合的配置方式
• 纯动态配置方式
• 实战

基础概念

Envoy 的强大功能之一是支持动态配置，当使用动态配置时，我们不需要重新启动 Envoy 进程就可以生效。Envoy 通过从磁盘文件或网络接口读取配置，动态地重新加载配置。动态配置使用所谓的发现服务 API，指向配置的特定部分。这些 API 也被统称为xDS 即 (xxx discovery service)

注意：
Envoy的发现API开发模式是，按照Envoy指定的接口名称，请求参数，响应值，自己开发，即需要满足Envoy的规范

Envoy动态配置支持文件方式，grpc接口和, restful接口，其中 grpc接口/REST接口 的配置提供者（自己开发的项目）也被称为控制平面

实现方式：

• 文件方式： 监听文件的变化动态修改
• grpc接口： 使用的tcp长连接
• REST接口： 使用的http轮询的方式实现

Envoy 动态配置API

API类型

Envoy 内部有多个发现服务 API （xDS）：

• 监听器发现服务（LDS listener discovery service） 使用 LDS，Envoy 可以在运行时发现监听器，包括所有的过滤器栈、HTTP 过滤器和对 RDS 的引用。（即动态配置 listener 类似nginx配置虚拟主机）
• 扩展配置发现服务（ECDS） 使用 ECDS，Envoy 可以独立于监听器获取扩展配置（例如，HTTP 过滤器配置）。
• 路由发现服务（RDS route discovery service） 使用 RDS，Envoy 可以在运行时发现 HTTP 连接管理器过滤器的整个路由配置。与 EDS 和 CDS 相结合，我们可以实现复杂的路由拓扑结构。（即动态配置路由）
• 虚拟主机发现服务（VHDS） 使用 VHDS 允许 Envoy 从路由配置中单独请求虚拟主机。当路由配置中有大量的虚拟主机时，就可以使用这个功能。
• 宽泛路由发现服务（SRDS） 使用 SRDS，可以把路由表分解成多个部分。当有大的路由表时，就可以使用这个 API。
• 集群发现服务（CDS cluster discovery service ） 使用 CDS，Envoy 可以发现上游集群。Envoy 将通过排空和重新连接所有现有的连接池来优雅地添加、更新或删除集群。Envoy 在初始化时不必知道所有的集群，因为我们可以在以后使用 CDS 配置它们。（即动态配置集群）
• 端点发现服务（EDS endpoint discovery service） 使用 EDS，Envoy 可以发现上游集群的成员。 （即动态配置后端服务类似nginx upstream）
• 秘密发现服务（SDS） 使用 SDS，Envoy 可以为其监听器发现秘密（证书和私钥，TLS 会话密钥），并为对等的证书验证逻辑进行配置。
• 运行时发现服务（RTDS） 使用 RTDS，Envoy 可以动态地发现运行时层。

API 版本

Envoy 的 API 有 v2 v3 目前主流版本是 v3

官方文档 https://www.envoyproxy.io/docs/envoy/latest/configuration/overview/xds_api

xDS API 可以使用restful接口和grpc接口开发，只要满足指定的接口名称和DiscoveryRequest,DiscoveryResponse参数和响应对象即可

例如以下就是一个EDS的接口
/v3/discovery:endpoints （即自己写的controller的mapping是/v3/discovery:endpoints）
@RequestMapping("/v3/discovery:endpoints")

配置方式

动静结合的配置方式

静态配置与动态配置结合

例如

static_resources:listeners:- name: my_listeneraddress:socket_address: protocol: TCPaddress: 0.0.0.0port_value: 15200filter_chains:- filters:- name: envoy.filters.network.http_connection_managertyped_config:"@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManagerstat_prefix: my-http-filterhttp_filters:- name: envoy.filters.http.routerstat_prefix: my_listener_httpcodec_type: AUTOroute_config:name: local_routevirtual_hosts:- name: local_servicedomains: ["*"]routes:- match:  prefix: "/" route:  cluster: user-service clusters:- name: user-servicetype: EDS #这里就是使用动态配置的方式实现endpoint的动态发现connect_timeout: 0.5seds_cluster_config: eds_config:resource_api_version: V3api_config_source:api_type: RESTtransport_api_version: V3cluster_names: [edscluster]refresh_delay: 10s - name: edsclustertype: STATICconnect_timeout: 0.5shosts:  - socket_address: address: 192.168.0.218port_value: 7590 

纯动态配置方式

使用dynamic_resources 配置动态内容

例如

dynamic_resources:ads_config:api_type: GRPCtransport_api_version: V3grpc_services:- envoy_grpc:cluster_name: xds_clustercds_config:resource_api_version: V3api_config_source:api_type: GRPCtransport_api_version: V3grpc_services:- envoy_grpc:cluster_name: xds_clusterlds_config:resource_api_version: V3api_config_source:api_type: GRPCtransport_api_version: V3grpc_services:- envoy_grpc:cluster_name: xds_cluster

当envoy没有读取到配置时会一直使用默认的配置，所以如果控制平面宕机后还是会保持配置

每个 xDS API 都有给定的资源类型:

v2版本

LDS :  envoy.api.v2.Listener
RDS : envoy.api.v2.RouteConfiguration
CDS : envoy.api.v2.Cluster
EDS ：envoy.api.v2.ClusterLoadAssignment (EDS就是配置 endpoint)

v3版本

envoy.config.listener.v3.Listener
envoy.config.route.v3.RouteConfiguration,
envoy.config.route.v3.ScopedRouteConfiguration,
envoy.config.route.v3.VirtualHost
envoy.config.cluster.v3.Cluster
envoy.config.endpoint.v3.ClusterLoadAssignment  (EDS endpoint 返回的resources 对象类型)
envoy.extensions.transport_sockets.tls.v3.Secret
envoy.service.runtime.v3.Runtime

即接口返回DiscoveryResponse 内部的resources 是以上类型

实战

本次测试 envoy的版本为v1.16.0 使用docker镜像部署

基于envoy xDS api v3版本 java restful实现

step1 配置 envoy.yaml

配置文件如下

node:cluster: myclusterid: test-id# 这是一段静态配置
static_resources:listeners:- name: my_listeneraddress:socket_address: protocol: TCPaddress: 0.0.0.0port_value: 15200 #配置一个静态的listener 监听来自任意IP的请求15200端口的http请求filter_chains:- filters:- name: envoy.filters.network.http_connection_manager #注意指定filterstyped_config:"@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManagerstat_prefix: my-http-filterhttp_filters:- name: envoy.filters.http.routerstat_prefix: my_listener_httpcodec_type: AUTOroute_config:name: local_routevirtual_hosts:- name: local_servicedomains: ["*"]  #任意域名的请求routes:- match:  prefix: "/"   # 任意url的请求route:  cluster: user-service  # 路由到user-service 集群# 配置集群    clusters:- name: user-servicetype: EDS  #模式指定为EDS  connect_timeout: 0.5s # 配置连接超时时间eds_cluster_config: eds_config:resource_api_version: V3 #指定使用V3版本接口api_config_source:api_type: REST  #使用restful的方式transport_api_version: V3   #指定使用V3版本接口cluster_names: [edscluster]refresh_delay: 10s  # 配置刷新频率# 这里配置的是envoy EDS接口的提供服务即控制平面       - name: edsclustertype: STATICconnect_timeout: 0.5s # 配置连接超时时间# envoy会去请求 192.168.0.218:7590/v3/discovery:endpoints 这个接口 获取endpoint配置信息# 代码见 my-docker-demo-envoy-plane/DataPlaneEndpointControllerV3.javahosts:  - socket_address: address: 192.168.0.218port_value: 7590  

启动 envoy 镜像

docker run  -p 5201:5201 -p 15200:15200  -v /ops/envoy:/etc/envoy  envoyproxy/envoy:v1.16.0 

envoy 启动后可以看到开始调用 EDS接口，由于还没启动服务此时会报错

step2 java 程序开发

EDS接口使用java springboot 开发

注意点如下：

• 1 接口必须是 /v3/discovery:endpoints
• 2 动态配置需要是一个json 字符串 并且满足endpoint需要的格式
• 3 返回值必须是一个DiscoveryResponse service.discovery.v3.DiscoveryResponse

DiscoveryResponse 格式如下

{"version_info": ...,"resources": [],"type_url": ...,"nonce": ...,"control_plane": {...}
}

整体的返回值json字符串如下

{"versionInfo": "1.0.0","resources": [{"@type": "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment","clusterName": "user-service","endpoints": [{"lbEndpoints": [ {"endpoint": {"address": {"socketAddress": {"address": "10.244.1.203","portValue": 5588}}}}]}]}]
}

如果自己拼接json字符串感觉比较麻烦，可以使用envoy-api包

<dependency><groupId>io.envoyproxy.controlplane</groupId><artifactId>api</artifactId><version>1.0.39</version></dependency>

这个包，里面有xDS中的各种资源对象 以及grpc接口

也可以使用官方提供的 java控制面板项目 打包编译后得到api包，里面也有xDS中的各种资源对象

java 代码如下

  @RequestMapping(value="/v3/discovery:endpoints" , produces = {"application/json;charset=UTF-8"})public String discovery(HttpServletRequest req) throws Exception {			//json 字符串拼接//String json = staticJson();/*** 构建返回EDS 配置json 字符串*/String json = useBean();return json;}/*** @return*/private String useBean() throws Exception {/*** 以下资源类出自* * <dependency><groupId>io.envoyproxy.controlplane</groupId><artifactId>api</artifactId><version>1.0.39</version></dependency>* *///配置上游服务（类似nginx upstream）SocketAddress sa1 = SocketAddress.newBuilder().setAddress("10.244.0.190").setPortValue(5588).build();SocketAddress sa2 = SocketAddress.newBuilder().setAddress("10.244.1.203").setPortValue(5588).build();Address address1 = Address.newBuilder().setSocketAddress(sa1).build();Address address2 = Address.newBuilder().setSocketAddress(sa2).build();Endpoint end1 = Endpoint.newBuilder().setAddress(address1).build();Endpoint end2 = Endpoint.newBuilder().setAddress(address2).build();LbEndpoint lb1 = LbEndpoint.newBuilder().setEndpoint(end1).build();LbEndpoint lb2 = LbEndpoint.newBuilder().setEndpoint(end2).build();LocalityLbEndpoints llb = LocalityLbEndpoints.newBuilder().addLbEndpoints(lb1).addLbEndpoints(lb2).build();ClusterLoadAssignment cla = ClusterLoadAssignment.newBuilder().setClusterName("user-service").addEndpoints(llb).build();DiscoveryResponse dr = DiscoveryResponse.newBuilder().setVersionInfo("1.0.0").addResources(Any.pack(cla)).build();JsonFormat.TypeRegistry typeRegistry = JsonFormat.TypeRegistry.newBuilder().add(ClusterLoadAssignment.getDescriptor()).build();JsonFormat.Printer printer = JsonFormat.printer().usingTypeRegistry(typeRegistry);return printer.print(dr);
}

基于 envoy xDS api v3版本 java grpc实现

grpc的关键

• 1 使用envoy api包 实现对应的grpc 服务
• 2 返回值需要指定typeUrl
• 3 配置文件需要加入 http2_protocol_options 指定使用http2

没使用http2_protocol_options 配置会出现如下异常

io.grpc.netty.shaded.io.netty.handler.codec.http2.Http2Exception: Unexpected HTTP/1.x request: POST /envoy.service.endpoint.v3.EndpointDiscoveryService/StreamEndpoints at io.grpc.netty.shaded.io.netty.handler.codec.http2.Http2Exception.connectionError(Http2Exception.java:109) ~[grpc-netty-shaded-1.48.1.jar:1.48.1]at io.grpc.netty.shaded.io.netty.handler.codec.http2.Http2ConnectionHandler$PrefaceDecoder.readClientPrefaceString(Http2ConnectionHandler.java:302) ~[grpc-netty-shaded-1.48.1.jar:1.48.1]at io.grpc.netty.shaded.io.netty.handler.codec.http2.Http2ConnectionHandler$PrefaceDecoder.decode(Http2ConnectionHandler.java:239) ~[grpc-netty-shaded-1.48.1.jar:1.48.1]at io.grpc.netty.shaded.io.netty.handler.codec.http2.Http2ConnectionHandler.decode(Http2ConnectionHandler.java:438) [grpc-netty-shaded-1.48.1.jar:1.48.1]

返回值未指定typeUrl

023-08-16 06:14:49.608][8][warning][config] [source/common/config/grpc_mux_impl.cc:155] Ignoring the message for type URL  as it has no current subscribers.

关键配置 envoy.yaml 如下

# 指定集群名称
# 动态配置需要指定节点集群名称        
node:cluster: myclusterid: test-id# 这是一段静态配置
static_resources:listeners:- name: my_listeneraddress:socket_address: protocol: TCPaddress: 0.0.0.0port_value: 15200 #配置一个静态的listener 监听来自任意IP的请求15200端口的http请求filter_chains:- filters:- name: envoy.filters.network.http_connection_manager #注意指定filterstyped_config:"@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManagerstat_prefix: my-http-filterhttp_filters:- name: envoy.filters.http.routerstat_prefix: my_listener_httpcodec_type: AUTOroute_config:name: local_routevirtual_hosts:- name: local_servicedomains: ["*"]  #任意域名的请求routes:- match:  prefix: "/"   # 任意url的请求route:  cluster: user-service  # 路由到user-service 集群# 配置集群    clusters:- name: user-servicetype: EDS  #模式指定为EDS  connect_timeout: 0.5s # 配置连接超时时间eds_cluster_config: eds_config:resource_api_version: V3 #指定使用V3版本接口api_config_source:api_type: GRPC  #使用grpc的方式transport_api_version: V3   #指定使用V3版本接口# 指定grpc_services 对应的集群# 这里将使用下面定义的集群grpc_services: - envoy_grpc: cluster_name: edscluster# 这里配置的是envoy EDS接口的提供服务即控制平面       - name: edsclustertype: STATICconnect_timeout: 0.5s # 配置连接超时时间# 这里是一个关键，必须指定http2_protocol_options 即使用http2http2_protocol_options: {}hosts:  - socket_address: address: 192.168.0.218port_value: 7899  

关键java代码

public class EndpointDiscoveryServiceGrpcImpl extends EndpointDiscoveryServiceGrpc.EndpointDiscoveryServiceImplBase {/*** 这个接口是客户端模式* */@Overridepublic io.grpc.stub.StreamObserver<io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest> streamEndpoints(io.grpc.stub.StreamObserver<io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse> responseObserver) {System.out.println("run grpc ...");/*** 创建StreamObserver<DiscoveryRequest>对象*/StreamObserver<DiscoveryRequest> so = new StreamObserver<DiscoveryRequest>() {@Overridepublic void onNext(DiscoveryRequest request) {//接收客户端每一次发送的数据，返回给客户端//showRequest(request);SocketAddress sa1 = SocketAddress.newBuilder().setAddress("10.244.0.214").setPortValue(5588).build();SocketAddress sa2 = SocketAddress.newBuilder().setAddress("10.244.0.201").setPortValue(5588).build();Address address1 = Address.newBuilder().setSocketAddress(sa1).build();Address address2 = Address.newBuilder().setSocketAddress(sa2).build();Endpoint end1 = Endpoint.newBuilder().setAddress(address1).build();Endpoint end2 = Endpoint.newBuilder().setAddress(address2).build();LbEndpoint lb1 = LbEndpoint.newBuilder().setEndpoint(end1).build();LbEndpoint lb2 = LbEndpoint.newBuilder().setEndpoint(end2).build();LocalityLbEndpoints llb = LocalityLbEndpoints.newBuilder().addLbEndpoints(lb1).addLbEndpoints(lb2).build();ClusterLoadAssignment cla = ClusterLoadAssignment.newBuilder()/*** 这里配置的ClusterName 应该是路由对应的cluster name 而不是 node中的cluster* route: { cluster: user-service }*/.setClusterName("user-service") .addEndpoints(llb).build();final DiscoveryResponse dr = DiscoveryResponse.newBuilder().setVersionInfo("1.0.0")							.setTypeUrl("type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment").addResources(Any.pack(cla)).build();/*** 客户端模式这里不会去关闭StreamObserver* 即不会调用 responseObserver.onCompleted();方法*/responseObserver.onNext(dr);System.out.println("send DiscoveryResponse ");}@Overridepublic void onError(Throwable t) {System.out.println("onError");t.printStackTrace();}@Overridepublic void onCompleted() {//当客户端数据发送完毕后调用此方法，返回客户端SocketAddress sa1 = SocketAddress.newBuilder().setAddress("10.244.0.214").setPortValue(5588).build();SocketAddress sa2 = SocketAddress.newBuilder().setAddress("10.244.0.201").setPortValue(5588).build();Address address1 = Address.newBuilder().setSocketAddress(sa1).build();Address address2 = Address.newBuilder().setSocketAddress(sa2).build();Endpoint end1 = Endpoint.newBuilder().setAddress(address1).build();Endpoint end2 = Endpoint.newBuilder().setAddress(address2).build();LbEndpoint lb1 = LbEndpoint.newBuilder().setEndpoint(end1).build();LbEndpoint lb2 = LbEndpoint.newBuilder().setEndpoint(end2).build();LocalityLbEndpoints llb = LocalityLbEndpoints.newBuilder().addLbEndpoints(lb1).addLbEndpoints(lb2).build();ClusterLoadAssignment cla = ClusterLoadAssignment.newBuilder().setClusterName("user-service").addEndpoints(llb).build();final DiscoveryResponse dr = DiscoveryResponse.newBuilder().setVersionInfo("1.0.0").addResources(Any.pack(cla)).build();System.out.println("onCompleted");responseObserver.onNext(dr);responseObserver.onCompleted();}};return so;