feat(clustering): introduce incremental sync for clustering

changelog/unreleased/kong/cp-dp-rpc.yml

@@ -0,0 +1,3 @@
+message: "Added a remote procedure call (RPC) framework for Hybrid mode deployments."
+type: feature
+scope: Clustering

changelog/unreleased/kong/dynamic-log-level-rpc.yml

@@ -0,0 +1,6 @@
+message: |
+  Dynamic log level over Hybrid mode RPC which allows setting DP log level
+  to a different level for specified duration before reverting back
+  to the `kong.conf` configured value.
+type: feature
+scope: Clustering

kong-3.9.0-0.rockspec

@@ -88,7 +88,6 @@ build = {
     ["kong.clustering.compat.checkers"] = "kong/clustering/compat/checkers.lua",
     ["kong.clustering.config_helper"] = "kong/clustering/config_helper.lua",
     ["kong.clustering.tls"] = "kong/clustering/tls.lua",
-    ["kong.clustering.services.debug"] = "kong/clustering/services/debug.lua",
 
     ["kong.clustering.rpc.callbacks"] = "kong/clustering/rpc/callbacks.lua",
     ["kong.clustering.rpc.future"] = "kong/clustering/rpc/future.lua",
@@ -99,6 +98,12 @@ build = {
     ["kong.clustering.rpc.utils"] = "kong/clustering/rpc/utils.lua",
     ["kong.clustering.rpc.concentrator"] = "kong/clustering/rpc/concentrator.lua",
 
+    ["kong.clustering.services.debug"] = "kong/clustering/services/debug.lua",
+    ["kong.clustering.services.sync"] = "kong/clustering/services/sync/init.lua",
+    ["kong.clustering.services.sync.rpc"] = "kong/clustering/services/sync/rpc.lua",
+    ["kong.clustering.services.sync.hooks"] = "kong/clustering/services/sync/hooks.lua",
+    ["kong.clustering.services.sync.strategies.postgres"] = "kong/clustering/services/sync/strategies/postgres.lua",
+
     ["kong.cluster_events"] = "kong/cluster_events/init.lua",
     ["kong.cluster_events.strategies.postgres"] = "kong/cluster_events/strategies/postgres.lua",
     ["kong.cluster_events.strategies.off"] = "kong/cluster_events/strategies/off.lua",
@@ -289,7 +294,6 @@ build = {
     ["kong.db.strategies.postgres.plugins"] = "kong/db/strategies/postgres/plugins.lua",
     ["kong.db.strategies.off"] = "kong/db/strategies/off/init.lua",
     ["kong.db.strategies.off.connector"] = "kong/db/strategies/off/connector.lua",
-    ["kong.db.strategies.off.tags"] = "kong/db/strategies/off/tags.lua",
 
     ["kong.db.migrations.state"] = "kong/db/migrations/state.lua",
     ["kong.db.migrations.subsystems"] = "kong/db/migrations/subsystems.lua",
@@ -316,6 +320,7 @@ build = {
     ["kong.db.migrations.core.021_340_to_350"] = "kong/db/migrations/core/021_340_to_350.lua",
     ["kong.db.migrations.core.022_350_to_360"] = "kong/db/migrations/core/022_350_to_360.lua",
     ["kong.db.migrations.core.023_360_to_370"] = "kong/db/migrations/core/023_360_to_370.lua",
+    ["kong.db.migrations.core.024_370_to_380"] = "kong/db/migrations/core/024_370_to_380.lua",
     ["kong.db.migrations.operations.200_to_210"] = "kong/db/migrations/operations/200_to_210.lua",
     ["kong.db.migrations.operations.212_to_213"] = "kong/db/migrations/operations/212_to_213.lua",
     ["kong.db.migrations.operations.280_to_300"] = "kong/db/migrations/operations/280_to_300.lua",

kong/clustering/rpc/manager.lua

@@ -42,6 +42,7 @@ function _M.new(conf, node_id)
     -- clients[node_id]: { socket1 => true, socket2 => true, ... }
     clients = {},
     client_capabilities = {},
+    client_ips = {},  -- store DP node's ip addr
     node_id = node_id,
     conf = conf,
     cluster_cert = assert(clustering_tls.get_cluster_cert(conf)),
@@ -75,16 +76,18 @@ end
 
 
 function _M:_remove_socket(socket)
-  local sockets = assert(self.clients[socket.node_id])
+  local node_id = socket.node_id
+  local sockets = assert(self.clients[node_id])
 
   assert(sockets[socket])
 
   sockets[socket] = nil
 
   if table_isempty(sockets) then
-    self.clients[socket.node_id] = nil
-    self.client_capabilities[socket.node_id] = nil
-    assert(self.concentrator:_enqueue_unsubscribe(socket.node_id))
+    self.clients[node_id] = nil
+    self.client_ips[node_id] = nil
+    self.client_capabilities[node_id] = nil
+    assert(self.concentrator:_enqueue_unsubscribe(node_id))
   end
 end
 
@@ -255,6 +258,9 @@ function _M:handle_websocket()
   local s = socket.new(self, wb, node_id)
   self:_add_socket(s, rpc_capabilities)
 
+  -- store DP's ip addr
+  self.client_ips[node_id] = ngx_var.remote_addr
+
   s:start()
   local res, err = s:join()
   self:_remove_socket(s)
@@ -362,4 +368,9 @@ function _M:get_peers()
 end
 
 
+function _M:get_peer_ip(node_id)
+  return self.client_ips[node_id]
+end
+
+
 return _M

kong/clustering/services/sync/hooks.lua

@@ -0,0 +1,179 @@
+local _M = {}
+local _MT = { __index = _M, }
+
+
+local hooks = require("kong.hooks")
+local EMPTY = require("kong.tools.table").EMPTY
+
+
+local ipairs = ipairs
+local ngx_null = ngx.null
+local ngx_log = ngx.log
+local ngx_ERR = ngx.ERR
+local ngx_DEBUG = ngx.DEBUG
+
+
+local DEFAULT_PAGE_SIZE = 512
+
+
+function _M.new(strategy)
+  local self = {
+    strategy = strategy,
+  }
+
+  return setmetatable(self, _MT)
+end
+
+
+local function get_all_nodes_with_sync_cap()
+  local res, err = kong.db.clustering_data_planes:page(DEFAULT_PAGE_SIZE)
+  if err then
+    return nil, "unable to query DB " .. err
+  end
+
+  if not res then
+    return EMPTY
+  end
+
+  local ret = {}
+  local ret_n = 0
+
+  for _, row in ipairs(res) do
+    for _, c in ipairs(row.rpc_capabilities) do
+      if c == "kong.sync.v2" then
+        ret_n = ret_n + 1
+        ret[ret_n] = row.id
+        break
+      end
+    end
+  end
+
+  return ret
+end
+
+
+function _M:notify_all_nodes()
+  local latest_version, err = self.strategy:get_latest_version()
+  if not latest_version then
+    ngx_log(ngx_ERR, "can not get the latest version: ", err)
+    return
+  end
+
+  local msg = { default = { new_version = latest_version, }, }
+
+  for _, node in ipairs(get_all_nodes_with_sync_cap()) do
+    local res, err = kong.rpc:call(node, "kong.sync.v2.notify_new_version", msg)
+    if not res then
+      if not err:find("requested capability does not exist", nil, true) then
+        ngx_log(ngx_ERR, "unable to notify new version: ", err)
+      end
+
+    else
+      ngx_log(ngx_DEBUG, "notified ", node, " ", latest_version)
+    end
+  end
+end
+
+
+function _M:entity_delta_writer(row, name, options, ws_id, is_delete)
+  local deltas = {
+    {
+      type = name,
+      id = row.id,
+      ws_id = ws_id,
+      row = is_delete and ngx_null or row,
+    },
+  }
+
+  local res, err = self.strategy:insert_delta(deltas)
+  if not res then
+    self.strategy:cancel_txn()
+    return nil, err
+  end
+
+  res, err = self.strategy:commit_txn()
+  if not res then
+    self.strategy:cancel_txn()
+    return nil, err
+  end
+
+  self:notify_all_nodes()
+
+  return row -- for other hooks
+end
+
+
+-- only control plane has these delta operations
+function _M:register_dao_hooks()
+  local function is_db_export(name)
+    local db_export = kong.db[name].schema.db_export
+    return db_export == nil or db_export == true
+  end
+
+  -- common hook functions (pre/fail/post)
+
+  local function pre_hook_func(entity, name, options)
+    if not is_db_export(name) then
+      return true
+    end
+
+    return self.strategy:begin_txn()
+  end
+
+  local function fail_hook_func(err, entity, name)
+    if not is_db_export(name) then
+      return
+    end
+
+    local res, err = self.strategy:cancel_txn()
+    if not res then
+      ngx_log(ngx_ERR, "unable to cancel cancel_txn: ", tostring(err))
+    end
+  end
+
+  local function post_hook_writer_func(row, name, options, ws_id)
+    if not is_db_export(name) then
+      return row
+    end
+
+    return self:entity_delta_writer(row, name, options, ws_id)
+  end
+
+  local function post_hook_delete_func(row, name, options, ws_id, cascade_entries)
+    if not is_db_export(name) then
+      return row
+    end
+
+    -- set lmdb value to ngx_null then return row
+    return self:entity_delta_writer(row, name, options, ws_id, true)
+  end
+
+  local dao_hooks = {
+    -- dao:insert
+    ["dao:insert:pre"]  = pre_hook_func,
+    ["dao:insert:fail"] = fail_hook_func,
+    ["dao:insert:post"] = post_hook_writer_func,
+
+    -- dao:delete
+    ["dao:delete:pre"]  = pre_hook_func,
+    ["dao:delete:fail"] = fail_hook_func,
+    ["dao:delete:post"] = post_hook_delete_func,
+
+    -- dao:update
+    ["dao:update:pre"]  = pre_hook_func,
+    ["dao:update:fail"] = fail_hook_func,
+    ["dao:update:post"] = post_hook_writer_func,
+
+    -- dao:upsert
+    ["dao:upsert:pre"]  = pre_hook_func,
+    ["dao:upsert:fail"] = fail_hook_func,
+    ["dao:upsert:post"] = post_hook_writer_func,
+  }
+
+  for ev, func in pairs(dao_hooks) do
+    hooks.register_hook(ev, func)
+  end
+end
+
+
+return _M

kong/clustering/services/sync/init.lua

@@ -0,0 +1,63 @@
+local _M = {}
+local _MT = { __index = _M, }
+
+
+local events = require("kong.clustering.events")
+local strategy = require("kong.clustering.services.sync.strategies.postgres")
+local rpc = require("kong.clustering.services.sync.rpc")
+
+
+-- TODO: what is the proper value?
+local FIRST_SYNC_DELAY = 0.5  -- seconds
+local EACH_SYNC_DELAY  = 30   -- seconds
+
+
+function _M.new(db, is_cp)
+  local strategy = strategy.new(db)
+
+  local self = {
+    db = db,
+    strategy = strategy,
+    rpc = rpc.new(strategy),
+    is_cp = is_cp,
+  }
+
+  -- only cp needs hooks
+  if is_cp then
+    self.hooks = require("kong.clustering.services.sync.hooks").new(strategy)
+  end
+
+  return setmetatable(self, _MT)
+end
+
+
+function _M:init(manager)
+  if self.hooks then
+    self.hooks:register_dao_hooks()
+  end
+  self.rpc:init(manager, self.is_cp)
+end
+
+
+function _M:init_worker()
+  -- is CP, enable clustering broadcasts
+  if self.is_cp then
+    events.init()
+
+    self.strategy:init_worker()
+    return
+  end
+
+  -- is DP, sync only in worker 0
+  if ngx.worker.id() ~= 0 then
+    return
+  end
+
+  -- sync to CP ASAP
+  assert(self.rpc:sync_once(FIRST_SYNC_DELAY))
+
+  assert(self.rpc:sync_every(EACH_SYNC_DELAY))
+end
+
+
+return _M