feat(core): add wasm integration

kong-3.4.0-0.rockspec

@@ -124,19 +124,20 @@ build = {
     ["kong.api.api_helpers"] = "kong/api/api_helpers.lua",
     ["kong.api.arguments"] = "kong/api/arguments.lua",
     ["kong.api.endpoints"] = "kong/api/endpoints.lua",
-    ["kong.api.routes.kong"] = "kong/api/routes/kong.lua",
-    ["kong.api.routes.health"] = "kong/api/routes/health.lua",
+    ["kong.api.routes.cache"] = "kong/api/routes/cache.lua",
+    ["kong.api.routes.certificates"] = "kong/api/routes/certificates.lua",
+    ["kong.api.routes.clustering"] = "kong/api/routes/clustering.lua",
     ["kong.api.routes.config"] = "kong/api/routes/config.lua",
     ["kong.api.routes.consumers"] = "kong/api/routes/consumers.lua",
+    ["kong.api.routes.debug"] = "kong/api/routes/debug.lua",
+    ["kong.api.routes.filter_chains"] = "kong/api/routes/filter_chains.lua",
+    ["kong.api.routes.health"] = "kong/api/routes/health.lua",
+    ["kong.api.routes.kong"] = "kong/api/routes/kong.lua",
     ["kong.api.routes.plugins"] = "kong/api/routes/plugins.lua",
-    ["kong.api.routes.cache"] = "kong/api/routes/cache.lua",
-    ["kong.api.routes.upstreams"] = "kong/api/routes/upstreams.lua",
-    ["kong.api.routes.targets"] = "kong/api/routes/targets.lua",
-    ["kong.api.routes.certificates"] = "kong/api/routes/certificates.lua",
     ["kong.api.routes.snis"] = "kong/api/routes/snis.lua",
     ["kong.api.routes.tags"] = "kong/api/routes/tags.lua",
-    ["kong.api.routes.clustering"] = "kong/api/routes/clustering.lua",
-    ["kong.api.routes.debug"] = "kong/api/routes/debug.lua",
+    ["kong.api.routes.targets"] = "kong/api/routes/targets.lua",
+    ["kong.api.routes.upstreams"] = "kong/api/routes/upstreams.lua",
 
     ["kong.admin_gui"] = "kong/admin_gui/init.lua",
 
@@ -174,6 +175,7 @@ build = {
     ["kong.runloop.plugin_servers.process"] = "kong/runloop/plugin_servers/process.lua",
     ["kong.runloop.plugin_servers.mp_rpc"] = "kong/runloop/plugin_servers/mp_rpc.lua",
     ["kong.runloop.plugin_servers.pb_rpc"] = "kong/runloop/plugin_servers/pb_rpc.lua",
+    ["kong.runloop.wasm"] = "kong/runloop/wasm.lua",
 
     ["kong.workspaces"] = "kong/workspaces/init.lua",
 
@@ -195,6 +197,7 @@ build = {
     ["kong.db.schema"] = "kong/db/schema/init.lua",
     ["kong.db.dao.keys"] = "kong/db/dao/keys.lua",
     ["kong.db.dao.key_sets"] = "kong/db/dao/key_sets.lua",
+    ["kong.db.dao.filter_chains"] = "kong/db/dao/filter_chains.lua",
     ["kong.db.schema.entities.keys"] = "kong/db/schema/entities/keys.lua",
     ["kong.db.schema.entities.key_sets"] = "kong/db/schema/entities/key_sets.lua",
     ["kong.db.schema.entities.consumers"] = "kong/db/schema/entities/consumers.lua",
@@ -212,6 +215,7 @@ build = {
     ["kong.db.schema.entities.workspaces"] = "kong/db/schema/entities/workspaces.lua",
     ["kong.db.schema.entities.clustering_data_planes"] = "kong/db/schema/entities/clustering_data_planes.lua",
     ["kong.db.schema.entities.parameters"] = "kong/db/schema/entities/parameters.lua",
+    ["kong.db.schema.entities.filter_chains"] = "kong/db/schema/entities/filter_chains.lua",
     ["kong.db.schema.others.migrations"] = "kong/db/schema/others/migrations.lua",
     ["kong.db.schema.others.declarative_config"] = "kong/db/schema/others/declarative_config.lua",
     ["kong.db.schema.entity"] = "kong/db/schema/entity.lua",

kong.conf.default

@@ -1951,3 +1951,15 @@
                         #
                         # Granularity can be adjusted through the `log_level`
                         # directive.
+
+
+#------------------------------------------------------------------------------
+# WASM
+#------------------------------------------------------------------------------
+#
+#
+#wasm = off                       # Use this setting to enable wasm, this allows running
+                                  # wasm filters to process request data.
+
+#wasm_filters_path =              # Path to the directory containing Wasm filters
+                                  # that Kong must load on startup.

kong/api/routes/filter_chains.lua

@@ -0,0 +1,185 @@
+local cjson = require "cjson"
+local endpoints = require "kong.api.endpoints"
+
+
+local kong = kong
+
+
+if kong.configuration.wasm == false then
+
+  local function wasm_disabled_error()
+    return kong.response.exit(400, {
+      message = "this endpoint is only available when wasm is enabled"
+    })
+  end
+
+  return {
+    ["/filter-chains"] = {
+      before = wasm_disabled_error,
+    },
+
+    ["/filter-chains/:filter_chains"] = {
+      before = wasm_disabled_error,
+    },
+
+    ["/filter-chains/:filter_chains/route"] = {
+      before = wasm_disabled_error,
+    },
+
+    ["/filter-chains/:filter_chains/service"] = {
+      before = wasm_disabled_error,
+    },
+
+    -- foreign key endpoints:
+
+    ["/routes/:routes/filter-chains"] = {
+      before = wasm_disabled_error,
+    },
+
+    ["/routes/:routes/filter-chains/:filter_chains"] = {
+      before = wasm_disabled_error,
+    },
+
+    ["/services/:services/filter-chains"] = {
+      before = wasm_disabled_error,
+    },
+
+    ["/services/:services/filter-chains/:filter_chains"] = {
+      before = wasm_disabled_error,
+    },
+
+    -- custom endpoints (implemented below):
+
+    ["/routes/:routes/filters/enabled"] = {
+      GET = wasm_disabled_error,
+    },
+
+    ["/routes/:routes/filters/disabled"] = {
+      GET = wasm_disabled_error,
+    },
+
+    ["/routes/:routes/filters/all"] = {
+      GET = wasm_disabled_error,
+    },
+  }
+end
+
+
+local function add_filters(filters, chain, from)
+  if not chain then
+    return
+  end
+
+  for _, filter in ipairs(chain.filters) do
+    table.insert(filters, {
+      name = filter.name,
+      config = filter.config,
+      from = from,
+      enabled = (chain.enabled == true and filter.enabled == true),
+      filter_chain = {
+        name = chain.name,
+        id = chain.id,
+      }
+    })
+  end
+end
+
+
+local function get_filters(self, db)
+  local route, _, err_t = endpoints.select_entity(self, db, db.routes.schema)
+  if err_t then
+    return nil, err_t
+  end
+
+  if not route then
+    return kong.response.exit(404, { message = "Not found" })
+  end
+
+  local route_chain
+  for chain, _, err_t in kong.db.filter_chains:each_for_route(route, nil, { nulls = true }) do
+    if not chain then
+      return nil, err_t
+    end
+
+    route_chain = chain
+  end
+
+  local service
+  local service_chain
+
+  if route.service then
+    service , _, err_t = kong.db.services:select(route.service)
+    if err_t then
+      return nil, err_t
+    end
+
+    for chain, _, err_t in kong.db.filter_chains:each_for_service(service, nil, { nulls = true }) do
+      if not chain then
+        return nil, err_t
+      end
+
+      service_chain = chain
+    end
+  end
+
+  local filters = setmetatable({}, cjson.array_mt)
+  add_filters(filters, service_chain, "service")
+  add_filters(filters, route_chain, "route")
+
+  return filters
+end
+
+
+return {
+  ["/routes/:routes/filters/all"] = {
+    GET = function(self, db)
+      local filters, err_t = get_filters(self, db)
+      if err_t then
+        return endpoints.handle_error(err_t)
+      end
+
+      return kong.response.exit(200, {
+        filters = filters,
+      })
+    end
+  },
+
+  ["/routes/:routes/filters/enabled"] = {
+    GET = function(self, db)
+      local filters, err_t = get_filters(self, db)
+      if err_t then
+        return endpoints.handle_error(err_t)
+      end
+
+      for i = #filters, 1, -1 do
+        if not filters[i].enabled then
+          table.remove(filters, i)
+        end
+      end
+
+      return kong.response.exit(200, {
+        filters = filters,
+      })
+    end
+  },
+
+  ["/routes/:routes/filters/disabled"] = {
+    GET = function(self, db)
+      local filters, err_t = get_filters(self, db)
+      if err_t then
+        return endpoints.handle_error(err_t)
+      end
+
+      for i = #filters, 1, -1 do
+        if filters[i].enabled then
+          table.remove(filters, i)
+        end
+      end
+
+      return kong.response.exit(200, {
+        filters = filters,
+      })
+    end
+  },
+
+}

kong/clustering/compat/init.lua

@@ -29,6 +29,8 @@ local COMPATIBILITY_CHECKERS = require("kong.clustering.compat.checkers")
 local CLUSTERING_SYNC_STATUS = constants.CLUSTERING_SYNC_STATUS
 local KONG_VERSION = meta.version
 
+local EMPTY = {}
+
 
 local _M = {}
 
@@ -176,6 +178,24 @@ function _M.check_configuration_compatibility(cp, dp)
     end
   end
 
+  if cp.conf.wasm then
+    local dp_filters = dp.filters or EMPTY
+    local missing
+    for name in pairs(cp.filters or EMPTY) do
+      if not dp_filters[name] then
+        missing = missing or {}
+        table.insert(missing, name)
+      end
+    end
+
+    if missing then
+      local msg = "data plane is missing one or more wasm filters "
+                  .. "(" .. table.concat(missing, ", ") .. ")"
+      return nil, msg, CLUSTERING_SYNC_STATUS.FILTER_SET_INCOMPATIBLE
+    end
+  end
+
+
   return true, nil, CLUSTERING_SYNC_STATUS.NORMAL
 end
 

kong/clustering/control_plane.lua

@@ -279,8 +279,10 @@ function _M:handle_cp_websocket()
 
   if self.deflated_reconfigure_payload then
     -- initial configuration compatibility for sync status variable
-    _, _, sync_status = self:check_configuration_compatibility(
-                              { dp_plugins_map = dp_plugins_map, })
+    _, _, sync_status = self:check_configuration_compatibility({
+      dp_plugins_map = dp_plugins_map,
+      filters = data.filters,
+    })
 
     table_insert(queue, RECONFIGURE_TYPE)
     queue.post()
@@ -397,8 +399,11 @@ function _M:handle_cp_websocket()
       assert(payload == RECONFIGURE_TYPE)
 
       local previous_sync_status = sync_status
-      ok, err, sync_status = self:check_configuration_compatibility(
-                                { dp_plugins_map = dp_plugins_map, })
+      ok, err, sync_status = self:check_configuration_compatibility({
+        dp_plugins_map = dp_plugins_map,
+        filters = data.filters,
+      })
+
       if not ok then
         ngx_log(ngx_WARN, _log_prefix, "unable to send updated configuration to data plane: ", err, log_suffix)
         if sync_status ~= previous_sync_status then
@@ -532,8 +537,9 @@ local function push_config_loop(premature, self, push_config_semaphore, delay)
 end
 
 
-function _M:init_worker(plugins_list)
+function _M:init_worker(basic_info)
   -- ROLE = "control_plane"
+  local plugins_list = basic_info.plugins
   self.plugins_list = plugins_list
   self.plugins_map = plugins_list_to_map(plugins_list)
 
@@ -547,6 +553,8 @@ function _M:init_worker(plugins_list)
     self.plugin_versions[plugin.name] = plugin.version
   end
 
+  self.filters = basic_info.filters
+
   local push_config_semaphore = semaphore.new()
 
   -- When "clustering", "push_config" worker event is received by a worker,

kong/clustering/data_plane.lua

@@ -73,10 +73,11 @@ function _M.new(clustering)
 end
 
 
-function _M:init_worker(plugins_list)
+function _M:init_worker(basic_info)
   -- ROLE = "data_plane"
 
-  self.plugins_list = plugins_list
+  self.plugins_list = basic_info.plugins
+  self.filters = basic_info.filters
 
   -- only run in process which worker_id() == 0
   assert(ngx.timer.at(0, function(premature)
@@ -147,6 +148,7 @@ function _M:communicate(premature)
   _, err = c:send_binary(cjson_encode({ type = "basic_info",
                                         plugins = self.plugins_list,
                                         process_conf = configuration,
+                                        filters = self.filters,
                                         labels = labels, }))
   if err then
     ngx_log(ngx_ERR, _log_prefix, "unable to send basic information to control plane: ", uri,

kong/clustering/init.lua

@@ -101,15 +101,27 @@ function _M:init_worker()
     return { name = p.name, version = p.handler.VERSION, }
   end, plugins_list)
 
+  local filters = {}
+  if kong.db.filter_chains.filters then
+    for _, filter in ipairs(kong.db.filter_chains.filters) do
+      filters[filter.name] = { name = filter.name }
+    end
+  end
+
+  local basic_info = {
+    plugins = plugins_list,
+    filters = filters,
+  }
+
   local role = self.conf.role
 
   if role == "control_plane" then
-    self:init_cp_worker(plugins_list)
+    self:init_cp_worker(basic_info)
     return
   end
 
   if role == "data_plane" then
-    self:init_dp_worker(plugins_list)
+    self:init_dp_worker(basic_info)
   end
 end