tests(clustering/rpc): add cases for sync.v2.get_delta (#14151)

KAG-6179

spec/02-integration/18-hybrid_rpc/07-notification_spec.lua

@@ -49,32 +49,26 @@ for _, strategy in helpers.each_strategy() do
         local name = nil
 
         -- cp logs
-        helpers.pwait_until(function()
-          assert.logfile(name).has.line(
-            "notification is hello", true)
-          assert.logfile(name).has.line(
-            "[rpc] notifying kong.test.notification(node_id:", true)
-          assert.logfile(name).has.line(
-            "[rpc] notification has no response", true)
-          assert.logfile(name).has.no.line(
-            "assert failed", true)
-          return true
-        end, 10)
+        assert.logfile(name).has.line(
+          "notification is hello", true, 10)
+        assert.logfile(name).has.line(
+          "[rpc] notifying kong.test.notification(node_id:", true, 10)
+        assert.logfile(name).has.line(
+          "[rpc] notification has no response", true, 10)
+        assert.logfile(name).has.no.line(
+          "assertion failed", true, 0)
 
         local name = "servroot2/logs/error.log"
 
         -- dp logs
-        helpers.pwait_until(function()
-          assert.logfile(name).has.line(
-            "[rpc] notifying kong.test.notification(node_id: control_plane) via local", true)
-          assert.logfile(name).has.line(
-            "notification is world", true)
-          assert.logfile(name).has.line(
-            "[rpc] notification has no response", true)
-          assert.logfile(name).has.no.line(
-            "assert failed", true)
-          return true
-        end, 10)
+        assert.logfile(name).has.line(
+          "[rpc] notifying kong.test.notification(node_id: control_plane) via local", true, 10)
+        assert.logfile(name).has.line(
+          "notification is world", true, 10)
+        assert.logfile(name).has.line(
+          "[rpc] notification has no response", true, 10)
+        assert.logfile(name).has.no.line(
+          "assertion failed", true, 0)
 
       end)
     end)

spec/02-integration/18-hybrid_rpc/08-sync_v2_get_delta_spec.lua

@@ -0,0 +1,72 @@
+local helpers = require "spec.helpers"
+
+
+-- register a rpc connected event in custom plugin rpc-get-delta-test
+-- ENABLE rpc sync on cp side for testing sync.v2.get_delta
+-- DISABLE rpc sync on dp side
+for _, strategy in helpers.each_strategy() do
+  describe("Hybrid Mode 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",
+        plugins = "bundled",
+        nginx_worker_processes = 4, -- multiple workers
+        cluster_rpc = "on", -- enable rpc
+        cluster_rpc_sync = "on", -- enable 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",
+        plugins = "bundled,rpc-get-delta-test",
+        nginx_worker_processes = 4, -- multiple workers
+        cluster_rpc = "on", -- enable rpc
+        cluster_rpc_sync = "off", -- disable rpc sync
+      }))
+    end)
+
+    lazy_teardown(function()
+      helpers.stop_kong("servroot2")
+      helpers.stop_kong()
+    end)
+
+    describe("sync.v2.get_delta works", function()
+      it("on cp side", function()
+        local name = "servroot2/logs/error.log"
+
+        -- dp logs
+        assert.logfile(name).has.line(
+          "kong.sync.v2.get_delta ok", true, 10)
+        assert.logfile(name).has.no.line(
+          "assertion failed", true, 0)
+        assert.logfile(name).has.no.line(
+          "[error]", true, 0)
+
+        local name = nil
+
+        -- cp logs
+        assert.logfile(name).has.no.line(
+          "assertion failed", true, 0)
+        assert.logfile(name).has.no.line(
+          "[error]", true, 0)
+
+      end)
+    end)
+  end)
+end -- for _, strategy

spec/fixtures/custom_plugins/kong/plugins/rpc-get-delta-test/handler.lua

@@ -0,0 +1,60 @@
+local rep = string.rep
+local isempty = require("table.isempty")
+
+
+local RpcSyncV2GetDeltaTestHandler = {
+  VERSION = "1.0",
+  PRIORITY = 1000,
+}
+
+
+function RpcSyncV2GetDeltaTestHandler:init_worker()
+  local worker_events = assert(kong.worker_events)
+
+  -- if rpc is ready we will send test calls
+  -- cp's version now is "v02_00000"
+  worker_events.register(function(capabilities_list)
+    local node_id = "control_plane"
+    local method = "kong.sync.v2.get_delta"
+
+    -- no field `default` for kong.sync.v2.get_delta
+    local msg = {}
+    local res, err = kong.rpc:call(node_id, method, msg)
+
+    assert(not res)
+    assert(err == "default namespace does not exist inside params")
+
+    -- version is invalid
+    local msg = { default = { version = rep("A", 32), }, }
+    local res, err = kong.rpc:call(node_id, method, msg)
+
+    assert(type(res) == "table")
+    assert(not isempty(res.default.deltas))
+    assert(res.default.wipe == true)
+    assert(not err)
+
+    -- dp's version is greater than cp's version
+    local msg = { default = { version = "v02_" .. rep("A", 28), }, }
+    local res, err = kong.rpc:call(node_id, method, msg)
+
+    assert(type(res) == "table")
+    assert(not isempty(res.default.deltas))
+    assert(res.default.wipe == true)
+    assert(not err)
+
+    -- dp's version is equal to cp's version
+    local msg = { default = { version = "v02_" .. rep("0", 28), }, }
+    local res, err = kong.rpc:call(node_id, method, msg)
+
+    assert(type(res) == "table")
+    assert(isempty(res.default.deltas))
+    assert(res.default.wipe == false)
+    assert(not err)
+
+    ngx.log(ngx.DEBUG, "kong.sync.v2.get_delta ok")
+
+  end, "clustering:jsonrpc", "connected")
+end
+
+
+return RpcSyncV2GetDeltaTestHandler

spec/fixtures/custom_plugins/kong/plugins/rpc-get-delta-test/schema.lua

@@ -0,0 +1,12 @@
+return {
+  name = "rpc-get-delta-test",
+  fields = {
+    {
+      config = {
+        type = "record",
+        fields = {
+        },
+      },
+    },
+  },
+}