api: Add API to kill connection to a particular host

For testing failure scenarios.

api/api-doc/error_injection.json

@@ -90,6 +90,30 @@
             }
          ]
       },
+      {
+         "path":"/v2/error_injection/disconnect/{ip}",
+         "operations":[
+            {
+               "method":"POST",
+               "summary":"Drop connection to a given IP",
+               "type":"void",
+               "nickname":"inject_disconnect",
+               "produces":[
+                  "application/json"
+               ],
+               "parameters":[
+                  {
+                     "name":"ip",
+                     "description":"IP address to disconnect from",
+                     "required":true,
+                     "allowMultiple":false,
+                     "type":"string",
+                     "paramType":"path"
+                  }
+               ]
+            }
+         ]
+      },
       {
          "path":"/v2/error_injection/injection",
          "operations":[

api/messaging_service.cc

@@ -10,6 +10,7 @@
 #include "message/messaging_service.hh"
 #include <seastar/rpc/rpc_types.hh>
 #include "api/api-doc/messaging_service.json.hh"
+#include "api/api-doc/error_injection.json.hh"
 #include <iostream>
 #include <sstream>
 
@@ -19,6 +20,8 @@ using namespace netw;
 
 namespace api {
 
+namespace hf = httpd::error_injection_json;
+
 using shard_info = messaging_service::shard_info;
 using msg_addr = messaging_service::msg_addr;
 
@@ -142,6 +145,14 @@ void set_messaging_service(http_context& ctx, routes& r, sharded<netw::messaging
             return make_ready_future<json::json_return_type>(res);
         });
     });
+
+    hf::inject_disconnect.set(r, [&ms] (std::unique_ptr<request> req) -> future<json::json_return_type> {
+        auto ip = msg_addr(req->param["ip"]);
+        co_await ms.invoke_on_all([ip] (netw::messaging_service& ms) {
+            ms.remove_rpc_client(ip);
+        });
+        co_return json::json_void();
+    });
 }
 
 void unset_messaging_service(http_context& ctx, routes& r) {
@@ -155,6 +166,7 @@ void unset_messaging_service(http_context& ctx, routes& r) {
     get_respond_completed_messages.unset(r);
     get_version.unset(r);
     get_dropped_messages_by_ver.unset(r);
+    hf::inject_disconnect.unset(r);
 }
 
 }

test/pylib/rest_client.py

@@ -238,6 +238,9 @@ class ScyllaRESTAPIClient():
     async def message_injection(self, node_ip: str, injection: str) -> None:
         await self.client.post(f"/v2/error_injection/injection/{injection}/message", host=node_ip)
 
+    async def inject_disconnect(self, node_ip: str, ip_to_disconnect_from: str) -> None:
+        await self.client.post(f"/v2/error_injection/disconnect/{ip_to_disconnect_from}", host=node_ip)
+
     async def get_logger_level(self, node_ip: str, logger: str) -> str:
         """Get logger level"""
         return await self.client.get_text(f"/system/logger/{logger}", host=node_ip)