feat: passive round-robin health check (#148)

shuaijinchao · tzssangglass · web-flow · commit 9995f38e4da9 · 2021-10-10T12:14:25.000+08:00
Co-authored-by: tzssangglass &lt;tzssangglass@gmail.com&gt;
diff --git a/health_check.md b/health_check.md
@@ -32,12 +32,14 @@ Get the current status of the target.
 
 | name         | Type    | Requirement | Default | Description                                                  |
 | ------------ | ------- | ----------- | ------- | ------------------------------------------------------------ |
-| shm_name     | string  | required    |         | the declarative `lua_shared_dict` is used to store the health status of endpoints. |
+| shm_name     | string  | optional    |         | the declarative `lua_shared_dict` is used to store the health status of endpoints, if this option is not set, the health check will return to [round-robin](#round-robin-based-health-check) check mode. |
 | fail_timeout | integer | optional    | 10s     | sets the time during which the specified number of unsuccessful attempts to communicate with the endpoint should happen to marker the endpoint unavailable, and also sets the period of time the endpoint will be marked unavailable. |
-| max_fails    | integer | optional    | 1       | sets the number of failed attempts that must occur during the `fail_timeout` period for the endpoint to be marked unavailable. |
+| max_fails    | integer | optional    | 1       | sets the number of failed attempts that must occur during the `fail_timeout` period for the endpoint to be marked unavailable. This configuration only takes effect in [policy](#policy-based-health-check) check mode |
 | retry        | bool    | optional    | false   | automatically retry another endpoint when operations failed. |
 
-lua example:
+## Example
+
+### Policy based health check
 
 ```lua
 local health_check, err = require("resty.etcd.health_check").init({
@@ -54,19 +56,36 @@ Health check mechanism would switch endpoint only when the previously choosed en
 
 The failure counter and health status of each etcd endpoint are shared across workers and by different etcd clients.
 
-Also note that the `fail_timeout`, `max_fails` and `retry` of the health check cannot be changed once it has been created.
+PS: the `fail_timeout`, `max_fails` and `retry` of the health check After initialization, they will only be reset when the health check mode is switched.
+
+### Round-robin based health check
+
+```lua
+local health_check, err = require("resty.etcd.health_check").init({
+    fail_timeout = 10,
+    retry = false,
+})
+```
+
+Round-robin health check. When a endpoint fails, the endpoint will be marked as unhealthy, and will not be connected to the endpoint within the time set by `fail_timeout` (select the next healthy endpoint to connect).
+
+Unhealthy nodes will be released to the selection pool of healthy endpoints after the `fail_timeout` time is exceeded.
+
+The status of etcd unhealthy endpoints is only valid in the current worker
 
 ##  Synopsis
 
 ```nginx
 http {
     # required declares a shared memory zone to store endpoints's health status
+    # if you use the round-robin method for health check, you don’t need to set this
     lua_shared_dict healthcheck_shm 1m;
 
     server {
         location = /healthcheck {
             content_by_lua_block {
                 # the health check feature is optional, and can be enabled with the following configuration.
+                # if you use the round-robin method for health check, you don’t need to set this
                 local health_check, err = require("resty.etcd.health_check").init({
                     shm_name = "healthcheck_shm",
                     fail_timeout = 10,
diff --git a/lib/resty/etcd/health_check.lua b/lib/resty/etcd/health_check.lua
@@ -1,31 +1,53 @@
 local ngx_shared    = ngx.shared
 local utils         = require("resty.etcd.utils")
+local type          = type
+local now           = os.time
 local conf
 
 local _M = {}
 
+local round_robin_unhealthy_target_hosts
+
+
 local function gen_unhealthy_key(etcd_host)
     return "unhealthy-" .. etcd_host
 end
 
 local function get_target_status(etcd_host)
-    if conf == nil then
-        return
+    if not conf then
+        return nil, "etcd health check uninitialized"
     end
 
-    local unhealthy_key = gen_unhealthy_key(etcd_host)
-    local unhealthy_endpoint, err = ngx_shared[conf.shm_name]:get(unhealthy_key)
-    if err then
-        utils.log_warn("failed to get unhealthy_key: ",
-                unhealthy_key, " err: ", err)
-        return
+    if type(etcd_host) ~= "string" then
+        return false, "etcd host invalid"
     end
 
-    if not unhealthy_endpoint then
-        return true
-    end
+    local unhealthy_key = gen_unhealthy_key(etcd_host)
+    if conf.shm_name ~= nil then
+        local unhealthy_endpoint, err = ngx_shared[conf.shm_name]:get(unhealthy_key)
+        if err then
+            utils.log_warn("failed to get unhealthy_key: ",
+                    unhealthy_key, " err: ", err)
+            return
+        end
 
-    return false
+        if not unhealthy_endpoint then
+            return true
+        end
+
+        return false
+    else
+        if type(round_robin_unhealthy_target_hosts) ~= "table" then
+            round_robin_unhealthy_target_hosts = {}
+        end
+
+        local target_fail_expired_time = round_robin_unhealthy_target_hosts[unhealthy_key]
+        if target_fail_expired_time and target_fail_expired_time >= now() then
+            return false, "endpoint: " .. etcd_host .. " is unhealthy"
+        else
+            return true
+        end
+    end
 end
 _M.get_target_status = get_target_status
 
@@ -44,37 +66,56 @@ end
 
 
 local function report_failure(etcd_host)
-    if conf == nil then
-        return
+    if not conf then
+        return nil, "etcd health check uninitialized"
     end
 
-    local fails, err = fault_count(etcd_host, conf.shm_name, conf.fail_timeout)
-    if err then
-        utils.log_error("failed to incr etcd endpoint fail times: ", err)
-        return
+    if type(etcd_host) ~= "string" then
+        return nil, "etcd host invalid"
     end
 
-    if fails >= conf.max_fails then
-        local unhealthy_key = gen_unhealthy_key(etcd_host)
-        local unhealthy_endpoint, _ = ngx_shared[conf.shm_name]:get(unhealthy_key)
-        if unhealthy_endpoint == nil then
-            ngx_shared[conf.shm_name]:set(unhealthy_key, etcd_host,
-                    conf.fail_timeout)
-            utils.log_warn("update endpoint: ", etcd_host, " to unhealthy")
+    if conf.shm_name ~= nil then
+        local fails, err = fault_count(etcd_host, conf.shm_name, conf.fail_timeout)
+        if err then
+            utils.log_error("failed to incr etcd endpoint fail times: ", err)
+            return nil, err
         end
+
+        if fails >= conf.max_fails then
+            local unhealthy_key = gen_unhealthy_key(etcd_host)
+            local unhealthy_endpoint, _ = ngx_shared[conf.shm_name]:get(unhealthy_key)
+            if unhealthy_endpoint == nil then
+                ngx_shared[conf.shm_name]:set(unhealthy_key, etcd_host,
+                        conf.fail_timeout)
+                utils.log_warn("update endpoint: ", etcd_host, " to unhealthy")
+            end
+        end
+    else
+        if type(round_robin_unhealthy_target_hosts) ~= "table" then
+            round_robin_unhealthy_target_hosts = {}
+        end
+        local unhealthy_key = gen_unhealthy_key(etcd_host)
+        round_robin_unhealthy_target_hosts[unhealthy_key] = now() + conf.fail_timeout
+        utils.log_warn("update endpoint: ", etcd_host, " to unhealthy")
     end
 end
 _M.report_failure = report_failure
 
 
 function _M.init(opts)
-    if conf == nil then
+    opts = opts or {}
+    if not conf or opts.shm_name ~= conf.shm_name then
         conf = {}
-        local shared_dict = ngx_shared[opts.shm_name]
-        if not shared_dict then
-            return nil, "failed to get ngx.shared dict: " .. opts.shm_name
+        if opts.shm_name and type(opts.shm_name) == "string" then
+            local shared_dict = ngx_shared[opts.shm_name]
+            if not shared_dict then
+                return nil, "failed to get ngx.shared dict: " .. opts.shm_name
+            end
+            conf.shm_name = opts.shm_name
+            utils.log_info("healthy check use ngx.shared dict: ", opts.shm_name)
+        else
+            utils.log_info("healthy check use round robin")
         end
-        conf.shm_name = opts.shm_name
         conf.fail_timeout = opts.fail_timeout or 10    -- 10 sec
         conf.max_fails = opts.max_fails or 1
         conf.retry = opts.retry or false
diff --git a/lib/resty/etcd/v3.lua b/lib/resty/etcd/v3.lua
@@ -23,7 +23,6 @@ local encode_json   = cjson.encode
 local encode_base64 = ngx.encode_base64
 local decode_base64 = ngx.decode_base64
 local semaphore     = require("ngx.semaphore")
-local INIT_COUNT_RESIZE = 2e8
 local health_check  = require("resty.etcd.health_check")
 
 local _M = {}
@@ -45,24 +44,14 @@ local refresh_jwt_token
 
 local function choose_endpoint(self)
     local endpoints = self.endpoints
-    local endpoints_len = #endpoints
 
-    if health_check.conf ~= nil then
-        for _, endpoint in ipairs(endpoints) do
-            if health_check.get_target_status(endpoint.http_host) then
-                return endpoint
-            end
+    for _, endpoint in ipairs(endpoints) do
+        if health_check.get_target_status(endpoint.http_host) then
+            return endpoint
         end
-        return nil, "has no healthy etcd endpoint available"
-    end
-
-    self.init_count = (self.init_count or -1) + 1
-    local pos = self.init_count % endpoints_len + 1
-    if self.init_count >= INIT_COUNT_RESIZE then
-        self.init_count = 0
     end
 
-    return endpoints[pos]
+    return nil, "has no healthy etcd endpoint available"
 end
 
 
@@ -92,17 +81,12 @@ local function http_request_uri(self, http_cli, method, uri, body, headers, keep
     })
 
     if err then
-        if health_check.conf ~= nil then
-            health_check.report_failure(endpoint.http_host)
-            err = endpoint.http_host .. ": " .. err
-        end
-        return nil, err
+        health_check.report_failure(endpoint.http_host)
+        return nil, endpoint.http_host .. ": " .. err
     end
 
     if res.status >= 500 then
-        if health_check.conf ~= nil then
-            health_check.report_failure(endpoint.http_host)
-        end
+        health_check.report_failure(endpoint.http_host)
         return nil, "invalid response code: " .. res.status
     end
 
@@ -156,12 +140,7 @@ local function _request_uri(self, method, uri, opts, timeout, ignore_auth)
     end
 
     local res
-    if health_check.conf == nil or not health_check.conf.retry then
-        res, err = http_request_uri(self, http_cli, method, uri, body, headers, keepalive)
-        if err then
-            return nil, err
-        end
-    else
+    if health_check.conf.retry then
         local max_retry = #self.endpoints * health_check.conf.max_fails + 1
         for _ = 1, max_retry do
             res, err = http_request_uri(self, http_cli, method, uri, body, headers, keepalive)
@@ -174,6 +153,11 @@ local function _request_uri(self, method, uri, opts, timeout, ignore_auth)
                 break
             end
         end
+    else
+        res, err = http_request_uri(self, http_cli, method, uri, body, headers, keepalive)
+        if err then
+            return nil, err
+        end
     end
 
     if not typeof.string(res.body) then
@@ -269,6 +253,10 @@ function _M.new(opts)
         return nil, err
     end
 
+    if health_check.conf == nil then
+        health_check.init()
+    end
+
     return setmetatable({
             last_auth_time = now(), -- save last Authentication time
             last_refresh_jwt_err = nil,
@@ -596,9 +584,7 @@ local function http_request_chunk(self, http_cli)
         ssl_server_name = self.sni,
     })
     if not ok then
-        if health_check.conf ~= nil then
-            health_check.report_failure(endpoint.http_host)
-        end
+        health_check.report_failure(endpoint.http_host)
         return nil, endpoint.http_host .. ": " .. err
     end
 
@@ -653,12 +639,7 @@ local function request_chunk(self, method, path, opts, timeout)
     end
 
     local endpoint
-    if health_check.conf == nil or not health_check.conf.retry then
-        endpoint, err = http_request_chunk(self, http_cli)
-        if err then
-            return nil, err
-        end
-    else
+    if health_check.conf.retry then
         local max_retry = #self.endpoints * health_check.conf.max_fails + 1
         for _ = 1, max_retry do
             endpoint, err = http_request_chunk(self, http_cli)
@@ -671,6 +652,11 @@ local function request_chunk(self, method, path, opts, timeout)
                 break
             end
         end
+    else
+        endpoint, err = http_request_chunk(self, http_cli)
+        if err then
+            return nil, err
+        end
     end
 
     local res
@@ -710,11 +696,9 @@ local function request_chunk(self, method, path, opts, timeout)
         if not body then
             return nil, "failed to decode json body: " .. (err or " unkwon")
         elseif body.error and body.error.http_code >= 500 then
-            if health_check.conf ~= nil then
-                -- health_check retry should do nothing here
-                -- and let connection closed to create a new one
-                health_check.report_failure(endpoint.http_host)
-            end
+            -- health_check retry should do nothing here
+            -- and let connection closed to create a new one
+            health_check.report_failure(endpoint.http_host)
             return nil, endpoint.http_host .. ": " .. body.error.http_status
         end
 
diff --git a/t/v3/auth.t b/t/v3/auth.t
@@ -200,17 +200,17 @@ GET /t
 --- response_body
 ok
 --- grep_error_log eval
-qr/(uri: .+, timeout: \d+|v3 refresh jwt last err: [^,]+|connection refused)/
+qr/(uri: .+, timeout: \d+|has no healthy [^,]+)/
 --- grep_error_log_out
 uri: /kv/put, timeout: 3
 uri: /auth/authenticate, timeout: 3
+has no healthy etcd endpoint available
 uri: /kv/put, timeout: 3
+uri: /auth/authenticate, timeout: 3
+has no healthy etcd endpoint available
 uri: /kv/put, timeout: 3
-connection refused
-v3 refresh jwt last err: connection refused
-connection refused
-v3 refresh jwt last err: connection refused
-connection refused
+uri: /auth/authenticate, timeout: 3
+has no healthy etcd endpoint available
 
 
 
diff --git a/t/v3/health_check.t b/t/v3/health_check.t
diff --git a/t/v3/mtls.t b/t/v3/mtls.t
