feat(elasticsearch-logger): support multi elasticsearch endpoints (#8604

)

apisix/plugins/elasticsearch-logger.lua

@@ -22,6 +22,7 @@ local bp_manager_mod  = require("apisix.utils.batch-processor-manager")
 
 local ngx             = ngx
 local str_format      = core.string.format
+local math_random     = math.random
 
 local plugin_name = "elasticsearch-logger"
 local batch_processor_manager = bp_manager_mod.new(plugin_name)
@@ -30,10 +31,19 @@ local batch_processor_manager = bp_manager_mod.new(plugin_name)
 local schema = {
     type = "object",
     properties = {
+        -- deprecated, use "endpoint_addrs" instead
         endpoint_addr = {
             type = "string",
             pattern = "[^/]$",
         },
+        endpoint_addrs = {
+            type = "array",
+            minItems = 1,
+            items = {
+                type = "string",
+                pattern = "[^/]$",
+            },
+        },
         field = {
             type = "object",
             properties = {
@@ -67,7 +77,10 @@ local schema = {
         }
     },
     encrypt_fields = {"auth.password"},
-    required = { "endpoint_addr", "field" },
+    oneOf = {
+        {required = {"endpoint_addr", "field"}},
+        {required = {"endpoint_addrs", "field"}}
+    },
 }
 
 
@@ -114,7 +127,13 @@ local function send_to_elasticsearch(conf, entries)
         return false, str_format("create http error: %s", err)
     end
 
-    local uri = conf.endpoint_addr .. "/_bulk"
+    local selected_endpoint_addr
+    if conf.endpoint_addr then
+        selected_endpoint_addr = conf.endpoint_addr
+    else
+        selected_endpoint_addr = conf.endpoint_addrs[math_random(#conf.endpoint_addrs)]
+    end
+    local uri = selected_endpoint_addr .. "/_bulk"
     local body = core.table.concat(entries, "")
     local headers = {["Content-Type"] = "application/x-ndjson"}
     if conf.auth then

docs/en/latest/plugins/elasticsearch-logger.md

@@ -37,7 +37,8 @@ When the Plugin is enabled, APISIX will serialize the request context informatio
 
 | Name          | Type    | Required | Default                     | Description                                                  |
 | ------------- | ------- | -------- | --------------------------- | ------------------------------------------------------------ |
-| endpoint_addr | string  | True     |                             | Elasticsearch API.                                            |
+| endpoint_addr | string  | Deprecated     |                             | Deprecated. Use `endpoint_addrs` instead. Elasticsearch API.                                            |
+| endpoint_addrs  | array  | True     |                             | Elasticsearch API. If multiple endpoints are configured, they will be written randomly.                                            |
 | field         | array   | True     |                             | Elasticsearch `field` configuration.                          |
 | field.index   | string  | True     |                             | Elasticsearch [_index field](https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-index-field.html#mapping-index-field). |
 | field.type    | string  | False    | Elasticsearch default value | Elasticsearch [_type field](https://www.elastic.co/guide/en/elasticsearch/reference/7.17/mapping-type-field.html#mapping-type-field). |

docs/zh/latest/plugins/elasticsearch-logger.md

@@ -38,7 +38,8 @@ description: 本文介绍了 API 网关 Apache APISIX 的 elasticsearch-logger 
 
 | 名称          | 类型    | 必选项 | 默认值               | 描述                                                         |
 | ------------- | ------- | -------- | -------------------- | ------------------------------------------------------------ |
-| endpoint_addr | string  | 是       |                      | Elasticsearch API。                                           |
+| endpoint_addr | string  | 废弃       |                      | Elasticsearch API。推荐使用 `endpoint_addrs`                                           |
+| endpoint_addrs | array  | 是       |                      | Elasticsearch API。如果配置多个 `endpoints`，日志将会随机写入到各个 `endpoints`。                                           |
 | field         | array   | 是       |                      | Elasticsearch `field`配置信息。                                |
 | field.index   | string  | 是       |                      | Elasticsearch `[_index field](https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-index-field.html#mapping-index-field)`。 |
 | field.type    | string  | 否       | Elasticsearch 默认值 | Elasticsearch `[_type field](https://www.elastic.co/guide/en/elasticsearch/reference/7.17/mapping-type-field.html#mapping-type-field)` |

t/plugin/elasticsearch-logger.t

@@ -106,8 +106,8 @@ __DATA__
 --- response_body_like
 passed
 passed
-property "endpoint_addr" is required
-property "field" is required
+value should match only one schema, but matches none
+value should match only one schema, but matches none
 property "field" validation failed: property "index" is required
 property "endpoint_addr" validation failed: failed to match pattern "\[\^/\]\$" with "http://127.0.0.1:9200/"
 
@@ -515,3 +515,70 @@ apisix:
 --- response_body
 123456
 PTQvJEaPcNOXcOHeErC0XQ==
+
+
+
+=== TEST 13: add plugin on routes using multi elasticsearch-logger
+--- config
+    location /t {
+        content_by_lua_block {
+            local t = require("lib.test_admin").test
+            local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, {
+                uri = "/hello",
+                upstream = {
+                    type = "roundrobin",
+                    nodes = {
+                        ["127.0.0.1:1980"] = 1
+                    }
+                },
+                plugins = {
+                    ["elasticsearch-logger"] = {
+                        endpoint_addrs = {"http://127.0.0.1:9200", "http://127.0.0.1:9201"},
+                        field = {
+                            index = "services"
+                        },
+                        batch_max_size = 1,
+                        inactive_timeout = 1
+                    }
+                }
+            })
+
+            if code >= 300 then
+                ngx.status = code
+            end
+            ngx.say(body)
+        }
+    }
+--- response_body
+passed
+
+
+
+=== TEST 14: to show that different endpoints will be chosen randomly
+--- config
+    location /t {
+        content_by_lua_block {
+            local code_count = {}
+            local t = require("lib.test_admin").test
+            for i = 1, 12 do
+                local code, body = t('/hello', ngx.HTTP_GET)
+                if code ~= 200 then
+                    ngx.say("code: ", code, " body: ", body)
+                end
+                code_count[code] = (code_count[code] or 0) + 1
+            end
+
+            local code_arr = {}
+            for code, count in pairs(code_count) do
+                table.insert(code_arr, {code = code, count = count})
+            end
+
+            ngx.say(require("toolkit.json").encode(code_arr))
+            ngx.exit(200)
+        }
+    }
+--- response_body
+[{"code":200,"count":12}]
+--- error_log
+http://127.0.0.1:9200/_bulk
+http://127.0.0.1:9201/_bulk