feature: add UDP logger plugin (#1070)

conf/config.yaml

@@ -94,6 +94,7 @@ plugins:                          # plugin list
   - redirect
   - response-rewrite
   - fault-injection
+  - udp-logger
 
 stream_plugins:
   - mqtt-proxy

lua/apisix/plugins/log-util.lua

@@ -0,0 +1,61 @@
+--
+-- 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 core     = require("apisix.core")
+
+local _M = {}
+
+local function get_full_log(ngx)
+    local ctx = ngx.ctx.api_ctx
+    local var = ctx.var
+    local service_id
+    local route_id
+    local url = var.scheme .. "://" .. var.host .. ":" .. var.server_port .. var.request_uri
+    local matched_route = ctx.matched_route and ctx.matched_route.value
+
+    if matched_route then
+        service_id = matched_route.service_id or ""
+        route_id = matched_route.id
+    else
+        service_id = var.host
+    end
+
+    return  {
+        request = {
+            url = url,
+            uri = var.request_uri,
+            method = ngx.req.get_method(),
+            headers = ngx.req.get_headers(),
+            querystring = ngx.req.get_uri_args(),
+            size = var.request_length
+        },
+        response = {
+            status = ngx.status,
+            headers = ngx.resp.get_headers(),
+            size = var.bytes_sent
+        },
+        upstream = var.upstream_addr,
+        service_id = service_id,
+        route_id = route_id,
+        consumer = ctx.consumer,
+        client_ip = core.request.get_remote_client_ip(ngx.ctx.api_ctx),
+        start_time = ngx.req.start_time() * 1000,
+        latency = (ngx.now() - ngx.req.start_time()) * 1000
+    }
+end
+
+_M.get_full_log = get_full_log
+return _M

lua/apisix/plugins/udp-logger.lua

@@ -0,0 +1,78 @@
+--
+-- 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 core     = require("apisix.core")
+local log_util = require("apisix.plugins.log-util")
+local plugin_name = "udp-logger"
+local ngx = ngx
+
+local timer_at = ngx.timer.at
+local udp = ngx.socket.udp
+
+local schema = {
+    type = "object",
+    properties = {
+        host = {type = "string"},
+        port = {type = "integer", minimum = 0},
+        timeout = {type = "integer", minimum = 1, default= 1000} -- timeout in milliseconds
+    },
+    required = {"host", "port"}
+}
+
+
+local _M = {
+    version = 0.1,
+    priority = 400,
+    name = plugin_name,
+    schema = schema,
+}
+
+function _M.check_schema(conf)
+    return core.schema.check(schema, conf)
+end
+
+local function log(premature, conf, log_message)
+    if premature then
+        return
+    end
+
+    local sock = udp()
+    sock:settimeout(conf.timeout)
+
+    local ok, err = sock:setpeername(conf.host, conf.port)
+    if not ok then
+        core.log.error("failed to connect to UDP server: host[", conf.host, "] port[", conf.port, "] ", err)
+        return
+    end
+
+    ok, err = sock:send(log_message)
+    if not ok then
+        core.log.error("failed to send data to UDP server: host[", conf.host, "] port[", conf.port, "] ", err)
+    end
+
+    ok, err = sock:close()
+    if not ok then
+        core.log.error("failed to close the UDP connection, host[", conf.host, "] port[", conf.port, "] ", err)
+    end
+end
+
+
+function _M.log(conf)
+    return timer_at(0, log, conf, core.json.encode(log_util.get_full_log(ngx)))
+end
+
+return _M
+

t/admin/plugins.t

@@ -30,6 +30,6 @@ __DATA__
 --- request
 GET /apisix/admin/plugins/list
 --- response_body_like eval
-qr/\["limit-req","limit-count","limit-conn","key-auth","basic-auth","prometheus","node-status","jwt-auth","zipkin","ip-restriction","grpc-transcode","serverless-pre-function","serverless-post-function","openid-connect","proxy-rewrite","redirect","response-rewrite","fault-injection"\]/
+qr/\["limit-req","limit-count","limit-conn","key-auth","basic-auth","prometheus","node-status","jwt-auth","zipkin","ip-restriction","grpc-transcode","serverless-pre-function","serverless-post-function","openid-connect","proxy-rewrite","redirect","response-rewrite","fault-injection",["udp-logger"]\]/
 --- no_error_log
 [error]

t/debug/debug-mode.t

@@ -71,6 +71,7 @@ loaded plugin and sort by priority: 900 name: redirect
 loaded plugin and sort by priority: 899 name: response-rewrite
 loaded plugin and sort by priority: 506 name: grpc-transcode
 loaded plugin and sort by priority: 500 name: prometheus
+loaded plugin and sort by priority: 400 name: udp-logger
 loaded plugin and sort by priority: 0 name: example-plugin
 loaded plugin and sort by priority: -1000 name: zipkin
 loaded plugin and sort by priority: -2000 name: serverless-post-function

t/plugin/udp-logger.t

@@ -0,0 +1,222 @@
+#
+# 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.
+#
+use t::APISIX 'no_plan';
+
+repeat_each(1);
+no_long_string();
+no_root_location();
+run_tests;
+
+__DATA__
+
+=== TEST 1: sanity
+--- config
+    location /t {
+        content_by_lua_block {
+            local plugin = require("apisix.plugins.udp-logger")
+            local ok, err = plugin.check_schema({host = "127.0.0.1", port = 3000})
+            if not ok then
+                ngx.say(err)
+            end
+
+            ngx.say("done")
+        }
+    }
+--- request
+GET /t
+--- response_body
+done
+--- no_error_log
+[error]
+
+
+=== TEST 2: missing host
+--- config
+    location /t {
+        content_by_lua_block {
+            local plugin = require("apisix.plugins.udp-logger")
+            local ok, err = plugin.check_schema({port = 3000})
+            if not ok then
+                ngx.say(err)
+            end
+
+            ngx.say("done")
+        }
+    }
+--- request
+GET /t
+--- response_body
+property "host" is required
+done
+--- no_error_log
+[error]
+
+
+=== TEST 3: wrong type of string
+--- config
+    location /t {
+        content_by_lua_block {
+            local plugin = require("apisix.plugins.udp-logger")
+            local ok, err = plugin.check_schema({host= "127.0.0.1", port = 3000, timeout = "10"})
+            if not ok then
+                ngx.say(err)
+            end
+
+            ngx.say("done")
+        }
+    }
+--- request
+GET /t
+--- response_body
+property "timeout" validation failed: wrong type: expected integer, got string
+done
+--- no_error_log
+[error]
+
+
+=== TEST 4: add plugin
+--- config
+    location /t {
+        content_by_lua_block {
+            local t = require("lib.test_admin").test
+            local code, body = t('/apisix/admin/routes/1',
+                 ngx.HTTP_PUT,
+                 [[{
+                        "plugins": {
+                            "udp-logger": {
+                                "host": "127.0.0.1",
+                                "port": 2000
+                            }
+                        },
+                        "upstream": {
+                            "nodes": {
+                                "127.0.0.1:1982": 1
+                            },
+                            "type": "roundrobin"
+                        },
+                        "uri": "/opentracing"
+                }]],
+                [[{
+                    "node": {
+                        "value": {
+                            "plugins": {
+                                "udp-logger": {
+                                    "host": "127.0.0.1",
+                                    "port": 2000
+                                }
+                            },
+                            "upstream": {
+                                "nodes": {
+                                    "127.0.0.1:1982": 1
+                                },
+                                "type": "roundrobin"
+                            },
+                            "uri": "/opentracing"
+                        },
+                        "key": "/apisix/routes/1"
+                    },
+                    "action": "set"
+                }]]
+                )
+
+            if code >= 300 then
+                ngx.status = code
+            end
+            ngx.say(body)
+        }
+    }
+--- request
+GET /t
+--- response_body
+passed
+--- no_error_log
+[error]
+
+
+=== TEST 5: access
+--- request
+GET /opentracing
+--- response_body
+opentracing
+--- no_error_log
+[error]
+--- wait: 0.2
+
+
+=== TEST 6: error log
+--- config
+    location /t {
+        content_by_lua_block {
+            local t = require("lib.test_admin").test
+            local code, body = t('/apisix/admin/routes/1',
+                 ngx.HTTP_PUT,
+                 [[{
+                        "plugins": {
+                            "udp-logger": {
+                                "host": "312.0.0.1",
+                                "port": 2000
+                            }
+                        },
+                        "upstream": {
+                            "nodes": {
+                                "127.0.0.1:1982": 1
+                            },
+                            "type": "roundrobin"
+                        },
+                        "uri": "/opentracing"
+                }]],
+                [[{
+                    "node": {
+                        "value": {
+                            "plugins": {
+                                "udp-logger": {
+                                    "host": "312.0.0.1",
+                                    "port": 2000
+                                }
+                            },
+                            "upstream": {
+                                "nodes": {
+                                    "127.0.0.1:1982": 1
+                                },
+                                "type": "roundrobin"
+                            },
+                            "uri": "/opentracing"
+                        },
+                        "key": "/apisix/routes/1"
+                    },
+                    "action": "set"
+                }]]
+                )
+
+            if code >= 300 then
+                ngx.status = code
+            end
+
+            ngx.say(body)
+
+            local http = require "resty.http"
+            local httpc = http.new()
+            local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/opentracing"
+            local res, err = httpc:request_uri(uri, {method = "GET"})
+        }
+    }
+--- request
+GET /t
+--- error_log
+failed to connect to UDP server: host[312.0.0.1] port[2000]
+[error]
+--- wait: 0.2