feat(clustering): do not enable kong.sync.v2 when connecting dp is older than cp

kong/clustering/rpc/manager.lua

@@ -12,25 +12,28 @@ local callbacks = require("kong.clustering.rpc.callbacks")
 local clustering_tls = require("kong.clustering.tls")
 local constants = require("kong.constants")
 local table_isempty = require("table.isempty")
-local pl_tablex = require("pl.tablex")
+local table_clone = require("table.clone")
+local table_remove = table.remove
 local cjson = require("cjson.safe")
 local string_tools = require("kong.tools.string")
 
 
 local ipairs = ipairs
 local ngx_var = ngx.var
 local ngx_ERR = ngx.ERR
+local ngx_INFO = ngx.INFO
 local ngx_DEBUG = ngx.DEBUG
 local ngx_log = ngx.log
 local ngx_exit = ngx.exit
 local ngx_time = ngx.time
 local exiting = ngx.worker.exiting
-local pl_tablex_makeset = pl_tablex.makeset
+local pl_tablex_makeset = require("pl.tablex").makeset
 local cjson_encode = cjson.encode
 local cjson_decode = cjson.decode
 local validate_client_cert = clustering_tls.validate_client_cert
 local CLUSTERING_PING_INTERVAL = constants.CLUSTERING_PING_INTERVAL
 local parse_proxy_url = require("kong.clustering.utils").parse_proxy_url
+local version_num = require("kong.clustering.compat.version").string_to_number
 
 
 local _log_prefix = "[rpc] "
@@ -42,7 +45,6 @@ local WS_OPTS = {
   timeout = constants.CLUSTERING_TIMEOUT,
   max_payload_len = kong.configuration.cluster_max_payload,
 }
-local KONG_VERSION = kong.version
 
 
 -- create a new RPC manager, node_id is own node_id
@@ -188,10 +190,36 @@ function _M:_handle_meta_call(c, cert)
   assert(type(info.kong_hostname) == "string")
   assert(type(info.kong_conf) == "table")
 
+  local rpc_capabilities = self.callbacks:get_capabilities_list()
+  -- For data planes older than the control plane, we don't want to enable
+  -- kong.sync.v2 because we decided to not add the compatibility layer to
+  -- it. The v1 sync implements the compatibility code, and thus by not
+  -- advertising the kong.sync.v2 the data plane will automatically fall
+  -- back to v1 sync.
+  --
+  -- In case we want to reverse the decision, the compatibility code for the
+  -- kong.sync.v2 can be found here: https://github.com/Kong/kong-ee/pull/11040
+  local dp_version = info.kong_version
+  if version_num(kong.version) > version_num(dp_version) then
+    local delta_index
+    for i, rpc_capability in ipairs(rpc_capabilities) do
+      if rpc_capability == "kong.sync.v2" then
+        delta_index = i
+        break
+      end
+    end
+    if delta_index then
+      ngx_log(ngx_INFO, "disabling kong.sync.v2 because the data plane is older ",
+                        "than the control plane, node_id: ", info.kong_node_id)
+      rpc_capabilities = table_clone(rpc_capabilities)
+      table_remove(rpc_capabilities, delta_index)
+    end
+  end
+
   local payload = {
     jsonrpc = jsonrpc.VERSION,
     result = {
-      rpc_capabilities = self.callbacks:get_capabilities_list(),
+      rpc_capabilities = rpc_capabilities,
       -- now we only support snappy
       rpc_frame_encoding = RPC_SNAPPY_FRAMED,
       },
@@ -239,7 +267,7 @@ function _M:_handle_meta_call(c, cert)
   -- store DP's ip addr
   self.client_info[node_id] = {
     ip = ngx_var.remote_addr,
-    version = info.kong_version,
+    version = dp_version,
     labels = labels,
     cert_details = cert_details,
   }
@@ -256,7 +284,7 @@ function _M:_meta_call(c, meta_cap, node_id)
     -- now we only support snappy
     rpc_frame_encodings =  { RPC_SNAPPY_FRAMED, },
 
-    kong_version = KONG_VERSION,
+    kong_version = kong.version,
     kong_hostname = kong.node.get_hostname(),
     kong_node_id = self.node_id,
     kong_conf = kong.configuration.remove_sensitive(),

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

@@ -3,9 +3,7 @@ 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()
-
+describe("CP disabled Sync RPC #" .. strategy, function()
   lazy_setup(function()
     helpers.get_db_utils(strategy, {
       "clustering_data_planes",
@@ -117,8 +115,91 @@ describe("CP diabled Sync RPC #" .. strategy, function()
       end, 10)
     end)
   end)
+end)
 
 
-end)
+describe("CP disables Sync RPC with older data planes #" .. strategy, function()
+  lazy_setup(function()
+    helpers.get_db_utils(strategy, {
+      "routes",
+      "services",
+      "clustering_data_planes",
+    }, {
+      "older-version",
+      "error-generator",
+      "error-generator-last",
+      "error-handler-log",
+    })
+
+    assert(helpers.start_kong({
+      role = "control_plane",
+      cluster_cert = "spec/fixtures/kong_clustering.crt",
+      cluster_cert_key = "spec/fixtures/kong_clustering.key",
+      database = strategy,
+      prefix = "servroot2",
+      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 = "on", -- CP ENABLE rpc sync
+    }))
+
+    assert(helpers.start_kong({
+      role = "data_plane",
+      database = "off",
+      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
+
+      plugins = "older-version,error-generator,error-generator-last,error-handler-log",
+      cluster_rpc = "on", -- DP ENABLE rpc
+      cluster_rpc_sync = "on", -- DP ENABLE rpc sync
+    }))
+  end)
+
+  lazy_teardown(function()
+    helpers.stop_kong()
+    helpers.stop_kong("servroot2")
+  end)
+
+  after_each(function()
+    helpers.clean_logfile()
+    helpers.clean_logfile("servroot2/logs/error.log")
+  end)
+
+  it("fallbacks to sync v1", 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("servroot2/logs/error.log").has.no.line("[rpc]", true)
+    assert.logfile("servroot2/logs/error.log").has.line(
+      "disabling kong.sync.v2 because the data plane is older than the control plane", true)
+
+    -- dp will not run rpc too
+    assert.logfile().has.line("rpc sync is disabled in CP")
+    assert.logfile().has.line("sync v1 is enabled due to rpc sync can not work.")
+  end)
+end)
 end -- for _, strategy

spec/fixtures/custom_plugins/kong/plugins/older-version/handler.lua

@@ -0,0 +1,31 @@
+local meta = require "kong.meta"
+
+
+local version = setmetatable({
+  major = 3,
+  minor = 9,
+  patch = 0,
+}, {
+  __tostring = function(t)
+    return string.format("%d.%d.%d%s", t.major, t.minor, t.patch,
+            t.suffix or "")
+  end
+})
+
+
+local OlderVersion =  {
+  VERSION = "1.0.0",
+  PRIORITY = 1000,
+}
+
+
+function OlderVersion:init_worker()
+  meta._VERSION = tostring(version)
+  meta._VERSION_TABLE = version
+  meta._SERVER_TOKENS = "kong/" .. tostring(version)
+  meta.version = tostring(version)
+  kong.version = meta._VERSION
+end
+
+
+return OlderVersion

spec/fixtures/custom_plugins/kong/plugins/older-version/schema.lua

@@ -0,0 +1,18 @@
+local typedefs = require "kong.db.schema.typedefs"
+
+
+return {
+  name = "older-version",
+  fields = {
+    {
+      protocols = typedefs.protocols { default = { "http", "https", "tcp", "tls", "grpc", "grpcs" } },
+    },
+    {
+      config = {
+        type = "record",
+        fields = {
+        },
+      },
+    },
+  },
+}