feat: allow user-defined balancer with metadata in node

Signed-off-by: spacewander <spacewanderlzx@gmail.com>

apisix/schema_def.lua

@@ -389,7 +389,6 @@ local upstream_schema = {
         type = {
             description = "algorithms of load balancing",
             type = "string",
-            enum = {"chash", "roundrobin", "ewma", "least_conn"}
         },
         checks = health_checker,
         hash_on = {

apisix/utils/upstream.lua

@@ -47,7 +47,7 @@ function _M.compare_upstream_node(up_conf, new_t)
     for i = 1, #new_t do
         local new_node = new_t[i]
         local old_node = old_t[i]
-        for _, name in ipairs({"host", "port", "weight", "priority"}) do
+        for _, name in ipairs({"host", "port", "weight", "priority", "metadata"}) do
             if new_node[name] ~= old_node[name] then
                 return false
             end

docs/en/latest/admin-api.md

@@ -568,6 +568,7 @@ In addition to the basic complex equalization algorithm selection, APISIX's Upst
 * `chash`: consistent hash
 * `ewma`: pick one of node which has minimum latency. See https://en.wikipedia.org/wiki/EWMA_chart for details.
 * `least_conn`: pick node which has the lowest `(active_conn + 1) / weight`. Note the `active connection` concept is the same with Nginx: it is a connection in used by a request.
+* user-defined balancer which can be loaed via `require("apisix.balancer.your_balancer")`.
 
 `hash_on` can be set to different types:
 

docs/zh/latest/admin-api.md

@@ -574,6 +574,7 @@ APISIX 的 Upstream 除了基本的负载均衡算法选择外，还支持对上
 - `chash`: 一致性哈希
 - `ewma`: 选择延迟最小的节点，计算细节参考 https://en.wikipedia.org/wiki/EWMA_chart
 - `least_conn`: 选择 `(active_conn + 1) / weight` 最小的节点。注意这里的 `active connection` 概念跟 Nginx 的相同：它是当前正在被请求使用的连接。
+- 用户自定义的 balancer，需要可以通过 `require("apisix.balancer.your_balancer")` 来加载。
 
 `hash_on` 比较复杂，这里专门说明下：
 

t/admin/upstream.t

@@ -453,7 +453,7 @@ passed
 
 
 
-=== TEST 13: invalid type
+=== TEST 13: unknown type
 --- config
     location /t {
         content_by_lua_block {
@@ -465,7 +465,7 @@ passed
                     "nodes": {
                         "127.0.0.1:8080": 1
                     },
-                    "type": "invalid_type"
+                    "type": "unknown"
                 }]]
                 )
 
@@ -475,9 +475,8 @@ passed
     }
 --- request
 GET /t
---- error_code: 400
---- response_body
-{"error_msg":"invalid configuration: property \"type\" validation failed: matches none of the enum values"}
+--- response_body chomp
+passed
 --- no_error_log
 [error]
 

t/node/invalid-upstream.t

@@ -89,36 +89,7 @@ passed
 
 
 
-=== TEST 4: invalid upstream(wrong type)
---- config
-    location /t {
-        content_by_lua_block {
-            local t = require("lib.test_admin").test
-            local code, body = t('/apisix/admin/upstreams/1',
-                ngx.HTTP_PUT,
-                [[{
-                    "nodes": {
-                        "127.0.0.1:1980": 1
-                    },
-                    "type": "roundrobin_invalid"
-                }]]
-            )
-
-            if code >= 300 then
-                ngx.status = code
-            end
-            ngx.print(body)
-        }
-    }
---- request
-GET /t
---- error_code: 400
---- response_body
-{"error_msg":"invalid configuration: property \"type\" validation failed: matches none of the enum values"}
-
-
-
-=== TEST 5: set valid upstream(id: 1)
+=== TEST 4: set valid upstream(id: 1)
 --- config
     location /t {
         content_by_lua_block {
@@ -145,7 +116,7 @@ qr/"nodes":\{"127.0.0.1:1980":1\}/
 
 
 
-=== TEST 6: no error log
+=== TEST 5: no error log
 --- config
     location /t {
         content_by_lua_block {

t/node/upstream-discovery.t

@@ -331,3 +331,96 @@ qr/proxy request to \S+/
 --- grep_error_log_out
 proxy request to 127.0.0.1:1979
 proxy request to 0.0.0.0:1980
+
+
+
+=== TEST 7: create new server picker when metadata change
+--- apisix_yaml
+routes:
+  -
+    uris:
+        - /hello
+    upstream_id: 1
+--- config
+    location /t {
+        content_by_lua_block {
+            local discovery = require("apisix.discovery.init").discovery
+            discovery.mock = {
+                nodes = function()
+                    return {
+                        {host = "127.0.0.1", port = 1980, weight = 1, metadata = {a = 1}},
+                        {host = "0.0.0.0", port = 1980, weight = 1, metadata = {}},
+                    }
+                end
+            }
+            local http = require "resty.http"
+            local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+            local httpc = http.new()
+            local res, err = httpc:request_uri(uri, {method = "GET", keepalive = false})
+            ngx.say(res.status)
+
+            discovery.mock = {
+                nodes = function()
+                    return {
+                        {host = "127.0.0.1", port = 1980, weight = 1, metadata = {a = 1}},
+                        {host = "0.0.0.0", port = 1980, weight = 1, metadata = {b = 1}},
+                    }
+                end
+            }
+            local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+            local httpc = http.new()
+            local res, err = httpc:request_uri(uri, {method = "GET", keepalive = false})
+        }
+    }
+--- grep_error_log eval
+qr/create_obj_fun\(\): upstream nodes:/
+--- grep_error_log_out
+create_obj_fun(): upstream nodes:
+create_obj_fun(): upstream nodes:
+
+
+
+=== TEST 8: don't create new server picker when metadata doesn't change
+--- apisix_yaml
+routes:
+  -
+    uris:
+        - /hello
+    upstream_id: 1
+--- config
+    location /t {
+        content_by_lua_block {
+            local discovery = require("apisix.discovery.init").discovery
+            local meta1 = {a = 1}
+            local meta2 = {b = 2}
+            discovery.mock = {
+                nodes = function()
+                    return {
+                        {host = "127.0.0.1", port = 1980, weight = 1, metadata = meta1},
+                        {host = "0.0.0.0", port = 1980, weight = 1, metadata = meta2},
+                    }
+                end
+            }
+            local http = require "resty.http"
+            local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+            local httpc = http.new()
+            local res, err = httpc:request_uri(uri, {method = "GET", keepalive = false})
+            ngx.say(res.status)
+
+            discovery.mock = {
+                nodes = function()
+                    return {
+                        {host = "127.0.0.1", port = 1980, weight = 1, metadata = meta1},
+                        {host = "0.0.0.0", port = 1980, weight = 1, metadata = meta2},
+                    }
+                end
+            }
+            local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+            local httpc = http.new()
+            local res, err = httpc:request_uri(uri, {method = "GET", keepalive = false})
+        }
+    }
+--- grep_error_log eval
+qr/create_obj_fun\(\): upstream nodes:/
+--- grep_error_log_out
+create_obj_fun(): upstream nodes: