feat(clustering/rpc): emit event when RPC becomes ready (Kong#14034)

This removes the artificial race condition on sleeping a few seconds before starting the sync poll, resulting
in cleaner code and less error logs.

KAG-5895
KAG-5891

kong/clustering/data_plane.lua

@@ -80,10 +80,50 @@ function _M:init_worker(basic_info)
   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)
-    self:communicate(premature)
-  end))
+  local function start_communicate()
+    assert(ngx.timer.at(0, function(premature)
+      self:communicate(premature)
+    end))
+  end
+
+  -- does not config rpc sync
+  if not kong.sync then
+    start_communicate()
+    return
+  end
+
+  local worker_events = assert(kong.worker_events)
+
+  -- if rpc is ready we will check then decide how to sync
+  worker_events.register(function(capabilities_list)
+    -- we only check once
+    if self.inited then
+      return
+    end
+
+    local has_sync_v2
+
+    -- check cp's capabilities
+    for _, v in ipairs(capabilities_list) do
+      if v == "kong.sync.v2" then
+        has_sync_v2 = true
+        break
+      end
+    end
+
+    -- cp supports kong.sync.v2
+    if has_sync_v2 then
+      return
+    end
+
+    ngx_log(ngx_WARN, "sync v1 is enabled due to rpc sync can not work.")
+
+    self.inited = true
+
+    -- only run in process which worker_id() == 0
+    start_communicate()
+
+  end, "clustering:jsonrpc", "connected")
 end
 
 

kong/clustering/rpc/manager.lua

@@ -270,6 +270,16 @@ function _M:_meta_call(c, meta_cap, node_id)
     list = capabilities_list,
   }
 
+  -- tell outside that rpc is ready
+  local worker_events = assert(kong.worker_events)
+
+  -- notify this worker
+  local ok, err = worker_events.post_local("clustering:jsonrpc", "connected",
+                                           capabilities_list)
+  if not ok then
+    ngx_log(ngx_ERR, _log_prefix, "unable to post rpc connected event: ", err)
+  end
+
   return true
 end
 

kong/clustering/services/sync/init.lua

@@ -53,10 +53,39 @@ function _M:init_worker()
     return
   end
 
-  -- sync to CP ASAP
-  assert(self.rpc:sync_once(FIRST_SYNC_DELAY))
+  local worker_events = assert(kong.worker_events)
 
-  assert(self.rpc:sync_every(EACH_SYNC_DELAY))
+  -- if rpc is ready we will start to sync
+  worker_events.register(function(capabilities_list)
+    -- we only check once
+    if self.inited then
+      return
+    end
+
+    local has_sync_v2
+
+    -- check cp's capabilities
+    for _, v in ipairs(capabilities_list) do
+      if v == "kong.sync.v2" then
+        has_sync_v2 = true
+        break
+      end
+    end
+
+    -- cp does not support kong.sync.v2
+    if not has_sync_v2 then
+      ngx.log(ngx.WARN, "rpc sync is disabled in CP.")
+      return
+    end
+
+    self.inited = true
+
+    -- sync to CP ASAP
+    assert(self.rpc:sync_once(FIRST_SYNC_DELAY))
+
+    assert(self.rpc:sync_every(EACH_SYNC_DELAY))
+
+  end, "clustering:jsonrpc", "connected")
 end
 
 

kong/clustering/utils.lua

@@ -161,6 +161,7 @@ end
 
 function _M.is_dp_worker_process()
   if kong.configuration.role == "data_plane"
+      and not kong.sync -- privileged agent is only enabled when rpc sync is off
       and kong.configuration.dedicated_config_processing == true then
     return process_type() == "privileged agent"
   end

kong/init.lua

@@ -763,6 +763,7 @@ function Kong.init()
 
   require("resty.kong.var").patch_metatable()
 
+  -- NOTE: privileged_agent is disabled when rpc sync is on
   if config.dedicated_config_processing and is_data_plane(config) and not kong.sync then
     -- TODO: figure out if there is better value than 4096
     -- 4096 is for the cocurrency of the lua-resty-timer-ng
@@ -881,19 +882,14 @@ function Kong.init_worker()
   end
 
   if kong.clustering then
-    local is_cp = is_control_plane(kong.configuration)
-    local is_dp_sync_v1 = is_data_plane(kong.configuration) and not kong.sync
     local using_dedicated = kong.configuration.dedicated_config_processing
 
     -- CP needs to support both v1 and v2 sync
-    -- v1 sync is only enabled for DP if v2 sync is disabled
-    if is_cp or is_dp_sync_v1 then
-      kong.clustering:init_worker()
-    end
+    -- v1 sync is only enabled for DP if v2 sync is unavailble
+    kong.clustering:init_worker()
 
     -- see is_dp_worker_process() in clustering/utils.lua
     if using_dedicated and process.type() == "privileged agent" then
-      assert(not is_cp)
       return
     end
   end

spec/02-integration/09-hybrid_mode/15-cp_inert_rpc_sync_spec.lua

@@ -0,0 +1,124 @@
+local helpers = require "spec.helpers"
+local cjson = require("cjson.safe")
+local CLUSTERING_SYNC_STATUS = require("kong.constants").CLUSTERING_SYNC_STATUS
+
+for _, strategy in helpers.each_strategy() do
+
+describe("CP diabled Sync RPC #" .. strategy, function()
+
+  lazy_setup(function()
+    helpers.get_db_utils(strategy, {
+      "clustering_data_planes",
+    }) -- runs migrations
+
+    assert(helpers.start_kong({
+      role = "control_plane",
+      cluster_cert = "spec/fixtures/kong_clustering.crt",
+      cluster_cert_key = "spec/fixtures/kong_clustering.key",
+      database = strategy,
+      cluster_listen = "127.0.0.1:9005",
+      nginx_conf = "spec/fixtures/custom_nginx.template",
+      nginx_worker_processes = 2, -- multiple workers
+
+      cluster_rpc = "on", -- CP ENABLE rpc
+      cluster_rpc_sync = "off", -- CP DISABLE rpc sync
+    }))
+
+    assert(helpers.start_kong({
+      role = "data_plane",
+      database = "off",
+      prefix = "servroot2",
+      cluster_cert = "spec/fixtures/kong_clustering.crt",
+      cluster_cert_key = "spec/fixtures/kong_clustering.key",
+      cluster_control_plane = "127.0.0.1:9005",
+      proxy_listen = "0.0.0.0:9002",
+      nginx_conf = "spec/fixtures/custom_nginx.template",
+      nginx_worker_processes = 2, -- multiple workers
+
+      cluster_rpc = "on", -- DP ENABLE rpc
+      cluster_rpc_sync = "on", -- DP ENABLE rpc sync
+    }))
+  end)
+
+  lazy_teardown(function()
+    helpers.stop_kong("servroot2")
+    helpers.stop_kong()
+  end)
+
+  after_each(function()
+    helpers.clean_logfile("servroot2/logs/error.log")
+    helpers.clean_logfile()
+  end)
+
+  describe("works", function()
+    it("shows DP status", function()
+      helpers.wait_until(function()
+        local admin_client = helpers.admin_client()
+        finally(function()
+          admin_client:close()
+        end)
+
+        local res = assert(admin_client:get("/clustering/data-planes"))
+        local body = assert.res_status(200, res)
+        local json = cjson.decode(body)
+
+        for _, v in pairs(json.data) do
+          if v.ip == "127.0.0.1" then
+            assert.near(14 * 86400, v.ttl, 3)
+            assert.matches("^(%d+%.%d+)%.%d+", v.version)
+            assert.equal(CLUSTERING_SYNC_STATUS.NORMAL, v.sync_status)
+            return true
+          end
+        end
+      end, 10)
+
+      -- cp will not run rpc
+      assert.logfile().has.no.line("[rpc]", true)
+
+      -- dp will not run rpc too
+      assert.logfile("servroot2/logs/error.log").has.line(
+        "rpc sync is disabled in CP")
+      assert.logfile("servroot2/logs/error.log").has.line(
+        "sync v1 is enabled due to rpc sync can not work.")
+    end)
+  end)
+
+  describe("sync works", function()
+    it("proxy on DP follows CP config", function()
+      local admin_client = helpers.admin_client(10000)
+      finally(function()
+        admin_client:close()
+      end)
+
+      local res = assert(admin_client:post("/services", {
+        body = { name = "mockbin-service", url = "https://127.0.0.1:15556/request", },
+        headers = {["Content-Type"] = "application/json"}
+      }))
+      assert.res_status(201, res)
+
+      res = assert(admin_client:post("/services/mockbin-service/routes", {
+        body = { paths = { "/" }, },
+        headers = {["Content-Type"] = "application/json"}
+      }))
+
+      helpers.wait_until(function()
+        local proxy_client = helpers.http_client("127.0.0.1", 9002)
+
+        res = proxy_client:send({
+          method  = "GET",
+          path    = "/",
+        })
+
+        local status = res and res.status
+        proxy_client:close()
+        if status == 200 then
+          return true
+        end
+      end, 10)
+    end)
+  end)
+
+
+end)
+
+end -- for _, strategy