feat(xRPC): support log filter

apisix/schema_def.lua

@@ -807,6 +807,28 @@ local xrpc_protocol_schema = {
             description = "protocol-specific configuration",
             type = "object",
         },
+        logger = {
+            type = "array",
+            items = {
+                properties = {
+                    name = {
+                        type = "string",
+                    },
+                    filter = {
+                        description = "logger filter rules",
+                        type = "array",
+                    },
+                    conf = {
+                        description = "logger plugin configuration",
+                        type = "object",
+                    },
+                },
+                dependencies = {
+                    name = {"filter", "conf"},
+                },
+            },
+        },
+
     },
     required = {"name"}
 }

apisix/stream/xrpc/runner.lua

@@ -14,13 +14,18 @@
 -- See the License for the specific language governing permissions and
 -- limitations under the License.
 --
+local require = require
 local core = require("apisix.core")
+local expr = require("resty.expr.v1")
 local pairs = pairs
 local ngx = ngx
 local ngx_now = ngx.now
 local OK = ngx.OK
 local DECLINED = ngx.DECLINED
 local DONE = ngx.DONE
+local pcall = pcall
+local ipairs = ipairs
+local tostring = tostring
 
 
 local _M = {}
@@ -70,9 +75,55 @@ local function put_req_ctx(session, ctx)
 end
 
 
+local function filter_logger(ctx, logger)
+    if not logger or not logger.filter or #logger.filter == 0 then
+        return false
+    end
+
+    local expr, err = expr.new(logger.filter)
+    if err then
+        core.log.error("failed to validate the 'filter' expression: ", err)
+        return false
+    end
+    return expr:eval(ctx)
+end
+
+
+local function run_log_plugin(ctx, logger)
+    local pkg_name = "apisix.stream.plugins." .. logger.name
+    local ok, plugin = pcall(require, pkg_name)
+    if not ok then
+        core.log.error("failed to load plugin [", logger.name, "] err: ", plugin)
+        return
+    end
+
+    -- we choose to initialize conf_id and conf_type here
+    -- because conf here refers specifically to the conf in the log phase,
+    -- to avoid overwriting the conf in other parts of the protocol
+    ctx.conf_id = tostring(logger.conf)
+    ctx.conf_type = "xrpc-logger"
+
+    local log_func = plugin.log
+    if log_func then
+        log_func(logger.conf, ctx)
+    end
+end
+
+
 local function finish_req(protocol, session, ctx)
     ctx._rpc_end_time = ngx_now()
 
+    local loggers = session._route.protocol.logger
+    if loggers and #loggers > 0 then
+        for _, logger in ipairs(loggers) do
+            local matched = filter_logger(ctx, logger)
+            core.log.info("log filter: ", logger.name, " filter result: ", matched)
+            if matched then
+                run_log_plugin(ctx, logger)
+            end
+        end
+    end
+
     protocol.log(session, ctx)
     put_req_ctx(session, ctx)
 end

t/xds-library/config_xds_2.t

@@ -205,8 +205,6 @@ decode the conf of [/routes/3] failed, err: Expected object key string but found
                 }
             }
             local data_str = core.json.encode(data)
-            ngx.log(ngx.WARN, "data_str : ", require("inspect")(data_str))
-
             ngx.shared["xds-config"]:set("/routes/3", data_str)
             ngx.update_time()
             ngx.shared["xds-config-version"]:set("version", ngx.now())

t/xrpc/apisix/stream/xrpc/protocols/pingpong/init.lua

@@ -76,6 +76,12 @@ local function to_int32(p, idx)
 end
 
 
+local function init_ctx_fileds(ctx)
+    -- init some ctx fields
+    core.ctx.set_vars_meta(ctx)
+end
+
+
 function _M.from_downstream(session, downstream)
     -- read a request from downstream
     -- return status and the new ctx
@@ -160,6 +166,8 @@ function _M.from_downstream(session, downstream)
     if typ == TYPE_UNARY_DYN_UP then
         ctx.len = ctx.len + 4
     end
+
+    init_ctx_fileds(ctx)
     return OK, ctx
 end