feat(router): enable flavor expressions in stream subsystem (#11071)

(cherry picked from commit 5c7a7b8)

.requirements

@@ -9,7 +9,7 @@ LUA_KONG_NGINX_MODULE=4d19e8d19c6dbc07eba5cf6f5ebacad95266f928 # 0.6.0
 LUA_RESTY_LMDB=951926f20b674a0622236a0e331b359df1c02d9b # 1.3.0
 LUA_RESTY_EVENTS=8448a92cec36ac04ea522e78f6496ba03c9b1fd8 # 0.2.0
 LUA_RESTY_WEBSOCKET=60eafc3d7153bceb16e6327074e0afc3d94b1316 # 0.4.0
-ATC_ROUTER=72cc8fddeac024c54c9c1fa5a25c28a72d79080e # 1.1.0
+ATC_ROUTER=b0d5e7e2a2ca59bb051959385d3e42d96c93bb98 # 1.2.0
 
 KONG_MANAGER=nightly
 NGX_WASM_MODULE_BRANCH=main

CHANGELOG.md

@@ -36,6 +36,9 @@
 
 #### Core
 
+- Enable `expressions` and `traditional_compatible` router flavor in stream subsystem.
+  [#11071](https://github.com/Kong/kong/pull/11071)
+
 #### Admin API
 
 #### Kong Manager
@@ -169,8 +172,9 @@ Meanwhile, the following Kong configurations cannot reference vaults as they are
 - Bumped pgmoon from 1.16.0 to 1.16.2 (Kong's fork)
   [#11181](https://github.com/Kong/kong/pull/11181)
   [#11229](https://github.com/Kong/kong/pull/11229)
-- Bumped atc-router from 1.0.5 to 1.1.0
+- Bumped atc-router from 1.0.5 to 1.2.0
   [#10100](https://github.com/Kong/kong/pull/10100)
+  [#11071](https://github.com/Kong/kong/pull/11071)
 - Bumped lua-resty-lmdb from 1.1.0 to 1.3.0
   [#11227](https://github.com/Kong/kong/pull/11227)
 

kong/router/atc.lua

@@ -18,6 +18,7 @@ local assert = assert
 local setmetatable = setmetatable
 local pairs = pairs
 local ipairs = ipairs
+local tonumber = tonumber
 
 
 local max = math.max
@@ -49,6 +50,9 @@ local LOGICAL_OR  = " || "
 local LOGICAL_AND = " && "
 
 
+local is_http = ngx.config.subsystem == "http"
+
+
 -- reuse buffer object
 local values_buf = buffer.new(64)
 
@@ -64,8 +68,11 @@ do
                  },
 
     ["Int"]    = {"net.port",
+                  "net.src.port", "net.dst.port",
                  },
 
+    ["IpAddr"] = {"net.src.ip", "net.dst.ip",
+                 },
   }
 
   CACHED_SCHEMA = schema.new()
@@ -343,7 +350,6 @@ end
 -- example.*:123 => example.*, 123
 local split_host_port
 do
-  local tonumber = tonumber
   local DEFAULT_HOSTS_LRUCACHE_SIZE = DEFAULT_MATCH_LRUCACHE_SIZE
 
   local memo_hp = lrucache.new(DEFAULT_HOSTS_LRUCACHE_SIZE)
@@ -381,6 +387,8 @@ do
 end
 
 
+if is_http then
+
 function _M:select(req_method, req_uri, req_host, req_scheme,
                    src_ip, src_port,
                    dst_ip, dst_port,
@@ -588,6 +596,140 @@ function _M:exec(ctx)
   return match_t
 end
 
+else  -- is stream subsystem
+
+function _M:select(_, _, _, scheme,
+                   src_ip, src_port,
+                   dst_ip, dst_port,
+                   sni)
+
+  check_select_params(nil, nil, nil, scheme,
+                      src_ip, src_port,
+                      dst_ip, dst_port,
+                      sni)
+
+  local c = context.new(self.schema)
+
+  for _, field in ipairs(self.fields) do
+    if field == "net.protocol" then
+      assert(c:add_value(field, scheme))
+
+    elseif field == "tls.sni" then
+      local res, err = c:add_value(field, sni)
+      if not res then
+        return nil, err
+      end
+
+    elseif field == "net.src.ip" then
+      assert(c:add_value(field, src_ip))
+
+    elseif field == "net.src.port" then
+      assert(c:add_value(field, src_port))
+
+    elseif field == "net.dst.ip" then
+      assert(c:add_value(field, dst_ip))
+
+    elseif field == "net.dst.port" then
+      assert(c:add_value(field, dst_port))
+
+    end -- if
+  end -- for
+
+  local matched = self.router:execute(c)
+  if not matched then
+    return nil
+  end
+
+  local uuid = c:get_result()
+
+  local service = self.services[uuid]
+  local matched_route = self.routes[uuid]
+
+  local service_protocol, _,  --service_type
+        service_host, service_port,
+        service_hostname_type = get_service_info(service)
+
+  return {
+    route          = matched_route,
+    service        = service,
+    upstream_url_t = {
+      type = service_hostname_type,
+      host = service_host,
+      port = service_port,
+    },
+    upstream_scheme = service_protocol,
+  }
+end
+
+
+function _M:exec(ctx)
+  local src_ip   = var.remote_addr
+  local dst_ip   = var.server_addr
+
+  local src_port = tonumber(var.remote_port, 10)
+  local dst_port = tonumber((ctx or ngx.ctx).host_port, 10) or
+                   tonumber(var.server_port, 10)
+
+  -- error value for non-TLS connections ignored intentionally
+  local sni = server_name()
+
+  -- fallback to preread SNI if current connection doesn't terminate TLS
+  if not sni then
+    sni = var.ssl_preread_server_name
+  end
+
+  local scheme
+  if var.protocol == "UDP" then
+    scheme = "udp"
+  else
+    scheme = sni and "tls" or "tcp"
+  end
+
+  -- when proxying TLS request in second layer or doing TLS passthrough
+  -- rewrite the dst_ip, port back to what specified in proxy_protocol
+  if var.kong_tls_passthrough_block == "1" or var.ssl_protocol then
+    dst_ip = var.proxy_protocol_server_addr
+    dst_port = tonumber(var.proxy_protocol_server_port)
+  end
+
+  local cache_key = (src_ip   or "") .. "|" ..
+                    (src_port or "") .. "|" ..
+                    (dst_ip   or "") .. "|" ..
+                    (dst_port or "") .. "|" ..
+                    (sni      or "")
+
+  local match_t = self.cache:get(cache_key)
+  if not match_t then
+    if self.cache_neg:get(cache_key) then
+      route_match_stat(ctx, "neg")
+      return nil
+    end
+
+    local err
+    match_t, err = self:select(nil, nil, nil, scheme,
+                               src_ip, src_port,
+                               dst_ip, dst_port,
+                               sni)
+    if not match_t then
+      if err then
+        ngx_log(ngx_ERR, "router returned an error: ", err)
+      end
+
+      self.cache_neg:set(cache_key, true)
+      return nil
+    end
+
+    self.cache:set(cache_key, match_t)
+
+  else
+    route_match_stat(ctx, "pos")
+  end
+
+  return match_t
+end
+
+end   -- if is_http
+
 
 function _M._set_ngx(mock_ngx)
   if type(mock_ngx) ~= "table" then