Embedded HA Proxy

Current HA in openLooKeng:

https://openlookeng.io/docs/docs/installation/deployment-ha.html

image.png

Active-Active High Availability Cluster

The main purpose of an active-active cluster is to achieve load balancing.

An active-active cluster is typically made up of at least two nodes, both actively running the same kind of service simultaneously.

Active-Passive High Availability Cluster

The passive (a.k.a. failover) server serves as a backup that's ready to take over as soon as the active (a.k.a. primary) server gets disconnected or is unable to serve.

PrestoServer 启一个服务器线程，里面包含创建一个ServerMainModule，用于根据节点的配置来创建CoordinatorModule或WorkerModule。换句话讲现在是一个Server对应一个CN或者WN。如果说加入反向代理，这个关系不需要改变，需要改变的是在某一个节点上新增一个Proxy线程，用于发现所有CN并做相应的负载均衡。

https://www.bbsmax.com/A/QW5Y0E0GJm/ （服务发现独立部署，解决Presto on Yarn Coordinator地址变化问题）

解析PrestoProxy：

public final class PrestoProxy

{

 private PrestoProxy() {}

 public static void start(Module... extraModules)

 {

 Bootstrap app = new Bootstrap(ImmutableList.<Module>builder()

 .add(new NodeModule())

 .add(new HttpServerModule())

 .add(new JsonModule())

 .add(new JaxrsModule())

 .add(new MBeanModule())

 .add(new JmxModule())

 .add(new LogJmxModule())

 .add(new TraceTokenModule())

 .add(new EventModule())

 .add(new ProxyModule())

 .add(extraModules)

 .build());

 Logger log = Logger.get(PrestoProxy.class);

 try {

 app.strictConfig().initialize();

 log.info("======== SERVER STARTED ========");

 }

 catch (Throwable t) {

 log.error(t);

 System.exit(1);

 }

 }

 public static void main(String[] args)

 {

 start();

 }

}

NodeModule：注册当前节点信息

HttpServerModule: 以http server提供给cli访问

ProxyModule.java

configBinder(binder).bindConfigDefaults(HttpServerConfig.class, httpServerConfig -> {

 httpServerConfig.setAdminEnabled(false);

 });

 httpClientBinder(binder).bindHttpClient("proxy", ForProxy.class);

ProxyResource.java

@Inject

 public ProxyResource(@ForProxy HttpClient httpClient, JsonWebTokenHandler jwtHandler, ProxyConfig config)

 {

 this.httpClient = requireNonNull(httpClient, "httpClient is null");

 this.jwtHandler = requireNonNull(jwtHandler, "jwtHandler is null");

 this.remoteUri = requireNonNull(config.getUri(), "uri is null");

 this.hmac = hmacSha256(loadSharedSecret(config.getSharedSecretFile()));

 }

ProxyModule:

ProxyConfig 一些与代理相关的配置，包括：

uri

sharedSecretFile ？？？？？

ProxyResource执行具体的请求及获取响应，下面逐行理解该类代码：

private static final String X509_ATTRIBUTE = "javax.servlet.request.X509Certificate";

private static final Duration ASYNC_TIMEOUT = new Duration(2, MINUTES);

private static final JsonFactory JSON_FACTORY = new JsonFactory().disable(CANONICALIZE_FIELD_NAMES);

private final ExecutorService executor = newCachedThreadPool(daemonThreadsNamed("proxy-%s")); // new一个执行线程？

private final HttpClient httpClient; // proxy提供给外部的httpclient实例，由外部注册进来

private final JsonWebTokenHandler jwtHandler;   

private final URI remoteUri; // proxy提供给外部访问的uri

private final HashFunction hmac;

RESTful API

@GET

 @Path("/v1/info") // 对外的接口名

 @Produces(APPLICATION_JSON)

 public void getInfo(

 @Context HttpServletRequest servletRequest,

 @Suspended AsyncResponse asyncResponse)

 {

 Request.Builder request = prepareGet()

 .setUri(uriBuilderFrom(remoteUri).replacePath("/v1/info").build()); // build request

  performRequest(servletRequest, asyncResponse, request, response ->

 responseWithHeaders(Response.ok(response.getBody()), response)); // 执行请求，获取结果

 }

 @POST

 @Path("/v1/statement") // 对外的接口名

 @Produces(APPLICATION_JSON)

 public void postStatement(

 String statement,

 @Context HttpServletRequest servletRequest,

 @Context UriInfo uriInfo,

 @Suspended AsyncResponse asyncResponse)

 {

 Request.Builder request = preparePost()

 .setUri(uriBuilderFrom(remoteUri).replacePath("/v1/statement").build())

 .setBodyGenerator(createStaticBodyGenerator(statement, UTF_8));

   performRequest(servletRequest, asyncResponse, request, response -> buildResponse(uriInfo, response));

 }

 @GET

 @Path("/v1/proxy") // 对外的接口名

 @Produces(APPLICATION_JSON)

 public void getNext(

 @QueryParam("uri") String uri,

 @QueryParam("hmac") String hash,

 @Context HttpServletRequest servletRequest,

 @Context UriInfo uriInfo,

 @Suspended AsyncResponse asyncResponse)

 {

 if (!hmac.hashString(uri, UTF_8).equals(HashCode.fromString(hash))) {

 throw badRequest(FORBIDDEN, "Failed to validate HMAC of URI");

 }

 Request.Builder request = prepareGet().setUri(URI.create(uri));

   performRequest(servletRequest, asyncResponse, request, response -> buildResponse(uriInfo, response));

 }

 @DELETE

 @Path("/v1/proxy") // 对外的接口名

 @Produces(APPLICATION_JSON)

 public void cancelQuery(

 @QueryParam("uri") String uri,

 @QueryParam("hmac") String hash,

 @Context HttpServletRequest servletRequest,

 @Suspended AsyncResponse asyncResponse)

 {

 if (!hmac.hashString(uri, UTF_8).equals(HashCode.fromString(hash))) {

 throw badRequest(FORBIDDEN, "Failed to validate HMAC of URI");

 }

 Request.Builder request = prepareDelete().setUri(URI.create(uri));

  performRequest(servletRequest, asyncResponse, request, response -> responseWithHeaders(noContent(), response));

 }

service-inventory.json这是一个比较重要的文件，里面记录了整个集群的信息，discovery集群利用这个配置文件获取集群的所有信息，知道集群中所有部署的情况及如何与其它节点进行通信。它的配置如下：

{

"environment": "test",

"services": [

{

"id": "C8A9EE64-0476-452C-8638-8E72F3EE3CA6",

"nodeId": "597A741E-9968-40E2-BB4D-7AF26DE18689",

"type": "discovery",

"pool": "general",

"location": "/172.17.31.245",

"state": "RUNNING",

"properties": {

"http": "[http://172.17.31.245:8411](http://172.17.31.245:8411/)"

}

},

{

"id": "370AF416-5F44-47D3-BFB6-D93A92676D49",

"nodeId": "0BA42FDB-5DBA-4A2C-BE26-9596B7B4368E",

"type": "discovery",

"pool": "general",

"location": "/172.17.31.246",

"state": "RUNNING",

"properties": {

"http": "[http://172.17.31.246:8411](http://172.17.31.246:8411/)"

}

}]

}

HetuDiscoveryModule 绑定 HetuServiceInventory，从共享StateStore里获取到所有discovery-service里的节点，并取第一个作为提供服务的节点；Prxoy模式下无法从ServiceInventory里获取到确切地连接地址信息，需要动态地从StateStore里读取到所有可用的Coordinator列表，并基于某种LB策略选取列表中的某个Coordinator来执行本次任务。