From 2f9c52bb70f2515358ac8845a57c5d50ba5ec746 Mon Sep 17 00:00:00 2001
From: xumin <xumin.zhou@konghq.com>
Date: Tue, 11 Feb 2025 16:36:14 +0800
Subject: [PATCH] feat(rpc): connection lost event

Precondition for KAG-6178
---
 kong/clustering/rpc/manager.lua | 17 ++++++++++++-----
 1 file changed, 12 insertions(+), 5 deletions(-)

diff --git a/kong/clustering/rpc/manager.lua b/kong/clustering/rpc/manager.lua
index bd33f9db0a6..21d85a74305 100644
--- a/kong/clustering/rpc/manager.lua
+++ b/kong/clustering/rpc/manager.lua
@@ -587,7 +587,7 @@ function _M:connect(premature, node_id, host, path, cert, key)
 
   local ok, err = c:connect(uri, opts)
   if not ok then
-    ngx_log(ngx_ERR, _log_prefix, "unable to connect to peer: ", err)
+    err = "unable to connect to peer: " .. (err or "unknown error")
     goto err
   end
 
@@ -596,7 +596,7 @@ function _M:connect(premature, node_id, host, path, cert, key)
     -- FIXME: resp_headers should not be case sensitive
 
     if not resp_headers or not resp_headers["sec_websocket_protocol"] then
-      ngx_log(ngx_ERR, _log_prefix, "peer did not provide sec_websocket_protocol, node_id: ", node_id)
+      err = "peer did not provide sec_websocket_protocol, node_id: " .. (node_id or "unknown")
       c:send_close() -- can't do much if this fails
       goto err
     end
@@ -605,7 +605,7 @@ function _M:connect(premature, node_id, host, path, cert, key)
     local meta_cap = resp_headers["sec_websocket_protocol"]
 
     if meta_cap ~= RPC_META_V1 then
-      ngx_log(ngx_ERR, _log_prefix, "did not support protocol : ", meta_cap)
+      err = "did not support protocol : " .. (type(meta_cap) == "string" and meta_cap or "unknown")
       c:send_close() -- can't do much if this fails
       goto err
     end
@@ -613,8 +613,8 @@ function _M:connect(premature, node_id, host, path, cert, key)
     -- if timeout (default is 5s) we will close the connection
     local ok, err = self:_meta_call(c, meta_cap, node_id)
     if not ok then
-      ngx_log(ngx_ERR, _log_prefix, "unable to handshake with server, node_id: ", node_id,
-                       " err: ", err)
+      err = "unable to handshake with server, node_id: " .. (node_id or "unknown")
+                    .. ", err: " .. (err or "unknown error")
       c:send_close() -- can't do much if this fails
       goto err
     end
@@ -634,6 +634,13 @@ function _M:connect(premature, node_id, host, path, cert, key)
   end
 
   ::err::
+  local worker_events = assert(kong.worker_events)
+
+  -- notify this worker
+  local ok, err = worker_events.post_local("clustering:jsonrpc", "connection_lost")
+  if not ok then
+    ngx_log(ngx_ERR, _log_prefix, "unable to post rpc connection_lost event: ", err)
+  end
 
   if not exiting() then
     c:close()