-
Notifications
You must be signed in to change notification settings - Fork 2.5k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
feat: add nacos support #3820
feat: add nacos support #3820
Changes from 57 commits
baab627
e053355
5b14149
613c221
f0b0942
221ee3a
8e3e6a5
69055a0
e412a70
9b48317
d8f819b
46fe9fd
9e51570
4a0dac6
bbc3f1f
b153240
962be01
26de944
c1e6f00
95ab302
63eba52
df1d54f
3ea6a66
d1f3bef
ec0c16d
397a197
dcb76c2
14de476
b53f4e9
dccd644
33e7c2e
ddff0f5
3ebf626
147c088
d55effc
ea93f5e
922ac50
5a91e81
0d1ef67
7398fbb
74f5512
b7d356c
4fa00d7
638cdc9
e24d641
dce8163
04a42a2
846fc86
952aa04
5ea5414
e1a9f13
6bbcaf1
f150a72
9ac15df
c97c0a2
4936b19
fb69097
6171c2e
83d7500
5af1d9e
69408be
dfd6921
8b196f3
09767bd
660d2b6
407fb42
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,315 @@ | ||
-- | ||
-- 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 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 error = error | ||
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 default_weight | ||
local applications | ||
local auth_path = "auth/login" | ||
local instance_list_path = "ns/instance/list?healthyOnly=true&serviceName=" | ||
|
||
local schema = { | ||
type = "object", | ||
properties = { | ||
host = { | ||
type = "array", | ||
minItems = 1, | ||
items = { | ||
type = "string", | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. add a pattern to limit the string to the host format |
||
}, | ||
}, | ||
fetch_interval = {type = "integer", minimum = 1, default = 30}, | ||
prefix = {type = "string", default = "/nacos/v1/"}, | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. ditto |
||
weight = {type = "integer", minimum = 1, default = 100}, | ||
timeout = { | ||
type = "object", | ||
properties = { | ||
connect = {type = "integer", minimum = 1, default = 2000}, | ||
send = {type = "integer", minimum = 1, default = 2000}, | ||
read = {type = "integer", minimum = 1, default = 5000}, | ||
}, | ||
default = { | ||
connect = 2000, | ||
send = 2000, | ||
read = 5000, | ||
} | ||
}, | ||
}, | ||
required = {"host"} | ||
} | ||
|
||
|
||
local _M = { | ||
version = 0.1, | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. this version is useless, we can drop it |
||
} | ||
|
||
|
||
local function request(request_uri, path, body, method, basic_auth) | ||
local url = request_uri .. path | ||
log.info("request url:", url) | ||
local headers = core.table.new(0, 0) | ||
headers['Accept'] = 'application/json' | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. for the string object, I think we use
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. and please fix the similar points |
||
|
||
if basic_auth then | ||
headers['Authorization'] = basic_auth | ||
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 | ||
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 username and password then | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. better style: if not username or not password then
return ""
end
... |
||
local data, err = post_url(base_uri, auth_path .. "?username=" .. username | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think we should call ngx.encode_args for |
||
.. "&password=" .. password, nil) | ||
if err then | ||
log.error("nacos login fail:", username, " ", password, " desc:", err) | ||
return nil, err | ||
end | ||
return "&accessToken=" .. data.accessToken | ||
end | ||
return "" | ||
end | ||
|
||
|
||
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 = str_find(url, "@") | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. we'd better use the plain text model, better performance. https://stackoverflow.com/questions/18973019/is-there-a-lua-string-find-without-pattern |
||
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 | ||
|
||
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 | ||
end | ||
|
||
|
||
local function iter_and_add_service(services, values) | ||
if not values then | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. bad indentation |
||
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 | ||
|
||
if up.discovery_type == "nacos" then | ||
core.table.insert(services, up.service_name) | ||
end | ||
::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_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) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. By the way,Are there 'discovery_type' config in service and route? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Those can have embedded upstream field. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I see,thanks. |
||
values = get_services() | ||
iter_and_add_service(services, values) | ||
return services | ||
end | ||
|
||
|
||
local function fetch_full_registry(premature) | ||
if premature then | ||
return | ||
end | ||
|
||
local up_apps = core.table.new(0, 0) | ||
local base_uri, username, password = get_base_uri() | ||
local token_param, err = get_token_param(base_uri, username, password) | ||
if err then | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. we need to writhe this error message to error log |
||
log.error("get_token_param error:", err) | ||
if not applications then | ||
applications = up_apps | ||
end | ||
return | ||
end | ||
|
||
local infos = get_nacos_services() | ||
if #infos == 0 then | ||
applications = up_apps | ||
return | ||
end | ||
|
||
local data, err | ||
for _, service_name in ipairs(infos) do | ||
data, err = get_url(base_uri, instance_list_path .. service_name .. token_param) | ||
spacewander marked this conversation as resolved.
Show resolved
Hide resolved
|
||
if err then | ||
log.error("get_url:", instance_list_path, " err:", err) | ||
if not applications then | ||
applications = up_apps | ||
end | ||
return | ||
end | ||
|
||
for _, host in ipairs(data.hosts) do | ||
local nodes = up_apps[service_name] | ||
if not nodes then | ||
nodes = core.table.new(0, 0) | ||
up_apps[service_name] = nodes | ||
end | ||
core.table.insert(nodes, { | ||
host = host.ip, | ||
port = host.port, | ||
weight = host.weight or default_weight, | ||
}) | ||
end | ||
end | ||
applications = up_apps | ||
end | ||
|
||
|
||
function _M.nodes(service_name) | ||
local logged = false | ||
while not applications do | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think we need to set a maximum waiting time , eg: 5 seconds |
||
if not logged then | ||
log.warn('wait init') | ||
logged = true | ||
end | ||
ngx.sleep(0.1) | ||
end | ||
return applications[service_name] | ||
end | ||
|
||
|
||
function _M.init_worker() | ||
if not local_conf.discovery.nacos or | ||
not local_conf.discovery.nacos.host or #local_conf.discovery.nacos.host == 0 then | ||
error("do not set nacos.host") | ||
return | ||
end | ||
|
||
local ok, err = core.schema.check(schema, local_conf.discovery.nacos) | ||
if not ok then | ||
error("invalid nacos configuration: " .. err) | ||
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) | ||
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 {}} | ||
end | ||
|
||
|
||
return _M |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We can initialize
applications = {}
so that we can save many codes like:There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If use applications = {},run test could fail cause maybe call _M.nodes method before finish init.