fix(clustering) DP errors when CP exits

Fix FT-3204

kong/clustering/init.lua

@@ -176,5 +176,12 @@ function _M:init_worker()
   end
 end
 
+function _M:exit_worker()
+  if self.conf.role == "control_plane" then
+    if not kong.configuration.legacy_hybrid_protocol then
+      self.wrpc_handler:exit_worker()
+    end
+  end
+end
 
 return _M

kong/clustering/wrpc_control_plane.lua

@@ -367,5 +367,16 @@ function _M:init_worker(plugins_list)
                self.conf.db_update_frequency)
 end
 
+function _M:exit_worker()
+  -- politely close all connections so they will not error on CP's exit
+  for ws, client in pairs(self.clients) do
+    ngx.thread.spawn(function()
+      client.peer:close()
+      -- wait for closing frame to be sent
+      client.peer:wait_threads()
+      self.clients[ws] = nil
+    end)
+  end
+end
 
 return _M

kong/init.lua

@@ -751,6 +751,10 @@ function Kong.exit_worker()
   if kong.configuration.role ~= "control_plane" and ngx.worker.id() == 0 then
     plugin_servers.stop()
   end
+
+  if kong.clustering then
+    kong.clustering:exit_worker()
+  end
 end
 
 

kong/tools/wrpc/threads.lua

@@ -76,7 +76,7 @@ local function step(wrpc_peer)
     msg, err = wrpc_peer:receive()
   end
 
-  if err ~= nil and not endswith(err, ": timeout") then
+  if err ~= nil and not endswith(err, ": timeout") and not endswith(err, ": closed") then
     ngx_log(NOTICE, "[wRPC] WebSocket frame: ", err)
     wrpc_peer.closing = true
     return false, err

spec/02-integration/09-hybrid_mode/02-start_stop_spec.lua

@@ -132,3 +132,37 @@ for cluster_protocol, conf in pairs(confs) do
     end
   end)
 end
+
+-- note that lagacy modes still error when CP exits
+describe("when CP exits DP does not error #t", function()
+  local need_exit = true
+  setup(function()
+    assert(helpers.start_kong({
+      role = "control_plane",
+      prefix = "servroot1",
+      cluster_cert = "spec/fixtures/kong_clustering.crt",
+      cluster_cert_key = "spec/fixtures/kong_clustering.key",
+      cluster_listen = "127.0.0.1:9005",
+    }))
+    assert(helpers.start_kong({
+      role = "data_plane",
+      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",
+      database = "off",
+    }))
+  end)
+  teardown(function()
+    if need_exit then
+      helpers.stop_kong("servroot1")
+    end
+    helpers.stop_kong("servroot2")
+  end)
+  it("", function ()
+    assert(helpers.stop_kong("servroot1"))
+    need_exit = false
+    assert.logfile("servroot2/logs/error.log").has.no.line("[error]", true)
+  end)
+end)