fix: ensure the plugin is always reloaded (#4319)

* fix: ensure the plugin is always reloaded

Only trigger a reset from admin when the etcd's data is different
from the one of admin. So there is no need to add check in node side.

Fix #4314

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

* fix reset check

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

* fix error handling

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

* update test

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

apisix/admin/init.lua

@@ -286,9 +286,22 @@ local function post_reload_plugins()
 end
 
 
-local function sync_local_conf_to_etcd()
-    core.log.warn("sync local conf to etcd")
+local function plugins_eq(old, new)
+    local old_set = {}
+    for _, p in ipairs(old) do
+        old_set[p.name] = p
+    end
+
+    local new_set = {}
+    for _, p in ipairs(new) do
+        new_set[p.name] = p
+    end
+
+    return core.table.set_eq(old_set, new_set)
+end
 
+
+local function sync_local_conf_to_etcd(reset)
     local local_conf = core.config.local_conf()
 
     local plugins = {}
@@ -305,6 +318,42 @@ local function sync_local_conf_to_etcd()
         })
     end
 
+    if reset then
+        local res, err = core.etcd.get("/plugins")
+        if not res then
+            core.log.error("failed to get current plugins: ", err)
+            return
+        end
+
+        if res.status == 404 then
+            -- nothing need to be reset
+            return
+        end
+
+        if res.status ~= 200 then
+            core.log.error("failed to get current plugins, status: ", res.status)
+            return
+        end
+
+        local stored_plugins = res.body.node.value
+        local revision = res.body.node.modifiedIndex
+        if plugins_eq(stored_plugins, plugins) then
+            core.log.info("plugins not changed, don't need to reset")
+            return
+        end
+
+        core.log.warn("sync local conf to etcd")
+
+        local res, err = core.etcd.atomic_set("/plugins", plugins, nil, revision)
+        if not res then
+            core.log.error("failed to set plugins: ", err)
+        end
+
+        return
+    end
+
+    core.log.warn("sync local conf to etcd")
+
     -- need to store all plugins name into one key so that it can be updated atomically
     local res, err = core.etcd.set("/plugins", plugins)
     if not res then
@@ -364,7 +413,8 @@ function _M.init_worker()
                 return
             end
 
-            sync_local_conf_to_etcd()
+            -- try to reset the /plugins to the current configuration in the admin
+            sync_local_conf_to_etcd(true)
         end)
 
         if not ok then

apisix/plugin.lua

@@ -137,25 +137,6 @@ local function load_plugin(name, plugins_list, is_stream_plugin)
 end
 
 
-local function plugins_eq(old, new)
-    local eq = core.table.set_eq(old, new)
-    if not eq then
-        core.log.info("plugin list changed")
-        return false
-    end
-
-    for name, plugin in pairs(old) do
-        eq = core.table.deep_eq(plugin.attr, plugin_attr(name))
-        if not eq then
-            core.log.info("plugin_attr of ", name, " changed")
-            return false
-        end
-    end
-
-    return true
-end
-
-
 local function load(plugin_names)
     local processed = {}
     for _, name in ipairs(plugin_names) do
@@ -164,12 +145,6 @@ local function load(plugin_names)
         end
     end
 
-    -- the same configure may be synchronized more than one
-    if plugins_eq(local_plugins_hash, processed) then
-        core.log.info("plugins not changed")
-        return true
-    end
-
     core.log.warn("new plugins: ", core.json.delay_encode(processed))
 
     for name in pairs(local_plugins_hash) do
@@ -212,12 +187,6 @@ local function load_stream(plugin_names)
         end
     end
 
-    -- the same configure may be synchronized more than one
-    if plugins_eq(stream_local_plugins_hash, processed) then
-        core.log.info("plugins not changed")
-        return true
-    end
-
     core.log.warn("new plugins: ", core.json.delay_encode(processed))
 
     for name in pairs(stream_local_plugins_hash) do

t/admin/plugins-reload.t

@@ -42,6 +42,8 @@ __DATA__
 location /t {
     content_by_lua_block {
         local t = require("lib.test_admin").test
+        -- now the plugin will be loaded twice,
+        -- one during startup and the other one by reload
         local code, _, org_body = t('/apisix/admin/plugins/reload',
                                     ngx.HTTP_PUT)
 
@@ -54,15 +56,15 @@ location /t {
 GET /t
 --- response_body
 done
+--- grep_error_log eval
+qr/sync local conf to etcd/
+--- grep_error_log_out
+sync local conf to etcd
 --- error_log
-load plugin times: 1
-load plugin times: 1
+load plugin times: 2
+load plugin times: 2
 start to hot reload plugins
 start to hot reload plugins
-load(): plugins not changed
-load_stream(): plugins not changed
-load(): plugins not changed
-load_stream(): plugins not changed
 
 
 
@@ -80,8 +82,7 @@ location /t {
             automatic = true,
             single_item = true,
             filter = function(item)
-                -- called twice before reload,
-                -- one for worker start, another for sync data from admin
+                -- called once before reload for sync data from admin
                 ngx.log(ngx.WARN, "reload plugins on node ",
                         before_reload and "before reload" or "after reload")
                 ngx.log(ngx.WARN, require("toolkit.json").encode(item.value))
@@ -122,7 +123,6 @@ done
 qr/reload plugins on node \w+ reload/
 --- grep_error_log_out
 reload plugins on node before reload
-reload plugins on node before reload
 reload plugins on node after reload
 --- error_log
 filter(): [{"name":"jwt-auth"},{"name":"mqtt-proxy","stream":true}]
@@ -180,6 +180,7 @@ plugin_attr:
         local code, _, org_body = t('/apisix/admin/plugins/reload',
                                     ngx.HTTP_PUT)
         ngx.say(org_body)
+        ngx.sleep(0.1)
     }
 }
 --- request
@@ -196,9 +197,9 @@ example-plugin get plugin attr val: 0
 example-plugin get plugin attr val: 1
 example-plugin get plugin attr val: 1
 example-plugin get plugin attr val: 1
---- error_log
-plugin_attr of example-plugin changed
-plugins not changed
+example-plugin get plugin attr val: 1
+example-plugin get plugin attr val: 1
+example-plugin get plugin attr val: 1
 
 
 
@@ -301,3 +302,84 @@ done
 qr/Instance report fails/
 --- grep_error_log_out
 Instance report fails
+
+
+
+=== TEST 6: check disabling plugin via etcd
+--- 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,
+                 [[{
+                        "plugins": {
+                            "echo": {
+                                "body":"hello upstream\n"
+                            }
+                        },
+                        "upstream": {
+                            "nodes": {
+                                "127.0.0.1:1980": 1
+                            },
+                            "type": "roundrobin"
+                        },
+                        "uri": "/hello"
+                }]]
+                )
+
+            if code >= 300 then
+                ngx.status = code
+            end
+            ngx.say(body)
+        }
+    }
+--- request
+GET /t
+--- response_body
+passed
+
+
+
+=== TEST 7: hit
+--- yaml_config
+apisix:
+  node_listen: 1984
+  enable_admin: false
+--- request
+GET /hello
+--- response_body
+hello upstream
+
+
+
+=== TEST 8: hit after disabling echo
+--- yaml_config
+apisix:
+  node_listen: 1984
+  enable_admin: false
+--- config
+location /t {
+    content_by_lua_block {
+        local t = require("lib.test_admin").test
+        local etcd = require("apisix.core.etcd")
+        assert(etcd.set("/plugins", {{name = "jwt-auth"}}))
+
+        ngx.sleep(0.2)
+
+        local http = require "resty.http"
+        local httpc = http.new()
+        local uri = "http://127.0.0.1:" .. ngx.var.server_port
+                    .. "/hello"
+        local res, err = httpc:request_uri(uri)
+        if not res then
+            ngx.say(err)
+            return
+        end
+        ngx.print(res.body)
+    }
+}
+--- request
+GET /t
+--- response_body
+hello world

t/config-center-yaml/plugin.t

@@ -82,7 +82,7 @@ load(): new plugins
 
 
 
-=== TEST 2: plugins not changed
+=== TEST 2: plugins not changed, but still need to reload
 --- yaml_config
 apisix:
     node_listen: 1984
@@ -104,12 +104,17 @@ plugins:
 GET /hello
 --- response_body
 hello world
---- error_log
-load(): loaded plugin and sort by priority: 3000 name: ip-restriction
-load(): loaded plugin and sort by priority: 2510 name: jwt-auth
-load_stream(): loaded stream plugin and sort by priority: 1000 name: mqtt-proxy
-load(): plugins not changed
-load_stream(): plugins not changed
+--- grep_error_log eval
+qr/loaded plugin and sort by priority: \d+ name: [^,]+/
+--- grep_error_log_out
+loaded plugin and sort by priority: 3000 name: ip-restriction
+loaded plugin and sort by priority: 2510 name: jwt-auth
+loaded plugin and sort by priority: 3000 name: ip-restriction
+loaded plugin and sort by priority: 2510 name: jwt-auth
+loaded plugin and sort by priority: 3000 name: ip-restriction
+loaded plugin and sort by priority: 2510 name: jwt-auth
+loaded plugin and sort by priority: 3000 name: ip-restriction
+loaded plugin and sort by priority: 2510 name: jwt-auth