feat: add error log skywalking reporter

apisix/plugins/error-log-logger.lua

@@ -20,6 +20,7 @@ local errlog = require("ngx.errlog")
 local batch_processor = require("apisix.utils.batch-processor")
 local plugin = require("apisix.plugin")
 local timers = require("apisix.timers")
+local http = require("resty.http")
 local plugin_name = "error-log-logger"
 local table = core.table
 local schema_def = core.schema
@@ -36,23 +37,46 @@ local lrucache = core.lrucache.new({
 local metadata_schema = {
     type = "object",
     properties = {
-        host = schema_def.host_def,
-        port = {type = "integer", minimum = 0},
-        tls = {type = "boolean", default = false},
-        tls_server_name = {type = "string"},
-        timeout = {type = "integer", minimum = 1, default = 3},
-        keepalive = {type = "integer", minimum = 1, default = 30},
+        type = {type = "string", default = "TCP", enum = {"TCP", "SKYWALKING"}},
+        tcp = {
+            type = "object",
+            properties = {
+                host = schema_def.host_def,
+                port = {type = "integer", minimum = 0},
+                tls = {type = "boolean", default = false},
+                tls_server_name = {type = "string"},
+            },
+            required = {"host", "port"}
+        },
+        skywalking = {
+            type = "object",
+            properties = {
+                endpoint_addr = schema_def.uri,
+                service_name = {type = "string", default = "APISIX"},
+                service_instance_name = {type="string", default = "APISIX Service Instance"},
+            },
+            required = {"endpoint_addr"}
+        },
+        host = {schema_def.host_def, description = "Deprecated, use `tcp.host` instead."},
+        port = {type = "integer", minimum = 0, description = "Deprecated, use `tcp.port` instead."},
+        tls = {type = "boolean", default = false,
+                description = "Deprecated, use `tcp.tls` instead."},
+        tls_server_name = {type = "string",
+                description = "Deprecated, use `tcp.tls_server_name` instead."},
         name = {type = "string", default = plugin_name},
         level = {type = "string", default = "WARN", enum = {"STDERR", "EMERG", "ALERT", "CRIT",
                 "ERR", "ERROR", "WARN", "NOTICE", "INFO", "DEBUG"}},
+        timeout = {type = "integer", minimum = 1, default = 3},
+        keepalive = {type = "integer", minimum = 1, default = 30},
         batch_max_size = {type = "integer", minimum = 0, default = 1000},
         max_retry_count = {type = "integer", minimum = 0, default = 0},
         retry_delay = {type = "integer", minimum = 0, default = 1},
         buffer_duration = {type = "integer", minimum = 1, default = 60},
         inactive_timeout = {type = "integer", minimum = 1, default = 3},
     },
-    required = {"host", "port"}
 }
+
+
 local schema = {
     type = "object",
 }
@@ -85,7 +109,15 @@ local _M = {
 }
 
 
-local function send_to_server(data)
+function _M.check_schema(conf, schema_type)
+    if schema_type == core.schema.TYPE_METADATA then
+        return core.schema.check(metadata_schema, conf)
+    end
+    return core.schema.check(schema, conf)
+end
+
+
+local function send_to_tcp_server(data)
     local sock, soc_err = tcp()
 
     if not sock then
@@ -94,33 +126,86 @@ local function send_to_server(data)
 
     sock:settimeout(config.timeout * 1000)
 
-    local ok, err = sock:connect(config.host, config.port)
+    local tcp_config = config.tcp
+    local ok, err = sock:connect(tcp_config.host, tcp_config.port)
     if not ok then
-        return false, "failed to connect the TCP server: host[" .. config.host
-            .. "] port[" .. tostring(config.port) .. "] err: " .. err
+        return false, "failed to connect the TCP server: host[" .. tcp_config.host
+            .. "] port[" .. tostring(tcp_config.port) .. "] err: " .. err
     end
 
-    if config.tls then
-        ok, err = sock:sslhandshake(false, config.tls_server_name, false)
+    if tcp_config.tls then
+        ok, err = sock:sslhandshake(false, tcp_config.tls_server_name, false)
         if not ok then
             sock:close()
             return false, "failed to perform TLS handshake to TCP server: host["
-                .. config.host .. "] port[" .. tostring(config.port) .. "] err: " .. err
+                .. tcp_config.host .. "] port[" .. tostring(tcp_config.port) .. "] err: " .. err
         end
     end
 
     local bytes, err = sock:send(data)
     if not bytes then
         sock:close()
-        return false, "failed to send data to TCP server: host[" .. config.host
-            .. "] port[" .. tostring(config.port) .. "] err: " .. err
+        return false, "failed to send data to TCP server: host[" .. tcp_config.host
+            .. "] port[" .. tostring(tcp_config.port) .. "] err: " .. err
     end
 
     sock:setkeepalive(config.keepalive * 1000)
     return true
 end
 
 
+local function send_to_skywalking(log_message)
+    local err_msg
+    local res = true
+    core.log.info("sending a batch logs to ", config.skywalking.endpoint_addr)
+
+    local httpc = http.new()
+    httpc:set_timeout(config.timeout * 1000)
+
+    local entries = {}
+    for i = 1, #log_message, 2 do
+        local content = {
+            service = config.skywalking.service_name,
+            serviceInstance = config.skywalking.service_instance_name,
+            endpoint = "",
+            body = {
+                text = {
+                    text = log_message[i]
+                }
+           }
+        }
+        table.insert(entries, content)
+    end
+
+    local httpc_res, httpc_err = httpc:request_uri(
+        config.skywalking.endpoint_addr,
+        {
+            method = "POST",
+            body = core.json.encode(entries),
+            keepalive_timeout = config.keepalive * 1000,
+            headers = {
+                ["Content-Type"] = "application/json",
+            }
+        }
+    )
+
+    if not httpc_res then
+        return false, "error while sending data to skywalking["
+            .. config.skywalking.endpoint_addr .. "] " .. httpc_err
+    end
+
+    -- some error occurred in the server
+    if httpc_res.status >= 400 then
+        res =  false
+        err_msg = "server returned status code[" .. httpc_res.status
+            .. "] skywalking[" .. config.skywalking.endpoint_addr
+            .. "] body[" .. httpc_res:read_body() .. "]"
+    end
+
+    return res, err_msg
+end
+
+
 local function update_filter(value)
     local level = log_level[string.upper(value.level)]
     local status, err = errlog.set_filter_level(level)
@@ -134,6 +219,14 @@ local function update_filter(value)
 end
 
 
+local function send(data)
+    if config.type == "TCP" then
+        return send_to_tcp_server(data)
+    end
+    return send_to_skywalking(data)
+end
+
+
 local function process()
     local metadata = plugin.plugin_metadata(plugin_name)
     if not (metadata and metadata.value and metadata.modifiedIndex) then
@@ -180,7 +273,7 @@ local function process()
     }
 
     local err
-    log_buffer, err = batch_processor:new(send_to_server, config_bat)
+    log_buffer, err = batch_processor:new(send, config_bat)
 
     if not log_buffer then
         core.log.warn("error when creating the batch processor: ", err)

t/plugin/error-log-logger.t

@@ -57,9 +57,9 @@ _EOC_
     $block->set_value("stream_config", $stream_single_server);
 
     my $stream_default_server = <<_EOC_;
-	    content_by_lua_block {
-	    	ngx.log(ngx.INFO, "a stream server")
-	    }
+        content_by_lua_block {
+            ngx.log(ngx.INFO, "a stream server")
+        }
 _EOC_
 
     $block->set_value("stream_server_config", $stream_default_server);
@@ -121,8 +121,10 @@ plugins:
             local code, body = t('/apisix/admin/plugin_metadata/error-log-logger',
                 ngx.HTTP_PUT,
                 [[{
-                    "port": 1999,
-                    "inactive_timeout": 1
+                    "tcp": {
+                        "port": 1999,
+                        "inactive_timeout": 1
+                    }
                 }]]
                 )
 
@@ -157,9 +159,11 @@ plugins:
             local code, body = t('/apisix/admin/plugin_metadata/error-log-logger',
                 ngx.HTTP_PUT,
                 [[{
-                    "host": "127.0.0.1",
-                    "port": 9999,
-                    "inactive_timeout": 1
+                    "tcp": {
+                        "host": "127.0.0.1",
+                        "port": 1999,
+                        "inactive_timeout": 1
+                    }
                 }]]
                 )
             ngx.sleep(2)
@@ -190,9 +194,11 @@ plugins:
             local code, body = t('/apisix/admin/plugin_metadata/error-log-logger',
                 ngx.HTTP_PUT,
                 [[{
-                    "host": "127.0.0.1",
-                    "port": 1999,
-                    "inactive_timeout": 1
+                    "tcp": {
+                        "host": "127.0.0.1",
+                        "port": 1999,
+                        "inactive_timeout": 1
+                    }
                 }]]
                 )
             ngx.sleep(2)
@@ -298,9 +304,11 @@ plugins:
                 [[{
                     "plugins": {
                         "error-log-logger": {
-                            "host": "127.0.0.1",
-                            "port": 1999,
-                            "inactive_timeout": 1
+                            "tcp": {
+                                "host": "127.0.0.1",
+                                "port": 1999,
+                                "inactive_timeout": 1
+                            }
                         }
                     },
                     "upstream": {