feat: validate the subordinate and check reference when deleting and adding…

apisix/admin/stream_routes.lua

@@ -16,6 +16,8 @@
 --
 local core = require("apisix.core")
 local utils = require("apisix.admin.utils")
+local config_util = require("apisix.core.config_util")
+local routes = require("apisix.stream.router.ip_port").routes
 local stream_route_checker = require("apisix.stream.router.ip_port").stream_route_checker
 local tostring = tostring
 
@@ -25,6 +27,51 @@ local _M = {
     need_v3_filter = true,
 }
 
+local function check_router_refer(items, key)
+    local refer_list = {}
+    for _, item in config_util.iterate_values(items) do
+        if item.value == nil then
+            goto CONTINUE
+        end
+        local route = item.value
+        if route.protocol and route.protocol.superior_id then
+	        local data
+            local setkey="/stream_routes/"..route.protocol.superior_id.."/refer"
+            local res, err = core.etcd.get(setkey,false)
+            if res then
+	            if #res.body.node.value == 0 then
+                    local v = core.json.decode("{}")
+	                local  r_id = item["key"]
+	                v[r_id]=1
+	                data = core.json.encode(v)
+                else
+                    local v = core.json.decode(res.body.node.value)
+	            local  r_id = item["key"]
+	            v[r_id]=1
+	            data = core.json.encode(v)
+                end
+            end 
+            local setres, err = core.etcd.set(setkey, data)
+            if not setres then
+                core.log.error("failed to put stream route[", key, "]: ", err)
+            end
+        end
+        ::CONTINUE::
+     end
+     local referkey = key .. "/refer"
+     local rescheck, err = core.etcd.get(referkey,false)
+     if rescheck then
+         if rescheck.body.node  ~= nil then
+             local refer_values=core.json.decode(rescheck.body.node.value)
+             for v,_ in pairs(refer_values) do
+	         table.insert(refer_list,v)
+             end
+         end
+     end
+     return refer_list
+end
+
+
 
 local function check_conf(id, conf, need_id)
     if not conf then
@@ -142,14 +189,23 @@ function _M.delete(id)
         return 400, {error_msg = "missing stream route id"}
     end
 
+    local items,_ = routes()
     local key = "/stream_routes/" .. id
     -- core.log.info("key: ", key)
+    local refer_list=check_router_refer(items,key)
+    local warn_message
+    if #refer_list >0 then
+        warn_message = key.." is refered by "..table.concat(refer_list,";;")
+    else 
+        warn_message = key.." is refered by None"
+    end
+
     local res, err = core.etcd.delete(key)
     if not res then
         core.log.error("failed to delete stream route[", key, "]: ", err)
         return 503, {error_msg = err}
     end
-
+    res.body["refer"]=warn_message
     return res.status, res.body
 end
 

apisix/cli/ngx_tpl.lua

@@ -447,6 +447,7 @@ http {
 
     init_worker_by_lua_block {
         apisix.http_init_worker()
+        apisix.stream_init_worker()
     }
 
     exit_worker_by_lua_block {

apisix/stream/router/ip_port.lua

@@ -63,23 +63,45 @@ local create_router
 do
     local sni_to_items = {}
     local tls_routes = {}
+    local routeid_to_protocols = {}
 
     function create_router(items)
         local tls_routes_idx = 1
         local other_routes_idx = 1
         core.table.clear(tls_routes)
         core.table.clear(other_routes)
         core.table.clear(sni_to_items)
-
+        core.table.clear(routeid_to_protocols)
         for _, item in config_util.iterate_values(items) do
             if item.value == nil then
                 goto CONTINUE
             end
+            local route = item.value
+            if route.protocol  then
+               routeid_to_protocols[item.key]=route.protocol.name
+            else
+               routeid_to_protocols[item.key]="No-Protocol"
+            end
+            ::CONTINUE::
+        end
 
+        for _, item in config_util.iterate_values(items) do
+            if item.value == nil then
+                goto CONTINUE
+            end
             local route = item.value
             if route.protocol and route.protocol.superior_id then
                 -- subordinate route won't be matched in the entry
-                -- TODO: check the subordinate relationship in the Admin API
+                local key="/apisix/stream_routes/"..route.protocol.superior_id
+                if routeid_to_protocols[key] == nil then
+	                core.log.warn("There is not exist stream_route: "..key)
+	            elseif routeid_to_protocols[key] == "No-Protocol" then
+	                core.log.warn("The stream_route: "..key.." may lacks procotol configuration")
+	            elseif routeid_to_protocols[key] == route.protocol.name then
+		            goto CONTINUE
+                else 
+                    core.log.warn("RPC procotol is different in stream_route:"..item.key.." and "..key)
+                end
                 goto CONTINUE
             end
 
@@ -146,7 +168,7 @@ do
         if router_ver ~= user_routes.conf_version then
             local err = create_router(user_routes.values)
             if err then
-                return false, "failed to create router: " .. err
+               return false, "failed to create router: " .. err
             end
 
             router_ver = user_routes.conf_version
@@ -236,6 +258,7 @@ function _M.stream_init_worker(filter)
             checker = stream_route_checker,
             filter = filter,
         })
+
 
     if not user_routes then
         error("failed to create etcd instance for fetching /stream_routes : "