optimize: Use lru to avoid resolving IP addresses repeatedly . (apache#1772)

* optimize: Use lru to avoid resolving IP addresses repeatedly .
Cached the global rules to `ctx` .

* optimzie: used a longer time interval for etcd and flush access log.

* optimize: return upstream node directly if the count is 1 .

* optimize: avoid to cache useless variable.

Author: membphis

apisix/balancer.lua

@@ -39,6 +39,9 @@ local lrucache_server_picker = core.lrucache.new({
 local lrucache_checker = core.lrucache.new({
     ttl = 300, count = 256
 })
+local lrucache_addr = core.lrucache.new({
+    ttl = 300, count = 1024 * 4
+})
 
 
 local _M = {
@@ -143,25 +146,47 @@ local function create_server_picker(upstream, checker)
 end
 
 
+local function parse_addr(addr)
+    local host, port, err = core.utils.parse_addr(addr)
+    return {host = host, port = port}, err
+end
+
+
 local function pick_server(route, ctx)
     core.log.info("route: ", core.json.delay_encode(route, true))
     core.log.info("ctx: ", core.json.delay_encode(ctx, true))
-    local healthcheck_parent = ctx.upstream_healthcheck_parent
     local up_conf = ctx.upstream_conf
-    local version = ctx.upstream_version
-    local key = ctx.upstream_key
-
     if up_conf.service_name then
         if not discovery then
-            return nil, nil, "discovery is uninitialized"
+            return nil, "discovery is uninitialized"
         end
         up_conf.nodes = discovery.nodes(up_conf.service_name)
     end
 
-    if not up_conf.nodes or #up_conf.nodes == 0 then
-        return nil, nil, "no valid upstream node"
+    local nodes_count = up_conf.nodes and #up_conf.nodes or 0
+    if nodes_count == 0 then
+        return nil, "no valid upstream node"
+    end
+
+    if up_conf.timeout then
+        local timeout = up_conf.timeout
+        local ok, err = set_timeouts(timeout.connect, timeout.send,
+                                     timeout.read)
+        if not ok then
+            core.log.error("could not set upstream timeouts: ", err)
+        end
     end
 
+    if nodes_count == 1 then
+        local node = up_conf.nodes[1]
+        ctx.balancer_ip = node.host
+        ctx.balancer_port = node.port
+        return node
+    end
+
+    local healthcheck_parent = ctx.upstream_healthcheck_parent
+    local version = ctx.upstream_version
+    local key = ctx.upstream_key
     local checker = fetch_healthchecker(up_conf, healthcheck_parent, version)
 
     ctx.balancer_try_count = (ctx.balancer_try_count or 0) + 1
@@ -197,28 +222,25 @@ local function pick_server(route, ctx)
     local server_picker = lrucache_server_picker(key, version,
                             create_server_picker, up_conf, checker)
     if not server_picker then
-        return nil, nil, "failed to fetch server picker"
+        return nil, "failed to fetch server picker"
     end
 
     local server, err = server_picker.get(ctx)
     if not server then
         err = err or "no valid upstream node"
-        return nil, nil, "failed to find valid upstream server, " .. err
+        return nil, "failed to find valid upstream server, " .. err
     end
 
-    if up_conf.timeout then
-        local timeout = up_conf.timeout
-        local ok, err = set_timeouts(timeout.connect, timeout.send, timeout.read)
-        if not ok then
-            core.log.error("could not set upstream timeouts: ", err)
-        end
+    local res, err = lrucache_addr(server, nil, parse_addr, server)
+    ctx.balancer_ip = res.host
+    ctx.balancer_port = res.port
+    -- core.log.info("proxy to ", host, ":", port)
+    if err then
+        core.log.error("failed to parse server addr: ", server, " err: ", err)
+        return core.response.exit(502)
     end
 
-    local ip, port, err = core.utils.parse_addr(server)
-    ctx.balancer_ip = ip
-    ctx.balancer_port = port
-    core.log.info("proxy to ", ip, ":", port)
-    return ip, port, err
+    return res
 end
 
 
@@ -227,16 +249,16 @@ _M.pick_server = pick_server
 
 
 function _M.run(route, ctx)
-    local ip, port, err = pick_server(route, ctx)
-    if err then
+    local server, err = pick_server(route, ctx)
+    if not server then
         core.log.error("failed to pick server: ", err)
         return core.response.exit(502)
     end
 
-    local ok, err = balancer.set_current_peer(ip, port)
+    local ok, err = balancer.set_current_peer(server.host, server.port)
     if not ok then
-        core.log.error("failed to set server peer [", ip, ":", port,
-                       "] err: ", err)
+        core.log.error("failed to set server peer [", server.host, ":",
+                       server.port, "] err: ", err)
         return core.response.exit(502)
     end
 

apisix/init.lua

@@ -112,7 +112,7 @@ local function run_plugin(phase, plugins, api_ctx)
     end
 
     plugins = plugins or api_ctx.plugins
-    if not plugins then
+    if not plugins or #plugins == 0 then
         return api_ctx
     end
 
@@ -262,6 +262,8 @@ function _M.http_access_phase()
         api_ctx.conf_type = nil
         api_ctx.conf_version = nil
         api_ctx.conf_id = nil
+
+        api_ctx.global_rules = router.global_rules
     end
 
     router.router_http.match(api_ctx)
@@ -433,25 +435,24 @@ function _M.grpc_access_phase()
 end
 
 
-local function common_phase(plugin_name)
+local function common_phase(phase_name)
     local api_ctx = ngx.ctx.api_ctx
     if not api_ctx then
         return
     end
 
-    if router.global_rules and router.global_rules.values
-            and #router.global_rules.values > 0
-    then
+    if api_ctx.global_rules then
         local plugins = core.tablepool.fetch("plugins", 32, 0)
-        local values = router.global_rules.values
+        local values = api_ctx.global_rules.values
         for _, global_rule in config_util.iterate_values(values) do
             core.table.clear(plugins)
             plugins = plugin.filter(global_rule, plugins)
-            run_plugin(plugin_name, plugins, api_ctx)
+            run_plugin(phase_name, plugins, api_ctx)
         end
         core.tablepool.release("plugins", plugins)
     end
-    run_plugin(plugin_name, nil, api_ctx)
+
+    run_plugin(phase_name, nil, api_ctx)
     return api_ctx
 end
 

apisix/plugin.lua

@@ -235,7 +235,8 @@ end
 function _M.filter(user_route, plugins)
     plugins = plugins or core.table.new(#local_plugins * 2, 0)
     local user_plugin_conf = user_route.value.plugins
-    if user_plugin_conf == nil then
+    if user_plugin_conf == nil or
+       core.table.nkeys(user_plugin_conf) == 0 then
         if local_conf and local_conf.apisix.enable_debug then
             core.response.set_header("Apisix-Plugins", "no plugin")
         end

benchmark/fake-apisix/conf/nginx.conf

@@ -24,7 +24,6 @@ pid logs/nginx.pid;
 worker_rlimit_nofile 20480;
 
 events {
-    accept_mutex off;
     worker_connections 10620;
 }
 
@@ -33,6 +32,9 @@ worker_shutdown_timeout 3;
 http {
     lua_package_path  "$prefix/lua/?.lua;;";
 
+    log_format main '$remote_addr - $remote_user [$time_local] $http_host "$request" $status $body_bytes_sent $request_time "$http_referer" "$http_user_agent" $upstream_addr $upstream_status $upstream_response_time';
+    access_log logs/access.log main buffer=16384 flush=5;
+
     init_by_lua_block {
         require "resty.core"
         apisix = require("apisix")
@@ -60,8 +62,6 @@ http {
 
         listen 9080;
 
-        access_log off;
-
         server_tokens off;
         more_set_headers 'Server: APISIX web server';
 
@@ -106,6 +106,10 @@ http {
                 apisix.http_header_filter_phase()
             }
 
+            body_filter_by_lua_block {
+                apisix.http_body_filter_phase()
+            }
+
             log_by_lua_block {
                 apisix.http_log_phase()
             }

benchmark/fake-apisix/lua/apisix.lua

@@ -25,7 +25,7 @@ end
 
 local function fake_fetch()
     ngx.ctx.ip = "127.0.0.1"
-    ngx.ctx.port = 80
+    ngx.ctx.port = 1980
 end
 
 function _M.http_access_phase()
@@ -42,6 +42,12 @@ function _M.http_header_filter_phase()
     end
 end
 
+function _M.http_body_filter_phase()
+    if ngx.ctx then
+        -- do something
+    end
+end
+
 function _M.http_log_phase()
     if ngx.ctx then
         -- do something

bin/apisix

@@ -229,7 +229,7 @@ http {
 
     log_format main '$remote_addr - $remote_user [$time_local] $http_host "$request" $status $body_bytes_sent $request_time "$http_referer" "$http_user_agent" $upstream_addr $upstream_status $upstream_response_time';
 
-    access_log {* http.access_log *} main buffer=16384 flush=1;
+    access_log {* http.access_log *} main buffer=16384 flush=3;
     open_file_cache  max=1000 inactive=60;
     client_max_body_size 0;
     keepalive_timeout {* http.keepalive_timeout *};

conf/config.yaml

@@ -121,7 +121,7 @@ etcd:
   host:                           # it's possible to define multiple etcd hosts addresses of the same etcd cluster.
     - "http://127.0.0.1:2379"     # multiple etcd address
   prefix: "/apisix"               # apisix configurations prefix
-  timeout: 3                      # 3 seconds
+  timeout: 30                     # 3 seconds
   # user: root                     # root username for etcd
   # password: 5tHkHhYkjr6cQY        # root password for etcd
 #eureka:

t/admin/balancer.t

@@ -33,13 +33,13 @@ add_block_preprocessor(sub {
         local balancer = require("apisix.balancer")
         local res = {}
         for i = 1, count or 12 do
-            local host, port, err = balancer.pick_server(route, ctx)
+            local server, err = balancer.pick_server(route, ctx)
             if err then
                 ngx.say("failed: ", err)
             end
 
-            core.log.warn("host: ", host, " port: ", port)
-            res[host] = (res[host] or 0) + 1
+            core.log.warn("host: ", server.host, " port: ", server.port)
+            res[server.host] = (res[server.host] or 0) + 1
         end
 
         local keys = {}
@@ -144,6 +144,7 @@ host: 39.97.63.217 count: 6
             -- cached by version
             up_conf.nodes = {
                 {host = "39.97.63.218", port = 80, weight = 1},
+                {host = "39.97.63.219", port = 80, weight = 0},
             }
             test(route, ctx)
 
@@ -192,6 +193,7 @@ host: 39.97.63.218 count: 12
             -- cached by version
             up_conf.nodes = {
                 {host = "39.97.63.218", port = 80, weight = 1},
+                {host = "39.97.63.219", port = 80, weight = 0},
             }
             test(route, ctx)
 
@@ -208,3 +210,41 @@ host: 39.97.63.215 count: 12
 host: 39.97.63.218 count: 12
 --- no_error_log
 [error]
+
+
+
+=== TEST 5: return item directly if only have one item in `nodes`
+--- config
+    location /t {
+        content_by_lua_block {
+            local up_conf = {
+                type = "roundrobin",
+                nodes = {
+                    {host = "39.97.63.215", port = 80, weight = 1},
+                    {host = "39.97.63.216", port = 81, weight = 1},
+                    {host = "39.97.63.217", port = 82, weight = 1},
+                }
+            }
+            local ctx = {}
+            ctx.upstream_conf = up_conf
+            ctx.upstream_version = 1
+            ctx.upstream_key = up_conf.type .. "#route_" .. "id"
+
+            test(route, ctx)
+
+            -- one item in nodes, return it directly
+            up_conf.nodes = {
+                {host = "39.97.63.218", port = 80, weight = 1},
+            }
+            test(route, ctx)
+        }
+    }
+--- request
+GET /t
+--- response_body
+host: 39.97.63.215 count: 4
+host: 39.97.63.216 count: 4
+host: 39.97.63.217 count: 4
+host: 39.97.63.218 count: 12
+--- no_error_log
+[error]