fix: syslog plugin doesn't work (apache#9425)

Makefile

@@ -337,9 +337,6 @@ install: runtime
 	$(ENV_INSTALL) -d $(ENV_INST_LUADIR)/apisix/plugins/serverless
 	$(ENV_INSTALL) apisix/plugins/serverless/*.lua $(ENV_INST_LUADIR)/apisix/plugins/serverless/
 
-	$(ENV_INSTALL) -d $(ENV_INST_LUADIR)/apisix/plugins/slslog
-	$(ENV_INSTALL) apisix/plugins/slslog/*.lua $(ENV_INST_LUADIR)/apisix/plugins/slslog/
-
 	$(ENV_INSTALL) -d $(ENV_INST_LUADIR)/apisix/plugins/syslog
 	$(ENV_INSTALL) apisix/plugins/syslog/*.lua $(ENV_INST_LUADIR)/apisix/plugins/syslog/
 

apisix/plugins/sls-logger.lua

@@ -21,7 +21,7 @@ local bp_manager_mod = require("apisix.utils.batch-processor-manager")
 
 local plugin_name = "sls-logger"
 local ngx = ngx
-local rf5424 = require("apisix.plugins.slslog.rfc5424")
+local rf5424 = require("apisix.utils.rfc5424")
 local tcp = ngx.socket.tcp
 local tostring = tostring
 local ipairs = ipairs
@@ -138,9 +138,14 @@ function _M.log(conf, ctx)
         return
     end
 
-    local rf5424_data = rf5424.encode("SYSLOG", "INFO", ctx.var.host,"apisix",
-                                      ctx.var.pid, conf.project, conf.logstore,
-                                      conf.access_key_id, conf.access_key_secret, json_str)
+    local structured_data = {
+        {name = "project", value = conf.project},
+        {name = "logstore", value = conf.logstore},
+        {name = "access-key-id", value = conf.access_key_id},
+        {name = "access-key-secret", value = conf.access_key_secret},
+    }
+    local rf5424_data = rf5424.encode("SYSLOG", "INFO", ctx.var.host, "apisix",
+                                      ctx.var.pid, json_str, structured_data)
 
     local process_context = {
         data = rf5424_data,

apisix/plugins/syslog/init.lua

@@ -18,6 +18,10 @@
 local core = require("apisix.core")
 local bp_manager_mod = require("apisix.utils.batch-processor-manager")
 local logger_socket = require("resty.logger.socket")
+local rfc5424 = require("apisix.utils.rfc5424")
+local ipairs = ipairs
+local table_insert = core.table.insert
+local table_concat = core.table.concat
 
 local batch_processor_manager = bp_manager_mod.new("sys logger")
 
@@ -63,7 +67,8 @@ local function send_syslog_data(conf, log_message, api_ctx)
     end
 
     -- reuse the logger object
-    local ok, err = logger:log(core.json.encode(log_message))
+    local ok, err = logger:log(log_message)
+
     if not ok then
         res = false
         err_msg = "failed to log message" .. err
@@ -75,28 +80,32 @@ end
 
 -- called in log phase of APISIX
 function _M.push_entry(conf, ctx, entry)
-    if batch_processor_manager:add_entry(conf, entry) then
+    local json_str, err = core.json.encode(entry)
+    if not json_str then
+        core.log.error('error occurred while encoding the data: ', err)
+        return
+    end
+
+    local rfc5424_data = rfc5424.encode("SYSLOG", "INFO", ctx.var.host,
+                                "apisix", ctx.var.pid, json_str)
+
+    if batch_processor_manager:add_entry(conf, rfc5424_data) then
         return
     end
 
     -- Generate a function to be executed by the batch processor
     local cp_ctx = core.table.clone(ctx)
-    local func = function(entries, batch_max_size)
-        local data, err
-        if batch_max_size == 1 then
-            data, err = core.json.encode(entries[1]) -- encode as single {}
-        else
-            data, err = core.json.encode(entries) -- encode as array [{}]
-        end
-
-        if not data then
-            return false, 'error occurred while encoding the data: ' .. err
+    local func = function(entries)
+        local items = {}
+        for _, e in ipairs(entries) do
+            table_insert(items, e)
+            core.log.debug("buffered logs:", e)
         end
 
-        return send_syslog_data(conf, data, cp_ctx)
+        return send_syslog_data(conf, table_concat(items), cp_ctx)
     end
 
-    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
+    batch_processor_manager:add_entry_to_new_processor(conf, rfc5424_data, ctx, func)
 end
 
 

apisix/plugins/slslog/rfc5424.lua → apisix/utils/rfc5424.lua

@@ -79,12 +79,13 @@ local Severity = {
 }
 
 local log_util = require("apisix.utils.log-util")
-
+local ipairs = ipairs
+local str_format = string.format
 
 local _M = { version = 0.1 }
 
-function _M.encode(facility, severity, hostname, appname, pid, project,
-                   logstore, access_key_id, access_key_secret, msg)
+
+function _M.encode(facility, severity, hostname, appname, pid, msg, structured_data)
     local pri = (Facility[facility] * 8 + Severity[severity])
     local t = log_util.get_rfc3339_zulu_timestamp()
     if not hostname then
@@ -95,10 +96,19 @@ function _M.encode(facility, severity, hostname, appname, pid, project,
         appname = "-"
     end
 
-    return "<" .. pri .. ">1 " .. t .. " " .. hostname .. " " .. appname .. " " .. pid
-           .. " - [logservice project=\"" .. project .. "\" logstore=\"" .. logstore
-           .. "\" access-key-id=\"" .. access_key_id .. "\" access-key-secret=\""
-           .. access_key_secret .. "\"] " .. msg .. "\n"
+    local structured_data_str = "-"
+
+    if structured_data then
+        structured_data_str = "[logservice"
+        for _, sd_param in ipairs(structured_data) do
+            structured_data_str = structured_data_str .. " " .. sd_param.name
+                                  .. "=\"" .. sd_param.value .. "\""
+        end
+        structured_data_str = structured_data_str .. "]"
+    end
+
+    return str_format("<%d>1 %s %s %s %d - %s %s\n", pri, t, hostname,
+                    appname, pid, structured_data_str, msg)
 end
 
 return _M

ci/pod/docker-compose.plugin.yml

@@ -319,6 +319,20 @@ services:
     ports:
       - '8888:8080'
 
+  vector:
+    image: timberio/vector:0.29.1-debian
+    container_name: vector
+    volumes:
+      - ./ci/pod/vector:/etc/vector/
+      - ./t/certs:/certs
+    ports:
+      - '3000:3000'
+      - '43000:43000'
+      - '5140:5140'
+      - '5150:5150/udp'
+    networks:
+      vector_net:
+
 networks:
   apisix_net:
   kafka_net:

ci/pod/vector/vector.toml

@@ -0,0 +1,75 @@
+#
+# 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.
+
+[sources.log-from-tcp]
+type = "socket"
+address = "0.0.0.0:3000"
+host_key = "host"
+mode = "tcp"
+port_key = "port"
+shutdown_timeout_secs = 30
+socket_file_mode = 511
+
+[sources.log-from-tls]
+type = "socket"
+address = "0.0.0.0:43000"
+host_key = "host"
+mode = "tcp"
+port_key = "port"
+tls.enabled = true
+tls.verify = true
+tls.ca_file = "/certs/vector_logs_ca.crt"
+tls.crt_file = "/certs/vector_logs_server.crt"
+tls.key_file = "/certs/vector_logs_server.key"
+
+[sources.log-from-syslog-tcp]
+type = "syslog"
+address = "0.0.0.0:5140"
+mode = "tcp"
+
+[sources.log-from-syslog-udp]
+type = "syslog"
+address = "0.0.0.0:5150"
+mode = "udp"
+
+[sinks.log-2-console]
+inputs = [ "log-from-tcp",  "log-from-tls", "log-from-syslog-tcp", "log-from-syslog-udp" ]
+type = "console"
+encoding.codec = "json"
+
+[sinks.log-2-tcp-file]
+inputs = [ "log-from-tcp" ]
+type = "file"
+encoding.codec = "text"
+path = "/etc/vector/tcp.log"
+
+[sinks.tls-log-2-file]
+inputs = [ "log-from-tls" ]
+type = "file"
+encoding.codec = "json"
+path = "/etc/vector/tls-datas.log"
+
+[sinks.log-2-syslog-tcp-file]
+inputs = [ "log-from-syslog-tcp" ]
+type = "file"
+encoding.codec = "text"
+path = "/etc/vector/syslog-tcp.log"
+
+[sinks.log-2-syslog-udp-file]
+inputs = [ "log-from-syslog-udp" ]
+type = "file"
+encoding.codec = "text"
+path = "/etc/vector/syslog-udp.log"

t/plugin/sls-logger.t

@@ -173,16 +173,20 @@ hello world
             end
 
             math.randomseed(os.time())
-            local rfc5424 = require("apisix.plugins.slslog.rfc5424")
+            local rfc5424 = require("apisix.utils.rfc5424")
             local m = 0
             -- because the millisecond value obtained by `ngx.now` may be `0`
             -- it is executed multiple times to ensure the accuracy of the test
             for i = 1, 5 do
                 ngx.sleep(string.format("%0.3f", math.random()))
+                local structured_data = {
+                    {name = "project", value = "apisix.apache.org"},
+                    {name = "logstore", value = "apisix.apache.org"},
+                    {name = "access-key-id", value = "apisix.sls.logger"},
+                    {name = "access-key-secret", value = "BD274822-96AA-4DA6-90EC-15940FB24444"}
+                }
                 local log_entry = rfc5424.encode("SYSLOG", "INFO", "localhost", "apisix",
-                                                 123456, "apisix.apache.org", "apisix.apache.log",
-                                                 "apisix.sls.logger", "BD274822-96AA-4DA6-90EC-15940FB24444",
-                                                 "hello world")
+                                                 123456, "hello world", structured_data)
                 m = get_syslog_timestamp_millisecond(log_entry) + m
             end
 
@@ -226,15 +230,13 @@ passed
 === TEST 9: access
 --- extra_init_by_lua
     local json = require("toolkit.json")
-    local rfc5424 = require("apisix.plugins.slslog.rfc5424")
+    local rfc5424 = require("apisix.utils.rfc5424")
     local old_f = rfc5424.encode
-    rfc5424.encode = function(facility, severity, hostname, appname, pid, project,
-                   logstore, access_key_id, access_key_secret, msg)
+    rfc5424.encode = function(facility, severity, hostname, appname, pid, msg, structured_data)
         local r = json.decode(msg)
         assert(r.client_ip == "127.0.0.1", r.client_ip)
         assert(r.host == "localhost", r.host)
-        return old_f(facility, severity, hostname, appname, pid, project,
-                     logstore, access_key_id, access_key_secret, msg)
+        return old_f(facility, severity, hostname, appname, pid, msg, structured_data)
     end
 --- request
 GET /hello
@@ -372,14 +374,12 @@ passed
 === TEST 13: access
 --- extra_init_by_lua
     local json = require("toolkit.json")
-    local rfc5424 = require("apisix.plugins.slslog.rfc5424")
+    local rfc5424 = require("apisix.utils.rfc5424")
     local old_f = rfc5424.encode
-    rfc5424.encode = function(facility, severity, hostname, appname, pid, project,
-                   logstore, access_key_id, access_key_secret, msg)
+    rfc5424.encode = function(facility, severity, hostname, appname, pid, msg, structured_data)
         local r = json.decode(msg)
         assert(r.vip == "127.0.0.1", r.vip)
-        return old_f(facility, severity, hostname, appname, pid, project,
-                     logstore, access_key_id, access_key_secret, msg)
+        return old_f(facility, severity, hostname, appname, pid, msg, structured_data)
     end
 --- request
 GET /hello