bzp2010 commented on code in PR #12263:
URL: https://github.com/apache/apisix/pull/12263#discussion_r2116198414
##########
apisix/discovery/nacos/init.lua:
##########
@@ -15,410 +15,196 @@
-- limitations under the License.
--
-local require = require
-local local_conf = require('apisix.core.config_local').local_conf()
-local http = require('resty.http')
-local core = require('apisix.core')
-local ipairs = ipairs
-local type = type
-local math = math
-local math_random = math.random
-local ngx = ngx
-local ngx_re = require('ngx.re')
-local ngx_timer_at = ngx.timer.at
-local ngx_timer_every = ngx.timer.every
-local string = string
-local string_sub = string.sub
-local str_byte = string.byte
-local str_find = core.string.find
-local log = core.log
+local core = require("apisix.core")
+local nacos_factory = require("apisix.discovery.nacos.factory")
+local utils = require("apisix.discovery.nacos.utils")
+local process = require("ngx.process")
+local ipairs = ipairs
+local require = require
+local table = require("apisix.core.table")
+local pcall = pcall
+local pairs = pairs
+local local_conf = require('apisix.core.config_local').local_conf()
+local ngx = ngx
+
+local shdict_name = "nacos"
+if ngx.config.subsystem == "stream" then
+ shdict_name = shdict_name .. "-stream"
+end
+
+local nacos_dict = ngx.shared[shdict_name]
+local OLD_CONFIG_ID = utils.old_config_id
+local _M = {}
+local nacos_clients = {}
-local default_weight
-local applications
-local auth_path = 'auth/login'
-local instance_list_path = 'ns/instance/list?healthyOnly=true&serviceName='
-local default_namespace_id = "public"
-local default_group_name = "DEFAULT_GROUP"
-local access_key
-local secret_key
-
-local events
-local events_list
-
-
-local _M = {}
-
-local function discovery_nacos_callback(data, event, source, pid)
- applications = data
- log.notice("update local variable application, event is: ", event,
- "source: ", source, "server pid:", pid,
- ", application: ", core.json.encode(applications, true))
-end
-
-
-local function request(request_uri, path, body, method, basic_auth)
- local url = request_uri .. path
- log.info('request url:', url)
- local headers = {}
- headers['Accept'] = 'application/json'
-
- if basic_auth then
- headers['Authorization'] = basic_auth
- end
+function _M.nodes(service_name, discovery_args)
+ local ns_id = discovery_args and discovery_args.namespace_id or
utils.default_namespace_id
+ local group_name = discovery_args and discovery_args.group_name or
utils.default_group_name
+ local id = discovery_args and discovery_args.id or OLD_CONFIG_ID
+ local key = utils.generate_key(id,
+ ns_id,
+ group_name,
+ service_name)
+ local value = nacos_dict:get_stale(key)
Review Comment:
Why is it necessary to retrieve potentially stale data using `get_stale`
instead of `get`?
##########
apisix/discovery/nacos/init.lua:
##########
@@ -15,410 +15,196 @@
-- limitations under the License.
--
-local require = require
-local local_conf = require('apisix.core.config_local').local_conf()
-local http = require('resty.http')
-local core = require('apisix.core')
-local ipairs = ipairs
-local type = type
-local math = math
-local math_random = math.random
-local ngx = ngx
-local ngx_re = require('ngx.re')
-local ngx_timer_at = ngx.timer.at
-local ngx_timer_every = ngx.timer.every
-local string = string
-local string_sub = string.sub
-local str_byte = string.byte
-local str_find = core.string.find
-local log = core.log
+local core = require("apisix.core")
+local nacos_factory = require("apisix.discovery.nacos.factory")
+local utils = require("apisix.discovery.nacos.utils")
+local process = require("ngx.process")
+local ipairs = ipairs
+local require = require
+local table = require("apisix.core.table")
+local pcall = pcall
+local pairs = pairs
+local local_conf = require('apisix.core.config_local').local_conf()
+local ngx = ngx
+
+local shdict_name = "nacos"
+if ngx.config.subsystem == "stream" then
+ shdict_name = shdict_name .. "-stream"
+end
+
+local nacos_dict = ngx.shared[shdict_name]
+local OLD_CONFIG_ID = utils.old_config_id
+local _M = {}
+local nacos_clients = {}
-local default_weight
-local applications
-local auth_path = 'auth/login'
-local instance_list_path = 'ns/instance/list?healthyOnly=true&serviceName='
-local default_namespace_id = "public"
-local default_group_name = "DEFAULT_GROUP"
-local access_key
-local secret_key
-
-local events
-local events_list
-
-
-local _M = {}
-
-local function discovery_nacos_callback(data, event, source, pid)
- applications = data
- log.notice("update local variable application, event is: ", event,
- "source: ", source, "server pid:", pid,
- ", application: ", core.json.encode(applications, true))
-end
-
-
-local function request(request_uri, path, body, method, basic_auth)
- local url = request_uri .. path
- log.info('request url:', url)
- local headers = {}
- headers['Accept'] = 'application/json'
-
- if basic_auth then
- headers['Authorization'] = basic_auth
- end
+function _M.nodes(service_name, discovery_args)
+ local ns_id = discovery_args and discovery_args.namespace_id or
utils.default_namespace_id
+ local group_name = discovery_args and discovery_args.group_name or
utils.default_group_name
+ local id = discovery_args and discovery_args.id or OLD_CONFIG_ID
+ local key = utils.generate_key(id,
+ ns_id,
+ group_name,
+ service_name)
+ local value = nacos_dict:get_stale(key)
+ local nodes = {}
+ if not value then
+ -- maximum waiting time: 5 seconds
Review Comment:
indent
##########
apisix/discovery/nacos/factory.lua:
##########
@@ -0,0 +1,292 @@
+--
+-- 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.
+--
+
+local require = require
+local core = require("apisix.core")
+local http = require('resty.http')
+local ngx = ngx
+local ngx_re = require('ngx.re')
+local utils = require("apisix.discovery.nacos.utils")
+local string = string
+local string_sub = string.sub
+local str_find = core.string.find
+local ngx_timer_at = ngx.timer.at
+local math_random = math.random
+local ipairs = ipairs
+local shdict_name = "nacos"
+if ngx.config.subsystem == "stream" then
+ shdict_name = shdict_name .. "-stream"
+end
+
+local nacos_dict = ngx.shared[shdict_name]
+local NACOS_LOGIN_PATH = "/auth/login"
+local NACOS_INSTANCE_PATH = "/ns/instance/list"
+local _M = {}
+
+
+local function _request(method, uri, params, headers, body, options)
+ local url = uri
+ if params ~= nil and params ~= {} then
+ url = uri .. "?" .. ngx.encode_args(params)
+ end
+ local httpc = http.new()
+ local timeout = options and options.timeout or {}
+ local connect_timeout = timeout.connect and timeout.connect * 1000 or 2000
+ local read_timeout = timeout.read and timeout.read * 1000 or 2000
+ local write_timeout = timeout.write and timeout.write * 1000 or 5000
+
+ httpc:set_timeouts(connect_timeout, read_timeout, write_timeout )
+ local res, err = httpc:request_uri(url, {
+ method = method,
+ headers = headers,
+ body = body,
+ ssl_verify = false,
+ })
+
+ if not res then
+ return nil, err
+ end
+
+ if not res.body or res.status ~= 200 then
+ return nil, 'status = ' .. res.status
+ end
+
+ core.log.info("request to nacos, uri: ", url, "response: ", res.body)
+
+ local data, err = core.json.decode(res.body)
+ if not data then
+ return nil, err
+ end
+
+ return data
+end
+
+
+local function get_base_uri(hosts)
+ local host = hosts
+ -- TODO Add health check to get healthy nodes.
+ local url = host[math_random(#host)]
+ local auth_idx = core.string.rfind_char(url, '@')
+ local username, password
+ if auth_idx then
+ local protocol_idx = str_find(url, '://')
+ local protocol = string_sub(url, 1, protocol_idx + 2)
+ local user_and_password = string_sub(url, protocol_idx + 3, auth_idx -
1)
+ local arr = ngx_re.split(user_and_password, ':')
+ if #arr == 2 then
+ username = arr[1]
+ password = arr[2]
+ end
+ local other = string_sub(url, auth_idx + 1)
+ url = protocol .. other
+ end
+ return url, username, password
+end
+
+local function request_login(self, host, username, password)
+ local params = {
+ ["username"] = username,
+ ["password"] = password,
+ }
+ -- backward compat: NACOS_LOGIN_PATH starts with "/" so
+ -- we need to remove the last "/" from prefix
+ if string_sub(self.config.prefix, -1) == "/" then
+ self.config.prefix = string_sub(self.config.prefix, 1, -2)
+ end
+ local uri = host .. self.config.prefix .. NACOS_LOGIN_PATH
+
+ local headers = {
+ ["Content-Type"] ="application/x-www-form-urlencoded"
+ }
+
+ local resp, err = _request("POST", uri, params, headers, nil,
{timeout=self.config.timeout})
+ if not resp then
+ core.log.error("failed to fetch token from nacos, uri: ", uri, " err:
", err)
+ return ""
+ end
+ return resp.accessToken
+end
+
+
+local function request_instance_list(self, params, host)
+ -- backward compat: NACOS_INSTANCE_PATH starts with "/" so
+ -- we need to remove the last "/" from prefix
+ if string_sub(self.config.prefix, -1) == "/" then
+ self.config.prefix = string_sub(self.config.prefix, 1, -2)
+ end
+ local uri = host .. self.config.prefix .. NACOS_INSTANCE_PATH
+
+ local resp, err = _request("GET", uri, params)
+ if not resp then
+ core.log.error("failed to fetch instances list from nacos, uri: ",
uri, " err: ", err)
+ return {}
+ end
+ return resp.hosts or {}
+end
+
+local function is_grpc(scheme)
+ if scheme == 'grpc' or scheme == 'grpcs' then
+ return true
+ end
+
+ return false
+end
+
+
+local function fetch_instances(self, serv)
+ local config = self.config
+
+ local params = {
+ ["namespaceId"] = serv.namespace_id or "",
+ ["groupName"] = serv.group_name or "DEFAULT_GROUP",
+ ["serviceName"] = serv.name,
+ ["healthyOnly"] = "true"
+ }
+
+ local auth = config.auth or {}
+ -- for backward compat:
+ -- In older method, we passed username and password inside of host
+ -- In new method its passed separately
+ local username, password, host
+ if config.old_conf then
+ -- extract username and password from host
+ host, username, password = get_base_uri(config.hosts)
+ else
+ host = config.hosts[math_random(#config.hosts)]
+ if (auth.username and auth.username ~= "") and (auth.password and
auth.password ~= "") then
+ username = auth.username
+ password = auth.password
+ end
+ end
+ if username and username ~= "" and password and password ~= "" then
+ local token = request_login(self, host, username, password)
+ params["accessToken"] = token
+ end
+
+ if auth.token and auth.token ~= "" then
+ params["accessToken"] = auth.token
+ end
+
+ if (auth.access_key and auth.access_key ~= "") and
+ (auth.secret_key and auth.secret_key ~= "") then
+ local ak, data, signature = utils.generate_signature(serv.group_name,
+ serv.name, auth.access_key, auth.secret_key)
+ params["ak"] = ak
+ params["data"] = data
+ params["signature"] = signature
+ end
+
+ local instances = request_instance_list(self, params, host)
+ local nodes = {}
+ for _, instance in ipairs(instances) do
+ local node = {
+ host = instance.ip,
+ port = instance.port,
+ weight = instance.weight or self.config.default_weight,
+ metadata = instance.metadata,
+ }
+ -- docs: https://github.com/yidongnan/grpc-spring-boot-starter/pull/496
+ if is_grpc(instance.scheme) and instance.metadata and
instance.metadata.gRPC_port then
+ node.port = host.metadata.gRPC_port
+ end
+ core.table.insert(nodes, node)
+ end
+ return nodes
+end
+
+
+local function fetch_full_registry(self)
+ return function (premature)
+ if premature then
+ return
+ end
+
+ local config = self.config
+ local services_in_use = utils.get_nacos_services(config.id)
+ for _, serv in ipairs(services_in_use) do
+ if self.stop_flag then
+ core.log.error("nacos client is exited, id: ", config.id)
+ return
+ end
+
+ local nodes = self:fetch_instances(serv)
+ if #nodes > 0 then
+ local content = core.json.encode(nodes)
+ local key = utils.generate_key(serv.id, serv.namespace_id,
+ serv.group_name, serv.name)
+ nacos_dict:set(key, content, self.config.fetch_interval * 10)
+ end
+ end
+
+ ngx_timer_at(self.config.fetch_interval, self:fetch_full_registry())
+ end
+end
+
+
+local function stop(self)
+ self.stop_flag = true
+
+ if self.checker then
+ self.checker:clear()
+ end
+end
+
+
+local function start(self)
+ ngx_timer_at(0, self:fetch_full_registry())
+end
+
+
+function _M.new(config)
+ local version = ngx.md5(core.json.encode(config, true))
+
+ local client = {
+ id = config.id,
+ version = version,
+ config = config,
+ stop_flag = false,
+
+ start = start,
+ stop = stop,
+ fetch_instances = fetch_instances,
+ fetch_full_registry = fetch_full_registry,
Review Comment:
According to APISIX convention, you should use the meta table and `_index`.
I don't remember us having such a precedent.
##########
apisix/discovery/nacos/factory.lua:
##########
@@ -0,0 +1,292 @@
+--
+-- 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.
+--
+
+local require = require
+local core = require("apisix.core")
+local http = require('resty.http')
+local ngx = ngx
+local ngx_re = require('ngx.re')
+local utils = require("apisix.discovery.nacos.utils")
+local string = string
+local string_sub = string.sub
+local str_find = core.string.find
+local ngx_timer_at = ngx.timer.at
+local math_random = math.random
+local ipairs = ipairs
+local shdict_name = "nacos"
+if ngx.config.subsystem == "stream" then
+ shdict_name = shdict_name .. "-stream"
+end
Review Comment:
Is it possible to share services directly between http and stream, i.e.,
using meta shdict?
The specific reasons are as follows:
We do not configure discovery separately for http and stream in the
configuration file; they always use the same provider or the same provider ID.
The only difference is that they may use different service names. However,
we do not seem to need to concern ourselves with this.
In fact, you have already set
https://github.com/apache/apisix/pull/12263/files#diff-ac8ce1b2b70b12a55abac8b1d508079cab4136f5bf830052ba325e543b04bf9eR130-R148,
which includes both routes and stream routes.
##########
apisix/discovery/nacos/factory.lua:
##########
@@ -0,0 +1,292 @@
+--
+-- 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.
+--
+
+local require = require
+local core = require("apisix.core")
+local http = require('resty.http')
+local ngx = ngx
+local ngx_re = require('ngx.re')
+local utils = require("apisix.discovery.nacos.utils")
+local string = string
+local string_sub = string.sub
+local str_find = core.string.find
+local ngx_timer_at = ngx.timer.at
+local math_random = math.random
+local ipairs = ipairs
+local shdict_name = "nacos"
+if ngx.config.subsystem == "stream" then
+ shdict_name = shdict_name .. "-stream"
+end
+
+local nacos_dict = ngx.shared[shdict_name]
+local NACOS_LOGIN_PATH = "/auth/login"
+local NACOS_INSTANCE_PATH = "/ns/instance/list"
+local _M = {}
+
+
+local function _request(method, uri, params, headers, body, options)
+ local url = uri
+ if params ~= nil and params ~= {} then
+ url = uri .. "?" .. ngx.encode_args(params)
+ end
+ local httpc = http.new()
+ local timeout = options and options.timeout or {}
+ local connect_timeout = timeout.connect and timeout.connect * 1000 or 2000
+ local read_timeout = timeout.read and timeout.read * 1000 or 2000
+ local write_timeout = timeout.write and timeout.write * 1000 or 5000
+
+ httpc:set_timeouts(connect_timeout, read_timeout, write_timeout )
+ local res, err = httpc:request_uri(url, {
+ method = method,
+ headers = headers,
+ body = body,
+ ssl_verify = false,
+ })
+
+ if not res then
+ return nil, err
+ end
+
+ if not res.body or res.status ~= 200 then
+ return nil, 'status = ' .. res.status
+ end
+
+ core.log.info("request to nacos, uri: ", url, "response: ", res.body)
+
+ local data, err = core.json.decode(res.body)
+ if not data then
+ return nil, err
+ end
+
+ return data
+end
+
+
+local function get_base_uri(hosts)
+ local host = hosts
+ -- TODO Add health check to get healthy nodes.
+ local url = host[math_random(#host)]
+ local auth_idx = core.string.rfind_char(url, '@')
+ local username, password
+ if auth_idx then
+ local protocol_idx = str_find(url, '://')
+ local protocol = string_sub(url, 1, protocol_idx + 2)
+ local user_and_password = string_sub(url, protocol_idx + 3, auth_idx -
1)
+ local arr = ngx_re.split(user_and_password, ':')
+ if #arr == 2 then
+ username = arr[1]
+ password = arr[2]
+ end
+ local other = string_sub(url, auth_idx + 1)
+ url = protocol .. other
+ end
+ return url, username, password
+end
+
+local function request_login(self, host, username, password)
+ local params = {
+ ["username"] = username,
+ ["password"] = password,
Review Comment:
```suggestion
username = username,
password = password,
```
##########
apisix/discovery/nacos/init.lua:
##########
@@ -15,410 +15,196 @@
-- limitations under the License.
--
-local require = require
-local local_conf = require('apisix.core.config_local').local_conf()
-local http = require('resty.http')
-local core = require('apisix.core')
-local ipairs = ipairs
-local type = type
-local math = math
-local math_random = math.random
-local ngx = ngx
-local ngx_re = require('ngx.re')
-local ngx_timer_at = ngx.timer.at
-local ngx_timer_every = ngx.timer.every
-local string = string
-local string_sub = string.sub
-local str_byte = string.byte
-local str_find = core.string.find
-local log = core.log
+local core = require("apisix.core")
+local nacos_factory = require("apisix.discovery.nacos.factory")
+local utils = require("apisix.discovery.nacos.utils")
+local process = require("ngx.process")
+local ipairs = ipairs
+local require = require
+local table = require("apisix.core.table")
+local pcall = pcall
+local pairs = pairs
+local local_conf = require('apisix.core.config_local').local_conf()
+local ngx = ngx
+
+local shdict_name = "nacos"
+if ngx.config.subsystem == "stream" then
+ shdict_name = shdict_name .. "-stream"
+end
+
+local nacos_dict = ngx.shared[shdict_name]
+local OLD_CONFIG_ID = utils.old_config_id
+local _M = {}
+local nacos_clients = {}
-local default_weight
-local applications
-local auth_path = 'auth/login'
-local instance_list_path = 'ns/instance/list?healthyOnly=true&serviceName='
-local default_namespace_id = "public"
-local default_group_name = "DEFAULT_GROUP"
-local access_key
-local secret_key
-
-local events
-local events_list
-
-
-local _M = {}
-
-local function discovery_nacos_callback(data, event, source, pid)
- applications = data
- log.notice("update local variable application, event is: ", event,
- "source: ", source, "server pid:", pid,
- ", application: ", core.json.encode(applications, true))
-end
-
-
-local function request(request_uri, path, body, method, basic_auth)
- local url = request_uri .. path
- log.info('request url:', url)
- local headers = {}
- headers['Accept'] = 'application/json'
-
- if basic_auth then
- headers['Authorization'] = basic_auth
- end
+function _M.nodes(service_name, discovery_args)
+ local ns_id = discovery_args and discovery_args.namespace_id or
utils.default_namespace_id
+ local group_name = discovery_args and discovery_args.group_name or
utils.default_group_name
+ local id = discovery_args and discovery_args.id or OLD_CONFIG_ID
+ local key = utils.generate_key(id,
+ ns_id,
+ group_name,
+ service_name)
+ local value = nacos_dict:get_stale(key)
+ local nodes = {}
+ if not value then
+ -- maximum waiting time: 5 seconds
+ local waiting_time = 5
+ local step = 0.1
+ local logged = false
+ while not value and waiting_time > 0 do
+ if not logged then
+ logged = true
+ end
- if body and 'table' == type(body) then
- local err
- body, err = core.json.encode(body)
- if not body then
- return nil, 'invalid body : ' .. err
+ ngx.sleep(step)
+ waiting_time = waiting_time - step
+ value = nacos_dict:get_stale(key)
end
Review Comment:
What does this code mean?
Currently, nodes have been reduced to a mechanism that only reads from
shdict. Before the code I selected, you did not attempt to initiate a request,
so nodes would not suddenly be filled.
I understand that filling shdict is done by a decoupled privileged process,
but I don't understand the value of using a spin lock-like mechanism for
waiting here.
In my opinion, it should fast fail, which would cause APISIX to return a 503
error and print the log.
No matter how I look at it, it's strange that the request was blocked for 5
seconds because the service discovery provider did not return nodes.
##########
apisix/discovery/nacos/factory.lua:
##########
@@ -0,0 +1,292 @@
+--
+-- 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.
+--
+
+local require = require
+local core = require("apisix.core")
+local http = require('resty.http')
+local ngx = ngx
+local ngx_re = require('ngx.re')
+local utils = require("apisix.discovery.nacos.utils")
+local string = string
+local string_sub = string.sub
+local str_find = core.string.find
+local ngx_timer_at = ngx.timer.at
+local math_random = math.random
+local ipairs = ipairs
+local shdict_name = "nacos"
+if ngx.config.subsystem == "stream" then
+ shdict_name = shdict_name .. "-stream"
+end
+
+local nacos_dict = ngx.shared[shdict_name]
+local NACOS_LOGIN_PATH = "/auth/login"
+local NACOS_INSTANCE_PATH = "/ns/instance/list"
+local _M = {}
+
+
+local function _request(method, uri, params, headers, body, options)
+ local url = uri
+ if params ~= nil and params ~= {} then
+ url = uri .. "?" .. ngx.encode_args(params)
+ end
+ local httpc = http.new()
+ local timeout = options and options.timeout or {}
+ local connect_timeout = timeout.connect and timeout.connect * 1000 or 2000
+ local read_timeout = timeout.read and timeout.read * 1000 or 2000
+ local write_timeout = timeout.write and timeout.write * 1000 or 5000
+
+ httpc:set_timeouts(connect_timeout, read_timeout, write_timeout )
+ local res, err = httpc:request_uri(url, {
+ method = method,
+ headers = headers,
+ body = body,
+ ssl_verify = false,
Review Comment:
It seems unsafe.
##########
apisix/discovery/nacos/init.lua:
##########
@@ -15,410 +15,196 @@
-- limitations under the License.
--
-local require = require
-local local_conf = require('apisix.core.config_local').local_conf()
-local http = require('resty.http')
-local core = require('apisix.core')
-local ipairs = ipairs
-local type = type
-local math = math
-local math_random = math.random
-local ngx = ngx
-local ngx_re = require('ngx.re')
-local ngx_timer_at = ngx.timer.at
-local ngx_timer_every = ngx.timer.every
-local string = string
-local string_sub = string.sub
-local str_byte = string.byte
-local str_find = core.string.find
-local log = core.log
+local core = require("apisix.core")
+local nacos_factory = require("apisix.discovery.nacos.factory")
+local utils = require("apisix.discovery.nacos.utils")
+local process = require("ngx.process")
+local ipairs = ipairs
+local require = require
+local table = require("apisix.core.table")
+local pcall = pcall
+local pairs = pairs
+local local_conf = require('apisix.core.config_local').local_conf()
+local ngx = ngx
+
+local shdict_name = "nacos"
+if ngx.config.subsystem == "stream" then
+ shdict_name = shdict_name .. "-stream"
+end
+
+local nacos_dict = ngx.shared[shdict_name]
+local OLD_CONFIG_ID = utils.old_config_id
+local _M = {}
+local nacos_clients = {}
-local default_weight
-local applications
-local auth_path = 'auth/login'
-local instance_list_path = 'ns/instance/list?healthyOnly=true&serviceName='
-local default_namespace_id = "public"
-local default_group_name = "DEFAULT_GROUP"
-local access_key
-local secret_key
-
-local events
-local events_list
-
-
-local _M = {}
-
-local function discovery_nacos_callback(data, event, source, pid)
- applications = data
- log.notice("update local variable application, event is: ", event,
- "source: ", source, "server pid:", pid,
- ", application: ", core.json.encode(applications, true))
-end
-
-
-local function request(request_uri, path, body, method, basic_auth)
- local url = request_uri .. path
- log.info('request url:', url)
- local headers = {}
- headers['Accept'] = 'application/json'
-
- if basic_auth then
- headers['Authorization'] = basic_auth
- end
+function _M.nodes(service_name, discovery_args)
+ local ns_id = discovery_args and discovery_args.namespace_id or
utils.default_namespace_id
+ local group_name = discovery_args and discovery_args.group_name or
utils.default_group_name
+ local id = discovery_args and discovery_args.id or OLD_CONFIG_ID
+ local key = utils.generate_key(id,
+ ns_id,
+ group_name,
+ service_name)
+ local value = nacos_dict:get_stale(key)
+ local nodes = {}
+ if not value then
+ -- maximum waiting time: 5 seconds
+ local waiting_time = 5
+ local step = 0.1
+ local logged = false
+ while not value and waiting_time > 0 do
+ if not logged then
+ logged = true
+ end
- if body and 'table' == type(body) then
- local err
- body, err = core.json.encode(body)
- if not body then
- return nil, 'invalid body : ' .. err
+ ngx.sleep(step)
+ waiting_time = waiting_time - step
+ value = nacos_dict:get_stale(key)
end
- headers['Content-Type'] = 'application/json'
- end
-
- local httpc = http.new()
- local timeout = local_conf.discovery.nacos.timeout
- local connect_timeout = timeout.connect
- local send_timeout = timeout.send
- local read_timeout = timeout.read
- log.info('connect_timeout:', connect_timeout, ', send_timeout:',
send_timeout,
- ', read_timeout:', read_timeout)
- httpc:set_timeouts(connect_timeout, send_timeout, read_timeout)
- local res, err = httpc:request_uri(url, {
- method = method,
- headers = headers,
- body = body,
- ssl_verify = true,
- })
- if not res then
- return nil, err
- end
-
- if not res.body or res.status ~= 200 then
- return nil, 'status = ' .. res.status
- end
-
- local json_str = res.body
- local data, err = core.json.decode(json_str)
- if not data then
- return nil, err
- end
- return data
-end
-
-
-local function get_url(request_uri, path)
- return request(request_uri, path, nil, 'GET', nil)
-end
-
-
-local function post_url(request_uri, path, body)
- return request(request_uri, path, body, 'POST', nil)
-end
-
-
-local function get_token_param(base_uri, username, password)
- if not username or not password then
- return ''
- end
-
- local args = { username = username, password = password}
- local data, err = post_url(base_uri, auth_path .. '?' ..
ngx.encode_args(args), nil)
- if err then
- log.error('nacos login fail:', username, ' ', password, ' desc:', err)
- return nil, err
- end
- return '&accessToken=' .. data.accessToken
-end
-
-
-local function get_namespace_param(namespace_id)
- local param = ''
- if namespace_id then
- local args = {namespaceId = namespace_id}
- param = '&' .. ngx.encode_args(args)
end
- return param
-end
-
-
-local function get_group_name_param(group_name)
- local param = ''
- if group_name then
- local args = {groupName = group_name}
- param = '&' .. ngx.encode_args(args)
- end
- return param
-end
-
-
-local function get_signed_param(group_name, service_name)
- local param = ''
- if access_key ~= '' and secret_key ~= '' then
- local str_to_sign = ngx.now() * 1000 .. '@@' .. group_name .. '@@' ..
service_name
- local args = {
- ak = access_key,
- data = str_to_sign,
- signature = ngx.encode_base64(ngx.hmac_sha1(secret_key,
str_to_sign))
- }
- param = '&' .. ngx.encode_args(args)
+ if not value then
+ core.log.error("nacos service not found: ", service_name)
+ return nodes
end
- return param
-end
+ nodes = core.json.decode(value)
-local function get_base_uri()
- local host = local_conf.discovery.nacos.host
- -- TODO Add health check to get healthy nodes.
- local url = host[math_random(#host)]
- local auth_idx = core.string.rfind_char(url, '@')
- local username, password
- if auth_idx then
- local protocol_idx = str_find(url, '://')
- local protocol = string_sub(url, 1, protocol_idx + 2)
- local user_and_password = string_sub(url, protocol_idx + 3, auth_idx -
1)
- local arr = ngx_re.split(user_and_password, ':')
- if #arr == 2 then
- username = arr[1]
- password = arr[2]
+ local res = {}
+ for _, node in ipairs(nodes) do
+ if discovery_args then
+ if utils.match_metdata(node.metadata, discovery_args.metadata) then
+ core.table.insert(res, node)
+ end
+ else
+ core.table.insert(res, node)
end
- local other = string_sub(url, auth_idx + 1)
- url = protocol .. other
end
- if local_conf.discovery.nacos.prefix then
- url = url .. local_conf.discovery.nacos.prefix
- end
-
- if str_byte(url, #url) ~= str_byte('/') then
- url = url .. '/'
- end
-
- return url, username, password
+ core.log.info("nacos service_name: ", service_name, " nodes: ",
core.json.encode(res))
+ return res
end
-
-local function de_duplication(services, namespace_id, group_name,
service_name, scheme)
- for _, service in ipairs(services) do
- if service.namespace_id == namespace_id and service.group_name ==
group_name
- and service.service_name == service_name and service.scheme ==
scheme then
- return true
- end
- end
- return false
+local function generate_new_config_from_old(discovery_conf)
+ local config = {
+ id = OLD_CONFIG_ID,
+ hosts = discovery_conf.host,
+ prefix = discovery_conf.prefix,
+ fetch_interval = discovery_conf.fetch_interval,
+ auth = {
+ access_key = discovery_conf.access_key,
+ secret_key = discovery_conf.secret_key,
+ },
+ default_weight = discovery_conf.weight,
+ timeout = discovery_conf.timeout,
+ old_conf = true
+ }
+ return {config}
end
+function _M.init_worker()
+ local local_conf = require("apisix.core.config_local").local_conf(true)
+ local discovery_conf = local_conf.discovery and local_conf.discovery.nacos
or {}
-local function iter_and_add_service(services, values)
- if not values then
+ if process.type() ~= "privileged agent" then
return
end
- for _, value in core.config_util.iterate_values(values) do
- local conf = value.value
- if not conf then
- goto CONTINUE
- end
-
- local up
- if conf.upstream then
- up = conf.upstream
- else
- up = conf
- end
-
- local namespace_id = (up.discovery_args and
up.discovery_args.namespace_id)
- or default_namespace_id
-
- local group_name = (up.discovery_args and up.discovery_args.group_name)
- or default_group_name
+ local keep = {}
+ -- support old way
+ if discovery_conf.host then
+ discovery_conf = generate_new_config_from_old(discovery_conf)
+ end
+ for _, val in ipairs(discovery_conf) do
+ local id = val.id
+ local version = ngx.md5(core.json.encode(val, true))
+ keep[id] = true
- local dup = de_duplication(services, namespace_id, group_name,
- up.service_name, up.scheme)
- if dup then
+ -- The nacos config has not been changed.
+ if nacos_clients[id] and nacos_clients[id].version == version then
goto CONTINUE
end
- if up.discovery_type == 'nacos' then
- core.table.insert(services, {
- service_name = up.service_name,
- namespace_id = namespace_id,
- group_name = group_name,
- scheme = up.scheme,
- })
+ if nacos_clients[id] then
+ nacos_clients[id]:stop()
end
+ local new_client = nacos_factory.new(val)
+ new_client:start()
+ nacos_clients[id] = new_client
+
::CONTINUE::
Review Comment:
Why check for configuration changes and rebuild the client?
1. init_worker is not called frequently; it is only called once when the
privileged process starts. Therefore, when do you check for configuration
changes?
2. You store nacos_clients in a global variable in the init.lua file, but
regardless, the scope of this variable is limited to the LuaVM. Restarting the
privileged process is equivalent to restarting the worker, which causes the
LuaVM to be rebuilt, thereby clearing the table.
Comparing the configuration version with an old client stored there seems
pointless, as it will always fall into a branch where the old client no longer
exists.
##########
apisix/discovery/nacos/factory.lua:
##########
@@ -0,0 +1,292 @@
+--
+-- 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.
+--
+
+local require = require
+local core = require("apisix.core")
+local http = require('resty.http')
+local ngx = ngx
+local ngx_re = require('ngx.re')
+local utils = require("apisix.discovery.nacos.utils")
+local string = string
+local string_sub = string.sub
+local str_find = core.string.find
+local ngx_timer_at = ngx.timer.at
+local math_random = math.random
+local ipairs = ipairs
+local shdict_name = "nacos"
+if ngx.config.subsystem == "stream" then
+ shdict_name = shdict_name .. "-stream"
+end
+
+local nacos_dict = ngx.shared[shdict_name]
+local NACOS_LOGIN_PATH = "/auth/login"
+local NACOS_INSTANCE_PATH = "/ns/instance/list"
+local _M = {}
+
+
+local function _request(method, uri, params, headers, body, options)
+ local url = uri
+ if params ~= nil and params ~= {} then
+ url = uri .. "?" .. ngx.encode_args(params)
+ end
+ local httpc = http.new()
+ local timeout = options and options.timeout or {}
+ local connect_timeout = timeout.connect and timeout.connect * 1000 or 2000
+ local read_timeout = timeout.read and timeout.read * 1000 or 2000
+ local write_timeout = timeout.write and timeout.write * 1000 or 5000
+
+ httpc:set_timeouts(connect_timeout, read_timeout, write_timeout )
+ local res, err = httpc:request_uri(url, {
+ method = method,
+ headers = headers,
+ body = body,
+ ssl_verify = false,
+ })
+
+ if not res then
+ return nil, err
+ end
+
+ if not res.body or res.status ~= 200 then
+ return nil, 'status = ' .. res.status
+ end
+
+ core.log.info("request to nacos, uri: ", url, "response: ", res.body)
+
+ local data, err = core.json.decode(res.body)
+ if not data then
+ return nil, err
+ end
+
+ return data
+end
+
+
+local function get_base_uri(hosts)
+ local host = hosts
+ -- TODO Add health check to get healthy nodes.
+ local url = host[math_random(#host)]
+ local auth_idx = core.string.rfind_char(url, '@')
+ local username, password
+ if auth_idx then
+ local protocol_idx = str_find(url, '://')
+ local protocol = string_sub(url, 1, protocol_idx + 2)
+ local user_and_password = string_sub(url, protocol_idx + 3, auth_idx -
1)
+ local arr = ngx_re.split(user_and_password, ':')
+ if #arr == 2 then
+ username = arr[1]
+ password = arr[2]
+ end
+ local other = string_sub(url, auth_idx + 1)
+ url = protocol .. other
+ end
+ return url, username, password
+end
+
+local function request_login(self, host, username, password)
+ local params = {
+ ["username"] = username,
+ ["password"] = password,
+ }
+ -- backward compat: NACOS_LOGIN_PATH starts with "/" so
+ -- we need to remove the last "/" from prefix
+ if string_sub(self.config.prefix, -1) == "/" then
+ self.config.prefix = string_sub(self.config.prefix, 1, -2)
+ end
+ local uri = host .. self.config.prefix .. NACOS_LOGIN_PATH
+
+ local headers = {
+ ["Content-Type"] ="application/x-www-form-urlencoded"
+ }
+
+ local resp, err = _request("POST", uri, params, headers, nil,
{timeout=self.config.timeout})
+ if not resp then
+ core.log.error("failed to fetch token from nacos, uri: ", uri, " err:
", err)
+ return ""
+ end
+ return resp.accessToken
+end
+
+
+local function request_instance_list(self, params, host)
+ -- backward compat: NACOS_INSTANCE_PATH starts with "/" so
+ -- we need to remove the last "/" from prefix
+ if string_sub(self.config.prefix, -1) == "/" then
+ self.config.prefix = string_sub(self.config.prefix, 1, -2)
+ end
+ local uri = host .. self.config.prefix .. NACOS_INSTANCE_PATH
+
+ local resp, err = _request("GET", uri, params)
+ if not resp then
+ core.log.error("failed to fetch instances list from nacos, uri: ",
uri, " err: ", err)
+ return {}
+ end
+ return resp.hosts or {}
+end
+
+local function is_grpc(scheme)
+ if scheme == 'grpc' or scheme == 'grpcs' then
+ return true
+ end
+
+ return false
+end
Review Comment:
```suggestion
local function is_grpc(scheme)
return scheme == "grpc" or scheme == "grpcs"
end
```
##########
apisix/discovery/nacos/init.lua:
##########
@@ -15,410 +15,196 @@
-- limitations under the License.
--
-local require = require
-local local_conf = require('apisix.core.config_local').local_conf()
-local http = require('resty.http')
-local core = require('apisix.core')
-local ipairs = ipairs
-local type = type
-local math = math
-local math_random = math.random
-local ngx = ngx
-local ngx_re = require('ngx.re')
-local ngx_timer_at = ngx.timer.at
-local ngx_timer_every = ngx.timer.every
-local string = string
-local string_sub = string.sub
-local str_byte = string.byte
-local str_find = core.string.find
-local log = core.log
+local core = require("apisix.core")
+local nacos_factory = require("apisix.discovery.nacos.factory")
+local utils = require("apisix.discovery.nacos.utils")
+local process = require("ngx.process")
+local ipairs = ipairs
+local require = require
+local table = require("apisix.core.table")
+local pcall = pcall
+local pairs = pairs
+local local_conf = require('apisix.core.config_local').local_conf()
+local ngx = ngx
+
+local shdict_name = "nacos"
+if ngx.config.subsystem == "stream" then
+ shdict_name = shdict_name .. "-stream"
+end
+
+local nacos_dict = ngx.shared[shdict_name]
+local OLD_CONFIG_ID = utils.old_config_id
+local _M = {}
+local nacos_clients = {}
-local default_weight
-local applications
-local auth_path = 'auth/login'
-local instance_list_path = 'ns/instance/list?healthyOnly=true&serviceName='
-local default_namespace_id = "public"
-local default_group_name = "DEFAULT_GROUP"
-local access_key
-local secret_key
-
-local events
-local events_list
-
-
-local _M = {}
-
-local function discovery_nacos_callback(data, event, source, pid)
- applications = data
- log.notice("update local variable application, event is: ", event,
- "source: ", source, "server pid:", pid,
- ", application: ", core.json.encode(applications, true))
-end
-
-
-local function request(request_uri, path, body, method, basic_auth)
- local url = request_uri .. path
- log.info('request url:', url)
- local headers = {}
- headers['Accept'] = 'application/json'
-
- if basic_auth then
- headers['Authorization'] = basic_auth
- end
+function _M.nodes(service_name, discovery_args)
+ local ns_id = discovery_args and discovery_args.namespace_id or
utils.default_namespace_id
+ local group_name = discovery_args and discovery_args.group_name or
utils.default_group_name
+ local id = discovery_args and discovery_args.id or OLD_CONFIG_ID
+ local key = utils.generate_key(id,
+ ns_id,
+ group_name,
+ service_name)
+ local value = nacos_dict:get_stale(key)
+ local nodes = {}
+ if not value then
+ -- maximum waiting time: 5 seconds
+ local waiting_time = 5
+ local step = 0.1
+ local logged = false
+ while not value and waiting_time > 0 do
+ if not logged then
+ logged = true
+ end
- if body and 'table' == type(body) then
- local err
- body, err = core.json.encode(body)
- if not body then
- return nil, 'invalid body : ' .. err
+ ngx.sleep(step)
+ waiting_time = waiting_time - step
+ value = nacos_dict:get_stale(key)
end
- headers['Content-Type'] = 'application/json'
- end
-
- local httpc = http.new()
- local timeout = local_conf.discovery.nacos.timeout
- local connect_timeout = timeout.connect
- local send_timeout = timeout.send
- local read_timeout = timeout.read
- log.info('connect_timeout:', connect_timeout, ', send_timeout:',
send_timeout,
- ', read_timeout:', read_timeout)
- httpc:set_timeouts(connect_timeout, send_timeout, read_timeout)
- local res, err = httpc:request_uri(url, {
- method = method,
- headers = headers,
- body = body,
- ssl_verify = true,
- })
- if not res then
- return nil, err
- end
-
- if not res.body or res.status ~= 200 then
- return nil, 'status = ' .. res.status
- end
-
- local json_str = res.body
- local data, err = core.json.decode(json_str)
- if not data then
- return nil, err
- end
- return data
-end
-
-
-local function get_url(request_uri, path)
- return request(request_uri, path, nil, 'GET', nil)
-end
-
-
-local function post_url(request_uri, path, body)
- return request(request_uri, path, body, 'POST', nil)
-end
-
-
-local function get_token_param(base_uri, username, password)
- if not username or not password then
- return ''
- end
-
- local args = { username = username, password = password}
- local data, err = post_url(base_uri, auth_path .. '?' ..
ngx.encode_args(args), nil)
- if err then
- log.error('nacos login fail:', username, ' ', password, ' desc:', err)
- return nil, err
- end
- return '&accessToken=' .. data.accessToken
-end
-
-
-local function get_namespace_param(namespace_id)
- local param = ''
- if namespace_id then
- local args = {namespaceId = namespace_id}
- param = '&' .. ngx.encode_args(args)
end
- return param
-end
-
-
-local function get_group_name_param(group_name)
- local param = ''
- if group_name then
- local args = {groupName = group_name}
- param = '&' .. ngx.encode_args(args)
- end
- return param
-end
-
-
-local function get_signed_param(group_name, service_name)
- local param = ''
- if access_key ~= '' and secret_key ~= '' then
- local str_to_sign = ngx.now() * 1000 .. '@@' .. group_name .. '@@' ..
service_name
- local args = {
- ak = access_key,
- data = str_to_sign,
- signature = ngx.encode_base64(ngx.hmac_sha1(secret_key,
str_to_sign))
- }
- param = '&' .. ngx.encode_args(args)
+ if not value then
+ core.log.error("nacos service not found: ", service_name)
+ return nodes
end
- return param
-end
+ nodes = core.json.decode(value)
-local function get_base_uri()
- local host = local_conf.discovery.nacos.host
- -- TODO Add health check to get healthy nodes.
- local url = host[math_random(#host)]
- local auth_idx = core.string.rfind_char(url, '@')
- local username, password
- if auth_idx then
- local protocol_idx = str_find(url, '://')
- local protocol = string_sub(url, 1, protocol_idx + 2)
- local user_and_password = string_sub(url, protocol_idx + 3, auth_idx -
1)
- local arr = ngx_re.split(user_and_password, ':')
- if #arr == 2 then
- username = arr[1]
- password = arr[2]
+ local res = {}
+ for _, node in ipairs(nodes) do
+ if discovery_args then
+ if utils.match_metdata(node.metadata, discovery_args.metadata) then
+ core.table.insert(res, node)
+ end
+ else
+ core.table.insert(res, node)
end
- local other = string_sub(url, auth_idx + 1)
- url = protocol .. other
end
- if local_conf.discovery.nacos.prefix then
- url = url .. local_conf.discovery.nacos.prefix
- end
-
- if str_byte(url, #url) ~= str_byte('/') then
- url = url .. '/'
- end
-
- return url, username, password
+ core.log.info("nacos service_name: ", service_name, " nodes: ",
core.json.encode(res))
+ return res
end
-
-local function de_duplication(services, namespace_id, group_name,
service_name, scheme)
- for _, service in ipairs(services) do
- if service.namespace_id == namespace_id and service.group_name ==
group_name
- and service.service_name == service_name and service.scheme ==
scheme then
- return true
- end
- end
- return false
+local function generate_new_config_from_old(discovery_conf)
+ local config = {
+ id = OLD_CONFIG_ID,
+ hosts = discovery_conf.host,
+ prefix = discovery_conf.prefix,
+ fetch_interval = discovery_conf.fetch_interval,
+ auth = {
+ access_key = discovery_conf.access_key,
+ secret_key = discovery_conf.secret_key,
+ },
+ default_weight = discovery_conf.weight,
+ timeout = discovery_conf.timeout,
+ old_conf = true
+ }
+ return {config}
end
+function _M.init_worker()
+ local local_conf = require("apisix.core.config_local").local_conf(true)
+ local discovery_conf = local_conf.discovery and local_conf.discovery.nacos
or {}
-local function iter_and_add_service(services, values)
- if not values then
+ if process.type() ~= "privileged agent" then
return
end
- for _, value in core.config_util.iterate_values(values) do
- local conf = value.value
- if not conf then
- goto CONTINUE
- end
-
- local up
- if conf.upstream then
- up = conf.upstream
- else
- up = conf
- end
-
- local namespace_id = (up.discovery_args and
up.discovery_args.namespace_id)
- or default_namespace_id
-
- local group_name = (up.discovery_args and up.discovery_args.group_name)
- or default_group_name
+ local keep = {}
+ -- support old way
+ if discovery_conf.host then
+ discovery_conf = generate_new_config_from_old(discovery_conf)
+ end
+ for _, val in ipairs(discovery_conf) do
+ local id = val.id
+ local version = ngx.md5(core.json.encode(val, true))
+ keep[id] = true
- local dup = de_duplication(services, namespace_id, group_name,
- up.service_name, up.scheme)
- if dup then
+ -- The nacos config has not been changed.
+ if nacos_clients[id] and nacos_clients[id].version == version then
goto CONTINUE
end
- if up.discovery_type == 'nacos' then
- core.table.insert(services, {
- service_name = up.service_name,
- namespace_id = namespace_id,
- group_name = group_name,
- scheme = up.scheme,
- })
+ if nacos_clients[id] then
+ nacos_clients[id]:stop()
end
+ local new_client = nacos_factory.new(val)
+ new_client:start()
+ nacos_clients[id] = new_client
+
::CONTINUE::
end
-end
-
-local function get_nacos_services()
- local services = {}
- -- here we use lazy load to work around circle dependency
- local get_upstreams = require('apisix.upstream').upstreams
- local get_routes = require('apisix.router').http_routes
- local get_stream_routes = require('apisix.router').stream_routes
- local get_services = require('apisix.http.service').services
- local values = get_upstreams()
- iter_and_add_service(services, values)
- values = get_routes()
- iter_and_add_service(services, values)
- values = get_services()
- iter_and_add_service(services, values)
- values = get_stream_routes()
- iter_and_add_service(services, values)
- return services
+ for id, client in pairs(nacos_clients) do
+ -- The nacos config has been deleted.
+ if not keep[client.id] then
+ client:stop()
+ nacos_clients[id] = nil
+ end
+ end
end
-local function is_grpc(scheme)
- if scheme == 'grpc' or scheme == 'grpcs' then
- return true
- end
- return false
+-- Now we use control plane to list the services
+function _M.list_all_services()
+ return {}
end
-local function fetch_full_registry(premature)
- if premature then
- return
+function _M.get_health_checkers()
+ local result = core.table.new(0, 4)
+ if nacos_clients == nil then
+ return result
end
- local up_apps = {}
- local base_uri, username, password = get_base_uri()
- local token_param, err = get_token_param(base_uri, username, password)
- if err then
- log.error('get_token_param error:', err)
- if not applications then
- applications = up_apps
+ for id in pairs(nacos_clients) do
+ local health_check = require("resty.healthcheck")
+ local list = health_check.get_target_list(id, "nacos")
+ if list then
+ result[id] = list
end
- return
end
- local infos = get_nacos_services()
- if #infos == 0 then
- applications = up_apps
- return
- end
+ return result
+end
- for _, service_info in ipairs(infos) do
- local data, err
- local namespace_id = service_info.namespace_id
- local group_name = service_info.group_name
- local scheme = service_info.scheme or ''
- local namespace_param = get_namespace_param(service_info.namespace_id)
- local group_name_param = get_group_name_param(service_info.group_name)
- local signature_param = get_signed_param(service_info.group_name,
service_info.service_name)
- local query_path = instance_list_path .. service_info.service_name
- .. token_param .. namespace_param ..
group_name_param
- .. signature_param
- data, err = get_url(base_uri, query_path)
- if err then
- log.error('get_url:', query_path, ' err:', err)
- goto CONTINUE
- end
+local cjson = require "cjson"
- if not up_apps[namespace_id] then
- up_apps[namespace_id] = {}
- end
+function _M.dump_data()
+ local applications = {}
+ local keys = nacos_dict:get_keys() or {}
- if not up_apps[namespace_id][group_name] then
- up_apps[namespace_id][group_name] = {}
+ for _, key in ipairs(keys) do
+ local parts = {}
+ for part in key:gmatch("[^/]+") do
+ table.insert(parts, part)
end
- for _, host in ipairs(data.hosts) do
- local nodes = up_apps[namespace_id]
- [group_name][service_info.service_name]
- if not nodes then
- nodes = {}
- up_apps[namespace_id]
- [group_name][service_info.service_name] = nodes
- end
-
- local node = {
- host = host.ip,
- port = host.port,
- weight = host.weight or default_weight,
- }
-
- -- docs:
https://github.com/yidongnan/grpc-spring-boot-starter/pull/496
- if is_grpc(scheme) and host.metadata and host.metadata.gRPC_port
then
- node.port = host.metadata.gRPC_port
+ if #parts == 4 then
+ local id, namespace_id,
+ group_name, service_name = parts[1], parts[2], parts[3],
parts[4]
+ local data_str = nacos_dict:get(key)
+
+ if data_str and data_str ~= "" then
+ -- Decode JSON string to Lua table
+ local success, data = pcall(cjson.decode, data_str)
+ if success then
+ applications[id] = applications[id] or {}
+ applications[id][namespace_id] =
applications[id][namespace_id] or {}
+ applications[id][namespace_id][group_name] =
applications[id]
+
[namespace_id][group_name] or {}
+ applications[id][namespace_id][group_name][service_name] =
data
+ else
+ ngx.log(ngx.ERR, "failed to decode data for key ", key, ":
", data)
+ end
end
-
- core.table.insert(nodes, node)
end
-
- ::CONTINUE::
- end
- local new_apps_md5sum = ngx.md5(core.json.encode(up_apps))
- local old_apps_md5sum = ngx.md5(core.json.encode(applications))
- if new_apps_md5sum == old_apps_md5sum then
- return
- end
- applications = up_apps
- local ok, err = events:post(events_list._source, events_list.updating,
- applications)
- if not ok then
- log.error("post_event failure with ", events_list._source,
- ", update application error: ", err)
- end
-end
-
-
-function _M.nodes(service_name, discovery_args)
- local namespace_id = discovery_args and
- discovery_args.namespace_id or default_namespace_id
- local group_name = discovery_args
- and discovery_args.group_name or default_group_name
-
- local logged = false
- -- maximum waiting time: 5 seconds
- local waiting_time = 5
- local step = 0.1
- while not applications and waiting_time > 0 do
- if not logged then
- log.warn('wait init')
- logged = true
- end
- ngx.sleep(step)
- waiting_time = waiting_time - step
- end
-
- if not applications or not applications[namespace_id]
- or not applications[namespace_id][group_name]
- then
- return nil
end
- return applications[namespace_id][group_name][service_name]
-end
-
-function _M.init_worker()
- events = require("apisix.events")
- events_list = events:event_list("discovery_nacos_update_application",
- "updating")
-
- if 0 ~= ngx.worker.id() then
- events:register(discovery_nacos_callback, events_list._source,
- events_list.updating)
- return
- end
-
- default_weight = local_conf.discovery.nacos.weight
- log.info('default_weight:', default_weight)
- local fetch_interval = local_conf.discovery.nacos.fetch_interval
- log.info('fetch_interval:', fetch_interval)
- access_key = local_conf.discovery.nacos.access_key
- secret_key = local_conf.discovery.nacos.secret_key
- ngx_timer_at(0, fetch_full_registry)
- ngx_timer_every(fetch_interval, fetch_full_registry)
-end
-
-
-function _M.dump_data()
- return {config = local_conf.discovery.nacos, services = applications or {}}
+ return {
+ config = local_conf.discovery.nacos,
Review Comment:
It seems that printing configurations are not secure.
If users are responsible for maintaining APISIX deployments, they can view
these in the configuration files. The Control API itself should also be used by
maintenance personnel, but we know that security controls on APIs are always
difficult to configure correctly.
--
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: notifications-unsubscr...@apisix.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
