Re: [PR] refactor: extract reusable building blocks from Consul discovery [apisix]
Baoyuantop merged PR #13230: URL: https://github.com/apache/apisix/pull/13230 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] refactor: extract reusable building blocks from Consul discovery [apisix]
nic-6443 closed pull request #13230: refactor: extract reusable building blocks from Consul discovery URL: https://github.com/apache/apisix/pull/13230 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] refactor: extract reusable building blocks from Consul discovery [apisix]
nic-6443 commented on code in PR #13230:
URL: https://github.com/apache/apisix/pull/13230#discussion_r3090726995
##
apisix/discovery/consul/init.lua:
##
@@ -252,496 +223,410 @@ local function show_dump_file()
end
-local function get_retry_delay(retry_delay)
-if not retry_delay or retry_delay >= max_retry_time then
-retry_delay = 1
-else
-retry_delay = retry_delay * 4
+-- ─── polling loop ─
+
+local function check_keepalive(reg, consul_server, retry_delay)
+if reg.stop_flag then
+return
end
-return retry_delay
+if consul_server.keepalive and not exiting() then
+local ok, err = ngx_timer_at(0, _M.connect, reg, consul_server,
retry_delay)
+if not ok then
+log.error("create ngx_timer_at got error: ", err)
+return
+end
+end
end
-local function get_opts(consul_server, is_catalog)
-local opts = {
-host = consul_server.host,
-port = consul_server.port,
-connect_timeout = consul_server.connect_timeout,
-read_timeout = consul_server.read_timeout,
-default_args = {
-token = consul_server.token,
-}
-}
-if not consul_server.keepalive then
-return opts
+function _M.connect(premature, reg, consul_server, retry_delay)
+if premature or reg.stop_flag then
+return
end
-opts.default_args.wait = consul_server.wait_timeout --blocked wait!=0;
unblocked by wait=0
+local catalog_thread, spawn_catalog_err =
thread_spawn(consul_client.watch_catalog,
+ consul_server)
+if not catalog_thread then
+local random_delay = math_random(default_random_range)
+log.error("failed to spawn thread watch catalog: ", spawn_catalog_err,
+", retry connecting consul after ", random_delay, " seconds")
+core_sleep(random_delay)
-if is_catalog then
-opts.default_args.index = consul_server.catalog_index
-else
-opts.default_args.index = consul_server.health_index
+check_keepalive(reg, consul_server, retry_delay)
+return
end
-return opts
-end
-
-
-local function watch_catalog(consul_server)
-local client = resty_consul:new(get_opts(consul_server, true))
-
-::RETRY::
-local watch_result, watch_err =
client:get(consul_server.consul_watch_catalog_url)
-local watch_error_info = (watch_err ~= nil and watch_err)
- or ((watch_result ~= nil and watch_result.status
~= 200)
- and watch_result.status)
-if watch_error_info then
-log.error("connect consul: ", consul_server.consul_server_url,
-" by sub url: ", consul_server.consul_watch_catalog_url,
-", got watch result: ", json_delay_encode(watch_result),
-", with error: ", watch_error_info)
+local health_thread, err = thread_spawn(consul_client.watch_health,
consul_server)
+if not health_thread then
+thread_kill(catalog_thread)
+local random_delay = math_random(default_random_range)
+log.error("failed to spawn thread watch health: ", err,
+", retry connecting consul after ", random_delay, " seconds")
+core_sleep(random_delay)
-return watch_type_catalog, default_catalog_error_index
+check_keepalive(reg, consul_server, retry_delay)
+return
end
-if consul_server.catalog_index > 0
-and consul_server.catalog_index ==
tonumber(watch_result.headers['X-Consul-Index']) then
+local thread_wait_ok, watch_type, index = thread_wait(catalog_thread,
health_thread)
+thread_kill(catalog_thread)
+thread_kill(health_thread)
+if not thread_wait_ok then
local random_delay = math_random(default_random_range)
-log.info("watch catalog has no change, re-watch consul after ",
random_delay, " seconds")
+log.error("failed to wait thread: ", watch_type, ", retry connecting
consul after ",
+random_delay, " seconds")
core_sleep(random_delay)
-goto RETRY
+
+check_keepalive(reg, consul_server, retry_delay)
+return
end
-return watch_type_catalog, watch_result.headers['X-Consul-Index']
-end
+if not consul_client.watch_result_is_valid(tonumber(watch_type),
+tonumber(index), consul_server.catalog_index,
consul_server.health_index) then
+retry_delay = consul_client.get_retry_delay(retry_delay)
+log.warn("get all svcs got err, retry connecting consul after ",
retry_delay, " seconds")
+core_sleep(retry_delay)
+check_keepalive(reg, consul_server, retry_delay)
+return
+end
-local function watch_health(consul_server)
-local client = resty_consul:new(get_opts(consul_server, false))
+if reg.stop_flag then
+return
+end
Re: [PR] refactor: extract reusable building blocks from Consul discovery [apisix]
nic-6443 commented on code in PR #13230:
URL: https://github.com/apache/apisix/pull/13230#discussion_r3090726507
##
apisix/discovery/consul/client.lua:
##
@@ -0,0 +1,479 @@
+--
+-- Licensed to the Apache Software Foundation (ASF) under one or more
+-- contributor license agreements. See the NOTICE file distributed with
+-- this work for additional information regarding copyright ownership.
+-- The ASF licenses this file to You under the Apache License, Version 2.0
+-- (the "License"); you may not use this file except in compliance with
+-- the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+--
+
+--- Reusable HTTP client primitives for Consul service discovery.
+--- Extracted from init.lua so that both static-config mode and
+--- dynamic-config mode can share the same core logic.
+
+local require= require
+local core = require("apisix.core")
+local core_sleep = require("apisix.core.utils").sleep
+local resty_consul = require('resty.consul')
+local http = require('resty.http')
+local ipairs = ipairs
+local pairs = pairs
+local unpack = unpack
+local tonumber = tonumber
+local type = type
+local next = next
+local ngx= ngx
+local math_random= math.random
+local log= core.log
+local json_delay_encode = core.json.delay_encode
+local pcall = pcall
+local null = ngx.null
+
+local _M = {}
+
+local default_random_range = 5
+local default_catalog_error_index = -1
+local default_health_error_index = -2
+local watch_type_catalog = 1
+local watch_type_health = 2
+local max_retry_time = 256
+
+
+-- ─── helpers ──
+
+function _M.get_retry_delay(retry_delay)
+if not retry_delay or retry_delay >= max_retry_time then
+retry_delay = 1
+else
+retry_delay = retry_delay * 4
+end
+
+return retry_delay
+end
+
+
+local function is_not_empty(value)
+if value == nil or value == null
+or (type(value) == "table" and not next(value))
+or (type(value) == "string" and value == "")
+then
+return false
+end
+
+return true
+end
+
+
+-- ─── sort comparators ─
+
+local function combine_sort_nodes_cmp(left, right)
+if left.host ~= right.host then
+return left.host < right.host
+end
+
+return left.port < right.port
+end
+
+
+local function port_sort_nodes_cmp(left, right)
+return left.port < right.port
+end
+
+
+local function host_sort_nodes_cmp(left, right)
+return left.host < right.host
+end
+
+
+function _M.sort_nodes(nodes, sort_type)
+if not nodes or not sort_type or sort_type == "origin" then
+return
+end
+
+if sort_type == "port_sort" then
+core.table.sort(nodes, port_sort_nodes_cmp)
+elseif sort_type == "host_sort" then
+core.table.sort(nodes, host_sort_nodes_cmp)
+elseif sort_type == "combine_sort" then
+core.table.sort(nodes, combine_sort_nodes_cmp)
+end
+end
+
+
+-- ─── resty.consul options ─
+
+local function get_opts(consul_server, is_catalog)
+local opts = {
+host = consul_server.host,
+port = consul_server.port,
+connect_timeout = consul_server.connect_timeout,
+read_timeout = consul_server.read_timeout,
+default_args = {
+token = consul_server.token,
+}
+}
+if not consul_server.keepalive then
+return opts
+end
+
+opts.default_args.wait = consul_server.wait_timeout
+
+if is_catalog then
+opts.default_args.index = consul_server.catalog_index
+else
+opts.default_args.index = consul_server.health_index
+end
+
+return opts
+end
+
+
+-- ─── blocking query watchers ──
+
+function _M.watch_catalog(consul_server)
+local client = resty_consul:new(get_opts(consul_server, true))
+
+::RETRY::
+local watch_result, watch_err =
client:get(consul_server.consul_watch_catalog_url)
+local watch_error_info = (watch_err ~= nil and watch_err)
+ or ((watch_result ~= nil and watch_result.status
~= 200)
+ and watch_result.status)
+if watch_error_info then
+log.error("connect consul: ", consul_server.consul_server_url,
+" by sub url: ", consul_server.consul_watch_catalog_url,
+", got watc
Re: [PR] refactor: extract reusable building blocks from Consul discovery [apisix]
nic-6443 commented on code in PR #13230:
URL: https://github.com/apache/apisix/pull/13230#discussion_r3090726016
##
apisix/discovery/consul/init.lua:
##
@@ -252,496 +223,410 @@ local function show_dump_file()
end
-local function get_retry_delay(retry_delay)
-if not retry_delay or retry_delay >= max_retry_time then
-retry_delay = 1
-else
-retry_delay = retry_delay * 4
+-- ─── polling loop ─
+
+local function check_keepalive(reg, consul_server, retry_delay)
+if reg.stop_flag then
+return
end
-return retry_delay
+if consul_server.keepalive and not exiting() then
+local ok, err = ngx_timer_at(0, _M.connect, reg, consul_server,
retry_delay)
+if not ok then
+log.error("create ngx_timer_at got error: ", err)
+return
+end
+end
end
-local function get_opts(consul_server, is_catalog)
-local opts = {
-host = consul_server.host,
-port = consul_server.port,
-connect_timeout = consul_server.connect_timeout,
-read_timeout = consul_server.read_timeout,
-default_args = {
-token = consul_server.token,
-}
-}
-if not consul_server.keepalive then
-return opts
+function _M.connect(premature, reg, consul_server, retry_delay)
+if premature or reg.stop_flag then
+return
end
-opts.default_args.wait = consul_server.wait_timeout --blocked wait!=0;
unblocked by wait=0
+local catalog_thread, spawn_catalog_err =
thread_spawn(consul_client.watch_catalog,
+ consul_server)
+if not catalog_thread then
+local random_delay = math_random(default_random_range)
+log.error("failed to spawn thread watch catalog: ", spawn_catalog_err,
+", retry connecting consul after ", random_delay, " seconds")
+core_sleep(random_delay)
-if is_catalog then
-opts.default_args.index = consul_server.catalog_index
-else
-opts.default_args.index = consul_server.health_index
+check_keepalive(reg, consul_server, retry_delay)
+return
end
-return opts
-end
-
-
-local function watch_catalog(consul_server)
-local client = resty_consul:new(get_opts(consul_server, true))
-
-::RETRY::
-local watch_result, watch_err =
client:get(consul_server.consul_watch_catalog_url)
-local watch_error_info = (watch_err ~= nil and watch_err)
- or ((watch_result ~= nil and watch_result.status
~= 200)
- and watch_result.status)
-if watch_error_info then
-log.error("connect consul: ", consul_server.consul_server_url,
-" by sub url: ", consul_server.consul_watch_catalog_url,
-", got watch result: ", json_delay_encode(watch_result),
-", with error: ", watch_error_info)
+local health_thread, err = thread_spawn(consul_client.watch_health,
consul_server)
+if not health_thread then
+thread_kill(catalog_thread)
+local random_delay = math_random(default_random_range)
+log.error("failed to spawn thread watch health: ", err,
+", retry connecting consul after ", random_delay, " seconds")
+core_sleep(random_delay)
-return watch_type_catalog, default_catalog_error_index
+check_keepalive(reg, consul_server, retry_delay)
+return
end
-if consul_server.catalog_index > 0
-and consul_server.catalog_index ==
tonumber(watch_result.headers['X-Consul-Index']) then
+local thread_wait_ok, watch_type, index = thread_wait(catalog_thread,
health_thread)
+thread_kill(catalog_thread)
+thread_kill(health_thread)
+if not thread_wait_ok then
local random_delay = math_random(default_random_range)
-log.info("watch catalog has no change, re-watch consul after ",
random_delay, " seconds")
+log.error("failed to wait thread: ", watch_type, ", retry connecting
consul after ",
+random_delay, " seconds")
core_sleep(random_delay)
-goto RETRY
+
+check_keepalive(reg, consul_server, retry_delay)
+return
end
-return watch_type_catalog, watch_result.headers['X-Consul-Index']
-end
+if not consul_client.watch_result_is_valid(tonumber(watch_type),
+tonumber(index), consul_server.catalog_index,
consul_server.health_index) then
+retry_delay = consul_client.get_retry_delay(retry_delay)
+log.warn("get all svcs got err, retry connecting consul after ",
retry_delay, " seconds")
+core_sleep(retry_delay)
+check_keepalive(reg, consul_server, retry_delay)
+return
+end
-local function watch_health(consul_server)
-local client = resty_consul:new(get_opts(consul_server, false))
+if reg.stop_flag then
+return
+end
Re: [PR] refactor: extract reusable building blocks from Consul discovery [apisix]
Copilot commented on code in PR #13230:
URL: https://github.com/apache/apisix/pull/13230#discussion_r3090717641
##
apisix/discovery/consul/init.lua:
##
@@ -252,496 +223,410 @@ local function show_dump_file()
end
-local function get_retry_delay(retry_delay)
-if not retry_delay or retry_delay >= max_retry_time then
-retry_delay = 1
-else
-retry_delay = retry_delay * 4
+-- ─── polling loop ─
+
+local function check_keepalive(reg, consul_server, retry_delay)
+if reg.stop_flag then
+return
end
-return retry_delay
+if consul_server.keepalive and not exiting() then
+local ok, err = ngx_timer_at(0, _M.connect, reg, consul_server,
retry_delay)
+if not ok then
+log.error("create ngx_timer_at got error: ", err)
+return
+end
+end
end
-local function get_opts(consul_server, is_catalog)
-local opts = {
-host = consul_server.host,
-port = consul_server.port,
-connect_timeout = consul_server.connect_timeout,
-read_timeout = consul_server.read_timeout,
-default_args = {
-token = consul_server.token,
-}
-}
-if not consul_server.keepalive then
-return opts
+function _M.connect(premature, reg, consul_server, retry_delay)
+if premature or reg.stop_flag then
+return
end
-opts.default_args.wait = consul_server.wait_timeout --blocked wait!=0;
unblocked by wait=0
+local catalog_thread, spawn_catalog_err =
thread_spawn(consul_client.watch_catalog,
+ consul_server)
+if not catalog_thread then
+local random_delay = math_random(default_random_range)
+log.error("failed to spawn thread watch catalog: ", spawn_catalog_err,
+", retry connecting consul after ", random_delay, " seconds")
+core_sleep(random_delay)
-if is_catalog then
-opts.default_args.index = consul_server.catalog_index
-else
-opts.default_args.index = consul_server.health_index
+check_keepalive(reg, consul_server, retry_delay)
+return
end
-return opts
-end
-
-
-local function watch_catalog(consul_server)
-local client = resty_consul:new(get_opts(consul_server, true))
-
-::RETRY::
-local watch_result, watch_err =
client:get(consul_server.consul_watch_catalog_url)
-local watch_error_info = (watch_err ~= nil and watch_err)
- or ((watch_result ~= nil and watch_result.status
~= 200)
- and watch_result.status)
-if watch_error_info then
-log.error("connect consul: ", consul_server.consul_server_url,
-" by sub url: ", consul_server.consul_watch_catalog_url,
-", got watch result: ", json_delay_encode(watch_result),
-", with error: ", watch_error_info)
+local health_thread, err = thread_spawn(consul_client.watch_health,
consul_server)
+if not health_thread then
+thread_kill(catalog_thread)
+local random_delay = math_random(default_random_range)
+log.error("failed to spawn thread watch health: ", err,
+", retry connecting consul after ", random_delay, " seconds")
+core_sleep(random_delay)
-return watch_type_catalog, default_catalog_error_index
+check_keepalive(reg, consul_server, retry_delay)
+return
end
-if consul_server.catalog_index > 0
-and consul_server.catalog_index ==
tonumber(watch_result.headers['X-Consul-Index']) then
+local thread_wait_ok, watch_type, index = thread_wait(catalog_thread,
health_thread)
+thread_kill(catalog_thread)
+thread_kill(health_thread)
+if not thread_wait_ok then
local random_delay = math_random(default_random_range)
-log.info("watch catalog has no change, re-watch consul after ",
random_delay, " seconds")
+log.error("failed to wait thread: ", watch_type, ", retry connecting
consul after ",
+random_delay, " seconds")
core_sleep(random_delay)
-goto RETRY
+
+check_keepalive(reg, consul_server, retry_delay)
+return
end
-return watch_type_catalog, watch_result.headers['X-Consul-Index']
-end
+if not consul_client.watch_result_is_valid(tonumber(watch_type),
+tonumber(index), consul_server.catalog_index,
consul_server.health_index) then
+retry_delay = consul_client.get_retry_delay(retry_delay)
+log.warn("get all svcs got err, retry connecting consul after ",
retry_delay, " seconds")
+core_sleep(retry_delay)
+check_keepalive(reg, consul_server, retry_delay)
+return
+end
-local function watch_health(consul_server)
-local client = resty_consul:new(get_opts(consul_server, false))
+if reg.stop_flag then
+return
+end
