wehub-resource-sync
404 lines
9.4 KiB
Lua
404 lines
9.4 KiB
Lua
local kong_global = require "kong.global"
|
|
local cjson = require "cjson.safe"
|
|
local grpc_tools = require "kong.tools.grpc"
|
|
local pb = require "pb"
|
|
local lpack = require "lua_pack"
|
|
local rpc_util = require "kong.runloop.plugin_servers.rpc.util"
|
|
|
|
local ngx = ngx
|
|
local kong = kong
|
|
local cjson_encode = cjson.encode
|
|
local t_unpack = table.unpack -- luacheck: ignore table
|
|
local st_pack = lpack.pack
|
|
local st_unpack = lpack.unpack
|
|
local str_find = string.find
|
|
|
|
local proto_fname = "kong/pluginsocket.proto"
|
|
local Rpc = {}
|
|
Rpc.__index = Rpc
|
|
|
|
local pb_unwrap
|
|
do
|
|
local structpb_value, structpb_list, structpb_struct
|
|
|
|
function structpb_value(v)
|
|
if type(v) ~= "table" then
|
|
return v
|
|
end
|
|
|
|
if v.list_value then
|
|
return structpb_list(v.list_value)
|
|
end
|
|
|
|
if v.struct_value then
|
|
return structpb_struct(v.struct_value)
|
|
end
|
|
|
|
return v.bool_value or v.string_value or v.number_value or v.null_value
|
|
end
|
|
|
|
function structpb_list(l)
|
|
local out = {}
|
|
if type(l) == "table" then
|
|
for i, v in ipairs(l.values or l) do
|
|
out[i] = structpb_value(v)
|
|
end
|
|
end
|
|
return out
|
|
end
|
|
|
|
function structpb_struct(struct)
|
|
if type(struct) ~= "table" then
|
|
return struct
|
|
end
|
|
|
|
local out = {}
|
|
for k, v in pairs(struct.fields or struct) do
|
|
out[k] = structpb_value(v)
|
|
end
|
|
return out
|
|
end
|
|
|
|
local function unwrap_val(d) return d.v end
|
|
|
|
pb_unwrap = {
|
|
[".google.protobuf.Empty"] = function() end,
|
|
[".google.protobuf.Value"] = structpb_value,
|
|
[".google.protobuf.ListValue"] = function(v)
|
|
return t_unpack(structpb_list(v))
|
|
end,
|
|
[".google.protobuf.Struct"] = structpb_struct,
|
|
|
|
[".kong_plugin_protocol.Bool"] = unwrap_val,
|
|
[".kong_plugin_protocol.Number"] = unwrap_val,
|
|
[".kong_plugin_protocol.Int"] = unwrap_val,
|
|
[".kong_plugin_protocol.String"] = unwrap_val,
|
|
[".kong_plugin_protocol.ByteString"] = unwrap_val,
|
|
[".kong_plugin_protocol.KV"] = function(d)
|
|
return d.k, structpb_value(d.v)
|
|
end,
|
|
[".kong_plugin_protocol.Target"] = function(d)
|
|
return d.host, d.port
|
|
end,
|
|
[".kong_plugin_protocol.ExitArgs"] = function (d)
|
|
return d.status, d.body, structpb_struct(d.headers)
|
|
end,
|
|
[".kong_plugin_protocol.ConsumerSpec"] = function (d)
|
|
return d.id, d.by_username
|
|
end,
|
|
[".kong_plugin_protocol.AuthenticateArgs"] = function (d)
|
|
return d.consumer, d.credential
|
|
end,
|
|
}
|
|
end
|
|
|
|
local pb_wrap
|
|
do
|
|
local structpb_value, structpb_list, structpb_struct
|
|
|
|
function structpb_value(v)
|
|
local t = type(v)
|
|
|
|
local bool_v = nil
|
|
if t == "boolean" then
|
|
bool_v = v
|
|
end
|
|
|
|
local list_v = nil
|
|
local struct_v = nil
|
|
|
|
if t == "table" then
|
|
if v[1] ~= nil then
|
|
list_v = structpb_list(v)
|
|
else
|
|
struct_v = structpb_struct(v)
|
|
end
|
|
end
|
|
|
|
return {
|
|
null_value = t == "nil" and 1 or nil,
|
|
bool_value = bool_v,
|
|
number_value = t == "number" and v or nil,
|
|
string_value = t == "string" and v or nil,
|
|
list_value = list_v,
|
|
struct_value = struct_v,
|
|
}
|
|
end
|
|
|
|
function structpb_list(l)
|
|
local out = {}
|
|
for i, v in ipairs(l) do
|
|
out[i] = structpb_value(v)
|
|
end
|
|
return { values = out }
|
|
end
|
|
|
|
function structpb_struct(d)
|
|
local out = {}
|
|
for k, v in pairs(d) do
|
|
out[k] = structpb_value(v)
|
|
end
|
|
return { fields = out }
|
|
end
|
|
|
|
local function wrap_val(v) return { v = v } end
|
|
|
|
pb_wrap = {
|
|
[".google.protobuf.Value"] = structpb_value,
|
|
[".google.protobuf.ListValue"] = structpb_list,
|
|
[".google.protobuf.Struct"] = structpb_struct,
|
|
|
|
[".kong_plugin_protocol.Bool"] = wrap_val,
|
|
[".kong_plugin_protocol.Number"] = wrap_val,
|
|
[".kong_plugin_protocol.Int"] = wrap_val,
|
|
[".kong_plugin_protocol.String"] = wrap_val,
|
|
[".kong_plugin_protocol.ByteString"] = wrap_val,
|
|
[".kong_plugin_protocol.RawBodyResult"] = function(v, err)
|
|
if type(v) == "string" then
|
|
return { content = v }
|
|
end
|
|
|
|
local path = ngx.req.get_body_file()
|
|
if path then
|
|
return { body_filepath = path }
|
|
end
|
|
|
|
return { error = err or "Can't read request body" }
|
|
end,
|
|
--[".kong_plugin_protocol.MemoryStats"] = - function(v)
|
|
-- return {
|
|
-- lua_shared_dicts = {
|
|
--
|
|
-- }
|
|
-- }
|
|
--end
|
|
}
|
|
end
|
|
|
|
local function load_service(pdk)
|
|
local p = grpc_tools.new()
|
|
local protoc_instance = p.protoc_instance
|
|
|
|
protoc_instance:loadfile(proto_fname)
|
|
local parsed = protoc_instance:parsefile(proto_fname)
|
|
|
|
local service = {}
|
|
for i, s in ipairs(parsed.service) do
|
|
for j, m in ipairs(s.method) do
|
|
local method_name = s.name .. '.' .. m.name
|
|
local lower_name = m.options and m.options.MethodName
|
|
or method_name
|
|
:gsub('_', '.')
|
|
:gsub('([a-z]%d*)([A-Z])', '%1_%2')
|
|
:lower()
|
|
|
|
service[lower_name] = {
|
|
method_name = method_name,
|
|
method = rpc_util.index_table(pdk, lower_name),
|
|
input_type = m.input_type,
|
|
output_type = m.output_type,
|
|
}
|
|
--print(("service[%q] = %s"):format(lower_name, pp(service[lower_name])))
|
|
end
|
|
end
|
|
|
|
return service
|
|
end
|
|
|
|
|
|
|
|
local rpc_service
|
|
|
|
|
|
local function identity_function(x)
|
|
return x
|
|
end
|
|
|
|
|
|
local function call_pdk(pdk, method_name, arg)
|
|
local method = rpc_service[method_name]
|
|
if not method then
|
|
return nil, ("method %q not found"):format(method_name)
|
|
end
|
|
|
|
local saved = pdk.get_saved_req_data()
|
|
if saved and saved.plugin_name then
|
|
kong_global.set_namespaced_log(kong, saved.plugin_name)
|
|
end
|
|
|
|
arg = assert(pb.decode(method.input_type, arg))
|
|
local unwrap = pb_unwrap[method.input_type] or identity_function
|
|
local wrap = pb_wrap[method.output_type] or identity_function
|
|
|
|
local reply = wrap(method.method(unwrap(arg)))
|
|
if reply == nil then
|
|
--kong.log.debug("no reply")
|
|
return ""
|
|
end
|
|
|
|
reply = assert(pb.encode(method.output_type, reply))
|
|
|
|
return reply
|
|
end
|
|
|
|
|
|
local function read_frame(c)
|
|
--kong.log.debug("reading frame...")
|
|
local msg, err = c:receive(4) -- uint32
|
|
if not msg then
|
|
return nil, err
|
|
end
|
|
local _, msg_len = st_unpack(msg, "I")
|
|
--kong.log.debug("len: ", msg_len)
|
|
|
|
msg, err = c:receive(msg_len)
|
|
if not msg then
|
|
return nil, err
|
|
end
|
|
--kong.log.debug(("data: %q"):format(msg))
|
|
|
|
return msg, nil
|
|
end
|
|
|
|
local function write_frame(c, msg)
|
|
assert (c:send(st_pack("I", #msg)))
|
|
assert (c:send(msg))
|
|
end
|
|
|
|
function Rpc:call(method, data, do_bridge_loop)
|
|
self.msg_id = self.msg_id + 1
|
|
local msg_id = self.msg_id
|
|
local c, err = ngx.socket.connect("unix:" .. self.plugin.server_def.socket)
|
|
if not c then
|
|
kong.log.err("trying to connect: ", err)
|
|
return nil, err
|
|
end
|
|
|
|
msg_id = msg_id + 1
|
|
--kong.log.debug("will encode: ", pp{sequence = msg_id, [method] = data})
|
|
local msg, err = assert(pb.encode(".kong_plugin_protocol.RpcCall", { -- luacheck: ignore err
|
|
sequence = msg_id,
|
|
[method] = data,
|
|
}))
|
|
--kong.log.debug("encoded len: ", #msg)
|
|
assert (c:send(st_pack("I", #msg)))
|
|
assert (c:send(msg))
|
|
|
|
while do_bridge_loop do
|
|
local method_name
|
|
method_name, err = read_frame(c)
|
|
if not method_name then
|
|
return nil, err
|
|
end
|
|
if method_name == "" then
|
|
break
|
|
end
|
|
|
|
--kong.log.debug(("pdk method: %q (%d)"):format(method_name, #method_name))
|
|
|
|
local args
|
|
args, err = read_frame(c)
|
|
if not args then
|
|
return nil, err
|
|
end
|
|
|
|
local reply
|
|
reply, err = call_pdk(self.plugin.exposed_api, method_name, args)
|
|
if not reply then
|
|
return nil, err
|
|
end
|
|
|
|
err = write_frame(c, reply)
|
|
if err then
|
|
return nil, err
|
|
end
|
|
end
|
|
|
|
msg, err = read_frame(c)
|
|
if not msg then
|
|
return nil, err
|
|
end
|
|
c:setkeepalive()
|
|
|
|
msg = assert(pb.decode(".kong_plugin_protocol.RpcReturn", msg))
|
|
--kong.log.debug("decoded: "..pp(msg))
|
|
assert(msg.sequence == msg_id)
|
|
|
|
return msg
|
|
end
|
|
|
|
|
|
function Rpc:call_start_instance(plugin_name, conf)
|
|
local status, err = self:call("cmd_start_instance", {
|
|
name = plugin_name,
|
|
config = cjson_encode(conf)
|
|
})
|
|
|
|
if status == nil then
|
|
return nil, err
|
|
end
|
|
|
|
kong.log.debug("started plugin server: seq ", conf.__seq__, ", worker ", ngx.worker.id(), ", instance id ",
|
|
status.instance_status.instance_id)
|
|
|
|
return {
|
|
id = status.instance_status.instance_id,
|
|
conf = conf,
|
|
seq = conf.__seq__,
|
|
Config = status.instance_status.config,
|
|
rpc = self,
|
|
}
|
|
end
|
|
|
|
function Rpc:call_close_instance(instance_id)
|
|
return self:call("cmd_close_instance", {
|
|
instance_id = instance_id,
|
|
})
|
|
end
|
|
|
|
|
|
function Rpc:handle_event(conf, phase)
|
|
local plugin_name = self.plugin.name
|
|
|
|
local instance_id, err = self.plugin.instance_callbacks.get_instance_id(self.plugin, conf)
|
|
local res
|
|
if not err then
|
|
res, err = self:call("cmd_handle_event", {
|
|
instance_id = instance_id,
|
|
event_name = phase,
|
|
}, true)
|
|
end
|
|
|
|
if not res or res == "" then
|
|
local err_lowered = err and err:lower() or "unknown error"
|
|
|
|
if str_find(err_lowered, "no plugin instance", nil, true)
|
|
or str_find(err_lowered, "closed", nil, true) then
|
|
self.plugin.instance_callbacks.reset_instance(plugin_name, conf)
|
|
kong.log.warn(err)
|
|
return self:handle_event(conf, phase)
|
|
|
|
else
|
|
kong.log.err("pluginserver error: ", err or "unknown error")
|
|
kong.response.error(500)
|
|
end
|
|
end
|
|
end
|
|
|
|
local function new(plugin)
|
|
if not rpc_service then
|
|
rpc_service = load_service(plugin.exposed_api)
|
|
end
|
|
|
|
local self = setmetatable({
|
|
msg_id = 0,
|
|
plugin = plugin,
|
|
}, Rpc)
|
|
|
|
return self
|
|
end
|
|
|
|
return {
|
|
new = new,
|
|
}
|