Revolyssup commented on code in PR #12364:
URL: https://github.com/apache/apisix/pull/12364#discussion_r2168129288
##########
apisix/plugins/elasticsearch-logger.lua:
##########
@@ -124,32 +124,94 @@ function _M.check_schema(conf, schema_type)
if schema_type == core.schema.TYPE_METADATA then
return core.schema.check(metadata_schema, conf)
end
+
local check = {"endpoint_addrs"}
core.utils.check_https(check, conf, plugin_name)
core.utils.check_tls_bool({"ssl_verify"}, conf, plugin_name)
-
return core.schema.check(schema, conf)
end
+local function get_es_major_version(uri, conf)
+ local httpc = http.new()
+ if not httpc then
+ return nil, "failed to create http client"
+ end
+ local headers = {}
+ if conf.auth then
+ local authorization = "Basic " .. ngx.encode_base64(
+ conf.auth.username .. ":" .. conf.auth.password
+ )
+ headers["Authorization"] = authorization
+ end
+ httpc:set_timeout(conf.timeout * 1000)
+ local res, err = httpc:request_uri(uri, {
+ ssl_verify = conf.ssl_verify,
+ method = "GET",
+ headers = headers,
+ })
+ if not res then
+ return false, err
+ end
+ if res.status ~= 200 then
+ return nil, str_format("server returned status: %d, body: %s",
+ res.status, res.body or "")
+ end
+ local json_body, err = core.json.decode(res.body)
+ if not json_body then
+ return nil, "failed to decode response body: " .. err
+ end
+ if not json_body.version or not json_body.version.number then
+ return nil, "failed to get version from response body"
+ end
+
+ local major_version = json_body.version.number:match("^(%d+)%.")
+ if not major_version then
+ return nil, "invalid version format: " .. json_body.version.number
+ end
+
+ return major_version
+end
+
+
local function get_logger_entry(conf, ctx)
local entry = log_util.get_log_entry(plugin_name, conf, ctx)
- return core.json.encode({
- create = {
- _index = conf.field.index,
- _type = conf.field.type
- }
- }) .. "\n" ..
+ local body = {
+ index = {
+ _index = conf.field.index
+ }
+ }
+ -- for older version type is required
+ if conf._version == "6" or conf._version == "5" then
+ body.index._type = "_doc"
+ end
+ return core.json.encode(body) .. "\n" ..
core.json.encode(entry) .. "\n"
end
+local function set_version(conf)
+ if not conf._version then
+ local selected_endpoint_addr
+ if conf.endpoint_addr then
+ selected_endpoint_addr = conf.endpoint_addr
+ else
+ selected_endpoint_addr =
conf.endpoint_addrs[math_random(#conf.endpoint_addrs)]
+ end
+ local major_version, err =
get_es_major_version(selected_endpoint_addr, conf)
+ if err then
+ return false, str_format("failed to get Elasticsearch version:
%s", err)
+ end
+ conf._version = major_version
+ end
+end
+
local function send_to_elasticsearch(conf, entries)
local httpc, err = http.new()
if not httpc then
return false, str_format("create http error: %s", err)
end
-
+ set_version(conf)
Review Comment:
this function is run later in batch processor. But we need the version in
the log phase when entry is created. That is why we need it in access phase.
--
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
