feat: stream subsystem support kubernetes service discovery (#8640)

Fixes #7779

apisix/cli/ngx_tpl.lua

@@ -145,6 +145,13 @@ stream {
     lua_shared_dict plugin-limit-conn-stream {* stream.lua_shared_dict["plugin-limit-conn-stream"] *};
     {% end %}
 
+    # for discovery shared dict
+    {% if discovery_shared_dicts then %}
+    {% for key, size in pairs(discovery_shared_dicts) do %}
+    lua_shared_dict {*key*}-stream {*size*};
+    {% end %}
+    {% end %}
+
     resolver {% for _, dns_addr in ipairs(dns_resolver or {}) do %} {*dns_addr*} {% end %} {% if dns_resolver_valid then %} valid={*dns_resolver_valid*}{% end %} ipv6={% if enable_ipv6 then %}on{% else %}off{% end %};
     resolver_timeout {*resolver_timeout*};
 

apisix/discovery/kubernetes/init.lua

@@ -25,7 +25,8 @@ local os = os
 local error = error
 local pcall = pcall
 local setmetatable = setmetatable
-local process = require("ngx.process")
+local is_http = ngx.config.subsystem == "http"
+local support_process, process = pcall(require, "ngx.process")
 local core = require("apisix.core")
 local util = require("apisix.cli.util")
 local local_conf = require("apisix.core.config_local").local_conf()
@@ -331,9 +332,24 @@ local function start_fetch(handle)
     ngx.timer.at(0, timer_runner)
 end
 
+local function get_endpoint_dict(id)
+    local shm = "kubernetes"
+
+    if id and #id > 0 then
+        shm = shm .. "-" .. id
+    end
+
+    if not is_http then
+        shm = shm .. "-stream"
+    end
+
+    return ngx.shared[shm]
+end
+
 
 local function single_mode_init(conf)
-    local endpoint_dict = ngx.shared.kubernetes
+    local endpoint_dict = get_endpoint_dict()
+
     if not endpoint_dict then
         error("failed to get lua_shared_dict: ngx.shared.kubernetes, " ..
                 "please check your APISIX version")
@@ -407,7 +423,7 @@ local function multiple_mode_worker_init(confs)
             error("duplicate id value")
         end
 
-        local endpoint_dict = ngx.shared["kubernetes-" .. id]
+        local endpoint_dict = get_endpoint_dict(id)
         if not endpoint_dict then
             error(string.format("failed to get lua_shared_dict: ngx.shared.kubernetes-%s, ", id) ..
                     "please check your APISIX version")
@@ -433,7 +449,7 @@ local function multiple_mode_init(confs)
             error("duplicate id value")
         end
 
-        local endpoint_dict = ngx.shared["kubernetes-" .. id]
+        local endpoint_dict = get_endpoint_dict(id)
         if not endpoint_dict then
             error(string.format("failed to get lua_shared_dict: ngx.shared.kubernetes-%s, ", id) ..
                     "please check your APISIX version")
@@ -504,6 +520,11 @@ end
 
 
 function _M.init_worker()
+    if not support_process then
+        core.log.error("kubernetes discovery not support in subsystem: ", ngx.config.subsystem,
+                       ", please check if your openresty version >= 1.19.9.1 or not")
+        return
+    end
     local discovery_conf = local_conf.discovery.kubernetes
     core.log.info("kubernetes discovery conf: ", core.json.delay_encode(discovery_conf))
     if #discovery_conf == 0 then

docs/en/latest/discovery/kubernetes.md

@@ -34,6 +34,12 @@ The [_Kubernetes_](https://kubernetes.io/) service discovery [_List-Watch_](http
 
 Discovery also provides a node query interface in accordance with the [_APISIX Discovery Specification_](https://github.com/apache/apisix/blob/master/docs/en/latest/discovery.md).
 
+:::note
+
+use kubernetes discovery in L4 require OpenResty version >= 1.19.9.1
+
+:::
+
 ## How To Use
 
 Kubernetes service discovery both support single-cluster and multi-cluster mode, applicable to the case where the service is distributed in a single or multiple Kubernetes clusters.

docs/zh/latest/discovery/kubernetes.md

@@ -34,6 +34,12 @@ Kubernetes 服务发现以 [_List-Watch_](https://kubernetes.io/docs/reference/u
 
 同时遵循 [_APISIX Discovery 规范_](https://github.com/apache/apisix/blob/master/docs/zh/latest/discovery.md) 提供了节点查询接口。
 
+:::note
+
+在四层中使用 Kubernetes 服务发现要求 OpenResty 版本大于等于 1.19.9.1
+
+:::
+
 ## Kubernetes 服务发现的使用
 
 目前 Kubernetes 服务发现支持单集群和多集群模式，分别适用于待发现的服务分布在单个或多个 Kubernetes 的场景。

t/APISIX.pm

@@ -393,6 +393,10 @@ _EOC_
     lua_shared_dict etcd-cluster-health-check-stream 10m;
     lua_shared_dict worker-events-stream 10m;
 
+    lua_shared_dict kubernetes-stream 1m;
+    lua_shared_dict kubernetes-first-stream 1m;
+    lua_shared_dict kubernetes-second-stream 1m;
+
     upstream apisix_backend {
         server 127.0.0.1:1900;
         balancer_by_lua_block {