Merge pull request #7 from Revolyssup/revolyssup/sync-1.05

sync with resty-redis-cluster version 1.05

Author: Revolyssup

lib/resty/rediscluster.lua

@@ -20,6 +20,7 @@ local string_find = string.find
 local redis_crc = xmodem.redis_crc
 
 local DEFAULT_SHARED_DICT_NAME = "redis_cluster_slot_locks"
+local DEFAULT_REFRESH_DICT_NAME = "refresh_lock"
 local DEFAULT_MAX_REDIRECTION = 5
 local DEFAULT_MAX_CONNECTION_ATTEMPTS = 3
 local DEFAULT_KEEPALIVE_TIMEOUT = 55000
@@ -97,22 +98,32 @@ local function split(s, delimiter)
 end
 
 local function try_hosts_slots(self, serv_list)
+    local start_time = ngx.now()
     local errors = {}
     local config = self.config
     if #serv_list < 1 then
         return nil, "failed to fetch slots, serv_list config is empty"
     end
+
     for i = 1, #serv_list do
         local ip = serv_list[i].ip
         local port = serv_list[i].port
         local redis_client = redis:new()
-        local ok, err
+        local ok, err, max_connection_timeout_err
         redis_client:set_timeouts(config.connect_timeout or DEFAULT_CONNECTION_TIMEOUT,
                                   config.send_timeout or DEFAULT_SEND_TIMEOUT,
                                   config.read_timeout or DEFAULT_READ_TIMEOUT)
 
         --attempt to connect DEFAULT_MAX_CONNECTION_ATTEMPTS times to redis
         for k = 1, config.max_connection_attempts or DEFAULT_MAX_CONNECTION_ATTEMPTS do
+            local total_connection_time_ms = (ngx.now() - start_time) * 1000
+            if (config.max_connection_timeout and total_connection_time_ms > config.max_connection_timeout) then
+                max_connection_timeout_err = "max_connection_timeout of " .. config.max_connection_timeout .. "ms reached."
+                ngx.log(ngx.ERR, max_connection_timeout_err)
+                table_insert(errors, max_connection_timeout_err)
+                break
+            end
+
             ok, err = redis_client:connect(ip, port, self.config.connect_opts)
             if ok then break end
             if err then
@@ -180,12 +191,14 @@ local function try_hosts_slots(self, serv_list)
                 table_insert(errors, nerr)
             end
             release_connection(redis_client, config)
-            
+
             -- refresh of slots and master nodes successful
             -- not required to connect/iterate over additional hosts
             if nodes_res and slots_info then
                 return true, nil
             end
+        elseif max_connection_timeout_err then
+            break
         else
             table_insert(errors, err)
         end
@@ -201,12 +214,19 @@ function _M.fetch_slots(self)
     local serv_list = self.config.serv_list
     local serv_list_cached = slot_cache[self.config.name .. "serv_list"]
 
-    local serv_list_combined = {}
+    local serv_list_combined
 
-    -- if a cached serv_list is present, use it
+    -- if a cached serv_list is present, start with that
     if serv_list_cached then
         serv_list_combined = serv_list_cached.serv_list
+
+        -- then append the serv_list from config, in the event that the entire
+        -- cached serv_list no longer points to anything usable
+        for _, s in ipairs(serv_list) do
+            table_insert(serv_list_combined, s)
+        end
     else
+        -- otherwise we bootstrap with our serv_list from config
         serv_list_combined = serv_list
     end
 
@@ -221,6 +241,30 @@ function _M.fetch_slots(self)
 end
 
 
+function _M.refresh_slots(self)
+    local worker_id = ngx.worker.id()
+    local lock, err, elapsed, ok
+    lock, err = resty_lock:new(self.config.dict_name or DEFAULT_SHARED_DICT_NAME, {time_out = 0})
+    if not lock then
+        ngx.log(ngx.ERR, "failed to create lock in refresh slot cache: ", err)
+        return nil, err
+    end
+
+    local refresh_lock_key = (self.config.refresh_lock_key or DEFAULT_REFRESH_DICT_NAME) .. worker_id
+    elapsed, err = lock:lock(refresh_lock_key)
+    if not elapsed then
+        return nil, 'race refresh lock fail, ' .. err
+    end
+
+    self:fetch_slots()
+    ok, err = lock:unlock()
+    if not ok then
+        ngx.log(ngx.ERR, "failed to unlock in refresh slot cache:", err)
+        return nil, err
+    end
+end
+
+
 function _M.init_slots(self)
     if slot_cache[self.config.name] then
         -- already initialized
@@ -396,7 +440,7 @@ local function handle_command_with_retry(self, target_ip, target_port, asking, c
             ip, port, slave, err = pick_node(self, serv_list, slot)
             if err then
                 ngx.log(ngx.ERR, "pickup node failed, will return failed for this request, meanwhile refereshing slotcache " .. err)
-                self:fetch_slots()
+                self:refresh_slots()
                 return nil, err
             end
         end
@@ -416,7 +460,7 @@ local function handle_command_with_retry(self, target_ip, target_port, asking, c
                 --set readonly
                 ok, err = redis_client:readonly()
                 if not ok then
-                    self:fetch_slots()
+                    self:refresh_slots()
                     return nil, err
                 end
             end
@@ -425,7 +469,7 @@ local function handle_command_with_retry(self, target_ip, target_port, asking, c
                 --executing asking
                 ok, err = redis_client:asking()
                 if not ok then
-                    self:fetch_slots()
+                    self:refresh_slots()
                     return nil, err
                 end
             end
@@ -445,7 +489,7 @@ local function handle_command_with_retry(self, target_ip, target_port, asking, c
                     release_connection(redis_client, config)
                     target_ip = nil
                     target_port = nil
-                    self:fetch_slots()
+                    self:refresh_slots()
                     need_to_retry = true
 
                 elseif string.sub(err, 1, 3) == "ASK" then
@@ -468,7 +512,7 @@ local function handle_command_with_retry(self, target_ip, target_port, asking, c
                     return nil, "Cannot executing command, cluster status is failed!"
                 else
                     --There might be node fail, we should also refresh slot cache
-                    self:fetch_slots()
+                    self:refresh_slots()
                     return nil, err
                 end
             end
@@ -478,7 +522,7 @@ local function handle_command_with_retry(self, target_ip, target_port, asking, c
             end
         else
             --There might be node fail, we should also refresh slot cache
-            self:fetch_slots()
+            self:refresh_slots()
             if k == config.max_redirection or k == DEFAULT_MAX_REDIRECTION then
                 -- only return after allowing for `k` attempts
                 return nil, connerr
@@ -561,7 +605,7 @@ local function construct_final_pipeline_resp(self, node_res_map, node_req_map)
                 --ngx.log(ngx.NOTICE, "handle moved signal for cmd:" .. reqs[i]["cmd"] .. " key:" .. reqs[i]["key"])
                 if need_to_fetch_slots then
                     -- if there is multiple signal for moved, we just need to fetch slot cache once, and do retry.
-                    self:fetch_slots()
+                    self:refresh_slots()
                     need_to_fetch_slots = false
                 end
                 local movedres, err = handle_command_with_retry(self, nil, nil, false, reqs[i]["cmd"], reqs[i]["key"], unpack(reqs[i]["args"]))
@@ -634,7 +678,7 @@ function _M.commit_pipeline(self)
             -- We must empty local reference to slots cache, otherwise there will be memory issue while
             -- coroutine swich happens(eg. ngx.sleep, cosocket), very important!
             slots = nil
-            self:fetch_slots()
+            self:refresh_slots()
             return nil, err
         end
 
@@ -671,7 +715,7 @@ function _M.commit_pipeline(self)
             --set readonly
             local ok, err = redis_client:readonly()
             if not ok then
-                self:fetch_slots()
+                self:refresh_slots()
                 return nil, err
             end
         end
@@ -692,7 +736,7 @@ function _M.commit_pipeline(self)
             local res, err = redis_client:commit_pipeline()
             if err then
                 --There might be node fail, we should also refresh slot cache
-                self:fetch_slots()
+                self:refresh_slots()
                 return nil, err .. " return from " .. tostring(ip) .. ":" .. tostring(port)
             end
 
@@ -703,7 +747,7 @@ function _M.commit_pipeline(self)
             node_res_map[k] = res
         else
             --There might be node fail, we should also refresh slot cache
-            self:fetch_slots()
+            self:refresh_slots()
             return nil, err .. "pipeline commit failed while connecting to " .. tostring(ip) .. ":" .. tostring(port)
         end
     end