[incubator-apisix] branch master updated: bugfix: reloaded the etcd configuation data if the index (#871)

[we...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

wenming pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-apisix.git


The following commit(s) were added to refs/heads/master by this push:
     new 863aef7  bugfix: reloaded the etcd configuation data if the index (#871)
863aef7 is described below

commit 863aef7510cb64e222eb98d269d03a2593caa53a
Author: YuanSheng Wang <me...@gmail.com>
AuthorDate: Fri Nov 29 15:46:35 2019 +0800

    bugfix: reloaded the etcd configuation data if the index (#871)
---
 lua/apisix/core/config_etcd.lua | 89 ++++++++++++++++++++++++++++-------------
 1 file changed, 61 insertions(+), 28 deletions(-)

diff --git a/lua/apisix/core/config_etcd.lua b/lua/apisix/core/config_etcd.lua
index d3eb909..26e4c73 100644
--- a/lua/apisix/core/config_etcd.lua
+++ b/lua/apisix/core/config_etcd.lua
@@ -54,23 +54,17 @@ local function readdir(etcd_cli, key)
         return nil, nil, "not inited"
     end
 
-    local data, err = etcd_cli:readdir(key, true)
-    if not data then
+    local res, err = etcd_cli:readdir(key, true)
+    if not res then
         -- log.error("failed to get key from etcd: ", err)
         return nil, nil, err
     end
 
-    local body = data.body
-
-    if type(body) ~= "table" then
-        return nil, nil, "failed to read etcd dir"
-    end
-
-    if body.message then
-        return nil, nil, body.message
+    if type(res.body) ~= "table" then
+        return nil, "failed to read etcd dir"
     end
 
-    return body.node, data.headers
+    return res
 end
 
 local function waitdir(etcd_cli, key, modified_index)
@@ -78,19 +72,17 @@ local function waitdir(etcd_cli, key, modified_index)
         return nil, nil, "not inited"
     end
 
-    local data, err = etcd_cli:waitdir(key, modified_index)
-    if not data then
+    local res, err = etcd_cli:waitdir(key, modified_index)
+    if not res then
         -- log.error("failed to get key from etcd: ", err)
-        return nil, nil, err
+        return nil, err
     end
 
-    local body = data.body or {}
-
-    if body.message then
-        return body.node, data.headers, body.message
+    if type(res.body) ~= "table" then
+        return nil, "failed to read etcd dir"
     end
 
-    return body.node, data.headers
+    return res
 end
 
 
@@ -125,8 +117,13 @@ local function sync_data(self)
         return nil, "missing 'key' arguments"
     end
 
-    if self.values == nil then
-        local dir_res, headers, err = readdir(self.etcd_cli, self.key)
+    if self.need_reload then
+        local res, err = readdir(self.etcd_cli, self.key)
+        if not res then
+            return false, err
+        end
+
+        local dir_res, headers = res.body.node, res.headers
         log.debug("readdir key: ", self.key, " res: ",
                   json.delay_encode(dir_res))
         if not dir_res then
@@ -141,6 +138,20 @@ local function sync_data(self)
             dir_res.nodes = {}
         end
 
+        if self.values then
+            for i, val in ipairs(self.values) do
+                if val and val.clean_handlers then
+                    for _, clean_handler in ipairs(val.clean_handlers) do
+                        clean_handler(val)
+                    end
+                    val.clean_handlers = nil
+                end
+            end
+
+            self.values = nil
+            self.values_hash = nil
+        end
+
         self.values = new_tab(#dir_res.nodes, 0)
         self.values_hash = new_tab(0, #dir_res.nodes)
 
@@ -185,18 +196,39 @@ local function sync_data(self)
         if changed then
             self.conf_version = self.conf_version + 1
         end
+
+        self.need_reload = false
         return true
     end
 
-    local res, headers, err = waitdir(self.etcd_cli, self.key,
-                                      self.prev_index + 1)
-    log.debug("waitdir key: ", self.key, " prev_index: ", self.prev_index + 1)
-    log.debug("res: ", json.delay_encode(res, true))
-    log.debug("headers: ", json.delay_encode(headers, true))
+    local dir_res, err = waitdir(self.etcd_cli, self.key, self.prev_index + 1)
+    log.info("waitdir key: ", self.key, " prev_index: ", self.prev_index + 1)
+    log.info("res: ", json.delay_encode(dir_res, true))
+    if not dir_res then
+        return false, err
+    end
+
+    local res = dir_res.body.node
+    local err_msg = dir_res.body.message
+    if err_msg then
+        if err_msg == "The event in requested index is outdated and cleared"
+           and dir_res.body.errorCode == 401 then
+            self.need_reload = true
+            log.warn("waitdir [", self.key, "] err: ", err_msg,
+                     ", need to fully reload")
+            return false
+        end
+        return false, err
+    end
+
     if not res then
-        if headers then
-          self:upgrade_version(headers["X-Etcd-Index"])
+        if err == "The event in requested index is outdated and cleared" then
+            self.need_reload = true
+            log.warn("waitdir [", self.key, "] err: ", err,
+                     ", need to fully reload")
+            return false
         end
+
         return false, err
     end
 
@@ -369,6 +401,7 @@ function _M.new(key, opts)
         running = true,
         conf_version = 0,
         values = nil,
+        need_reload = true,
         routes_hash = nil,
         prev_index = nil,
         last_err = nil,