使用 JavaScript 脚本来进行复杂的查询改写

有这么一个需求：

网关里怎样对跨集群搜索进行支持的呢？我想实现: 输入的搜索请求是 lp:9200/index1/_search 这个索引在3个集群上，需要跨集群检索，也就是网关能否改成 lp:9200/cluster01:index1,cluster02,index1,cluster03:index1/_search 呢？ 索引有一百多个，名称不一定是 app, 还可能多个索引一起的。

极限网关自带的过滤器 content_regex_replace 虽然可以实现字符正则替换，但是这个需求是带参数的变量替换，稍微复杂一点，没有办法直接用这个正则替换实现，有什么其他办法实现么？

使用脚本过滤器

当然有的，上面的这个需求，理论上我们只需要将其中的索引 index1 匹配之后，替换为 cluster01:index1,cluster02,index1,cluster03:index1 就行了。

答案就是使用自定义脚本来做，再复杂的业务逻辑都不是问题，都能通过自定义脚本来实现，一行脚本不行，那就两行。

使用极限网关提供的 JavaScript 过滤器可以很灵活的实现这个功能，具体继续看。

定义过滤器

首先创建一个脚本文件，放在网关数据目录的 scripts 子目录下面，如下：

➜  gateway ✗ tree data 
data
└── gateway
    └── nodes
        └── c9bpg0ai4h931o4ngs3g
            ├── kvdb
            ├── queue
            ├── scripts
            │   └── index_path_rewrite.js
            └── stats

这个脚本的内容如下：

function process(context) {
    var originalPath = context.Get("_ctx.request.path");
    var matches = originalPath.match(/\/?(.*?)\/_search/)
    var indexNames = [];
    if(matches && matches.length > 1) {
        indexNames = matches[1].split(",")
    }
    var resultNames = []
    var clusterNames = ["cluster01", "cluster02"]
    if(indexNames.length > 0) {
        for(var i=0; i<indexNames.length; i++){
            if(indexNames[i].length > 0) {
                for(var j=0; j<clusterNames.length; j++){
                    resultNames.push(clusterNames[j]+":"+indexNames[i])
                }
            }
        }
    }

    if (resultNames.length>0){
        var newPath="/"+resultNames.join(",")+"/_search";
        context.Put("_ctx.request.path",newPath);
    }
}

和普通的 JavaScript 一样，定义一个特定的函数 process 来处理请求里面的上下文信息，_ctx.request.path 是网关内置上下文的一个变量，用来获取请求的路径，通过 context.Get("_ctx.request.path") 在脚本里面进行访问。

中间我们使用了 JavaScript 的正则匹配和字符处理，做了一些字符拼接，得到新的路径 newPath 变量，最后使用 context.Put("_ctx.request.path",newPath) 更新网关请求的路径信息，从而实现查询条件里面的参数替换。

有关网关内置上下文的变量列表，请访问 Request Context

接下来，创建一个网关配置，并使用 javascript 过滤器调用该脚本，如下：

entry:
  - name: my_es_entry
    enabled: true
    router: my_router
    max_concurrency: 10000
    network:
      binding: 0.0.0.0:8000

flow:
  - name: default_flow
    filter:
      - dump:
          context:
            - _ctx.request.path
      - javascript:
          file: index_path_rewrite.js
      - dump:
          context:
          - _ctx.request.path
      - elasticsearch:
          elasticsearch: dev
router:
  - name: my_router
    default_flow: default_flow

elasticsearch:
- name: dev
  enabled: true
  schema: http
  hosts:
    - 192.168.3.188:9206

上面的例子中，使用了一个 javascript 过滤器，并且指定了加载的脚本文件为 index_path_rewrite.js，并使用了两个 dump 过滤器来输出脚本运行前后的路径信息，最后再使用一个 elasticsearch 过滤器来转发请求给 Elasticsearch 进行查询。

我们启动网关测试一下，如下：

➜  gateway ✗ ./bin/gateway
   ___   _   _____  __  __    __  _       
  / _ \ /_\ /__   \/__\/ / /\ \ \/_\ /\_/\
 / /_\///_\\  / /\/_\  \ \/  \/ //_\\\_ _/
/ /_\\/  _  \/ / //__   \  /\  /  _  \/ \ 
\____/\_/ \_/\/  \__/    \/  \/\_/ \_/\_/ 

[GATEWAY] A light-weight, powerful and high-performance elasticsearch gateway.
[GATEWAY] 1.0.0_SNAPSHOT, 2022-04-18 07:11:09, 2023-12-31 10:10:10, 8062c4bc6e57a3fefcce71c0628d2d4141e46953
[04-19 11:41:29] [INF] [app.go:174] initializing gateway.
[04-19 11:41:29] [INF] [app.go:175] using config: /Users/medcl/go/src/infini.sh/gateway/gateway.yml.
[04-19 11:41:29] [INF] [instance.go:72] workspace: /Users/medcl/go/src/infini.sh/gateway/data/gateway/nodes/c9bpg0ai4h931o4ngs3g
[04-19 11:41:29] [INF] [app.go:283] gateway is up and running now.
[04-19 11:41:30] [INF] [api.go:262] api listen at: http://0.0.0.0:2900
[04-19 11:41:30] [INF] [entry.go:312] entry [my_es_entry] listen at: http://0.0.0.0:8000
[04-19 11:41:30] [INF] [module.go:116] all modules are started
[04-19 11:41:30] [INF] [actions.go:349] elasticsearch [dev] is available

运行下面的查询来验证查询结果，如下：

curl localhost:8000/abc,efg/_search

可以看到网关通过 dump 过滤器输出的调试信息：

---- DUMPING CONTEXT ---- 
_ctx.request.path  :  /abc,efg/_search
---- DUMPING CONTEXT ---- 
_ctx.request.path  :  /cluster01:abc,cluster02:abc,cluster01:efg,cluster02:efg/_search

查询条件按照我们的需求进行了改写，Nice！

重写 DSL 查询语句

好吧，我们刚刚只是修改了查询的索引而已，那么查询请求的 DSL 呢？行不行？

那自然是可以的嘛，瞧下面的例子:

function process(context) {
    var originalDSL = context.Get("_ctx.request.body");
    if (originalDSL.length >0){
        var jsonObj=JSON.parse(originalDSL);
        jsonObj.size=123;
        jsonObj.aggs= {
            "test1": {
                "terms": {
                    "field": "abc",
                        "size": 10
                }
            }
        }
        context.Put("_ctx.request.body",JSON.stringify(jsonObj));
    }
}

先是获取查询请求，然后转换成 JSON 对象，之后任意修改查询对象就行了，保存回去，搞掂。

测试一下:

 curl -XPOST   localhost:8000/abc,efg/_search -d'{"query":{}}'

输出:

---- DUMPING CONTEXT ---- 
_ctx.request.path  :  /abc,efg/_search
_ctx.request.body  :  {"query":{}}
[04-19 18:14:24] [INF] [reverseproxy.go:255] elasticsearch [dev] hosts: [] => [192.168.3.188:9206]
---- DUMPING CONTEXT ---- 
_ctx.request.path  :  /abc,efg/_search
_ctx.request.body  :  {"query":{},"size":123,"aggs":{"test1":{"terms":{"field":"abc","size":10}}}}

是不是感觉解锁了新的世界？

结论

通过使用 Javascript 脚本过滤器，我们可以非常灵活的进行复杂逻辑的操作来满足我们的业务需求。

配置

上一篇我们先学习了极限网关的安装和启动，今天学习配置。

读写分离

现在我们遇到读写分离的需求，用网关该怎么做呢？ 假设服务端现在从 http://127.0.0.1:8000 写入数据，从 http://127.0.0.1:9000 读取数据，怎么设计呢？

首先查看文档配置文档

我们在 gateway.yml 中定义两个 entry，分别绑定不同的端口，配置不同的 router

entry:
  - name: write_es
    enabled: true
    router: write_router
    network:
      binding: 0.0.0.0:8000
  - name: read_es
    enabled: true
    router: read_router
    network:
      binding: 0.0.0.0:9000

router:
  - name: write_router
    default_flow: default_flow
    tracing_flow: logging
  - name: read_router
    default_flow: default_flow
    tracing_flow: logging

为了演示效果，只配置一个 Elasticsearch

elasticsearch:
- name: dev
  enabled: true
  schema: http
  hosts:
    - 192.168.3.188:9206

启动项目

我们从 http://127.0.0.1:8000 写入一条数据，再从 http://127.0.0.1:9000 读取该条数据

添加接口

返回字符串

我们想自定义添加一个接口，怎么在不写代码的情况下通过配置实现返回字符串

flow:
  - name: hello_flow
    filter:
      - echo:
          message: "hello flow"

router:
  - name: read_router
    default_flow: hello_flow

修改配置后启动

返回 json 数据

返回字符串不符合标准的 restful 接口规范，怎么返回给调用方标准 json 数据？

filter:
  - set_response:
      content_type: application/json
      body: '{"message":"hello world"}'

修改配置后启动

修改路由

我们已经新加了接口，返回 json 数据，但是接口是直接定义在 http://127.0.0.1:9000 中，之前网关的接口就无法使用，所以我们需要单独为自定义的接口指定单独的路由

router:
  - name: read_router
    default_flow: default_flow
    tracing_flow: logging
    rules:
      - method:
          - GET
        pattern:
          - "/hello"
        flow:
          - hello_flow

default_flow: 默认的处理流，也就是业务处理的主流程，请求转发、过滤、缓存等操作都在这里面进行

tracing_flow：用于追踪请求状态的流，用于记录请求日志、统计等

如果我们有过开发经验，了解 MVC 模式，flow 就类似 MVC 中的 Controller，rules 中类似路由规则，当请求匹配到配置中的路由规则时，由配置的 flow 处理业务逻辑。

数据整体流向，从服务端发到网关，网关为每个 Elasticsearch 绑定不同的 IP 地址，每个 Elasticsearch 都有唯一一个 router 和它对应，根据请求的 method 和 path 匹配到 router 中的一个 flow，flow 中包含多个 filter 处理对数据进行流式处理。

如下图所示

流式处理是什么，假设水从一个管子里面流出来，管子旁边每一段依次站了几个人，第一个人往水里放点鱼，鱼和水到了第二个人，第二个人往水里放点草，鱼、水和草到了第三人等等，每个人对水做一定的操作，水经过这些操作后最后到达水池里。

我们可以把数据当成水，filter 是管子旁边的人，水池就是 Elasticsearch

总结

在学习了router/flow/filter后，我们已经对极限网关的配置有了初步的了解，后续开发的时候直接查阅文档。

产品介绍

极限网关（INFINI Gateway）是一个面向 Elasticsearch 的高性能应用网关。特性丰富，使用简单。

它和其他业务型网关最大的区别是业务网关把请求转发给各个底层微服务，而它把请求转发给 Elasticsearch，更多是类似 Mycat 的中间件的作用。

没有使用网关之前，服务端请求多个节点

使用网关后

下载地址

打开 下载地址，根据操作系统版本选择。

Windows 安装和启动

安装

下载 gateway-1.6.0_SNAPSHOT-597-windows-amd64.zip，解压如下。 gateway-windows-amd64.exe 是启动文件，gateway.yml 是默认配置文件。

启动失败

当 gateway.yml 的 elasticsearch 选项中的 hosts 不能正常响应请求的时候，启动界面如下。

为什么 elasticsearch 不能访问的时候，网关还要继续提供服务呢，为什么不像业务接口启动时在基础业务组件如 MySQL/Redis 不能正常响应就直接 panic？

一方面网关作为 elasticsearch 抵挡流量冲击的城墙，在 elasticsearch 不能提供服务的时候，对之前成功的请求缓存结果，继续提供有限度的服务，为 elasticsearch 修复后上线争取时间。

另一方面业务接口和基础组件是强耦合关系，没有基础组件就完全无法对外提供数据读写服务，而网关与 elasticsearch 是松耦合关系，网关在没有 elasticsearch 的情况下也能对外提供有限度的服务。

在 gateway.yml 的 elasticsearch 选项中的 hosts 改成能够正常响应的 elasticsearch 请求地址。

启动成功

双击 gateway-windows-amd64.exe 文件，启动成功界面如下

访问

API 访问

由启动后终端显示可知，网关的 API 接口地址是 http://localhost:2900

[api.go:262] api listen at: http://0.0.0.0:2900

打开浏览器输入 http://localhost:2900，显示所有可以对外提供的 API 接口

我们选择其中一个，在浏览器中输入 http://localhost:2900/_framework/api/_version 从路由上看该接口是查询产品的版本信息，显示如下

gateway.yml 中可以看到有被注释掉的一段配置，看起来应该是配置 api 地址的地方。

#api:
#  enabled: true
#  network:
#    binding: 127.0.0.1:2900

把注释去掉后尝试把端口改成 2901。

api:
  enabled: true
  network:
    binding: 127.0.0.1:2901

改完后启动 打开浏览器先输入 http://localhost:2900，无法正常响应请求，再输入 http://localhost:2901，可以正常响应，界面和修改配置前访问 http://localhost:2900 的界面一样，说明 API 请求地址成功修改

Elasticsearch 访问

启动日志中显示监听 8000 端口，猜测应该是 elasticsearch 请求地址，打开浏览器输入 http://127.0.0.1:8000/

 entry [my_es_entry] listen at: http://0.0.0.0:8000

gateway.yml 中可以看到 my_es_entry 的 network 绑定 8000 端口，显而易见的这部分就是配置代理转发给 elasticsearch 的地址，所以安装后只需要把以前请求 elasticsearch 的地址修改为该地址。

entry:
  - name: my_es_entry
    enabled: true
    router: my_router
    max_concurrency: 10000
    network:
      binding: 0.0.0.0:8000

总结

我们成功安装和启动极限网关，接下来我们学习怎么根据需求修改配置。