feat(redis): add metrics

apisix/plugins/prometheus/exporter.lua

@@ -37,6 +37,8 @@ local get_stream_routes = router.stream_routes
 local get_protos = require("apisix.plugins.grpc-transcode.proto").protos
 local service_fetch = require("apisix.http.service").get
 local latency_details = require("apisix.utils.log-util").latency_details_in_ms
+local xrpc = require("apisix.stream.xrpc")
+
 
 local ngx_capture
 if ngx.config.subsystem == "http" then
@@ -70,6 +72,8 @@ local function init_stream_metrics()
     metrics.stream_connection_total = prometheus:counter("stream_connection_total",
         "Total number of connections handled per stream route in APISIX",
         {"route"})
+
+    xrpc.init_metrics(prometheus)
 end
 
 

apisix/stream/xrpc.lua

@@ -16,6 +16,7 @@
 --
 local require = require
 local core = require("apisix.core")
+local metrics = require("apisix.stream.xrpc.metrics")
 local ipairs = ipairs
 local pairs = pairs
 local ngx_exit = ngx.exit
@@ -45,7 +46,7 @@ end
 
 
 function _M.init()
-    local local_conf = core.config.local_conf(true)
+    local local_conf = core.config.local_conf()
     if not local_conf.xrpc then
         return
     end
@@ -67,9 +68,26 @@ function _M.init()
 end
 
 
+function _M.init_metrics(collector)
+    local local_conf = core.config.local_conf()
+    if not local_conf.xrpc then
+        return
+    end
+
+    local prot_conf = local_conf.xrpc.protocols
+    if not prot_conf then
+        return
+    end
+
+    for _, prot in ipairs(prot_conf) do
+        metrics.store(collector, prot.name)
+    end
+end
+
+
 function _M.init_worker()
-    for _, prot in pairs(registered_protocols) do
-        if prot.init_worker then
+    for name, prot in pairs(registered_protocols) do
+        if not is_http and prot.init_worker then
             prot.init_worker()
         end
     end

apisix/stream/xrpc/metrics.lua

@@ -0,0 +1,50 @@
+--
+-- 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 require = require
+local core = require("apisix.core")
+local pairs = pairs
+local pcall = pcall
+
+
+local _M = {}
+local hubs = {}
+
+
+function _M.store(prometheus, name)
+    local ok, m = pcall(require, "apisix.stream.xrpc.protocols." .. name .. ".metrics")
+    if not ok then
+        core.log.notice("no metric for protocol ", name)
+        return
+    end
+
+    local hub = {}
+    for metric, conf in pairs(m) do
+        core.log.notice("register metric ", metric, " for protocol ", name)
+        hub[metric] = prometheus[conf.type](prometheus, name .. '_' .. metric,
+                                            conf.help, conf.labels, conf.buckets)
+    end
+
+    hubs[name] = hub
+end
+
+
+function _M.load(name)
+    return hubs[name]
+end
+
+
+return _M

apisix/stream/xrpc/protocols/redis/init.lua

@@ -39,6 +39,7 @@ end)
 
 -- redis protocol spec: https://redis.io/docs/reference/protocol-spec/
 -- There is no plan to support inline command format
+local protocol_name = "redis"
 local _M = {}
 local MAX_LINE_LEN = 128
 local MAX_VALUE_LEN = 128
@@ -107,6 +108,7 @@ function _M.init_downstream(session)
 
     session.req_id_seq = 0
     session.resp_id_seq = 0
+    session.cmd_labels = {session.route.id, ""}
     return xrpc_socket.downstream.socket()
 end
 
@@ -482,6 +484,13 @@ end
 
 
 function _M.log(session, ctx)
+    local metrics = sdk.get_metrics(session, protocol_name)
+    if metrics then
+        session.cmd_labels[2] = ctx.cmd
+        metrics.commands_total:inc(1, session.cmd_labels)
+        metrics.commands_latency_seconds:observe(ctx.var.rpc_time, session.cmd_labels)
+    end
+
     core.tablepool.release("xrpc_redis_cmd_line", ctx.cmd_line)
     ctx.cmd_line = nil
 end

apisix/stream/xrpc/protocols/redis/metrics.lua

@@ -0,0 +1,33 @@
+--
+-- 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 _M = {
+    commands_total = {
+        type = "counter",
+        help = "Total number of requests for a specific Redis command",
+        labels = {"route", "command"},
+    },
+    commands_latency_seconds = {
+        type = "histogram",
+        help = "Latency of requests for a specific Redis command",
+        labels = {"route", "command"},
+        -- latency buckets, 1ms to 1s:
+        buckets = {0.001, 0.002, 0.005, 0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1}
+    },
+}
+
+
+return _M

apisix/stream/xrpc/sdk.lua

@@ -21,6 +21,7 @@
 local core = require("apisix.core")
 local config_util = require("apisix.core.config_util")
 local router = require("apisix.stream.router.ip_port")
+local metrics = require("apisix.stream.xrpc.metrics")
 local apisix_upstream = require("apisix.upstream")
 local xrpc_socket = require("resty.apisix.stream.xrpc.socket")
 local ngx_now = ngx.now
@@ -182,4 +183,20 @@ function _M.set_upstream(session, conf)
 end
 
 
+---
+-- Returns the protocol specific metrics object
+--
+-- @function xrpc.sdk.get_metrics
+-- @tparam table session xrpc session
+-- @tparam string protocol_name protocol name
+-- @treturn nil|table the metrics under the specific protocol if available
+function _M.get_metrics(session, protocol_name)
+    local metric_conf = session.route.protocol.metric
+    if not (metric_conf and metric_conf.enable) then
+        return nil
+    end
+    return metrics.load(protocol_name)
+end
+
+
 return _M

docs/en/latest/xrpc.md

@@ -172,6 +172,31 @@ The protocol itself defines the granularity of the specific request, and the xRP
 
 For example, in the Redis protocol, the execution of a command is considered a request.
 
+### Dynamic metrics
+
+xRPC also supports gathering metrics on the fly and exposing them via Prometheus.
+
+To know how to enable Prometheus metrics for TCP and collect them, please refer to [prometheus](./plugins/prometheus.md).
+
+To get the protocol-specific metrics, you need to:
+
+1. Make sure the Prometheus is enabled for TCP
+2. Add the metric field to the specific route and ensure the `enable` is true:
+
+```json
+{
+    ...
+    "protocol": {
+        "name": "redis",
+        "metric": {
+            "enable": true
+        }
+    }
+}
+```
+
+Different protocols will have different metrics. Please refer to the `Metrics` section of their own documentation.
+
 ## How to write your own protocol
 
 Assuming that your protocol is named `my_proto`, you need to create a directory that can be introduced by `require "apisix.stream.xrpc.protocols.my_proto"`.

docs/en/latest/xrpc/redis.md

@@ -63,6 +63,22 @@ Fields under an entry of `faults`:
 | key | string        | False    |                                               | "blahblah"  | Key fault is restricted to |
 | delay | number        | True    |                                               | 0.1  | Duration of the delay in seconds |
 
+## Metrics
+
+* `apisix_redis_commands_total`: Total number of requests for a specific Redis command.
+
+    | Labels        | Description             |
+    | ------------- | --------------------    |
+    | route         | matched stream route ID |
+    | command       | the Redis command       |
+
+* `apisix_redis_commands_latency_seconds`: Latency of requests for a specific Redis command.
+
+    | Labels        | Description             |
+    | ------------- | --------------------    |
+    | route         | matched stream route ID |
+    | command       | the Redis command       |
+
 ## Example usage
 
 Assumed the APISIX is proxying TCP on port `9101`, and the Redis is listening on port `6379`.