fix: plugin hot reload should work on node

FAQ.md

@@ -259,17 +259,8 @@ Now you can trace the info level log in logs/error.log.
 
 ## How to reload your own plugin
 
-The Apache APISIX plugin supports hot reloading. If your APISIX node has the Admin API turned on, then for scenarios such as adding / deleting / modifying plugins, you can hot reload the plugin by calling the HTTP interface without restarting the service.
-
-```shell
-curl http://127.0.0.1:9080/apisix/admin/plugins/reload -H 'X-API-KEY: edd1c9f034335f136f87ad84b625c8f1' -X PUT
-```
-
-If your APISIX node does not open the Admin API, then you can manually load the plug-in by reloading APISIX.
-
-```shell
-apisix reload
-```
+The Apache APISIX plugin supports hot reloading.
+See the `Hot reload` section in [plugins](./doc/plugins.md) for how to do that.
 
 ## How to make APISIX listen on multiple ports when handling HTTP or HTTPS requests?
 

FAQ_CN.md

@@ -209,17 +209,9 @@ Server: APISIX web server
 
 ## 如何加载自己编写的插件
 
-Apache APISIX 的插件支持热加载，如果你的 APISIX 节点打开了 Admin API，那么对于新增/删除/修改插件等场景，均可以通过调用 HTTP 接口的方式热加载插件，不需要重启服务。
+Apache APISIX 的插件支持热加载。
 
-```shell
-curl http://127.0.0.1:9080/apisix/admin/plugins/reload -H 'X-API-KEY: edd1c9f034335f136f87ad84b625c8f1' -X PUT
-```
-
-如果你的 APISIX 节点并没有打开 Admin API，那么你可以通过手动 reload APISIX 的方式加载插件。
-
-```shell
-apisix reload
-```
+具体怎么做参考 [插件](./doc/zh-cn/plugins.md) 中关于“热加载”的部分。
 
 ## 如何让 APISIX 在处理 HTTP 或 HTTPS 请求时监听多个端口
 

apisix/admin/init.lua

@@ -20,10 +20,14 @@ local route = require("resty.radixtree")
 local plugin = require("apisix.plugin")
 local ngx = ngx
 local get_method = ngx.req.get_method
+local ngx_time = ngx.time
+local ngx_timer_at = ngx.timer.at
+local ngx_worker_id = ngx.worker.id
 local tonumber = tonumber
 local str_lower = string.lower
 local reload_event = "/apisix/admin/plugins/reload"
 local ipairs = ipairs
+local error = error
 local events
 local MAX_REQ_BODY = 1024 * 1024 * 1.5      -- 1.5 MiB
 
@@ -245,7 +249,7 @@ local function post_reload_plugins()
         core.response.exit(401)
     end
 
-    local success, err = events.post(reload_event, get_method(), ngx.time())
+    local success, err = events.post(reload_event, get_method(), ngx_time())
     if not success then
         core.response.exit(500, err)
     end
@@ -254,9 +258,40 @@ local function post_reload_plugins()
 end
 
 
+local function sync_local_conf_to_etcd()
+    core.log.warn("sync local conf to etcd")
+
+    local local_conf = core.config.local_conf()
+
+    local plugins = {}
+    for _, name in ipairs(local_conf.plugins) do
+        core.table.insert(plugins, {
+            name = name,
+        })
+    end
+
+    for _, name in ipairs(local_conf.stream_plugins) do
+        core.table.insert(plugins, {
+            name = name,
+            stream = true,
+        })
+    end
+
+    -- 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
+        core.log.error("failed to set plugins: ", err)
+    end
+end
+
+
 local function reload_plugins(data, event, source, pid)
     core.log.info("start to hot reload plugins")
     plugin.load()
+
+    if ngx_worker_id() == 0 then
+        sync_local_conf_to_etcd()
+    end
 end
 
 
@@ -294,6 +329,20 @@ function _M.init_worker()
     events = require("resty.worker.events")
 
     events.register(reload_plugins, reload_event, "PUT")
+
+    if ngx_worker_id() == 0 then
+        local ok, err = ngx_timer_at(0, function(premature)
+            if premature then
+                return
+            end
+
+            sync_local_conf_to_etcd()
+        end)
+
+        if not ok then
+            error("failed to sync local configure to etcd: " .. err)
+        end
+    end
 end
 
 

apisix/core/config_etcd.lua

@@ -185,10 +185,6 @@ local function sync_data(self)
             return false, err
         end
 
-        if not dir_res.nodes then
-            dir_res.nodes = {}
-        end
-
         if self.values then
             for i, val in ipairs(self.values) do
                 if val and val.clean_handlers then
@@ -203,19 +199,14 @@ local function sync_data(self)
             self.values_hash = nil
         end
 
-        self.values = new_tab(#dir_res.nodes, 0)
-        self.values_hash = new_tab(0, #dir_res.nodes)
-
         local changed = false
-        for _, item in ipairs(dir_res.nodes) do
-            local key = short_key(self, item.key)
-            local data_valid = true
-            if type(item.value) ~= "table" then
-                data_valid = false
-                log.error("invalid item data of [", self.key .. "/" .. key,
-                          "], val: ", item.value,
-                          ", it shoud be a object")
-            end
+
+        if self.single_item then
+            self.values = new_tab(1, 0)
+            self.values_hash = new_tab(0, 1)
+
+            local item = dir_res
+            local data_valid = item.value ~= nil
 
             if data_valid and self.item_schema then
                 data_valid, err = check_schema(self.item_schema, item.value)
@@ -228,8 +219,8 @@ local function sync_data(self)
             if data_valid then
                 changed = true
                 insert_tab(self.values, item)
-                self.values_hash[key] = #self.values
-                item.value.id = key
+                self.values_hash[self.key] = #self.values
+
                 item.clean_handlers = {}
 
                 if self.filter then
@@ -238,6 +229,48 @@ local function sync_data(self)
             end
 
             self:upgrade_version(item.modifiedIndex)
+
+        else
+            if not dir_res.nodes then
+                dir_res.nodes = {}
+            end
+
+            self.values = new_tab(#dir_res.nodes, 0)
+            self.values_hash = new_tab(0, #dir_res.nodes)
+
+            for _, item in ipairs(dir_res.nodes) do
+                local key = short_key(self, item.key)
+                local data_valid = true
+                if type(item.value) ~= "table" then
+                    data_valid = false
+                    log.error("invalid item data of [", self.key .. "/" .. key,
+                              "], val: ", item.value,
+                              ", it shoud be a object")
+                end
+
+                if data_valid and self.item_schema then
+                    data_valid, err = check_schema(self.item_schema, item.value)
+                    if not data_valid then
+                        log.error("failed to check item data of [", self.key,
+                                  "] err:", err, " ,val: ", json.encode(item.value))
+                    end
+                end
+
+                if data_valid then
+                    changed = true
+                    insert_tab(self.values, item)
+                    self.values_hash[key] = #self.values
+
+                    item.value.id = key
+                    item.clean_handlers = {}
+
+                    if self.filter then
+                        self.filter(item)
+                    end
+                end
+
+                self:upgrade_version(item.modifiedIndex)
+            end
         end
 
         if headers then
@@ -285,9 +318,16 @@ local function sync_data(self)
     end
 
     local res_copy = res
+    -- waitdir will return [res] even for self.single_item = true
     for _, res in ipairs(res_copy) do
-        local key = short_key(self, res.key)
-        if res.value and type(res.value) ~= "table" then
+        local key
+        if self.single_item then
+            key = self.key
+        else
+            key = short_key(self, res.key)
+        end
+
+        if res.value and not self.single_item and type(res.value) ~= "table" then
             self:upgrade_version(res.modifiedIndex)
             return false, "invalid item data of [" .. self.key .. "/" .. key
                             .. "], val: " .. res.value
@@ -314,10 +354,6 @@ local function sync_data(self)
             return false
         end
 
-        if self.filter then
-            self.filter(res)
-        end
-
         local pre_index = self.values_hash[key]
         if pre_index then
             local pre_val = self.values[pre_index]
@@ -329,7 +365,10 @@ local function sync_data(self)
             end
 
             if res.value then
-                res.value.id = key
+                if not self.single_item then
+                    res.value.id = key
+                end
+
                 self.values[pre_index] = res
                 res.clean_handlers = {}
                 log.info("update data by key: ", key)
@@ -345,7 +384,10 @@ local function sync_data(self)
             res.clean_handlers = {}
             insert_tab(self.values, res)
             self.values_hash[key] = #self.values
-            res.value.id = key
+            if not self.single_item then
+                res.value.id = key
+            end
+
             log.info("insert data by key: ", key)
         end
 
@@ -372,6 +414,12 @@ local function sync_data(self)
             self.sync_times = 0
         end
 
+        -- /plugins' filter need to known self.values when it is called
+        -- so the filter should be called after self.values set.
+        if self.filter then
+            self.filter(res)
+        end
+
         self.conf_version = self.conf_version + 1
     end
 
@@ -476,6 +524,7 @@ function _M.new(key, opts)
     local item_schema = opts and opts.item_schema
     local filter_fun = opts and opts.filter
     local timeout = opts and opts.timeout
+    local single_item = opts and opts.single_item
 
     local obj = setmetatable({
         etcd_cli = nil,
@@ -493,6 +542,7 @@ function _M.new(key, opts)
         last_err = nil,
         last_err_time = nil,
         timeout = timeout,
+        single_item = single_item,
         filter = filter_fun,
     }, mt)